services/image_api_adapter.py

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"
    # Complete request examples without any auth header/key are strong evidence
    # for a public endpoint. This is intentionally only a hint for the adapter.
    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}