zfc/guzhi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
guzhi/app/schemas/valuation.py
邹方成 b10c357a56 feat(valuation): 添加update_calc方法并更新相关字段类型 
为ValuationController添加update_calc方法用于计算更新
将updated_at等字段改为Optional类型
修复heritage_level字段获取方式
更新docker镜像版本至v2.7
2025-12-02 18:50:46 +08:00

433 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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: Optional[datetime] = Field(None, 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: Optional[datetime] = Field(None, 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: Optional[datetime] = Field(None, 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: Optional[datetime] = Field(None, description="更新时间")
class Config:
from_attributes = True
json_encoders = {
datetime: lambda v: v.isoformat(),
Decimal: lambda v: float(v)
}
Reference in New Issue View Git Blame Copy Permalink