该文件提供用户行为埋点和 API 路径映射管理相关接口,包括:
所有接口均需要 Token 认证。
路由前缀:/apiv1(以 routers/__init__.py 中注册为准)
/apiv1/tracking/record — 记录埋点功能说明: 记录用户访问某个 API 的行为。自动获取客户端 IP 和请求方法,生成 UUID 作为请求 ID。
是否需要认证: 是(通过 request.state.user 获取用户信息)
请求方式: POST
请求体(JSON,Pydantic 模型 TrackingRequest):
| 字段 | 类型 | 必填 | 默认值 | 说明 |
|------|------|------|--------|------|
| api_path | string | 是 | — | 被记录的 API 路径 |
| api_name | string | 否 | "" | API 名称/描述 |
业务逻辑:
request.state.user 获取 user_idrequest.client.host 获取客户端 IP(client 为 None 时为空字符串)request_idTrackingRecord 记录并提交request_id自动记录的字段: | 字段 | 来源 | 说明 | |------|------|------| | user_id | request.state.user.user_id | 当前用户 ID | | method | request.method | HTTP 请求方法(此处固定为 POST) | | request_id | uuid.uuid4() | 唯一请求标识 | | ip_address | request.client.host | 客户端 IP 地址 | | created_at | get_unix() | 当前 Unix 时间戳 |
成功响应字段: | 字段 | 类型 | 说明 | |------|------|------| | data.request_id | string | UUID 格式的请求 ID |
测试用例:
// 请求
POST /apiv1/tracking/record
token: <有效Token>
Content-Type: application/json
{
"api_path": "/apiv1/chat/send",
"api_name": "发送聊天消息"
}
// 预期响应 (HTTP 200)
{
"statusCode": 200,
"msg": "记录成功",
"data": {
"request_id": "550e8400-e29b-41d4-a716-446655440000"
}
}
request_id为 UUID v4 格式字符串,每次不同。
// 请求
POST /apiv1/tracking/record
token: <有效Token>
Content-Type: application/json
{
"api_path": "/apiv1/get_hot_question"
}
// 预期响应 (HTTP 200)
{
"statusCode": 200,
"msg": "记录成功",
"data": {
"request_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
}
}
数据库中
api_name存储为空字符串。
// 请求
POST /apiv1/tracking/record
token: <有效Token>
Content-Type: application/json
{
"api_name": "测试接口"
}
// 预期响应 (HTTP 422)
{
"detail": [
{
"loc": ["body", "api_path"],
"msg": "field required",
"type": "value_error.missing"
}
]
}
// 请求
POST /apiv1/tracking/record
Content-Type: application/json
{
"api_path": "/apiv1/chat/send"
}
// 预期响应 (HTTP 401)
{
"statusCode": 401,
"msg": "未认证"
}
// 请求(第一次)
POST /apiv1/tracking/record
token: <有效Token>
Content-Type: application/json
{
"api_path": "/apiv1/chat/send",
"api_name": "发送消息"
}
// 预期响应 (HTTP 200)
{
"statusCode": 200,
"msg": "记录成功",
"data": {
"request_id": "uuid-1"
}
}
// 请求(第二次,相同 api_path)
POST /apiv1/tracking/record
token: <有效Token>
Content-Type: application/json
{
"api_path": "/apiv1/chat/send",
"api_name": "发送消息"
}
// 预期响应 (HTTP 200)
// request_id 与第一次不同
{
"statusCode": 200,
"msg": "记录成功",
"data": {
"request_id": "uuid-2"
}
}
同一 API 路径可被多次记录,不做去重。
/apiv1/tracking/records — 获取埋点记录功能说明: 查询当前用户的埋点记录,按创建时间倒序排列,支持限制返回条数。
是否需要认证: 是
请求方式: GET
请求参数(Query): | 字段 | 类型 | 必填 | 默认值 | 说明 | |------|------|------|--------|------| | limit | int | 否 | 100 | 返回的最大记录条数 |
业务逻辑:
user.user_id 过滤当前用户的记录created_at 倒序排列(最新的在前)limit 限制返回条数(非分页模式)成功响应字段: | 字段 | 类型 | 说明 | |------|------|------| | data[].id | int | 记录 ID | | data[].api_path | string | API 路径 | | data[].api_name | string | API 名称 | | data[].created_at | int | 创建时间(Unix 时间戳) |
注意:响应中不包含
method、request_id、ip_address等字段。
测试用例:
// 请求
GET /apiv1/tracking/records
token: <有效Token>
// 预期响应 (HTTP 200)
{
"statusCode": 200,
"msg": "success",
"data": [
{
"id": 50,
"api_path": "/apiv1/chat/send",
"api_name": "发送聊天消息",
"created_at": 1700000500
},
{
"id": 49,
"api_path": "/apiv1/get_hot_question",
"api_name": "获取热点问题",
"created_at": 1700000400
}
]
}
// 请求
GET /apiv1/tracking/records?limit=5
token: <有效Token>
// 预期响应 (HTTP 200)
{
"statusCode": 200,
"msg": "success",
"data": [...] // 最多 5 条记录
}
// 请求
GET /apiv1/tracking/records
token: <新用户Token,无埋点记录>
// 预期响应 (HTTP 200)
{
"statusCode": 200,
"msg": "success",
"data": []
}
// 请求
GET /apiv1/tracking/records
// 预期响应 (HTTP 401)
{
"statusCode": 401,
"msg": "未认证"
}
// 请求
GET /apiv1/tracking/records
token: <用户A的Token>
// 预期行为:仅返回用户A的埋点记录,不包含其他用户的记录
// 预期响应 (HTTP 200)
{
"statusCode": 200,
"msg": "success",
"data": [...] // 仅包含 user_id 为用户A的记录
}
/apiv1/tracking/api_mapping — 添加 API 映射功能说明: 添加一条 API 路径与名称的映射关系。同一 api_path 不可重复添加。
是否需要认证: 是(代码中显式检查 if not user)
请求方式: POST
请求体(JSON,Pydantic 模型 ApiMappingRequest):
| 字段 | 类型 | 必填 | 默认值 | 说明 |
|------|------|------|--------|------|
| api_path | string | 是 | — | API 路径 |
| api_name | string | 是 | — | API 名称 |
| api_desc | string | 否 | "" | API 描述 |
业务逻辑:
ApiPathMapping 表中是否已存在相同 api_pathApiPathMapping 记录成功响应字段: | 字段 | 类型 | 说明 | |------|------|------| | data.id | int | 新创建的映射记录 ID |
测试用例:
// 请求
POST /apiv1/tracking/api_mapping
token: <有效Token>
Content-Type: application/json
{
"api_path": "/apiv1/chat/send",
"api_name": "发送聊天消息",
"api_desc": "用户发送聊天消息给AI,支持流式响应"
}
// 预期响应 (HTTP 200)
{
"statusCode": 200,
"msg": "添加成功",
"data": {
"id": 1
}
}
// 请求
POST /apiv1/tracking/api_mapping
token: <有效Token>
Content-Type: application/json
{
"api_path": "/apiv1/get_hot_question",
"api_name": "获取热点问题"
}
// 预期响应 (HTTP 200)
{
"statusCode": 200,
"msg": "添加成功",
"data": {
"id": 2
}
}
// 请求(/apiv1/chat/send 已在用例1中添加)
POST /apiv1/tracking/api_mapping
token: <有效Token>
Content-Type: application/json
{
"api_path": "/apiv1/chat/send",
"api_name": "发送消息(重复)",
"api_desc": "重复添加"
}
// 预期响应 (HTTP 200, 业务码 400)
{
"statusCode": 400,
"msg": "API映射已存在"
}
// 请求
POST /apiv1/tracking/api_mapping
token: <有效Token>
Content-Type: application/json
{
"api_path": "/apiv1/test"
}
// 预期响应 (HTTP 422)
{
"detail": [
{
"loc": ["body", "api_name"],
"msg": "field required",
"type": "value_error.missing"
}
]
}
// 请求
POST /apiv1/tracking/api_mapping
Content-Type: application/json
{
"api_path": "/apiv1/test",
"api_name": "测试"
}
// 预期响应 (HTTP 200, 业务码 401)
{
"statusCode": 401,
"msg": "未授权"
}
注意:此接口返回的未认证消息为"未授权",而非"未认证"。
/apiv1/tracking/api_mappings — 获取所有 API 映射功能说明: 查询所有 API 路径映射关系。返回全量数据,无分页、无过滤。
是否需要认证: 是(代码中显式检查 if not user)
请求方式: GET
请求参数: 无
成功响应字段: | 字段 | 类型 | 说明 | |------|------|------| | data[].id | int | 映射 ID | | data[].api_path | string | API 路径 | | data[].api_name | string | API 名称 | | data[].api_desc | string | API 描述 | | data[].status | int/string | 状态(具体含义取决于模型定义) |
测试用例:
// 请求
GET /apiv1/tracking/api_mappings
token: <有效Token>
// 预期响应 (HTTP 200)
{
"statusCode": 200,
"msg": "success",
"data": [
{
"id": 1,
"api_path": "/apiv1/chat/send",
"api_name": "发送聊天消息",
"api_desc": "用户发送聊天消息给AI,支持流式响应",
"status": 1
},
{
"id": 2,
"api_path": "/apiv1/get_hot_question",
"api_name": "获取热点问题",
"api_desc": "",
"status": 1
}
]
}
// 请求
GET /apiv1/tracking/api_mappings
token: <有效Token>
// 预期响应 (HTTP 200)
{
"statusCode": 200,
"msg": "success",
"data": []
}
// 请求
GET /apiv1/tracking/api_mappings
// 预期响应 (HTTP 200, 业务码 401)
{
"statusCode": 401,
"msg": "未授权"
}
| 依赖项 | 说明 |
|---|---|
database.get_db |
SQLAlchemy 数据库会话(Depends 注入) |
database.get_unix |
获取当前 Unix 时间戳的工具函数 |
models.tracking.TrackingRecord |
埋点记录模型(字段:id, user_id, api_path, api_name, method, request_id, ip_address, created_at) |
models.tracking.ApiPathMapping |
API 路径映射模型(字段:id, api_path, api_name, api_desc, status, created_at) |
uuid |
Python 标准库,用于生成 UUID v4 |
request.state.user |
从中间件注入的用户信息(含 user_id) |
record 接口使用 Pydantic 模型 TrackingRequest 接收参数, 比 points.py 中的 request.json() 方式更规范,自动处理参数校验。record 接口的 method 字段记录的是当前请求的 HTTP 方法(POST), 而非被埋点的目标 API 的请求方法。如果需要记录目标 API 的方法,需要在请求体中额外传入。records 接口使用 limit 而非分页模式, 无法获取第 100 条之后的数据(需调大 limit)。不返回 total 总数。records 接口响应字段较少, 不包含 method、request_id、ip_address 等存储在数据库中的字段。api_mapping 和 api_mappings 接口使用显式的 if not user 检查, 返回的错误消息为"未授权"(而非其他接口常用的"未认证")。api_mapping 通过 api_path 做唯一性检查, 同一路径不可重复添加映射。但不支持更新或删除已有映射。api_mappings 返回全量数据, 无分页、无筛选条件。当映射数据量较大时可能存在性能问题。get_user_data_id 已移至 routers/total.py, 避免路由重复注册。