from __future__ import annotations import logging import re from decimal import Decimal from pathlib import Path from typing import Any import xlrd from .models import GroundTruthRow logger = logging.getLogger("dscons.takeoff_training.gt_loader") class GroundTruthLoader: """Đọc và trích xuất dữ liệu dự toán thực tế chuẩn Kỹ sư từ file Excel (.xls / .xlsx).""" @classmethod def load_ground_truth_from_excel( cls, file_path: str | Path ) -> list[GroundTruthRow]: """Trích xuất toàn bộ các dòng đầu việc và diễn giải công thức đối chứng.""" p = Path(file_path) if not p.exists(): logger.warning("[GT_LOADER] File not found: %s", p) return [] if p.suffix.lower() == ".xls": return cls._parse_xls(p) elif p.suffix.lower() == ".xlsx": return cls._parse_xlsx(p) return [] @classmethod def _parse_xls(cls, path: Path) -> list[GroundTruthRow]: items: list[GroundTruthRow] = [] try: wb = xlrd.open_workbook(str(path), on_demand=True) sheet_names = wb.sheet_names() target_sheet_name = None for candidate in [ "Công trình", "Tiên lượng", "Chi tiết KL", "Dự toán", "Sheet1", ]: for name in sheet_names: if candidate.lower() in name.lower(): target_sheet_name = name break if target_sheet_name: break sh = ( wb.sheet_by_name(target_sheet_name) if target_sheet_name else wb.sheet_by_index(0) ) rows_data = [] for r in range(sh.nrows): row_vals = [sh.cell_value(r, c) for c in range(min(sh.ncols, 35))] rows_data.append(row_vals) items = cls._extract_from_matrix(rows_data) except Exception as exc: logger.error("[GT_LOADER] Error reading xls: %s", exc) return items @classmethod def _parse_xlsx(cls, path: Path) -> list[GroundTruthRow]: import openpyxl items: list[GroundTruthRow] = [] try: wb = openpyxl.load_workbook(str(path), data_only=True) sh = wb.active rows_data = [] for r in range(1, sh.max_row + 1): row_vals = [ sh.cell(row=r, column=c).value for c in range(1, min(sh.max_column + 1, 35)) ] rows_data.append(row_vals) items = cls._extract_from_matrix(rows_data) except Exception as exc: logger.error("[GT_LOADER] Error reading xlsx: %s", exc) return items @classmethod def _extract_from_matrix(cls, matrix: list[list[Any]]) -> list[GroundTruthRow]: items: list[GroundTruthRow] = [] header_idx = -1 col_stt, col_norm, col_name, col_unit, col_qty = 0, 2, 5, 6, -1 for idx, row in enumerate(matrix[:20]): txt_row = " ".join([str(c or "").lower() for c in row]) if ("stt" in txt_row or "mã hiệu" in txt_row) and ( "đơn vị" in txt_row or "công tác" in txt_row or "khối lượng" in txt_row ): header_idx = idx for c_idx, cell in enumerate(row): cell_str = str(cell or "").lower().replace("\n", " ") if "stt" in cell_str and col_stt == 0: col_stt = c_idx elif ( "mã hiệu" in cell_str or "mã đm" in cell_str ) and "công tác" not in cell_str: col_norm = c_idx elif ( "danh mục" in cell_str or "công tác" in cell_str or "nội dung" in cell_str or "diễn giải" in cell_str ): col_name = c_idx elif "đơn vị" in cell_str or "đvt" in cell_str: col_unit = c_idx elif "khối lượng" in cell_str or cell_str == "kl": col_qty = c_idx break current_item: GroundTruthRow | None = None current_formulas: list[str] = [] start_row = header_idx + 1 if header_idx >= 0 else 0 for r in range(start_row, len(matrix)): row = matrix[r] if not row or all(c is None or str(c).strip() == "" for c in row): continue stt_val = str( row[col_stt] if col_stt >= 0 and col_stt < len(row) and row[col_stt] is not None else "" ).strip() name_val = str( row[col_name] if col_name >= 0 and col_name < len(row) and row[col_name] is not None else "" ).strip() norm_val = str( row[col_norm] if col_norm >= 0 and col_norm < len(row) and row[col_norm] is not None else "" ).strip() unit_val = str( row[col_unit] if col_unit >= 0 and col_unit < len(row) and row[col_unit] is not None else "" ).strip() is_main_item = False qty_val = Decimal("0.0") if ( stt_val and ( stt_val.replace(".0", "").isdigit() or re.match(r"^[0-9]+(\.[0-9]+)*$", stt_val) ) or norm_val and re.match(r"^[A-Z]{2}\.[0-9]+", norm_val) ): is_main_item = True if col_qty >= 0 and col_qty < len(row): raw_q = row[col_qty] if isinstance(raw_q, (int, float)) and raw_q != 0: qty_val = Decimal(str(round(float(raw_q), 4))) elif ( isinstance(raw_q, str) and raw_q.strip().replace(".", "", 1).replace("-", "", 1).isdigit() ): try: qty_val = Decimal(raw_q.strip()) except Exception: pass if is_main_item and name_val: if current_item: current_item.formula_breakdown = ( "\n".join(current_formulas) if current_formulas else None ) if current_item.quantity == Decimal("0.0") and current_formulas: current_item.quantity = cls._sum_formulas(current_formulas) items.append(current_item) current_formulas = [] current_item = GroundTruthRow( stt=stt_val.replace(".0", ""), norm_code=norm_val if norm_val and not norm_val.startswith("HM") else None, task_name=name_val, unit=unit_val if unit_val else "m3", quantity=qty_val, ) elif current_item and name_val: if "=" in name_val or any( op in name_val for op in ["*", "+", "/", "-"] ): current_formulas.append(name_val) if current_item: current_item.formula_breakdown = ( "\n".join(current_formulas) if current_formulas else None ) if current_item.quantity == Decimal("0.0") and current_formulas: current_item.quantity = cls._sum_formulas(current_formulas) items.append(current_item) return items @staticmethod def _sum_formulas(formulas: list[str]) -> Decimal: total = Decimal("0.0") for f in formulas: if "=" in f: parts = f.split("=") try: val = parts[-1].strip().replace(",", ".") total += Decimal(str(round(float(val), 4))) except Exception: pass return total