zfc/guzhi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' of https://git.1024tool.vip/zfc/guzhi

Browse Source
This commit is contained in:
若拙_233 2025-10-10 15:48:42 +08:00
commit 42b5ce887c
27 changed files with 5230 additions and 250 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
aaa.json Normal file
View File

@ -0,0 +1,83 @@
{
"asset_name": "资产名称",
"institution": "所属机构",
"industry": "农业",
"annual_revenue": "22",
"rd_investment": "33",
"three_year_income": [
"11",
"22",
"33"
],
"funding_status": "国家级资助",
"sales_volume": "22",
"link_views": "22",
"circulation": "0",
"last_market_activity": "0",
"monthly_transaction": "0",
"price_fluctuation": [
"2",
"3"
],
"application_maturity": "0",
"application_coverage": "0",
"cooperation_depth": "1",
"offline_activities": "3",
"online_accounts": [
"0",
"333"
],
"inheritor_level": "国家级传承人",
"inheritor_age_count": [
"55",
"66",
"77"
],
"inheritor_certificates": [
"http://example.com/国家级非遗传承人证书.jpg"
],
"heritage_level": "0",
"historical_evidence": {
"artifacts": "22",
"ancient_literature": "33",
"inheritor_testimony": "66"
},
"patent_certificates": [
"http://example.com/专利证书1.jpg",
"http://example.com/专利证书2.jpg"
],
"pattern_images": [
"pattern1.jpg"
],
"patent_application_no": "22",
"heritage_asset_level": "国家级非遗",
"inheritor_ages": [
"55",
"66",
"77"
],
"implementation_stage": "成熟应用",
"coverage_area": "全球覆盖",
"collaboration_type": "品牌联名",
"platform_accounts": {
"bilibili": {
"followers_count": 8000,
"likes": 1000,
"comments": 500,
"shares": 500
},
"douyin": {
"followers_count": 8000,
"likes": 1000,
"comments": 500,
"shares": 500
}
},
"scarcity_level": "孤品：全球唯一，不可复制（如特定版权、唯一实物）",
"market_activity_time": "近一周",
"price_range": {
"highest": "2",
"lowest": "3"
},
"monthly_transaction_amount": "月交易额<100万元"
}

2
app/__init__.py
View File

@ -35,7 +35,7 @@ def create_app() -> FastAPI:
lifespan=lifespan,
)
# 注册静态文件目录
app.mount("/static", StaticFiles(directory="app/static"), name="static")
# app.mount("/static", StaticFiles(directory="app/static"), name="static")
register_exceptions(app)
register_routers(app, prefix="/api")
return app

4
app/api/v1/__init__.py
View File

@ -24,7 +24,9 @@ v1_router = APIRouter()
v1_router.include_router(base_router, prefix="/base")
v1_router.include_router(app_users_router, prefix="/app-user") # AppUser路由，无需权限依赖
v1_router.include_router(app_valuations_router, prefix="/app-valuations") # 用户端估值评估路由，需要认证
# 注意：app-valuations 路由在各自的端点内部使用 get_current_app_user 进行认证
# 这样可以保持App用户认证系统的独立性，不与后台管理权限系统混合
v1_router.include_router(app_valuations_router, prefix="/app-valuations") # 用户端估值评估路由
v1_router.include_router(users_router, prefix="/user", dependencies=[DependAuth, DependPermission])
v1_router.include_router(roles_router, prefix="/role", dependencies=[DependAuth, DependPermission])
v1_router.include_router(menus_router, prefix="/menu", dependencies=[DependAuth, DependPermission])

74
app/api/v1/app_valuations/app_valuations.py
View File

@ -9,8 +9,10 @@ from app.schemas.valuation import (
UserValuationOut,
UserValuationDetail
)
from app.models.user import AppUser
from app.schemas.base import Success, SuccessExtra
from app.utils.app_user_jwt import get_current_app_user_id
from app.utils.app_user_jwt import get_current_app_user
app_valuations_router = APIRouter(tags=["用户端估值评估"])
@ -18,19 +20,70 @@ app_valuations_router = APIRouter(tags=["用户端估值评估"])
@app_valuations_router.post("/", summary="创建估值评估")
async def create_valuation(
data: UserValuationCreate,
user_id: int = Depends(get_current_app_user_id)
current_user: AppUser = Depends(get_current_app_user)
):
"""
用户创建估值评估申请
"""
try:
result = await user_valuation_controller.create_valuation(
user_id=user_id,
user_id=current_user.id,
data=data
)
# 使用model_dump_json()来正确序列化datetime，然后解析为dict
import json
result_dict = json.loads(result.model_dump_json())
# 开始计算 估值 信息
# 1 # 经济价值B1模块: EconomicValueB1Calculator | BasicValueB11Calculator | TrafficFactorB12Calculator | PolicyMultiplierB13Calculator
# 1.1 EconomicValueB1Calculator
# input_data = {
# # 基础价值B11相关参数
# 'three_year_income': data.three_year_income,
# 'patent_score': data.pa, # 专利分
# 'popularity_score': data.popularity_score, # 普及地域分值
# 'infringement_score': data.infringement_score, # 侵权分
# 'innovation_ratio': data.innovation_ratio,
# 'esg_score':data.esg_score,
# 'industry_coefficient':data.industry_coefficient,
# # 流量因子B12相关参数
# 'search_index_s1': 4500.0,
# 'industry_average_s2': 5000.0,
# # 'social_media_spread_s3': social_media_spread_s3,
# 'likes': 4, # 点赞
# 'comments': 5, # 评论
# 'shares': 6, # 转发
# 'followers': 7, # 粉丝数
# 'click_count': 1000,# 点击量
# 'view_count': 100, # 内容浏览量
# # 政策乘数B13相关参数
# 'policy_match_score': 10.0, # 政策匹配度
# 'implementation_stage': 10.0, # 实施阶段评分
# 'funding_support': 10.0 # 资金支持度
# }
# 1.2 BasicValueB11Calculator
# 1.3 TrafficFactorB12Calculator
# 1.4 PolicyMultiplierB13Calculator
# 2 # 文化价值B2模块: CulturalValueB2Calculator | LivingHeritageB21Calculator | PatternGeneB22Calculator
# 2.1 CulturalValueB2Calculator
# 2.2 LivingHeritageB21Calculator
# 2.3 PatternGeneB22Calculator
# 3 # 风险调整系数B3模块: RiskAdjustmentB3Calculator
# 3.1 RiskAdjustmentB3Calculator
# 4 # 市场估值C模块: MarketValueCCalculator | MarketBiddingC1Calculator | HeatCoefficientC2Calculator | ScarcityMultiplierC3Calculator | TemporalDecayC4Calculator
# 4.1 MarketValueCCalculator
# 4.2 MarketBiddingC1Calculator
# 4.3 HeatCoefficientC2Calculator
# 4.4 ScarcityMultiplierC3Calculator
# 4.5 TemporalDecayC4Calculator
# 5 # 最终估值A模块: FinalValueACalculator
# 5.1 FinalValueACalculator
return Success(data=result_dict, msg="估值评估申请提交成功")
except Exception as e:
raise HTTPException(
@ -42,14 +95,15 @@ async def create_valuation(
@app_valuations_router.get("/", summary="获取我的估值评估列表")
async def get_my_valuations(
query: UserValuationQuery = Depends(),
user_id: int = Depends(get_current_app_user_id)
current_user: AppUser = Depends(get_current_app_user)
):
"""
获取当前用户的估值评估列表
"""
try:
result = await user_valuation_controller.get_user_valuations(
user_id=user_id,
user_id=current_user.id,
query=query
)
# 使用model_dump_json()来正确序列化datetime，然后解析为dict列表
@ -59,7 +113,7 @@ async def get_my_valuations(
data=serialized_items,
total=result.total,
page=result.page,
size=result.size,
page_size=result.size,
pages=result.pages,
msg="获取估值评估列表成功"
)
@ -73,14 +127,14 @@ async def get_my_valuations(
@app_valuations_router.get("/{valuation_id}", summary="获取估值评估详情")
async def get_valuation_detail(
valuation_id: int,
user_id: int = Depends(get_current_app_user_id)
current_user: AppUser = Depends(get_current_app_user)
):
"""
获取指定估值评估的详细信息
"""
try:
result = await user_valuation_controller.get_user_valuation_detail(
user_id=user_id,
user_id=current_user.id,
valuation_id=valuation_id
)
@ -105,14 +159,14 @@ async def get_valuation_detail(
@app_valuations_router.get("/statistics/overview", summary="获取我的估值评估统计")
async def get_my_valuation_statistics(
user_id: int = Depends(get_current_app_user_id)
current_user: AppUser = Depends(get_current_app_user)
):
"""
获取当前用户的估值评估统计信息
"""
try:
result = await user_valuation_controller.get_user_valuation_statistics(
user_id=user_id
user_id=current_user.id
)
return Success(data=result, msg="获取统计信息成功")
except Exception as e:

67
app/api/v1/calculation/calcuation.py
View File

@ -15,6 +15,7 @@ from app.schemas.valuation import (
from app.schemas.base import Success, SuccessExtra
from app.utils.app_user_jwt import get_current_app_user_id
from app.utils.calculation_engine import FinalValueACalculator
from app.utils.calculation_engine.drp import DynamicPledgeRateCalculator
from app.utils.calculation_engine.economic_value_b1.sub_formulas.basic_value_b11 import calculate_popularity_score, \
calculate_infringement_score, calculate_patent_usage_score
@ -40,6 +41,64 @@ async def calculate_valuation(
计算估值评估
根据用户提交的估值评估数据，调用计算引擎进行经济价值B1计算
请求示例JSON (仅包含用户填写部分):
{
"asset_name": "传统刺绣工艺",
"institution": "某文化传承机构",
"industry": "传统手工艺",
// 财务状况 (用户填写)
"annual_revenue": "500", // 近12个月机构营收/万元
"rd_investment": "50", // 近12个月机构研发投入/万元
"three_year_income": [400, 450, 500], // 近三年机构收益/万元
// 非遗等级与技术 (用户填写)
"inheritor_level": "国家级传承人", // 非遗传承人等级
"inheritor_ages": [45, 60, 75], // 传承人年龄列表
"heritage_level": "国家级", // 非遗等级
"patent_application_no": "CN202310123456.7", // 专利申请号
"patent_remaining_years": "15", // 专利剩余年限
"historical_evidence": { // 历史证明证据及数量
"历史文献": 3,
"考古发现": 2,
"传承谱系": 5
},
"pattern_images": ["demo.jpg"], // 非遗纹样图片
// 非遗应用与推广 (用户填写)
"application_maturity": "成熟应用", // 应用成熟度
"application_coverage": "全国覆盖", // 应用覆盖范围
"cooperation_depth": "0.5", // 跨界合作深度
"offline_activities": "12", // 近12个月线下宣讲活动次数
"online_accounts": [ // 线上宣传账号信息
{"platform": "抖音", "account": "传统刺绣大师"},
{"platform": "微博", "account": "非遗传承人"}
],
// 市场信息 (用户填写)
"sales_volume": "1000", // 近12个月销售量
"link_views": "5000", // 近12个月链接浏览量
"circulation": "限量", // 发行量
"last_market_activity": "2024-01-15", // 最近一次市场活动时间
"price_fluctuation": [95.0, 105.0], // 近30天价格波动区间
"manual_bids": [48000.0, 50000.0, 52000.0], // 手动收集的竞价列表
// 政策相关 (用户填写)
"funding_status": "国家级资助", // 资金支持情况
"implementation_stage": "成熟应用" // 实施阶段
}
API获取参数 (系统自动获取，无需用户填写):
- 搜索指数: 百度、微信、微博搜索指数
- 社交媒体数据: 点赞数、评论数、转发数、粉丝数
- 交易数据: 近3个月加权平均价格
- 热度数据: 近7日日均浏览量、收藏数
- ESG评分: 根据行业自动匹配
- 行业系数: 根据行业ROE计算
- 政策匹配度: 根据行业自动匹配
- 专利验证: 通过API验证专利有效性
- 侵权记录: 通过API查询侵权诉讼历史
"""
try:
start_ts = time.monotonic()
@ -126,6 +185,14 @@ async def calculate_valuation(
# 计算最终估值A（统一计算）
calculation_result = calculator.calculate_complete_final_value_a(input_data)
# 计算动态质押
drp_c = DynamicPledgeRateCalculator()
'''
monthly_amount (float): 月交易额（万元）
heritage_level (str): 非遗等级
'''
drp_result = drp_c.calculate_dynamic_pledge_rate(float(data.monthly_transaction_amount),data.heritage_asset_level)
# 结构化日志：关键分值
try:
duration_ms = int((time.monotonic() - start_ts) * 1000)

21
app/api/v1/third_party_api/third_party_api.py
View File

@ -11,6 +11,7 @@ from app.schemas.third_party_api import (
OCRRequest,
XiaohongshuNoteRequest,
JizhiliaoSearchRequest,
DifyWorkflowRequest,
APIResponse,
APIProviderInfo,
APIEndpointInfo,
@ -110,4 +111,22 @@ async def search_jizhiliao_index(request: JizhiliaoSearchRequest):
return Fail(message=result.message)
except Exception as e:
logger.error(f"极致聊指数搜索失败: {e}")
return Fail(message=f"极致聊指数搜索失败: {str(e)}")
return Fail(message=f"极致聊指数搜索失败: {str(e)}")
@router.post("/dify/workflows/run", summary="运行Dify工作流")
async def run_dify_workflow(request: DifyWorkflowRequest):
"""运行Dify工作流"""
try:
result = await third_party_api_controller.run_dify_workflow(
zl_img=request.zl_img,
response_mode=request.response_mode,
user=request.user
)
if result.success:
return Success(data=result.data, message=result.message)
else:
return Fail(message=result.message)
except Exception as e:
logger.error(f"运行Dify工作流失败: {e}")
return Fail(message=f"运行Dify工作流失败: {str(e)}")

2
app/api/v1/upload/__init__.py
View File

@ -2,4 +2,4 @@ from fastapi import APIRouter
from .upload import router as upload_router
router = APIRouter()
router.include_router(upload_router, prefix="/upload", tags=["文件上传"])
router.include_router(upload_router, tags=["文件上传"])

11
app/api/v1/valuations/valuations.py
View File

@ -21,7 +21,10 @@ valuations_router = APIRouter(tags=["估值评估"])
async def create_valuation(data: ValuationAssessmentCreate):
"""创建新的估值评估记录"""
try:
result = await valuation_controller.create(data)
# 获取当前用户ID
user_id = CTX_USER_ID.get()
print(user_id)
result = await valuation_controller.create(data, user_id)
return Success(data=result, msg="创建成功")
except Exception as e:
raise HTTPException(status_code=400, detail=f"创建失败: {str(e)}")
@ -88,9 +91,9 @@ async def get_valuations(
data=result.items,
total=result.total,
page=result.page,
size=result.size,
page_size=result.size,
pages=result.pages,
msg="获取成功"
msg="获取估值评估列表成功"
)
@ -106,7 +109,7 @@ async def search_valuations(
data=result.items,
total=result.total,
page=result.page,
size=result.size,
page_size=result.size,
pages=result.pages,
msg="搜索成功"
)

20
app/controllers/third_party_api.py
View File

@ -78,9 +78,9 @@ class ThirdPartyAPIController:
endpoint="patent",
params={
"searchKey": company_name,
"pageNo": 1,
"range": 100,
"searchType": 0,
"pageNo": "1",
"range": "100",
"searchType": '1',
"ChinazVer": chinaz_ver
}
)
@ -125,6 +125,20 @@ class ThirdPartyAPIController:
timeout=timeout
)
async def run_dify_workflow(
self,
zl_img: str,
response_mode: str = "blocking",
user: str = "default_user"
) -> APIResponse:
"""运行Dify工作流"""
params = {
"inputs": {"zl_img": zl_img},
"response_mode": response_mode,
"user": user
}
return await self.make_api_request("dify", "workflows_run", params)
# 创建全局控制器实例
third_party_api_controller = ThirdPartyAPIController()

5
app/controllers/upload.py
View File

@ -3,6 +3,7 @@ from pathlib import Path
from typing import List
from fastapi import UploadFile
from app.schemas.upload import ImageUploadResponse
from app.settings.config import settings
class UploadController:
"""文件上传控制器"""
@ -44,8 +45,8 @@ class UploadController:
with open(file_path, "wb") as f:
f.write(content)
# 返回文件URL
# 返回完整的可访问URL
return ImageUploadResponse(
url=f"/static/images/{filename}",
url=f"{settings.BASE_URL}/static/images/{filename}",
filename=filename
)

34
app/controllers/user_valuation.py
View File

@ -86,16 +86,7 @@ class UserValuationController:
async def _to_user_out(self, valuation: ValuationAssessment) -> UserValuationOut:
"""转换为用户端输出模型"""
return UserValuationOut(
id=valuation.id,
asset_name=valuation.asset_name,
institution=valuation.institution,
industry=valuation.industry,
status=valuation.status,
admin_notes=valuation.admin_notes,
created_at=valuation.created_at,
updated_at=valuation.updated_at
)
return UserValuationOut.model_validate(valuation)
async def _to_user_detail(self, valuation: ValuationAssessment) -> UserValuationDetail:
"""转换为用户端详细模型"""
@ -109,24 +100,47 @@ class UserValuationController:
three_year_income=valuation.three_year_income,
funding_status=valuation.funding_status,
inheritor_level=valuation.inheritor_level,
inheritor_ages=valuation.inheritor_ages,
inheritor_age_count=valuation.inheritor_age_count,
inheritor_certificates=valuation.inheritor_certificates,
heritage_level=valuation.heritage_level,
heritage_asset_level=valuation.heritage_asset_level,
patent_application_no=valuation.patent_application_no,
patent_remaining_years=valuation.patent_remaining_years,
historical_evidence=valuation.historical_evidence,
patent_certificates=valuation.patent_certificates,
pattern_images=valuation.pattern_images,
application_maturity=valuation.application_maturity,
implementation_stage=valuation.implementation_stage,
application_coverage=valuation.application_coverage,
coverage_area=valuation.coverage_area,
cooperation_depth=valuation.cooperation_depth,
collaboration_type=valuation.collaboration_type,
offline_activities=valuation.offline_activities,
offline_teaching_count=valuation.offline_teaching_count,
online_accounts=valuation.online_accounts,
platform_accounts=valuation.platform_accounts,
sales_volume=valuation.sales_volume,
link_views=valuation.link_views,
circulation=valuation.circulation,
scarcity_level=valuation.scarcity_level,
last_market_activity=valuation.last_market_activity,
market_activity_time=valuation.market_activity_time,
monthly_transaction=valuation.monthly_transaction,
monthly_transaction_amount=valuation.monthly_transaction_amount,
price_fluctuation=valuation.price_fluctuation,
price_range=valuation.price_range,
market_price=valuation.market_price,
infringement_record=valuation.infringement_record,
patent_count=valuation.patent_count,
esg_value=valuation.esg_value,
policy_matching=valuation.policy_matching,
online_course_views=valuation.online_course_views,
pattern_complexity=valuation.pattern_complexity,
normalized_entropy=valuation.normalized_entropy,
legal_risk=valuation.legal_risk,
base_pledge_rate=valuation.base_pledge_rate,
flow_correction=valuation.flow_correction,
status=valuation.status,
admin_notes=valuation.admin_notes,
created_at=valuation.created_at,

8
app/controllers/valuation.py
View File

@ -18,9 +18,12 @@ class ValuationController:
model = ValuationAssessment
async def create(self, data: ValuationAssessmentCreate) -> ValuationAssessmentOut:
async def create(self, data: ValuationAssessmentCreate, user_id: int) -> ValuationAssessmentOut:
"""创建估值评估"""
valuation = await self.model.create(**data.model_dump())
# 将用户ID添加到数据中
create_data = data.model_dump()
create_data['user_id'] = user_id
valuation = await self.model.create(**create_data)
return ValuationAssessmentOut.model_validate(valuation)
async def get_by_id(self, valuation_id: int) -> Optional[ValuationAssessmentOut]:
@ -66,7 +69,6 @@ class ValuationController:
# 转换为输出模型
items = [ValuationAssessmentOut.model_validate(v) for v in valuations]
# 计算总页数
pages = (total + query.size - 1) // query.size

1
app/core/init_app.py
View File

@ -46,6 +46,7 @@ def make_middlewares():
"/api/v1/base/access_token",
"/docs",
"/openapi.json",
"/static", # 排除静态文件路径
],
),
]

35
app/core/middlewares.py
View File

@ -62,22 +62,31 @@ class HttpAuditLogMiddleware(BaseHTTPMiddleware):
# 获取请求体
if request.method in ["POST", "PUT", "PATCH"]:
try:
body = await request.json()
args.update(body)
except json.JSONDecodeError:
# 检查Content-Type来决定如何解析请求体
content_type = request.headers.get("content-type", "")
if "multipart/form-data" in content_type or "application/x-www-form-urlencoded" in content_type:
# 处理表单数据（包括文件上传）
try:
body = await request.form()
# args.update(body)
for k, v in body.items():
if hasattr(v, "filename"): # 文件上传行为
args[k] = v.filename
elif isinstance(v, list) and v and hasattr(v[0], "filename"):
args[k] = [file.filename for file in v]
else:
args[k] = v
# 对于文件上传，不要在中间件中消费request.form()
# 因为这会导致FastAPI无法再次读取请求体
pass
except Exception:
pass
elif "application/json" in content_type:
# 处理JSON数据
try:
body = await request.json()
args.update(body)
except (json.JSONDecodeError, UnicodeDecodeError):
pass
else:
# 尝试解析为JSON，如果失败则跳过
try:
body = await request.json()
args.update(body)
except (json.JSONDecodeError, UnicodeDecodeError):
pass
return args

27
app/models/valuation.py
View File

@ -20,28 +20,53 @@ class ValuationAssessment(Model):
# 非遗等级与技术
inheritor_level = fields.CharField(max_length=50, null=True, description="非遗传承人等级")
inheritor_ages = fields.JSONField(null=True, description="传承人年龄列表")
inheritor_age_count = fields.JSONField(null=True, description="非遗传承人年龄水平及数量")
inheritor_certificates = fields.JSONField(null=True, description="非遗传承人等级证书")
heritage_level = fields.CharField(max_length=50, null=True, description="非遗等级")
heritage_asset_level = fields.CharField(max_length=50, null=True, description="非遗资产等级")
patent_application_no = fields.CharField(max_length=100, null=True, description="非遗资产所用专利的申请号")
patent_remaining_years = fields.CharField(max_length=50, null=True, description="专利剩余年限")
historical_evidence = fields.JSONField(null=True, description="非遗资产历史证明证据及数量")
patent_certificates = fields.JSONField(null=True, description="非遗资产所用专利的证书")
pattern_images = fields.JSONField(null=True, description="非遗纹样图片")
# 非遗应用与推广
application_maturity = fields.CharField(max_length=100, null=True, description="非遗资产应用成熟度")
implementation_stage = fields.CharField(max_length=100, null=True, description="非遗资产应用成熟度")
application_coverage = fields.CharField(max_length=100, null=True, description="非遗资产应用覆盖范围")
coverage_area = fields.CharField(max_length=100, null=True, description="非遗资产应用覆盖范围")
cooperation_depth = fields.CharField(max_length=100, null=True, description="非遗资产跨界合作深度")
collaboration_type = fields.CharField(max_length=100, null=True, description="非遗资产跨界合作深度")
offline_activities = fields.CharField(max_length=50, null=True, description="近12个月线下相关宣讲活动次数")
offline_teaching_count = fields.IntField(null=True, description="近12个月线下相关演讲活动次数")
online_accounts = fields.JSONField(null=True, description="线上相关宣传账号信息")
platform_accounts = fields.JSONField(null=True, description="线上相关宣传账号信息")
# 非遗资产衍生商品信息
sales_volume = fields.CharField(max_length=50, null=True, description="该商品近12个月销售量")
link_views = fields.CharField(max_length=50, null=True, description="该商品近12个月的链接浏览量")
circulation = fields.CharField(max_length=50, null=True, description="该商品的发行量")
scarcity_level = fields.CharField(max_length=50, null=True, description="稀缺等级")
last_market_activity = fields.CharField(max_length=100, null=True, description="该商品最近一次市场活动时间")
market_activity_time = fields.CharField(max_length=100, null=True, description="市场活动的时间")
monthly_transaction = fields.CharField(max_length=50, null=True, description="月交易额")
price_fluctuation = fields.CharField(max_length=100, null=True, description="该商品近30天价格波动区间")
monthly_transaction_amount = fields.CharField(max_length=50, null=True, description="月交易额")
price_fluctuation = fields.JSONField(null=True, description="该商品近30天价格波动区间")
price_range = fields.JSONField(null=True, description="资产商品的价格波动率")
market_price = fields.FloatField(null=True, description="市场价格（单位：万元）")
# 内置API计算字段
infringement_record = fields.CharField(max_length=100, null=True, description="侵权记录")
patent_count = fields.CharField(max_length=50, null=True, description="专利使用量")
esg_value = fields.CharField(max_length=50, null=True, description="ESG关联价值")
policy_matching = fields.CharField(max_length=50, null=True, description="政策匹配度")
online_course_views = fields.IntField(null=True, description="线上课程点击量")
pattern_complexity = fields.CharField(max_length=50, null=True, description="结构复杂度")
normalized_entropy = fields.CharField(max_length=50, null=True, description="归一化信息熵")
legal_risk = fields.CharField(max_length=100, null=True, description="法律风险-侵权诉讼历史")
base_pledge_rate = fields.CharField(max_length=50, null=True, description="基础质押率")
flow_correction = fields.CharField(max_length=50, null=True, description="流量修正系数")
# 系统字段
user = fields.ForeignKeyField("models.AppUser", related_name="valuations", description="提交用户")

6
app/schemas/third_party_api.py
View File

@ -28,6 +28,12 @@ class JizhiliaoSearchRequest(BaseAPIRequest):
page: int = Field(1, description="页码")
size: int = Field(10, description="每页数量")
class DifyWorkflowRequest(BaseAPIRequest):
"""Dify工作流请求模型"""
zl_img: str = Field(..., description="资料图片字段")
response_mode: str = Field("blocking", description="响应模式，blocking或streaming")
user: str = Field(..., description="用户标识")
class APIResponse(BaseModel):
"""API响应模型"""
success: bool

66
app/schemas/valuation.py
View File

@ -1,5 +1,5 @@
from datetime import datetime
from typing import List, Optional, Any
from typing import List, Optional, Any, Dict, Union
from pydantic import BaseModel, Field
@ -13,33 +13,58 @@ class ValuationAssessmentBase(BaseModel):
# 财务状况
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="近三年机构收益/万元")
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[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="非遗资产所用专利的申请号")
historical_evidence: Optional[List[Any]] = Field(None, description="非遗资产历史证明证据及数量")
patent_certificates: Optional[List[Any]] = Field(None, description="非遗资产所用专利的证书")
pattern_images: Optional[List[Any]] = 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="非遗纹样图片")
# 非遗应用与推广
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, 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="月交易额")
price_fluctuation: Optional[str] = Field(None, description="该商品近30天价格波动区间")
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="市场价格（单位：万元）")
# 内置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="流量修正系数")
class ValuationAssessmentCreate(ValuationAssessmentBase):
@ -83,13 +108,13 @@ class ValuationAssessmentUpdate(BaseModel):
circulation: Optional[str] = Field(None, description="该商品的发行量")
last_market_activity: Optional[str] = Field(None, description="该商品最近一次市场活动时间")
monthly_transaction: Optional[str] = Field(None, description="月交易额")
price_fluctuation: Optional[str] = Field(None, description="该商品近30天价格波动区间")
price_fluctuation: Optional[List[Union[str, int, float]]] = Field(None, description="该商品近30天价格波动区间")
is_active: Optional[bool] = Field(None, description="是否激活")
class ValuationAssessmentOut(ValuationAssessmentBase):
"""输出估值评估模型"""
"""估值评估输出模型"""
id: int = Field(..., description="主键ID")
user_id: int = Field(..., description="用户ID")
status: str = Field(..., description="评估状态")
@ -100,6 +125,11 @@ class ValuationAssessmentOut(ValuationAssessmentBase):
class Config:
from_attributes = True
json_encoders = {
datetime: lambda v: v.isoformat()
}
# 确保所有字段都被序列化，包括None值
exclude_none = False
# 用户端专用模式
@ -108,22 +138,22 @@ class UserValuationCreate(ValuationAssessmentBase):
pass
class UserValuationOut(BaseModel):
"""用户端输出估值评估模型"""
class UserValuationOut(ValuationAssessmentBase):
"""用户端估值评估输出模型"""
id: int = Field(..., description="主键ID")
asset_name: str = Field(..., description="资产名称")
institution: str = Field(..., description="所属机构")
industry: str = Field(..., description="所属行业")
user_id: Optional[int] = Field(None, description="用户ID")
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
class UserValuationDetail(ValuationAssessmentBase):
@ -148,6 +178,14 @@ class UserValuationList(BaseModel):
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):

5
app/settings/config.py
View File

@ -16,6 +16,11 @@ class Settings(BaseSettings):
CORS_ALLOW_HEADERS: typing.List = ["*"]
DEBUG: bool = True
# 服务器配置
SERVER_HOST: str = "127.0.0.1"
SERVER_PORT: int = 9999
BASE_URL: str = f"http://{SERVER_HOST}:{SERVER_PORT}"
PROJECT_ROOT: str = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir))
BASE_DIR: str = os.path.abspath(os.path.join(PROJECT_ROOT, os.pardir))

251
app/utils/api_config.py
View File

@ -32,7 +32,7 @@ class APIConfig:
"base_url": "https://openapi.chinaz.net",
"endpoints": {
"copyright_software": {
"api_key": os.getenv("CHINAZ_COPYRIGHT_API_KEY", "YOUR_API_KEY"),
"APIKey": os.getenv("CHINAZ_COPYRIGHT_API_KEY", "YOUR_API_KEY"),
"path": "/v1/1036/copyrightsoftware",
"method": "POST",
"description": "企业软件著作权查询",
@ -40,7 +40,7 @@ class APIConfig:
"optional_params": ["sign"]
},
"patent": {
"api_key": os.getenv("CHINAZ_PATENT_API_KEY", "YOUR_API_KEY"),
"APIKey": os.getenv("CHINAZ_PATENT_API_KEY", "apiuser_quantity_045e9c832d253a1fdfd41edd1a85a254_4e9f3a38e384414e97fdbd33cca13124"),
"path": "/v1/1036/patent",
"method": "POST",
"description": "企业专利信息查询",
@ -48,7 +48,7 @@ class APIConfig:
"optional_params": ["sign", "searchType"]
},
"judgement": {
"api_key": os.getenv("CHINAZ_JUDGEMENT_API_KEY", "YOUR_API_KEY"),
"APIKey": os.getenv("CHINAZ_JUDGEMENT_API_KEY", "YOUR_API_KEY"),
"path": "/v1/1036/judgementdetailv4",
"method": "POST",
"description": "司法综合数据查询",
@ -56,7 +56,7 @@ class APIConfig:
"optional_params": ["sign", "q", "idCardNo", "datatype", "id", "pageNo"]
},
"recognition_ocr": {
"api_key": os.getenv("CHINAZ_OCR_API_KEY", "YOUR_API_KEY"),
"APIKey": os.getenv("CHINAZ_OCR_API_KEY", "apiuser_quantity_d0848e65f7b7ae50ff6f1ae37e413586_31466987e2ef402ba3142b72680a92fc"),
"path": "/v1/1024/recognition_ocr",
"method": "POST",
"description": "图片OCR识别",
@ -72,184 +72,75 @@ class APIConfig:
"timeout": 30,
"retries": 3,
"endpoints": {
"xiaohongshu": {
"api_key": os.getenv("XIAOHONGSHU_TOKEN", "YNSbIjdU"),
"base_url": "https://api.justoneapi.com",
"timeout": 30,
"retries": 3,
"endpoints": {
"xiaohongshu_note_detail": {
"path": "/api/xiaohongshu/get-note-detail/v7",
"method": "GET",
"description": "小红书笔记详情查询",
"required_params": ["noteId"],
"optional_params": ["token"]
},
"xiaohongshu_user_info": {
"path": "/api/xiaohongshu/get-user-info/v7",
"method": "GET",
"description": "小红书用户信息查询",
"required_params": ["userId"],
"optional_params": ["token"]
},
"xiaohongshu_search_notes": {
"path": "/api/xiaohongshu/search-notes/v7",
"method": "GET",
"description": "小红书笔记搜索",
"required_params": ["keyword"],
"optional_params": ["page", "size", "token"]
},
"weibo_user_detail": {
"path": "/api/weibo/get-user-detail/v1", # 微博只有粉丝数接口
"method": "GET",
"description": "获取用户信息 包括昵称、头像、用户ID、粉丝数、关注数、简介、认证状态等公开信息",
"required_params": ["token", "uid"],
"optional_params": ["token", "uid"]
},
"weixin_user_detail": {
"path": "/api/weixin/get-user-post/v1",
"method": "GET",
"description": "公众号发布的文章内容，包括标题、作者、发布时间、内容摘要以及阅读数、点赞数与转发数等互动数据",
"required_params": ["token", "wxid"],
"optional_params": ["token", "wxid"]
},
"douyin_video_detail": {
"path": "/api/douyin/get-video-detail/v2",
"method": "GET",
"description": "该接口用于获取指定抖音视频的详细信息，包括视频地址、描述文案、作者信息、发布时间、播放量、点赞数、评论数与分享数等",
"required_params": ["token", "videoId"],
"optional_params": ["token", "videoId"]
},
"kuaishou_video_detail": {
"path": "/api/kuaishou/get-video-detail/v2",
"method": "GET",
"description": "该接口用于获取指定抖音视频的详细信息，包括视频地址、描述文案、作者信息、发布时间、播放量、点赞数、评论数与分享数等",
"required_params": ["token", "videoId"],
"optional_params": ["token", "videoId"]
},
"bilibili_video_detail": {
"path": "/api/bilibili/get-video-detail/v2",
"method": "GET",
"description": "该接口用于获取指定抖音视频的详细信息，包括视频地址、描述文案、作者信息、发布时间、播放量、点赞数、评论数与分享数等",
"required_params": ["token", "bvid"],
"optional_params": ["token", "bvid"]
}
}
"xiaohongshu_note_detail": {
"path": "/api/xiaohongshu/get-note-detail/v7",
"method": "GET",
"description": "小红书笔记详情查询",
"required_params": ["noteId"],
"optional_params": ["token"]
},
"weibo": {
"api_key": os.getenv("XIAOHONGSHU_TOKEN", "YNSbIjdU"),
"base_url": "https://api.justoneapi.com",
"timeout": 30,
"retries": 3,
"endpoints": {
"weibo_user_detail": {
"path": "/api/weibo/get-user-detail/v1", # 微博只有粉丝数接口
"method": "GET",
"description": "获取用户信息 包括昵称、头像、用户ID、粉丝数、关注数、简介、认证状态等公开信息",
"required_params": ["token", "uid"],
"optional_params": ["token", "uid"]
},
}
"xiaohongshu_user_info": {
"path": "/api/xiaohongshu/get-user-info/v7",
"method": "GET",
"description": "小红书用户信息查询",
"required_params": ["userId"],
"optional_params": ["token"]
},
"weixin": {
"api_key": os.getenv("XIAOHONGSHU_TOKEN", "YNSbIjdU"),
"base_url": "https://api.justoneapi.com",
"timeout": 30,
"retries": 3,
"endpoints": {
"weixin_user_detail": {
"path": "/api/weixin/get-user-post/v1",
"method": "GET",
"description": "公众号发布的文章内容，包括标题、作者、发布时间、内容摘要以及阅读数、点赞数与转发数等互动数据",
"required_params": ["token", "wxid"],
"optional_params": ["token", "wxid"]
}
}
"xiaohongshu_search_notes": {
"path": "/api/xiaohongshu/search-notes/v7",
"method": "GET",
"description": "小红书笔记搜索",
"required_params": ["keyword"],
"optional_params": ["page", "size", "token"]
},
"douyin": {
"api_key": os.getenv("XIAOHONGSHU_TOKEN", "YNSbIjdU"),
"base_url": "https://api.justoneapi.com",
"timeout": 30,
"retries": 3,
"endpoints": {
"douyin_video_detail": {
"path": "/api/douyin/get-video-detail/v2",
"method": "GET",
"description": "该接口用于获取指定抖音视频的详细信息，包括视频地址、描述文案、作者信息、发布时间、播放量、点赞数、评论数与分享数等",
"required_params": ["token", "videoId"],
"optional_params": ["token", "videoId"]
}
}
"weibo_user_detail": {
"path": "/api/weibo/get-user-detail/v1",
"method": "GET",
"description": "获取用户信息 包括昵称、头像、用户ID、粉丝数、关注数、简介、认证状态等公开信息",
"required_params": ["token", "uid"],
"optional_params": []
},
"bilibili": {
"api_key": os.getenv("XIAOHONGSHU_TOKEN", "YNSbIjdU"),
"base_url": "https://api.justoneapi.com",
"timeout": 30,
"retries": 3,
"endpoints": {
"bilibili_video_detail": {
"path": "/api/bilibili/get-video-detail/v2",
"method": "GET",
"description": "该接口用于获取指定抖音视频的详细信息，包括视频地址、描述文案、作者信息、发布时间、播放量、点赞数、评论数与分享数等",
"required_params": ["token", "bvid"],
"optional_params": ["token", "bvid"]
}
}
"weixin_user_detail": {
"path": "/api/weixin/get-user-post/v1",
"method": "GET",
"description": "公众号发布的文章内容，包括标题、作者、发布时间、内容摘要以及阅读数、点赞数与转发数等互动数据",
"required_params": ["token", "wxid"],
"optional_params": []
},
"kuaishou": {
"api_key": os.getenv("XIAOHONGSHU_TOKEN", "YNSbIjdU"),
"base_url": "https://api.justoneapi.com",
"timeout": 30,
"retries": 3,
"endpoints": {
"kuaishou_video_detail": {
"path": "/api/kuaishou/get-video-detail/v2",
"method": "GET",
"description": "该接口用于获取指定抖音视频的详细信息，包括视频地址、描述文案、作者信息、发布时间、播放量、点赞数、评论数与分享数等",
"required_params": ["token", "videoId"],
"optional_params": ["token", "videoId"]
}
}
"douyin_video_detail": {
"path": "/api/douyin/get-video-detail/v2",
"method": "GET",
"description": "该接口用于获取指定抖音视频的详细信息，包括视频地址、描述文案、作者信息、发布时间、播放量、点赞数、评论数与分享数等",
"required_params": ["token", "videoId"],
"optional_params": []
},
"other_apis": {
"jizhiliao": {
"api_key": os.getenv("JIZHILIAO_API_KEY", "JZL089ef0b7d0315d96"),
"base_url": "https://api.justoneapi.com",
"timeout": 30,
"retries": 3,
"verifycode": os.getenv("JIZHILIAO_VERIFYCODE", ""),
"endpoints": {
"index_search": {
"path": "/fbmain/monitor/v3/web_search",
"method": "POST",
"description": "极致聊指数搜索",
"required_params": ["keyword", "mode", "BusinessType", "sub_search_type"],
"optional_params": ["key", "verifycode"]
},
}
},
"other": {
# 可以添加其他第三方API配置
"example_api": {
"api_key": os.getenv("EXAMPLE_API_KEY", ""),
"base_url": "https://api.example.com",
"endpoints": {}
}
"kuaishou_video_detail": {
"path": "/api/kuaishou/get-video-detail/v2",
"method": "GET",
"description": "该接口用于获取指定快手视频的详细信息，包括视频地址、描述文案、作者信息、发布时间、播放量、点赞数、评论数与分享数等",
"required_params": ["token", "videoId"],
"optional_params": []
},
"common_settings": {
"timeout": 60,
"retries": 3,
"sign_prefix": "634xz"
"bilibili_video_detail": {
"path": "/api/bilibili/get-video-detail/v2",
"method": "GET",
"description": "该接口用于获取指定B站视频的详细信息，包括视频地址、描述文案、作者信息、发布时间、播放量、点赞数、评论数与分享数等",
"required_params": ["token", "bvid"],
"optional_params": []
},
"jizhiliao_index_search": {
"path": "/fbmain/monitor/v3/web_search",
"method": "POST",
"description": "极致聊指数搜索",
"api_key": os.getenv("JIZHILIAO_API_KEY", "JZL089ef0b7d0315d96"),
"verifycode": os.getenv("JIZHILIAO_VERIFYCODE", ""),
"required_params": ["keyword", "mode", "BusinessType", "sub_search_type"],
"optional_params": ["key", "verifycode"]
}
}
}
},
# 微信指数
"dajiala": {
"api_key": "",
"base_url": "https://www.dajiala.com",
"timeout": 30,
"retries": 3,
@ -263,6 +154,26 @@ class APIConfig:
}
}
},
# dify
"dify": {
"base_url": "http://106.14.211.94",
"api_key": "app-FJXEWdKv63oq1F4rHb4I8kvE",
"timeout": 30,
"retries": 3,
"endpoints": {
"workflows_run": {
"path": "/v1/workflows/run",
"method": "POST",
"description": "运行工作流",
"required_params": ["inputs", "response_mode", "user"],
"optional_params": [],
"headers": {
"Authorization": "Bearer {api_key}",
"Content-Type": "application/json"
}
}
}
}
}
# 尝试从文件加载配置

12
app/utils/app_user_jwt.py
View File

@ -1,8 +1,7 @@
from datetime import datetime, timedelta
from typing import Optional
import jwt
from fastapi import HTTPException, status, Depends
from fastapi.security import HTTPBearer
from fastapi import HTTPException, status, Depends, Header
from app.controllers.app_user import app_user_controller
from app.schemas.app_user import AppUserJWTPayload
from app.settings import settings
@ -12,8 +11,6 @@ SECRET_KEY = settings.SECRET_KEY
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7 # 7天
security = HTTPBearer()
def create_app_user_access_token(user_id: int, phone: str) -> str:
"""
@ -51,7 +48,7 @@ def verify_app_user_token(token: str) -> Optional[AppUserJWTPayload]:
return None
def get_current_app_user_id(token: str = Depends(security)) -> int:
def get_current_app_user_id(token: str = Header(None)) -> int:
"""
从令牌中获取当前AppUser ID
"""
@ -61,7 +58,10 @@ def get_current_app_user_id(token: str = Depends(security)) -> int:
headers={"WWW-Authenticate": "Bearer"},
)
payload = verify_app_user_token(token.credentials)
if not token:
raise credentials_exception
payload = verify_app_user_token(token)
if payload is None:
raise credentials_exception

9
app/utils/calculation_engine/drp/__init__.py Normal file
View File

@ -0,0 +1,9 @@
"""
动态质押率DPR计算包
动态质押率DPR=基础质押率*(1+ 流量修正系数)+ 政策加成系数- 流动性调节因子
"""
from .sub_formulas.dynamic_pledge_rate import DynamicPledgeRateCalculator
__all__ = [
"DynamicPledgeRateCalculator"
]

7
app/utils/calculation_engine/drp/sub_formulas/__init__.py Normal file
View File

@ -0,0 +1,7 @@
from .dynamic_pledge_rate import DynamicPledgeRateCalculator
__all__ = [
"DynamicPledgeRateCalculator"
]

106
app/utils/calculation_engine/drp/sub_formulas/dynamic_pledge_rate.py Normal file
View File

@ -0,0 +1,106 @@
"""
动态质押率DPR计算模块
动态质押率DPR=基础质押率*(1+ 流量修正系数)+ 政策加成系数- 流动性调节因子
"""
class DynamicPledgeRateCalculator:
"""DRP计算器"""
def __init__(self):
pass
def calculate_dynamic_pledge_rate(self,
monthly_amount: float,
heritage_level: str,
) -> float:
"""
动态质押率 DPR 计算公式：
DPR = 基础质押率 * (1 + 流量修正系数) + 政策加成系数 - 流动性调节因子
参数:
monthly_amount (float): 月交易额（万元）
heritage_level (str): 非遗等级
liquidity_adjustment (float): 流动性调节因子（例如 0.05 表示5%）
固定参数:
基础质押率 = 0.5
流量修正系数 = 0.3
"""
base_rate = 0.5
flow_correction = 0.3
# ① 交易额得分影响可嵌入流量修正或单独展示，这里保持原式不变
policy_bonus = self.get_heritage_level_score(heritage_level)
liquidity_adjustment = self.get_monthly_transaction_score(monthly_amount)
dpr = base_rate * (1 + flow_correction) + policy_bonus - liquidity_adjustment
return round(dpr, 4)
def get_heritage_level_score(self, level: str) -> float:
"""
根据用户所选的非遗等级匹配系数分数。
规则：
① 国家级非遗 => +0.05
② 省级非遗 => +0.03
③ 纳入《国家文化数字化战略清单》 => +0.02
④ 无 => 0.0
参数:
level (str): 用户选择的非遗等级描述
返回:
float: 对应的分数
"""
level = level.strip()
if "国家级" in level:
return 0.05
elif "省级" in level:
return 0.03
elif "国家文化数字化战略清单" in level:
return 0.02
else:
return 0.0
def get_monthly_transaction_score(self, monthly_amount: float) -> float:
"""
根据用户月交易额区间匹配评分。
规则：
① 月交易额 < 100万 => -0.1
② 月交易额 ∈ [100万, 500万) => 0
③ 月交易额 ≥ 500万 => +0.15
参数:
monthly_amount (float): 月交易额（单位：万元）
返回:
float: 对应的评分
"""
if monthly_amount < 100:
return -0.1
elif monthly_amount < 500:
return 0.0
else:
return 0.15
# 示例使用
if __name__ == "__main__":
# 创建计算器实例
calculator = DynamicPledgeRateCalculator()
# 示例1：国家级非遗，月交易额500万，流动性调节因子0.05
print(calculator.calculate_dynamic_pledge_rate(500, "国家级非遗"))
# 结果：0.55
# 示例2：省级非遗，流动性调节因子0.02
print(calculator.calculate_dynamic_pledge_rate(300, "省级非遗"))
# 结果：0.68
# 示例3：无非遗，流动性调节因子0.1
print(calculator.calculate_dynamic_pledge_rate(50, "无"))
# 0.75

67
app/utils/universal_api_manager.py
View File

@ -53,7 +53,14 @@ class UniversalAPIManager:
endpoint_config = self._get_endpoint_config(provider, endpoint)
if not endpoint_config:
return None
return endpoint_config.get('api_key')
# 优先从端点配置中获取APIKey
if 'APIKey' in endpoint_config:
return endpoint_config['APIKey']
# 如果端点配置中没有，则从提供商配置中获取api_key
provider_config = self._get_provider_config(provider)
return provider_config.get('api_key')
def make_request(self, provider: str, endpoint: str, params: Dict[str, Any], timeout: Optional[int] = None) -> Dict[str, Any]:
"""
@ -89,7 +96,7 @@ class UniversalAPIManager:
# 添加API密钥到参数中
params['APIKey'] = api_key
print(params)
# 发送请求
try:
response = self.session.request(
@ -156,6 +163,17 @@ class UniversalAPIManager:
# 对于GET请求，将参数添加到URL中
query_string = urllib.parse.urlencode(params)
full_url = f"{full_url}?{query_string}"
elif provider == 'chinaz' and method.upper() == 'POST':
# 对于Chinaz POST请求，只将APIKey和ChinazVer添加到URL中
url_params = {}
if 'APIKey' in params:
url_params['APIKey'] = params['APIKey']
if 'ChinazVer' in params:
url_params['ChinazVer'] = params['ChinazVer']
if url_params:
query_string = urllib.parse.urlencode(url_params)
full_url = f"{full_url}?{query_string}"
return full_url
@ -176,6 +194,9 @@ class UniversalAPIManager:
missing_params = []
for param in required_params:
# 如果参数在端点配置中已经定义（如APIKey），则跳过验证
if param in endpoint_config:
continue
if param not in params or params[param] is None:
missing_params.append(param)
@ -195,6 +216,7 @@ class UniversalAPIManager:
Dict[str, Any]: 处理后的参数
"""
provider_config = self._get_provider_config(provider)
endpoint_config = self._get_endpoint_config(provider, endpoint)
prepared_params = params.copy()
# 根据不同的API提供商添加特定参数
@ -202,6 +224,10 @@ class UniversalAPIManager:
# 站长之家API需要签名
if 'sign' not in prepared_params:
prepared_params['sign'] = self._generate_chinaz_sign()
# 添加APIKey参数
if 'APIKey' not in prepared_params and endpoint_config and 'APIKey' in endpoint_config:
prepared_params['APIKey'] = endpoint_config['APIKey']
elif provider == 'xiaohongshu':
# 小红书接口使用 JustOneAPI 平台，需要 token 参数
@ -228,6 +254,11 @@ class UniversalAPIManager:
if default_verifycode is not None:
prepared_params['verifycode'] = default_verifycode
elif provider == 'dify':
# Dify API需要在请求头中设置Authorization
# 这里不需要修改params，Authorization会在make_request中处理
pass
return prepared_params
@ -264,16 +295,38 @@ class UniversalAPIManager:
# 构建URL
url = self._build_url(provider, endpoint, prepared_params)
logger.info(f"发送{method}请求到: {url}")
# 发送请求
for attempt in range(retries + 1):
try:
if method == 'GET':
response = self.session.get(url, timeout=timeout)
elif method == 'POST':
response = self.session.post(url, json=prepared_params, timeout=timeout)
if provider == 'chinaz':
# 对于Chinaz API，从prepared_params中移除URL参数，只发送body参数
body_params = prepared_params.copy()
body_params.pop('APIKey', None)
body_params.pop('ChinazVer', None)
# Chinaz API使用application/x-www-form-urlencoded格式
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
response = self.session.post(url, data=body_params, headers=headers, timeout=timeout)
print(response.json())
elif provider == 'dify':
# 对于Dify API，需要设置Authorization头
provider_config = self._get_provider_config(provider)
endpoint_config = self._get_endpoint_config(provider, endpoint)
# 获取API密钥
api_key = provider_config.get('api_key', 'app-FJXEWdKv63oq1F4rHb4I8kvE')
# 设置请求头
headers = {
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
}
response = self.session.post(url, json=prepared_params, headers=headers, timeout=timeout)
else:
response = self.session.post(url, json=prepared_params, timeout=timeout)
else:
raise ValueError(f"不支持的HTTP方法: {method}")
@ -349,7 +402,7 @@ class UniversalAPIManager:
'searchKey': company_name,
'pageNo': '1',
'range': '100',
'searchType': '0',
'searchType': '1',
'ChinazVer': chinaz_ver
}
return self.make_request('chinaz', 'patent', params)

4551
app/utils/专利.json Normal file
View File

File diff suppressed because it is too large Load Diff

2
demo_api.py
View File

@ -372,7 +372,7 @@ def run_tests():
test_change_password()
# 测试登出
test_logout()
# test_logout()
print("\n===== 所有测试通过 =====")
except AssertionError as e:

4
估值字段.txt
View File

@ -37,5 +37,5 @@
export DOCKER_DEFAULT_PLATFORM=linux/amd64
docker build -t zfc931912343/guzhi-fastapi-admin:v1.1 .
docker push zfc931912343/guzhi-fastapi-admin:v1.1
docker build -t zfc931912343/guzhi-fastapi-admin:v1.2 .
docker push zfc931912343/guzhi-fastapi-admin:v1.2