from __future__ import annotations import logging import openpyxl from decimal import Decimal from typing import Any, Dict, List, Optional from datetime import datetime from app.modules.projects.domain.ports.project_repository_port import ProjectDataIntegrityPort logger = logging.getLogger("DataIntegrityService") class DataIntegrityService: """Dịch vụ kiểm toán tính toàn vẹn và phát hiện dị thường dữ liệu dự án ERP Định Sơn.""" EXCEL_PATH = "C:/Projects/DSCons/CongTrinh/Bảng theo dõi công trình.xlsx" def __init__(self, repo: ProjectDataIntegrityPort | None = None): if repo is None: from app.core.postgres.erp_client import ErpDatabaseClient repo = ErpDatabaseClient() self.db = repo def audit_all_projects_integrity(self) -> Dict[str, Any]: """Quét và phân tích toàn diện 142 dự án trong CSDL: - Kiểm tra độ khớp giữa Giá trị HĐ và Cây WBS - Kiểm tra tính hợp lệ của Đợt thanh toán (IPC) - Kiểm tra tính duy nhất và không bị duplicate - Đối soát sai lệch với Bảng theo dõi công trình Excel """ anomalies: List[Dict[str, Any]] = [] with self.db.get_connection() as conn: with conn.cursor() as cur: # 1. Lấy toàn bộ dự án cur.execute(""" SELECT id, project_code, project_name, client_name, contract_number, contract_value, budget_amount, status FROM projects ORDER BY project_code; """) projects = cur.fetchall() # 2. Lấy thống kê WBS theo từng dự án cur.execute(""" SELECT project_id, count(id) as total_wbs, count(DISTINCT wbs_code) as distinct_codes, sum(total_amount_vnd) FILTER (WHERE task_type = 'task') as task_sum, sum(total_amount_vnd) FILTER (WHERE task_type = 'phase') as phase_sum FROM erp_project_wbs GROUP BY project_id; """) wbs_stats = {str(r["project_id"]): r for r in cur.fetchall()} # 3. Lấy thống kê IPC theo từng dự án cur.execute(""" SELECT project_id, count(id) as total_ipcs, count(DISTINCT ipc_number) as distinct_ipcs, sum(net_certified_amount_vnd) as certified_sum, sum(gross_claimed_amount_vnd) as claimed_sum FROM erp_project_ipcs GROUP BY project_id; """) ipc_stats = {str(r["project_id"]): r for r in cur.fetchall()} # 4. Kiểm tra dị thường nội bộ DB for p in projects: pid = str(p["id"]) pcode = p["project_code"] pname = p["project_name"] c_val = Decimal(str(p["contract_value"])) if p["contract_value"] is not None else Decimal("0") # Check WBS if pid in wbs_stats: w_stat = wbs_stats[pid] task_sum = Decimal(str(w_stat["task_sum"] or 0)) # Nếu có WBS task và lệch lớn hơn 100.000 VNĐ so với HĐ if task_sum > 0 and abs(task_sum - c_val) > 100000: anomalies.append({ "type": "WBS_VALUE_MISMATCH", "severity": "high", "project_code": pcode, "project_name": pname, "message": f"Tổng WBS ({task_sum:,.0f} đ) lệch so với Giá trị HĐ ({c_val:,.0f} đ)", "expected": float(c_val), "actual": float(task_sum) }) if w_stat["total_wbs"] > 0 and w_stat["total_wbs"] > w_stat["distinct_codes"] * 2: anomalies.append({ "type": "WBS_DUPLICATE_SUSPECT", "severity": "medium", "project_code": pcode, "project_name": pname, "message": f"Cảnh báo trùng mã WBS ({w_stat['total_wbs']} dòng nhưng chỉ có {w_stat['distinct_codes']} mã)", }) # Check IPC if pid in ipc_stats: i_stat = ipc_stats[pid] if i_stat["total_ipcs"] > i_stat["distinct_ipcs"]: anomalies.append({ "type": "IPC_DUPLICATE_NUMBER", "severity": "critical", "project_code": pcode, "project_name": pname, "message": f"Phát hiện đợt thanh toán IPC bị trùng lặp số đợt ({i_stat['total_ipcs']} bản ghi nhưng chỉ có {i_stat['distinct_ipcs']} đợt)", }) cert_sum = Decimal(str(i_stat["certified_sum"] or 0)) if c_val > 0 and cert_sum > c_val: anomalies.append({ "type": "IPC_OVERPAYMENT", "severity": "critical", "project_code": pcode, "project_name": pname, "message": f"Tổng nghiệm thu thanh toán ({cert_sum:,.0f} đ) vượt quá giá trị Hợp đồng ({c_val:,.0f} đ)", "expected": float(c_val), "actual": float(cert_sum) }) # 5. Đối soát sai lệch với File Excel excel_drift = self._compare_with_excel(projects) for drift in excel_drift: anomalies.append(drift) total_p = len(projects) healthy_p = total_p - len({a["project_code"] for a in anomalies if "project_code" in a}) integrity_score = round((healthy_p / total_p) * 100, 1) if total_p > 0 else 100.0 return { "status": "healthy" if len(anomalies) == 0 else "warning", "total_projects": total_p, "healthy_projects": healthy_p, "integrity_score_percent": integrity_score, "anomaly_count": len(anomalies), "anomalies": anomalies, "excel_path": self.EXCEL_PATH, "audited_at": datetime.now().isoformat() } def _compare_with_excel(self, db_projects: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """So khớp dữ liệu CSDL với Bảng theo dõi công trình.xlsx để phát hiện row-shift.""" drifts = [] try: wb = openpyxl.load_workbook(self.EXCEL_PATH, data_only=True) except Exception as e: logger.warning(f"Không thể mở file Excel để đối soát: {e}") return drifts excel_projects_by_code: Dict[str, Dict[str, Any]] = {} for sname in ["2026", "2025", "2024", "2023", "2022"]: if sname not in wb.sheetnames: continue ws = wb[sname] gt_col = 14 if sname == "2024" else 7 for r in range(5, ws.max_row + 1): stt = ws.cell(r, 1).value name = ws.cell(r, 2).value if not stt and not name: continue try: stt_int = int(stt) if stt else 0 c_val = ws.cell(r, gt_col).value val_float = float(c_val) if isinstance(c_val, (int, float)) else None code_candidate = f"DA-{sname}-{stt_int:02d}" excel_projects_by_code[code_candidate] = { "name": str(name).strip() if name else "", "contract_num": str(ws.cell(r, 5).value or "").strip(), "contract_value": val_float, "row": r, "sheet": sname } except Exception: continue for p in db_projects: code = p["project_code"] if code in excel_projects_by_code: x_item = excel_projects_by_code[code] if x_item["contract_value"] is not None and p["contract_value"] is not None: db_v = float(p["contract_value"]) ex_v = float(x_item["contract_value"]) diff = abs(db_v - ex_v) # Nếu chênh lệch quá 1000 đồng -> Báo động lệch số liệu / lệch dòng if diff > 1000: drifts.append({ "type": "EXCEL_DB_ROW_SHIFT_DRIFT", "severity": "critical", "project_code": code, "project_name": p["project_name"], "message": f"Lệch số liệu giữa CSDL ({db_v:,.0f} đ) và Excel Sheet {x_item['sheet']} Dòng {x_item['row']} ({ex_v:,.0f} đ)", "db_value": db_v, "excel_value": ex_v, "difference": diff }) return drifts