baodan/api/insurance/plan_data/conversion.py

"""旧提取结构到 PlanData v1 的无推断转换。"""
from __future__ import annotations

import re
from typing import Any

from insurance.plan_data.field_values import extracted_field_value, missing_field_value


IDENTITY_FIELDS = {
    "productName": ("product_name",),
    "productType": ("product_type", "plan_type"),
}
INSURED_FIELDS = {
    "insuredName": ("name",),
    "insuredAge": ("age",),
    "insuredGender": ("gender",),
    "insuredSmoker": ("smoker",),
}
POLICY_FIELDS = {
    "currency": ("currency",),
    "sumAssured": ("sum_insured", "sumAssured"),
    "annualPremium": ("annual_premium", "annualPremium"),
    "premiumPaymentPeriod": ("premium_payment_period", "payYears"),
    "coveragePeriod": ("coverage_period", "coveragePeriod"),
    "contractualTotalPremium": ("total_premium", "contractualTotalPremium"),
}


def legacy_to_plan_data(data: dict, evidence_entries: list[dict] | None = None) -> dict:
    evidence_by_path = _evidence_map(evidence_entries or (data.get("meta") or {}).get("evidence") or [])
    identity = {
        target: _field_value(_first(data, *aliases), evidence_by_path, aliases)
        for target, aliases in IDENTITY_FIELDS.items()
    }
    insured = data.get("insured") or {}
    identity.update({
        target: _field_value(
            _first(insured, *aliases),
            evidence_by_path,
            tuple(f"insured.{item}" for item in aliases),
        )
        for target, aliases in INSURED_FIELDS.items()
    })

    policy = data.get("policy") or {}
    currency = _first(policy, "currency") or data.get("currency")
    policy_values = {}
    for target, aliases in POLICY_FIELDS.items():
        value = _first(policy, *aliases)
        if value is None:
            value = _first(data, *aliases)
        paths = tuple(f"policy.{item}" for item in aliases) + aliases
        policy_values[target] = _field_value(
            value,
            evidence_by_path,
            paths,
            currency=(
                str(currency).upper()
                if target not in {"currency", "premiumPaymentPeriod", "coveragePeriod"} and currency
                else None
            ),
        )
    consumed_policy_keys = {alias for aliases in POLICY_FIELDS.values() for alias in aliases}
    for key, value in policy.items():
        if key in consumed_policy_keys:
            continue
        policy_values[_camel_case(key)] = _field_value(
            value,
            evidence_by_path,
            (f"policy.{key}",),
            currency=str(currency).upper() if currency and _looks_amount_field(key) else None,
        )

    scenarios: dict[str, list[dict]] = {}
    for row in data.get("benefit_illustration") or data.get("benefit_table") or []:
        if not isinstance(row, dict):
            continue
        scenario = str(row.get("scenario_type") or row.get("scenarioType") or "base")
        scenarios.setdefault(scenario, []).append(_benefit_row(row, scenario, evidence_by_path, currency))

    withdrawals = [
        _plain_row(row, currency)
        for row in data.get("withdrawal_illustration") or data.get("withdrawals") or []
        if isinstance(row, dict)
    ]
    return {
        "identity": identity,
        "policy": policy_values,
        "benefitScenarios": [
            {"scenarioType": scenario, "rows": rows}
            for scenario, rows in sorted(scenarios.items())
        ],
        "withdrawals": withdrawals,
        "riders": list(data.get("riders") or data.get("coverage_items") or []),
        "assumptions": dict(data.get("assumptions") or {}),
    }


def legacy_path_to_plan_path(data: dict, path: str) -> str | None:
    """把复核界面的旧字段路径映射到 PlanData 字段路径。"""
    normalized = str(path or "").replace("[", ".").replace("]", "").strip(".")
    direct = {
        "product_name": "identity.productName",
        "product_type": "identity.productType",
        "insured.name": "identity.insuredName",
        "insured.age": "identity.insuredAge",
        "insured.gender": "identity.insuredGender",
        "insured.smoker": "identity.insuredSmoker",
        "policy.currency": "policy.currency",
        "policy.sum_insured": "policy.sumAssured",
        "policy.annual_premium": "policy.annualPremium",
        "policy.premium_payment_period": "policy.premiumPaymentPeriod",
        "policy.coverage_period": "policy.coveragePeriod",
        "policy.total_premium": "policy.contractualTotalPremium",
    }
    if normalized in direct:
        return direct[normalized]
    parts = normalized.split(".")
    if parts and parts[0] == "policy" and len(parts) == 2:
        return f"policy.{_camel_case(parts[1])}"
    if len(parts) < 3 or parts[0] != "benefit_illustration" or not parts[1].isdigit():
        return None
    source_rows = data.get("benefit_illustration") or []
    source_index = int(parts[1])
    if source_index >= len(source_rows) or not isinstance(source_rows[source_index], dict):
        return None
    source_row = source_rows[source_index]
    scenario = str(source_row.get("scenario_type") or source_row.get("scenarioType") or "base")
    scenario_names = sorted({
        str(row.get("scenario_type") or row.get("scenarioType") or "base")
        for row in source_rows if isinstance(row, dict)
    })
    scenario_index = scenario_names.index(scenario)
    scenario_row_index = sum(
        1 for row in source_rows[:source_index]
        if isinstance(row, dict)
        and str(row.get("scenario_type") or row.get("scenarioType") or "base") == scenario
    )
    return (
        f"benefitScenarios.{scenario_index}.rows.{scenario_row_index}."
        f"{_camel_case(parts[2])}"
    )


def _benefit_row(row: dict, scenario: str, evidence_by_path: dict, currency) -> dict:
    year = _first(row, "policy_year", "policyYear", "year")
    variant = str(row.get("row_variant") or row.get("rowVariant") or "default")
    result = {
        "policyYear": _field_value(year, evidence_by_path, (f"benefit.{scenario}.{year}.{variant}.policy_year",)),
        "rowVariant": variant,
    }
    aliases = {
        "age": ("age",),
        "totalPremiumPaid": ("total_premium_paid", "totalPremiumPaid"),
        "guaranteedCashValue": ("guaranteed_cash_value", "guaranteedCashValue"),
        "nonGuaranteedCashValue": ("non_guaranteed_cash_value", "nonGuaranteedCashValue"),
        "totalSurrenderValue": ("total_surrender_value", "totalSurrenderValue", "total_surrender"),
        "deathBenefit": ("death_benefit", "deathBenefit"),
        "accountValue": ("account_value", "accountValue"),
    }
    for target, source_aliases in aliases.items():
        paths = tuple(f"benefit.{scenario}.{year}.{variant}.{item}" for item in source_aliases)
        result[target] = _field_value(
            _first(row, *source_aliases),
            evidence_by_path,
            paths,
            currency=str(currency).upper() if target != "age" and currency else None,
        )
    consumed = {alias for source_aliases in aliases.values() for alias in source_aliases}
    consumed.update({"scenario_type", "scenarioType", "row_variant", "rowVariant", "source_page", "sourcePage"})
    for key, value in row.items():
        if key in consumed:
            continue
        result[_camel_case(key)] = _field_value(
            value,
            evidence_by_path,
            (f"benefit.{scenario}.{year}.{variant}.{key}",),
            currency=str(currency).upper() if currency and _looks_amount_field(key) else None,
        )
    return result


def _plain_row(row: dict, currency) -> dict:
    result = {}
    for key, value in row.items():
        result[_camel_case(key)] = (
            extracted_field_value(
                value,
                currency=str(currency).upper() if currency and _looks_amount_field(key) else None,
            )
            if value is not None else missing_field_value()
        )
    return result


def _field_value(value: Any, evidence_by_path: dict, paths, currency: str | None = None) -> dict:
    evidence = []
    for path in paths:
        evidence.extend(evidence_by_path.get(path, ()))
    if value is None or (isinstance(value, str) and not value.strip()):
        return missing_field_value(currency=currency)
    return extracted_field_value(value, currency=currency, evidence=evidence)


def _evidence_map(entries: list[dict]) -> dict[str, list[dict]]:
    result: dict[str, list[dict]] = {}
    for entry in entries:
        if not isinstance(entry, dict):
            continue
        path = str(entry.get("fieldPath") or "")
        ref = entry.get("evidence") if "evidence" in entry else entry
        if path and isinstance(ref, dict):
            result.setdefault(path, []).append(dict(ref))
    return result


def _first(mapping: dict, *keys):
    for key in keys:
        if key in mapping and mapping[key] is not None and mapping[key] != "":
            return mapping[key]
    return None


def _looks_amount_field(key: str) -> bool:
    token = str(key).lower()
    return any(word in token for word in ("premium", "value", "benefit", "amount", "withdraw"))


def _camel_case(value: str) -> str:
    return re.sub(r"_([a-z0-9])", lambda match: match.group(1).upper(), str(value))