23 KiB
LSFX Mock Random Hit Rule 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 稳定随机命中一部分,同时补齐 LARGE_PURCHASE_TRANSACTION 所需的最小数据库基线数据。
Architecture: 保持现有 FileService -> StatementService -> 缓存分页 主链路不变,不新增平行模块或兼容性双轨。FileService 只负责生成并保存规则命中计划,statement_rule_samples.py 只负责按规则代码装配命中样本,StatementService 负责把命中样本与噪声流水合并后缓存;LARGE_PURCHASE_TRANSACTION 不伪造成银行流水,单独通过 SQL 基线脚本补齐真实来源表数据。
Tech Stack: Python 3, FastAPI, pytest, PyMySQL, MySQL, Bash
File Structure
lsfx-mock-server/services/file_service.py: 在FileRecord中保存稳定随机的规则命中计划,并让上传链路与行内流水链路都具备一致的计划字段。lsfx-mock-server/services/statement_rule_samples.py: 维护大额交易与第一期新增规则的样本 builder 映射,按命中计划生成最小命中样本集合。lsfx-mock-server/services/statement_service.py: 从FileRecord读取规则命中计划,拼装命中样本与噪声流水,统一分配 ID 并缓存。lsfx-mock-server/tests/test_file_service.py: 锁定命中计划的生成与持久化语义。lsfx-mock-server/tests/test_statement_service.py: 锁定样本装配、缓存稳定性与按规则子集命中。lsfx-mock-server/tests/integration/test_full_workflow.py: 验证getJZFileOrZjrcuFile -> getBSByLogId端到端链路。sql/migration/2026-03-20-lsfx-mock-random-hit-rule-purchase-baseline.sql: 为LARGE_PURCHASE_TRANSACTION补最小采购业务数据。docs/reports/implementation/2026-03-20-lsfx-mock-random-hit-rule-backend-record.md: 记录本次后端实施内容。docs/tests/records/2026-03-20-lsfx-mock-random-hit-rule-backend-verification.md: 记录测试与数据库验证结果。
Task 1: 在 FileService 中持久化稳定随机的规则命中计划
Files:
-
Modify:
lsfx-mock-server/services/file_service.py -
Modify:
lsfx-mock-server/tests/test_file_service.py -
Reference:
docs/design/2026-03-20-lsfx-mock-random-hit-rule-design.md -
Step 1: Write the failing test
在 lsfx-mock-server/tests/test_file_service.py 中先补两条最小失败用例,锁定“同一 logId 命中计划稳定”和“fetch_inner_flow() 会把命中计划落到 FileRecord”:
def test_build_rule_hit_plan_should_be_deterministic_for_same_log_id():
service = FileService(staff_identity_repository=FakeStaffIdentityRepository())
plan1 = service._build_rule_hit_plan(10001)
plan2 = service._build_rule_hit_plan(10001)
assert plan1 == plan2
assert 2 <= len(plan1["large_transaction_hit_rules"]) <= 4
assert 2 <= len(plan1["phase1_hit_rules"]) <= 4
def test_fetch_inner_flow_should_persist_rule_hit_plan(monkeypatch):
service = FileService(staff_identity_repository=FakeStaffIdentityRepository())
monkeypatch.setattr(
service,
"_build_rule_hit_plan",
lambda log_id: {
"large_transaction_hit_rules": ["HOUSE_OR_CAR_EXPENSE", "TAX_EXPENSE"],
"phase1_hit_rules": ["GAMBLING_SENSITIVE_KEYWORD", "FOREX_BUY_AMT"],
},
)
response = service.fetch_inner_flow(
{
"groupId": 1001,
"customerNo": "test_customer_001",
"dataChannelCode": "test_code",
"requestDateId": 20240101,
"dataStartDateId": 20240101,
"dataEndDateId": 20240131,
"uploadUserId": 902001,
}
)
log_id = response["data"][0]
record = service.file_records[log_id]
assert record.large_transaction_hit_rules == ["HOUSE_OR_CAR_EXPENSE", "TAX_EXPENSE"]
assert record.phase1_hit_rules == ["GAMBLING_SENSITIVE_KEYWORD", "FOREX_BUY_AMT"]
- Step 2: Run test to verify it fails
Run:
cd lsfx-mock-server
pytest tests/test_file_service.py -k "rule_hit_plan or persist_rule_hit_plan" -v
Expected:
-
FAIL -
原因是
FileRecord尚未保存规则命中计划,FileService也没有_build_rule_hit_plan()能力 -
Step 3: Write minimal implementation
在 lsfx-mock-server/services/file_service.py 中只做最小改动:
- 为
FileRecord新增两个字段,采用现有 Python 代码风格的 snake_case:
large_transaction_hit_rules: List[str] = field(default_factory=list)
phase1_hit_rules: List[str] = field(default_factory=list)
- 新增固定规则池与稳定随机 helper:
def _build_rule_hit_plan(self, log_id: int) -> dict:
rng = random.Random(f"rule-plan:{log_id}")
return {
"large_transaction_hit_rules": self._pick_rule_subset(rng, LARGE_TRANSACTION_RULE_CODES, 2, 4),
"phase1_hit_rules": self._pick_rule_subset(rng, PHASE1_RULE_CODES, 2, 4),
}
- 在
_create_file_record()、upload_file()、fetch_inner_flow()中都把规则命中计划写入FileRecord,避免getBSByLogId读取上传链路记录时出现字段缺失。
- Step 4: Run test to verify it passes
Run:
cd lsfx-mock-server
pytest tests/test_file_service.py -k "rule_hit_plan or persist_rule_hit_plan" -v
Expected:
-
PASS -
FileRecord已具备稳定随机的规则命中计划 -
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: 将 statement_rule_samples.py 拆成按规则代码装配的样本生成器
Files:
-
Modify:
lsfx-mock-server/services/statement_rule_samples.py -
Modify:
lsfx-mock-server/tests/test_statement_service.py -
Step 1: Write the failing test
在 lsfx-mock-server/tests/test_statement_service.py 中补两条失败用例,锁定“只生成被选中的规则样本,而不是默认全量命中”:
def test_build_seed_statements_for_rule_plan_should_only_include_requested_phase1_rules():
plan = {
"large_transaction_hit_rules": [],
"phase1_hit_rules": ["GAMBLING_SENSITIVE_KEYWORD", "FOREX_BUY_AMT"],
}
statements = build_seed_statements_for_rule_plan(
group_id=1000,
log_id=20001,
rule_plan=plan,
)
assert any("游戏" in item["userMemo"] for item in statements)
assert any("购汇" in item["userMemo"] for item in statements)
assert not any("证券" in item["userMemo"] for item in statements)
def test_build_seed_statements_for_rule_plan_should_generate_withdraw_cnt_samples():
plan = {
"large_transaction_hit_rules": [],
"phase1_hit_rules": ["WITHDRAW_CNT"],
}
statements = build_seed_statements_for_rule_plan(
group_id=1000,
log_id=20001,
rule_plan=plan,
)
assert len([
item for item in statements
if "微信提现" in item["userMemo"] or "支付宝提现" in item["userMemo"]
]) >= 4
- Step 2: Run test to verify it fails
Run:
cd lsfx-mock-server
pytest tests/test_statement_service.py -k "rule_plan_should_only_include or withdraw_cnt_samples" -v
Expected:
-
FAIL -
原因是当前样本模块仍然是固定大额交易全量样本,没有“按规则计划装配”入口
-
Step 3: Write minimal implementation
在 lsfx-mock-server/services/statement_rule_samples.py 中做结构性改造,但只落最短路径:
- 保留现有基础 identity 与
_build_statement()工具函数 - 将大额交易样本拆成按规则代码可调用的 builder
- 新增第一期规则 builder 映射
- 提供统一入口
build_seed_statements_for_rule_plan(...)
最小结构如下:
LARGE_TRANSACTION_BUILDERS = {
"HOUSE_OR_CAR_EXPENSE": build_house_or_car_samples,
"TAX_EXPENSE": build_tax_samples,
"SINGLE_LARGE_INCOME": build_single_large_income_samples,
"CUMULATIVE_INCOME": build_cumulative_income_samples,
"ANNUAL_TURNOVER": build_annual_turnover_supporting_samples,
"LARGE_CASH_DEPOSIT": build_large_cash_deposit_samples,
"FREQUENT_CASH_DEPOSIT": build_frequent_cash_deposit_samples,
"LARGE_TRANSFER": build_large_transfer_samples,
}
PHASE1_RULE_BUILDERS = {
"GAMBLING_SENSITIVE_KEYWORD": build_gambling_sensitive_keyword_samples,
"SPECIAL_AMOUNT_TRANSACTION": build_special_amount_transaction_samples,
"SUSPICIOUS_INCOME_KEYWORD": build_suspicious_income_keyword_samples,
"FOREX_BUY_AMT": build_forex_buy_samples,
"FOREX_SELL_AMT": build_forex_sell_samples,
"STOCK_TFR_LARGE": build_stock_transfer_large_samples,
"LARGE_STOCK_TRADING": build_large_stock_trading_samples,
"WITHDRAW_CNT": build_withdraw_cnt_samples,
}
def build_seed_statements_for_rule_plan(group_id, log_id, rule_plan, **kwargs):
statements = []
for rule_code in rule_plan["large_transaction_hit_rules"]:
statements.extend(LARGE_TRANSACTION_BUILDERS[rule_code](group_id, log_id, **kwargs))
for rule_code in rule_plan["phase1_hit_rules"]:
statements.extend(PHASE1_RULE_BUILDERS[rule_code](group_id, log_id, **kwargs))
return statements
要求:
-
不再默认返回全量命中样本
-
每个 builder 只构造命中该规则所需的最小流水集合
-
WITHDRAW_CNT虽然是对象型规则,但依然通过银行流水样本构造触发条件 -
Step 4: Run test to verify it passes
Run:
cd lsfx-mock-server
pytest tests/test_statement_service.py -k "rule_plan_should_only_include or withdraw_cnt_samples" -v
Expected:
-
PASS -
样本模块已按规则子集装配,不再默认全量命中
-
Step 5: Commit
git add lsfx-mock-server/services/statement_rule_samples.py lsfx-mock-server/tests/test_statement_service.py
git commit -m "拆分Mock规则样本构造器"
Task 3: 让 StatementService 按命中计划生成样本并保持缓存稳定
Files:
-
Modify:
lsfx-mock-server/services/statement_service.py -
Modify:
lsfx-mock-server/tests/test_statement_service.py -
Modify:
lsfx-mock-server/tests/integration/test_full_workflow.py -
Step 1: Write the failing test
先在 lsfx-mock-server/tests/test_statement_service.py 补一条服务层测试,明确 StatementService 必须读取 FileRecord 中的命中计划,而不是继续默认全量样本:
def test_generate_statements_should_follow_rule_hit_plan_from_file_record():
file_service = FileService(staff_identity_repository=FakeStaffIdentityRepository())
statement_service = StatementService(file_service=file_service)
response = file_service.fetch_inner_flow(
{
"groupId": 1001,
"customerNo": "customer_rule_plan",
"dataChannelCode": "test_code",
"requestDateId": 20240101,
"dataStartDateId": 20240101,
"dataEndDateId": 20240131,
"uploadUserId": 902001,
}
)
log_id = response["data"][0]
record = file_service.file_records[log_id]
record.large_transaction_hit_rules = ["HOUSE_OR_CAR_EXPENSE"]
record.phase1_hit_rules = ["GAMBLING_SENSITIVE_KEYWORD"]
statements = statement_service._generate_statements(group_id=1001, log_id=log_id, count=200)
assert any("房产首付款" in item["userMemo"] for item in statements)
assert any("游戏" in item["userMemo"] for item in statements)
assert not any("购汇" in item["userMemo"] for item in statements)
再在 lsfx-mock-server/tests/integration/test_full_workflow.py 补一条链路测试,锁定同一 logId 首次生成后分页稳定:
def test_inner_flow_bank_statement_should_keep_same_rule_subset(client):
fetch_response = client.post(
"/watson/api/project/getJZFileOrZjrcuFile",
data={
"groupId": 1001,
"customerNo": "customer_subset",
"dataChannelCode": "test_code",
"requestDateId": 20240101,
"dataStartDateId": 20240101,
"dataEndDateId": 20240131,
"uploadUserId": 902001,
},
)
log_id = fetch_response.json()["data"][0]
page1 = client.post(
"/watson/api/project/getBSByLogId",
data={"groupId": 1001, "logId": log_id, "pageNow": 1, "pageSize": 10},
).json()
page2 = client.post(
"/watson/api/project/getBSByLogId",
data={"groupId": 1001, "logId": log_id, "pageNow": 1, "pageSize": 10},
).json()
assert page1["data"]["bankStatementList"] == page2["data"]["bankStatementList"]
- Step 2: Run test to verify it fails
Run:
cd lsfx-mock-server
pytest tests/test_statement_service.py -k "follow_rule_hit_plan" -v
pytest tests/integration/test_full_workflow.py -k "same_rule_subset" -v
Expected:
-
FAIL -
原因是
StatementService还没有从FileRecord读取规则命中计划,也没有按子集装配样本 -
Step 3: Write minimal implementation
在 lsfx-mock-server/services/statement_service.py 中接通规则命中计划:
from services.statement_rule_samples import build_seed_statements_for_rule_plan
def _generate_statements(self, group_id: int, log_id: int, count: int) -> List[Dict]:
record = self.file_service.get_file_record(log_id) if self.file_service is not None else None
rule_plan = {
"large_transaction_hit_rules": record.large_transaction_hit_rules if record else [],
"phase1_hit_rules": record.phase1_hit_rules if record else [],
}
seeded_statements = build_seed_statements_for_rule_plan(
group_id=group_id,
log_id=log_id,
rule_plan=rule_plan,
primary_enterprise_name=primary_enterprise_name,
primary_account_no=primary_account_no,
staff_id_card=record.staff_id_card if record else None,
family_id_cards=record.family_id_cards if record else None,
)
total_count = max(count, len(seeded_statements))
statements = list(seeded_statements)
for _ in range(total_count - len(seeded_statements)):
statements.append(
self._generate_random_statement(
group_id,
log_id,
primary_enterprise_name,
primary_account_no,
allowed_identity_cards,
rng,
)
)
return self._assign_statement_ids(statements, group_id, log_id)
要求:
-
命中样本生成完后仍统一走
_assign_statement_ids() -
继续保留 200 条固定总数和缓存语义
-
未命中的规则绝不回退为默认全量样本
-
Step 4: Run test to verify it passes
Run:
cd lsfx-mock-server
pytest tests/test_statement_service.py -k "follow_rule_hit_plan or fixed_total_count_200 or cached_result" -v
pytest tests/integration/test_full_workflow.py -k "same_rule_subset or share_same_primary_binding" -v
Expected:
-
PASS -
同一
logId的规则命中子集稳定,且主绑定信息没有回归 -
Step 5: Commit
git add lsfx-mock-server/services/statement_service.py lsfx-mock-server/tests/test_statement_service.py lsfx-mock-server/tests/integration/test_full_workflow.py
git commit -m "接通Mock随机命中流水生成链路"
Task 4: 为 LARGE_PURCHASE_TRANSACTION 补最小数据库基线脚本
Files:
-
Create:
sql/migration/2026-03-20-lsfx-mock-random-hit-rule-purchase-baseline.sql -
Reference:
sql/ccdi_purchase_transaction.sql -
Reference:
ccdi-project/src/main/resources/mapper/ccdi/project/CcdiBankTagAnalysisMapper.xml -
Step 1: Write the failing verification query
先准备最小验证脚本思路,确认当前库里还没有本次联调专用的采购事项基线数据:
import pymysql
cursor.execute(
"SELECT COUNT(1) AS cnt FROM ccdi_purchase_transaction WHERE purchase_id = 'LSFXMOCKPUR001'"
)
assert cursor.fetchone()["cnt"] == 0
- Step 2: Run verification to confirm baseline data is absent
Run:
python3 - <<'PY'
import pymysql
from pathlib import Path
import re
text = Path('ruoyi-admin/src/main/resources/application-dev.yml').read_text(encoding='utf-8')
match = re.search(r"url:\s*jdbc:mysql://(?P<host>[^:/?#]+):(?P<port>\d+)/(?P<db>[^?\n]+).*?\n\s*username:\s*(?P<user>[^\n]+)\n\s*password:\s*(?P<pwd>[^\n]+)", text, re.S)
conn = pymysql.connect(
host=match.group('host'),
port=int(match.group('port')),
user=match.group('user').strip(),
password=match.group('pwd').strip(),
database=match.group('db').strip(),
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor,
)
with conn, conn.cursor() as cursor:
cursor.execute("SELECT COUNT(1) AS cnt FROM ccdi_purchase_transaction WHERE purchase_id = 'LSFXMOCKPUR001'")
print(cursor.fetchone()['cnt'])
PY
Expected:
-
输出
0 -
说明
LARGE_PURCHASE_TRANSACTION所需联调基线数据尚未落库 -
Step 3: Write minimal implementation
创建 sql/migration/2026-03-20-lsfx-mock-random-hit-rule-purchase-baseline.sql,只插入最小必要采购记录,不扩展多余字段。脚本建议采用“先删后插”的幂等方式:
DELETE FROM ccdi_purchase_transaction WHERE purchase_id = 'LSFXMOCKPUR001';
INSERT INTO ccdi_purchase_transaction (
purchase_id, purchase_category, project_name, subject_name, subject_desc,
purchase_qty, budget_amount, bid_amount, actual_amount, contract_amount, settlement_amount,
purchase_method, supplier_name, contact_person, contact_phone, supplier_uscc, supplier_bank_account,
apply_date, plan_approve_date, announce_date, bid_open_date, contract_sign_date,
expected_delivery_date, actual_delivery_date, acceptance_date, settlement_date,
applicant_id, applicant_name, apply_department,
purchase_leader_id, purchase_leader_name, purchase_department,
created_by, updated_by
)
SELECT
'LSFXMOCKPUR001', '设备采购', 'LSFX Mock 联调',
'反洗钱终端设备采购', '用于命中 LARGE_PURCHASE_TRANSACTION 真实规则',
1, 188000.00, 186000.00, 186000.00, 186000.00, 186000.00,
'竞争性谈判', '兰溪市联调供应链有限公司', '联调联系人', '13800000000', '91330781MOCKPUR001', '6222000000001234',
CURRENT_DATE, CURRENT_DATE, CURRENT_DATE, CURRENT_DATE, CURRENT_DATE,
CURRENT_DATE, CURRENT_DATE, CURRENT_DATE, CURRENT_DATE,
CAST(s.staff_id AS CHAR), s.name, '纪检初核部',
NULL, NULL, NULL,
'admin', 'admin'
FROM ccdi_base_staff s
WHERE COALESCE(TRIM(CAST(s.staff_id AS CHAR)), '') <> ''
AND COALESCE(TRIM(s.name), '') <> ''
LIMIT 1;
要求:
-
只补一条采购记录
-
actual_amount必须大于 100000 -
不手写
mysql -e直接执行中文 SQL,后续执行必须使用bin/mysql_utf8_exec.sh -
Step 4: Run script and verify it works
Run:
bin/mysql_utf8_exec.sh sql/migration/2026-03-20-lsfx-mock-random-hit-rule-purchase-baseline.sql
python3 - <<'PY'
import pymysql
from pathlib import Path
import re
text = Path('ruoyi-admin/src/main/resources/application-dev.yml').read_text(encoding='utf-8')
match = re.search(r"url:\s*jdbc:mysql://(?P<host>[^:/?#]+):(?P<port>\d+)/(?P<db>[^?\n]+).*?\n\s*username:\s*(?P<user>[^\n]+)\n\s*password:\s*(?P<pwd>[^\n]+)", text, re.S)
conn = pymysql.connect(
host=match.group('host'),
port=int(match.group('port')),
user=match.group('user').strip(),
password=match.group('pwd').strip(),
database=match.group('db').strip(),
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor,
)
with conn, conn.cursor() as cursor:
cursor.execute("""
SELECT purchase_id, actual_amount, supplier_name
FROM ccdi_purchase_transaction
WHERE purchase_id = 'LSFXMOCKPUR001'
""")
print(cursor.fetchone())
PY
Expected:
-
SQL 脚本执行成功
-
查询结果返回
LSFXMOCKPUR001 -
actual_amount大于100000 -
Step 5: Commit
git add sql/migration/2026-03-20-lsfx-mock-random-hit-rule-purchase-baseline.sql
git commit -m "补齐Mock采购规则数据库基线"
Task 5: 做后端回归并沉淀实施与验证文档
Files:
-
Create:
docs/reports/implementation/2026-03-20-lsfx-mock-random-hit-rule-backend-record.md -
Create:
docs/tests/records/2026-03-20-lsfx-mock-random-hit-rule-backend-verification.md -
Reference:
docs/design/2026-03-20-lsfx-mock-random-hit-rule-design.md -
Reference:
docs/plans/backend/2026-03-20-lsfx-mock-random-hit-rule-backend-implementation.md -
Step 1: Run the full targeted regression
Run:
cd lsfx-mock-server
pytest tests/test_file_service.py tests/test_statement_service.py tests/test_api.py tests/integration/test_full_workflow.py -v
Expected:
-
PASS -
随机命中计划、样本装配、缓存稳定与端到端链路均通过
-
Step 2: Verify LARGE_PURCHASE_TRANSACTION baseline separately
Run:
python3 - <<'PY'
import pymysql
from pathlib import Path
import re
text = Path('ruoyi-admin/src/main/resources/application-dev.yml').read_text(encoding='utf-8')
match = re.search(r"url:\s*jdbc:mysql://(?P<host>[^:/?#]+):(?P<port>\d+)/(?P<db>[^?\n]+).*?\n\s*username:\s*(?P<user>[^\n]+)\n\s*password:\s*(?P<pwd>[^\n]+)", text, re.S)
conn = pymysql.connect(
host=match.group('host'),
port=int(match.group('port')),
user=match.group('user').strip(),
password=match.group('pwd').strip(),
database=match.group('db').strip(),
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor,
)
with conn, conn.cursor() as cursor:
cursor.execute("""
SELECT COUNT(1) AS cnt
FROM ccdi_purchase_transaction
WHERE purchase_id = 'LSFXMOCKPUR001'
AND actual_amount > 100000
""")
print(cursor.fetchone()['cnt'])
PY
Expected:
-
输出
1 -
说明采购基线数据已满足真实规则命中门槛
-
Step 3: Write implementation and verification records
在 docs/reports/implementation/2026-03-20-lsfx-mock-random-hit-rule-backend-record.md 中记录:
- 规则命中计划如何生成
- 样本模块如何按规则子集装配
StatementService如何读取计划并缓存LARGE_PURCHASE_TRANSACTION为什么单独补数据库基线
在 docs/tests/records/2026-03-20-lsfx-mock-random-hit-rule-backend-verification.md 中记录:
-
pytest执行时间与结果 -
SQL 基线脚本执行结果
-
采购基线查询结果
-
是否发现回归
-
Step 4: Check Git change scope
Run:
git status --short
Expected:
-
只包含本次任务相关文件
-
.DS_Store不纳入提交 -
Step 5: Commit
git add docs/reports/implementation/2026-03-20-lsfx-mock-random-hit-rule-backend-record.md docs/tests/records/2026-03-20-lsfx-mock-random-hit-rule-backend-verification.md
git commit -m "补充Mock随机命中后端实施记录"