From 4656d4b96c5cd527b8b02e162d02c607a322a023 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=82=B9=E6=96=B9=E6=88=90?= Date: Thu, 9 Oct 2025 22:13:30 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E9=9D=9E=E9=81=97IP?= =?UTF-8?q?=E8=B5=84=E4=BA=A7=E4=BC=B0=E5=80=BCAPI=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E8=84=9A=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实现完整的API测试流程，包括用户注册、登录、信息管理及估值申请功能 --- demo_api.py | 384 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 384 insertions(+) create mode 100644 demo_api.py diff --git a/demo_api.py b/demo_api.py new file mode 100644 index 0000000..a22431f --- /dev/null +++ b/demo_api.py @@ -0,0 +1,384 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +import requests +import json +import random +import time + +# API基础URL +BASE_URL = "http://127.0.0.1:9999/api/v1" + +# 测试数据 +test_phone = f"1380000{random.randint(1000, 9999)}" +test_password = test_phone[-6:] # 默认密码是手机号后6位 +access_token = None +user_id = None +valuation_id = None + +def test_register(): + """测试用户注册功能""" + print("\n===== 测试用户注册 =====") + url = f"{BASE_URL}/app-user/register" + data = { + "phone": test_phone + } + + response = requests.post(url, json=data) + print(f"请求URL: {url}") + print(f"请求数据: {data}") + print(f"响应状态码: {response.status_code}") + print(f"响应内容: {response.text}") + + assert response.status_code == 200, "注册请求失败" + result = response.json() + assert result["code"] == 200, "注册失败" + assert result["data"]["phone"] == test_phone, "返回的手机号不匹配" + assert result["data"]["default_password"] == test_phone[-6:], "默认密码不正确" + + print("✅ 用户注册测试通过") + return result + +def test_login(): + """测试用户登录功能""" + global access_token + + print("\n===== 测试用户登录 =====") + url = f"{BASE_URL}/app-user/login" + data = { + "phone": test_phone, + "password": test_password + } + + response = requests.post(url, json=data) + print(f"请求URL: {url}") + print(f"请求数据: {data}") + print(f"响应状态码: {response.status_code}") + print(f"响应内容: {response.text}") + + assert response.status_code == 200, "登录请求失败" + result = response.json() + assert "access_token" in result, "登录失败，未返回token" + + # 保存token供后续请求使用 + access_token = result["access_token"] + + print("✅ 用户登录测试通过") + return result + +def test_create_valuation(): + """测试创建估值评估申请""" + global access_token + + print("\n===== 测试创建估值评估申请 =====") + url = f"{BASE_URL}/app-valuations/" + + # 准备请求头，包含授权token + headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json" + } + + # 估值评估申请数据 - 根据估值字段.txt更新 + data = { + # 02 - 基础信息 - 非遗IP资产的基本信息 + "asset_name": f"蜀绣-{random.randint(1000, 9999)}", # 资产名称：必须是企业全名称 + "institution": "有数", # 所属机构：拥有或管理该非遗IP资产的机构名称 + "industry": "文化艺术", # 所属行业：非遗IP资产所属的行业分类 + + # 03 - 财务状况 + "rd_investment": "5", # ：近12个月的研发费用（单位：万元） + "annual_revenue": "100", # 近12个月的营收额（单位：万元），用于计算创新投入比 + "three_year_income": [100, 120, 150], # 近三年的每年收益：资产近3年的年收益数据（单位：万元），用于计算年均收益和增长率 + "funding_status": "国家级资助", # 资金支持: 国家级资助(10分)、省级资助(7分)、无资助(0分) + + # 04 - 非遗等级与技术 + "inheritor_level": "国家级", # 传承人等级 + "inheritor_ages": [60, 42, 35], # 传承人年龄 + "inheritor_certificates": ["http://example.com/国家级非遗传承人证书.jpg"], # 传承人证书：传承人资质证明材料 + "heritage_asset_level": "国家级非遗", # 非遗资产等级 + "patent_remaining_years": "8", # [实际上就是专利号 通过API查询到的 ]专利剩余年限：资产相关专利的剩余保护期，8年对应7分 + "historical_evidence": { # 资产历史证据类型+数量：历史传承的证据材料 + "artifacts": 1, # 出土实物数量 + "ancient_literature": 2, # 古代文献数量 + "inheritor_testimony": 3, # 传承人佐证数量 + "modern_research": 1 # 现代研究数量 + }, + # 专利证书： + "patent_certificates": ["http://example.com/专利证书1.jpg", "http://example.com/专利证书2.jpg"], + "pattern_images": ["pattern1.jpg"], # 纹样图片：资产相关的纹样图片文件 + + # 04 - 非遗应用与推广 + "implementation_stage": "成熟应用", # 非遗资产应用成熟度 + "coverage_area": "区域覆盖", # 非遗资产应用覆盖范围 + "collaboration_type": "品牌联名", # 非遗资产跨界合作深度 + "offline_teaching_count": 12, # 近12个月线下相关演讲活动次数 + "platform_accounts": { # 线上相关宣传账号信息 + "bilibili": { + "followers_count": 8000, # 粉丝数量 + "likes": 1000, # 点赞数 + "comments": 500, # 评论数 + "shares": 500 # 转发数 + }, # B站账号 + "douyin": { + "followers_count": 8000, # 粉丝数量 + "likes": 1000, # 点赞数 + "comments": 500, # 评论数 + "shares": 500 # 转发数 + } # 抖音账号 + }, + + # 06 - 非遗资产衍生商品信息 + #该商品近12个月销售量 + "sales_volume": "1000", # 近12个月销售量：资产衍生商品的近12个月销售量（单位：件） 链接购买量 + # 该商品近12个月的链接浏览量 + "link_views": "10000", # 近12个月链接浏览量：资产衍生商品相关链接的近12个月浏览量（单位：次） 浏览量 + "scarcity_level": "流通", # 稀缺等级：资产的稀缺程度，流通（发行量>1000份）对应0.1分 + "market_activity_time": "近一月", # 市场活动的时间 + "monthly_transaction_amount": "<100万元", # 月交易额：资产衍生商品的月交易额水平，<100万元对应-0.1 + "price_range": { # 资产商品的价格波动率：近30天商品价格的波动情况 + "highest": 239, # 最高价（单位：元） + "lowest": 189 # 最低价（单位：元） + }, + "market_price": 0, # 直接提供的市场价格（单位：万元） 用户输入: 专家审核 或者 系统默认 专家审核 + + # 内置API 计算字段 + "infringement_record": "无侵权记录", # 侵权记录：资产的侵权历史情况，无侵权记录对应10分 + "patent_count": "1", # 专利使用量：资产相关的专利数量，每引用一项专利+2.5分 + "esg_value": "10", # ESG关联价值：根据行业匹配的ESG（环境、社会、治理）关联价值 + "policy_matching": "10", # 政策匹配度：根据行业自动匹配的政策匹配度分值 + "online_course_views": 2000, # 线上课程点击量：抖音/快手播放量按100:1折算为学习人次，B站课程按50:1折算 + "pattern_complexity": "1.459", # 结构复杂度：纹样的结构复杂度值 搞一个默认值: 0.0 + "normalized_entropy": "9.01", # 归一化信息熵：纹样的归一化信息熵值 搞一个默认值: 0.0 + "legal_risk": "无诉讼", # 法律风险-侵权诉讼历史：资产所属机构的诉讼历史，无诉讼对应10分 + # 动态质押率DPR：计算公式=基础质押率*(1+流量修正系数)+政策加成系数-流动性调节因子 + "base_pledge_rate": "50%", # 基础质押率：基础质押率固定值50% + "flow_correction": "0.3", # 流量修正系数：固定值0.3 + } + + response = requests.post(url, headers=headers, json=data) + print(f"请求URL: {url}") + print(f"请求头: {headers}") + print(f"请求数据: {json.dumps(data, ensure_ascii=False, indent=2)}") + print(f"响应状态码: {response.status_code}") + print(f"响应内容: {response.text}") + + assert response.status_code == 200, "创建估值评估申请请求失败" + result = response.json() + assert result["code"] == 200, "创建估值评估申请失败" + + print("✅ 创建估值评估申请测试通过") + return result + +def test_get_profile(): + """测试获取用户个人信息""" + global access_token, user_id + + print("\n===== 测试获取用户个人信息 =====") + url = f"{BASE_URL}/app-user/profile" + + headers = { + "Authorization": f"Bearer {access_token}" + } + + response = requests.get(url, headers=headers) + print(f"请求URL: {url}") + print(f"请求头: {headers}") + print(f"响应状态码: {response.status_code}") + print(f"响应内容: {response.text}") + + assert response.status_code == 200, "获取用户信息请求失败" + result = response.json() + user_id = result["id"] # 保存用户ID供后续使用 + + print("✅ 获取用户个人信息测试通过") + return result + +def test_change_password(): + """测试修改密码""" + global access_token + + print("\n===== 测试修改密码 =====") + url = f"{BASE_URL}/app-user/change-password" + + headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json" + } + + new_password = "new" + test_password + data = { + "old_password": test_password, + "new_password": new_password + } + + response = requests.post(url, headers=headers, json=data) + print(f"请求URL: {url}") + print(f"请求头: {headers}") + print(f"请求数据: {data}") + print(f"响应状态码: {response.status_code}") + print(f"响应内容: {response.text}") + + assert response.status_code == 200, "修改密码请求失败" + result = response.json() + assert result["code"] == 200, "修改密码失败" + + print("✅ 修改密码测试通过") + return result + +def test_update_profile(): + """测试更新用户信息""" + global access_token + + print("\n===== 测试更新用户信息 =====") + url = f"{BASE_URL}/app-user/profile" + + headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json" + } + + data = { + "nickname": f"测试用户{random.randint(100, 999)}", + "avatar": "https://example.com/avatar.jpg", + "gender": "male", + "email": f"test{random.randint(100, 999)}@example.com" + } + + response = requests.put(url, headers=headers, json=data) + print(f"请求URL: {url}") + print(f"请求头: {headers}") + print(f"请求数据: {data}") + print(f"响应状态码: {response.status_code}") + print(f"响应内容: {response.text}") + + assert response.status_code == 200, "更新用户信息请求失败" + result = response.json() + # 更新用户信息接口直接返回用户对象，不包含code字段 + assert "id" in result, "更新用户信息失败" + + print("✅ 更新用户信息测试通过") + return result + +def test_logout(): + """测试用户登出""" + global access_token + + print("\n===== 测试用户登出 =====") + url = f"{BASE_URL}/app-user/logout" + + headers = { + "Authorization": f"Bearer {access_token}" + } + + response = requests.post(url, headers=headers) + print(f"请求URL: {url}") + print(f"请求头: {headers}") + print(f"响应状态码: {response.status_code}") + print(f"响应内容: {response.text}") + + assert response.status_code == 200, "登出请求失败" + result = response.json() + assert result["code"] == 200, "登出失败" + + print("✅ 用户登出测试通过") + return result + +def test_get_valuation_list(): + """测试获取用户估值列表""" + global access_token + + print("\n===== 测试获取用户估值列表 =====") + url = f"{BASE_URL}/app-valuations/" + + headers = { + "Authorization": f"Bearer {access_token}" + } + + response = requests.get(url, headers=headers) + print(f"请求URL: {url}") + print(f"请求头: {headers}") + print(f"响应状态码: {response.status_code}") + print(f"响应内容: {response.text}") + + assert response.status_code == 200, "获取估值列表请求失败" + result = response.json() + assert result["code"] == 200, "获取估值列表失败" + + print("✅ 获取用户估值列表测试通过") + return result + +def test_get_valuation_detail(): + """测试获取估值详情""" + global access_token, valuation_id + + # 先获取估值列表，获取第一个估值ID + if not valuation_id: + list_result = test_get_valuation_list() + if list_result["data"] and len(list_result["data"]) > 0: + valuation_id = list_result["data"][0]["id"] + else: + print("⚠️ 没有可用的估值记录，跳过估值详情测试") + return None + + print("\n===== 测试获取估值详情 =====") + url = f"{BASE_URL}/app-valuations/{valuation_id}" + + headers = { + "Authorization": f"Bearer {access_token}" + } + + response = requests.get(url, headers=headers) + print(f"请求URL: {url}") + print(f"请求头: {headers}") + print(f"响应状态码: {response.status_code}") + print(f"响应内容: {response.text}") + + assert response.status_code == 200, "获取估值详情请求失败" + result = response.json() + assert result["code"] == 200, "获取估值详情失败" + + print("✅ 获取估值详情测试通过") + return result + +def run_tests(): + """运行所有测试""" + try: + # 测试注册 + test_register() + + # 等待一秒，确保数据已保存 + time.sleep(1) + + # 测试登录 + test_login() + + # 测试获取用户个人信息 + test_get_profile() + + # 测试更新用户信息 + test_update_profile() + + # 测试创建估值评估申请 + test_create_valuation() + + # 测试获取估值列表 + test_get_valuation_list() + + # 测试获取估值详情 + test_get_valuation_detail() + + # 测试修改密码 + test_change_password() + + # 测试登出 + test_logout() + + print("\n===== 所有测试通过 =====") + except AssertionError as e: + print(f"\n❌ 测试失败: {e}") + except Exception as e: + print(f"\n❌ 发生错误: {e}") + +if __name__ == "__main__": + run_tests() \ No newline at end of file