433 lines
23 KiB
Python
433 lines
23 KiB
Python
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(待审核), approved(已通过), 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)
|
||
}
|