from __future__ import annotations """Dịch vụ Phân bổ Vật tư Đầu vào dựa trên Dự toán (BoQ) và Hạch toán xuất thẳng chân công trường.""" import logging from decimal import Decimal from typing import Any from app.core.postgres.erp_client import ErpDatabaseClient logger = logging.getLogger("BoQMaterialAllocationService") class BoQMaterialAllocationService: def __init__(self, db: ErpDatabaseClient | None = None): self.db = db or ErpDatabaseClient() def get_project_boq_materials(self, project_id: str | None = None) -> list[dict[str, Any]]: """Lấy danh sách định mức vật tư dự toán và hạn mức còn lại.""" with self.db.get_connection() as conn: with conn.cursor() as cur: sql = """ SELECT bm.id, bm.project_id, p.project_code, p.project_name, p.project_type, bm.material_code, bm.material_name, bm.material_group, bm.unit, bm.boq_quantity, bm.unit_price_boq_vnd, bm.total_budget_boq_vnd, bm.invoiced_quantity, bm.invoiced_amount_vnd, bm.remaining_quantity, bm.allowed_loss_percent, bm.status FROM erp_project_boq_materials bm JOIN projects p ON bm.project_id = p.id """ params: list[Any] = [] if project_id: sql += " WHERE bm.project_id = %s" params.append(project_id) sql += " ORDER BY p.project_code, bm.material_group, bm.material_name;" cur.execute(sql, tuple(params)) return [dict(r) for r in cur.fetchall()] def determine_candidate_projects( self, item_name: str, material_group: str, active_boq_items: list[dict[str, Any]], ) -> list[dict[str, Any]]: """Xác định danh sách dự án ứng viên dựa trên đặc thù kỹ thuật (Technical Spec / DNA).""" name_lower = item_name.lower() candidates: list[dict[str, Any]] = [] # 1. Nhóm đặc thù Dân dụng (Chỉ ưu tiên dự án Dân dụng như NVH Đại Thắng) civil_keywords = [ "tôn mạ", "tôn lợp", "thép hộp", "gạch xây", "gạch không nung", "gạch tuynel", "gạch lát", "ngói", "thạch cao", "cửa nhôm", "sơn nước", "ống ppr", "thiết bị vệ sinh", ] is_civil_specific = any(k in name_lower for k in civil_keywords) # 2. Nhóm đặc thù Thủy lợi (Chỉ ưu tiên dự án Thủy lợi như Cống Bến Kem, Kè Đa Độ) hydraulic_keywords = [ "cừ larsen", "cọc tre", "cừ tràm", "đá hộc", "rọ đá", "vải địa", "bạt xác rắn", "cống hộp", "chịu mặn", "bền sunfat", "kè mái", "bê tông cống", ] is_hydraulic_specific = any(k in name_lower for k in hydraulic_keywords) for boq in active_boq_items: p_type = str(boq.get("project_type") or "").lower() mat_grp = str(boq.get("material_group") or "").upper() boq_name = str(boq.get("material_name") or "").lower() # Lọc sơ bộ theo nhóm vật tư tương đồng group_match = ( mat_grp == material_group.upper() or any(w in boq_name for w in name_lower.split() if len(w) > 3) ) if not group_match: continue if is_civil_specific: if "dân dụng" in p_type or "nhà văn hóa" in str(boq.get("project_name")).lower(): candidates.append(boq) elif is_hydraulic_specific: if "thủy lợi" in p_type or "hạ tầng" in p_type or "cống" in str(boq.get("project_name")).lower(): candidates.append(boq) else: # Vật tư xây dựng cơ bản dùng chung (Xi măng, cát, đá 1x2, thép kết cấu) candidates.append(boq) return candidates def calculate_item_allocations( self, invoice_item: dict[str, Any], candidate_boq_items: list[dict[str, Any]], ) -> list[dict[str, Any]]: """Tính toán tỷ trọng phân bổ dòng hàng vào các dự án theo hạn mức còn lại (Remaining Quota).""" qty = Decimal(str(invoice_item.get("quantity") or 0)) amt_before_vat = Decimal(str(invoice_item.get("amount_before_vat_vnd") or 0)) vat_amt = Decimal(str(invoice_item.get("vat_amount_vnd") or 0)) tot_amt = Decimal(str(invoice_item.get("total_item_amount_vnd") or (amt_before_vat + vat_amt))) if not candidate_boq_items: return [] # Nếu chỉ có 1 dự án phù hợp duy nhất if len(candidate_boq_items) == 1: boq = candidate_boq_items[0] boq_qty = Decimal(str(boq.get("boq_quantity") or 0)) invoiced_qty = Decimal(str(boq.get("invoiced_quantity") or 0)) loss_pct = Decimal(str(boq.get("allowed_loss_percent") or 3)) max_allowed = boq_qty * (Decimal("1") + loss_pct / Decimal("100")) is_overrun = (invoiced_qty + qty) > max_allowed return [ { "project_id": str(boq["project_id"]), "project_code": str(boq["project_code"]), "boq_material_id": str(boq["id"]), "allocated_quantity": qty, "allocated_amount_before_vat_vnd": amt_before_vat, "allocated_vat_amount_vnd": vat_amt, "allocated_total_amount_vnd": tot_amt, "allocation_method": "boq_auto_direct", "debit_account": f"621_{boq['project_code']}", "credit_account": "331", "confidence_score": Decimal("98.00"), "is_overrun": is_overrun, "overrun_ratio": float((invoiced_qty + qty) / boq_qty) if boq_qty > 0 else 1.0, "notes": f"Xuất thẳng chân CT {boq['project_code']} (Khớp dự toán 100%)", } ] # Trường hợp 1 dòng hóa đơn giao cho nhiều công trình: Phân bổ theo tỷ trọng Remaining_Quota total_remaining = sum( Decimal(str(b.get("remaining_quantity") or 0)) for b in candidate_boq_items ) allocations: list[dict[str, Any]] = [] allocated_qty_acc = Decimal("0.0000") allocated_tot_acc = Decimal("0.0000") for idx, boq in enumerate(candidate_boq_items): rem = Decimal(str(boq.get("remaining_quantity") or 0)) if total_remaining > 0: weight = rem / total_remaining else: weight = Decimal("1") / Decimal(len(candidate_boq_items)) # Dự án cuối cùng nhận số dư còn lại để tránh sai số làm tròn if idx == len(candidate_boq_items) - 1: alloc_qty = qty - allocated_qty_acc alloc_tot = tot_amt - allocated_tot_acc alloc_before = amt_before_vat - (amt_before_vat * allocated_tot_acc / tot_amt if tot_amt > 0 else 0) alloc_vat = vat_amt - (vat_amt * allocated_tot_acc / tot_amt if tot_amt > 0 else 0) else: alloc_qty = (qty * weight).quantize(Decimal("0.0001")) alloc_tot = (tot_amt * weight).quantize(Decimal("0.0001")) alloc_before = (amt_before_vat * weight).quantize(Decimal("0.0001")) alloc_vat = (vat_amt * weight).quantize(Decimal("0.0001")) allocated_qty_acc += alloc_qty allocated_tot_acc += alloc_tot boq_qty = Decimal(str(boq.get("boq_quantity") or 0)) invoiced_qty = Decimal(str(boq.get("invoiced_quantity") or 0)) is_overrun = (invoiced_qty + alloc_qty) > (boq_qty * Decimal("1.05")) allocations.append( { "project_id": str(boq["project_id"]), "project_code": str(boq["project_code"]), "boq_material_id": str(boq["id"]), "allocated_quantity": alloc_qty, "allocated_amount_before_vat_vnd": alloc_before, "allocated_vat_amount_vnd": alloc_vat, "allocated_total_amount_vnd": alloc_tot, "allocation_method": "proportional_split", "debit_account": f"621_{boq['project_code']}", "credit_account": "331", "confidence_score": Decimal("92.00"), "is_overrun": is_overrun, "overrun_ratio": float((invoiced_qty + alloc_qty) / boq_qty) if boq_qty > 0 else 1.0, "notes": f"Phân bổ tỷ trọng BoQ còn thiếu ({weight * 100:.1f}%) cho CT {boq['project_code']}", } ) return allocations def save_allocations( self, invoice_id: str, invoice_item_id: str, allocations: list[dict[str, Any]], ) -> None: """Lưu vết phân bổ vào CSDL và cập nhật lũy kế dự toán.""" if not allocations: return insert_sql = """ INSERT INTO erp_invoice_item_allocations ( invoice_id, invoice_item_id, project_id, boq_material_id, allocated_quantity, allocated_amount_before_vat_vnd, allocated_vat_amount_vnd, allocated_total_amount_vnd, allocation_method, debit_account, credit_account, confidence_score, notes ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ); """ update_boq_sql = """ UPDATE erp_project_boq_materials SET invoiced_quantity = invoiced_quantity + %s, invoiced_amount_vnd = invoiced_amount_vnd + %s, updated_at = CURRENT_TIMESTAMP WHERE id = %s; """ with self.db.get_connection() as conn: with conn.cursor() as cur: for alloc in allocations: cur.execute( insert_sql, ( invoice_id, invoice_item_id, alloc["project_id"], alloc.get("boq_material_id"), alloc["allocated_quantity"], alloc["allocated_amount_before_vat_vnd"], alloc["allocated_vat_amount_vnd"], alloc["allocated_total_amount_vnd"], alloc["allocation_method"], alloc["debit_account"], alloc["credit_account"], alloc["confidence_score"], alloc["notes"], ), ) if alloc.get("boq_material_id"): cur.execute( update_boq_sql, ( alloc["allocated_quantity"], alloc["allocated_total_amount_vnd"], alloc["boq_material_id"], ), ) conn.commit() def batch_allocate_construction_invoices(self) -> dict[str, Any]: """Tự động phân bổ toàn bộ hóa đơn Trụ Cột 1 (Thi công Xây lắp & Thủy lợi) vào dự án.""" logger.info("Khởi động quy trình phân bổ vật tư tự động theo Dự toán BoQ...") # 1. Lấy danh sách định mức BoQ hiện hành active_boq = self.get_project_boq_materials() if not active_boq: logger.warning("Chưa có dữ liệu định mức BoQ trong hệ thống!") return {"status": "error", "message": "No active BoQ materials"} # 2. Xóa các phân bổ cũ để chạy lại chuẩn xác with self.db.get_connection() as conn: with conn.cursor() as cur: cur.execute("TRUNCATE TABLE erp_invoice_item_allocations;") cur.execute("UPDATE erp_project_boq_materials SET invoiced_quantity = 0, invoiced_amount_vnd = 0;") # Lấy danh sách hóa đơn Trụ Cột 1 có dòng hàng cur.execute(""" SELECT i.id, i.invoice_number, i.invoice_series, i.issue_date, i.seller_name, i.notes FROM erp_invoices i WHERE (i.notes LIKE '%Trụ Cột 1%' OR i.notes LIKE '%Trụ cột 1%') ORDER BY i.issue_date ASC; """) invoices = cur.fetchall() cur.execute(""" SELECT it.id, it.invoice_id, it.item_name, it.unit, it.quantity, it.amount_before_vat_vnd, it.vat_amount_vnd, it.total_item_amount_vnd, it.cost_category FROM erp_invoice_items it JOIN erp_invoices i ON it.invoice_id = i.id WHERE (i.notes LIKE '%Trụ Cột 1%' OR i.notes LIKE '%Trụ cột 1%') ORDER BY it.invoice_id, it.item_order; """) all_items = cur.fetchall() conn.commit() items_by_inv: dict[str, list[dict]] = {} for it in all_items: iid = str(it["invoice_id"]) if iid not in items_by_inv: items_by_inv[iid] = [] items_by_inv[iid].append(it) stats = { "total_invoices_scanned": len(invoices), "allocated_items_count": 0, "project_summaries": {}, "overrun_alerts": [], } # Đọc lại BoQ active_boq = self.get_project_boq_materials() with self.db.get_connection() as conn: with conn.cursor() as cur: for inv in invoices: inv_id = str(inv["id"]) items = items_by_inv.get(inv_id, []) if not items: continue inv_matched_projects: set[str] = set() for it in items: name = str(it["item_name"]) cat = str(it.get("cost_category") or "OTHER") candidates = self.determine_candidate_projects(name, cat, active_boq) if not candidates: continue allocations = self.calculate_item_allocations(it, candidates) if allocations: self.save_allocations(inv_id, str(it["id"]), allocations) stats["allocated_items_count"] += 1 for al in allocations: p_code = al["project_code"] inv_matched_projects.add(p_code) if p_code not in stats["project_summaries"]: stats["project_summaries"][p_code] = { "total_amount_vnd": Decimal("0.0000"), "items_count": 0, } stats["project_summaries"][p_code]["total_amount_vnd"] += al["allocated_total_amount_vnd"] stats["project_summaries"][p_code]["items_count"] += 1 if al.get("is_overrun"): stats["overrun_alerts"].append( { "invoice": f"{inv.get('invoice_series')}-{inv.get('invoice_number')}", "project": p_code, "item": name, "overrun_ratio": al.get("overrun_ratio"), } ) # Cập nhật matched_project_id trên hóa đơn if inv_matched_projects: primary_proj_code = list(inv_matched_projects)[0] cur.execute("SELECT id FROM projects WHERE project_code = %s LIMIT 1;", (primary_proj_code,)) p_row = cur.fetchone() if p_row: p_id = str(p_row["id"]) proj_tags = ", ".join(inv_matched_projects) cur.execute( """ UPDATE erp_invoices SET matched_project_id = %s, notes = notes || %s, updated_at = CURRENT_TIMESTAMP WHERE id = %s; """, (p_id, f" [Giao thẳng chân CT: {proj_tags}]", inv_id), ) conn.commit() logger.info(f"Hoàn thành phân bổ BoQ cho {stats['allocated_items_count']} dòng hàng hóa đơn!") return stats def get_allocation_summary(self) -> dict[str, Any]: """Tổng hợp tình hình phân bổ vật tư và cảnh báo vượt định mức BoQ.""" with self.db.get_connection() as conn: with conn.cursor() as cur: cur.execute(""" SELECT p.project_code, p.project_name, COUNT(a.id) as allocation_records_count, SUM(a.allocated_total_amount_vnd) as total_allocated_amount_vnd FROM erp_invoice_item_allocations a JOIN projects p ON a.project_id = p.id GROUP BY p.project_code, p.project_name ORDER BY total_allocated_amount_vnd DESC; """) projects_summary = [dict(r) for r in cur.fetchall()] cur.execute(""" SELECT bm.material_name, bm.material_code, p.project_code, bm.boq_quantity, bm.invoiced_quantity, bm.unit, (bm.invoiced_quantity / NULLIF(bm.boq_quantity, 0)) as burn_rate FROM erp_project_boq_materials bm JOIN projects p ON bm.project_id = p.id WHERE bm.invoiced_quantity > (bm.boq_quantity * 1.05) ORDER BY burn_rate DESC; """) overruns = [dict(r) for r in cur.fetchall()] return { "status": "success", "projects_summary": projects_summary, "overrun_alerts": overruns, }