wkc/ccdi

Files

2026-03-18 15:55:41 +08:00

15 KiB

Raw Blame History

LSFX Mock LogId Primary Binding Backend Implementation Plan

For agentic workers: REQUIRED: Use superpowers:subagent-driven-development (if subagents available) or superpowers:executing-plans to implement this plan. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 让 lsfx-mock-server 为上传文件和拉取本行信息生成的每个 logId 绑定唯一且稳定的本方主体与本方账号，并在上传响应、上传状态和银行流水查询中保持一致。

Architecture: 在 FileService 中统一生成并维护 logId -> 主体账号绑定，以 FileRecord 作为单一事实来源；上传文件与拉取本行信息两条链路都创建完整记录；StatementService 通过只读查询能力按 logId 取回绑定并写入每条流水的 leName/accountMaskNo，不再自行随机本方主体或本方账号。

Tech Stack: Python 3, FastAPI, pytest, httpx TestClient

执行结果

2026-03-18 已按计划完成 Task 1 至 Task 5，功能实现、实施记录与最终验证均已落地。
代码提交顺序：
- 0120d09 收敛Mock文件记录主体账号绑定模型
- 6fb7287 让拉取本行信息链路复用Mock主体账号绑定
- 0a85c09 统一Mock上传状态主体账号绑定优先级
- 5195617 让Mock流水查询复用logId主体账号绑定
最终验证已通过：
- python3 -m pytest tests/test_file_service.py -v
- python3 -m pytest tests/test_statement_service.py -v
- python3 -m pytest tests/test_api.py -v
- python3 -m pytest tests/integration/test_full_workflow.py -v
- python3 verify_implementation.py
实施细节见 docs/reports/implementation/2026-03-18-lsfx-logid-primary-binding-implementation.md

Task 1: 收敛 FileRecord 为单一主体账号绑定模型

Files:

Modify: lsfx-mock-server/services/file_service.py
Create: lsfx-mock-server/tests/test_file_service.py
Step 1: Write the failing test

先新增 test_file_service.py，锁定 FileRecord 和上传响应的最小语义：一个 logId 只对应一个主体与一个账号。

import pytest
from fastapi import BackgroundTasks, UploadFile
from io import BytesIO

from services.file_service import FileService


@pytest.mark.asyncio
async def test_upload_file_should_create_single_primary_binding():
    service = FileService()
    upload = UploadFile(filename="测试流水.csv", file=BytesIO(b"demo"))

    response = await service.upload_file(1000, upload, BackgroundTasks())

    log = response["data"]["uploadLogList"][0]
    assert len(log["enterpriseNameList"]) == 1
    assert len(log["accountNoList"]) == 1
    assert response["data"]["accountsOfLog"][str(log["logId"])][0]["accountName"] == log["enterpriseNameList"][0]
    assert response["data"]["accountsOfLog"][str(log["logId"])][0]["accountNo"] == log["accountNoList"][0]

Step 2: Run test to verify it fails

Run:

cd lsfx-mock-server
pytest tests/test_file_service.py -k primary_binding -v

Expected:

FAIL
原因是 tests/test_file_service.py 尚不存在，且 FileRecord 还没有明确的单值主体账号语义
Step 3: Write minimal implementation

在 file_service.py 中增加单值字段和统一生成方法，例如：

@dataclass
class FileRecord:
    primary_enterprise_name: str = ""
    primary_account_no: str = ""

PRIMARY_ENTERPRISE_POOL = [
    "测试主体A",
    "测试主体B",
    "测试主体C",
    "兰溪测试主体一部",
    "兰溪测试主体二部",
]


def _generate_primary_account_binding(self) -> tuple[str, str]:
    enterprise_name = random.choice(PRIMARY_ENTERPRISE_POOL)
    account_no = f"{random.randint(100000000000000, 999999999999999)}"
    return enterprise_name, account_no

并在 upload_file() 中回填：

enterprise_name, account_no = self._generate_primary_account_binding()
file_record.primary_enterprise_name = enterprise_name
file_record.primary_account_no = account_no
file_record.enterprise_name_list = [enterprise_name]
file_record.account_no_list = [account_no]

Step 4: Run test to verify it passes

Run:

cd lsfx-mock-server
pytest tests/test_file_service.py -k primary_binding -v

Expected:

PASS
说明上传接口返回已经围绕单一主体账号绑定工作
Step 5: Commit

git add lsfx-mock-server/services/file_service.py lsfx-mock-server/tests/test_file_service.py
git commit -m "收敛Mock文件记录主体账号绑定模型"

Task 2: 让拉取本行信息链路也创建并保存绑定记录

Files:

Modify: lsfx-mock-server/services/file_service.py
Modify: lsfx-mock-server/tests/test_file_service.py
Modify: lsfx-mock-server/tests/test_api.py
Step 1: Write the failing test

补两个测试，一个测服务层，一个测 API 层：

def test_fetch_inner_flow_should_persist_file_record_for_returned_log_id():
    service = FileService()

    response = service.fetch_inner_flow({"groupId": 1000})
    log_id = response["data"][0]

    assert log_id in service.file_records
    record = service.file_records[log_id]
    assert record.primary_enterprise_name
    assert record.primary_account_no

def test_fetch_inner_flow_followed_by_upload_status_should_keep_same_binding(client, sample_inner_flow_request):
    flow_response = client.post("/watson/api/project/getJZFileOrZjrcuFile", data=sample_inner_flow_request)
    log_id = flow_response.json()["data"][0]

    status_response = client.get(f"/watson/api/project/bs/upload?groupId=1000&logId={log_id}")
    log = status_response.json()["data"]["logs"][0]

    assert len(log["enterpriseNameList"]) == 1
    assert len(log["accountNoList"]) == 1

Step 2: Run test to verify it fails

Run:

cd lsfx-mock-server
pytest tests/test_file_service.py -k fetch_inner_flow -v
pytest tests/test_api.py -k "fetch_inner_flow_followed_by_upload_status" -v

Expected:

FAIL
原因是 fetch_inner_flow() 当前只返回随机 logId，不会创建 FileRecord
Step 3: Write minimal implementation

把 fetch_inner_flow() 改成复用和上传相同的记录创建逻辑：

def _create_file_record(self, group_id: int, file_name: str, bank_name: str, template_name: str) -> FileRecord:
    ...

def fetch_inner_flow(self, request: Union[Dict, object]) -> Dict:
    group_id = request.get("groupId", 1000) if isinstance(request, dict) else request.groupId
    self.log_counter += 1
    log_id = self.log_counter
    file_record = self._create_file_record(group_id, f"inner_flow_{log_id}.csv", "ZJRCU", "ZJRCU_T251114")
    self.file_records[log_id] = file_record
    return {"code": "200", "data": [log_id], "status": "200", "successResponse": True}

Step 4: Run test to verify it passes

Run:

cd lsfx-mock-server
pytest tests/test_file_service.py -k fetch_inner_flow -v
pytest tests/test_api.py -k "fetch_inner_flow_followed_by_upload_status" -v

Expected:

PASS
说明拉取本行信息返回的 logId 已经能在后续链路中复用同一组主体账号
Step 5: Commit

git add lsfx-mock-server/services/file_service.py lsfx-mock-server/tests/test_file_service.py lsfx-mock-server/tests/test_api.py
git commit -m "让拉取本行信息链路复用Mock主体账号绑定"

Task 3: 统一上传状态接口优先读取真实记录

Files:

Modify: lsfx-mock-server/services/file_service.py
Modify: lsfx-mock-server/tests/test_file_service.py
Modify: lsfx-mock-server/tests/test_api.py
Step 1: Write the failing test

补一组“真实记录优先级”测试，防止 get_upload_status() 继续走即时随机分支覆盖真实绑定：

import pytest


@pytest.mark.asyncio
async def test_get_upload_status_should_prefer_existing_file_record_binding():
    service = FileService()
    upload = UploadFile(filename="测试流水.csv", file=BytesIO(b"demo"))
    response = await service.upload_file(1000, upload, BackgroundTasks())
    log_id = response["data"]["uploadLogList"][0]["logId"]

    status = service.get_upload_status(1000, log_id)
    log = status["data"]["logs"][0]

    assert log["enterpriseNameList"] == response["data"]["uploadLogList"][0]["enterpriseNameList"]
    assert log["accountNoList"] == response["data"]["uploadLogList"][0]["accountNoList"]

Step 2: Run test to verify it fails

Run:

cd lsfx-mock-server
pytest tests/test_file_service.py -k "prefer_existing_file_record" -v

Expected:

FAIL
原因是 get_upload_status() 当前只按 logId 做即时随机生成，不会优先读取 self.file_records
Step 3: Write minimal implementation

把 get_upload_status() 改成：

def get_upload_status(self, group_id: int, log_id: int = None) -> dict:
    logs = []
    if log_id:
        if log_id in self.file_records:
            logs.append(self._build_log_detail(self.file_records[log_id]))
        else:
            random.seed(log_id)
            logs.append(self._generate_deterministic_record(log_id, group_id))

同时确保 _generate_deterministic_record() 也遵循单一绑定语义：

enterpriseNameList 长度为 1
accountNoList 长度为 1
Step 4: Run test to verify it passes

Run:

cd lsfx-mock-server
pytest tests/test_file_service.py -v
pytest tests/test_api.py -k "upload_status or deterministic_data_generation" -v

Expected:

PASS
已有上传状态接口的确定性测试仍然成立
对真实记录会优先返回真实绑定
Step 5: Commit

git add lsfx-mock-server/services/file_service.py lsfx-mock-server/tests/test_file_service.py lsfx-mock-server/tests/test_api.py
git commit -m "统一Mock上传状态主体账号绑定优先级"

Task 4: 让 StatementService 按 logId 注入本方主体和账号

Files:

Modify: lsfx-mock-server/services/statement_service.py
Modify: lsfx-mock-server/services/file_service.py
Modify: lsfx-mock-server/routers/api.py
Create: lsfx-mock-server/tests/test_statement_service.py
Modify: lsfx-mock-server/tests/integration/test_full_workflow.py
Step 1: Write the failing test

新增服务层测试，锁定同一 logId 的所有流水都要复用同一组 leName/accountMaskNo：

from services.file_service import FileService
from services.statement_service import StatementService


def test_bank_statement_should_reuse_file_record_primary_binding():
    file_service = FileService()
    log_id = file_service.fetch_inner_flow({"groupId": 1000})["data"][0]
    record = file_service.file_records[log_id]

    service = StatementService(file_service=file_service)
    result = service.get_bank_statement({"groupId": 1000, "logId": log_id, "pageNow": 1, "pageSize": 20})

    statements = result["data"]["bankStatementList"]
    assert all(item["leName"] == record.primary_enterprise_name for item in statements)
    assert all(item["accountMaskNo"] == record.primary_account_no for item in statements)

再补一个集成断言：

def test_complete_workflow_should_keep_same_primary_binding_between_status_and_statement(client):
    ...

Step 2: Run test to verify it fails

Run:

cd lsfx-mock-server
pytest tests/test_statement_service.py -v
pytest tests/integration/test_full_workflow.py -k primary_binding -v

Expected:

FAIL
原因是 StatementService 目前仍写死 leName = 张传伟 并独立随机 accountMaskNo
Step 3: Write minimal implementation

给 StatementService 增加对 FileService 的只读依赖：

class StatementService:
    def __init__(self, file_service=None):
        self.file_service = file_service
        self._cache = {}

def _resolve_primary_binding(self, group_id: int, log_id: int) -> tuple[str, str]:
    if self.file_service and log_id in self.file_service.file_records:
        record = self.file_service.file_records[log_id]
        return record.primary_enterprise_name, record.primary_account_no
    return "测试主体A", f"{random.randint(100000000000000, 999999999999999)}"

在 _generate_random_statement() 中替换：

enterprise_name, account_no = self._resolve_primary_binding(group_id, log_id)
...
"accountMaskNo": account_no,
...
"leName": enterprise_name,

并在 lsfx-mock-server/routers/api.py 中把服务实例装配改成共享同一份 file_service：

file_service = FileService()
statement_service = StatementService(file_service=file_service)

Step 4: Run test to verify it passes

Run:

cd lsfx-mock-server
pytest tests/test_statement_service.py -v
pytest tests/integration/test_full_workflow.py -v

Expected:

PASS
同一 logId 的分页流水在 leName/accountMaskNo 上保持一致
Step 5: Commit

git add lsfx-mock-server/services/statement_service.py lsfx-mock-server/services/file_service.py lsfx-mock-server/tests/test_statement_service.py lsfx-mock-server/tests/integration/test_full_workflow.py
git commit -m "让Mock流水查询复用logId主体账号绑定"

Task 5: 补实施记录并完成最终验证

Files:

Create: docs/reports/implementation/2026-03-18-lsfx-logid-primary-binding-implementation.md
Modify: docs/plans/backend/2026-03-18-lsfx-logid-primary-binding-backend-implementation.md
Step 1: Write the implementation report skeleton

记录以下内容：

FileRecord 新增的绑定字段
上传文件/拉取本行信息/上传状态/查流水四处的联动
实际执行过的 pytest 命令
兼容逻辑优先级说明

建议骨架：

# LSFX Mock LogId 主体账号绑定实施记录

## 变更概述
- ...

## 验证记录
- `cd lsfx-mock-server && pytest tests/test_file_service.py -v`
- `cd lsfx-mock-server && pytest tests/test_statement_service.py -v`
- `cd lsfx-mock-server && pytest tests/test_api.py -v`

Step 2: Run final verification

Run:

cd lsfx-mock-server
pytest tests/test_file_service.py -v
pytest tests/test_statement_service.py -v
pytest tests/test_api.py -v
pytest tests/integration/test_full_workflow.py -v
python verify_implementation.py

Expected:

所有测试通过
接口字段完整性检查仍然通过
Step 3: Stop any started services

如果验证过程中启动了：

python main.py
uvicorn main:app --reload --host 0.0.0.0 --port 8000

结束时必须主动关闭进程，避免残留端口占用。

Step 4: Commit

git add docs/reports/implementation/2026-03-18-lsfx-logid-primary-binding-implementation.md docs/plans/backend/2026-03-18-lsfx-logid-primary-binding-backend-implementation.md
git commit -m "补充Mock主体账号绑定实施记录"

15 KiB Raw Blame History Unescape Escape

LSFX Mock LogId Primary Binding Backend Implementation Plan

执行结果

Task 1: 收敛 FileRecord 为单一主体账号绑定模型

Task 2: 让拉取本行信息链路也创建并保存绑定记录

Task 3: 统一上传状态接口优先读取真实记录

Task 4: 让 StatementService 按 logId 注入本方主体和账号

Task 5: 补实施记录并完成最终验证

15 KiB

Raw Blame History