#!/usr/bin/env python3 """Generate the approved Apifox automation scope and evidence checklist. Only two folders count toward automation completion: * 01-store/PC端 * 02-ai_api/AI运动营养师 The generated rows start as NOT_RUN. A row may become PASS only after the real request, named assertions, API read-back (for writes), and direct database read-back have all produced evidence for the selected database environment. """ from __future__ import annotations import csv import json import sys from collections import Counter from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from import_openapi import METHODS, TARGET_PROJECT_ID, export_openapi, load_token # noqa: E402 ROOT = Path(__file__).resolve().parents[1] INVENTORY_DIR = ROOT / "inventory" STATUS_FIELDS = ( "mysql_status", "dm8_status", "last_run_at", "evidence_link", "defect_id", "notes", ) STORE_SCENES = { "S00-PC环境、首页与人员权限": ("首页", "人员管理", "身份管理"), "S01-膳食调查、监控与配餐": ("膳食调查", "膳食监控", "配餐管理"), "S02-餐厅、菜品与排菜": ("餐厅管理", "餐厅设置", "大众智慧餐厅", "AI菜品管理"), "S03-进销存与供应链": ("进销存",), "S04-订单、账户、卡务与账单": ("订单管理", "账户管理", "卡务管理", "账单记录", "中国银行", "邮储收款账户"), "S05-营销、评价与营养品": ("营销中心", "就餐评价", "营养品展示分类", "营养品管理", "营养资讯"), "S06-系统、审批、知识库与报表": ("系统管理", "审批流", "知识库管理", "统计报表", "AI"), } AI_SCENES = { "A00-AI营养师环境与认证": ("用户认证模块",), "A01-个人健康与设备数据": ("我的", "华为健康", "英吉多设备", "上禾体检一体机", "餐盘扫码"), "A02-订餐、订单与国信业务": ("国信订餐", "订餐", "订单"), "A03-AI问答与会话能力": ("AI问答", "CozeAI"), } def operation_folder(operation: dict) -> str: return str(operation.get("x-apifox-folder") or ((operation.get("tags") or ["未分类"])[0])) def scenario_for(service: str, folder: str) -> str: if service == "store": relative = folder.removeprefix("01-store/PC端").strip("/") if not relative: return "S00-PC环境、首页与人员权限" module = relative.split("/", 1)[0] mapping = STORE_SCENES else: relative = folder.removeprefix("02-ai_api/AI运动营养师").strip("/") if not relative: return "A00-AI营养师环境与认证" module = relative.split("/", 1)[0] mapping = AI_SCENES for scene, modules in mapping.items(): if module in modules: return scene raise ValueError(f"No scenario mapping for {service}: {folder}") def classify_risk(method: str, path: str, folder: str, summary: str) -> tuple[str, str]: text = " ".join((path, folder, summary)).lower() if path in {"/p/common/start", "/p/test/test"}: return "OBSOLETE_DOC", "开发确认无现行实现与前端调用；正式场景禁用，文档待清理" if path == "/p/login/login": return "AUTH_WITH_AUDIT", "登录成功、Token 脱敏断言；直接回查安全审计表" if path == "/p/menu/treeMenu": return "READ_ONLY", "HTTP+业务断言；菜单数量与首项 UUID、名称直接核对 ydy_menu" if path == "/p/Restaurant/index": return "READ_ONLY", "HTTP+业务断言；菜品总数、启用数和本月新增数直接核对 ydy_dishes" if path == "/api/identity/lists": return "READ_ONLY", "控制器和服务均为查询；核对分页、筛选及 ydy_identity_type 原表" if any(word in text for word in ("支付", "退款", "银行", "充值", "补贴", "pay", "refund", "bank")): return "PAYMENT", "仅可使用获批沙箱；否则 BLOCKED_EXTERNAL，不执行真实资金动作" if any(word in text for word in ("华为", "英吉多", "上禾", "coze", "设备", "device")): return "EXTERNAL_OR_DEVICE", "验证可控边界和响应契约；缺少设备/沙箱时标记 BLOCKED_EXTERNAL" if any(word in text for word in ("上传", "导入", "导出", "下载", "文件", "upload", "import", "export", "download")): return "FILE_OR_BULK", "真实小文件/小批量；核对响应、回查结果和资源清理" if method == "GET": return "READ_ONLY", "HTTP+业务断言；关键字段与数据库 SELECT 结果对照" if method == "DELETE" or any(word in text for word in ("删除", "remove", "delete")): return "DELETE_OR_STATE_CHANGE", "先创建隔离数据，再删除并通过 API 与数据库双重确认" if method in {"POST", "PUT", "PATCH"}: return "DB_WRITE_OR_COMMAND", "真实业务入参；写后 API 回读并直接查询数据库" return "REVIEW_REQUIRED", "执行前人工判定副作用与证据要求" def iter_rows(document: dict): for path, path_item in sorted((document.get("paths") or {}).items()): for method, operation in path_item.items(): upper = method.upper() if method.lower() not in METHODS or not isinstance(operation, dict): continue folder = operation_folder(operation) if folder == "01-store/PC端" or folder.startswith("01-store/PC端/"): service = "store" prefix = "01-store/PC端" elif folder == "02-ai_api/AI运动营养师" or folder.startswith("02-ai_api/AI运动营养师/"): service = "ai_api" prefix = "02-ai_api/AI运动营养师" else: continue relative = folder.removeprefix(prefix).strip("/") module = relative.split("/", 1)[0] if relative else "根目录" summary = str(operation.get("summary") or operation.get("operationId") or f"{upper} {path}") risk, evidence = classify_risk(upper, path, folder, summary) yield { "service": service, "folder": folder, "module": module, "scenario": scenario_for(service, folder), "method": upper, "path": path, "summary": summary, "operation_id": str(operation.get("operationId") or ""), "risk": risk, "evidence_required": evidence, "mysql_status": "NOT_RUN", "dm8_status": "NOT_RUN", "last_run_at": "", "evidence_link": "", "defect_id": "", "notes": "", } def load_existing_statuses(csv_path: Path) -> dict[tuple[str, str, str], dict[str, str]]: """Keep manual execution evidence when the live Apifox scope is regenerated.""" if not csv_path.exists(): return {} with csv_path.open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle) return { (row["service"], row["method"], row["path"]): { field: row.get(field, "") for field in STATUS_FIELDS } for row in reader } def main() -> int: token = load_token() status, document = export_openapi(token, TARGET_PROJECT_ID) if status != 200: raise RuntimeError(f"Apifox export failed: HTTP {status}") rows = list(iter_rows(document)) service_counts = Counter(row["service"] for row in rows) expected = {"store": 302, "ai_api": 46} if dict(service_counts) != expected: raise RuntimeError(f"Scope drift: expected {expected}, got {dict(service_counts)}") scene_counts = Counter(row["scenario"] for row in rows) expected_scenes = { "S00-PC环境、首页与人员权限": 22, "S01-膳食调查、监控与配餐": 28, "S02-餐厅、菜品与排菜": 64, "S03-进销存与供应链": 52, "S04-订单、账户、卡务与账单": 53, "S05-营销、评价与营养品": 27, "S06-系统、审批、知识库与报表": 56, "A00-AI营养师环境与认证": 10, "A01-个人健康与设备数据": 13, "A02-订餐、订单与国信业务": 9, "A03-AI问答与会话能力": 14, } if dict(scene_counts) != expected_scenes: raise RuntimeError(f"Scenario drift: expected {expected_scenes}, got {dict(scene_counts)}") csv_path = INVENTORY_DIR / "apifox-automation-scope.csv" INVENTORY_DIR.mkdir(parents=True, exist_ok=True) existing_statuses = load_existing_statuses(csv_path) for row in rows: preserved = existing_statuses.get((row["service"], row["method"], row["path"])) if preserved: row.update(preserved) if row["service"] == "store" and row["path"] == "/p/common/start": row.update({ "mysql_status": "NOT_APPLICABLE", "dm8_status": "NOT_APPLICABLE", "evidence_link": "reports/runs/S00_COMMON_START_NOT_APPLICABLE_20261008/report.md", "notes": "开发线程确认仅为过期 OpenAPI 条目，无现行代码实现或前端调用；Apifox 步骤已持久禁用", }) if row["service"] == "store" and row["path"] == "/p/test/test": row.update({ "mysql_status": "NOT_APPLICABLE", "dm8_status": "NOT_APPLICABLE", "evidence_link": "reports/runs/S00_TEST_TEMPLATE_NOT_APPLICABLE_20261008/report.md", "notes": "开发线程确认遗留模板接口，固定占位异常、无现行 PC 前端调用；Apifox 步骤已持久禁用", }) with csv_path.open("w", encoding="utf-8-sig", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) summary = { "apifoxProjectId": TARGET_PROJECT_ID, "scope": { "storeFolder": "01-store/PC端", "aiApiFolder": "02-ai_api/AI运动营养师", }, "total": len(rows), "services": dict(sorted(service_counts.items())), "scenarios": dict(sorted(scene_counts.items())), "methods": dict(sorted(Counter(row["method"] for row in rows).items())), "risks": dict(sorted(Counter(row["risk"] for row in rows).items())), "statuses": { "mysql": dict(Counter(row["mysql_status"] for row in rows)), "dm8": dict(Counter(row["dm8_status"] for row in rows)), }, "passRule": "请求断言 + 写后API回读 + 数据库直接回查；缺一项不得标记PASS", } summary_path = INVENTORY_DIR / "apifox-automation-scope-summary.json" summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps({"csv": str(csv_path), "summary": str(summary_path), **summary}, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())