Files
Aislo/B08_wf5_Quantity/B08_wf5_Quantity_Repository_Calculation.py
T
2026-07-25 19:41:05 +09:00

189 lines
8.3 KiB
Python

import json
from uuid import uuid4
from B08_wf5_Quantity.B08_wf5_Quantity_Engine_Estimate import aggregate_by_wbs
from B08_wf5_Quantity.B08_wf5_Quantity_Schema_Calculation import (
CostTotals,
EstimateLine,
FinalCost,
)
class CalculationRepository:
def __init__(self, connection):
self.db = connection
async def save(
self,
project_id: str,
versions: dict,
input_hash: str,
snapshots: list[dict],
lines: list[EstimateLine],
totals: CostTotals,
final: FinalCost,
user_id: int | None,
) -> str:
run_id = str(uuid4())
async with self.db.cursor() as cursor:
await cursor.execute(
"""INSERT INTO b08_calculation_runs(
id,project_id,calculation_type,basis_version,price_version,
quantity_version,rule_version,input_hash,status,error_json,created_by
) VALUES(%s,%s,'FINAL',%s,%s,%s,%s,%s,'COMPLETE','[]',%s)""",
(run_id, project_id, versions["basis_version"], versions["price_version"],
versions["quantity_version"], versions["rule_version"], input_hash, user_id),
)
for snapshot in snapshots:
await cursor.execute(
"""INSERT INTO b08_calculation_inputs(
run_id,input_type,reference_id,snapshot_json
) VALUES(%s,'QUANTITY_ITEM',%s,%s)""",
(run_id, snapshot["quantity_item_id"],
json.dumps(snapshot, ensure_ascii=False)),
)
for line in lines:
await cursor.execute(
"""INSERT INTO b08_estimate_lines(
run_id,quantity_item_id,wbs_id,quantity,unit_labor,
unit_material,unit_expense,labor_amount,material_amount,
expense_amount,total_amount,trace_json
) VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
(run_id, line.quantity_item_id, line.wbs_id, line.quantity,
line.unit_labor, line.unit_material, line.unit_expense,
line.labor_amount, line.material_amount, line.expense_amount,
line.total_amount, json.dumps(line.trace, ensure_ascii=False)),
)
await self._save_aggregates(cursor, run_id, lines, totals)
for sort_order, result in enumerate(final.indirect_results):
await cursor.execute(
"""INSERT INTO b08_indirect_cost_results(
run_id,rule_code,base_amount,rate_value,result_amount,
trace_json,sort_order
) VALUES(%s,%s,%s,%s,%s,%s,%s)""",
(run_id, result.rule_code, result.base_amount, result.rate_value,
result.result_amount, json.dumps(result.trace, ensure_ascii=False),
sort_order),
)
await cursor.execute(
"""INSERT INTO b08_final_cost_results(
run_id,direct_cost,net_cost,general_admin,profit,total_cost,
vat,contract_cost,government_material,procurement_fee,
total_project_cost,trace_json
) VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
(run_id, final.direct_cost, final.net_cost, final.general_admin,
final.profit, final.total_cost, final.vat, final.contract_cost,
final.government_material, final.procurement_fee,
final.total_project_cost, json.dumps(final.trace, ensure_ascii=False)),
)
await self.db.commit()
return run_id
@staticmethod
async def _save_aggregates(cursor, run_id: str, lines: list[EstimateLine], totals: CostTotals) -> None:
grouped = aggregate_by_wbs(lines)
grouped["__TOTAL__"] = totals
for group_key, total in grouped.items():
aggregate_type = "TOTAL" if group_key == "__TOTAL__" else "WBS"
total_amount = (
total.direct_cost + total.government_material + total.excluded_amount
)
await cursor.execute(
"""INSERT INTO b08_cost_aggregates(
run_id,aggregate_type,group_key,labor_amount,material_amount,
expense_amount,total_amount
) VALUES(%s,%s,%s,%s,%s,%s,%s)""",
(run_id, aggregate_type, group_key, total.labor, total.material,
total.expense, total_amount),
)
async def latest_detail(self, project_id: str) -> dict | None:
async with self.db.cursor() as cursor:
await cursor.execute(
"""SELECT id FROM b08_calculation_runs
WHERE project_id=%s AND status='COMPLETE'
ORDER BY created_at DESC,id DESC LIMIT 1""",
(project_id,),
)
run = await cursor.fetchone()
if not run:
return None
return await self.detail(project_id, run["id"])
async def detail(self, project_id: str, run_id: str) -> dict:
async with self.db.cursor() as cursor:
await cursor.execute(
"""SELECT * FROM b08_calculation_runs
WHERE project_id=%s AND id=%s AND status='COMPLETE'""",
(project_id, run_id),
)
run = await cursor.fetchone()
if not run:
raise LookupError("계산 실행을 찾을 수 없습니다.")
await cursor.execute(
"SELECT * FROM b08_estimate_lines WHERE run_id=%s ORDER BY id",
(run_id,),
)
lines = list(await cursor.fetchall())
await cursor.execute(
"""SELECT * FROM b08_cost_aggregates
WHERE run_id=%s AND aggregate_type='TOTAL'""",
(run_id,),
)
aggregate = await cursor.fetchone() or {}
await cursor.execute(
"""SELECT * FROM b08_indirect_cost_results
WHERE run_id=%s ORDER BY sort_order""",
(run_id,),
)
indirect = list(await cursor.fetchall())
await cursor.execute(
"SELECT * FROM b08_final_cost_results WHERE run_id=%s",
(run_id,),
)
final = await cursor.fetchone()
if not final:
raise LookupError("최종공사비 결과를 찾을 수 없습니다.")
for line in lines:
line["trace"] = json.loads(line.pop("trace_json"))
for result in indirect:
result["trace"] = json.loads(result.pop("trace_json"))
result["rule_name"] = result["trace"].get(
"rule_name", result["rule_code"]
)
final_trace = json.loads(final.pop("trace_json"))
final["trace"] = final_trace
final["indirect_results"] = indirect
totals = {
"labor": aggregate.get("labor_amount", 0),
"material": aggregate.get("material_amount", 0),
"expense": aggregate.get("expense_amount", 0),
"direct_cost": final["direct_cost"],
"government_material": final["government_material"],
"excluded_amount": final_trace.get("excluded_amount", 0),
}
versions = {
"basis_version": run["basis_version"],
"price_version": run["price_version"],
"quantity_version": run["quantity_version"],
"rule_version": run["rule_version"],
}
return {
"run_id": run_id,
"created_at": run["created_at"],
"versions": versions,
"lines": lines,
"totals": totals,
"final": final,
}
async def history(self, project_id: str) -> list[dict]:
async with self.db.cursor() as cursor:
await cursor.execute(
"""SELECT r.*, f.total_project_cost
FROM b08_calculation_runs r
LEFT JOIN b08_final_cost_results f ON f.run_id=r.id
WHERE r.project_id=%s ORDER BY r.created_at DESC""",
(project_id,),
)
return list(await cursor.fetchall())