270 lines
8.6 KiB
Python
270 lines
8.6 KiB
Python
import argparse
|
||
import json
|
||
import time
|
||
from typing import Dict, Any, Optional
|
||
|
||
import requests
|
||
|
||
|
||
def _print(title: str, payload: Any) -> None:
|
||
print(f"\n[{title}]\n{json.dumps(payload, ensure_ascii=False, indent=2)}")
|
||
|
||
|
||
def _url(base: str, path: str) -> str:
|
||
return f"{base}{path}"
|
||
|
||
|
||
class AppClient:
|
||
"""
|
||
用户端客户端，会话维持与常用接口封装
|
||
|
||
参数:
|
||
base: API 基础地址，如 http://127.0.0.1:9991/api/v1
|
||
|
||
属性:
|
||
session: requests.Session 会话对象，携带 token
|
||
"""
|
||
|
||
def __init__(self, base: str) -> None:
|
||
self.base = base.rstrip("/")
|
||
self.session = requests.Session()
|
||
|
||
def set_token(self, token: str) -> None:
|
||
"""
|
||
设置用户端 token 到请求头
|
||
|
||
参数:
|
||
token: 登录接口返回的 access_token
|
||
返回:
|
||
None
|
||
"""
|
||
self.session.headers.update({"token": token})
|
||
|
||
def register(self, phone: str) -> Dict[str, Any]:
|
||
"""
|
||
用户注册
|
||
|
||
参数:
|
||
phone: 手机号
|
||
返回:
|
||
注册响应 dict
|
||
"""
|
||
resp = self.session.post(_url(self.base, "/app-user/register"), json={"phone": phone})
|
||
return _safe_json(resp)
|
||
|
||
def login(self, phone: str, password: str) -> Optional[str]:
|
||
"""
|
||
用户登录
|
||
|
||
参数:
|
||
phone: 手机号
|
||
password: 密码
|
||
返回:
|
||
access_token 或 None
|
||
"""
|
||
resp = self.session.post(_url(self.base, "/app-user/login"), json={"phone": phone, "password": password})
|
||
data = _safe_json(resp)
|
||
token = data.get("access_token") if isinstance(data, dict) else None
|
||
if token:
|
||
self.set_token(token)
|
||
return token
|
||
|
||
def profile(self) -> Dict[str, Any]:
|
||
resp = self.session.get(_url(self.base, "/app-user/profile"))
|
||
return _safe_json(resp)
|
||
|
||
def dashboard(self) -> Dict[str, Any]:
|
||
resp = self.session.get(_url(self.base, "/app-user/dashboard"))
|
||
return _safe_json(resp)
|
||
|
||
def quota(self) -> Dict[str, Any]:
|
||
resp = self.session.get(_url(self.base, "/app-user/quota"))
|
||
return _safe_json(resp)
|
||
|
||
def submit_valuation(self, payload: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""
|
||
提交估值评估
|
||
|
||
参数:
|
||
payload: 估值评估输入数据
|
||
返回:
|
||
提交响应 dict
|
||
"""
|
||
resp = self.session.post(_url(self.base, "/app-valuations/"), json=payload)
|
||
return _safe_json(resp)
|
||
|
||
def list_valuations(self) -> Dict[str, Any]:
|
||
resp = self.session.get(_url(self.base, "/app-valuations/"))
|
||
return _safe_json(resp)
|
||
|
||
def valuation_detail(self, valuation_id: int) -> Dict[str, Any]:
|
||
resp = self.session.get(_url(self.base, f"/app-valuations/{valuation_id}"))
|
||
return _safe_json(resp)
|
||
|
||
|
||
class AdminClient:
|
||
"""
|
||
后台客户端，会话维持与接口封装
|
||
|
||
参数:
|
||
base: API 基础地址
|
||
"""
|
||
|
||
def __init__(self, base: str) -> None:
|
||
self.base = base.rstrip("/")
|
||
self.session = requests.Session()
|
||
|
||
def set_token(self, token: str) -> None:
|
||
self.session.headers.update({"token": token})
|
||
|
||
def login(self, username: str, password: str) -> Optional[str]:
|
||
resp = self.session.post(_url(self.base, "/base/access_token"), json={"username": username, "password": password})
|
||
data = _safe_json(resp)
|
||
token = data.get("data", {}).get("access_token") if isinstance(data, dict) else None
|
||
if token:
|
||
self.set_token(token)
|
||
return token
|
||
|
||
def list_valuations(self) -> Dict[str, Any]:
|
||
resp = self.session.get(_url(self.base, "/valuations/"))
|
||
return _safe_json(resp)
|
||
|
||
def valuation_detail(self, valuation_id: int) -> Dict[str, Any]:
|
||
resp = self.session.get(_url(self.base, f"/valuations/{valuation_id}"))
|
||
return _safe_json(resp)
|
||
|
||
def valuation_steps(self, valuation_id: int) -> Dict[str, Any]:
|
||
resp = self.session.get(_url(self.base, f"/valuations/{valuation_id}/steps"))
|
||
return _safe_json(resp)
|
||
|
||
|
||
def _safe_json(resp: requests.Response) -> Dict[str, Any]:
|
||
try:
|
||
return resp.json()
|
||
except Exception:
|
||
return {"status_code": resp.status_code, "text": resp.text}
|
||
|
||
|
||
def build_sample_payload() -> Dict[str, Any]:
|
||
"""
|
||
构建估值评估示例输入（精简版）
|
||
|
||
返回:
|
||
dict: 估值评估输入
|
||
"""
|
||
# 使用你提供的参数，保持后端计算逻辑不变
|
||
payload = {
|
||
"asset_name": "马王堆",
|
||
"institution": "成都文化产权交易所",
|
||
"industry": "文化艺术业",
|
||
"annual_revenue": "10000",
|
||
"rd_investment": "6000",
|
||
"three_year_income": ["8000", "9000", "9500"],
|
||
"funding_status": "省级资助",
|
||
"sales_volume": "60000",
|
||
"link_views": "350000",
|
||
"circulation": "3",
|
||
"last_market_activity": "0",
|
||
"monthly_transaction": "1",
|
||
"price_fluctuation": [402, 445],
|
||
"application_maturity": "0",
|
||
"application_coverage": "0",
|
||
"cooperation_depth": "0",
|
||
"offline_activities": "20",
|
||
"online_accounts": ["1", "成都文交所", "500000", "89222", "97412"],
|
||
"inheritor_level": "省级传承人",
|
||
"inheritor_age_count": [200, 68, 20],
|
||
"inheritor_certificates": [],
|
||
"heritage_level": "2",
|
||
"historical_evidence": {"artifacts": "58", "ancient_literature": "789", "inheritor_testimony": "100"},
|
||
"patent_certificates": [],
|
||
"pattern_images": [],
|
||
"patent_application_no": "",
|
||
"heritage_asset_level": "纳入《国家文化数字化战略清单》",
|
||
"inheritor_ages": [200, 68, 20],
|
||
"implementation_stage": "成熟应用",
|
||
"coverage_area": "全球覆盖",
|
||
"collaboration_type": "无",
|
||
"scarcity_level": "流通：总发行份数 >1000份，或二级市场流通率 ≥ 5%",
|
||
"market_activity_time": "近一周",
|
||
"monthly_transaction_amount": "月交易额＞100万＜500万",
|
||
"platform_accounts": {
|
||
"douyin": {"account": "成都文交所", "likes": "500000", "comments": "89222", "shares": "97412", "views": "100000"}
|
||
}
|
||
}
|
||
# 若 application_coverage 为占位，则用 coverage_area 回填
|
||
if payload.get("application_coverage") in (None, "0", "") and payload.get("coverage_area"):
|
||
payload["application_coverage"] = payload["coverage_area"]
|
||
return payload
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="估值二期 API 冒烟测试")
|
||
parser.add_argument("--base", default="http://127.0.0.1:9991/api/v1", help="API基础地址")
|
||
parser.add_argument("--phone", default="13800138001", help="测试手机号")
|
||
args = parser.parse_args()
|
||
|
||
base = args.base.rstrip("/")
|
||
phone = args.phone
|
||
default_pwd = phone[-6:]
|
||
|
||
app = AppClient(base)
|
||
admin = AdminClient(base)
|
||
|
||
# 用户注册
|
||
reg = app.register(phone)
|
||
_print("用户注册", reg)
|
||
|
||
# 用户登录
|
||
token = app.login(phone, default_pwd)
|
||
_print("用户登录token", {"access_token": token})
|
||
if not token:
|
||
print("登录失败，终止测试")
|
||
return
|
||
|
||
# 用户相关接口
|
||
_print("用户信息", app.profile())
|
||
_print("首页摘要", app.dashboard())
|
||
_print("剩余估值次数", app.quota())
|
||
|
||
# 提交估值
|
||
payload = build_sample_payload()
|
||
submit = app.submit_valuation(payload)
|
||
_print("提交估值", submit)
|
||
|
||
# 轮询估值列表抓取最新记录
|
||
valuation_id = None
|
||
for _ in range(10):
|
||
lst = app.list_valuations()
|
||
_print("我的估值列表", lst)
|
||
try:
|
||
items = lst.get("data", []) if isinstance(lst, dict) else []
|
||
if items:
|
||
valuation_id = items[0].get("id") or items[-1].get("id")
|
||
if valuation_id:
|
||
break
|
||
except Exception:
|
||
pass
|
||
time.sleep(0.8)
|
||
|
||
if valuation_id:
|
||
detail = app.valuation_detail(valuation_id)
|
||
_print("估值详情", detail)
|
||
else:
|
||
print("未获得估值ID，跳过详情")
|
||
|
||
# 后台登录
|
||
admin_token = admin.login("admin", "123456")
|
||
_print("后台登录token", {"access_token": admin_token})
|
||
if admin_token:
|
||
vlist = admin.list_valuations()
|
||
_print("后台估值列表", vlist)
|
||
if valuation_id:
|
||
vdetail = admin.valuation_detail(valuation_id)
|
||
_print("后台估值详情", vdetail)
|
||
vsteps = admin.valuation_steps(valuation_id)
|
||
_print("后台估值计算步骤", vsteps)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |