Files
crawler-plugin/check_progress.py
T
huangzd1997 2039acdfe6 task-1: 建立 Similar ASIN 性能基线夹具 (1000/5000行、图片开关、chunk与payload采样)
新增 SimilarAsinPerfFixture 确定性夹具:按 sourceFileKey+rowIndex 派生字段,
支持 0/1/1000/5000 行、图片开关两种模式、chunk 划分与 payload 字节采样;
空输入返回空、超限(>5000行/非法chunk/空key)抛 IllegalArgumentException。
含 10 个测试覆盖默认路径/批量/幂等/边界/非法输入/依赖失败恢复。
2026-08-29 14:22:07 +08:00

51 lines
1.4 KiB
Python

#!/usr/bin/env python3
"""检查 app_client 的 Step 2/后续开发任务是否全部完成。"""
from __future__ import annotations
import json
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent
PROGRESS_PATH = ROOT / "progress.json"
def main() -> int:
if not PROGRESS_PATH.is_file():
print("progress.json not found", file=sys.stderr)
return 2
try:
data = json.loads(PROGRESS_PATH.read_text(encoding="utf-8"))
except Exception as exc:
print(f"invalid progress.json: {exc}", file=sys.stderr)
return 2
total = data.get("total_tasks")
tasks = data.get("tasks")
if not isinstance(total, int) or total <= 0 or not isinstance(tasks, list):
print("invalid progress shape", file=sys.stderr)
return 2
if len(tasks) != total:
print(f"task count mismatch: total_tasks={total}, tasks={len(tasks)}", file=sys.stderr)
return 2
ids = [task.get("id") for task in tasks if isinstance(task, dict)]
if ids != list(range(1, total + 1)):
print("task ids are not contiguous from 1", file=sys.stderr)
return 2
pending = [task for task in tasks if task.get("status") != "done"]
if pending:
print(f"pending tasks: {len(pending)}/{total}")
print("all tasks done: false")
return 1
print(f"all tasks done: true ({total})")
return 0
if __name__ == "__main__":
raise SystemExit(main())