baodan/api/insurance/poster/product_source_resolver.py

"""统一解析公共产品和用户私有小册子来源。"""
import json


class ProductSourceError(ValueError):
    def __init__(self, code: int, message: str):
        super().__init__(message)
        self.code = code
        self.message = message


def normalize_product_source(data: dict) -> tuple[str, str]:
    """兼容新 productSource 和旧 productId。"""
    source = data.get("productSource") if isinstance(data.get("productSource"), dict) else {}
    source_type = str(
        source.get("type")
        or data.get("productSourceType")
        or ("library_product" if data.get("productId") else "")
    )
    source_id = str(
        source.get("id")
        or data.get("productSourceId")
        or data.get("productId")
        or ""
    )
    if source_type not in ("library_product", "user_material") or not source_id:
        raise ProductSourceError(1001, "请选择产品或上传产品小册子")
    return source_type, source_id


def resolve_product_source(user_id: str, data: dict) -> dict:
    """返回生成链路使用的统一产品上下文。"""
    source_type, source_id = normalize_product_source(data)
    if source_type == "user_material":
        from insurance.config import get_config

        enabled = get_config("POSTER_USER_MANUAL_UPLOAD_ENABLED", "true")
        if enabled.strip().lower() not in ("1", "true", "yes", "on"):
            raise ProductSourceError(4108, "用户上传产品小册子功能暂未开放")
    if source_type == "library_product":
        return _resolve_library_product(source_id)
    return _resolve_user_material(user_id, source_id)


def _resolve_library_product(product_id: str) -> dict:
    from insurance.models.ppt_config import PptCompany, PptProduct

    product = PptProduct.query.filter_by(
        id=product_id,
        status=1,
        manual_parse_status="reviewed",
    ).first()
    if not product:
        raise ProductSourceError(1002, "产品不存在、未启用或尚未审核")
    company = PptCompany.query.filter_by(id=product.company_id, status=1).first()
    if not company:
        raise ProductSourceError(1002, "产品所属保司不存在或已停用")

    try:
        rules = json.loads(product.manual_parsed_rules) if product.manual_parsed_rules else {}
    except (json.JSONDecodeError, TypeError):
        rules = {}
    product_data = product.to_dict()
    rules.setdefault("product_name", product.display_name)
    rules.setdefault("company_name", company.display_name)
    rules.setdefault("coverage_period", product.coverage_period or "")
    rules.setdefault("payment_period", product.payment_period or "")
    rules.setdefault("insured_age_range", product.insured_age_range or "")
    if not rules.get("features") and product_data.get("highlights"):
        rules["features"] = [
            {"title": item, "summary": ""} if isinstance(item, str) else item
            for item in product_data["highlights"]
        ]

    return {
        "sourceType": "library_product",
        "sourceId": str(product.id),
        "productId": str(product.id),
        "productName": product.display_name,
        "companyId": company.id,
        "companyName": company.display_name,
        "planType": product.plan_type,
        "rules": rules,
        "confirmedAt": product.manual_reviewed_at.isoformat() if product.manual_reviewed_at else None,
        "productData": product_data,
        "companyData": company.to_dict(),
    }


def _resolve_user_material(user_id: str, material_id: str) -> dict:
    from insurance.models.user_product_material import UserProductMaterial

    try:
        numeric_id = int(material_id)
    except (TypeError, ValueError):
        raise ProductSourceError(4106, "产品资料不存在或无权访问")

    material = UserProductMaterial.query.filter(
        UserProductMaterial.id == numeric_id,
        UserProductMaterial.owner_user_id == user_id,
        UserProductMaterial.status == 1,
        UserProductMaterial.deleted_at.is_(None),
    ).first()
    if not material:
        raise ProductSourceError(4106, "产品资料不存在或无权访问")
    if material.parse_status in ("pending", "queued", "parsing"):
        raise ProductSourceError(4103, "产品小册子仍在解析中")
    if material.parse_status == "failed":
        raise ProductSourceError(4104, "产品小册子解析失败，请重新解析")
    if not material.confirmed_rules or not material.confirmed_at:
        raise ProductSourceError(4105, "请先确认产品小册子解析结果")

    try:
        rules = json.loads(material.confirmed_rules)
    except (json.JSONDecodeError, TypeError):
        raise ProductSourceError(4105, "产品小册子确认数据无效，请重新确认")

    return {
        "sourceType": "user_material",
        "sourceId": str(material.id),
        "productId": None,
        "productName": material.product_name,
        "companyId": material.company_id,
        "companyName": material.company_name,
        "planType": material.plan_type or "other",
        "rules": rules,
        "confirmedAt": material.confirmed_at.isoformat(),
        "productData": {
            "id": str(material.id),
            "displayName": material.product_name,
            "maskedDisplayName": "",
            "planType": material.plan_type or "other",
        },
        "companyData": {
            "id": material.company_id,
            "displayName": material.company_name,
            "maskedDisplayName": "",
        },
    }


_PRODUCT_SNAPSHOT_FIELDS = (
    "id", "companyId", "planType", "displayName", "maskedDisplayName",
    "productCode", "productType", "coveragePeriod", "paymentPeriod",
    "insuredAgeRange", "waitingPeriod", "highlights",
)
_COMPANY_SNAPSHOT_FIELDS = (
    "id", "displayName", "maskedDisplayName", "companyIntro",
    "companyHighlights", "rating", "foundedYear", "nameZh", "nameEn",
    "shortEn", "logoUrl",
)


def product_snapshot(context: dict) -> dict:
    """移除内部展示字段，生成可审计、可序列化的产品快照。"""
    product_data = context.get("productData") or {}
    company_data = context.get("companyData") or {}
    context = {
        **context,
        "productData": {
            key: product_data.get(key)
            for key in _PRODUCT_SNAPSHOT_FIELDS
            if key in product_data
        },
        "companyData": {
            key: company_data.get(key)
            for key in _COMPANY_SNAPSHOT_FIELDS
            if key in company_data
        },
    }
    return {
        key: context.get(key)
        for key in (
            "sourceType",
            "sourceId",
            "productId",
            "productName",
            "companyId",
            "companyName",
            "planType",
            "rules",
            "confirmedAt",
            "productData",
            "companyData",
        )
    }


def case_matches_product(case, context: dict) -> bool:
    case_type = case.product_source_type or "library_product"
    case_id = case.product_source_id or case.product_id
    return case_type == context["sourceType"] and str(case_id or "") == context["sourceId"]