baodan/api/insurance/ppt/normalizer.py

"""归一化模块 — 将 LLM 提取的原始数据转为标准结构。"""
import re
import hashlib
from typing import Optional


def _safe_number(value) -> float:
    """安全数值转换。"""
    if value is None:
        return 0
    try:
        parsed = float(value)
        return parsed if parsed == parsed else 0  # NaN check
    except (ValueError, TypeError):
        return 0


def _extract_years(value) -> int:
    """从字符串提取年数（如 '5年' → 5）。"""
    if value is None:
        return 0
    match = re.search(r"\d+(?:\.\d+)?", str(value))
    return int(float(match.group())) if match else 0


def _sha256(file_path: Optional[str]) -> str:
    """计算文件 SHA-256。"""
    if not file_path:
        return ""
    try:
        h = hashlib.sha256()
        with open(file_path, "rb") as f:
            for chunk in iter(lambda: f.read(8192), b""):
                h.update(chunk)
        return h.hexdigest()
    except Exception:
        return ""


def normalize_savings_plan(raw: dict, pdf_path: str = None, parser: str = "llm-json") -> dict:
    """归一化储蓄险提取数据。"""
    insured_age = _safe_number(raw.get("insured", {}).get("age"))
    insured = raw.get("insured", {})

    # 归一化利益演示行
    benefit_rows = []
    for row in raw.get("benefit_illustration", []):
        if not isinstance(row, dict):
            continue
        policy_year = _safe_number(row.get("policy_year"))
        if policy_year <= 0:
            continue
        age = _safe_number(row.get("age")) or (insured_age + policy_year)
        benefit_rows.append({
            "policyYear": int(policy_year),
            "age": int(age),
            "totalPremiumPaid": _safe_number(row.get("total_premium_paid")),
            "guaranteedCashValue": _safe_number(row.get("guaranteed_cash_value")),
            "reversionaryBonus": _safe_number(row.get("reversionary_bonus")),
            "terminalDividend": _safe_number(row.get("terminal_dividend")),
            "totalSurrenderValue": _safe_number(row.get("total_surrender_value")),
            "deathBenefit": _safe_number(row.get("death_benefit")),
            "sourcePage": int(_safe_number(row.get("source_page"))) if row.get("source_page") else None,
        })
    benefit_rows.sort(key=lambda r: r["policyYear"])

    # 归一化退保行
    cumulative = 0.0
    withdrawal_rows = []
    for row in raw.get("withdrawal_illustration", []):
        if not isinstance(row, dict):
            continue
        policy_year = _safe_number(row.get("policy_year"))
        if policy_year <= 0:
            continue
        annual = _safe_number(row.get("annual_withdrawal"))
        total_withdrawn = _safe_number(row.get("total_withdrawn") or row.get("cumulative_withdrawal"))
        cumulative = total_withdrawn if total_withdrawn > 0 else cumulative + annual
        age = _safe_number(row.get("age")) or (insured_age + policy_year)
        withdrawal_rows.append({
            "policyYear": int(policy_year),
            "age": int(age),
            "totalPremiumPaid": _safe_number(row.get("total_premium_paid")),
            "annualWithdrawal": annual,
            "cumulativeWithdrawal": cumulative,
            "surrenderValueAfter": _safe_number(row.get("surrender_value_after")),
            "guaranteedValueAfter": _safe_number(row.get("guaranteed_value_after")),
            "basicSumInsuredAfter": _safe_number(row.get("basic_sum_insured_after")),
            "sourcePage": int(_safe_number(row.get("source_page"))) if row.get("source_page") else None,
        })
    withdrawal_rows.sort(key=lambda r: r["policyYear"])

    # 保单信息
    policy = raw.get("policy", {})
    annual_premium = _safe_number(policy.get("annual_premium"))
    pay_years = _extract_years(policy.get("premium_payment_period"))
    raw_product_name = raw.get("product_name") or policy.get("product_name", "")

    return {
        "kind": "savings",
        "productName": raw_product_name,
        "rawProductName": raw_product_name,
        "insured": {
            "name": insured.get("name") or "客户",
            "age": int(insured_age),
            "gender": insured.get("gender") or "",
        },
        "policy": {
            "currency": policy.get("currency") or "USD",
            "annualPremium": annual_premium,
            "annualPremiumWithLevy": policy.get("total_premium_with_levy"),
            "payYears": pay_years,
            "contractualTotalPremium": annual_premium * pay_years,
            "coveragePeriod": policy.get("coverage_period", ""),
        },
        "benefitRows": benefit_rows,
        "withdrawalRows": withdrawal_rows,
        "withdrawalProvenance": "official_extracted" if withdrawal_rows else "missing",
        "source": {
            "pdfHash": _sha256(pdf_path),
            "pdfPath": pdf_path,
            "parser": parser,
        },
    }


def normalize_ci_plan(raw: dict, pdf_path: str = None, parser: str = "llm-json") -> dict:
    """归一化重疾险提取数据。"""
    insured = raw.get("insured", {})
    policy = raw.get("policy", {})
    insured_age = _safe_number(insured.get("age"))

    # 保障项目
    coverage_items = []
    for item in raw.get("coverage_items", []):
        if not isinstance(item, dict):
            continue
        name = item.get("name") or item.get("label") or ""
        coverage_items.append({
            "name": name,
            "amount": _safe_number(item.get("amount")),
            "description": item.get("description") or "",
            "sourcePage": int(_safe_number(item.get("source_page"))) if item.get("source_page") else None,
        })

    # 利益演示行
    benefit_rows = []
    for row in raw.get("benefit_illustration", []):
        if not isinstance(row, dict):
            continue
        policy_year = _safe_number(row.get("policy_year"))
        if policy_year <= 0:
            continue
        death_benefit = _safe_number(row.get("death_benefit"))
        if death_benefit == 0:
            death_benefit = _safe_number(policy.get("sum_insured"))
        benefit_rows.append({
            "policyYear": int(policy_year),
            "totalPremiumPaid": _safe_number(row.get("total_premium_paid")),
            "deathBenefit": death_benefit,
            "totalSurrenderValue": death_benefit,  # 渲染器统一字段（CI 用身故赔付作为主值）
            "ciBenefit": _safe_number(row.get("ci_benefit")) if row.get("ci_benefit") else None,
            "sourcePage": int(_safe_number(row.get("source_page"))) if row.get("source_page") else None,
        })
    benefit_rows.sort(key=lambda r: r["policyYear"])

    annual_premium = _safe_number(policy.get("annual_premium"))
    pay_years = _extract_years(policy.get("premium_payment_period"))
    base_sum_insured = _safe_number(
        raw.get("base_sum_insured")
        or policy.get("basic_sum_insured")
        or policy.get("sum_insured")
    )

    return {
        "kind": "ci",
        "productName": raw.get("product_name", ""),
        "insured": {
            "name": insured.get("name") or "客户",
            "age": int(insured_age),
            "gender": insured.get("gender") or "",
            "smoker": insured.get("smoker") or "",
        },
        "policy": {
            "currency": policy.get("currency") or "USD",
            "sumInsured": _safe_number(policy.get("sum_insured")),
            "baseSumInsured": base_sum_insured,
            "upgradeBenefitAmount": _safe_number(raw.get("upgrade_benefit_amount")),
            "upgradeBenefitYears": _safe_number(raw.get("upgrade_benefit_years")),
            "annualPremium": annual_premium,
            "annualPremiumWithLevy": policy.get("total_premium_with_levy"),
            "payYears": pay_years,
            "totalPremium": annual_premium * pay_years,
            "coveragePeriod": policy.get("coverage_period", ""),
        },
        "coverageSummary": {
            "majorCiCount": int(_safe_number(raw.get("major_ci_count"))),
            "earlyCiCount": int(_safe_number(raw.get("early_ci_count"))),
        },
        "coverageItems": coverage_items,
        "icuBenefitRules": raw.get("icu_benefit_rules", []),
        "multiClaimRules": raw.get("multi_claim", []),
        "premiumWaiverRiders": raw.get("premium_waiver_riders", []),
        "benefitRows": benefit_rows,
        "source": {
            "pdfHash": _sha256(pdf_path),
            "pdfPath": pdf_path,
            "parser": parser,
        },
    }


def normalize_iul_plan(raw: dict, pdf_path: str = None, parser: str = "llm-json") -> dict:
    """归一化 IUL 提取数据。"""
    insured = raw.get("insured", {})
    policy = raw.get("policy", {})
    insured_age = _safe_number(insured.get("age"))

    # 指数账户
    index_accounts = []
    for acc in raw.get("index_accounts", []):
        if not isinstance(acc, dict):
            continue
        allocation = acc.get("allocation", 0)
        if isinstance(allocation, str):
            try:
                allocation = float(allocation)
            except ValueError:
                allocation = 0
        index_accounts.append({
            "name": acc.get("name", ""),
            "allocation": allocation,
            "assumedRate": acc.get("current_assumed_rate") or acc.get("current_rate") or "",
            "floorRate": acc.get("guaranteed_floor_rate") or acc.get("guaranteed_floor") or "",
            "capRate": acc.get("cap_rate") or "",
            "participationRate": acc.get("participation_rate") or "",
        })

    # 利益演示行
    benefit_rows = []
    for row in raw.get("benefit_illustration", []):
        if not isinstance(row, dict):
            continue
        policy_year = _safe_number(row.get("policy_year"))
        if policy_year <= 0:
            continue
        age = _safe_number(row.get("age")) or (insured_age + policy_year)

        # 回填非保证字段
        non_guaranteed_account = _safe_number(row.get("non_guaranteed_account_value") or row.get("account_value"))
        non_guaranteed_cash = _safe_number(row.get("non_guaranteed_cash_value") or row.get("cash_value"))
        non_guaranteed_death = _safe_number(row.get("non_guaranteed_death_benefit") or row.get("death_benefit"))

        benefit_rows.append({
            "policyYear": int(policy_year),
            "age": int(age),
            "totalPremiumPaid": _safe_number(row.get("total_premium_paid")),
            "guaranteedCashValue": _safe_number(row.get("guaranteed_cash_value")),
            "nonGuaranteedCashValue": non_guaranteed_cash,
            "totalSurrenderValue": non_guaranteed_cash,  # 渲染器统一字段
            "guaranteedDeathBenefit": _safe_number(row.get("guaranteed_death_benefit")),
            "nonGuaranteedDeathBenefit": non_guaranteed_death,
            "sourcePage": int(_safe_number(row.get("source_page"))) if row.get("source_page") else None,
        })
    benefit_rows.sort(key=lambda r: r["policyYear"])

    annual_premium = _safe_number(policy.get("annual_premium"))
    payment_period = policy.get("premium_payment_period", "")
    pay_years = _extract_years(payment_period)

    return {
        "kind": "iul",
        "productName": raw.get("product_name", ""),
        "insured": {
            "name": insured.get("name") or "客户",
            "age": int(insured_age),
            "gender": insured.get("gender") or "",
            "smoker": insured.get("smoker") or "",
        },
        "policy": {
            "currency": policy.get("currency") or "USD",
            "sumInsured": _safe_number(policy.get("sum_insured")),
            "initialPremium": _safe_number(policy.get("initial_premium")),
            "annualPremium": annual_premium,
            "payYears": pay_years,
            "totalPremium": annual_premium * pay_years,
            "paymentPeriod": str(payment_period),
            "coveragePeriod": policy.get("coverage_period", ""),
        },
        "indexAccounts": index_accounts,
        "benefitRows": benefit_rows,
        "source": {
            "pdfHash": _sha256(pdf_path),
            "pdfPath": pdf_path,
            "parser": parser,
        },
    }


def map_savings_metrics(data: dict) -> dict:
    """提取储蓄险关键指标。"""
    insured = data.get("insured", {})
    policy = data.get("policy", {})
    benefit_rows = data.get("benefit_illustration", [])
    if not isinstance(benefit_rows, list):
        benefit_rows = []

    annual_premium = _safe_number(policy.get("annual_premium"))
    pay_years = _extract_years(policy.get("premium_payment_period"))
    total_premium = annual_premium * pay_years

    # 回本年度
    breakeven_year = None
    for row in benefit_rows:
        if not isinstance(row, dict):
            continue
        tp = _safe_number(row.get("total_premium_paid"))
        sv = _safe_number(row.get("total_surrender_value"))
        if tp > 0 and sv >= tp:
            breakeven_year = int(_safe_number(row.get("policy_year")))
            break

    # 20年/30年倍数
    multiple_20 = None
    multiple_30 = None
    for row in benefit_rows:
        if not isinstance(row, dict):
            continue
        py = int(_safe_number(row.get("policy_year")))
        tp = _safe_number(row.get("total_premium_paid"))
        sv = _safe_number(row.get("total_surrender_value"))
        if py == 20 and tp > 0:
            multiple_20 = round(sv / tp, 2)
        if py == 30 and tp > 0:
            multiple_30 = round(sv / tp, 2)

    # 退保开始年度
    withdraw_start_year = None
    withdraw_start_age = None
    for row in data.get("withdrawal_illustration", []):
        if not isinstance(row, dict):
            continue
        if _safe_number(row.get("annual_withdrawal")) > 0:
            withdraw_start_year = int(_safe_number(row.get("policy_year")))
            withdraw_start_age = int(_safe_number(row.get("age")))
            break

    return {
        "insuredName": insured.get("name") or "",
        "insuredAge": int(_safe_number(insured.get("age"))),
        "insuredGender": insured.get("gender") or "",
        "productName": data.get("product_name") or policy.get("product_name") or "",
        "currency": policy.get("currency") or "",
        "annualPremium": annual_premium,
        "payYears": pay_years,
        "totalPremium": total_premium,
        "breakevenYear": breakeven_year,
        "multiple20": multiple_20,
        "multiple30": multiple_30,
        "withdrawStartYear": withdraw_start_year,
        "withdrawStartAge": withdraw_start_age,
    }