from __future__ import annotations import logging from pathlib import Path from typing import Any logger = logging.getLogger(__name__) class DrawingPricingStateMixin: def _get_state_prices_context(self) -> str: prices_context = "" try: with self._erp_client.get_connection() as conn: with conn.cursor() as cur: cur.execute( "SELECT material_name, state_unit_price_vnd, document_reference, source_url FROM erp_state_published_prices ORDER BY publish_period DESC, material_name ASC LIMIT 640;" ) rows = cur.fetchall() if rows: prices_context = ( "BẢNG GIÁ VẬT LIỆU LIÊN SỞ & THỊ TRƯỜNG THAM KHẢO 2026:\n" ) for r in rows: url = ( r.get("source_url") or f"/dashboard/material-price-comparison?q={r['material_name']}" ) prices_context += f"- {r['material_name']}: {float(r['state_unit_price_vnd'])} VNĐ (Nguồn: {url})\n" except Exception: pass return prices_context def auto_discover_project_hsmt( self, file_path_str: str = "", filename: str = "", drawing_hint: str = "", ) -> dict[str, Any] | None: """Chủ động rà soát kho dữ liệu dự án HĐ-2026 và cơ sở dữ liệu để tìm file HSMT / Dự toán tương ứng.""" combined = f"{filename} {drawing_hint} {file_path_str}".lower() # Target directory base_dir = Path("HĐ-2026") if not base_dir.exists(): return None candidates = [] try: for p in base_dir.glob("**/*"): if p.is_file() and p.suffix.lower() in [ ".xls", ".xlsx", ".doc", ".docx", ".pdf", ]: full_p_str = str(p).lower() score = 0 if any( k in combined for k in [ "ong den", "ông đen", "kênh ông đen", "kenh ong den", "ttdvsnc kiến minh", "ttdvsnc", ] ) and any( k in full_p_str for k in [ "ong den", "ông đen", "ttdvsnc", "biểu_mẫu", "bieu_mau", ] ): score += 90 if "biểu_mẫu" in full_p_str or "bieu_mau" in full_p_str: score += 50 elif p.suffix.lower() in [".xls", ".xlsx"]: score += 40 elif any( k in combined for k in ["đại thắng", "dai thang", "nvh"] ) and any( k in full_p_str for k in ["đại thắng", "dai thang", "nvh"] ): if p.suffix.lower() in [".xls", ".xlsx"]: score += 60 elif p.suffix.lower() in [".doc", ".docx"]: score += 40 if any( k in p.name.lower() for k in ["đại thắng", "dai thang", "nvh", "thôn", "thon"] ): score += 50 score += 30 elif any( k in combined for k in [ "kiến minh", "kien minh", "bến kem", "ben kem", "phong cầu", "phong cau", "cống", "cong", ] ) and any( k in full_p_str for k in [ "kiến minh", "kien minh", "bến kem", "ben kem", "muong", "cong", "phong cầu", ] ): if "biểu_mẫu" in full_p_str or "bieu_mau" in full_p_str: score += 80 elif p.suffix.lower() in [".xls", ".xlsx"]: score += 60 elif p.suffix.lower() in [".doc", ".docx"]: score += 40 score += 30 elif any( k in combined for k in ["đa độ", "da do", "kè", "ke"] ) and any(k in full_p_str for k in ["đa độ", "da do", "kè", "ke"]): score += 30 if score >= 30: candidates.append((score, p)) except Exception as scan_err: logger.debug("[TAKEOFF] Lỗi khi tự động quét HSMT: %s", scan_err) if not candidates: return None candidates.sort(key=lambda x: x[0], reverse=True) best_path = candidates[0][1] return { "file_path": str(best_path), "file_name": best_path.name, "project_folder": best_path.parent.name, "file_ext": best_path.suffix.lower(), } @staticmethod def _normalize_search_text(s: str) -> str: import re s = (s or "").lower().strip() s = re.sub(r"[_\-\.,\t\r\n]+", " ", s) s = re.sub(r"\s+", " ", s) return s def match_and_apply_state_prices( self, items: list[dict[str, Any]] ) -> list[dict[str, Any]]: """Tự động khớp nối đơn giá định mức & công bố giá Sở Xây Dựng Hải Phòng 2026 (erp_state_published_prices) với tỷ lệ đơn vị chuẩn xác.""" updated = [] for itm in items: c_itm = dict(itm) name = c_itm.get("item_name", "") unit = str(c_itm.get("unit", "")).strip() nl = self._normalize_search_text(name) ul = self._normalize_search_text(unit) # Check if item already has authentic valid unit price and norm code cur_price = float(c_itm.get("unit_price_vnd") or 0) # 1. Đơn giá công bố Sở Xây Dựng Hải Phòng 2026 (Chuẩn 100% thực tế Định Sơn) # Chỉ áp dụng khi hạng mục chưa có đơn giá (cur_price <= 0) hoặc là bóc tách vật liệu thô # 1.1 Cát xây dựng (Rule 11: Cát vàng BT 710k, Cát xây trát 510k, Cát đen 215k) if ( ("cát vàng" in nl or "cát bê tông" in nl or "cát hạt lớn" in nl) and ("m3" in ul or "100m3" in ul) and cur_price <= 0 ): c_itm["unit_price_vnd"] = 710000.0 c_itm["price_source_url"] = ( "/dashboard/material-price-comparison?q=TBG-CAT-VANG-BT" ) elif ( ("cát xây trát" in nl or "cát hạt trung" in nl or "cát mịn" in nl) and ("m3" in ul or "100m3" in ul) and cur_price <= 0 ): c_itm["unit_price_vnd"] = 510000.0 c_itm["price_source_url"] = ( "/dashboard/material-price-comparison?q=TBG-CAT-VANG-TRUNG" ) elif ( ("cát đen" in nl or "cát san" in nl or "đệm cát" in nl) and ("m3" in ul or "100m3" in ul) and cur_price <= 0 ): if "100m3" in ul or "100 m3" in ul: c_itm["unit_price_vnd"] = 21500000.0 else: c_itm["unit_price_vnd"] = 215000.0 c_itm["price_source_url"] = ( "/dashboard/material-price-comparison?q=TBG-CAT-DEN-SAN" ) # 1.2 Đất đắp K95, Đào móng & Vận chuyển đất bùn elif ( ("đất đắp" in nl or "k95" in nl or "hoàn trả" in nl) and ("m3" in ul or "100m3" in ul) and cur_price <= 0 ): if "100m3" in ul or "100 m3" in ul: c_itm["unit_price_vnd"] = 16500000.0 else: c_itm["unit_price_vnd"] = 165000.0 c_itm["price_source_url"] = ( "/dashboard/material-price-comparison?q=TBG-DAT-DAP-K95" ) elif ( ("đào móng" in nl or "đào đất" in nl or "đào hố" in nl) and "vận chuyển" not in nl and cur_price <= 0 ): if "100m3" in ul or "100 m3" in ul: c_itm["unit_price_vnd"] = 4800000.0 else: c_itm["unit_price_vnd"] = 48000.0 c_itm["price_source_url"] = ( "/dashboard/material-price-comparison?q=TBG-MAY-DAO-08M3" ) elif "vận chuyển" in nl and ("đất" in nl or "bùn" in nl) and cur_price <= 0: if "100m3" in ul or "100 m3" in ul: c_itm["unit_price_vnd"] = 7500000.0 else: c_itm["unit_price_vnd"] = 75000.0 c_itm["price_source_url"] = ( "/dashboard/material-price-comparison?q=TBG-VC-OTO-TU-DO" ) # 1.3 Bê tông thương phẩm & Bê tông lót (Chỉ áp dụng khi cur_price <= 0 cho ĐVT m3 / 100m3, LOẠI TRỪ vữa trát / láng m2) elif ( ("m3" in ul or "100m3" in ul) and ("bê tông" in nl or "bt" in nl) and not any( k in nl for k in ["láng", "trát", "vữa", "sơn", "đục", "phá", "tháo", "ván"] ) and cur_price <= 0 ): if "bê tông lót" in nl or "bt lót" in nl or "m100" in nl: c_itm["unit_price_vnd"] = 1050000.0 c_itm["price_source_url"] = ( "/dashboard/material-price-comparison?q=Bê+tông+M100" ) elif "m250" in nl or "bản đáy" in nl or "thành" in nl or "nắp" in nl: c_itm["unit_price_vnd"] = 1240000.0 c_itm["price_source_url"] = ( "/dashboard/material-price-comparison?q=Bê+tông+M250" ) elif ( "m300" in nl or "m350" in nl or "dầm" in nl or "cột" in nl or "sàn" in nl ): c_itm["unit_price_vnd"] = 1350000.0 c_itm["price_source_url"] = ( "/dashboard/material-price-comparison?q=Bê+tông+M300" ) # 1.4 Thép xây dựng & Cốt thép (Chỉ áp dụng khi cur_price <= 0 và ĐVT là tấn / kg / 100kg, LOẠI TRỪ dây sét, sơn, dàn giáo, tháo dỡ, xà gồ, cửa) elif ( ( ("cốt thép" in nl) or ( "thép" in nl and not any( k in nl for k in [ "dây", "sét", "sơn", "dàn giáo", "giàn giáo", "tháo dỡ", "cừ", "ván", "cửa", "hoa sắt", "lan can", "xà gồ", "khung sắt", "cấu kiện", "tấm đan", ] ) ) ) and any(u in ul for u in ["tấn", "tan", "kg", "100kg", "100 kg"]) and cur_price <= 0 ): is_cb400 = "cb400" in nl base_ton_price = 16450000.0 if is_cb400 else 16150000.0 if "tấn" in ul or "tan" in ul: c_itm["unit_price_vnd"] = base_ton_price elif "100kg" in ul or "100 kg" in ul: c_itm["unit_price_vnd"] = base_ton_price / 10.0 # 1.615.000 đ/100kg elif "kg" in ul: c_itm["unit_price_vnd"] = base_ton_price / 1000.0 # 16.150 đ/kg c_itm["price_source_url"] = ( "/dashboard/material-price-comparison?q=Thép+CB300" ) # 1.5 Ván khuôn (Chỉ áp dụng khi cur_price <= 0 và ĐVT là m2 / 100m2) elif ( "ván khuôn" in nl and ("m2" in ul or "100m2" in ul or "100 m2" in ul) and cur_price <= 0 ): if "100m2" in ul or "100 m2" in ul: c_itm["unit_price_vnd"] = 2850000.0 elif "m2" in ul: c_itm["unit_price_vnd"] = 215000.0 if "phim" in nl else 165000.0 c_itm["price_source_url"] = ( "/dashboard/material-price-comparison?q=Ván+khuôn" ) # 1.6 Cọc tre & Cừ Larsen elif "cọc tre" in nl and cur_price <= 0: if "100m" in ul or "100 m" in ul: c_itm["unit_price_vnd"] = 2650000.0 else: c_itm["unit_price_vnd"] = 24500.0 c_itm["price_source_url"] = ( "/dashboard/material-price-comparison?q=Cọc+tre" ) elif "cừ larsen" in nl and cur_price <= 0: if "100m" in ul or "100 m" in ul: c_itm["unit_price_vnd"] = 6200000.0 else: c_itm["unit_price_vnd"] = 125000.0 c_itm["price_source_url"] = ( "/dashboard/material-price-comparison?q=Cừ+Larsen" ) # 1.7 Trung chuyển & bốc xếp vật liệu elif ("bốc xếp" in nl or "bốc dỡ" in nl) and cur_price <= 0: if "thép" in nl or "tấn" in ul: c_itm["unit_price_vnd"] = 55000.0 elif "cát" in nl: c_itm["unit_price_vnd"] = 20000.0 else: c_itm["unit_price_vnd"] = 45000.0 elif ( "vận chuyển" in nl and ("cát" in nl or "đá" in nl or "xi măng" in nl) and cur_price <= 0 ): if "thép" in nl or "tấn" in ul: c_itm["unit_price_vnd"] = 120000.0 elif "cát" in nl: c_itm["unit_price_vnd"] = 35000.0 elif "xi măng" in nl: c_itm["unit_price_vnd"] = 95000.0 else: c_itm["unit_price_vnd"] = 75000.0 # 2. Strict Unit Scale Guard (Chỉ scale khi unit price quá nhỏ bất thường dưới ngưỡng ĐVT lớn) p_val = float(c_itm.get("unit_price_vnd") or 0) if ( ("100m3" in ul or "100 m3" in ul) and 0 < p_val < 500000 or ("100m2" in ul or "100 m2" in ul) and 0 < p_val < 50000 or ("100m" in ul or "100 m" in ul) and 0 < p_val < 50000 ): c_itm["unit_price_vnd"] = p_val * 100.0 c_itm["total_amount_vnd"] = round( float(c_itm.get("quantity") or 0) * float(c_itm.get("unit_price_vnd") or 0) ) updated.append(c_itm) return updated return updated def enforce_learned_rules( self, items: list[dict[str, Any]], drawing_type: str = "general", takeoff_id: str | None = None, ) -> list[dict[str, Any]]: """Thực thi bắt buộc toàn bộ Quy tắc Rút Kinh Nghiệm đã học từ Kỹ Sư Trưởng QS (erp_ai_takeoff_learned_rules).""" import re client = getattr(self, "_erp_client", None) if not client: return items rules = client.list_takeoff_learned_rules(is_active=True) if not rules: return items updated = [dict(it) for it in items] applied_rule_ids = set() for r in rules: r_id = str(r.get("id")) rule_type = (r.get("drawing_type") or "all").lower() err_pat = self._normalize_search_text(r.get("error_pattern") or "") corr_rule = r.get("correction_rule") or "" corr_norm = self._normalize_search_text(corr_rule) # Check drawing type compatibility if rule_type not in ["all", "general", drawing_type.lower()]: any_match = any( self._normalize_search_text(it.get("item_name", "")) in corr_norm for it in updated ) if not any_match: continue # Match against items for itm in updated: item_name = itm.get("item_name", "") nl = self._normalize_search_text(item_name) ul = self._normalize_search_text(itm.get("unit", "")) is_target = False if ( nl in corr_norm or nl in err_pat or ("đào" in nl and "đất" in nl) and ("đào vận chuyển đất" in corr_norm or "đào đất" in corr_norm) or ("vận chuyển" in nl and "bùn" in nl) and ("vận chuyển bùn" in corr_norm or "vận chuyển đất" in corr_norm) or "đài móng" in nl and "đài móng" in corr_norm or ( ("cốt thép" in nl or "thép cb" in nl or "thép vằn" in nl) and not any( k in nl for k in [ "cừ", "thuê", "ván", "khuôn", "sơn", "dây", "sét", "bốc", "vận chuyển", "hoa sắt", "lan can", "cửa", "dàn giáo", "giàn giáo", ] ) ) and "cb400" in corr_norm and any(u in ul for u in ["tấn", "tan", "kg", "100kg", "100 kg"]) ): is_target = True if is_target: # 1. Check unit price directives if ( ("100m3" in ul or "100 m3" in ul or "100m3" in corr_norm) and ( "7500000" in corr_norm or "7 500 000" in corr_norm or "75000" in corr_norm or "75 000" in corr_norm ) and ( "đào" in nl or "vận chuyển" in nl or "bùn" in nl or "đất" in nl ) ): itm["unit_price_vnd"] = 7500000.0 itm["unit"] = "100m3" elif ( "thép cb400" in corr_norm and any( u in ul for u in ["tấn", "tan", "kg", "100kg", "100 kg"] ) and not any( k in nl for k in [ "cừ", "thuê", "ván", "khuôn", "sơn", "dây", "sét", "bốc", "vận chuyển", "hoa sắt", "lan can", "cửa", "dàn giáo", "giàn giáo", ] ) ): if "tấn" in ul or "tan" in ul: itm["unit_price_vnd"] = 16450000.0 elif "100kg" in ul or "100 kg" in ul: itm["unit_price_vnd"] = 1645000.0 elif "kg" in ul: itm["unit_price_vnd"] = 16450.0 # 2. Check quantity directives (e.g. 'phải là 45 m3', 'khối lượng là 45') qty_match = re.search( r"(?:phải là|khối lượng là|kl là|chỉ có)\s*(\d+(?:[.,]\d+)?)\s*(?:m3|m2|m|tấn|cọc|bộ)?", corr_rule, re.IGNORECASE, ) if qty_match: q_val = float(qty_match.group(1).replace(",", ".")) if 0 < q_val < 10000: itm["quantity"] = q_val # 3. Unit scale guard p_val = float(itm.get("unit_price_vnd") or 0) if ( ("100m3" in ul or "100 m3" in ul) and 0 < p_val < 500000 or ("100m2" in ul or "100 m2" in ul) and 0 < p_val < 50000 or ("100m" in ul or "100 m" in ul) and 0 < p_val < 50000 ): itm["unit_price_vnd"] = p_val * 100.0 itm["total_amount_vnd"] = round( float(itm.get("quantity") or 0) * float(itm.get("unit_price_vnd") or 0) ) applied_rule_ids.add(r_id) # Check for Supplemental Item (e.g. 'bổ sung ván khuôn đài móng 110 m2') if "bổ sung" in corr_norm or "thêm" in corr_norm: supp_match = re.search( r"(?:bổ sung|thêm)\s+([a-zA-Zà-ỹÀ-Ỹ0-9\s]+?)\s+(\d+(?:[.,]\d+)?)\s*(m3|m2|m|tấn|cọc|bộ)", corr_rule, re.IGNORECASE, ) if supp_match: supp_name = supp_match.group(1).strip() supp_qty = float(supp_match.group(2).replace(",", ".")) supp_unit = supp_match.group(3).strip() exists = any( self._normalize_search_text(supp_name) in self._normalize_search_text(it.get("item_name", "")) for it in updated ) if not exists: cat = ( "formwork" if "ván khuôn" in supp_name.lower() else "concrete" if "bê tông" in supp_name.lower() else "earthwork" ) u_price = 215000.0 if cat == "formwork" else 1240000.0 new_item = { "item_order": len(updated) + 1, "wbs_code": f"1.{len(updated) + 1}", "norm_code": "AF.81110" if cat == "formwork" else "AF.11110", "item_name": f"Sản xuất lắp dựng {supp_name}", "category": cat, "dimension_formula": "Bổ sung theo quy tắc kinh nghiệm Kỹ sư QS", "unit": supp_unit, "quantity": supp_qty, "unit_price_vnd": u_price, "total_amount_vnd": round(supp_qty * u_price), "price_source_url": "/dashboard/material-price-comparison?q=Ván+khuôn", } updated.append(new_item) applied_rule_ids.add(r_id) # Record rule application counters in DB for rid in applied_rule_ids: try: client.record_learned_rule_application(rid) except Exception as rec_err: logger.debug( "[TAKEOFF] Failed updating rule application count: %s", rec_err ) return updated def apply_learned_rules_and_state_prices( self, items: list[dict[str, Any]], drawing_type: str = "general", takeoff_id: str | None = None, ) -> list[dict[str, Any]]: """Hệ thống chuẩn hóa kép: Áp đơn giá công bố Sở Xây Dựng 2026 -> Thực thi Quy tắc Rút kinh nghiệm QS -> Kiểm tra độ chuẩn xác toán học.""" step1 = self.match_and_apply_state_prices(items) step2 = self.enforce_learned_rules( step1, drawing_type=drawing_type, takeoff_id=takeoff_id ) sanitized = [] for idx, itm in enumerate(step2): c_itm = dict(itm) c_itm["item_order"] = idx + 1 qty = float(c_itm.get("quantity") or 0) u_price = float(c_itm.get("unit_price_vnd") or 0) c_itm["quantity"] = qty c_itm["unit_price_vnd"] = u_price c_itm["total_amount_vnd"] = round(qty * u_price) sanitized.append(c_itm) return sanitized def reapply_rules_and_state_prices_to_takeoff( self, takeoff_id: str ) -> dict[str, Any]: """Tái áp dụng toàn bộ Quy tắc Rút kinh nghiệm và Đơn giá Sở Xây Dựng 2026 vào hồ sơ bóc tách.""" from uuid import uuid4 client = getattr(self, "_erp_client", None) if not client: raise ValueError("ErpDatabaseClient is not available") takeoff = client.get_drawing_takeoff(takeoff_id) if not takeoff: raise ValueError(f"Không tìm thấy hồ sơ bóc tách ID: {takeoff_id}") items = client.list_drawing_takeoff_items(takeoff_id) or [] if not items: return { "status": "success", "message": "Không có hạng mục nào để cập nhật", "items_count": 0, } drawing_type = takeoff.get("drawing_type") or "general" updated_items = self.apply_learned_rules_and_state_prices( items=items, drawing_type=drawing_type, takeoff_id=takeoff_id, ) # Wipe and recreate items client.delete_drawing_takeoff_items(takeoff_id) for itm in updated_items: itm_payload = { "id": str(uuid4()), "takeoff_id": takeoff_id, "item_order": itm.get("item_order", 1), "wbs_code": itm.get("wbs_code", ""), "norm_code": itm.get("norm_code", ""), "item_name": itm.get("item_name", ""), "category": itm.get("category", "concrete"), "dimension_formula": itm.get("dimension_formula", ""), "unit": itm.get("unit", ""), "quantity": float(itm.get("quantity", 0)), "unit_price_vnd": float(itm.get("unit_price_vnd", 0)), "total_amount_vnd": float(itm.get("total_amount_vnd", 0)), "price_source_url": itm.get("price_source_url", ""), } client.create_drawing_takeoff_item(itm_payload) # Recalculate KPIs tot_cost = sum(float(itm.get("total_amount_vnd", 0)) for itm in updated_items) tot_concrete = sum( float(itm.get("quantity", 0)) for itm in updated_items if itm.get("category") == "concrete" ) tot_rebar = sum( float(itm.get("quantity", 0)) for itm in updated_items if itm.get("category") == "rebar" ) tot_formwork = sum( float(itm.get("quantity", 0)) for itm in updated_items if itm.get("category") == "formwork" ) tot_earthwork = sum( float(itm.get("quantity", 0)) for itm in updated_items if itm.get("category") == "earthwork" ) client.update_drawing_takeoff( takeoff_id, { "total_estimated_cost_vnd": tot_cost, "total_concrete_volume_m3": tot_concrete, "total_rebar_weight_tons": tot_rebar, "total_formwork_area_m2": tot_formwork, "total_earthwork_volume_m3": tot_earthwork, }, ) return { "status": "success", "message": f"Đã áp dụng thành công Quy tắc Rút kinh nghiệm & Đơn giá SXD 2026 cho {len(updated_items)} hạng mục.", "takeoff_id": takeoff_id, "items_count": len(updated_items), "total_estimated_cost_vnd": tot_cost, }