from datetime import datetime from typing import List, Optional, Any, Dict, Union from pydantic import BaseModel, Field, field_validator, model_validator from decimal import Decimal class ValuationAssessmentBase(BaseModel): """估值评估基础模型""" # 基础信息 asset_name: str = Field(..., description="资产名称") institution: str = Field(..., description="所属机构") industry: str = Field(..., description="所属行业") # 财务状况 annual_revenue: Optional[str] = Field(None, description="近12个月机构营收/万元") rd_investment: Optional[str] = Field(None, description="近12个月机构研发投入/万元") three_year_income: Optional[List[Union[int, float]]] = Field(None, description="近三年机构收益/万元") funding_status: Optional[str] = Field(None, description="资产受资助情况") # 非遗等级与技术 inheritor_level: Optional[str] = Field(None, description="非遗传承人等级") inheritor_ages: Optional[List[int]] = Field(None, description="传承人年龄列表") inheritor_age_count: Optional[List[Any]] = Field(None, description="非遗传承人年龄水平及数量") inheritor_certificates: Optional[List[str]] = Field(None, description="非遗传承人等级证书") heritage_level: Optional[str] = Field(None, description="非遗等级") heritage_asset_level: Optional[str] = Field(None, description="非遗资产等级") patent_application_no: Optional[str] = Field(None, description="非遗资产所用专利的申请号") patent_remaining_years: Optional[str] = Field(None, description="专利剩余年限") # 未使用 historical_evidence: Optional[Dict[str, int]] = Field(None, description="非遗资产历史证明证据及数量") patent_certificates: Optional[List[str]] = Field(None, description="非遗资产所用专利的证书") pattern_images: Optional[List[str]] = Field(None, description="非遗纹样图片") report_url: Optional[str] = Field(None, description="评估报告URL") certificate_url: Optional[str] = Field(None, description="证书URL") # 非遗应用与推广 application_maturity: Optional[str] = Field(None, description="非遗资产应用成熟度") implementation_stage: Optional[str] = Field(None, description="非遗资产应用成熟度") application_coverage: Optional[str] = Field(None, description="非遗资产应用覆盖范围") coverage_area: Optional[str] = Field(None, description="非遗资产应用覆盖范围") cooperation_depth: Optional[str] = Field(None, description="非遗资产跨界合作深度") collaboration_type: Optional[str] = Field(None, description="非遗资产跨界合作深度") offline_activities: Optional[str] = Field(None, description="近12个月线下相关宣讲活动次数") offline_teaching_count: Optional[int] = Field(None, description="近12个月线下相关演讲活动次数") # 未使用 online_accounts: Optional[List[Any]] = Field(None, description="线上相关宣传账号信息") platform_accounts: Optional[Dict[str, Dict[str, Union[str, int]]]] = Field(None, description="线上相关宣传账号信息") # 非遗资产衍生商品信息 sales_volume: Optional[str] = Field(None, description="该商品近12个月销售量") link_views: Optional[str] = Field(None, description="该商品近12个月的链接浏览量") circulation: Optional[str] = Field(None, description="该商品的发行量") scarcity_level: Optional[str] = Field(None, description="稀缺等级") last_market_activity: Optional[str] = Field(None, description="该商品最近一次市场活动时间") market_activity_time: Optional[str] = Field(None, description="市场活动的时间") monthly_transaction: Optional[str] = Field(None, description="月交易额") monthly_transaction_amount: Optional[str] = Field(None, description="月交易额") price_fluctuation: Optional[List[Union[str, int, float]]] = Field(None, description="该商品近30天价格波动区间") price_range: Optional[Dict[str, Union[int, float]]] = Field(None, description="资产商品的价格波动率") # 未使用 market_price: Optional[Union[int, float]] = Field(None, description="市场价格（单位：万元）") # 未使用 credit_code_or_id: Optional[str] = Field(None, description="统一社会信用代码或身份证号") biz_intro: Optional[str] = Field(None, description="业务/传承介绍") # 内置API计算字段 infringement_record: Optional[str] = Field(None, description="侵权记录") patent_count: Optional[str] = Field(None, description="专利使用量") esg_value: Optional[str] = Field(None, description="ESG关联价值") policy_matching: Optional[str] = Field(None, description="政策匹配度") online_course_views: Optional[int] = Field(None, description="线上课程点击量") pattern_complexity: Optional[str] = Field(None, description="结构复杂度") normalized_entropy: Optional[str] = Field(None, description="归一化信息熵") legal_risk: Optional[str] = Field(None, description="法律风险-侵权诉讼历史") base_pledge_rate: Optional[str] = Field(None, description="基础质押率") flow_correction: Optional[str] = Field(None, description="流量修正系数") # 计算结果字段 model_value_b: Optional[float] = Field(None, description="模型估值B（万元）") market_value_c: Optional[float] = Field(None, description="市场估值C（万元）") final_value_ab: Optional[float] = Field(None, description="最终估值AB（万元）") dynamic_pledge_rate: Optional[float] = Field(None, description="动态质押率") calculation_result: Optional[Dict[str, Any]] = Field(None, description="完整计算结果JSON") calculation_input: Optional[Dict[str, Any]] = Field(None, description="计算输入参数JSON") class ValuationAssessmentCreate(ValuationAssessmentBase): """创建估值评估模型""" pass class ValuationAssessmentUpdate(BaseModel): """更新估值评估模型""" # 基础信息 asset_name: Optional[str] = Field(None, description="资产名称") institution: Optional[str] = Field(None, description="所属机构") industry: Optional[str] = Field(None, description="所属行业") # 财务状况 annual_revenue: Optional[str] = Field(None, description="近12个月机构营收/万元") rd_investment: Optional[str] = Field(None, description="近12个月机构研发投入/万元") three_year_income: Optional[List[Any]] = Field(None, description="近三年机构收益/万元") funding_status: Optional[str] = Field(None, description="资产受资助情况") # 非遗等级与技术 inheritor_level: Optional[str] = Field(None, description="非遗传承人等级") inheritor_ages: Optional[List[int]] = Field(None, description="传承人年龄列表") inheritor_age_count: Optional[List[Any]] = Field(None, description="非遗传承人年龄水平及数量") inheritor_certificates: Optional[List[Any]] = Field(None, description="非遗传承人等级证书") heritage_level: Optional[str] = Field(None, description="非遗等级") heritage_asset_level: Optional[str] = Field(None, description="非遗资产等级") patent_application_no: Optional[str] = Field(None, description="非遗资产所用专利的申请号") patent_remaining_years: Optional[str] = Field(None, description="专利剩余年限") historical_evidence: Optional[Dict[str, int]] = Field(None, description="非遗资产历史证明证据及数量") patent_certificates: Optional[List[Any]] = Field(None, description="非遗资产所用专利的证书") pattern_images: Optional[List[Any]] = Field(None, description="非遗纹样图片") report_url: Optional[str] = Field(None, description="评估报告URL") certificate_url: Optional[str] = Field(None, description="证书URL") # 非遗应用与推广 implementation_stage: Optional[str] = Field(None, description="非遗资产应用成熟度") application_maturity: Optional[str] = Field(None, description="非遗资产应用成熟度") application_coverage: Optional[str] = Field(None, description="非遗资产应用覆盖范围") coverage_area: Optional[str] = Field(None, description="应用覆盖范围") cooperation_depth: Optional[str] = Field(None, description="非遗资产跨界合作深度") collaboration_type: Optional[str] = Field(None, description="跨界合作类型") offline_activities: Optional[str] = Field(None, description="近12个月线下相关宣讲活动次数") offline_teaching_count: Optional[int] = Field(None, description="近12个月线下相关演讲活动次数") online_accounts: Optional[List[Any]] = Field(None, description="线上相关宣传账号信息") platform_accounts: Optional[Dict[str, Dict[str, Union[str, int]]]] = Field(None, description="线上相关宣传账号信息") # 非遗资产衍生商品信息 sales_volume: Optional[str] = Field(None, description="该商品近12个月销售量") link_views: Optional[str] = Field(None, description="该商品近12个月的链接浏览量") circulation: Optional[str] = Field(None, description="该商品的发行量") scarcity_level: Optional[str] = Field(None, description="稀缺等级") last_market_activity: Optional[str] = Field(None, description="该商品最近一次市场活动时间") market_activity_time: Optional[str] = Field(None, description="市场活动的时间") monthly_transaction: Optional[str] = Field(None, description="月交易额") monthly_transaction_amount: Optional[str] = Field(None, description="月交易额") price_fluctuation: Optional[List[Union[str, int, float]]] = Field(None, description="该商品近30天价格波动区间") price_range: Optional[Dict[str, Union[int, float]]] = Field(None, description="资产商品的价格波动率") market_price: Optional[Union[int, float]] = Field(None, description="市场价格（单位：万元）") credit_code_or_id: Optional[str] = Field(None, description="统一社会信用代码或身份证号") biz_intro: Optional[str] = Field(None, description="业务/传承介绍") # 内置API计算字段 infringement_record: Optional[str] = Field(None, description="侵权记录") patent_count: Optional[str] = Field(None, description="专利使用量") esg_value: Optional[str] = Field(None, description="ESG关联价值") policy_matching: Optional[str] = Field(None, description="政策匹配度") online_course_views: Optional[int] = Field(None, description="线上课程点击量") pattern_complexity: Optional[str] = Field(None, description="结构复杂度") normalized_entropy: Optional[str] = Field(None, description="归一化信息熵") legal_risk: Optional[str] = Field(None, description="法律风险-侵权诉讼历史") base_pledge_rate: Optional[str] = Field(None, description="基础质押率") flow_correction: Optional[str] = Field(None, description="流量修正系数") # 计算结果字段 model_value_b: Optional[float] = Field(None, description="模型估值B（万元）") market_value_c: Optional[float] = Field(None, description="市场估值C（万元）") final_value_ab: Optional[float] = Field(None, description="最终估值AB（万元）") dynamic_pledge_rate: Optional[float] = Field(None, description="动态质押率") calculation_result: Optional[Dict[str, Any]] = Field(None, description="完整计算结果JSON") calculation_input: Optional[Dict[str, Any]] = Field(None, description="计算输入参数JSON") # 系统字段 status: Optional[str] = Field(None, description="评估状态: pending(待审核), success(已通过), fail(已拒绝)") admin_notes: Optional[str] = Field(None, description="管理员备注") is_active: Optional[bool] = Field(None, description="是否激活") @field_validator('report_url', 'certificate_url', mode='before') @classmethod def _coerce_url(cls, v): if v is None: return None if isinstance(v, list) and v: v = v[0] if isinstance(v, str): s = v.strip() if s.startswith('`') and s.endswith('`'): s = s[1:-1].strip() return s return v class ValuationAssessmentOut(ValuationAssessmentBase): """估值评估输出模型""" id: int = Field(..., description="主键ID") user_id: int = Field(..., description="用户ID") user_phone: Optional[str] = Field(None, description="用户手机号") report_url: List[str] = Field(default_factory=list, description="评估报告URL列表") certificate_url: List[str] = Field(default_factory=list, description="证书URL列表") report_download_urls: List[str] = Field(default_factory=list, description="评估报告下载地址列表") certificate_download_urls: List[str] = Field(default_factory=list, description="证书下载地址列表") status: str = Field(..., description="评估状态") admin_notes: Optional[str] = Field(None, description="管理员备注") created_at: datetime = Field(..., description="创建时间") updated_at: datetime = Field(..., description="更新时间") audited_at: Optional[datetime] = Field(None, description="审核时间") is_active: bool = Field(..., description="是否激活") class Config: from_attributes = True json_encoders = { datetime: lambda v: v.isoformat() } # 确保所有字段都被序列化，包括None值 exclude_none = False @field_validator('report_url', 'certificate_url', mode='before') @classmethod def _to_list(cls, v): def clean(s: str) -> str: s = s.strip() if s.startswith('`') and s.endswith('`'): s = s[1:-1].strip() return s if v is None: return [] if isinstance(v, list): return [clean(str(i)) for i in v if i is not None and str(i).strip() != ""] if isinstance(v, str): s = clean(v) return [s] if s else [] return [] @model_validator(mode='after') def _fill_downloads(self): self.report_download_urls = list(self.report_url or []) self.certificate_download_urls = list(self.certificate_url or []) return self # 用户端专用模式 class UserValuationCreate(ValuationAssessmentBase): """用户端创建估值评估模型""" pass class UserValuationOut(ValuationAssessmentBase): """用户端估值评估输出模型""" id: int = Field(..., description="主键ID") user_id: Optional[int] = Field(None, description="用户ID") report_url: List[str] = Field(default_factory=list, description="评估报告URL列表") certificate_url: List[str] = Field(default_factory=list, description="证书URL列表") report_download_urls: List[str] = Field(default_factory=list, description="评估报告下载地址列表") certificate_download_urls: List[str] = Field(default_factory=list, description="证书下载地址列表") status: str = Field(..., description="评估状态") admin_notes: Optional[str] = Field(None, description="管理员备注") created_at: datetime = Field(..., description="创建时间") updated_at: datetime = Field(..., description="更新时间") is_active: Optional[bool] = Field(None, description="是否激活") class Config: from_attributes = True json_encoders = { datetime: lambda v: v.isoformat() } exclude_none = False @field_validator('report_url', 'certificate_url', mode='before') @classmethod def _to_list(cls, v): def clean(s: str) -> str: s = s.strip() if s.startswith('`') and s.endswith('`'): s = s[1:-1].strip() return s if v is None: return [] if isinstance(v, list): return [clean(str(i)) for i in v if i is not None and str(i).strip() != ""] if isinstance(v, str): s = clean(v) return [s] if s else [] return [] @model_validator(mode='after') def _fill_downloads(self): self.report_download_urls = list(self.report_url or []) self.certificate_download_urls = list(self.certificate_url or []) return self class UserValuationDetail(ValuationAssessmentBase): """用户端详细估值评估模型""" id: int = Field(..., description="主键ID") report_url: List[str] = Field(default_factory=list, description="评估报告URL列表") certificate_url: List[str] = Field(default_factory=list, description="证书URL列表") report_download_urls: List[str] = Field(default_factory=list, description="评估报告下载地址列表") certificate_download_urls: List[str] = Field(default_factory=list, description="证书下载地址列表") status: str = Field(..., description="评估状态") admin_notes: Optional[str] = Field(None, description="管理员备注") created_at: datetime = Field(..., description="创建时间") updated_at: datetime = Field(..., description="更新时间") class Config: from_attributes = True json_encoders = { datetime: lambda v: v.isoformat() } @field_validator('report_url', 'certificate_url', mode='before') @classmethod def _to_list(cls, v): def clean(s: str) -> str: s = s.strip() if s.startswith('`') and s.endswith('`'): s = s[1:-1].strip() return s if v is None: return [] if isinstance(v, list): return [clean(str(i)) for i in v if i is not None and str(i).strip() != ""] if isinstance(v, str): s = clean(v) return [s] if s else [] return [] @model_validator(mode='after') def _fill_downloads(self): self.report_download_urls = list(self.report_url or []) self.certificate_download_urls = list(self.certificate_url or []) return self class UserValuationList(BaseModel): """用户端估值评估列表模型""" items: List[UserValuationOut] = Field(..., description="估值评估列表") total: int = Field(..., description="总数量") page: int = Field(..., description="当前页码") size: int = Field(..., description="每页数量") pages: int = Field(..., description="总页数") class Config: from_attributes = True json_encoders = { datetime: lambda v: v.isoformat() } exclude_none = False class UserValuationQuery(BaseModel): """用户端估值评估查询模型""" status: Optional[str] = Field(None, description="评估状态") asset_name: Optional[str] = Field(None, description="资产名称") page: int = Field(1, ge=1, description="页码") size: int = Field(10, ge=1, le=100, description="每页数量") class ValuationAssessmentList(BaseModel): """估值评估列表模型""" items: List[ValuationAssessmentOut] = Field(..., description="估值评估列表") total: int = Field(..., description="总数量") page: int = Field(..., description="当前页码") size: int = Field(..., description="每页数量") pages: int = Field(..., description="总页数") class ValuationAssessmentQuery(BaseModel): """估值评估查询模型""" asset_name: Optional[str] = Field(None, description="资产名称") institution: Optional[str] = Field(None, description="所属机构") industry: Optional[str] = Field(None, description="所属行业") heritage_level: Optional[str] = Field(None, description="非遗等级") status: Optional[str] = Field(None, description="评估状态: pending(待审核), success(已通过), rejected(已拒绝)") is_active: Optional[bool] = Field(None, description="是否激活") phone: Optional[str] = Field(None, description="手机号模糊查询") submitted_start: Optional[str] = Field(None, description="提交时间开始（毫秒时间戳或ISO字符串）") submitted_end: Optional[str] = Field(None, description="提交时间结束（毫秒时间戳或ISO字符串）") audited_start: Optional[str] = Field(None, description="审核时间开始（证书修改时间，毫秒时间戳或ISO字符串）") audited_end: Optional[str] = Field(None, description="审核时间结束（证书修改时间，毫秒时间戳或ISO字符串）") page: int = Field(1, ge=1, description="页码") size: int = Field(10, ge=1, le=100, description="每页数量") # 管理端审核相关模型 class ValuationApprovalRequest(BaseModel): """估值评估审核请求模型""" admin_notes: Optional[str] = Field(None, description="管理员备注") class ValuationAdminNotesUpdate(BaseModel): """管理员备注更新模型""" admin_notes: str = Field(..., description="管理员备注") class ValuationCalculationStepBase(BaseModel): """估值计算步骤基础模型""" step_order: Decimal = Field(..., description="步骤顺序") step_name: str = Field(..., description="步骤名称") step_description: Optional[str] = Field(None, description="步骤描述") input_params: Optional[Dict[str, Any]] = Field(None, description="输入参数") output_result: Optional[Dict[str, Any]] = Field(None, description="输出结果") status: str = Field(..., description="步骤状态: processing/completed/failed") error_message: Optional[str] = Field(None, description="错误信息") formula_code: Optional[str] = Field(None, description="公式编码") formula_name: Optional[str] = Field(None, description="公式名称") formula_text: Optional[str] = Field(None, description="公式说明") parent_formula_code: Optional[str] = Field(None, description="父级公式编码") group_code: Optional[str] = Field(None, description="分组编码") @field_validator('step_order', mode='before') @classmethod def _coerce_step_order(cls, v): if isinstance(v, Decimal): return v if isinstance(v, (int, float, str)): try: return Decimal(str(v)) except Exception: raise ValueError('Invalid step_order') raise ValueError('Invalid step_order type') class ValuationCalculationStepCreate(ValuationCalculationStepBase): """创建估值计算步骤模型""" valuation_id: int = Field(..., description="关联的估值评估ID") class ValuationCalculationStepOut(ValuationCalculationStepBase): """估值计算步骤输出模型""" id: int = Field(..., description="主键ID") valuation_id: int = Field(..., description="关联的估值评估ID") created_at: datetime = Field(..., description="创建时间") updated_at: datetime = Field(..., description="更新时间") class Config: from_attributes = True json_encoders = { datetime: lambda v: v.isoformat(), Decimal: lambda v: float(v) }