#!/usr/bin/env python3 """Import the existing store/ai_api OpenAPI catalogs into Apifox project 8900430. The script keeps credentials outside the repository, writes backups to a private temporary directory, prefixes component names to avoid cross-project collisions, and assigns every endpoint to one of the two approved top-level folders. """ from __future__ import annotations import argparse import copy import json import os import tempfile import urllib.error import urllib.request from pathlib import Path API_BASE = "https://api.apifox.com" API_VERSION = "2024-03-28" TARGET_PROJECT_ID = "8900430" METHODS = {"get", "post", "put", "patch", "delete", "head", "options", "trace"} def load_token() -> str: env_file = Path.home() / ".codex/skills/apifox-api-sync/.local.env" if env_file.exists(): for raw in env_file.read_text(encoding="utf-8").splitlines(): if "=" not in raw or raw.lstrip().startswith("#"): continue key, value = raw.split("=", 1) os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'")) token = os.environ.get("APIFOX_ACCESS_TOKEN") if not token: raise RuntimeError("Missing APIFOX_ACCESS_TOKEN") return token def request_json(token: str, method: str, url: str, body: dict | None = None, timeout: int = 300): data = None if body is None else json.dumps(body, ensure_ascii=False).encode("utf-8") request = urllib.request.Request( url, data=data, method=method, headers={ "Authorization": f"Bearer {token}", "X-Apifox-Api-Version": API_VERSION, "Content-Type": "application/json", "Accept": "application/json", "User-Agent": "Codex-Apifox-Sync/1.0", }, ) try: with urllib.request.urlopen(request, timeout=timeout) as response: raw = response.read().decode("utf-8") return response.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[:2000]} return exc.code, parsed def export_openapi(token: str, project_id: str): return request_json( token, "POST", f"{API_BASE}/v1/projects/{project_id}/export-openapi?locale=zh-CN", { "scope": {"type": "ALL"}, "options": {"includeApifoxExtensionProperties": True, "addFoldersToTags": True}, "oasVersion": "3.0", "exportFormat": "JSON", }, ) def rewrite_refs(value, prefix: str, component_categories: set[str]): if isinstance(value, dict): result = {} for key, child in value.items(): if key == "$ref" and isinstance(child, str) and child.startswith("#/components/"): parts = child.split("/") if len(parts) >= 4 and parts[2] in component_categories: parts[3] = prefix + parts[3] child = "/".join(parts) elif key == "security" and isinstance(child, list): child = [ {prefix + name: scopes for name, scopes in item.items()} if isinstance(item, dict) else item for item in child ] result[key] = rewrite_refs(child, prefix, component_categories) return result if isinstance(value, list): return [rewrite_refs(item, prefix, component_categories) for item in value] return value def transform(source: dict, top_folder: str, prefix: str, title: str) -> dict: document = copy.deepcopy(source) document["info"] = dict(document.get("info") or {}, title=title) document["servers"] = [] components = document.get("components") or {} categories = set(components) document = rewrite_refs(document, prefix, categories) components = document.get("components") or {} for category in list(components): if isinstance(components[category], dict): components[category] = {prefix + name: value for name, value in components[category].items()} document["components"] = components tags = {} for _, path_item in (document.get("paths") or {}).items(): for method, operation in path_item.items(): if method.lower() not in METHODS or not isinstance(operation, dict): continue raw_folder = operation.get("x-apifox-folder") or ((operation.get("tags") or ["未分类"])[0]) raw_folder = str(raw_folder or "未分类").strip("/").replace("(未分类)", "未分类") if raw_folder == top_folder: raw_folder = "" elif raw_folder.startswith(top_folder + "/"): raw_folder = raw_folder[len(top_folder) + 1 :] folder = top_folder + (("/" + raw_folder) if raw_folder else "") operation["tags"] = [folder] operation["x-apifox-folder"] = folder if operation.get("operationId"): operation["operationId"] = prefix + str(operation["operationId"]) description = operation.get("description") or "" ownership = f"\n\n---\n接口归属：{top_folder}；双库验收以 Apifox 环境和数据库回查证据为准。" if ownership.strip() not in description: operation["description"] = description + ownership tags[folder] = {"name": folder} document["tags"] = list(tags.values()) return document def operation_count(document: dict) -> int: return sum( 1 for path_item in (document.get("paths") or {}).values() for method in path_item if method.lower() in METHODS ) def import_document(token: str, document: dict, overwrite_behavior: str): return request_json( token, "POST", f"{API_BASE}/v1/projects/{TARGET_PROJECT_ID}/import-openapi?locale=zh-CN", { "input": json.dumps(document, ensure_ascii=False), "options": { "targetEndpointFolderId": 0, "targetSchemaFolderId": 0, "endpointOverwriteBehavior": overwrite_behavior, "schemaOverwriteBehavior": "KEEP_EXISTING", "updateFolderOfChangedEndpoint": True, "prependBasePath": False, }, }, ) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("stage", choices=["store", "ai_api", "verify"]) args = parser.parse_args() token = load_token() evidence_dir = Path(tempfile.mkdtemp(prefix=f"apifox-{TARGET_PROJECT_ID}-{args.stage}-")) status, before = export_openapi(token, TARGET_PROJECT_ID) if status != 200: raise RuntimeError(f"Target backup failed: HTTP {status}") (evidence_dir / "target_before.json").write_text( json.dumps(before, ensure_ascii=False), encoding="utf-8" ) if args.stage == "verify": folders = {} for path_item in (before.get("paths") or {}).values(): for method, operation in path_item.items(): if method.lower() not in METHODS or not isinstance(operation, dict): continue folder = operation.get("x-apifox-folder") or ((operation.get("tags") or ["未分类"])[0]) folders[folder] = folders.get(folder, 0) + 1 print(json.dumps({ "httpStatus": status, "pathCount": len(before.get("paths") or {}), "operationCount": operation_count(before), "topFolders": folders, "evidenceDir": str(evidence_dir), }, ensure_ascii=False)) return 0 if args.stage == "store": source_project, top_folder, prefix, title = ( "5737553", "01-store", "store__", "01-store 智慧餐厅 API" ) overwrite = "OVERWRITE_EXISTING" else: source_project, top_folder, prefix, title = ( "6067185", "02-ai_api", "aiapi__", "02-ai_api AI运动营养师 API" ) overwrite = "KEEP_EXISTING" status, source = export_openapi(token, source_project) if status != 200: raise RuntimeError(f"Source export failed: HTTP {status}") document = transform(source, top_folder, prefix, title) (evidence_dir / f"{args.stage}_import.json").write_text( json.dumps(document, ensure_ascii=False), encoding="utf-8" ) status, response = import_document(token, document, overwrite) (evidence_dir / f"{args.stage}_import_response.json").write_text( json.dumps(response, ensure_ascii=False, indent=2), encoding="utf-8" ) print(json.dumps({ "stage": args.stage, "sourceOperationCount": operation_count(document), "httpStatus": status, "counters": (response.get("data") or {}).get("counters", {}), "evidenceDir": str(evidence_dir), }, ensure_ascii=False)) return 0 if status == 200 else 1 if __name__ == "__main__": raise SystemExit(main())