Aislo/B08_wf5_Quantity/B08_wf5_Quantity_Repository_CalculationSource.py

import hashlib
import json
from decimal import Decimal

from B08_wf5_Quantity.B08_wf5_Quantity_Schema_Basis import RatePolicy
from B08_wf5_Quantity.B08_wf5_Quantity_Schema_Calculation import (
    CalculationVersions,
    EstimateInput,
    UnitPriceBreakdown,
)


class CalculationSourceRepository:
    def __init__(self, connection):
        self.db = connection

    async def load(
        self, project_id: str, versions: CalculationVersions
    ) -> tuple[list[EstimateInput], list[RatePolicy], list[dict]]:
        async with self.db.cursor() as cursor:
            await self._validate_versions(cursor, project_id, versions)
            await cursor.execute(self._input_query(), (versions.price_version, project_id))
            rows = list(await cursor.fetchall())
            await cursor.execute(
                """SELECT * FROM b08_rate_policies
                   WHERE project_id=%s AND basis_version=%s AND status='APPROVED'
                   ORDER BY sort_order""",
                (project_id, versions.rule_version),
            )
            policy_rows = list(await cursor.fetchall())
        if not rows:
            raise ValueError("확정된 설계수량 항목이 없습니다.")
        missing = [row for row in rows if not row["reference_found"]]
        if missing:
            raise ValueError(f"확정 적용단가를 찾을 수 없는 항목 {len(missing)}건이 있습니다.")
        await self._validate_quantity_hash(project_id, versions.quantity_version, rows)
        inputs = [self._to_input(row) for row in rows]
        policies = [self._to_policy(row) for row in policy_rows]
        snapshots = [self._snapshot(row) for row in rows]
        return inputs, policies, snapshots

    async def _validate_versions(self, cursor, project_id: str, versions: CalculationVersions) -> None:
        await cursor.execute(
            """SELECT 1 FROM b08_basis_versions
               WHERE project_id=%s AND version=%s AND status='CONFIRMED'""",
            (project_id, versions.basis_version),
        )
        if not await cursor.fetchone():
            raise ValueError("확정된 기준정보 버전이 아닙니다.")
        if versions.rule_version != versions.basis_version:
            raise ValueError("요율 규칙 버전은 기준정보 버전과 같아야 합니다.")
        await cursor.execute(
            """SELECT 1 FROM b08_price_books
               WHERE project_id=%s AND price_version=%s AND basis_version=%s
                 AND status='CONFIRMED'""",
            (project_id, versions.price_version, versions.basis_version),
        )
        if not await cursor.fetchone():
            raise ValueError("확정 가격판과 기준정보 버전이 일치하지 않습니다.")
        await cursor.execute(
            """SELECT quantity_version FROM b08_quantity_confirmations
               WHERE project_id=%s ORDER BY quantity_version DESC LIMIT 1""",
            (project_id,),
        )
        latest = await cursor.fetchone()
        if not latest or latest["quantity_version"] != versions.quantity_version:
            raise ValueError("최신 확정 수량 버전을 사용해야 합니다.")

    async def _validate_quantity_hash(
        self, project_id: str, quantity_version: int, rows: list[dict]
    ) -> None:
        snapshot = [
            {"id": row["id"], "quantity": row["confirmed_quantity"]}
            for row in sorted(rows, key=lambda item: item["id"])
        ]
        digest = hashlib.sha256(
            json.dumps(snapshot, default=str, sort_keys=True).encode()
        ).hexdigest()
        async with self.db.cursor() as cursor:
            await cursor.execute(
                """SELECT input_hash FROM b08_quantity_confirmations
                   WHERE project_id=%s AND quantity_version=%s""",
                (project_id, quantity_version),
            )
            confirmation = await cursor.fetchone()
        if not confirmation or confirmation["input_hash"] != digest:
            raise ValueError("확정 후 설계수량이 변경되어 다시 확정해야 합니다.")

    @staticmethod
    def _input_query() -> str:
        return """
            SELECT q.*,
              CASE q.reference_type
                WHEN 'CATALOG' THEN IF(ci.cost_type='LABOR',ap.applied_price,0)
                ELSE COALESCE(uc.labor_price,eq.labor_price,cb.labor_price,0)
              END AS unit_labor,
              CASE q.reference_type
                WHEN 'CATALOG' THEN IF(ci.cost_type='MATERIAL',ap.applied_price,0)
                ELSE COALESCE(uc.material_price,eq.material_price,cb.material_price,0)
              END AS unit_material,
              CASE q.reference_type
                WHEN 'CATALOG' THEN IF(ci.cost_type='EXPENSE',ap.applied_price,0)
                ELSE COALESCE(uc.expense_price,eq.expense_price,cb.expense_price,0)
              END AS unit_expense,
              CASE q.reference_type
                WHEN 'CATALOG' THEN ap.item_id IS NOT NULL
                WHEN 'UNIT_COST' THEN uc.id IS NOT NULL
                WHEN 'EQUIPMENT' THEN eq.id IS NOT NULL
                WHEN 'COST_BASIS' THEN cb.id IS NOT NULL
                ELSE 0
              END AS reference_found
            FROM b08_design_quantity_items q
            LEFT JOIN b08_catalog_items ci
              ON q.reference_type='CATALOG' AND ci.id=q.reference_id
             AND ci.project_id=q.project_id
            LEFT JOIN b08_applied_prices ap
              ON ap.project_id=q.project_id AND ap.item_id=ci.id AND ap.price_version=%s
            LEFT JOIN b08_unit_costs uc
              ON q.reference_type='UNIT_COST' AND uc.id=q.reference_id
             AND uc.project_id=q.project_id AND uc.status='CONFIRMED'
            LEFT JOIN b08_equipment_rates eq
              ON q.reference_type='EQUIPMENT' AND eq.id=q.reference_id
             AND eq.project_id=q.project_id AND eq.status='CONFIRMED'
            LEFT JOIN b08_cost_basis cb
              ON q.reference_type='COST_BASIS' AND cb.id=q.reference_id
             AND cb.project_id=q.project_id AND cb.status='CONFIRMED'
            WHERE q.project_id=%s AND q.status='CONFIRMED' AND q.excluded=0
            ORDER BY q.id
        """

    @staticmethod
    def _to_input(row: dict) -> EstimateInput:
        return EstimateInput(
            quantity_item_id=row["id"],
            wbs_id=row["wbs_id"],
            quantity=Decimal(str(row["confirmed_quantity"])),
            unit_price=UnitPriceBreakdown(
                labor=Decimal(str(row["unit_labor"] or 0)),
                material=Decimal(str(row["unit_material"] or 0)),
                expense=Decimal(str(row["unit_expense"] or 0)),
            ),
            procurement_type=row["procurement_type"],
        )

    @staticmethod
    def _to_policy(row: dict) -> RatePolicy:
        values = dict(row)
        condition = values.get("condition_json")
        values["condition_json"] = json.loads(condition) if isinstance(condition, str) else (condition or {})
        return RatePolicy.model_validate(values)

    @staticmethod
    def _snapshot(row: dict) -> dict:
        return {
            "quantity_item_id": row["id"],
            "wbs_id": row["wbs_id"],
            "reference_type": row["reference_type"],
            "reference_id": row["reference_id"],
            "quantity": str(row["confirmed_quantity"]),
            "unit_labor": str(row["unit_labor"] or 0),
            "unit_material": str(row["unit_material"] or 0),
            "unit_expense": str(row["unit_expense"] or 0),
            "procurement_type": row["procurement_type"],
        }