diff --git a/src/equipment/service.py b/src/equipment/service.py index 45974ad..98447e6 100644 --- a/src/equipment/service.py +++ b/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]: @@ -428,25 +551,58 @@ async def update( if "proportion" in update_data or "category_no" in update_data: category_no = update_data.get("category_no", equipment.category_no) proportion = update_data.get("proportion", equipment.proportion) - + print(f"DEBUG: Detected change - category_no={category_no}, proportion={proportion}") - - acquisition_data_query = Select(AcquisitionData).filter( - AcquisitionData.category_no == category_no + + resolved_category_no, resolved_proportion = apply_category_proportion_rules( + 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 acquisition_data: - print(f"DEBUG: cost_unit_3={acquisition_data.cost_unit_3}") - - if acquisition_data and acquisition_data.cost_unit_3: - new_acquisition_cost = (proportion * 0.01) * acquisition_data.cost_unit_3 - print(f"DEBUG: Calculated new_acquisition_cost={new_acquisition_cost}") - equipment.acquisition_cost = new_acquisition_cost + + if resolved_category_no != category_no: + print( + "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 not effective_category_no: + print("DEBUG: Missing category_no after applying rules; skip cost update") 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: