fix proportion for acquisition cost calculation

src/equipment/service.py

@@ -8,7 +8,7 @@ from src.acquisition_cost.model import AcquisitionData
 from src.yeardata.model import Yeardata
 from ..equipment_master.model import EquipmentMaster
 from .schema import EquipmentCreate, EquipmentUpdate, MasterBase
-from typing import Optional
+from typing import Optional, TypedDict
 
 from src.database.core import DbSession, CollectorDbSession
 from src.auth.service import CurrentUser
@@ -23,6 +23,129 @@ import math
 from sqlalchemy import text
 
 
+class CategoryRule(TypedDict, total=False):
+    category_no: str
+    proportion: float
+
+
+# Rules extracted from the provided proportion/category mapping sheet.
+# Each entry tells the service to reuse the acquisition data (and optionally
+# override the proportion) of the referenced category. Update the mapping below
+# whenever business rules change — the logic elsewhere will pick it up
+# automatically.
+CATEGORY_PROPORTION_RULES: dict[str, CategoryRule] = {
+    "1.2": {"category_no": "1.1"},
+    "1.5": {"category_no": "1.1"},
+    "2.7": {"category_no": "2.1"},
+    "3.2": {"category_no": "3.1"},
+    "3.3": {"category_no": "3.1"},
+    "3.4": {"category_no": "3.1"},
+    "3.5": {"category_no": "3.1"},
+    "3.6": {"category_no": "3.1"},
+    "3.7": {"category_no": "3.1"},
+    "3.8": {"category_no": "3.1"},
+    "3.9": {"category_no": "3.1"},
+    "3.10": {"category_no": "3.1"},
+    "4.2": {"category_no": "4.1"},
+    "4.4": {"category_no": "4.1"},
+    "4.6": {"category_no": "4.1"},
+    "5.2": {"category_no": "5.1"},
+    "5.4": {"category_no": "5.3"},
+    "6.2": {"category_no": "6.1"},
+    "6.3": {"category_no": "6.1"},
+    "6.5": {"category_no": "6.4"},
+    "8.2": {"category_no": "8.1"},
+    "8.4": {"category_no": "8.8"},
+    "8.11": {"category_no": "8.8"},
+    "8.12": {"category_no": "8.8"},
+    "8.14": {"category_no": "8.8"},
+    "8.22": {"category_no": "8.21"},
+    "8.23": {"category_no": "8.21"},
+    "8.32": {"category_no": "8.31"},
+    "8.33": {"category_no": "8.8"},
+    "8.34": {"category_no": "8.8"},
+    "8.35": {"category_no": "8.31"},
+    "8.36": {"category_no": "8.8"},
+    "8.37": {"category_no": "8.39"},
+    "8.38": {"category_no": "8.39"},
+    "8.40": {"category_no": "8.39"},
+    "8.41": {"category_no": "8.39"},
+    "8.42": {"category_no": "8.39"},
+    "9.5": {"category_no": "9.4"},
+}
+
+
+def apply_category_proportion_rules(
+    *, category_no: Optional[str], proportion: Optional[float]
+) -> tuple[Optional[str], Optional[float]]:
+    """Normalize category/proportion based on pre-defined business rules."""
+
+    if not category_no:
+        return category_no, proportion
+
+    normalized_category = category_no
+    normalized_proportion = proportion
+    visited: set[str] = set()
+
+    # Follow chained mappings (e.g. 8.35 -> 8.31 -> ...).
+    while normalized_category in CATEGORY_PROPORTION_RULES and normalized_category not in visited:
+        visited.add(normalized_category)
+        rule = CATEGORY_PROPORTION_RULES[normalized_category]
+
+        target_category = rule.get("category_no")
+        if isinstance(target_category, str):
+            normalized_category = target_category
+
+        target_proportion = rule.get("proportion")
+        if isinstance(target_proportion, (int, float)):
+            normalized_proportion = float(target_proportion)
+
+    return normalized_category, normalized_proportion
+
+
+def _build_category_rollup_children() -> dict[str, set[str]]:
+    rollups: dict[str, set[str]] = {}
+    for alias in CATEGORY_PROPORTION_RULES:
+        resolved, _ = apply_category_proportion_rules(category_no=alias, proportion=None)
+        if resolved and resolved != alias:
+            rollups.setdefault(resolved, set()).add(alias)
+    return rollups
+
+
+CATEGORY_ROLLUP_CHILDREN = _build_category_rollup_children()
+
+
+async def fetch_acquisition_cost_with_rollup(
+    *, db_session: DbSession, base_category_no: str
+) -> tuple[Optional[AcquisitionData], Optional[float]]:
+    """Return base acquisition data and aggregated cost_unit_3 for related categories."""
+
+    related_categories = {base_category_no}
+    related_categories.update(CATEGORY_ROLLUP_CHILDREN.get(base_category_no, set()))
+
+    if not related_categories:
+        return None, None
+
+    acquisition_data_query = Select(AcquisitionData).filter(
+        AcquisitionData.category_no.in_(tuple(related_categories))
+    )
+    acquisition_data_result = await db_session.execute(acquisition_data_query)
+    acquisition_records = acquisition_data_result.scalars().all()
+
+    base_record: Optional[AcquisitionData] = None
+    total_cost_unit_3 = 0.0
+    has_cost_unit = False
+
+    for record in acquisition_records:
+        if record.category_no == base_category_no:
+            base_record = record
+        if record.cost_unit_3 is not None:
+            has_cost_unit = True
+            total_cost_unit_3 += record.cost_unit_3
+
+    return base_record, total_cost_unit_3 if has_cost_unit else None
+
+
 async def get_master_by_assetnum(
     *, db_session: DbSession, collector_db_session: CollectorDbSession, assetnum: str
 ) -> tuple[list[EquipmentTransactionRecords], float | None]:
@@ -431,22 +554,55 @@ async def update(
 
         print(f"DEBUG: Detected change - category_no={category_no}, proportion={proportion}")
 
-        acquisition_data_query = Select(AcquisitionData).filter(
+        resolved_category_no, resolved_proportion = apply_category_proportion_rules(
-            AcquisitionData.category_no == category_no
+            category_no=category_no, proportion=proportion
         )
-        acquisition_data_result = await db_session.execute(acquisition_data_query)
-        acquisition_data = acquisition_data_result.scalars().one_or_none()
 
-        print(f"DEBUG: AcquisitionData found: {acquisition_data is not None}")
+        if resolved_category_no != category_no:
-        if acquisition_data:
+            print(
-            print(f"DEBUG: cost_unit_3={acquisition_data.cost_unit_3}")
+                "DEBUG: category alias rule applied - "
+                f"{category_no} -> {resolved_category_no}"
+            )
+
+        effective_category_no = resolved_category_no or category_no
+
+        if resolved_proportion is not None and resolved_proportion != proportion:
+            print(
+                "DEBUG: proportion overridden by rule - "
+                f"{proportion} -> {resolved_proportion}"
+            )
+            proportion = resolved_proportion
+            update_data["proportion"] = resolved_proportion
 
-        if acquisition_data and acquisition_data.cost_unit_3:
+        if not effective_category_no:
-            new_acquisition_cost = (proportion * 0.01) * acquisition_data.cost_unit_3
+            print("DEBUG: Missing category_no after applying rules; skip cost update")
-            print(f"DEBUG: Calculated new_acquisition_cost={new_acquisition_cost}")
-            equipment.acquisition_cost = new_acquisition_cost
         else:
-            print(f"DEBUG: No acquisition_data or cost_unit_3 available")
+            acquisition_data, aggregated_cost_unit_3 = await fetch_acquisition_cost_with_rollup(
+                db_session=db_session, base_category_no=effective_category_no
+            )
+
+            print(f"DEBUG: AcquisitionData found: {acquisition_data is not None}")
+            if acquisition_data:
+                print(f"DEBUG: base cost_unit_3={acquisition_data.cost_unit_3}")
+
+            related_categories = {effective_category_no}
+            related_categories.update(
+                CATEGORY_ROLLUP_CHILDREN.get(effective_category_no, set())
+            )
+            print(
+                "DEBUG: Aggregated categories="
+                f"{sorted(related_categories)} cost_unit_3={aggregated_cost_unit_3}"
+            )
+
+            if aggregated_cost_unit_3 is not None and proportion is not None:
+                new_acquisition_cost = (proportion * 0.01) * aggregated_cost_unit_3
+                print(f"DEBUG: Calculated new_acquisition_cost={new_acquisition_cost}")
+                equipment.acquisition_cost = new_acquisition_cost
+                update_data["acquisition_cost"] = new_acquisition_cost
+            elif aggregated_cost_unit_3 is None:
+                print("DEBUG: No cost_unit_3 available across related categories")
+            else:
+                print("DEBUG: Proportion missing; acquisition cost not updated")
 
     for field in data:
         if field in update_data: