197 lines
7.5 KiB
Python
197 lines
7.5 KiB
Python
"""验证模块 — 数据完整性和导出就绪检查。"""
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
|
|
@dataclass
|
|
class FormalDeckIssue:
|
|
code: str
|
|
level: str # "error" | "warn"
|
|
message: str
|
|
|
|
|
|
@dataclass
|
|
class ValidationIssue:
|
|
field: str
|
|
message: str
|
|
level: str # "error" | "warn"
|
|
|
|
|
|
def _is_continuous(rows: list[dict]) -> bool:
|
|
"""检查保单年度是否连续。"""
|
|
years = sorted(r.get("policyYear", 0) for r in rows if isinstance(r, dict))
|
|
for i in range(1, len(years)):
|
|
if years[i] != years[i - 1] + 1:
|
|
return False
|
|
return True
|
|
|
|
|
|
def validate_formal_savings_plan(plan: dict) -> list[FormalDeckIssue]:
|
|
"""验证归一化储蓄险数据的导出就绪性。"""
|
|
issues = []
|
|
|
|
def err(code, msg):
|
|
issues.append(FormalDeckIssue(code, "error", msg))
|
|
|
|
def warn(code, msg):
|
|
issues.append(FormalDeckIssue(code, "warn", msg))
|
|
|
|
if not plan.get("productName"):
|
|
err("PRODUCT_NAME_MISSING", "产品名称缺失")
|
|
|
|
insured = plan.get("insured", {})
|
|
if not insured.get("age"):
|
|
err("INSURED_AGE_MISSING", "被保险人年龄缺失")
|
|
|
|
policy = plan.get("policy", {})
|
|
if _safe_number(policy.get("annualPremium")) <= 0:
|
|
err("ANNUAL_PREMIUM_INVALID", "年缴保费必须大于 0")
|
|
if _safe_number(policy.get("payYears")) <= 0:
|
|
err("PAY_YEARS_INVALID", "缴费年期必须大于 0")
|
|
|
|
benefit_rows = plan.get("benefitRows", [])
|
|
if len(benefit_rows) < 20:
|
|
err("BENEFIT_ROWS_INCOMPLETE", f"利益演示行数不足（当前 {len(benefit_rows)} 行，至少需要 20 行）")
|
|
if benefit_rows and not _is_continuous(benefit_rows):
|
|
err("BENEFIT_ROWS_DISCONTINUOUS", "利益演示保单年度不连续")
|
|
|
|
source = plan.get("source", {})
|
|
if not source.get("pdfHash"):
|
|
err("SOURCE_HASH_MISSING", "缺少 PDF 哈希（来源追溯）")
|
|
|
|
if not any(r.get("sourcePage") for r in benefit_rows):
|
|
warn("BENEFIT_SOURCE_PAGE_MISSING", "利益演示缺少来源页码")
|
|
|
|
withdrawal_rows = plan.get("withdrawalRows", [])
|
|
if withdrawal_rows and not _is_continuous(withdrawal_rows):
|
|
err("WITHDRAWAL_ROWS_DISCONTINUOUS", "退保演示保单年度不连续")
|
|
if withdrawal_rows and not any(r.get("sourcePage") for r in withdrawal_rows):
|
|
warn("WITHDRAWAL_SOURCE_PAGE_MISSING", "退保演示缺少来源页码")
|
|
if not withdrawal_rows:
|
|
warn("WITHDRAWAL_ROWS_MISSING", "无退保数据（正式 PPT 将隐藏退保页面）")
|
|
|
|
return issues
|
|
|
|
|
|
def validate_formal_ci_plan(plan: dict) -> list[FormalDeckIssue]:
|
|
"""验证归一化重疾险数据。"""
|
|
issues = []
|
|
|
|
def err(code, msg):
|
|
issues.append(FormalDeckIssue(code, "error", msg))
|
|
|
|
if not plan.get("productName"):
|
|
err("CI_PRODUCT_NAME_MISSING", "产品名称缺失")
|
|
if not plan.get("insured", {}).get("age"):
|
|
err("CI_INSURED_AGE_MISSING", "被保险人年龄缺失")
|
|
if _safe_number(plan.get("policy", {}).get("sumInsured")) <= 0:
|
|
err("CI_SUM_INSURED_INVALID", "保额必须大于 0")
|
|
if _safe_number(plan.get("policy", {}).get("annualPremium")) <= 0:
|
|
err("CI_ANNUAL_PREMIUM_INVALID", "年缴保费必须大于 0")
|
|
if _safe_number(plan.get("policy", {}).get("payYears")) <= 0:
|
|
err("CI_PAY_YEARS_INVALID", "缴费年期必须大于 0")
|
|
if not plan.get("coverageItems"):
|
|
err("CI_COVERAGE_ITEMS_MISSING", "保障项目列表为空")
|
|
|
|
return issues
|
|
|
|
|
|
def validate_formal_iul_plan(plan: dict) -> list[FormalDeckIssue]:
|
|
"""验证归一化 IUL 数据。"""
|
|
issues = []
|
|
|
|
def err(code, msg):
|
|
issues.append(FormalDeckIssue(code, "error", msg))
|
|
|
|
def warn(code, msg):
|
|
issues.append(FormalDeckIssue(code, "warn", msg))
|
|
|
|
if not plan.get("productName"):
|
|
err("IUL_PRODUCT_NAME_MISSING", "产品名称缺失")
|
|
if not plan.get("insured", {}).get("age"):
|
|
err("IUL_INSURED_AGE_MISSING", "被保险人年龄缺失")
|
|
if _safe_number(plan.get("policy", {}).get("sumInsured")) <= 0:
|
|
err("IUL_SUM_INSURED_INVALID", "保额必须大于 0")
|
|
if not plan.get("indexAccounts"):
|
|
err("IUL_INDEX_ACCOUNT_MISSING", "指数账户配置为空")
|
|
|
|
benefit_rows = plan.get("benefitRows", [])
|
|
if len(benefit_rows) < 20:
|
|
err("IUL_BENEFIT_ROWS_INCOMPLETE", f"利益演示行数不足（当前 {len(benefit_rows)} 行）")
|
|
is_continuous = _is_continuous(benefit_rows) if benefit_rows else True
|
|
if benefit_rows and not is_continuous:
|
|
warn("IUL_BENEFIT_ROWS_DISCONTINUOUS", "利益演示保单年度不连续，请确认 PDF 是否只提供里程碑年度")
|
|
|
|
source = plan.get("source", {})
|
|
if not source.get("pdfHash"):
|
|
err("IUL_SOURCE_HASH_MISSING", "缺少 PDF 哈希")
|
|
|
|
if benefit_rows and not any(r.get("sourcePage") for r in benefit_rows):
|
|
warn("IUL_SOURCE_PAGE_MISSING", "利益演示缺少来源页码")
|
|
|
|
# 缴费年期一致性检查
|
|
payment_period = plan.get("policy", {}).get("paymentPeriod", "")
|
|
if benefit_rows and payment_period and is_continuous:
|
|
# 通过数据检测实际缴费年数
|
|
detected_years = 0
|
|
for i in range(1, len(benefit_rows)):
|
|
prev = _safe_number(benefit_rows[i - 1].get("totalPremiumPaid"))
|
|
curr = _safe_number(benefit_rows[i].get("totalPremiumPaid"))
|
|
if curr > prev:
|
|
detected_years += 1
|
|
# 解析声明的缴费年期
|
|
stated_years = _extract_years(payment_period)
|
|
if stated_years > 0 and detected_years > 0 and abs(detected_years - stated_years) > 1:
|
|
err("IUL_PAY_TERM_MISMATCH", f"缴费年期不一致：声明 {stated_years} 年，数据检测 {detected_years} 年")
|
|
|
|
# 年龄合理性检查
|
|
for row in benefit_rows:
|
|
if isinstance(row, dict):
|
|
age = _safe_number(row.get("age"))
|
|
if age > 0 and (age < 0 or age > 150):
|
|
err("IUL_AGE_OUT_OF_RANGE", f"年龄超出合理范围: {age}")
|
|
break
|
|
|
|
return issues
|
|
|
|
|
|
def validate_savings_metrics(metrics: dict) -> list[ValidationIssue]:
|
|
"""验证储蓄险关键指标。"""
|
|
issues = []
|
|
|
|
if not metrics.get("productName"):
|
|
issues.append(ValidationIssue("productName", "产品名称缺失", "error"))
|
|
if _safe_number(metrics.get("insuredAge")) <= 0:
|
|
issues.append(ValidationIssue("insuredAge", "被保险人年龄无效", "error"))
|
|
if not metrics.get("currency"):
|
|
issues.append(ValidationIssue("currency", "货币缺失", "error"))
|
|
if _safe_number(metrics.get("annualPremium")) <= 0:
|
|
issues.append(ValidationIssue("annualPremium", "年缴保费无效", "error"))
|
|
if _safe_number(metrics.get("payYears")) <= 0:
|
|
issues.append(ValidationIssue("payYears", "缴费年期无效", "error"))
|
|
if metrics.get("multiple20") is None:
|
|
issues.append(ValidationIssue("multiple20", "20年倍数缺失", "warn"))
|
|
if metrics.get("multiple30") is None:
|
|
issues.append(ValidationIssue("multiple30", "30年倍数缺失", "warn"))
|
|
if metrics.get("breakevenYear") is None:
|
|
issues.append(ValidationIssue("breakevenYear", "回本年度缺失", "warn"))
|
|
|
|
return issues
|
|
|
|
|
|
def _safe_number(value) -> float:
|
|
if value is None:
|
|
return 0
|
|
try:
|
|
return float(value)
|
|
except (ValueError, TypeError):
|
|
return 0
|
|
|
|
|
|
def _extract_years(value) -> int:
|
|
import re
|
|
if value is None:
|
|
return 0
|
|
match = re.search(r"\d+(?:\.\d+)?", str(value))
|
|
return int(float(match.group())) if match else 0
|