#!/usr/bin/env python3 """Build and optionally apply an all-Chinese Apifox endpoint catalogue. The source must be a fresh OpenAPI export. Existing paths, methods, operationId values and schemas are preserved; only display summaries and endpoint folders are normalized. The generated mapping keeps the Apifox endpoint ID parsed from ``x-run-in-apifox`` so the post-import audit can prove endpoint identity. """ from __future__ import annotations import argparse import copy import importlib.util import json import re import time import urllib.error import urllib.request from pathlib import Path from typing import Any PROJECT_ID = 8_881_630 API_BASE = "https://api.apifox.com" METHODS = {"get", "post", "put", "patch", "delete"} SCRIPT_DIR = Path(__file__).resolve().parent SHARED_SCRIPT = SCRIPT_DIR / "sync_school_s00_s06_scenarios.py" FOLDER_RULES: list[tuple[str, str]] = [ ("/auth", "登录认证"), ("/school/ai", "学校端/人工智能事件"), ("/school/analytics", "学校端/数据分析"), ("/school/assets", "学校端/卫生与资产巡检"), ("/school/hygiene", "学校端/卫生与资产巡检"), ("/school/canteen", "学校端/食堂档案"), ("/school/cater", "学校端/供餐分类"), ("/school/certificates", "学校端/证件管理"), ("/school/credentials", "学校端/资质证照管理"), ("/school/companion-meals", "学校端/陪餐管理"), ("/school/device-iot", "学校端/物联网设备配置"), ("/school/device-safety", "学校端/设备安全"), ("/school/devices", "学校端/设备管理"), ("/school/dine", "学校端/就餐管理"), ("/school/disinfection", "学校端/消毒管理"), ("/school/emergency", "学校端/应急管理"), ("/school/foods", "学校端/菜品管理"), ("/school/home", "学校端/学校首页"), ("/school/inventory", "学校端/库存与进销存"), ("/school/meal-plans", "学校端/配餐计划"), ("/school/messages", "学校端/学校消息"), ("/school/morning-checks", "学校端/晨检管理"), ("/school/personnel", "学校端/人员管理"), ("/school/pesticide", "学校端/农残检测"), ("/school/public", "学校端/公示与公众反馈"), ("/school/risk", "学校端/风险管理"), ("/school/sampling", "学校端/食品留样"), ("/school/system", "学校端/系统管理"), ("/school/waste", "学校端/餐厨废弃物"), ("/school", "学校端/其他接口"), ("/reg/ai", "监管端/人工智能事件"), ("/reg/alerts", "监管端/预警管理"), ("/reg/credentials", "监管端/资质证照监管"), ("/reg/food-waste", "监管端/餐厨废弃物监管"), ("/reg/inspections", "监管端/检查任务"), ("/reg/ledgers", "监管端/台账监管"), ("/reg/overview", "监管端/首页总览"), ("/reg/reports", "监管端/统计报告"), ("/reg/schools", "监管端/学校管理"), ("/reg/system", "监管端/系统管理"), ("/reg", "监管端/其他接口"), ("/screen", "大屏端/数据分析"), ("/bright-local", "明厨亮灶/本地视频管理"), ("/bright", "明厨亮灶/视频监管"), ("/gateway/openApi/aet6control", "人工智能盒子开放接口/设备控制"), ("/gateway/openApi/AnalysisType", "人工智能盒子开放接口/算法类型"), ("/gateway/openApi/autoinfo", "人工智能盒子开放接口/车辆信息"), ("/gateway/openApi/autolib", "人工智能盒子开放接口/车辆库"), ("/gateway/openApi/autorecord", "人工智能盒子开放接口/车辆记录"), ("/gateway/openApi/channel", "人工智能盒子开放接口/通道管理"), ("/gateway/openApi/faceinfo", "人工智能盒子开放接口/人员信息"), ("/gateway/openApi/facelib", "人工智能盒子开放接口/人脸库"), ("/gateway/openApi/token", "人工智能盒子开放接口/认证"), ("/gateway/openApi/videoSheet", "人工智能盒子开放接口/视频任务"), ("/gateway/openApi", "人工智能盒子开放接口/系统管理"), ("/openApi/view", "人工智能盒子开放接口/视频查看"), ("/openApi/info", "人工智能盒子开放接口/设备信息"), ("/openApi", "人工智能盒子开放接口/设备上报"), ("/gatewayGBS", "第三方设备数据接入/视频平台"), ("/trustivs", "第三方设备数据接入/视频平台代理"), ("/sakura", "第三方设备数据接入/留样仪器"), ("/sample", "第三方设备数据接入/留样仪器"), ("/push", "第三方设备数据接入/设备同步"), ("/wosapi", "第三方设备数据接入/晨检设备"), ("/deviceIot", "第三方设备数据接入/晨检设备"), ("/device", "第三方设备数据接入/晨检设备"), ("/inspectionRecord", "第三方设备数据接入/晨检设备"), ("/uploadInspectionRecord", "第三方设备数据接入/晨检设备"), ("/system/group/device", "第三方设备数据接入/晨检设备"), ("/dewo", "第三方设备数据接入/晨检设备"), ("/control", "第三方设备数据接入/设备上报"), ("/cabinet", "第三方设备数据接入/设备心跳"), ("/regulator", "第三方设备数据接入/监管数据"), ("/api", "第三方设备数据接入/通用集成"), ("/iot", "物联网设备/视频设备"), ("/audit", "审计日志"), ("/files", "文件服务"), ("/static", "文件服务"), ("/common", "文件服务"), ("/health", "公共与系统/健康检查"), ("/home", "公共与系统/首页数据"), ("/system/storage", "公共与系统/存储配置"), ("/reports", "公共与系统/统计报告"), ("/public", "公共与系统/公众反馈"), ("/sampling", "公共与系统/留样人员"), ("/", "公共与系统/根接口"), ] RESOURCE_RULES: list[tuple[str, str]] = [ ("/auth/change-password", "当前用户密码"), ("/auth/zhct-login", "智慧食安外部登录"), ("/auth/login", "用户登录"), ("/auth/logout", "用户退出"), ("/auth/refresh", "登录令牌"), ("/auth/me", "当前用户信息"), ("/school/ai/events/summary", "人工智能事件汇总"), ("/school/ai/events", "人工智能事件"), ("/school/ai/home/summary", "首页人工智能事件汇总"), ("/school/ai/strangers", "陌生人事件"), ("/school/ai/summary/pie", "人工智能事件分类统计"), ("/school/ai/types", "人工智能事件类型"), ("/school/analytics/personnel-health", "人员健康分析"), ("/school/analytics/alerts", "预警分析"), ("/school/analytics/dashboard", "数据分析看板"), ("/school/analytics/health", "健康数据分析"), ("/school/analytics/stream", "数据分析实时流"), ("/school/assets/maintenance", "资产维护记录"), ("/school/hygiene/inspections", "卫生检查记录"), ("/school/hygiene/stream", "卫生检查实时流"), ("/school/canteen-detail", "食堂详情"), ("/school/canteen-licenses", "食堂证照"), ("/school/canteens/summary", "食堂汇总"), ("/school/canteens", "食堂"), ("/school/cater/categories", "供餐分类"), ("/school/cater/select", "外部供餐数据"), ("/school/certificates", "学校证件"), ("/school/credentials/canteens", "食堂资质"), ("/school/credentials/suppliers", "供应商资质"), ("/school/credentials/workers", "从业人员资质"), ("/school/credentials/exceptions", "资质异常"), ("/school/credentials/stream", "资质异常实时流"), ("/school/companion-meals", "陪餐记录"), ("/school/device-iot", "设备接入配置"), ("/school/device-safety", "设备安全记录"), ("/school/devices/statuses", "设备状态"), ("/school/devices/types", "设备类型"), ("/school/devices", "学校设备"), ("/school/dine/meal-options", "餐别选项"), ("/school/dine/meals", "就餐汇总"), ("/school/dine/records", "就餐记录"), ("/school/dine/stream", "就餐实时流"), ("/school/dine/qr", "就餐二维码"), ("/school/dine/camera", "就餐摄像头数据"), ("/school/disinfection/records", "消毒记录"), ("/school/disinfection/stream", "消毒实时流"), ("/school/disinfection/device", "消毒设备数据"), ("/school/emergency/alerts", "应急告警"), ("/school/emergency/cameras", "应急摄像头"), ("/school/emergency/duty", "应急值班"), ("/school/emergency/events", "应急事件"), ("/school/emergency/meetings", "应急会议"), ("/school/emergency/overview", "应急总览"), ("/school/emergency/plans", "应急预案"), ("/school/emergency/resources", "应急资源"), ("/school/emergency/stream", "应急实时流"), ("/school/emergency/tasks", "应急任务"), ("/school/emergency/notify", "应急通知"), ("/school/foods", "学校菜品"), ("/school/home/overview", "学校首页总览"), ("/school/inventory/additives", "食品添加剂"), ("/school/inventory/categories", "库存分类"), ("/school/inventory/inbound/doc", "入库单据"), ("/school/inventory/inbound", "入库记录"), ("/school/inventory/outbound/batches", "出库批次"), ("/school/inventory/outbound", "出库记录"), ("/school/inventory/products", "库存商品"), ("/school/inventory/stock", "库存"), ("/school/inventory/suppliers", "供应商"), ("/school/inventory/tickets", "采购票据"), ("/school/inventory/warehouses", "仓库"), ("/school/inventory/stream", "库存实时流"), ("/school/meal-plans", "配餐计划"), ("/school/messages/rectify", "学校整改消息"), ("/school/morning-checks", "晨检记录"), ("/school/personnel", "学校人员"), ("/school/pesticide/records", "农残检测记录"), ("/school/pesticide/stream", "农残检测实时流"), ("/school/pesticide/device", "农残检测设备数据"), ("/school/public/config/feedback/qr", "公众反馈二维码"), ("/school/public/config/audit", "公示配置审计记录"), ("/school/public/config", "公示配置"), ("/school/public/feedback", "公众反馈"), ("/school/risk/catalog", "风险目录"), ("/school/risk/reports", "风险报告"), ("/school/risk/tasks", "风险任务"), ("/school/sampling/operators", "留样人员"), ("/school/sampling/records", "食品留样记录"), ("/school/sampling/stream", "食品留样实时流"), ("/school/sampling/device", "留样设备数据"), ("/school/system/announcements", "系统公告"), ("/school/system/apps", "学校应用"), ("/school/system/canteen", "学校食堂配置"), ("/school/system/linkage", "学校联动申请"), ("/school/system/meals", "学校餐别配置"), ("/school/system/news", "学校新闻"), ("/school/system/org-type", "学校机构类型"), ("/school/system/permissions", "学校权限"), ("/school/system/roles", "学校角色"), ("/school/system/users", "学校用户"), ("/school/waste/categories", "餐厨废弃物分类"), ("/school/waste/records", "餐厨废弃物记录"), ("/reg/ai/broadcast", "人工智能事件播报"), ("/reg/ai/events", "监管人工智能事件"), ("/reg/ai/methods", "人工智能处置方式"), ("/reg/ai/schools", "人工智能监管学校"), ("/reg/ai/tasks", "人工智能监管任务"), ("/reg/alerts/config", "预警配置"), ("/reg/alerts/events", "预警事件"), ("/reg/alerts/stream", "预警实时流"), ("/reg/alerts/summary", "预警汇总"), ("/reg/alerts/notify", "预警通知"), ("/reg/credentials/canteens", "食堂资质监管"), ("/reg/credentials/exceptions", "资质异常监管"), ("/reg/credentials/suppliers", "供应商资质监管"), ("/reg/credentials/workers", "从业人员资质监管"), ("/reg/credentials/overview", "资质监管总览"), ("/reg/credentials/export", "资质监管数据"), ("/reg/food-waste/ranking", "餐厨废弃物学校排名"), ("/reg/food-waste/trend", "餐厨废弃物趋势"), ("/reg/inspections/config", "检查任务配置"), ("/reg/inspections/inspectors", "检查人员"), ("/reg/inspections/tasks", "检查任务"), ("/reg/ledgers/data-stats", "台账数据统计"), ("/reg/ledgers/submit-stats", "台账上报统计"), ("/reg/ledgers/detail", "台账详情"), ("/reg/ledgers/dine", "就餐台账"), ("/reg/ledgers/disinfection", "消毒台账"), ("/reg/ledgers/pesticide", "农残台账"), ("/reg/ledgers/sampling", "留样台账"), ("/reg/ledgers/search", "台账综合查询"), ("/reg/ledgers/stats", "台账统计"), ("/reg/ledgers/waste", "废弃物台账"), ("/reg/overview", "监管首页总览"), ("/reg/reports", "监管统计报告"), ("/reg/schools", "监管学校"), ("/reg/system/info", "监管系统信息"), ("/reg/system/permissions", "监管权限"), ("/reg/system/roles", "监管角色"), ("/reg/system/settings", "监管系统设置"), ("/reg/system/units", "监管单位"), ("/reg/system/users", "监管用户"), ("/screen/analytics", "大屏分析数据"), ("/screen/analytics/ai/events", "大屏人工智能事件"), ("/screen/analytics/ai/pie", "大屏人工智能事件分类统计"), ("/screen/analytics/cameras", "大屏摄像头"), ("/screen/analytics/devices/statuses", "大屏设备状态"), ("/screen/analytics/devices/types", "大屏设备类型"), ("/screen/analytics/devices", "大屏设备"), ("/screen/analytics/personnel-health", "大屏人员健康数据"), ("/bright-local/ai/events", "本地人工智能事件"), ("/bright-local/aibox-boxes", "本地智能分析盒"), ("/bright-local/channels", "本地视频通道"), ("/bright-local/playback", "本地视频回放"), ("/bright-local/strangers", "本地陌生人记录"), ("/bright-local/refresh", "本地视频数据"), ("/bright/reg", "监管视频"), ("/bright/reg/download", "监管视频下载地址"), ("/bright/reg/playback", "监管视频回放"), ("/bright/reg/snapshots", "监管视频抓拍"), ("/bright/school", "学校视频"), ("/bright/school/download", "学校视频下载地址"), ("/bright/school/playback", "学校视频回放"), ("/bright/school/snapshots", "学校视频抓拍"), ("/bright/school/trustivs-camera-options", "学校视频摄像头选项"), ("/bright/school/trustivs-cameras", "学校视频摄像头"), ("/bright/channels", "明厨亮灶通道"), ("/bright/start", "明厨亮灶视频流"), ("/audit/logs", "审计日志"), ("/files", "文件"), ("/static", "静态文件"), ("/health", "系统健康状态"), ("/health/db", "数据库健康状态"), ("/health/routes", "系统路由"), ("/home", "公共首页数据"), ("/home/devices", "首页设备数据"), ("/home/hygiene", "首页卫生数据"), ("/home/inbound", "首页入库数据"), ("/home/outbound", "首页出库数据"), ("/home/parent-feedback", "首页家长反馈"), ("/api/devices", "集成设备"), ("/api/employees", "集成人员"), ("/api/morning-checks", "集成晨检记录"), ("/api/integrations/morning-checks/mego", "晨检数据接入"), ("/iot", "物联网视频"), ("/", "系统根接口"), ] REPLACEMENTS = { "AI": "人工智能", "AIBox": "智能分析盒", "CSV": "表格文件", "PDF": "文档文件", "JSON": "结构化数据", "JWT": "登录令牌", "MQTT": "设备消息", "OSS": "对象存储", "WebSocket": "实时通道", "deviceId": "设备编号", "schoolId": "学校编号", "cameraId": "摄像头编号", "id": "编号", "ID": "编号", } TOKEN_TRANSLATIONS = { "ai": "人工智能", "aibox": "智能分析盒", "aisle": "通道", "all": "全部", "api": "开放接口", "alg": "算法", "analysis": "分析", "analysistypes": "算法类型", "back": "回放", "beep": "蜂鸣器", "bit": "码", "box": "盒", "beat": "心跳", "by": "", "camera": "摄像头", "channel": "通道", "channels": "通道", "check": "晨检", "company": "单位", "config": "配置", "cust": "客流", "data": "数据", "dep": "部门", "device": "设备", "download": "下载地址", "encode": "编码", "employee": "人员", "employees": "人员", "face": "人脸", "feature": "特征", "file": "文件", "food": "菜品", "frames": "帧率", "from": "来源", "hum": "人体", "image": "图片", "info": "信息", "infos": "信息", "heart": "设备", "id": "编号", "last": "最近", "led": "指示灯", "list": "列表", "live": "实时", "local": "本地", "log": "日志", "meal": "餐别", "menu": "菜单", "mode": "模式", "morning": "晨检", "operate": "操作", "open": "开放", "or": "", "param": "参数", "people": "人员", "pixel": "分辨率", "plat": "平台", "progress": "进度", "pwr": "电源", "record": "记录", "records": "记录", "robot": "设备", "score": "相似度", "sdk": "设备开发包", "serial": "序列号", "server": "服务器", "sheet": "任务", "snap": "抓拍", "sn": "序列号", "staff": "人员", "statistic": "统计", "status": "状态", "stranger": "陌生人", "stream": "视频流", "stream0": "主码流", "system": "系统", "template": "模板", "third": "第三方", "time": "时间", "token": "令牌", "url": "地址", "usb": "外接", "user": "用户", "value": "条件", "vehicle": "车辆", "video": "视频", "view": "查看", "number": "编号", } TOKEN_ACTIONS = { "add": "新增", "create": "新增", "delete": "删除", "disable": "停用", "discover": "发现", "download": "获取", "enable": "启用", "find": "查询", "get": "查询", "init": "初始化", "list": "查询", "login": "登录", "page": "分页查询", "push": "推送", "query": "查询", "reboot": "重启", "refresh": "刷新", "report": "上报", "save": "保存", "select": "查询", "send": "发送", "set": "设置", "show": "查询", "stop": "停止", "update": "更新", "upload": "上传", } def prefix_value(path: str, rules: list[tuple[str, str]], default: str) -> str: for prefix, value in sorted(rules, key=lambda item: len(item[0]), reverse=True): if prefix == "/" and path == "/": return value if path == prefix or path.startswith(prefix + "/"): return value return default def folder_for(path: str) -> str: return prefix_value(path, FOLDER_RULES, "公共与系统/其他接口") def resource_for(path: str) -> str: direct = prefix_value(path, RESOURCE_RULES, "") if direct: return direct folder = folder_for(path).split("/")[-1] return folder if folder not in {"其他接口", "根接口"} else "业务数据" def contains_latin(value: str) -> bool: return bool(re.search(r"[A-Za-z]", value)) def normalize_chinese_summary(value: str) -> str: result = value for source, target in sorted(REPLACEMENTS.items(), key=lambda item: len(item[0]), reverse=True): result = result.replace(source, target) result = re.sub(r"[（(][A-Za-z0-9_./+ -]+[）)]", "", result) result = re.sub(r"[A-Za-z_]+", "", result) result = re.sub(r"\s+", "", result) return result.strip(" -—_/.") def split_identifier(value: str) -> list[str]: value = re.sub(r"\.[A-Za-z0-9]+$", "", value) value = re.sub(r"([a-z0-9])([A-Z])", r"\1 \2", value) value = re.sub(r"([A-Z]+)([A-Z][a-z])", r"\1 \2", value) return [item.lower() for item in re.split(r"[-_ .]+", value) if item] def legacy_endpoint_title(path: str, resource: str) -> str | None: last = path.rstrip("/").split("/")[-1] parts = split_identifier(last) if not parts: return None action: str | None = None translated: list[str] = [] for part in parts: if part in TOKEN_ACTIONS: next_action = TOKEN_ACTIONS[part] if action is None: action = next_action elif next_action != action: action = action + "或" + next_action continue word = TOKEN_TRANSLATIONS.get(part) if word: translated.append(word) elif word == "": continue elif part.isdigit(): translated.append(part) else: return None if action and translated: return action + "".join(translated) if action: return action + resource return None def action_for(method: str, path: str) -> str: lower = path.lower() last = lower.rstrip("/").split("/")[-1] if lower == "/auth/login" or lower.endswith("/login"): return "登录" if lower.endswith("/logout"): return "退出" if lower.endswith("/change-password"): return "修改" if lower.endswith("/refresh") or lower.endswith("/employees/refresh"): return "刷新" if "export" in last or last.endswith(".csv"): return "导出" if "import" in lower or "upload" in last: return "导入" if last in {"stream", "online"}: return "查询" if last in {"detail", "get", "view", "statuses", "types", "summary", "overview", "history", "audit"} or (last == "status" and method == "GET"): return "查询" if last in {"exists"}: return "检查" if last in {"delete", "remove", "clear", "clear-mem"}: return "删除" if lower.endswith("/handle/batch"): return "批量处理" if last in {"handle", "measure", "review", "accept"}: return "处理" if last in {"submit", "push", "report", "ingest"}: return "提交" if last in {"notify", "notify-rectify", "broadcast"}: return "发送" if last in {"enable", "disable", "enabled", "status"}: return "设置" if lower.endswith("/enable/batch"): return "批量设置" if last in {"copy-week"}: return "复制" if last in {"stocktake", "stocktake-batch"}: return "盘点" if last in {"callback"}: return "接收" if last in {"start-plan", "start"}: return "启动" if last in {"invite"}: return "邀请" if method == "POST" and lower.startswith("/push/"): return "同步" if last in TOKEN_ACTIONS: return TOKEN_ACTIONS[last] if method == "GET": return "查询" if method == "POST": return "新增" if method in {"PUT", "PATCH"}: return "更新" if method == "DELETE": return "删除" return "调用" def summary_for(method: str, path: str, old_summary: str) -> str: special = { ("POST", "/auth/login"): "用户登录", ("POST", "/auth/logout"): "用户退出登录", ("POST", "/auth/refresh"): "刷新登录令牌", ("GET", "/auth/me"): "查询当前用户信息", ("POST", "/auth/change-password"): "修改当前用户密码", ("POST", "/deviceIot/getDeviceIdBySerialNumber"): "根据序列号查询设备编号", ("POST", "/gateway/openApi/videoSheet/saveOrUpdateOpenApi"): "保存或更新视频任务", ("POST", "/openApi/info/getAlgAnalysisDetail"): "查询算法分析详情", ("POST", "/openApi/info/getStream0Config"): "查询主码流配置", }.get((method, path)) if special: return special normalized = normalize_chinese_summary(old_summary) if old_summary and not contains_latin(old_summary) and normalized: return normalized resource = resource_for(path) legacy_title = legacy_endpoint_title(path, resource) if legacy_title: return legacy_title action = action_for(method, path) suffix = "" lower = path.lower() if lower.endswith("/stream"): suffix = "实时数据" elif "export" in lower or lower.endswith(".csv"): suffix = "文件" elif lower.endswith("/detail") or (method == "GET" and "{id}" in lower): suffix = "详情" elif method == "GET" and not any(key in lower for key in ("summary", "overview", "status", "types", "history", "config", "audit", "download", "playback", "snapshot", "snapshots")): suffix = "列表" title = f"{action}{resource}{suffix}" title = normalize_chinese_summary(title) if not title or contains_latin(title): raise ValueError(f"Unable to produce Chinese summary for {method} {path}: {title!r}") return title def endpoint_id(operation: dict[str, Any]) -> int | None: run_url = str(operation.get("x-run-in-apifox") or "") match = re.search(r"/api-(\d+)-run", run_url) return int(match.group(1)) if match else None def transform(source: dict[str, Any]) -> tuple[dict[str, Any], list[dict[str, Any]]]: result = copy.deepcopy(source) result.setdefault("info", {})["title"] = "食品安全接口" mapping: list[dict[str, Any]] = [] used_folders: set[str] = set() for path, path_item in result.get("paths", {}).items(): for method, operation in path_item.items(): if method not in METHODS or not isinstance(operation, dict): continue old_summary = str(operation.get("summary") or "") old_folder = str(operation.get("x-apifox-folder") or ((operation.get("tags") or [""])[0])) new_folder = folder_for(path) new_summary = summary_for(method.upper(), path, old_summary) operation["summary"] = new_summary operation["tags"] = [new_folder] operation["x-apifox-folder"] = new_folder used_folders.add(new_folder) mapping.append( { "endpointId": endpoint_id(operation), "method": method.upper(), "path": path, "oldSummary": old_summary, "newSummary": new_summary, "oldFolder": old_folder, "newFolder": new_folder, } ) result["tags"] = [{"name": name} for name in sorted(used_folders)] return result, mapping def desktop_authorization() -> str: spec = importlib.util.spec_from_file_location("shared_apifox", SHARED_SCRIPT) if spec is None or spec.loader is None: raise RuntimeError("Unable to load Apifox desktop token helper") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module.desktop_authorization() def request_json(url: str, body: dict[str, Any]) -> tuple[int, dict[str, Any]]: request = urllib.request.Request( url, data=json.dumps(body, ensure_ascii=False).encode("utf-8"), headers={ "Authorization": desktop_authorization(), "X-Apifox-Api-Version": "2024-03-28", "Accept": "application/json", "Content-Type": "application/json", "User-Agent": "Codex-Apifox-Catalog-Cleanup/1.0", }, method="POST", ) try: with urllib.request.urlopen(request, timeout=180) as response: raw = response.read() return response.status, json.loads(raw or b"{}") except urllib.error.HTTPError as error: raw = error.read().decode("utf-8", "replace") try: detail = json.loads(raw) except json.JSONDecodeError: detail = {"raw": raw} return error.code, detail def operation_rows(spec: dict[str, Any]) -> list[tuple[str, str, dict[str, Any]]]: return [ (method.upper(), path, operation) for path, path_item in (spec.get("paths") or {}).items() for method, operation in path_item.items() if method in METHODS and isinstance(operation, dict) ] def validate(spec: dict[str, Any], mapping: list[dict[str, Any]]) -> dict[str, Any]: rows = operation_rows(spec) latin_summaries = [f"{method} {path}" for method, path, operation in rows if contains_latin(str(operation.get("summary") or ""))] latin_folders = [ f"{method} {path}" for method, path, operation in rows if contains_latin(str(operation.get("x-apifox-folder") or "")) ] login_rows = [ (method, path, operation) for method, path, operation in rows if method == "POST" and path == "/auth/login" ] missing_ids = [f"{item['method']} {item['path']}" for item in mapping if not item.get("endpointId")] return { "operationCount": len(rows), "pathCount": len(spec.get("paths") or {}), "folderCount": len(spec.get("tags") or []), "latinSummaryCount": len(latin_summaries), "latinFolderCount": len(latin_folders), "loginCount": len(login_rows), "loginFolder": login_rows[0][2].get("x-apifox-folder") if len(login_rows) == 1 else None, "missingEndpointIdCount": len(missing_ids), "latinSummaries": latin_summaries, "latinFolders": latin_folders, "missingEndpointIds": missing_ids, } def write_json(path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8") def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--source", required=True, type=Path) parser.add_argument("--output-dir", required=True, type=Path) parser.add_argument("--apply", action="store_true") args = parser.parse_args() source = json.loads(args.source.read_text(encoding="utf-8")) transformed, mapping = transform(source) validation = validate(transformed, mapping) ts = time.strftime("%Y%m%d_%H%M%S") spec_path = args.output_dir / f"food-safety-catalog-cleanup_openapi_{ts}.json" mapping_path = args.output_dir / f"food-safety-catalog-cleanup_id-map_{ts}.json" validation_path = args.output_dir / f"food-safety-catalog-cleanup_validation_{ts}.json" write_json(spec_path, transformed) write_json(mapping_path, mapping) write_json(validation_path, validation) result: dict[str, Any] = { "source": str(args.source), "spec": str(spec_path), "mapping": str(mapping_path), "validation": validation, "applied": False, } if any((validation["latinSummaryCount"], validation["latinFolderCount"], validation["loginCount"] != 1, validation["loginFolder"] != "登录认证")): print(json.dumps(result, ensure_ascii=False, indent=2)) return 1 if args.apply: payload = { "input": json.dumps(transformed, ensure_ascii=False), "options": { "targetEndpointFolderId": 0, "targetSchemaFolderId": 0, "endpointOverwriteBehavior": "OVERWRITE_EXISTING", "schemaOverwriteBehavior": "KEEP_EXISTING", "updateFolderOfChangedEndpoint": True, "prependBasePath": False, }, } payload_path = args.output_dir / f"food-safety-catalog-cleanup_import-payload_{ts}.json" response_path = args.output_dir / f"food-safety-catalog-cleanup_import-response_{ts}.json" write_json(payload_path, payload) status, response = request_json( f"{API_BASE}/v1/projects/{PROJECT_ID}/import-openapi?locale=zh-CN", payload, ) write_json(response_path, response) result.update({"applied": status == 200, "importStatus": status, "importResponse": response, "payload": str(payload_path)}) if status != 200: print(json.dumps(result, ensure_ascii=False, indent=2)) return 1 print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())