249 lines
10 KiB
Python
249 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
SMETA_VALUE_PATHS: list[tuple[str, tuple[str, ...]]] = [
|
|
("section_code", ("section_code",)),
|
|
("name", ("name",)),
|
|
("plan_support_q1", ("plan", "support", "q1")),
|
|
("plan_support_q2", ("plan", "support", "q2")),
|
|
("plan_support_q3", ("plan", "support", "q3")),
|
|
("plan_support_q4", ("plan", "support", "q4")),
|
|
("plan_support_year", ("plan", "support", "year")),
|
|
("plan_development_q1", ("plan", "development", "q1")),
|
|
("plan_development_q2", ("plan", "development", "q2")),
|
|
("plan_development_q3", ("plan", "development", "q3")),
|
|
("plan_development_q4", ("plan", "development", "q4")),
|
|
("plan_development_year", ("plan", "development", "year")),
|
|
("plan_total_year", ("plan", "total_year")),
|
|
("approved_support_q1", ("approved", "support", "q1")),
|
|
("approved_support_q2", ("approved", "support", "q2")),
|
|
("approved_support_q3", ("approved", "support", "q3")),
|
|
("approved_support_q4", ("approved", "support", "q4")),
|
|
("approved_support_year", ("approved", "support", "year")),
|
|
("approved_development_q1", ("approved", "development", "q1")),
|
|
("approved_development_q2", ("approved", "development", "q2")),
|
|
("approved_development_q3", ("approved", "development", "q3")),
|
|
("approved_development_q4", ("approved", "development", "q4")),
|
|
("approved_development_year", ("approved", "development", "year")),
|
|
("approved_total_year", ("approved", "total_year")),
|
|
("fact_support_q1", ("fact", "support", "q1")),
|
|
("fact_support_q2", ("fact", "support", "q2")),
|
|
("fact_support_q3", ("fact", "support", "q3")),
|
|
("fact_support_q4", ("fact", "support", "q4")),
|
|
("fact_support_year", ("fact", "support", "year")),
|
|
("fact_development_q1", ("fact", "development", "q1")),
|
|
("fact_development_q2", ("fact", "development", "q2")),
|
|
("fact_development_q3", ("fact", "development", "q3")),
|
|
("fact_development_q4", ("fact", "development", "q4")),
|
|
("fact_development_year", ("fact", "development", "year")),
|
|
("fact_total_year", ("fact", "total_year")),
|
|
("corrected_support_q2", ("corrected", "support", "q2")),
|
|
("corrected_support_q3", ("corrected", "support", "q3")),
|
|
("corrected_support_q4", ("corrected", "support", "q4")),
|
|
("corrected_development_q2", ("corrected", "development", "q2")),
|
|
("corrected_development_q3", ("corrected", "development", "q3")),
|
|
("corrected_development_q4", ("corrected", "development", "q4")),
|
|
]
|
|
QUARTER_KEYS = ("q1", "q2", "q3", "q4", "totals")
|
|
FORM1_SECTION_BLOCK_KEYS = (
|
|
"sequestration",
|
|
"allocation",
|
|
"reserve",
|
|
"approved",
|
|
"collegial",
|
|
"ckk",
|
|
"contract",
|
|
"plan",
|
|
)
|
|
FORM1_METRIC_BLOCK_KEYS = (
|
|
"plan",
|
|
"reserve",
|
|
"approved",
|
|
"collegial",
|
|
"allocation",
|
|
"sequestration",
|
|
"contract",
|
|
"ckk",
|
|
"contract_summary",
|
|
)
|
|
|
|
|
|
class ExcelValueSerializer:
|
|
def to_excel_value(self, value: Any) -> Any:
|
|
if value is None or isinstance(value, (str, int, float, bool)):
|
|
return value
|
|
if isinstance(value, dict):
|
|
for preferred_key in ("name", "title", "label", "value"):
|
|
preferred_value = value.get(preferred_key)
|
|
if isinstance(preferred_value, (str, int, float, bool)) and preferred_value is not None:
|
|
return preferred_value
|
|
if len(value) == 1:
|
|
only_value = next(iter(value.values()))
|
|
if isinstance(only_value, (str, int, float, bool)) or only_value is None:
|
|
return only_value
|
|
return self._dict_to_readable_text(value)
|
|
if isinstance(value, (list, tuple)):
|
|
if all(isinstance(item, (str, int, float, bool)) or item is None for item in value):
|
|
return ", ".join("" if item is None else str(item) for item in value)
|
|
return json.dumps(value, ensure_ascii=False)
|
|
return str(value)
|
|
|
|
def _dict_to_readable_text(self, value: dict[str, Any]) -> Any:
|
|
parts: list[str] = []
|
|
for key, raw in value.items():
|
|
normalized = self.to_excel_value(raw)
|
|
if self._is_empty_excel_value(normalized):
|
|
continue
|
|
parts.append(f"{key}={normalized}")
|
|
if not parts:
|
|
return None
|
|
return "; ".join(parts)
|
|
|
|
def dict_to_readable_text(self, value: dict[str, Any]) -> Any:
|
|
return self._dict_to_readable_text(value)
|
|
|
|
@staticmethod
|
|
def _is_empty_excel_value(value: Any) -> bool:
|
|
if value is None:
|
|
return True
|
|
if isinstance(value, str) and value.strip() == "":
|
|
return True
|
|
return False
|
|
|
|
|
|
class Form1RowNormalizer:
|
|
def __init__(self, serializer: ExcelValueSerializer):
|
|
self.serializer = serializer
|
|
|
|
def normalize(self, data: dict[str, Any]) -> dict[str, Any]:
|
|
normalized = dict(data)
|
|
self._merge_header_fields_into_row(normalized=normalized, original=data)
|
|
self._promote_quarter_values_from_section_blocks(normalized=normalized)
|
|
self._collapse_metric_blocks_to_scalar(normalized=normalized)
|
|
self._fill_contract_fields(normalized=normalized)
|
|
return normalized
|
|
|
|
def _merge_header_fields_into_row(self, normalized: dict[str, Any], original: dict[str, Any]) -> None:
|
|
header = original.get("header")
|
|
if isinstance(header, dict):
|
|
if "section" in header and "section" not in normalized:
|
|
normalized["section"] = header.get("section")
|
|
if "item_id" in header and "item_id" not in normalized:
|
|
normalized["item_id"] = header.get("item_id")
|
|
if "name" in header and "name" not in normalized:
|
|
normalized["name"] = header.get("name")
|
|
normalized.pop("header", None)
|
|
normalized.pop("line_id", None)
|
|
|
|
def _promote_quarter_values_from_section_blocks(self, normalized: dict[str, Any]) -> None:
|
|
for section_key in FORM1_SECTION_BLOCK_KEYS:
|
|
section_payload = normalized.get(section_key)
|
|
if not isinstance(section_payload, dict):
|
|
continue
|
|
for quarter_key in QUARTER_KEYS:
|
|
if normalized.get(quarter_key) is None and section_payload.get(quarter_key) is not None:
|
|
normalized[quarter_key] = section_payload.get(quarter_key)
|
|
|
|
def _collapse_metric_blocks_to_scalar(self, normalized: dict[str, Any]) -> None:
|
|
for block_key in FORM1_METRIC_BLOCK_KEYS:
|
|
block_value = normalized.get(block_key)
|
|
if isinstance(block_value, dict):
|
|
normalized[block_key] = self._extract_preferred_metric(block_value)
|
|
|
|
def _fill_contract_fields(self, normalized: dict[str, Any]) -> None:
|
|
contract_summary = normalized.get("contract_summary")
|
|
contract_detail = normalized.get("contract_detail")
|
|
if normalized.get("contract") is None:
|
|
normalized["contract"] = self._extract_contract_label(contract_detail, contract_summary)
|
|
if isinstance(contract_detail, dict):
|
|
normalized["contract_sum"] = self._extract_preferred_metric(contract_detail)
|
|
|
|
def _extract_preferred_metric(self, payload: dict[str, Any]) -> Any:
|
|
priority_keys = (
|
|
"total_year",
|
|
"year",
|
|
"totals",
|
|
"total",
|
|
"sum",
|
|
"amount",
|
|
"value",
|
|
"approved_amount",
|
|
"reserved_amount",
|
|
"allocated_amount",
|
|
"booked_amount",
|
|
)
|
|
for key in priority_keys:
|
|
if payload.get(key) is not None:
|
|
return payload.get(key)
|
|
for quarter_key in ("q1", "q2", "q3", "q4"):
|
|
if payload.get(quarter_key) is not None:
|
|
return payload.get(quarter_key)
|
|
return self.serializer.dict_to_readable_text(payload)
|
|
|
|
@staticmethod
|
|
def _extract_contract_label(contract_detail: Any, contract_summary: Any) -> Any:
|
|
for payload in (contract_detail, contract_summary):
|
|
if not isinstance(payload, dict):
|
|
continue
|
|
for key in ("reference", "contract_ref", "counterparty", "subject"):
|
|
value = payload.get(key)
|
|
if value not in (None, ""):
|
|
return value
|
|
return None
|
|
|
|
|
|
class SmetaRowNormalizer:
|
|
def normalize(self, data: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
field: self._pick_nested_value(data, *path)
|
|
for field, path in SMETA_VALUE_PATHS
|
|
}
|
|
|
|
@staticmethod
|
|
def _pick_nested_value(container: Any, *path: str) -> Any:
|
|
cur = container
|
|
for key in path:
|
|
if not isinstance(cur, dict):
|
|
return None
|
|
cur = cur.get(key)
|
|
return cur
|
|
|
|
|
|
class Form3ReportRowNormalizer:
|
|
def normalize(self, data: dict[str, Any]) -> dict[str, Any]:
|
|
normalized = dict(data)
|
|
header = data.get("header")
|
|
if isinstance(header, dict):
|
|
if "section_code" in header and "section" not in normalized:
|
|
normalized["section"] = header.get("section_code")
|
|
if "item_id" in header and "item_id" not in normalized:
|
|
normalized["item_id"] = header.get("item_id")
|
|
if "name" in header and "name" not in normalized:
|
|
normalized["name"] = header.get("name")
|
|
for quarter_key in ("q1", "q2", "q3", "q4"):
|
|
normalized[quarter_key] = self._extract_quarter_metric(normalized.get(quarter_key))
|
|
normalized["totals"] = self._extract_totals_metric(normalized.get("totals"))
|
|
normalized.pop("header", None)
|
|
normalized.pop("line_id", None)
|
|
return normalized
|
|
|
|
@staticmethod
|
|
def _extract_quarter_metric(value: Any) -> Any:
|
|
if not isinstance(value, dict):
|
|
return value
|
|
for key in ("quarter_actual", "total_corr", "economy"):
|
|
if value.get(key) is not None:
|
|
return value.get(key)
|
|
return None
|
|
|
|
@staticmethod
|
|
def _extract_totals_metric(value: Any) -> Any:
|
|
if not isinstance(value, dict):
|
|
return value
|
|
for key in ("total_actual", "total_corr", "economy"):
|
|
if value.get(key) is not None:
|
|
return value.get(key)
|
|
return None
|