#!/usr/bin/env python3 """Conservative Apifox OpenAPI import with backup and verification.""" from __future__ import annotations import argparse import json import os import sys import time import urllib.error import urllib.request from pathlib import Path API_BASE = "https://api.apifox.com" API_VERSION = "2024-03-28" DEFAULT_CONFIG = Path(__file__).resolve().parents[1] / "references/apifox_ai_sports_project.json" def load_env_file(path: Path) -> None: if not path.exists(): return for raw_line in path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip().strip('"').strip("'") if key and key not in os.environ: os.environ[key] = value def request_json(method: str, url: str, token: str, body: dict | None = None) -> tuple[int, dict]: data = None headers = { "X-Apifox-Api-Version": API_VERSION, "Authorization": f"Bearer {token}", "Content-Type": "application/json", } if body is not None: data = json.dumps(body, ensure_ascii=False).encode("utf-8") req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=60) as resp: raw = resp.read().decode("utf-8") return resp.status, json.loads(raw) if raw else {} except urllib.error.HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") try: parsed = json.loads(raw) except json.JSONDecodeError: parsed = {"raw": raw} return exc.code, parsed def export_openapi(project_id: str, token: str) -> tuple[int, dict]: body = { "scope": {"type": "ALL"}, "options": {"includeApifoxExtensionProperties": True, "addFoldersToTags": True}, "oasVersion": "3.0", "exportFormat": "JSON", } return request_json("POST", f"{API_BASE}/v1/projects/{project_id}/export-openapi?locale=zh-CN", token, body) def import_openapi(args: argparse.Namespace, token: str, spec: dict) -> tuple[int, dict, dict]: payload = { "input": json.dumps(spec, ensure_ascii=False), "options": { "targetEndpointFolderId": args.endpoint_folder_id, "targetSchemaFolderId": args.schema_folder_id, "endpointOverwriteBehavior": args.endpoint_overwrite_behavior, "schemaOverwriteBehavior": args.schema_overwrite_behavior, "updateFolderOfChangedEndpoint": args.update_folder, "prependBasePath": args.prepend_base_path, }, } status, resp = request_json( "POST", f"{API_BASE}/v1/projects/{args.project_id}/import-openapi?locale=zh-CN", token, payload, ) return status, resp, payload def write_json(path: Path, data: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") def load_config(path: Path) -> dict: return json.loads(path.read_text(encoding="utf-8")) def folder_choices(config: dict) -> list[dict]: return config.get("endpointFolders", []) def print_folder_choices(config: dict) -> None: print(f"Apifox project: {config.get('projectName')} (projectId={config.get('projectId')})") for index, folder in enumerate(folder_choices(config), 1): print(f"{index}. {folder['name']} ({folder['id']}) key={folder['key']}") def resolve_folder(config: dict, folder_key: str | None, folder_id: int | None) -> dict | None: folders = folder_choices(config) if folder_id: return next((item for item in folders if int(item["id"]) == int(folder_id)), None) or { "key": str(folder_id), "name": str(folder_id), "id": int(folder_id), } if folder_key: return next((item for item in folders if item["key"] == folder_key or item["name"] == folder_key), None) return None def verify_paths(exported: dict, verify_paths: list[str]) -> list[str]: paths = exported.get("paths") or {} missing = [] for item in verify_paths: if ":" in item: method, path = item.split(":", 1) methods = paths.get(path) or {} if method.lower() not in methods: missing.append(item) elif item not in paths: missing.append(item) return missing def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--openapi", help="Path to incremental OpenAPI JSON") parser.add_argument("--config", default=str(DEFAULT_CONFIG)) parser.add_argument("--list-folders", action="store_true") parser.add_argument("--project-id", default=None) parser.add_argument("--folder-key", default=None, help="Configured folder key or folder name") parser.add_argument("--endpoint-folder-id", type=int, default=None) parser.add_argument("--schema-folder-id", type=int, default=0) parser.add_argument("--token-env", default="APIFOX_ACCESS_TOKEN") parser.add_argument("--endpoint-overwrite-behavior", default="OVERWRITE_EXISTING") parser.add_argument("--schema-overwrite-behavior", default="KEEP_EXISTING") parser.add_argument( "--token-env-file", default=str(Path.home() / ".codex/skills/apifox-api-sync/.local.env"), help="Optional personal env file. Never commit this file.", ) parser.add_argument("--update-folder", action=argparse.BooleanOptionalAction, default=True) parser.add_argument("--prepend-base-path", action=argparse.BooleanOptionalAction, default=False) parser.add_argument("--verify-path", action="append", default=[], help="Path or METHOD:/path to verify after import") parser.add_argument("--output-dir", default=None) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--yes", action="store_true", help="Skip interactive confirmation") args = parser.parse_args() config = load_config(Path(args.config).expanduser()) if args.list_folders: print_folder_choices(config) return 0 if not args.openapi: print("--openapi is required unless --list-folders is used", file=sys.stderr) return 2 args.project_id = args.project_id or str(config.get("projectId", "6067185")) target_folder = resolve_folder(config, args.folder_key, args.endpoint_folder_id) if not target_folder: print("Missing target folder. Choose one with --folder-key or --endpoint-folder-id:\n", file=sys.stderr) print_folder_choices(config) return 2 args.endpoint_folder_id = int(target_folder["id"]) load_env_file(Path(args.token_env_file).expanduser()) token = os.environ.get(args.token_env) if not token: print(f"Missing token env var: {args.token_env}", file=sys.stderr) return 2 openapi_path = Path(args.openapi) spec = json.loads(openapi_path.read_text(encoding="utf-8")) ts = time.strftime("%Y%m%d_%H%M%S") out_dir = Path(args.output_dir) if args.output_dir else openapi_path.parent stem = openapi_path.stem operations = [f"{method.upper()} {path}" for path, methods in sorted((spec.get("paths") or {}).items()) for method in sorted(methods)] print(f"准备写入 Apifox 项目：{config.get('projectName')}（projectId={args.project_id}）") print(f"目标目录：{target_folder['name']}（{target_folder['id']}）") print(f"本次计划写入：{len(operations)} 个接口") for item in operations: print(f"- {item}") if not args.yes: answer = input(f"确认写入目录“{target_folder['name']}”？请输入目录名称或 yes：").strip() if answer not in {"yes", "YES", "y", "Y", target_folder["name"], str(target_folder["id"]), target_folder["key"]}: print("Canceled.") return 130 status, before = export_openapi(args.project_id, token) write_json(out_dir / f"{stem}_export_before_{ts}.json", before) if status != 200: print(json.dumps({"stage": "export_before", "status": status, "response": before}, ensure_ascii=False, indent=2)) return 1 if args.dry_run: print(json.dumps({"dryRun": True, "paths": len(spec.get("paths", {}))}, ensure_ascii=False, indent=2)) return 0 status, imported, payload = import_openapi(args, token, spec) write_json(out_dir / f"{stem}_import_payload_{ts}.json", payload) write_json(out_dir / f"{stem}_import_response_{ts}.json", imported) if status != 200: print(json.dumps({"stage": "import", "status": status, "response": imported}, ensure_ascii=False, indent=2)) return 1 status, after = export_openapi(args.project_id, token) write_json(out_dir / f"{stem}_export_after_{ts}.json", after) missing = verify_paths(after, args.verify_path) result = { "importStatus": status, "counters": imported.get("data", {}).get("counters", {}), "verifyMissing": missing, "exportAfterPathCount": len(after.get("paths", {})), } print(json.dumps(result, ensure_ascii=False, indent=2)) return 1 if missing else 0 if __name__ == "__main__": raise SystemExit(main())