"""统一解析公共产品和用户私有小册子来源。""" 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", deleted_at=None, ).first() if not product: raise ProductSourceError(1002, "产品不存在、未启用或尚未审核") company = PptCompany.query.filter_by( id=product.company_id, status=1, deleted_at=None ).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", "maskingEnabled", "productCode", "productType", "coveragePeriod", "paymentPeriod", "insuredAgeRange", "waitingPeriod", "highlights", ) _COMPANY_SNAPSHOT_FIELDS = ( "id", "displayName", "maskedDisplayName", "companyIntro", "maskingEnabled", "logoEnabled", "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"]