GaoYiYuan_0626
22f2a41192
阶段二失败（或阶段 1.5 引擎失败）后，仅凭 Core 侧数据把「详情 + 预警」两件事补回来
（PRD §7.1 / 架构 §5.4）。
落地（改 3 + 新增 2 脚本 + 测试 2 文件）
- convert_service：新增公开 compensate_convert(group_id, ...) —— 补偿的服务端单点入口
· 锁键 convert:rerun:{gid}，与 convert_fund 幂等重试路径同一个键
· 详情侧：非 completed 才补写，复用 _finalize_from_core（不另写第二份阶段二）
· 预警侧：幂等锚点 = 转出端 out_trade_id，复用 find_alerts_by_trade；
命中即 skipped，否则跑 process_convert_event
- risk_repository：has_engine_error_audit 加 decision 参数（默认值不变）
· convert 线阶段 1.5 失败审计用 engine_error，普通交易用 risk_engine_error，不是同一个码
- rebuild_alerts.py：新增 --convert-group（与 location 参数 trade_ids 互斥），薄封装
- 新增 scripts/agent/cleanup_pending_convert.py（架构 §2 与开发计划 §9 指定路径）：
超 convert_compensate_sla_hours 的 pending 占位 → status='expired'（标记不硬删，S2）
- 新增 scripts/dev/verify_convert_compensate.py：真库验证脚本（MySQL 8.0.46）
- 测试 +12：test_convert_service +4（补写 / 幂等 / missing / locked）、
test_demo_scripts +8（--convert-group 分派与接线 + cleanup 脚本）
state 四态与 CLI 退出码
- rebuilt(0) / skipped(0 幂等) / missing(1 零写入) / locked(3)
- 退出码 2 保留给 argparse 用法错误，故 locked 取 3
真库专属证据（sqlite 单测给不了的，本任务核心增量）
- status='expired' 在 MySQL ENUM 上被接受（sqlite 该列是 VARCHAR，写什么都收）
- created_at < cutoff 在 DATETIME(3) 上的时间边界正确（超时进候选 / 未超时不进 / 复跑幂等）
- input_summary 是真 JSON 列，而 has_engine_error_audit 用 LIKE 判定：脚本先断言
information_schema 的 DATA_TYPE='json' 再验命中，并反向断言决策码不匹配则不命中
顺带收口（用户指示）
- core_ro.concentration_profile 补 h.qty > 0，与 list_holdings 真正同口径
· ratio 不变（归零行市值为 0），但 rows 不再多出已清仓产品、不虚占截断判定位
· 新用例含跨出口一致性断言；突变验证：去掉 qty > 0 → 精准 1 条红
验证
- pytest -q → 731 passed / 3 skipped（基线 719 加 12，零回归）
- 突变验证 3 组精准命中：去掉幂等锚点（1 红）/ 去掉 status 过滤（2 红）/ 补偿无视锁（1 红）
- 真库 verify_convert_compensate.py 34/34，隔离数据零残留
- 全套 7 个真库脚本复跑零回归：seed 全 PASS / apply 24 / service 35 / engine 31 / lots 20 / tools 14 / compensate 34
104 lines
3.8 KiB
Python
104 lines
3.8 KiB
Python
"""超时 `pending` 转换占位清理（T-12 · 架构 §5.4 · PRD §7.1 I-3）。
|
||
|
||
背景：convert 阶段零先落一行 `risk_convert_detail(status='pending')` 占位，阶段二
|
||
回写 `completed`。若进程在阶段零与阶段一之间挂掉，这行占位会**永久滞留**，既污染
|
||
巡检又让人误判「有一笔转换在飞」。本脚本按 SLA（`convert_compensate_sla_hours`，
|
||
默认 24h）把超时占位置 `status='expired'`。
|
||
|
||
**标记不硬删（评审 S2）**：只 UPDATE 状态，**绝不 DELETE** —— 留痕供对账
|
||
（`expired` 是建表即含的 ENUM 值，无需 ALTER）。这与
|
||
`ConvertRepository.mark_expired` 是**同一份实现**，本脚本不复制 SQL。
|
||
|
||
用法与退出码：
|
||
python scripts/agent/cleanup_pending_convert.py # 按 SLA 24h 清理
|
||
python scripts/agent/cleanup_pending_convert.py --hours 1 # 覆盖 SLA（演练/测试）
|
||
python scripts/agent/cleanup_pending_convert.py --dry-run # 只报告不写库
|
||
|
||
0 = 清理完成（含「无候选」）；1 = MySQL 不可达或写库失败。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
# ① sys.path 引导项目根（rebuild_alerts / escalation_scan 先例）
|
||
ROOT = Path(__file__).resolve().parents[2]
|
||
sys.path.insert(0, str(ROOT))
|
||
|
||
from app.config.settings import settings # noqa: E402
|
||
from app.repository.convert_repository import ConvertRepository # noqa: E402
|
||
from app.utils.trace import new_trace # noqa: E402
|
||
|
||
|
||
def cleanup(repo: ConvertRepository, hours: int, *, dry_run: bool = False) -> dict:
|
||
"""扫出超 `hours` 的 pending 占位并置 `expired`；返回 JSON 友好摘要。
|
||
|
||
幂等：`mark_expired` 只把 `pending` 行改状态，已 `expired` 的行**不再进候选**
|
||
（`list_expired_candidates` 硬编码 `status='pending'`），故重复执行第二次为空操作。
|
||
"""
|
||
candidates = repo.list_expired_candidates(hours)
|
||
expired: list[str] = []
|
||
if not dry_run:
|
||
for row in candidates:
|
||
group_id = str(row["convert_group_id"])
|
||
repo.mark_expired(group_id)
|
||
expired.append(group_id)
|
||
return {
|
||
"sla_hours": hours,
|
||
"dry_run": dry_run,
|
||
"candidate_count": len(candidates),
|
||
"expired_count": len(expired),
|
||
"expired": expired,
|
||
"candidates": [
|
||
{
|
||
"convert_group_id": str(r["convert_group_id"]),
|
||
"client_request_id": r.get("client_request_id"),
|
||
"created_at": r.get("created_at"),
|
||
}
|
||
for r in candidates
|
||
],
|
||
}
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(
|
||
description="超时 pending 转换占位 → status='expired'（标记不硬删，S2）"
|
||
)
|
||
parser.add_argument(
|
||
"--hours",
|
||
type=int,
|
||
default=None,
|
||
help=f"覆盖 SLA 小时数（缺省取 settings.convert_compensate_sla_hours="
|
||
f"{settings.convert_compensate_sla_hours}）",
|
||
)
|
||
parser.add_argument("--dry-run", action="store_true", help="只报告候选，不写库")
|
||
args = parser.parse_args()
|
||
|
||
hours = args.hours if args.hours is not None else settings.convert_compensate_sla_hours
|
||
new_trace()
|
||
repo = ConvertRepository()
|
||
|
||
# 连接自检（零写入探针）
|
||
try:
|
||
repo.list_expired_candidates(hours)
|
||
except Exception as exc:
|
||
print(f"MySQL 不可达：{exc}", file=sys.stderr)
|
||
print("请确认本机 MySQL 服务已启动（见 FLOW §0 本机状态）。", file=sys.stderr)
|
||
return 1
|
||
|
||
try:
|
||
summary = cleanup(repo, hours, dry_run=args.dry_run)
|
||
except Exception as exc: # noqa: BLE001
|
||
print(f"清理失败：{exc}", file=sys.stderr)
|
||
return 1
|
||
|
||
print(json.dumps(summary, ensure_ascii=False, default=str, indent=2))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|