ccdi/lsfx-mock-server

流水分析 Mock 服务器

基于 Python + FastAPI 的独立 Mock 服务器，用于模拟流水分析平台的 7 个核心接口，并提供征信解析 Mock 能力。

✨ 特性

• ✅ 完整的接口模拟 - 实现所有 7 个核心接口
• ✅ 文件解析延迟 - 使用 FastAPI 后台任务模拟 4 秒解析延迟
• ✅ 错误场景触发 - 通过 error_XXXX 标记触发所有 8 个错误码
• ✅ 征信解析 Mock - 支持稳定随机生成三大主题域征信 payload
• ✅ 自动 API 文档 - Swagger UI 和 ReDoc 自动生成
• ✅ 配置驱动 - JSON 模板文件，易于修改响应数据
• ✅ 零配置启动 - 开箱即用，无需数据库

🚀 快速开始

1. 安装依赖

pip install -r requirements.txt

2. 启动服务

python main.py --rule-hit-mode subset
python main.py --rule-hit-mode all

热重载启动请使用项目脚本入口：

python dev.py --reload --rule-hit-mode subset
python dev.py --reload --rule-hit-mode all

• subset：默认模式，按 logId 稳定随机命中部分规则
• all：全部兼容规则命中模式，会命中当前可共存的全部规则

补充说明：

• fetch_inner_flow 与上传链路会在内部生成 abnormal_account_hit_rules
• 当前异常账户规则样本包含 SUDDEN_ACCOUNT_CLOSURE 与 DORMANT_ACCOUNT_LARGE_ACTIVATION
• /watson/api/project/getBSByLogId 会沿用现有种子流水主链路，自动混入与异常账户事实匹配的命中流水样本

3. 访问 API 文档

• Swagger UI: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc

📖 使用示例

正常流程

import requests

# 1. 获取 Token
response = requests.post(
    "http://localhost:8000/account/common/getToken",
    json={
        "projectNo": "test_project_001",
        "entityName": "测试企业",
        "userId": "902001",
        "userName": "902001",
        "appId": "remote_app",
        "appSecretCode": "test_secret_code_12345",
        "role": "VIEWER",
        "orgCode": "902000",
        "departmentCode": "902000"
    }
)
token_data = response.json()
project_id = token_data["data"]["projectId"]

# 2. 上传文件
files = {"file": ("test.csv", open("test.csv", "rb"), "text/csv")}
response = requests.post(
    "http://localhost:8000/watson/api/project/remoteUploadSplitFile",
    files=files,
    data={"groupId": project_id}
)
log_id = response.json()["data"]["uploadLogList"][0]["logId"]

# 3. 轮询检查解析状态
import time
for i in range(10):
    response = requests.post(
        "http://localhost:8000/watson/api/project/upload/getpendings",
        json={"groupId": project_id, "inprogressList": str(log_id)}
    )
    result = response.json()
    if not result["data"]["parsing"]:
        print("解析完成")
        break
    time.sleep(1)

# 4. 获取银行流水
response = requests.post(
    "http://localhost:8000/watson/api/project/getBSByLogId",
    json={
        "groupId": project_id,
        "logId": log_id,
        "pageNow": 1,
        "pageSize": 10
    }
)

征信解析 Mock

curl -s -X POST http://localhost:8000/xfeature-mngs/conversation/htmlEval \
  -F model=LXCUSTALL \
  -F hType=PERSON \
  -F file=@./sample-credit.html

成功时返回：

{
  "message": "成功",
  "status_code": "0",
  "payload": {
    "lx_header": {},
    "lx_debt": {},
    "lx_publictype": {}
  }
}

调试错误码时，可在 model 中追加错误标记：

curl -s -X POST http://localhost:8000/xfeature-mngs/conversation/htmlEval \
  -F model=error_ERR_10001 \
  -F hType=PERSON \
  -F file=@./sample-credit.html

健康检查：

curl -s http://localhost:8000/credit/health

错误场景测试

# 触发 40101 错误（appId错误）
response = requests.post(
    "http://localhost:8000/account/common/getToken",
    json={
        "projectNo": "test_error_40101",  # 包含错误标记
        "entityName": "测试企业",
        "userId": "902001",
        "userName": "902001",
        "appId": "remote_app",
        "appSecretCode": "test_secret_code_12345",
        "role": "VIEWER",
        "orgCode": "902000",
        "departmentCode": "902000"
    }
)
# 返回: {"code": "40101", "message": "appId错误", ...}

🔧 配置说明

环境变量

创建 .env 文件（参考 .env.example）：

# 应用配置
APP_NAME=流水分析Mock服务
APP_VERSION=1.0.0
DEBUG=true

# 服务器配置
HOST=0.0.0.0
PORT=8000

# 模拟配置
PARSE_DELAY_SECONDS=4
MAX_FILE_SIZE=10485760
RULE_HIT_MODE=subset

响应模板

修改 config/responses/ 下的 JSON 文件可以自定义响应数据：

• token.json - Token 响应模板
• upload.json - 上传文件响应模板
• parse_status.json - 解析状态响应模板
• bank_statement.json - 银行流水响应模板

🐳 Docker 部署

使用 Docker

# 构建镜像
docker build -t lsfx-mock-server .

# 运行容器
docker run -d -p 8000:8000 --name lsfx-mock lsfx-mock-server

使用 Docker Compose

docker-compose up -d

📁 项目结构

lsfx-mock-server/
├── main.py                 # 应用入口
├── config/
│   ├── settings.py         # 全局配置
│   └── responses/          # 响应模板
├── models/
│   ├── request.py          # 请求模型
│   └── response.py         # 响应模型
├── services/
│   ├── token_service.py    # Token 管理
│   ├── file_service.py     # 文件上传和解析
│   └── statement_service.py # 流水数据管理
├── routers/
│   └── api.py              # API 路由
├── utils/
│   ├── error_simulator.py  # 错误模拟
│   └── response_builder.py # 响应构建器
└── tests/                 # 测试套件

🧪 运行测试

# 运行所有测试
pytest tests/ -v

# 生成覆盖率报告
pytest tests/ -v --cov=. --cov-report=html

🔌 API 接口列表

接口	方法	路径	描述
1	POST	/account/common/getToken	获取 Token
2	POST	/watson/api/project/remoteUploadSplitFile	上传文件
3	POST	/watson/api/project/getJZFileOrZjrcuFile	拉取行内流水（返回随机logId）
4	POST	/watson/api/project/upload/getpendings	检查解析状态
5	POST	/watson/api/project/batchDeleteUploadFile	删除文件
6	POST	/watson/api/project/getBSByLogId	获取银行流水
7	POST	/xfeature-mngs/conversation/htmlEval	征信解析 Mock
8	GET	/credit/health	征信解析健康检查

⚠️ 错误码列表

错误码	描述
40101	appId错误
40102	appSecretCode错误
40104	可使用项目次数为0，无法创建项目
40105	只读模式下无法新建项目
40106	错误的分析类型，不在规定的取值范围内
40107	当前系统不支持的分析类型
40108	当前用户所属行社无权限
501014	无行内流水文件

🛠️ 开发指南

添加新接口

1. 在 models/request.py 和 models/response.py 中添加模型
2. 在 services/ 中添加服务类
3. 在 routers/api.py 中添加路由
4. 在 config/responses/ 中添加响应模板
5. 编写测试

修改响应数据

直接编辑 config/responses/ 下的 JSON 文件，重启服务即可生效。

📝 License

MIT

🤝 Contributing

欢迎提交 Issue 和 Pull Request！