| from __future__ import annotations |
|
|
| import html |
| import json |
| import re |
| from html.parser import HTMLParser |
| from typing import Any |
| from urllib.parse import urlparse |
|
|
|
|
| ADAPTER_MODEL = "gpt-5-5" |
| MAX_ADAPTER_DOCUMENT_CHARS = 600_000 |
| ALLOWED_PLACEHOLDERS = { |
| "prompt", |
| "model", |
| "size", |
| "n", |
| "quality", |
| "response_format", |
| "api_key", |
| "image", |
| "image_base64", |
| "image_data_url", |
| "images", |
| "images_base64", |
| "images_data_urls", |
| "task_id", |
| } |
| ALLOWED_CONTENT_TYPES = {"auto", "json", "multipart"} |
| ALLOWED_AUTH_TYPES = {"bearer", "x-api-key", "body", "none"} |
| PLACEHOLDER_RE = re.compile(r"\{\{\s*([a-zA-Z0-9_]+)\s*\}\}") |
| SECRET_NAME_RE = re.compile(r"(?:\bkey\b|api[_\s-]?key|access[_\s-]?token|secret|authorization)", re.I) |
| RESPONSE_PATH_RE = re.compile(r"^[A-Za-z0-9_\-]+(?:\.[A-Za-z0-9_\-]+)*$") |
| NO_AUTH_RE = re.compile( |
| r"(?:无需|不需要|不使用|免|无|不)\s*(?:API\s*)?(?:鉴权|认证|密钥|key)|" |
| r"(?:no|without)\s+(?:api\s+key|auth(?:entication|orization)?)\s+(?:is\s+)?required|" |
| r"public\s+(?:api|endpoint)", |
| re.I, |
| ) |
| AUTH_EVIDENCE_RE = re.compile( |
| r"Authorization\s*:|Bearer\s+|x-api-key|api[_\s-]?key|access[_\s-]?token|" |
| r"(?:请求头|header).{0,30}(?:鉴权|认证|密钥|key)", |
| re.I, |
| ) |
|
|
|
|
| class _ReadableHTMLParser(HTMLParser): |
| def __init__(self) -> None: |
| super().__init__(convert_charrefs=True) |
| self.parts: list[str] = [] |
| self.hidden_depth = 0 |
|
|
| def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: |
| if tag.lower() in {"script", "style", "svg", "noscript"}: |
| self.hidden_depth += 1 |
| elif tag.lower() in {"p", "div", "br", "li", "tr", "h1", "h2", "h3", "h4", "pre", "code"}: |
| self.parts.append("\n") |
|
|
| def handle_endtag(self, tag: str) -> None: |
| if tag.lower() in {"script", "style", "svg", "noscript"} and self.hidden_depth: |
| self.hidden_depth -= 1 |
| elif tag.lower() in {"p", "div", "li", "tr", "h1", "h2", "h3", "h4", "pre", "code"}: |
| self.parts.append("\n") |
|
|
| def handle_data(self, data: str) -> None: |
| if not self.hidden_depth and data.strip(): |
| self.parts.append(data) |
|
|
|
|
| def document_to_text(document: object, filename: object = "") -> str: |
| raw = str(document or "") |
| if not raw.strip(): |
| raise ValueError("导入的 API 文档不能为空") |
| if len(raw) > MAX_ADAPTER_DOCUMENT_CHARS: |
| raise ValueError("API 文档过大，请控制在 600 KB 以内") |
| lowered_name = str(filename or "").lower() |
| looks_like_html = lowered_name.endswith((".html", ".htm")) or bool(re.search(r"<\s*(?:html|body|main|article|div)\b", raw, re.I)) |
| if looks_like_html: |
| parser = _ReadableHTMLParser() |
| parser.feed(raw) |
| text = " ".join(parser.parts) |
| else: |
| text = raw |
| text = html.unescape(text).replace("\x00", " ") |
| text = re.sub(r"[ \t]+", " ", text) |
| text = re.sub(r"\n\s*\n\s*\n+", "\n\n", text) |
| return text.strip() |
|
|
|
|
| def redact_document_secrets(value: object) -> str: |
| text = str(value or "") |
| text = re.sub(r"(?i)\bBearer\s+[A-Za-z0-9._~+/=-]{8,}", "Bearer [REDACTED]", text) |
| text = re.sub(r"\bsk-[A-Za-z0-9_-]{8,}\b", "[REDACTED_API_KEY]", text) |
| text = re.sub( |
| r"(?i)((?:api[_\s-]?key|access[_\s-]?token|secret|authorization|\"key\"|'key')\s*[:=]\s*[\"']?)([^\s\"'&,}<>]{8,})", |
| r"\1[REDACTED]", |
| text, |
| ) |
| return text |
|
|
|
|
| def infer_document_auth_type(document_text: object) -> str: |
| """Return a conservative authentication hint derived from API examples.""" |
| text = str(document_text or "") |
| if NO_AUTH_RE.search(text): |
| return "none" |
| if AUTH_EVIDENCE_RE.search(text): |
| return "required" |
| |
| |
| if re.search(r"(?:curl\s+(?:-[^\n]+\s+)*https?://|fetch\s*\(|POST\s*/|GET\s*/)", text, re.I): |
| return "none" |
| return "unknown" |
|
|
|
|
| def adapter_system_prompt() -> str: |
| placeholders = ", ".join(f"{{{{{name}}}}}" for name in sorted(ALLOWED_PLACEHOLDERS)) |
| return ( |
| "You analyze third-party image generation/edit API documentation and return one safe declarative JSON adapter. " |
| "Return JSON only, without Markdown. Never copy credentials, example keys, tokens, cookies, or secrets from the document. " |
| "Do not return code, JavaScript, Python, shell, expressions, callbacks, or executable logic. " |
| "Schema: {\"summary\": string, \"models\": [string], \"template\": {" |
| "\"version\": 2, \"method\": \"POST\", " |
| "\"auth_type\": \"bearer\"|\"x-api-key\"|\"body\"|\"none\", \"auth_name\": string, " |
| "\"generate\": operation, \"edit\": operation|null, \"models\": models_operation|null}}. Each operation is {" |
| "\"endpoint\": string, \"content_type\": \"auto\"|\"json\"|\"multipart\", " |
| "\"body\": object, \"image_field\": string, " |
| "\"poll\": null|{\"endpoint\": string, \"id_path\": string, " |
| "\"interval_seconds\": integer, \"max_attempts\": integer}, " |
| "\"response\": {\"image_paths\": [string], \"error_paths\": [string], " |
| "\"success_path\": string, \"success_values\": [string|number|boolean], " |
| "\"failure_values\": [string|number|boolean]}}. " |
| "models_operation is {\"endpoint\": string, \"response_path\": string, \"id_path\": string}. " |
| "When the documentation exposes a model-list endpoint, describe its GET endpoint and the paths used to reach the model collection " |
| "and each model id; otherwise set models to null. response_path may point to an object whose values contain model arrays. " |
| "Use dot-separated response paths. image_paths must point to an image URL/base64 value, object, or list. " |
| "Always describe text-to-image in generate. When the documentation supports image editing, also describe image-to-image in edit, " |
| "including its own endpoint, request format, body, image field, and response paths; otherwise set edit to null. " |
| "Prefer the documented synchronous generation mode so the configured image_paths are available in the same response; " |
| "include every required sync flag such as task_type=sync when the documentation shows one. " |
| "When an operation is asynchronous, configure poll: id_path points to the task id in the initial response, " |
| "endpoint is the documented GET status URL and must contain {{task_id}}, and response describes the final polled payload. " |
| "Set poll to null for synchronous operations. " |
| "For JSON image editing, use {{image_data_url}} for a documented single image string or " |
| "{{images_data_urls}} for a documented image array. " |
| f"The only allowed dynamic placeholders are: {placeholders}. " |
| "For custom adapters, {{size}} is the user's original size or aspect ratio such as 16:9; do not convert it to pixel dimensions. " |
| "Use {{api_key}} only through auth settings or a body field documented for authentication. " |
| "If examples work without an Authorization header, API key header, cookie, or key body field, use auth_type=none. " |
| "Never invent authentication merely because many APIs normally require it. " |
| "If the documentation omits a property, use an empty string/list/object instead of inventing it." |
| ) |
|
|
|
|
| def adapter_user_prompt( |
| document_text: str, |
| *, |
| filename: object = "", |
| base_url: object = "", |
| auth_hint: object = "", |
| ) -> str: |
| return ( |
| f"Filename: {str(filename or '').strip() or 'imported-document'}\n" |
| f"Configured API base URL: {str(base_url or '').strip() or '(not provided)'}\n\n" |
| f"Authentication evidence: {str(auth_hint or '').strip() or 'unknown'}\n" |
| "Treat 'none' as authoritative: emit auth_type=none and do not add an api_key placeholder.\n\n" |
| "Analyze this redacted API documentation and create the adapter JSON:\n" |
| f"<api_document>\n{document_text}\n</api_document>" |
| ) |
|
|
|
|
| def parse_model_json(value: object) -> dict[str, Any]: |
| text = str(value or "").strip() |
| if text.startswith("```"): |
| text = re.sub(r"^```(?:json)?\s*", "", text, flags=re.I) |
| text = re.sub(r"\s*```$", "", text) |
| try: |
| parsed = json.loads(text) |
| except json.JSONDecodeError: |
| start = text.find("{") |
| end = text.rfind("}") |
| if start < 0 or end <= start: |
| raise ValueError("AI 未返回有效的 JSON 调用模板") |
| try: |
| parsed = json.loads(text[start:end + 1]) |
| except json.JSONDecodeError as exc: |
| raise ValueError("AI 返回的调用模板不是有效 JSON") from exc |
| if not isinstance(parsed, dict): |
| raise ValueError("AI 返回的调用模板必须是 JSON 对象") |
| return parsed |
|
|
|
|
| def _validate_template_value(value: object, *, field_name: str = "body") -> Any: |
| if value is None or isinstance(value, (bool, int, float)): |
| return value |
| if isinstance(value, list): |
| if len(value) > 50: |
| raise ValueError("调用模板中的数组过长") |
| return [_validate_template_value(item, field_name=field_name) for item in value] |
| if isinstance(value, dict): |
| if len(value) > 80: |
| raise ValueError("调用模板中的字段过多") |
| result: dict[str, Any] = {} |
| for key, item in value.items(): |
| clean_key = str(key or "").strip() |
| if not clean_key or len(clean_key) > 100: |
| raise ValueError("调用模板包含无效字段名") |
| normalized_item = _validate_template_value(item, field_name=clean_key) |
| if SECRET_NAME_RE.search(clean_key) and not ( |
| isinstance(normalized_item, str) and normalized_item in {"{{api_key}}", ""} |
| ): |
| raise ValueError(f"调用模板字段 {clean_key} 不允许写入静态密钥") |
| result[clean_key] = normalized_item |
| return result |
| text = str(value) |
| placeholders = PLACEHOLDER_RE.findall(text) |
| unknown = [item for item in placeholders if item not in ALLOWED_PLACEHOLDERS] |
| if unknown: |
| raise ValueError(f"调用模板包含不支持的占位符：{unknown[0]}") |
| remainder = PLACEHOLDER_RE.sub("", text) |
| if "{{" in remainder or "}}" in remainder: |
| raise ValueError("调用模板包含无效占位符") |
| if len(text) > 4000: |
| raise ValueError("调用模板字段内容过长") |
| return text |
|
|
|
|
| def _normalize_adapter_endpoint(value: object, *, base_url: object = "") -> str: |
| endpoint = str(value or "").strip() |
| if not endpoint or len(endpoint) > 2000: |
| raise ValueError("调用模板缺少有效 endpoint") |
| parsed_endpoint = urlparse(endpoint) |
| if parsed_endpoint.scheme or parsed_endpoint.netloc: |
| if parsed_endpoint.scheme not in {"http", "https"} or not parsed_endpoint.netloc: |
| raise ValueError("模板 endpoint 必须是 http/https 地址") |
| configured = urlparse(str(base_url or "").strip()) |
| if configured.netloc and parsed_endpoint.netloc.lower() != configured.netloc.lower(): |
| raise ValueError("模板 endpoint 必须与 API URL 使用相同域名") |
| elif not endpoint.startswith("/"): |
| endpoint = f"/{endpoint}" |
| return endpoint |
|
|
|
|
| def _normalize_response_config(value: object) -> dict[str, Any]: |
| raw_response = value if isinstance(value, dict) else {} |
| image_paths = [str(item).strip() for item in raw_response.get("image_paths", []) if str(item).strip()] if isinstance(raw_response.get("image_paths"), list) else [] |
| if not image_paths: |
| raise ValueError("调用模板至少需要一个图片返回路径") |
| error_paths = [str(item).strip() for item in raw_response.get("error_paths", []) if str(item).strip()] if isinstance(raw_response.get("error_paths"), list) else [] |
| if any(not RESPONSE_PATH_RE.fullmatch(path) for path in image_paths + error_paths): |
| raise ValueError("调用模板包含无效返回路径") |
| success_path = str(raw_response.get("success_path") or "").strip() |
| if success_path and not RESPONSE_PATH_RE.fullmatch(success_path): |
| raise ValueError("调用模板 success_path 无效") |
| success_values = raw_response.get("success_values") if isinstance(raw_response.get("success_values"), list) else [] |
| failure_values = raw_response.get("failure_values") if isinstance(raw_response.get("failure_values"), list) else [] |
| return { |
| "image_paths": image_paths[:20], |
| "error_paths": error_paths[:20], |
| "success_path": success_path, |
| "success_values": [item for item in success_values if item is None or isinstance(item, (str, int, float, bool))][:20], |
| "failure_values": [item for item in failure_values if item is None or isinstance(item, (str, int, float, bool))][:20], |
| } |
|
|
|
|
| def _normalize_poll_config(value: object, *, base_url: object = "") -> dict[str, Any] | None: |
| if value is None: |
| return None |
| if not isinstance(value, dict): |
| raise ValueError("调用模板 poll 必须是对象或 null") |
| endpoint = _normalize_adapter_endpoint(value.get("endpoint"), base_url=base_url) |
| if "{{task_id}}" not in endpoint: |
| raise ValueError("异步轮询 endpoint 必须包含 {{task_id}}") |
| id_path = str(value.get("id_path") or "").strip() |
| if not id_path or not RESPONSE_PATH_RE.fullmatch(id_path): |
| raise ValueError("异步轮询 id_path 无效") |
| try: |
| interval_seconds = int(value.get("interval_seconds") or 3) |
| max_attempts = int(value.get("max_attempts") or 100) |
| except (TypeError, ValueError) as exc: |
| raise ValueError("异步轮询次数或间隔无效") from exc |
| return { |
| "endpoint": endpoint, |
| "id_path": id_path, |
| "interval_seconds": max(1, min(30, interval_seconds)), |
| "max_attempts": max(1, min(180, max_attempts)), |
| } |
|
|
|
|
| def _normalize_adapter_operation(value: object, *, base_url: object = "") -> dict[str, Any]: |
| raw = value if isinstance(value, dict) else {} |
| content_type = str(raw.get("content_type") or "json").strip().lower() |
| if content_type not in ALLOWED_CONTENT_TYPES: |
| raise ValueError("调用模板 content_type 仅支持 auto/json/multipart") |
| image_field = str(raw.get("image_field") or "").strip() |
| if image_field and not re.fullmatch(r"[A-Za-z0-9_.\[\]-]{1,100}", image_field): |
| raise ValueError("调用模板 image_field 无效") |
| return { |
| "endpoint": _normalize_adapter_endpoint(raw.get("endpoint"), base_url=base_url), |
| "content_type": content_type, |
| "body": _validate_template_value(raw.get("body") if isinstance(raw.get("body"), dict) else {}), |
| "image_field": image_field, |
| "poll": _normalize_poll_config(raw.get("poll"), base_url=base_url), |
| "response": _normalize_response_config(raw.get("response")), |
| } |
|
|
|
|
| def _normalize_models_operation(value: object, *, base_url: object = "") -> dict[str, str] | None: |
| if value is None: |
| return None |
| if not isinstance(value, dict): |
| raise ValueError("调用模板 models 必须是对象或 null") |
| response_path = str(value.get("response_path") or "").strip() |
| id_path = str(value.get("id_path") or "id").strip() |
| if response_path and not RESPONSE_PATH_RE.fullmatch(response_path): |
| raise ValueError("调用模板 models.response_path 无效") |
| if not id_path or not RESPONSE_PATH_RE.fullmatch(id_path): |
| raise ValueError("调用模板 models.id_path 无效") |
| return { |
| "endpoint": _normalize_adapter_endpoint(value.get("endpoint"), base_url=base_url), |
| "response_path": response_path, |
| "id_path": id_path, |
| } |
|
|
|
|
| def normalize_adapter_template(value: object, *, base_url: object = "") -> dict[str, Any]: |
| if not isinstance(value, dict): |
| raise ValueError("调用模板不能为空") |
| method = str(value.get("method") or "POST").strip().upper() |
| if method != "POST": |
| raise ValueError("当前仅允许 POST 调用方式") |
| auth_type = str(value.get("auth_type") or "bearer").strip().lower() |
| if auth_type not in ALLOWED_AUTH_TYPES: |
| raise ValueError("调用模板 auth_type 不受支持") |
| auth_name = str(value.get("auth_name") or "").strip() |
| if auth_type == "x-api-key": |
| auth_name = auth_name or "x-api-key" |
| if not re.fullmatch(r"[A-Za-z0-9_-]{1,80}", auth_name): |
| raise ValueError("调用模板 auth_name 无效") |
| elif auth_type == "body": |
| auth_name = auth_name or "key" |
| if not re.fullmatch(r"[A-Za-z0-9_.\[\]-]{1,80}", auth_name): |
| raise ValueError("调用模板 auth_name 无效") |
| else: |
| auth_name = "" |
|
|
| if not isinstance(value.get("generate"), dict): |
| raise ValueError("调用模板缺少文生图 generate 配置") |
| if "edit" not in value: |
| raise ValueError("调用模板缺少图生图 edit 配置") |
| edit_source = value.get("edit") |
| if edit_source is not None and not isinstance(edit_source, dict): |
| raise ValueError("调用模板 edit 必须是对象或 null") |
| generate = _normalize_adapter_operation(value["generate"], base_url=base_url) |
| edit = _normalize_adapter_operation(edit_source, base_url=base_url) if edit_source is not None else None |
| models = _normalize_models_operation(value.get("models"), base_url=base_url) |
| return { |
| "version": 2, |
| "method": "POST", |
| "auth_type": auth_type, |
| "auth_name": auth_name, |
| "generate": generate, |
| "edit": edit, |
| "models": models, |
| } |
|
|
|
|
| def normalize_adapter_result( |
| value: object, |
| *, |
| base_url: object = "", |
| auth_hint: object = "", |
| ) -> dict[str, Any]: |
| parsed = value if isinstance(value, dict) else {} |
| template_value = parsed.get("template") if isinstance(parsed.get("template"), dict) else parsed |
| template = normalize_adapter_template(template_value, base_url=base_url) |
| if str(auth_hint or "").strip().lower() == "none": |
| template["auth_type"] = "none" |
| template["auth_name"] = "" |
| for operation_name in ("generate", "edit"): |
| operation = template.get(operation_name) |
| if not isinstance(operation, dict) or not isinstance(operation.get("body"), dict): |
| continue |
| operation["body"] = { |
| key: item |
| for key, item in operation["body"].items() |
| if not ( |
| SECRET_NAME_RE.search(str(key)) |
| and isinstance(item, str) |
| and item.strip() in {"", "{{api_key}}"} |
| ) |
| } |
| models = [] |
| if isinstance(parsed.get("models"), list): |
| models = list(dict.fromkeys(str(item).strip() for item in parsed["models"] if str(item).strip()))[:100] |
| summary = str(parsed.get("summary") or "已根据导入文档生成自定义调用模板").strip()[:500] |
| return {"summary": summary, "models": models, "template": template} |
|
|
