| import base64 |
| import json |
| import mimetypes |
| import os |
| import random |
| import re |
| import threading |
| import time |
| import zipfile |
| from dataclasses import dataclass |
| from io import BytesIO |
| from pathlib import Path |
| from typing import Any, Callable, Dict, Iterator, Optional |
| from urllib.parse import quote, unquote, urlparse |
|
|
| from curl_cffi import requests |
| from PIL import Image |
| import xlrd |
|
|
| from services.account_service import ( |
| account_service, |
| build_chatgpt_client_fingerprint, |
| extract_chatgpt_account_id, |
| sanitize_registered_proxy, |
| ) |
| from services.config import config |
| from services.proxy_service import proxy_settings |
| from utils.helper import UpstreamHTTPError, ensure_ok, iter_sse_payloads, new_uuid, split_image_model |
| from utils.log import logger |
| from utils.pow import build_legacy_requirements_token, build_proof_token, parse_pow_resources |
| from services.sentinel_solver_client import solve_so, solve_turnstile, solver_user_agent |
|
|
|
|
| class InvalidAccessTokenError(RuntimeError): |
| pass |
|
|
|
|
| class UpstreamBlockedError(RuntimeError): |
| """上游 CDN（Cloudflare）拦截了本次请求 —— 与「token 无效」是两回事。 |
| |
| 典型场景：容器出口 IP 被 CF 拦，``/backend-api/accounts/check`` 返回 403 |
| （body 常为 HTML 拦截页）。本地直连同一个 token 却正常返回 200。此时**不能** |
| 把账号判为失效，否则导入/刷新会被整体误杀。 |
| """ |
|
|
| def __init__(self, message: str, status_code: int = 403, body: str = "") -> None: |
| super().__init__(message) |
| self.status_code = status_code |
| self.body = body |
|
|
|
|
| class ImagePollTimeoutError(RuntimeError): |
| def __init__( |
| self, |
| message: str = "", |
| conversation_id: str = "", |
| upstream_text: str = "", |
| ) -> None: |
| super().__init__(message) |
| self.conversation_id = conversation_id |
| |
| |
| |
| self.upstream_text = upstream_text |
|
|
|
|
| class ImageStreamHardTimeoutError(RuntimeError): |
| pass |
|
|
|
|
| class ImageTextOnlyAnswer(RuntimeError): |
| """上游这一轮已经用「纯文字」收尾了：不会再出图，文字就是它的最终回答。 |
| |
| 和超时的区别是**结论**：超时是「不知道还会不会出图」，这里是「确定不会出图了」。 |
| 所以调用方应当直接把 ``text`` 当结果返回（前台原样展示上游原话），而不是报超时。 |
| """ |
|
|
| def __init__(self, text: str = "", conversation_id: str = "") -> None: |
| super().__init__(text or "upstream answered with text only") |
| self.text = text |
| self.conversation_id = conversation_id |
|
|
|
|
| class UpstreamFileDeleteError(RuntimeError): |
| def __init__(self, message: str, partial_result: Dict[str, Any]) -> None: |
| self.partial_result = partial_result |
| super().__init__(message) |
|
|
|
|
| def image_poll_timeout_message(timeout_secs: float) -> str: |
| timeout_label = f"{timeout_secs:g}" |
| return ( |
| f"生图超时 (已等待 {timeout_label} 秒）。" |
| |
| |
| |
| ) |
|
|
|
|
| @dataclass |
| class ImageResolutionResult: |
| urls: list[str] |
| message_id: str = "" |
|
|
|
|
| @dataclass |
| class ChatRequirements: |
| """保存一次对话请求所需的 sentinel token。""" |
| token: str |
| proof_token: str = "" |
| turnstile_token: str = "" |
| so_token: str = "" |
| raw_finalize: Optional[Dict[str, Any]] = None |
|
|
|
|
| |
| |
| |
| DEFAULT_CLIENT_VERSION = "prod-a194cd50d4416d3c0b47c740f206b12ce60f5887" |
| DEFAULT_CLIENT_BUILD_NUMBER = "6708908" |
| |
| |
| DEFAULT_IMPERSONATE = "chrome146" |
| LEGACY_IMPERSONATES = {"chrome110"} |
| UPSTREAM_BROWSER_IMPERSONATE = DEFAULT_IMPERSONATE |
| UPSTREAM_SEC_CH_UA_ARCH = '"x86"' |
| UPSTREAM_SEC_CH_UA_BITNESS = '"64"' |
| UPSTREAM_SEC_CH_UA_FULL_VERSION = '"143.0.3650.96"' |
| UPSTREAM_SEC_CH_UA_FULL_VERSION_LIST = ( |
| '"Microsoft Edge";v="143.0.3650.96", "Chromium";v="143.0.7499.147", ' |
| '"Not A(Brand";v="24.0.0.0"' |
| ) |
| UPSTREAM_SEC_CH_UA_PLATFORM_VERSION = '"19.0.0"' |
| DEFAULT_POW_SCRIPT = "https://chatgpt.com/backend-api/sentinel/sdk.js" |
| CODEX_IMAGE_MODEL = "codex-gpt-image-2" |
| SEARCH_MODEL = "gpt-5-6-thinking" |
|
|
| SEARCH_TIMEOUT_SECS = 300.0 |
|
|
| SEARCH_POLL_INTERVAL_SECS = 3.0 |
|
|
| SEARCH_DONE_STATUS = {"finished_successfully", "finished_partial_completion"} |
|
|
| SEARCH_CONVERSATION_ID_RE = re.compile(r'"conversation_id"\s*:\s*"([^"]+)"') |
|
|
| SEARCH_URL_RE = re.compile(r"https?://[^\s\"'<>）)\]}]+") |
|
|
| EDITABLE_FILE_MODEL = "gpt-5-5-thinking" |
|
|
| EDITABLE_FILE_THINKING_EFFORT = "extended" |
|
|
| EDITABLE_FILE_TIMEOUT_SECS = 1200.0 |
|
|
| EDITABLE_FILE_POLL_INTERVAL_SECS = 5.0 |
|
|
| EDITABLE_FILE_CLIENT_VERSION = "prod-bede35f9dcd856d080e012478f0c1031faa2588e" |
|
|
| EDITABLE_FILE_CLIENT_BUILD_NUMBER = "6631702" |
|
|
| EDITABLE_FILE_PSD_OUTPUT_DIR = "data/files/psd" |
|
|
| EDITABLE_FILE_PPT_OUTPUT_DIR = "data/files/ppt" |
|
|
| EDITABLE_FILE_PPT_PROMPT = """我需要你根据用户的需求，来制作一个可以编辑的PPT，你可以使用Agent来做，你不要再继续询问用户问题，内容风格、版式、配色、内容结构和页面信息你可以自行补充并直接执行。整体的流程如下： |
| 1. 用生图的方式，帮我生成一个精美的产品介绍ppt，5-6个页面 |
| 2. 帮我把以上涉及到的所有图像和形状素材拆分成单独png，每个素材单独一张图片，不要有遗漏，让我可以直接在ppt里拼接素材还原，不要文字 |
| 3. 利用以上所有图片和形状素材，帮我还原你第一次生成的展示ppt，我需要是可编辑的ppt格式，主要部分需要你单独还原插入，文字需要可以编辑 |
| 最后只需要给我生成一个PPT文件，以及生成中遇到的各种素材压缩包zip文件就行。""" |
|
|
| EDITABLE_FILE_PSD_PROMPT = "帮我生成这个图像，把这张海报分成若干图像，包括背景图，每个元素不要改位置，这样子我可以直接在 平时里无需拖动，底色为白色，不要伪透明底。再帮我将以上拆分的图像拼合成一个psd文件，去除白色底，不要改变每个图层的相应位置，保留每个元素所在图层的相应位置，保留每个元素的图层，最后只需要给我输出psd文件，以及每个图层的zip文件" |
|
|
| EDITABLE_ASSET_POINTER_RE = re.compile(r"(?:file-service|sediment)://([A-Za-z0-9_-]+)") |
|
|
| EDITABLE_ZIP_MIME_TYPES = {"application/zip", "application/x-zip-compressed"} |
|
|
| EDITABLE_PSD_MIME_TYPES = {"image/vnd.adobe.photoshop", "application/vnd.adobe.photoshop"} |
|
|
| EDITABLE_PPT_MIME_TYPES = { |
| "application/vnd.openxmlformats-officedocument.presentationml.presentation", |
| "application/vnd.ms-powerpoint", |
| } |
|
|
| EDITABLE_PSD_EXPORT_FILE_RE = re.compile(r"(?:sandbox:)?(/mnt/data/[^\s\"'\)\]]+\.(?:psd|zip))", re.IGNORECASE) |
|
|
| EDITABLE_PPT_EXPORT_FILE_RE = re.compile(r"(?:sandbox:)?(/mnt/data/[^\s\"'\)\]]+\.(?:pptx?|zip))", re.IGNORECASE) |
|
|
| GENERIC_SANDBOX_EXPORT_FILE_RE = re.compile(r"(?:sandbox:)?(/mnt/data/[^\s<>\]\)\"']+)", re.IGNORECASE) |
|
|
| FILE_ID_RE = re.compile(r"\b(file[-_](?!service\b)[A-Za-z0-9_-]+)\b") |
|
|
| @dataclass |
| class EditableFileArtifact: |
| attachment_id: str = "" |
| file_id: str = "" |
| name: str = "" |
| mime_type: str = "" |
| create_time: float = 0.0 |
| author_role: str = "" |
| sandbox_path: str = "" |
| message_id: str = "" |
|
|
| @dataclass |
| class EditableFileExportResult: |
| conversation_id: str |
| primary_path: Path |
| zip_path: Path |
|
|
| FILE_SERVICE_ID_RE = re.compile(r"file-service://([A-Za-z0-9_-]+)") |
| |
| |
| REAL_IMAGE_FILE_ID_RE = re.compile(r"\bfile_00000000[a-f0-9]{24}\b") |
| SEDIMENT_ID_RE = re.compile(r"sediment://([A-Za-z0-9_-]+)") |
|
|
|
|
| class OpenAIBackendAPI: |
| """ChatGPT Web 后端封装。 |
| |
| 说明： |
| - 传入 `access_token` 时，聊天和模型列表都会走已登录链路 |
| 例如 `/backend-api/sentinel/chat-requirements`、`/backend-api/conversation` |
| - 不传 `access_token` 时，会走未登录链路 |
| 例如 `/backend-anon/sentinel/chat-requirements`、`/backend-anon/conversation` |
| - `stream_conversation()` 是底层统一流式入口 |
| - 协议兼容转换放在 `services.protocol` |
| """ |
|
|
| def __init__( |
| self, |
| access_token: str = "", |
| *, |
| account_context: Optional[Dict[str, Any]] = None, |
| fp: Optional[Dict[str, Any]] = None, |
| account_id: str = "", |
| ) -> None: |
| """初始化后端客户端。 |
| |
| 参数： |
| - `access_token`：可选。传入后表示使用已登录链路；不传则使用未登录链路。 |
| """ |
| self.base_url = "https://chatgpt.com" |
| self.client_version = DEFAULT_CLIENT_VERSION |
| self.client_build_number = DEFAULT_CLIENT_BUILD_NUMBER |
| self.access_token = access_token |
| stored_account = account_service.get_account(access_token) if access_token else None |
| self.account_context: Dict[str, Any] = dict(stored_account or {}) |
| if isinstance(account_context, dict): |
| self.account_context.update(account_context) |
| if isinstance(fp, dict): |
| self.account_context["fp"] = dict(fp) |
| if account_id: |
| self.account_context["account_id"] = str(account_id).strip() |
| self.account_key = str(self.account_context.get("account_key") or "").strip() |
| self.account_id = ( |
| extract_chatgpt_account_id(self.account_context, self.access_token) |
| or str(account_id or "").strip() |
| ) |
| self.fp = self._build_fp() |
| self.user_agent = self.fp["user-agent"] |
| self.device_id = self.fp["oai-device-id"] |
| self.session_id = self.fp["oai-session-id"] |
| self.registered_proxy = sanitize_registered_proxy(self.account_context.get("registered_proxy")) |
| self.pow_script_sources: list[str] = [] |
| self.pow_data_build = "" |
| self.progress_callback: Callable[[str], None] | None = None |
| session_kwargs: Dict[str, Any] = { |
| "impersonate": UPSTREAM_BROWSER_IMPERSONATE, |
| "verify": True, |
| } |
| if self.registered_proxy: |
| session_kwargs["proxy"] = self.registered_proxy |
| else: |
| session_kwargs = proxy_settings.build_session_kwargs(**session_kwargs) |
| self.session = requests.Session(**session_kwargs) |
| self.session.headers.update({ |
| "User-Agent": self.user_agent, |
| "Origin": self.base_url, |
| "Referer": self.base_url + "/", |
| "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8,en-US;q=0.7", |
| "Cache-Control": "no-cache", |
| "Pragma": "no-cache", |
| "Priority": "u=1, i", |
| "Sec-Ch-Ua": self.fp["sec-ch-ua"], |
| "Sec-Ch-Ua-Arch": UPSTREAM_SEC_CH_UA_ARCH, |
| "Sec-Ch-Ua-Bitness": UPSTREAM_SEC_CH_UA_BITNESS, |
| "Sec-Ch-Ua-Full-Version": UPSTREAM_SEC_CH_UA_FULL_VERSION, |
| "Sec-Ch-Ua-Full-Version-List": UPSTREAM_SEC_CH_UA_FULL_VERSION_LIST, |
| "Sec-Ch-Ua-Mobile": self.fp["sec-ch-ua-mobile"], |
| "Sec-Ch-Ua-Model": '""', |
| "Sec-Ch-Ua-Platform": self.fp["sec-ch-ua-platform"], |
| "Sec-Ch-Ua-Platform-Version": UPSTREAM_SEC_CH_UA_PLATFORM_VERSION, |
| "Sec-Fetch-Dest": "empty", |
| "Sec-Fetch-Mode": "cors", |
| "Sec-Fetch-Site": "same-origin", |
| "OAI-Device-Id": self.device_id, |
| "OAI-Session-Id": self.session_id, |
| "OAI-Language": "zh-CN", |
| "OAI-Client-Version": self.client_version, |
| "OAI-Client-Build-Number": self.client_build_number, |
| }) |
| if self.access_token: |
| self.session.headers["Authorization"] = f"Bearer {self.access_token}" |
| if self.account_id: |
| self.session.headers["ChatGPT-Account-Id"] = self.account_id |
| try: |
| self.session.cookies.set("oai-did", self.device_id, domain=".chatgpt.com") |
| except Exception: |
| pass |
|
|
| def _build_fp(self) -> Dict[str, str]: |
| raw_fp = self.account_context.get("fp") |
| if not isinstance(raw_fp, dict): |
| raw_fp = self.account_context |
| |
| |
| if isinstance(raw_fp, dict) and raw_fp.get("impersonate") in LEGACY_IMPERSONATES: |
| raw_fp = {key: value for key, value in raw_fp.items() if key != "impersonate"} |
| return build_chatgpt_client_fingerprint( |
| self.access_token, |
| raw_fp, |
| account_key=self.account_key, |
| ) |
|
|
| def close(self) -> None: |
| self.session.close() |
|
|
| def __enter__(self) -> "OpenAIBackendAPI": |
| return self |
|
|
| def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None: |
| self.close() |
|
|
| def _headers(self, path: str, extra: Optional[Dict[str, str]] = None) -> Dict[str, str]: |
| """构造请求头，并补上 web 端要求的 target path/route。""" |
| headers = dict(self.session.headers) |
| headers["X-OpenAI-Target-Path"] = path |
| headers["X-OpenAI-Target-Route"] = path |
| if self.account_id: |
| headers["ChatGPT-Account-Id"] = self.account_id |
| if extra: |
| headers.update(extra) |
| return headers |
|
|
| @staticmethod |
| def _extract_quota_and_restore_at(limits_progress: list[Any]) -> tuple[int, str | None, bool]: |
| for item in limits_progress: |
| if isinstance(item, dict) and item.get("feature_name") == "image_gen": |
| return int(item.get("remaining") or 0), str(item.get("reset_after") or "") or None, False |
| return 0, None, True |
|
|
| def _get_me(self) -> Dict[str, Any]: |
| path = "/backend-api/me" |
| response = self.session.get(self.base_url + path, headers=self._headers(path), timeout=20) |
| if response.status_code != 200: |
| if response.status_code == 401: |
| raise InvalidAccessTokenError(f"{path} failed: HTTP {response.status_code}") |
| raise RuntimeError(f"{path} failed: HTTP {response.status_code}") |
| return response.json() |
|
|
| def _get_conversation_init(self) -> Dict[str, Any]: |
| path = "/backend-api/conversation/init" |
| response = self.session.post( |
| self.base_url + path, |
| headers=self._headers(path, {"Content-Type": "application/json"}), |
| json={ |
| "gizmo_id": None, |
| "requested_default_model": None, |
| "conversation_id": None, |
| "timezone_offset_min": -480, |
| }, |
| timeout=20, |
| ) |
| if response.status_code != 200: |
| if response.status_code == 401: |
| raise InvalidAccessTokenError(f"{path} failed: HTTP {response.status_code}") |
| raise RuntimeError(f"{path} failed: HTTP {response.status_code}") |
| return response.json() |
|
|
| @staticmethod |
| def _looks_like_cdn_block(response: Any) -> bool: |
| """判断响应是不是 Cloudflare/CDN 的拦截页，而不是上游应用层的 JSON 报错。 |
| |
| 拦截页特征：content-type 是 text/html，或 body 以 ``<!DOCTYPE``/``<html`` 开头， |
| 或带 ``cf-ray``/``server: cloudflare`` 且 body 不是 JSON。 |
| """ |
| try: |
| ctype = str(response.headers.get("content-type") or "").lower() |
| body = (getattr(response, "text", "") or "").lstrip() |
| except Exception: |
| return False |
| if "json" in ctype: |
| return False |
| if body[:1] == "{" or body[:1] == "[": |
| return False |
| if "html" in ctype: |
| return True |
| head = body[:200].lower() |
| return head.startswith("<!doctype") or head.startswith("<html") or "<html" in head |
|
|
| def _get_default_account(self) -> Dict[str, Any]: |
| route = "/backend-api/accounts/check/v4-2023-04-27" |
| attempts = 2 |
| last_status = 0 |
| last_body = "" |
| for attempt in range(attempts): |
| response = self.session.get(self.base_url + route + "?timezone_offset_min=-480", |
| headers=self._headers(route), timeout=20) |
| last_status = int(response.status_code) |
| if last_status == 200: |
| payload = response.json() |
| logger.debug({"event": "backend_user_info_account_payload", "account_payload": payload}) |
| return ((payload.get("accounts") or {}).get("default") or {}).get("account") or {} |
| if last_status == 401: |
| raise InvalidAccessTokenError(f"{route} failed: HTTP {last_status}") |
| last_body = (getattr(response, "text", "") or "")[:400] |
| |
| |
| if last_status in (403, 429, 500, 502, 503, 504) and self._looks_like_cdn_block(response): |
| logger.warning({ |
| "event": "accounts_check_cdn_blocked", |
| "status": last_status, |
| "body_head": last_body[:120], |
| "attempt": attempt + 1, |
| }) |
| if attempt + 1 < attempts: |
| time.sleep(1.0 + attempt) |
| continue |
| raise UpstreamBlockedError( |
| "/backend-api/accounts/check blocked by upstream CDN: HTTP %d" % last_status, |
| status_code=last_status, |
| body=last_body, |
| ) |
| raise RuntimeError(f"/backend-api/accounts/check failed: HTTP {last_status}") |
| raise RuntimeError(f"/backend-api/accounts/check failed: HTTP {last_status}") |
|
|
| def get_user_info(self) -> Dict[str, Any]: |
| """获取当前 token 的账号信息。""" |
| if not self.access_token: |
| raise RuntimeError("access_token is required") |
| logger.debug({"event": "backend_user_info_start"}) |
| default_account = self._get_default_account() |
| discovered_account_id = ( |
| extract_chatgpt_account_id(default_account) |
| or str(default_account.get("id") or "").strip() |
| or self.account_id |
| ) |
| if discovered_account_id and discovered_account_id != self.account_id: |
| self.account_id = discovered_account_id |
| self.session.headers["ChatGPT-Account-Id"] = discovered_account_id |
| if discovered_account_id: |
| stored_account = account_service.get_account(self.access_token) |
| if stored_account and str(stored_account.get("account_id") or "").strip() != discovered_account_id: |
| account_service.update_account(self.access_token, {"account_id": discovered_account_id}) |
| me_payload = self._get_me() |
| init_payload = self._get_conversation_init() |
|
|
| plan_type = str(default_account.get("plan_type") or "free") |
|
|
| limits_progress = init_payload.get("limits_progress") |
| limits_progress = limits_progress if isinstance(limits_progress, list) else [] |
| quota, restore_at, image_quota_unknown = self._extract_quota_and_restore_at(limits_progress) |
| |
| |
| |
| if image_quota_unknown and plan_type.strip().lower() == "pro": |
| quota = 1000 |
| image_quota_unknown = False |
| result = { |
| "email": me_payload.get("email"), |
| "user_id": me_payload.get("id"), |
| "account_id": discovered_account_id or None, |
| "type": plan_type, |
| "quota": quota, |
| "image_quota_unknown": image_quota_unknown, |
| "limits_progress": limits_progress, |
| "default_model_slug": init_payload.get("default_model_slug"), |
| "restore_at": restore_at, |
| "status": "正常" if image_quota_unknown and plan_type.lower() != "free" else ("限流" if quota == 0 else "正常"), |
| } |
| logger.debug({ |
| "event": "backend_user_info_result", |
| "email": result.get("email"), |
| "user_id": result.get("user_id"), |
| "type": result.get("type"), |
| "quota": result.get("quota"), |
| "image_quota_unknown": result.get("image_quota_unknown"), |
| "default_model_slug": result.get("default_model_slug"), |
| "restore_at": result.get("restore_at"), |
| "status": result.get("status"), |
| }) |
| return result |
|
|
| def _bootstrap_headers(self) -> Dict[str, str]: |
| """构造首页预热请求头。""" |
| return { |
| "User-Agent": self.user_agent, |
| "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8", |
| "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", |
| "Sec-Ch-Ua": self.session.headers["Sec-Ch-Ua"], |
| "Sec-Ch-Ua-Mobile": self.session.headers["Sec-Ch-Ua-Mobile"], |
| "Sec-Ch-Ua-Platform": self.session.headers["Sec-Ch-Ua-Platform"], |
| "Sec-Fetch-Dest": "document", |
| "Sec-Fetch-Mode": "navigate", |
| "Sec-Fetch-Site": "none", |
| "Sec-Fetch-User": "?1", |
| "Upgrade-Insecure-Requests": "1", |
| } |
|
|
| def _sentinel_user_agent(self) -> str: |
| """requirements / proof token 里编码的 UA：应与实际发请求的浏览器环境一致。 |
| |
| 接入 sentinel 求解服务时改用求解页 UA（与 Turnstile 环境匹配，否则仍会被判低可信）； |
| 未配置时回落到本会话 UA。 |
| """ |
| return solver_user_agent() or self.user_agent |
|
|
| def _build_requirements(self, data: Dict[str, Any], source_p: str = "") -> ChatRequirements: |
| """把 sentinel 响应整理成后续对话需要的 token 集合。""" |
| if (data.get("arkose") or {}).get("required"): |
| raise RuntimeError("chat requirements requires arkose token, which is not implemented") |
|
|
| proof_token = "" |
| proof_info = data.get("proofofwork") or {} |
| if proof_info.get("required"): |
| proof_token = build_proof_token( |
| proof_info.get("seed", ""), |
| proof_info.get("difficulty", ""), |
| self._sentinel_user_agent(), |
| script_sources=self.pow_script_sources, |
| data_build=self.pow_data_build, |
| ) |
|
|
| turnstile_token = "" |
| turnstile_info = data.get("turnstile") or {} |
| if turnstile_info.get("required") and turnstile_info.get("dx"): |
| turnstile_token = solve_turnstile(turnstile_info["dx"], source_p) |
|
|
| return ChatRequirements( |
| token=data.get("token", ""), |
| proof_token=proof_token, |
| turnstile_token=turnstile_token, |
| so_token=data.get("so_token", ""), |
| raw_finalize=data, |
| ) |
|
|
| def _conversation_headers(self, path: str, requirements: ChatRequirements) -> Dict[str, str]: |
| """根据当前 requirements 构造对话 SSE 请求头。""" |
| headers = { |
| "Accept": "text/event-stream", |
| "Content-Type": "application/json", |
| "OpenAI-Sentinel-Chat-Requirements-Token": requirements.token, |
| } |
| if requirements.proof_token: |
| headers["OpenAI-Sentinel-Proof-Token"] = requirements.proof_token |
| if requirements.turnstile_token: |
| headers["OpenAI-Sentinel-Turnstile-Token"] = requirements.turnstile_token |
| if requirements.so_token: |
| headers["OpenAI-Sentinel-SO-Token"] = requirements.so_token |
| return self._headers(path, headers) |
|
|
| def _api_messages_to_conversation_messages(self, messages: list[Dict[str, Any]]) -> list[Dict[str, Any]]: |
| """把标准 chat messages 转成 web conversation 所需的 messages。""" |
| conversation_messages = [] |
| for item in messages: |
| role = item.get("role", "user") |
| content = item.get("content", "") |
| if isinstance(content, str): |
| conversation_messages.append({ |
| "id": new_uuid(), |
| "author": {"role": role}, |
| "content": {"content_type": "text", "parts": [content]}, |
| }) |
| continue |
| if not isinstance(content, list): |
| raise RuntimeError("only string or list message content is supported") |
| text_parts: list[str] = [] |
| image_inputs: list[tuple[bytes, str, str]] = [] |
| file_inputs: list[tuple[bytes, str, str]] = [] |
| for part in content: |
| if not isinstance(part, dict): |
| continue |
| part_type = str(part.get("type") or "") |
| if part_type == "text": |
| text_parts.append(str(part.get("text") or "")) |
| elif part_type == "image": |
| data = part.get("data") |
| mime = str(part.get("mime") or "image/png") |
| if isinstance(data, (bytes, bytearray)): |
| image_inputs.append((bytes(data), mime, "")) |
| elif part_type == "file": |
| data = part.get("data") |
| mime = str(part.get("mime") or "application/octet-stream") |
| name = str(part.get("name") or "attachment") |
| if isinstance(data, (bytes, bytearray)): |
| file_data = bytes(data) |
| archive_files = self._expand_zip_attachment(file_data, name, mime) |
| if archive_files is not None: |
| names: list[str] = [] |
| for archive_data, archive_name, archive_mime in archive_files: |
| names.append(archive_name) |
| if archive_mime in {"image/png", "image/jpeg", "image/webp", "image/gif"}: |
| image_inputs.append((archive_data, archive_mime, archive_name)) |
| else: |
| inline_text = ( |
| self._inline_text_attachment(archive_data, archive_name, archive_mime) |
| or self._inline_spreadsheet_attachment(archive_data, archive_name, archive_mime) |
| ) |
| if inline_text: |
| text_parts.append(inline_text) |
| else: |
| file_inputs.append((archive_data, archive_name, archive_mime)) |
| text_parts.append( |
| f'\n\n<attached_archive name={json.dumps(Path(name).name, ensure_ascii=False)} ' |
| f'files={len(names)}>\n' + "\n".join(names) + "\n</attached_archive>" |
| ) |
| continue |
| inline_text = self._inline_text_attachment(file_data, name, mime) |
| if inline_text: |
| |
| |
| |
| |
| |
| text_parts.append(inline_text) |
| file_inputs.append((file_data, name, mime)) |
| continue |
| spreadsheet_text = self._inline_spreadsheet_attachment(file_data, name, mime) |
| if spreadsheet_text: |
| |
| |
| |
| text_parts.append(spreadsheet_text) |
| else: |
| file_inputs.append((file_data, name, mime)) |
| if not image_inputs and not file_inputs: |
| conversation_messages.append({ |
| "id": new_uuid(), |
| "author": {"role": role}, |
| "content": {"content_type": "text", "parts": ["".join(text_parts)]}, |
| }) |
| continue |
| if not self.access_token: |
| raise RuntimeError("authenticated upstream account required for file input") |
| uploaded: list[Dict[str, Any]] = [] |
| for idx, (data, mime, source_name) in enumerate(image_inputs, start=1): |
| ext_part = mime.split("/", 1)[1].split("+")[0] if "/" in mime else "png" |
| extension = "jpg" if ext_part == "jpeg" else (ext_part or "png") |
| b64 = base64.b64encode(data).decode("ascii") |
| uploaded.append(self._upload_image( |
| f"data:{mime};base64,{b64}", |
| Path(source_name).name if source_name else f"image_{idx}.{extension}", |
| )) |
| for data, name, mime in file_inputs: |
| uploaded.append(self._upload_file(data, name, mime)) |
| parts: list[Any] = [] |
| for ref in uploaded: |
| if not ref.get("is_image"): |
| continue |
| parts.append({ |
| "content_type": "image_asset_pointer", |
| "asset_pointer": f"file-service://{ref['file_id']}", |
| "width": ref["width"], |
| "height": ref["height"], |
| "size_bytes": ref["file_size"], |
| }) |
| text = "".join(text_parts) |
| if text: |
| parts.append(text) |
| has_uploaded_image = any(ref.get("is_image") for ref in uploaded) |
| conversation_messages.append({ |
| "id": new_uuid(), |
| "author": {"role": role}, |
| "content": { |
| "content_type": "multimodal_text" if has_uploaded_image else "text", |
| "parts": parts if has_uploaded_image else [text], |
| }, |
| "metadata": { |
| "attachments": [{ |
| "id": ref["file_id"], |
| "name": ref["file_name"], |
| "size": ref["file_size"], |
| **({ |
| "mimeType": ref["mime_type"], |
| "width": ref["width"], |
| "height": ref["height"], |
| } if ref.get("is_image") else { |
| "mimeType": ref["mime_type"], |
| "fileTokenSize": ref.get("file_token_size") or max(1, len(ref["file_name"])), |
| }), |
| } for ref in uploaded], |
| }, |
| }) |
| return conversation_messages |
|
|
| @staticmethod |
| def _inline_text_attachment(data: bytes, file_name: str, mime_type: str) -> str: |
| """Keep text/code attachments readable even if upstream attachment indexing is delayed.""" |
| suffix = Path(str(file_name or "")).suffix.lower() |
| normalized_mime = str(mime_type or "").split(";", 1)[0].strip().lower() |
| text_suffixes = { |
| ".txt", ".text", ".md", ".markdown", ".json", ".html", ".htm", ".xml", ".csv", ".tsv", |
| ".js", ".mjs", ".jsx", ".ts", ".tsx", ".py", ".java", ".c", ".cc", ".cpp", ".h", ".hpp", |
| ".cs", ".go", ".rs", ".rb", ".php", ".sh", ".sql", ".yaml", ".yml", ".toml", ".css", |
| ".scss", ".less", ".log", ".srt", ".vtt", ".svg", ".env", |
| } |
| text_mimes = { |
| "application/json", "application/xml", "application/javascript", "application/typescript", |
| "application/x-yaml", "application/yaml", "application/toml", "application/x-toml", "image/svg+xml", |
| } |
| known_text = normalized_mime.startswith("text/") or normalized_mime in text_mimes or suffix in text_suffixes |
| binary_suffixes = { |
| ".pdf", ".doc", ".docx", ".dot", ".rtf", ".odt", ".ppt", ".pptx", ".pps", |
| ".xls", ".xlsx", ".zip", ".gz", ".rar", ".7z", ".exe", ".dll", |
| } |
| if not known_text and suffix in binary_suffixes: |
| return "" |
| if not known_text and normalized_mime.startswith(("image/", "audio/", "video/")): |
| return "" |
| if b"\x00" in data[:4096] and not data.startswith((b"\xff\xfe", b"\xfe\xff")): |
| return "" |
| decoded = "" |
| for encoding in ("utf-8-sig", "utf-16", "gb18030"): |
| try: |
| decoded = data.decode(encoding) |
| break |
| except UnicodeDecodeError: |
| continue |
| if not decoded.strip(): |
| return "" |
| if not known_text: |
| sample = decoded[:4096] |
| printable = sum(character.isprintable() or character in "\r\n\t" for character in sample) |
| if not sample or printable / len(sample) < 0.9: |
| return "" |
| limit = 200_000 |
| truncated = len(decoded) > limit |
| body = decoded[:limit] |
| name = Path(str(file_name or "attachment")).name or "attachment" |
| suffix_notice = "\n[内容过长，已截断]" if truncated else "" |
| return f'\n\n<attached_file name={json.dumps(name, ensure_ascii=False)}>\n{body}{suffix_notice}\n</attached_file>' |
|
|
| @staticmethod |
| def _inline_spreadsheet_attachment(data: bytes, file_name: str, mime_type: str) -> str: |
| """Extract legacy XLS cell contents so the model can answer without delayed indexing. |
| |
| ChatGPT's web file endpoint currently rejects some legacy ``.xls`` |
| attachments at conversation time even after the blob upload succeeds. |
| Extracting the workbook locally keeps the user-visible upload useful and |
| gives the model deterministic sheet/cell context. Embedded pictures are |
| intentionally left to image/archive handling; this path covers cell data. |
| """ |
| suffix = Path(str(file_name or "")).suffix.lower() |
| normalized_mime = str(mime_type or "").split(";", 1)[0].strip().lower() |
| if suffix != ".xls" and normalized_mime != "application/vnd.ms-excel": |
| return "" |
| try: |
| workbook = xlrd.open_workbook(file_contents=data, on_demand=True) |
| except (xlrd.XLRDError, OSError, ValueError): |
| return "" |
|
|
| limit = 200_000 |
| chunks = [ |
| f'<attached_spreadsheet name={json.dumps(Path(file_name).name, ensure_ascii=False)} ' |
| f'sheets={len(workbook.sheet_names())}>' |
| ] |
| truncated = False |
| try: |
| for sheet_name in workbook.sheet_names(): |
| sheet = workbook.sheet_by_name(sheet_name) |
| chunks.append( |
| f'<sheet name={json.dumps(sheet_name, ensure_ascii=False)} ' |
| f'rows={sheet.nrows} columns={sheet.ncols}>' |
| ) |
| for row_index in range(sheet.nrows): |
| cells: list[str] = [] |
| for column_index in range(sheet.ncols): |
| value = sheet.cell_value(row_index, column_index) |
| if value in ("", None): |
| continue |
| if isinstance(value, float) and value.is_integer(): |
| value = int(value) |
| address = f"{xlrd.formula.colname(column_index)}{row_index + 1}" |
| cells.append(f"{address}={json.dumps(str(value), ensure_ascii=False)}") |
| if not cells: |
| continue |
| line = " | ".join(cells) |
| if sum(len(item) + 1 for item in chunks) + len(line) > limit: |
| truncated = True |
| break |
| chunks.append(line) |
| chunks.append("</sheet>") |
| if truncated: |
| break |
| finally: |
| workbook.release_resources() |
| if truncated: |
| chunks.append("[内容过长，已截断]") |
| chunks.append("</attached_spreadsheet>") |
| return "\n\n" + "\n".join(chunks) |
|
|
| @staticmethod |
| def _expand_zip_attachment(data: bytes, file_name: str, mime_type: str) -> Optional[list[tuple[bytes, str, str]]]: |
| """Safely expose ZIP members because upstream models do not reliably index archives.""" |
| suffix = Path(str(file_name or "")).suffix.lower() |
| normalized_mime = str(mime_type or "").split(";", 1)[0].strip().lower() |
| if suffix != ".zip" and normalized_mime not in {"application/zip", "application/x-zip-compressed"}: |
| return None |
| max_members = 20 |
| max_member_bytes = 25 * 1024 * 1024 |
| max_total_bytes = 50 * 1024 * 1024 |
| archive_stem = Path(str(file_name or "archive.zip")).stem or "archive" |
| extracted: list[tuple[bytes, str, str]] = [] |
| total_bytes = 0 |
| try: |
| with zipfile.ZipFile(BytesIO(data)) as archive: |
| members = [item for item in archive.infolist() if not item.is_dir() and not item.filename.startswith("__MACOSX/")] |
| if not members: |
| raise ValueError(f"ZIP 文件为空：{file_name}") |
| if len(members) > max_members: |
| raise ValueError(f"ZIP 文件包含 {len(members)} 个文件，最多支持 {max_members} 个") |
| for item in members: |
| if item.flag_bits & 0x1: |
| raise ValueError(f"ZIP 内含加密文件，无法读取：{item.filename}") |
| if item.file_size > max_member_bytes or total_bytes + item.file_size > max_total_bytes: |
| raise ValueError(f"ZIP 解压后内容超过限制：{item.filename}") |
| with archive.open(item, "r") as member_file: |
| member_data = member_file.read(max_member_bytes + 1) |
| if len(member_data) > max_member_bytes: |
| raise ValueError(f"ZIP 内单个文件超过 25MB：{item.filename}") |
| total_bytes += len(member_data) |
| base_name = item.filename.replace("\\", "/").rsplit("/", 1)[-1] or "file" |
| safe_name = f"{archive_stem}-{base_name}" |
| member_mime = mimetypes.guess_type(base_name)[0] or "application/octet-stream" |
| extracted.append((member_data, safe_name, member_mime)) |
| except zipfile.BadZipFile as exc: |
| raise ValueError(f"ZIP 文件损坏或格式无效：{file_name}") from exc |
| return extracted |
|
|
| @staticmethod |
| def _normalize_thinking_effort(value: str) -> str: |
| normalized = str(value or "").strip().lower() |
| if normalized in {"", "none", "auto"}: |
| return "" |
| if normalized in {"low", "medium", "high", "standard", "max"}: |
| return normalized |
| if normalized in {"xhigh", "extended"}: |
| return "extended" |
| return "" |
|
|
| def _conversation_payload( |
| self, |
| messages: list[Dict[str, Any]], |
| model: str, |
| timezone: str, |
| thinking_effort: str = "", |
| ) -> Dict[str, Any]: |
| """把标准 messages 构造成 web 对话请求体。""" |
| payload = { |
| "action": "next", |
| "messages": self._api_messages_to_conversation_messages(messages), |
| "model": model, |
| "parent_message_id": new_uuid(), |
| "conversation_mode": {"kind": "primary_assistant"}, |
| "conversation_origin": None, |
| "force_paragen": False, |
| "force_paragen_model_slug": "", |
| "force_rate_limit": False, |
| "force_use_sse": True, |
| "history_and_training_disabled": True, |
| "reset_rate_limits": False, |
| "suggestions": [], |
| "supported_encodings": [], |
| "system_hints": [], |
| "timezone": timezone, |
| "timezone_offset_min": -480, |
| "variant_purpose": "comparison_implicit", |
| "websocket_request_id": new_uuid(), |
| "client_contextual_info": { |
| "is_dark_mode": False, |
| "time_since_loaded": 120, |
| "page_height": 900, |
| "page_width": 1400, |
| "pixel_ratio": 2, |
| "screen_height": 1440, |
| "screen_width": 2560, |
| }, |
| } |
| normalized_effort = self._normalize_thinking_effort(thinking_effort or config.default_thinking_effort) |
| if normalized_effort: |
| payload["thinking_effort"] = normalized_effort |
| return payload |
|
|
| def _image_model_settings(self, model: str) -> tuple[str, str]: |
| """把标准图片模型名映射为上游模型及思考强度。""" |
| _, base_model = split_image_model(model) |
| if not base_model: |
| return "auto", "" |
| if base_model == "gpt-image-2": |
| upstream_model = config.default_upstream_model_name |
| elif base_model == CODEX_IMAGE_MODEL: |
| upstream_model = base_model |
| else: |
| return "auto", "" |
| model_name, separator, suffix = upstream_model.rpartition("-") |
| if separator and suffix.lower() in {"standard", "extended", "max"}: |
| return model_name, suffix.lower() |
| return upstream_model, self._normalize_thinking_effort(config.default_thinking_effort) |
|
|
| def _image_headers(self, path: str, requirements: ChatRequirements, conduit_token: str = "", accept: str = "*/*") -> \ |
| Dict[str, str]: |
| """构造图片链路请求头。""" |
| headers = { |
| "Content-Type": "application/json", |
| "Accept": accept, |
| "OpenAI-Sentinel-Chat-Requirements-Token": requirements.token, |
| } |
| if requirements.proof_token: |
| headers["OpenAI-Sentinel-Proof-Token"] = requirements.proof_token |
| |
| if requirements.turnstile_token: |
| headers["OpenAI-Sentinel-Turnstile-Token"] = requirements.turnstile_token |
| if requirements.so_token: |
| headers["OpenAI-Sentinel-SO-Token"] = requirements.so_token |
| if conduit_token: |
| headers["X-Conduit-Token"] = conduit_token |
| if accept == "text/event-stream": |
| headers["X-Oai-Turn-Trace-Id"] = new_uuid() |
| return self._headers(path, headers) |
|
|
| def _prepare_image_conversation( |
| self, |
| prompt: str, |
| requirements: ChatRequirements, |
| model: str, |
| conversation_id: str = "", |
| parent_message_id: str = "", |
| request_timeout: float = 60.0, |
| ) -> str: |
| """为图片生成准备 conduit token。""" |
| path = "/backend-api/f/conversation/prepare" |
| parent_message_id = parent_message_id or new_uuid() |
| upstream_model, thinking_effort = self._image_model_settings(model) |
| payload = { |
| "action": "next", |
| "fork_from_shared_post": False, |
| "parent_message_id": parent_message_id, |
| "model": upstream_model, |
| "client_prepare_state": "success", |
| "timezone_offset_min": -480, |
| "timezone": "Asia/Shanghai", |
| "conversation_mode": {"kind": "primary_assistant"}, |
| "system_hints": ["picture_v2"], |
| "partial_query": { |
| "id": new_uuid(), |
| "author": {"role": "user"}, |
| "content": {"content_type": "text", "parts": [prompt]}, |
| }, |
| "supports_buffering": True, |
| "supported_encodings": ["v1"], |
| "client_contextual_info": {"app_name": "chatgpt.com"}, |
| } |
| if thinking_effort: |
| payload["thinking_effort"] = thinking_effort |
| if conversation_id: |
| payload["conversation_id"] = conversation_id |
| response = self.session.post( |
| self.base_url + path, |
| headers=self._image_headers(path, requirements), |
| json=payload, |
| timeout=max(1.0, request_timeout), |
| ) |
| ensure_ok(response, path) |
| return response.json().get("conduit_token", "") |
|
|
| def _decode_image_base64(self, image: str) -> bytes: |
| """把 base64 图片字符串或本地路径解码成二进制。""" |
| if ( |
| image |
| and len(image) < 512 |
| and not image.startswith("data:") |
| and "\n" not in image |
| and "\r" not in image |
| ): |
| file_path = Path(os.path.expanduser(image)) |
| if file_path.exists() and file_path.is_file(): |
| return file_path.read_bytes() |
| payload = image.split(",", 1)[1] if image.startswith("data:") and "," in image else image |
| return base64.b64decode(payload) |
|
|
| def _upload_image(self, image: str, file_name: str = "image.png") -> Dict[str, Any]: |
| """上传一张 base64 图片，返回底层文件元数据。""" |
| data = self._decode_image_base64(image) |
| if ( |
| image |
| and len(image) < 512 |
| and not image.startswith("data:") |
| and "\n" not in image |
| and "\r" not in image |
| ): |
| candidate_path = Path(os.path.expanduser(image)) |
| if candidate_path.exists() and candidate_path.is_file(): |
| file_name = candidate_path.name |
| image = Image.open(BytesIO(data)) |
| width, height = image.size |
| mime_type = Image.MIME.get(image.format, "image/png") |
| path = "/backend-api/files" |
| response = self.session.post( |
| self.base_url + path, |
| headers=self._headers(path, {"Content-Type": "application/json", "Accept": "application/json"}), |
| json={"file_name": file_name, "file_size": len(data), "use_case": "multimodal", "width": width, |
| "height": height}, |
| timeout=60, |
| ) |
| ensure_ok(response, path) |
| upload_meta = response.json() |
| time.sleep(0.5) |
| response = self.session.put( |
| upload_meta["upload_url"], |
| headers={ |
| "Content-Type": mime_type, |
| "x-ms-blob-type": "BlockBlob", |
| "x-ms-version": "2020-04-08", |
| "Origin": self.base_url, |
| "Referer": self.base_url + "/", |
| "User-Agent": self.user_agent, |
| "Accept": "application/json, text/plain, */*", |
| "Accept-Language": "en-US,en;q=0.8", |
| }, |
| data=data, |
| timeout=120, |
| ) |
| ensure_ok(response, "image_upload") |
| path = f"/backend-api/files/{upload_meta['file_id']}/uploaded" |
| response = self.session.post( |
| self.base_url + path, |
| headers=self._headers(path, {"Content-Type": "application/json", "Accept": "application/json"}), |
| data="{}", |
| timeout=60, |
| ) |
| ensure_ok(response, path) |
| return { |
| "file_id": upload_meta["file_id"], |
| "file_name": file_name, |
| "file_size": len(data), |
| "mime_type": mime_type, |
| "width": width, |
| "height": height, |
| "is_image": True, |
| } |
|
|
| def _upload_file(self, data: bytes, file_name: str, mime_type: str = "application/octet-stream") -> Dict[str, Any]: |
| """Upload a non-image attachment through ChatGPT's multimodal file flow.""" |
| if not data: |
| raise ValueError("file data is empty") |
| safe_name = Path(str(file_name or "attachment")).name or "attachment" |
| normalized_mime = str(mime_type or "").strip().lower() or mimetypes.guess_type(safe_name)[0] or "application/octet-stream" |
| document_mimes = { |
| "application/msword", |
| "application/pdf", |
| "application/rtf", |
| "application/vnd.ms-excel", |
| "application/vnd.ms-powerpoint", |
| "application/vnd.oasis.opendocument.presentation", |
| "application/vnd.oasis.opendocument.spreadsheet", |
| "application/vnd.oasis.opendocument.text", |
| "application/vnd.openxmlformats-officedocument.presentationml.presentation", |
| "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", |
| "application/vnd.openxmlformats-officedocument.wordprocessingml.document", |
| "text/csv", |
| "text/markdown", |
| "text/plain", |
| } |
| use_case = "my_files" if normalized_mime in document_mimes else "ace_upload" |
| path = "/backend-api/files" |
| response = self.session.post( |
| self.base_url + path, |
| headers=self._headers(path, {"Content-Type": "application/json", "Accept": "application/json"}), |
| json={ |
| "file_name": safe_name, |
| "file_size": len(data), |
| "use_case": use_case, |
| "timezone_offset_min": -480, |
| "reset_rate_limits": False, |
| }, |
| timeout=60, |
| ) |
| ensure_ok(response, path) |
| upload_meta = response.json() |
| upload_url = str(upload_meta.get("upload_url") or "") |
| file_id = str(upload_meta.get("file_id") or "") |
| if not upload_url or not file_id: |
| raise RuntimeError(f"invalid upload response: {upload_meta}") |
| response = self.session.put( |
| upload_url, |
| headers={ |
| "Content-Type": normalized_mime, |
| "x-ms-blob-type": "BlockBlob", |
| "x-ms-version": "2020-04-08", |
| "Origin": self.base_url, |
| "Referer": self.base_url + "/", |
| "User-Agent": self.user_agent, |
| "Accept": "application/json, text/plain, */*", |
| "Accept-Language": "en-US,en;q=0.8", |
| }, |
| data=data, |
| timeout=120, |
| ) |
| ensure_ok(response, "file_upload") |
| path = f"/backend-api/files/{file_id}/uploaded" |
| uploaded_status = "" |
| uploaded_payload: Dict[str, Any] = {} |
| for attempt in range(5): |
| response = self.session.post( |
| self.base_url + path, |
| headers=self._headers(path, {"Content-Type": "application/json", "Accept": "application/json"}), |
| data="{}", |
| timeout=60, |
| ) |
| ensure_ok(response, path) |
| uploaded_payload = response.json() |
| uploaded_status = str(uploaded_payload.get("status") or "").strip().lower() |
| if uploaded_status in {"success", "uploaded", "processed"}: |
| break |
| if uploaded_status in {"error", "failed"}: |
| raise RuntimeError(f"file processing failed: {uploaded_payload}") |
| if attempt < 4: |
| time.sleep(1) |
| else: |
| raise RuntimeError(f"file processing timed out: status={uploaded_status or 'unknown'}") |
| return { |
| "file_id": file_id, |
| "file_name": safe_name, |
| "file_size": len(data), |
| "mime_type": normalized_mime, |
| "file_token_size": ( |
| uploaded_payload.get("fileTokenSize") |
| or uploaded_payload.get("file_token_size") |
| or max(1, len(safe_name)) |
| ), |
| "is_image": False, |
| } |
|
|
| def _start_image_generation( |
| self, |
| prompt: str, |
| requirements: ChatRequirements, |
| conduit_token: str, |
| model: str, |
| references: Optional[list[Dict[str, Any]]] = None, |
| conversation_id: str = "", |
| parent_message_id: str = "", |
| request_timeout: float = 300.0, |
| ) -> requests.Response: |
| """启动图片生成或编辑的 SSE 请求。""" |
| upstream_model, thinking_effort = self._image_model_settings(model) |
| references = references or [] |
| parent_message_id = parent_message_id or new_uuid() |
| parts = [{ |
| "content_type": "image_asset_pointer", |
| "asset_pointer": f"file-service://{item['file_id']}", |
| "width": item["width"], |
| "height": item["height"], |
| "size_bytes": item["file_size"], |
| } for item in references] |
| parts.append(prompt) |
| content = {"content_type": "multimodal_text", "parts": parts} if references else {"content_type": "text", |
| "parts": [prompt]} |
| metadata = { |
| "developer_mode_connector_ids": [], |
| "selected_github_repos": [], |
| "selected_all_github_repos": False, |
| "system_hints": ["picture_v2"], |
| "serialization_metadata": {"custom_symbol_offsets": []}, |
| } |
| if references: |
| metadata["attachments"] = [{ |
| "id": item["file_id"], |
| "mimeType": item["mime_type"], |
| "name": item["file_name"], |
| "size": item["file_size"], |
| "width": item["width"], |
| "height": item["height"], |
| } for item in references] |
| payload = { |
| "action": "next", |
| "messages": [{ |
| "id": new_uuid(), |
| "author": {"role": "user"}, |
| "create_time": time.time(), |
| "content": content, |
| "metadata": metadata, |
| }], |
| "parent_message_id": parent_message_id, |
| "model": upstream_model, |
| "client_prepare_state": "sent", |
| "timezone_offset_min": -480, |
| "timezone": "Asia/Shanghai", |
| "conversation_mode": {"kind": "primary_assistant"}, |
| "enable_message_followups": True, |
| "system_hints": ["picture_v2"], |
| "supports_buffering": True, |
| "supported_encodings": ["v1"], |
| "client_contextual_info": { |
| "is_dark_mode": False, |
| "time_since_loaded": 1200, |
| "page_height": 1072, |
| "page_width": 1724, |
| "pixel_ratio": 1.2, |
| "screen_height": 1440, |
| "screen_width": 2560, |
| "app_name": "chatgpt.com", |
| }, |
| "paragen_cot_summary_display_override": "allow", |
| "force_parallel_switch": "auto", |
| } |
| if thinking_effort: |
| payload["thinking_effort"] = thinking_effort |
| if conversation_id: |
| payload["conversation_id"] = conversation_id |
| path = "/backend-api/f/conversation" |
| response = self.session.post( |
| self.base_url + path, |
| headers=self._image_headers(path, requirements, conduit_token, "text/event-stream"), |
| json=payload, |
| timeout=max(1.0, request_timeout), |
| stream=True, |
| ) |
| ensure_ok(response, path) |
| return response |
|
|
| def _get_conversation(self, conversation_id: str) -> Dict[str, Any]: |
| """获取完整 conversation 详情。""" |
| path = f"/backend-api/conversation/{conversation_id}" |
| response = self.session.get(self.base_url + path, headers=self._headers(path, {"Accept": "application/json"}), |
| timeout=60) |
| ensure_ok(response, path) |
| return response.json() |
|
|
| def hide_conversation(self, conversation_id: str) -> bool: |
| """Hide a ChatGPT official conversation from the account history.""" |
| conversation_id = str(conversation_id or "").strip() |
| if not conversation_id: |
| raise RuntimeError("conversation_id is required") |
| path = f"/backend-api/conversation/{conversation_id}" |
| response = self.session.patch( |
| self.base_url + path, |
| headers=self._headers(path, {"Accept": "application/json", "Content-Type": "application/json"}), |
| json={"is_visible": False}, |
| timeout=30, |
| ) |
| ensure_ok(response, path) |
| logger.info({"event": "conversation_hidden", "conversation_id": conversation_id}) |
| return True |
|
|
| def delete_conversation(self, conversation_id: str) -> Dict[str, Any]: |
| """兼容上游命名：实际通过 is_visible=false 隐藏会话。""" |
| self.hide_conversation(conversation_id) |
| return {"success": True} |
|
|
| def create_share_link(self, conversation_id: str) -> Dict[str, Any]: |
| """为一条上游会话创建 ChatGPT 官方匿名分享链接。 |
| |
| ⚠️ ``is_public`` 必须在创建时就传 ``True``：实测只传 ``is_anonymous`` 时 |
| 创建虽然成功，但匿名访客打开链接会拿到 404 ``no_access``，浏览器直接跳回首页。 |
| 另外 ``current_node_id`` 必须取自会话详情的 ``current_node``，缺了会 422。 |
| """ |
| conversation_id = str(conversation_id or "").strip() |
| if not conversation_id: |
| raise RuntimeError("conversation_id is required") |
|
|
| detail = self._get_conversation(conversation_id) |
| current_node_id = str(detail.get("current_node") or "").strip() |
| if not current_node_id: |
| raise RuntimeError("上游会话没有 current_node，无法创建分享链接") |
|
|
| payload = self._upstream_maintenance_json( |
| "POST", |
| "/backend-api/share/create", |
| { |
| "conversation_id": conversation_id, |
| "current_node_id": current_node_id, |
| "is_anonymous": True, |
| "is_public": True, |
| }, |
| ) |
| share_id = str(payload.get("share_id") or "").strip() |
| share_url = str(payload.get("share_url") or "").strip() |
| if not share_id or not share_url: |
| raise RuntimeError("上游未返回 share_id / share_url，无法确认分享已建立") |
| logger.info({"event": "conversation_shared", "conversation_id": conversation_id, "share_id": share_id}) |
| return payload |
|
|
| def delete_share_link(self, share_id: str) -> bool: |
| """删除一条官方分享链接。删除后链接立即失效（``share_link_deleted``）。""" |
| share_id = str(share_id or "").strip() |
| if not share_id: |
| raise RuntimeError("share_id is required") |
| self._upstream_maintenance_json("DELETE", f"/backend-api/share/{quote(share_id, safe='')}") |
| logger.info({"event": "conversation_share_deleted", "share_id": share_id}) |
| return True |
|
|
| def share_link_alive(self, share_id: str, *, timeout: float = 20) -> bool: |
| """判断分享链接是否仍然可访问。 |
| |
| ⚠️ 分享页 HTML 永远返回 200（SPA 外壳），判活只能看正文接口 |
| ``GET /backend-api/share/{share_id}``：失效时是 404 + ``shared_conversation_deleted``。 |
| """ |
| share_id = str(share_id or "").strip() |
| if not share_id: |
| return False |
| try: |
| self._upstream_maintenance_json( |
| "GET", |
| f"/backend-api/share/{quote(share_id, safe='')}", |
| timeout=timeout, |
| ) |
| except Exception: |
| return False |
| return True |
|
|
| def prepare_upstream_maintenance(self) -> None: |
| """Bootstrap the authenticated browser session before account maintenance.""" |
| if not self.access_token: |
| raise RuntimeError("access_token is required") |
| self._bootstrap() |
|
|
| def _upstream_maintenance_json( |
| self, |
| method: str, |
| path: str, |
| body: Optional[Dict[str, Any]] = None, |
| *, |
| timeout: float = 60, |
| ) -> Dict[str, Any]: |
| extra_headers = {"Accept": "application/json"} |
| if body is not None: |
| extra_headers["Content-Type"] = "application/json" |
| response = self.session.request( |
| method, |
| self.base_url + path, |
| headers=self._headers(path, extra_headers), |
| json=body, |
| timeout=timeout, |
| ) |
| ensure_ok(response, path) |
| raw_body = str(getattr(response, "text", "") or "").strip() |
| if not raw_body: |
| return {} |
| try: |
| payload = response.json() |
| except Exception as exc: |
| raise RuntimeError(f"{path} returned invalid JSON") from exc |
| return payload if isinstance(payload, dict) else {} |
|
|
| def disable_upstream_memory(self) -> Dict[str, Any]: |
| """Ask ChatGPT to disable the account memory feature. |
| |
| The current ChatGPT web client sends this PATCH with query parameters and |
| no JSON body. Some accounts are migration-locked by ChatGPT and return |
| ``IMMUTABLE_SETTING``; keep that as an explicit, actionable failure |
| instead of pretending the switch was changed. |
| """ |
| path = "/backend-api/settings/account_user_setting?feature=sunshine&value=false" |
| try: |
| self._upstream_maintenance_json("PATCH", path) |
| except UpstreamHTTPError as exc: |
| detail = exc.body.get("detail") if isinstance(exc.body, dict) else None |
| error_code = "" |
| upstream_message = "" |
| if isinstance(detail, dict): |
| error_code = str(detail.get("error_code") or detail.get("code") or "").strip() |
| upstream_message = str(detail.get("message") or "").strip() |
| if exc.status_code == 403 and error_code.upper().replace("-", "_") == "IMMUTABLE_SETTING": |
| message = upstream_message or "请更新 ChatGPT，以更改你的记忆设置。" |
| raise RuntimeError( |
| f"关闭记忆被 ChatGPT 拒绝（IMMUTABLE_SETTING）：{message}" |
| ) from exc |
| raise |
|
|
| settings_path = "/backend-api/settings/user" |
| settings_payload = self._upstream_maintenance_json("GET", settings_path) |
| settings = settings_payload.get("settings") |
| if not isinstance(settings, dict): |
| settings = settings_payload |
| if "sunshine" not in settings: |
| raise RuntimeError("关闭记忆复核失败：上游设置未返回 sunshine") |
| if settings.get("sunshine") is not False: |
| raise RuntimeError("关闭记忆复核失败：sunshine 仍未关闭") |
| return {"sunshine": False} |
|
|
| def hide_upstream_conversations(self) -> Dict[str, Any]: |
| """Hide all upstream conversations without physically deleting them.""" |
| payload = self._upstream_maintenance_json( |
| "PATCH", |
| "/backend-api/conversations", |
| {"is_visible": False}, |
| ) |
| if payload.get("success") is not True: |
| raise RuntimeError("隐藏历史对话失败：上游未确认 success=true") |
| return payload |
|
|
| def delete_upstream_files(self, page_limit: int = 100) -> Dict[str, Any]: |
| """Enumerate the complete upstream file library, then delete and verify it.""" |
| try: |
| normalized_limit = max(1, min(100, int(page_limit))) |
| except (TypeError, ValueError): |
| normalized_limit = 100 |
|
|
| file_ids: list[str] = [] |
| seen_file_ids: set[str] = set() |
| files_deleted = 0 |
| pages_scanned = 0 |
| cursor = "" |
| seen_cursors: set[str] = set() |
|
|
| def partial_result(*, complete: bool = False) -> Dict[str, Any]: |
| return { |
| "files_seen": len(file_ids), |
| "files_deleted": files_deleted, |
| "pages_scanned": pages_scanned, |
| "complete": complete, |
| } |
|
|
| def parse_file_ids(payload: Dict[str, Any]) -> list[str]: |
| raw_items = payload.get("items") |
| if not isinstance(raw_items, list): |
| raise RuntimeError("附件列表响应中的 items 必须是数组") |
| parsed: list[str] = [] |
| for index, item in enumerate(raw_items): |
| if not isinstance(item, dict): |
| raise RuntimeError(f"附件列表第 {index + 1} 项不是对象") |
| raw_file_id = item.get("file_id") |
| if not isinstance(raw_file_id, str) or not raw_file_id.strip(): |
| raise RuntimeError(f"附件列表第 {index + 1} 项缺少有效 file_id") |
| parsed.append(raw_file_id.strip()) |
| return parsed |
|
|
| enumeration_complete = False |
| for _ in range(100): |
| request_body: Dict[str, Any] = {"limit": normalized_limit} |
| if cursor: |
| request_body["cursor"] = cursor |
| try: |
| payload = self._upstream_maintenance_json( |
| "POST", |
| "/backend-api/files/library", |
| request_body, |
| ) |
| pages_scanned += 1 |
| page_file_ids = parse_file_ids(payload) |
| except Exception as exc: |
| raise UpstreamFileDeleteError( |
| f"附件列表读取失败（已删除 {files_deleted} 个）：{exc}", |
| partial_result(), |
| ) from exc |
|
|
| for file_id in page_file_ids: |
| if file_id not in seen_file_ids: |
| seen_file_ids.add(file_id) |
| file_ids.append(file_id) |
|
|
| raw_next_cursor = payload.get("cursor") |
| if raw_next_cursor is None or raw_next_cursor == "": |
| enumeration_complete = True |
| break |
| if not isinstance(raw_next_cursor, str): |
| raise UpstreamFileDeleteError( |
| "附件列表读取失败：cursor 必须是字符串", |
| partial_result(), |
| ) |
| next_cursor = raw_next_cursor.strip() |
| if not next_cursor: |
| enumeration_complete = True |
| break |
| if next_cursor in seen_cursors: |
| raise UpstreamFileDeleteError( |
| "附件列表读取失败：上游返回了重复 cursor，无法确认已完整枚举", |
| partial_result(), |
| ) |
| seen_cursors.add(next_cursor) |
| cursor = next_cursor |
|
|
| if not enumeration_complete: |
| raise UpstreamFileDeleteError( |
| "附件列表超过 100 页仍未结束，已截断且未执行删除", |
| partial_result(), |
| ) |
|
|
| for file_id in file_ids: |
| try: |
| delete_payload = self._upstream_maintenance_json( |
| "DELETE", |
| f"/backend-api/files/{quote(file_id, safe='')}", |
| ) |
| if delete_payload.get("success") is False: |
| raise RuntimeError("上游明确返回 success=false") |
| except Exception as exc: |
| raise UpstreamFileDeleteError( |
| f"附件删除中断（已删除 {files_deleted} 个）：{exc}", |
| partial_result(), |
| ) from exc |
| files_deleted += 1 |
|
|
| try: |
| verification_payload = self._upstream_maintenance_json( |
| "POST", |
| "/backend-api/files/library", |
| {"limit": normalized_limit}, |
| ) |
| remaining_file_ids = parse_file_ids(verification_payload) |
| except Exception as exc: |
| raise UpstreamFileDeleteError( |
| f"附件删除复核失败（已删除 {files_deleted} 个）：{exc}", |
| partial_result(), |
| ) from exc |
| if remaining_file_ids: |
| raise UpstreamFileDeleteError( |
| f"附件删除复核失败：上游仍返回 {len(remaining_file_ids)} 个附件", |
| partial_result(), |
| ) |
|
|
| return partial_result(complete=True) |
|
|
| @staticmethod |
| def _add_unique(values: list[str], candidates: list[str]) -> None: |
| for candidate in candidates: |
| if candidate and candidate not in values: |
| values.append(candidate) |
|
|
| @classmethod |
| def _extract_image_reference_ids(cls, payload: Any) -> tuple[list[str], list[str]]: |
| file_ids: list[str] = [] |
| sediment_ids: list[str] = [] |
|
|
| def walk(value: Any) -> None: |
| if isinstance(value, str): |
| cls._add_unique(file_ids, FILE_SERVICE_ID_RE.findall(value)) |
| cls._add_unique(file_ids, REAL_IMAGE_FILE_ID_RE.findall(value)) |
| cls._add_unique(sediment_ids, SEDIMENT_ID_RE.findall(value)) |
| return |
| if isinstance(value, dict): |
| for item in value.values(): |
| walk(item) |
| return |
| if isinstance(value, list): |
| for item in value: |
| walk(item) |
|
|
| walk(payload) |
| return file_ids, sediment_ids |
|
|
| @classmethod |
| def _has_image_asset_pointer(cls, payload: Any) -> bool: |
| if isinstance(payload, dict): |
| if str(payload.get("content_type") or "") == "image_asset_pointer": |
| return True |
| asset_pointer = str(payload.get("asset_pointer") or "") |
| if asset_pointer.startswith(("file-service://", "sediment://")): |
| return True |
| return any(cls._has_image_asset_pointer(item) for item in payload.values()) |
| if isinstance(payload, list): |
| return any(cls._has_image_asset_pointer(item) for item in payload) |
| return False |
|
|
| def _extract_image_tool_records( |
| self, |
| data: Dict[str, Any], |
| *, |
| after_message_id: str = "", |
| min_create_time: float = 0.0, |
| ) -> list[Dict[str, Any]]: |
| """从 conversation 明细里提取图片工具输出记录。""" |
| mapping = data.get("mapping") or {} |
| after_message_id = str(after_message_id or "").strip() |
| after_create_time = 0.0 |
| if after_message_id: |
| after_node = mapping.get(after_message_id) or {} |
| after_message = (after_node or {}).get("message") or {} |
| try: |
| after_create_time = float(after_message.get("create_time") or 0) |
| except (TypeError, ValueError): |
| after_create_time = 0.0 |
|
|
| def is_descendant(message_id: str, ancestor_id: str) -> bool: |
| if not message_id or not ancestor_id or message_id == ancestor_id: |
| return False |
| seen: set[str] = set() |
| current = message_id |
| while current and current not in seen: |
| seen.add(current) |
| node = mapping.get(current) or {} |
| parent = str((node or {}).get("parent") or "").strip() |
| if parent == ancestor_id: |
| return True |
| current = parent |
| return False |
|
|
| records = [] |
| for message_id, node in mapping.items(): |
| message = (node or {}).get("message") or {} |
| author = message.get("author") or {} |
| metadata = message.get("metadata") or {} |
| content = message.get("content") or {} |
| role = str(author.get("role") or "").strip().lower() |
| try: |
| create_time = float(message.get("create_time") or 0) |
| except (TypeError, ValueError): |
| create_time = 0.0 |
| if after_message_id: |
| is_after_parent = is_descendant(str(message_id), after_message_id) |
| is_after_time = bool( |
| (after_create_time and create_time > after_create_time) |
| or (not after_create_time and min_create_time and create_time >= min_create_time) |
| ) |
| if not is_after_parent and not is_after_time: |
| continue |
| elif min_create_time and create_time and create_time < min_create_time: |
| continue |
| if role not in {"tool", "assistant"}: |
| continue |
| is_image_gen = metadata.get("async_task_type") == "image_gen" |
| has_asset_pointer = self._has_image_asset_pointer(content) or self._has_image_asset_pointer(metadata) |
| if role == "assistant" and not (is_image_gen or has_asset_pointer): |
| continue |
| file_ids, sediment_ids = self._extract_image_reference_ids({"content": content, "metadata": metadata}) |
| if not is_image_gen and not has_asset_pointer and not file_ids and not sediment_ids: |
| continue |
| records.append( |
| {"message_id": message_id, "create_time": create_time, "file_ids": file_ids, |
| "sediment_ids": sediment_ids}) |
| return sorted(records, key=lambda item: item["create_time"]) |
|
|
| def _poll_image_results( |
| self, |
| conversation_id: str, |
| timeout_secs: float = 120.0, |
| *, |
| after_message_id: str = "", |
| min_create_time: float = 0.0, |
| initial_file_ids: list[str] | None = None, |
| initial_sediment_ids: list[str] | None = None, |
| ) -> tuple[list[str], list[str], str]: |
| """Poll the conversation document until image file ids appear or budget runs out. |
| |
| - Sleeps image_poll_initial_wait_secs first (default 10s, +jitter). ChatGPT |
| image generation takes ~30s; polling immediately wastes requests and trips |
| a transient 429 the upstream returns within ~200ms of the SSE stream |
| closing (the conversation document is not yet committed). |
| - Subsequent polls are image_poll_interval_secs apart (default 10s). |
| - 每次轮询没有图片记录时会先判断「上游是不是已经用纯文字收尾了」 |
| （``current_node`` 是终局的 assistant 文本叶子）——是就立刻抛 |
| ``ImageTextOnlyAnswer`` 把原话带回去，不再白等整段预算。 |
| - On upstream 429 / 5xx or network errors, backs off exponentially |
| (capped at 16s, +jitter) honoring Retry-After when present. |
| - All sleeps stay within timeout_secs; on exhaustion raises ImagePollTimeoutError. |
| """ |
| start = time.time() |
| attempt = 0 |
| interval = float(config.image_poll_interval_secs) |
| initial_wait = float(config.image_poll_initial_wait_secs) |
| file_ids: list[str] = [] |
| sediment_ids: list[str] = [] |
| self._add_unique(file_ids, initial_file_ids or []) |
| self._add_unique(sediment_ids, initial_sediment_ids or []) |
| has_initial_ids = bool(file_ids or sediment_ids) |
| result_message_id = "" |
| |
| last_conversation: Dict[str, Any] | None = None |
| last_hit_key: tuple[tuple[str, ...], tuple[str, ...]] | None = ( |
| (tuple(file_ids), tuple(sediment_ids)) if has_initial_ids else None |
| ) |
| logger.info({ |
| "event": "image_poll_start", |
| "conversation_id": conversation_id, |
| "timeout_secs": timeout_secs, |
| "initial_wait_secs": initial_wait, |
| "interval_secs": interval, |
| "after_message_id": after_message_id, |
| "initial_file_ids": file_ids, |
| "initial_sediment_ids": sediment_ids, |
| }) |
|
|
| def _remaining() -> float: |
| return timeout_secs - (time.time() - start) |
|
|
| if has_initial_ids and config.image_settle_enabled: |
| settle_for = min(float(config.image_settle_secs), max(0.0, _remaining())) |
| if settle_for > 0: |
| time.sleep(settle_for) |
| elif initial_wait > 0: |
| jitter = random.uniform(0, min(2.0, initial_wait * 0.2)) |
| sleep_for = min(initial_wait + jitter, max(0.0, _remaining())) |
| if sleep_for > 0: |
| time.sleep(sleep_for) |
|
|
| def _retry_sleep(reason: str, status_code: int | None, error: str | None, retry_after: int | None) -> bool: |
| |
| base = retry_after if retry_after is not None else min(2 ** min(attempt, 4), 16) |
| backoff = base + random.uniform(0, 0.5) |
| remaining = _remaining() |
| if remaining <= 0: |
| return False |
| sleep_for = min(backoff, remaining) |
| log_payload: Dict[str, Any] = { |
| "event": "image_poll_retry", |
| "conversation_id": conversation_id, |
| "attempt": attempt, |
| "reason": reason, |
| "sleep_secs": round(sleep_for, 2), |
| } |
| if status_code is not None: |
| log_payload["status_code"] = status_code |
| if error is not None: |
| log_payload["error"] = error |
| logger.warning(log_payload) |
| time.sleep(sleep_for) |
| return True |
|
|
| while _remaining() > 0: |
| attempt += 1 |
| try: |
| conversation = self._get_conversation(conversation_id) |
| except UpstreamHTTPError as exc: |
| if exc.status_code in (429, 500, 502, 503, 504): |
| if _retry_sleep("upstream_status", exc.status_code, None, exc.retry_after): |
| continue |
| break |
| raise |
| except requests.exceptions.RequestException as exc: |
| if _retry_sleep("network", None, str(exc), None): |
| continue |
| break |
|
|
| last_conversation = conversation |
| records = self._extract_image_tool_records( |
| conversation, |
| after_message_id=after_message_id, |
| min_create_time=min_create_time, |
| ) |
| for record in records: |
| result_message_id = str(record.get("message_id") or result_message_id) |
| self._add_unique(file_ids, [str(item) for item in record["file_ids"]]) |
| self._add_unique(sediment_ids, [str(item) for item in record["sediment_ids"]]) |
| logger.debug({"event": "image_poll_check", "conversation_id": conversation_id, "attempt": attempt, |
| "file_ids": file_ids, "sediment_ids": sediment_ids, "record_count": len(records)}) |
| if file_ids or sediment_ids: |
| hit_key = (tuple(file_ids), tuple(sediment_ids)) |
| logger.info({"event": "image_poll_hit", "conversation_id": conversation_id, "file_ids": file_ids, |
| "sediment_ids": sediment_ids, "message_id": result_message_id, "attempt": attempt}) |
| if not config.image_check_before_hit_enabled: |
| return file_ids, sediment_ids, result_message_id |
| if last_hit_key == hit_key: |
| return file_ids, sediment_ids, result_message_id |
| last_hit_key = hit_key |
| if not config.image_settle_enabled: |
| return file_ids, sediment_ids, result_message_id |
| wait = min(float(config.image_settle_secs), max(0.0, _remaining())) |
| if wait <= 0: |
| return file_ids, sediment_ids, result_message_id |
| if wait > 0: |
| time.sleep(wait) |
| continue |
| logger.debug({"event": "image_poll_wait", "conversation_id": conversation_id, |
| "elapsed_secs": round(time.time() - start, 1)}) |
| |
| |
| text_only = self._terminal_text_answer( |
| conversation, |
| after_message_id=after_message_id, |
| min_create_time=min_create_time, |
| ) |
| if text_only: |
| logger.info({ |
| "event": "image_poll_terminal_text_answer", |
| "conversation_id": conversation_id, |
| "attempt": attempt, |
| "chars": len(text_only), |
| "elapsed_secs": round(time.time() - start, 1), |
| }) |
| raise ImageTextOnlyAnswer(text_only, conversation_id=conversation_id or "") |
| wait = min(interval, max(0.0, _remaining())) |
| if wait > 0: |
| time.sleep(wait) |
| logger.info({ |
| "event": "image_poll_timeout", |
| "conversation_id": conversation_id, |
| "timeout_secs": timeout_secs, |
| "attempts_made": attempt, |
| |
| "initial_wait_exhausted_budget": attempt == 0, |
| }) |
| |
| |
| |
| try: |
| final_conversation = self._get_conversation(conversation_id) |
| except Exception as exc: |
| logger.debug({ |
| "event": "image_poll_timeout_final_fetch_failed", |
| "conversation_id": conversation_id, |
| "error": repr(exc), |
| }) |
| final_conversation = None |
| if final_conversation: |
| last_conversation = final_conversation |
| final_text = self._terminal_text_answer( |
| final_conversation, |
| after_message_id=after_message_id, |
| min_create_time=min_create_time, |
| ) |
| if final_text: |
| logger.info({ |
| "event": "image_poll_terminal_text_answer", |
| "conversation_id": conversation_id, |
| "attempt": attempt, |
| "chars": len(final_text), |
| "elapsed_secs": round(time.time() - start, 1), |
| "found_at": "timeout_final_fetch", |
| }) |
| raise ImageTextOnlyAnswer(final_text, conversation_id=conversation_id or "") |
| upstream_text = self._latest_assistant_text(last_conversation) |
| if upstream_text: |
| logger.info({ |
| "event": "image_poll_timeout_upstream_text", |
| "conversation_id": conversation_id, |
| "chars": len(upstream_text), |
| }) |
| raise ImagePollTimeoutError( |
| image_poll_timeout_message(timeout_secs), |
| conversation_id or "", |
| upstream_text=upstream_text, |
| ) |
|
|
| @staticmethod |
| def _node_is_after_cutoff( |
| mapping: Dict[str, Any], |
| message_id: str, |
| create_time: float, |
| *, |
| after_message_id: str = "", |
| after_create_time: float = 0.0, |
| min_create_time: float = 0.0, |
| ) -> bool: |
| """该节点是否落在「本次提问之后」。 |
| |
| 两个界标互相兜底：``after_message_id``（多轮续写时上一轮的叶子）先按父链判后代， |
| 父链判不出来再按 ``create_time``；界标本身不在文档里时只认 ``min_create_time``。 |
| **判不出来一律当作「不在之后」**——宁可多说一句超时，也别把上一轮的文案 |
| 当成这一轮的结论。 |
| """ |
| if not after_message_id and not min_create_time: |
| return True |
| if not after_message_id: |
| return create_time >= min_create_time |
| if message_id == after_message_id: |
| return False |
| seen: set[str] = set() |
| current = message_id |
| while current and current not in seen: |
| seen.add(current) |
| node = mapping.get(current) or {} |
| parent = str((node or {}).get("parent") or "").strip() |
| if parent == after_message_id: |
| return True |
| current = parent |
| if after_create_time: |
| return create_time > after_create_time |
| return bool(min_create_time) and create_time >= min_create_time |
|
|
| def _latest_assistant_text( |
| self, |
| conversation: Dict[str, Any] | None, |
| *, |
| after_message_id: str = "", |
| min_create_time: float = 0.0, |
| ) -> str: |
| """取会话里最后一条对用户可见的 assistant 文本。 |
| |
| 轮询本身只关心图片 id，但超时后要回答「GPT 到底回了什么」——被策略拦下时 |
| 上游常常只回一段文字、既没调用生图工具也没有任何图片资产，这时文案是唯一线索。 |
| |
| ``after_message_id`` / ``min_create_time`` 用来把范围限制在**本次提问之后**： |
| 多轮（记忆）会话里旧回合的 assistant 文案同样是「可见文本」，不设界就会把 |
| 上一轮的回答当成这一轮的结论报给用户。 |
| """ |
| if not isinstance(conversation, dict): |
| return "" |
| mapping = conversation.get("mapping") or {} |
| if not isinstance(mapping, dict): |
| return "" |
| after_message_id = str(after_message_id or "").strip() |
| after_create_time = 0.0 |
| if after_message_id: |
| after_node = mapping.get(after_message_id) or {} |
| after_message = (after_node or {}).get("message") or {} |
| try: |
| after_create_time = float(after_message.get("create_time") or 0) |
| except (TypeError, ValueError): |
| after_create_time = 0.0 |
|
|
| def _is_after(message_id: str, create_time: float) -> bool: |
| return self._node_is_after_cutoff( |
| mapping, |
| message_id, |
| create_time, |
| after_message_id=after_message_id, |
| after_create_time=after_create_time, |
| min_create_time=min_create_time, |
| ) |
|
|
| best_time = float("-inf") |
| best_text = "" |
| for message_id, node in mapping.items(): |
| if not isinstance(node, dict): |
| continue |
| message = node.get("message") |
| if not isinstance(message, dict): |
| continue |
| author = message.get("author") if isinstance(message.get("author"), dict) else {} |
| if str(author.get("role") or "") != "assistant": |
| continue |
| metadata = message.get("metadata") if isinstance(message.get("metadata"), dict) else {} |
| if metadata.get("is_visually_hidden_from_conversation"): |
| continue |
| text = self._search_message_text(message) |
| if not text: |
| continue |
| try: |
| create_time = float(message.get("create_time") or 0) |
| except (TypeError, ValueError): |
| create_time = 0.0 |
| if not _is_after(str(message_id), create_time): |
| continue |
| if create_time >= best_time: |
| best_time = create_time |
| best_text = text |
| return best_text |
|
|
| def _terminal_text_answer( |
| self, |
| conversation: Dict[str, Any] | None, |
| *, |
| after_message_id: str = "", |
| min_create_time: float = 0.0, |
| ) -> str: |
| """上游是否已经用「纯文字」收尾（本轮永远不会再出图）？是则返回那段文字。 |
| |
| 真机取证（`6ace1f9f-…` 会话，用户只发「把图片上的文字改成英文」没带图）： |
| 整棵树只有 3 个节点 —— user → assistant(`reasoning_recap`) → assistant(`text`)， |
| 最后那条是**叶子**，`status=finished_successfully`、`metadata.is_complete=true`、 |
| `finish_details.type="stop"`、`recipient="all"`，**没有任何 image_gen 工具节点**。 |
| 回合本身只花了 8 秒，我们却按 `turn_use_case=="image gen"` 轮询了 111 秒。 |
| |
| 判据（缺一不可，任何一条不满足就返回空串 → 调用方继续常规轮询，宁慢勿错）： |
| ``current_node`` 指向的节点是 assistant 的 ``content_type=="text"``、真叶子 |
| （``children`` 为空）、``recipient=="all"``（工具调用中间态是 `image_gen`/`dalle`）、 |
| ``status=="finished_successfully"``、**有结束标记**、不在 |
| ``is_visually_hidden_from_conversation`` 里，且落在本次提问之后的界标内。 |
| |
| 「有结束标记」有三种等价形态（任一即可）： |
| ``metadata.is_complete is True`` / ``metadata.finish_details.type == "stop"`` / |
| ``metadata.is_error is True``。 |
| |
| 第三种来自第二条真机取证（`6ace48ab-…`，让 GPT 生成迪士尼壁纸被内容策略拦下）： |
| 整棵树是 user → assistant(`recipient=<工具id>`, `content_type=code`) → assistant 文本叶子， |
| 叶子 ``metadata`` 里**既没有 is_complete 也没有 finish_details**，只有 |
| ``{"is_error": true, "can_save": …, "permissions": …}`` —— 上游是用「错误/拒绝」收尾的， |
| 同样是终局（不会再出图）。漏掉这一种就会把 21 秒就出现的拒绝文案当成普通超时， |
| 白轮询掉整段预算（实测 86.8s / 91.8s / 30s 三次都多等了这么久）。 |
| """ |
| if not isinstance(conversation, dict): |
| return "" |
| mapping = conversation.get("mapping") or {} |
| if not isinstance(mapping, dict): |
| return "" |
| current_node_id = str(conversation.get("current_node") or "").strip() |
| if not current_node_id: |
| return "" |
| node = mapping.get(current_node_id) |
| if not isinstance(node, dict): |
| return "" |
| if node.get("children"): |
| |
| return "" |
| message = node.get("message") |
| if not isinstance(message, dict): |
| return "" |
| author = message.get("author") if isinstance(message.get("author"), dict) else {} |
| if str(author.get("role") or "") != "assistant": |
| return "" |
| if str(message.get("recipient") or "") != "all": |
| return "" |
| content = message.get("content") if isinstance(message.get("content"), dict) else {} |
| if str(content.get("content_type") or "") != "text": |
| return "" |
| metadata = message.get("metadata") if isinstance(message.get("metadata"), dict) else {} |
| if metadata.get("is_visually_hidden_from_conversation"): |
| return "" |
| if str(message.get("status") or "") != "finished_successfully": |
| return "" |
| finish_details = metadata.get("finish_details") if isinstance(metadata.get("finish_details"), dict) else {} |
| if not ( |
| metadata.get("is_complete") is True |
| or str(finish_details.get("type") or "") == "stop" |
| or metadata.get("is_error") is True |
| ): |
| return "" |
| text = self._search_message_text(message) |
| if not text: |
| return "" |
| try: |
| create_time = float(message.get("create_time") or 0) |
| except (TypeError, ValueError): |
| create_time = 0.0 |
| after_message_id = str(after_message_id or "").strip() |
| after_create_time = 0.0 |
| if after_message_id: |
| after_node = mapping.get(after_message_id) or {} |
| after_message = (after_node or {}).get("message") or {} |
| try: |
| after_create_time = float(after_message.get("create_time") or 0) |
| except (TypeError, ValueError): |
| after_create_time = 0.0 |
| if not self._node_is_after_cutoff( |
| mapping, |
| current_node_id, |
| create_time, |
| after_message_id=after_message_id, |
| after_create_time=after_create_time, |
| min_create_time=min_create_time, |
| ): |
| return "" |
| return text |
|
|
| def _get_file_download_url(self, file_id: str) -> str: |
| """获取文件下载地址。""" |
| path = f"/backend-api/files/{file_id}/download" |
| response = self.session.get(self.base_url + path, headers=self._headers(path, {"Accept": "application/json"}), |
| timeout=60) |
| ensure_ok(response, path) |
| data = response.json() |
| return data.get("download_url") or data.get("url") or "" |
|
|
| def _get_attachment_download_url(self, conversation_id: str, attachment_id: str) -> str: |
| """通过 conversation 附件接口获取下载地址。""" |
| path = f"/backend-api/conversation/{conversation_id}/attachment/{attachment_id}/download" |
| response = self.session.get(self.base_url + path, headers=self._headers(path, {"Accept": "application/json"}), |
| timeout=60) |
| ensure_ok(response, path) |
| data = response.json() |
| return data.get("download_url") or data.get("url") or "" |
|
|
| def _resolve_image_urls(self, conversation_id: str, file_ids: list[str], sediment_ids: list[str]) -> list[str]: |
| """把图片结果 id 解析成可下载 URL。""" |
| urls = [] |
| skip_patterns = {"file_upload"} |
| |
| |
| |
| |
| |
| |
| resolved_ids: set[str] = set() |
| for file_id in file_ids: |
| if file_id in skip_patterns: |
| logger.debug({ |
| "event": "image_file_id_skipped", |
| "source": "file", |
| "conversation_id": conversation_id, |
| "id": file_id, |
| }) |
| continue |
| try: |
| url = self._get_file_download_url(file_id) |
| except Exception as exc: |
| logger.debug({ |
| "event": "image_download_url_failed", |
| "source": "file", |
| "conversation_id": conversation_id, |
| "id": file_id, |
| "error": repr(exc), |
| }) |
| continue |
| if url: |
| resolved_ids.add(file_id) |
| if url not in urls: |
| urls.append(url) |
| else: |
| logger.debug({ |
| "event": "image_download_url_empty", |
| "source": "file", |
| "conversation_id": conversation_id, |
| "id": file_id, |
| }) |
| if not conversation_id: |
| logger.debug({ |
| "event": "image_urls_resolved", |
| "conversation_id": conversation_id, |
| "file_ids": file_ids, |
| "sediment_ids": sediment_ids, |
| "urls": urls, |
| }) |
| return urls |
| for sediment_id in sediment_ids: |
| if sediment_id in resolved_ids: |
| |
| logger.debug({ |
| "event": "image_sediment_id_already_resolved", |
| "conversation_id": conversation_id, |
| "id": sediment_id, |
| }) |
| continue |
| try: |
| url = self._get_attachment_download_url(conversation_id, sediment_id) |
| except Exception as exc: |
| logger.debug({ |
| "event": "image_download_url_failed", |
| "source": "sediment", |
| "conversation_id": conversation_id, |
| "id": sediment_id, |
| "error": repr(exc), |
| }) |
| continue |
| if url: |
| if url not in urls: |
| urls.append(url) |
| else: |
| logger.debug({ |
| "event": "image_download_url_empty", |
| "source": "sediment", |
| "conversation_id": conversation_id, |
| "id": sediment_id, |
| }) |
| logger.debug({ |
| "event": "image_urls_resolved", |
| "conversation_id": conversation_id, |
| "file_ids": file_ids, |
| "sediment_ids": sediment_ids, |
| "urls": urls, |
| }) |
| return urls |
|
|
| def resolve_conversation_image_result( |
| self, |
| conversation_id: str, |
| file_ids: list[str], |
| sediment_ids: list[str], |
| poll: bool = True, |
| after_message_id: str = "", |
| min_create_time: float = 0.0, |
| timeout_secs: float | None = None, |
| ) -> ImageResolutionResult: |
| file_ids = [item for item in file_ids if item != "file_upload"] |
| sediment_ids = list(sediment_ids) |
| result_message_id = "" |
| timeout = config.image_poll_timeout_secs if timeout_secs is None else timeout_secs |
| filtered_poll = bool(after_message_id) or bool(min_create_time) |
| initial_file_ids = list(file_ids) |
| initial_sediment_ids = list(sediment_ids) |
| should_poll = poll and conversation_id and ( |
| not file_ids |
| and not sediment_ids |
| or filtered_poll |
| or config.image_check_before_hit_enabled |
| or config.image_settle_enabled |
| ) |
| if should_poll: |
| if filtered_poll: |
| file_ids = [] |
| sediment_ids = [] |
| initial_file_ids = [] |
| initial_sediment_ids = [] |
| logger.info({ |
| "event": "image_resolve_poll_needed", |
| "conversation_id": conversation_id, |
| "after_message_id": after_message_id, |
| "initial_file_ids": initial_file_ids, |
| "initial_sediment_ids": initial_sediment_ids, |
| }) |
| try: |
| polled_file_ids, polled_sediment_ids, result_message_id = self._poll_image_results( |
| conversation_id, |
| timeout, |
| after_message_id=after_message_id, |
| min_create_time=min_create_time, |
| initial_file_ids=initial_file_ids, |
| initial_sediment_ids=initial_sediment_ids, |
| ) |
| except Exception as exc: |
| if filtered_poll or not (initial_file_ids or initial_sediment_ids): |
| raise |
| logger.warning({ |
| "event": "image_resolve_poll_failed_use_initial_ids", |
| "conversation_id": conversation_id, |
| "initial_file_ids": initial_file_ids, |
| "initial_sediment_ids": initial_sediment_ids, |
| "error": repr(exc), |
| }) |
| else: |
| file_ids.extend(item for item in polled_file_ids if item and item not in file_ids) |
| sediment_ids.extend(item for item in polled_sediment_ids if item and item not in sediment_ids) |
| return ImageResolutionResult( |
| urls=self._resolve_image_urls(conversation_id, file_ids, sediment_ids), |
| message_id=result_message_id, |
| ) |
|
|
| def resolve_conversation_image_urls( |
| self, |
| conversation_id: str, |
| file_ids: list[str], |
| sediment_ids: list[str], |
| poll: bool = True, |
| after_message_id: str = "", |
| min_create_time: float = 0.0, |
| ) -> list[str]: |
| return self.resolve_conversation_image_result( |
| conversation_id, |
| file_ids, |
| sediment_ids, |
| poll=poll, |
| after_message_id=after_message_id, |
| min_create_time=min_create_time, |
| ).urls |
|
|
| def download_image_bytes(self, urls: list[str]) -> list[bytes]: |
| images = [] |
| for url in urls: |
| response = self.session.get(url, timeout=120) |
| ensure_ok(response, "image_download") |
| images.append(response.content) |
| return images |
|
|
| def stream_conversation( |
| self, |
| messages: Optional[list[Dict[str, Any]]] = None, |
| model: str = "auto", |
| prompt: str = "", |
| images: Optional[list[str]] = None, |
| system_hints: Optional[list[str]] = None, |
| upstream_conversation_id: str = "", |
| upstream_parent_message_id: str = "", |
| deadline_at: float = 0.0, |
| timeout_secs: float = 0.0, |
| thinking_effort: str = "", |
| ) -> Iterator[str]: |
| system_hints = system_hints or [] |
| if "picture_v2" in system_hints: |
| yield from self._stream_picture_conversation( |
| prompt, |
| model, |
| images or [], |
| upstream_conversation_id=upstream_conversation_id, |
| upstream_parent_message_id=upstream_parent_message_id, |
| deadline_at=deadline_at, |
| timeout_secs=timeout_secs, |
| ) |
| return |
|
|
| normalized = messages or [{"role": "user", "content": prompt}] |
| self._bootstrap() |
| requirements = self._get_chat_requirements() |
| path, timezone = self._chat_target() |
| payload = self._conversation_payload(normalized, model, timezone, thinking_effort=thinking_effort) |
| response = self.session.post( |
| self.base_url + path, |
| headers=self._conversation_headers(path, requirements), |
| json=payload, |
| timeout=300, |
| stream=True, |
| ) |
| ensure_ok(response, path) |
| try: |
| yield from iter_sse_payloads(response) |
| finally: |
| response.close() |
|
|
| def _report_progress(self, step: str) -> None: |
| if self.progress_callback: |
| try: |
| self.progress_callback(step) |
| except Exception: |
| pass |
|
|
| def _stream_picture_conversation( |
| self, |
| prompt: str, |
| model: str, |
| images: list[str], |
| upstream_conversation_id: str = "", |
| upstream_parent_message_id: str = "", |
| deadline_at: float = 0.0, |
| timeout_secs: float = 0.0, |
| ) -> Iterator[str]: |
| def remaining_timeout(default: float) -> float: |
| if deadline_at <= 0: |
| return default |
| remaining = deadline_at - time.time() |
| if remaining <= 0: |
| raise ImagePollTimeoutError(image_poll_timeout_message(timeout_secs or config.image_poll_timeout_secs)) |
| return max(1.0, remaining) |
|
|
| if not self.access_token: |
| raise RuntimeError("access_token is required for image endpoints") |
| try: |
| if images: |
| self._report_progress("uploading") |
| references = [self._upload_image(image, f"image_{idx}.png") for idx, image in enumerate(images, start=1)] |
| self._report_progress("bootstrapping") |
| self._bootstrap() |
| self._report_progress("getting_token") |
| requirements = self._get_chat_requirements() |
| parent_message_id = upstream_parent_message_id or new_uuid() |
| self._report_progress("preparing_conversation") |
| conduit_token = self._prepare_image_conversation( |
| prompt, |
| requirements, |
| model, |
| conversation_id=upstream_conversation_id, |
| parent_message_id=parent_message_id, |
| request_timeout=remaining_timeout(60.0), |
| ) |
| self._report_progress("starting_generation") |
| response = self._start_image_generation( |
| prompt, |
| requirements, |
| conduit_token, |
| model, |
| references, |
| conversation_id=upstream_conversation_id, |
| parent_message_id=parent_message_id, |
| request_timeout=remaining_timeout(300.0), |
| ) |
| self._report_progress("generating") |
| except requests.exceptions.Timeout as exc: |
| if deadline_at > 0: |
| raise ImagePollTimeoutError(image_poll_timeout_message(timeout_secs or config.image_poll_timeout_secs)) from exc |
| raise |
| hard_cap_secs = float(config.image_poll_timeout_secs) |
| if deadline_at > 0: |
| hard_cap_secs = min(hard_cap_secs, max(1.0, deadline_at - time.time())) |
| try: |
| yield from self._iter_sse_payloads_capped(response, hard_cap_secs) |
| except requests.exceptions.Timeout as exc: |
| if deadline_at > 0: |
| raise ImagePollTimeoutError(image_poll_timeout_message(timeout_secs or config.image_poll_timeout_secs)) from exc |
| raise |
| finally: |
| response.close() |
|
|
| def _iter_sse_payloads_capped(self, response: Any, hard_cap_secs: float) -> Iterator[str]: |
| hard_cap_secs = max(1.0, float(hard_cap_secs)) |
| deadline = time.monotonic() + hard_cap_secs |
| watchdog = threading.Timer(hard_cap_secs, response.close) |
| watchdog.daemon = True |
| watchdog.start() |
| timeout_message = f"图片生成流已超过硬上限 {int(hard_cap_secs)} 秒，已强制中断（上游可能未生成图片）" |
| try: |
| for payload in iter_sse_payloads(response): |
| yield payload |
| if time.monotonic() >= deadline: |
| raise ImageStreamHardTimeoutError(timeout_message) |
| except ImageStreamHardTimeoutError: |
| raise |
| except Exception as exc: |
| if time.monotonic() >= deadline: |
| raise ImageStreamHardTimeoutError(timeout_message) from exc |
| raise |
| finally: |
| watchdog.cancel() |
| try: |
| response.close() |
| except Exception: |
| pass |
|
|
| def _list_recent_conversations(self, limit: int = 5, timeout_secs: float = 10.0) -> list[Dict[str, Any]]: |
| """列出最近的对话列表，按更新时间倒序。 |
| |
| 当 SSE 流太短导致 conversation_id 丢失时，可以通过此方法 |
| 查找最近创建的对话来恢复 conversation_id。 |
| """ |
| path = f"/backend-api/conversations?offset=0&limit={limit}&order=updated&conversation_filter=all" |
| try: |
| response = self.session.get( |
| self.base_url + path, |
| headers=self._headers(path, {"Accept": "application/json"}), |
| timeout=timeout_secs, |
| ) |
| ensure_ok(response, path) |
| data = response.json() |
| return data.get("items") or data.get("conversations") or [] |
| except Exception as exc: |
| logger.debug({"event": "list_conversations_failed", "error": str(exc)}) |
| return [] |
|
|
| def find_conversation_by_prompt(self, prompt: str, started_at: float, timeout_secs: float = 10.0) -> str: |
| """根据 prompt 和开始时间，从最近对话列表中查找匹配的 conversation_id。 |
| |
| 当 SSE 流太短导致 conversation_id 丢失时，使用此方法恢复。 |
| 通过对比 prompt 关键词和时间戳来匹配最可能的对话。 |
| |
| 参数： |
| prompt: 用户输入的 prompt 文本 |
| started_at: 请求开始的时间戳（epoch seconds） |
| timeout_secs: 请求超时秒数 |
| |
| 返回： |
| 匹配的 conversation_id，如果未找到返回空字符串 |
| """ |
| items = self._list_recent_conversations(limit=10, timeout_secs=timeout_secs) |
| if not items: |
| return "" |
| |
| |
| prompt_lower = str(prompt or "").lower().strip() |
| best_match = "" |
| best_score = 0.0 |
| for item in items: |
| |
| conv_id = str(item.get("id") or item.get("conversation_id") or "") |
| if not conv_id: |
| continue |
| |
| updated_at = float(item.get("update_time") or item.get("updated_at") or 0) |
| if updated_at and started_at and (updated_at < started_at - 30 or updated_at > started_at + 600): |
| continue |
| |
| title = str(item.get("title") or "").lower() |
| |
| score = 0.0 |
| if prompt_lower and title: |
| |
| prompt_words = set(prompt_lower.split()) |
| title_words = set(title.split()) |
| common = prompt_words & title_words |
| if common: |
| score = len(common) / max(len(prompt_words), 1) |
| |
| if title.startswith("image"): |
| score += 0.3 |
| if score > best_score: |
| best_score = score |
| best_match = conv_id |
| if best_match and best_score > 0.1: |
| logger.info({ |
| "event": "conversation_prompt_match_found", |
| "conversation_id": best_match, |
| "match_score": round(best_score, 2), |
| }) |
| return best_match |
| |
| for item in items: |
| conv_id = str(item.get("id") or item.get("conversation_id") or "") |
| updated_at = float(item.get("update_time") or item.get("updated_at") or 0) |
| if conv_id and updated_at and started_at and updated_at >= started_at - 30: |
| logger.info({ |
| "event": "conversation_latest_match", |
| "conversation_id": conv_id, |
| "updated_at": updated_at, |
| }) |
| return conv_id |
| return "" |
|
|
| def search(self, prompt: str, model: str = SEARCH_MODEL, timeout_secs: float = SEARCH_TIMEOUT_SECS, |
| poll_interval_secs: float = SEARCH_POLL_INTERVAL_SECS) -> Dict[str, Any]: |
| if not self.access_token: |
| raise RuntimeError("access_token is required for search") |
| conduit_token = self._prepare_search_conversation(prompt, model) |
| self._bootstrap() |
| conversation_id = self._run_search_conversation(prompt, conduit_token, model) |
| return self._wait_search_result(conversation_id, timeout_secs, poll_interval_secs) |
|
|
| def _prepare_search_conversation(self, prompt: str, model: str) -> str: |
| path = "/backend-api/f/conversation/prepare" |
| response = self.session.post( |
| self.base_url + path, |
| headers=self._headers(path, {"Accept": "*/*", "Content-Type": "application/json", "X-Conduit-Token": "no-token"}), |
| json={ |
| "action": "next", |
| "fork_from_shared_post": False, |
| "parent_message_id": "client-created-root", |
| "model": model, |
| "client_prepare_state": "success", |
| "timezone_offset_min": -480, |
| "timezone": "Asia/Shanghai", |
| "conversation_mode": {"kind": "primary_assistant"}, |
| "system_hints": ["search"], |
| "partial_query": {"id": new_uuid(), "author": {"role": "user"}, "content": {"content_type": "text", "parts": [prompt]}}, |
| "supports_buffering": True, |
| "supported_encodings": ["v1"], |
| "client_contextual_info": {"app_name": "chatgpt.com"}, |
| }, |
| timeout=60, |
| ) |
| ensure_ok(response, path) |
| token = str(response.json().get("conduit_token") or "") |
| if not token: |
| raise RuntimeError("missing conduit_token") |
| return token |
|
|
| def _run_search_conversation(self, prompt: str, conduit_token: str, model: str) -> str: |
| requirements = self._get_chat_requirements() |
| path = "/backend-api/f/conversation" |
| response = self.session.post( |
| self.base_url + path, |
| headers=self._image_headers(path, requirements, conduit_token, "text/event-stream"), |
| json={ |
| "action": "next", |
| "messages": [{ |
| "id": new_uuid(), |
| "author": {"role": "user"}, |
| "create_time": time.time(), |
| "content": {"content_type": "text", "parts": [prompt]}, |
| "metadata": { |
| "developer_mode_connector_ids": [], |
| "selected_github_repos": [], |
| "selected_all_github_repos": False, |
| "system_hints": ["search"], |
| "serialization_metadata": {"custom_symbol_offsets": []}, |
| }, |
| }], |
| "parent_message_id": "client-created-root", |
| "model": model, |
| "client_prepare_state": "success", |
| "timezone_offset_min": -480, |
| "timezone": "Asia/Shanghai", |
| "conversation_mode": {"kind": "primary_assistant"}, |
| "enable_message_followups": True, |
| "system_hints": [], |
| "supports_buffering": True, |
| "supported_encodings": ["v1"], |
| "force_use_search": True, |
| "client_reported_search_source": "conversation_composer_web_icon", |
| "client_contextual_info": {"is_dark_mode": False, "time_since_loaded": 36, "page_height": 925, "page_width": 886, "pixel_ratio": 2, "screen_height": 1440, "screen_width": 2560, "app_name": "chatgpt.com"}, |
| "paragen_cot_summary_display_override": "allow", |
| "force_parallel_switch": "auto", |
| }, |
| timeout=300, |
| stream=True, |
| ) |
| ensure_ok(response, path) |
| conversation_id = "" |
| try: |
| for payload in iter_sse_payloads(response): |
| conversation_id = conversation_id or self._find_search_value(payload, "conversation_id") |
| if payload == "[DONE]": |
| break |
| finally: |
| response.close() |
| if not conversation_id: |
| raise RuntimeError("conversation_id not found in stream") |
| return conversation_id |
|
|
| def _wait_search_result(self, conversation_id: str, timeout_secs: float, poll_interval_secs: float) -> Dict[str, Any]: |
| deadline = time.time() + timeout_secs |
| last_result: Dict[str, Any] | None = None |
| last_answer = "" |
| stable_hits = 0 |
| while time.time() < deadline: |
| try: |
| last_result = self._extract_search_result(conversation_id, self._get_search_conversation(conversation_id)) |
| except UpstreamHTTPError as exc: |
| if exc.status_code not in {404, 409, 423, 429, 500, 502, 503, 504}: |
| raise |
| if last_result and last_result.get("answer"): |
| if last_result.get("status") in SEARCH_DONE_STATUS: |
| return last_result |
| answer = str(last_result.get("answer") or "") |
| stable_hits = stable_hits + 1 if answer == last_answer else 0 |
| last_answer = answer |
| if stable_hits >= 2: |
| return last_result |
| time.sleep(poll_interval_secs) |
| if last_result: |
| return last_result |
| raise RuntimeError(f"timed out waiting for search result: {conversation_id}") |
|
|
| def _get_search_conversation(self, conversation_id: str) -> Dict[str, Any]: |
| path = f"/backend-api/conversation/{conversation_id}" |
| headers = self._headers(path, {"Accept": "*/*"}) |
| headers["Referer"] = f"{self.base_url}/c/{conversation_id}" |
| headers["X-OpenAI-Target-Route"] = "/backend-api/conversation/{conversation_id}" |
| response = self.session.get(self.base_url + path, headers=headers, timeout=60) |
| ensure_ok(response, path) |
| return response.json() |
|
|
| def _extract_search_result(self, conversation_id: str, conversation: Dict[str, Any]) -> Dict[str, Any]: |
| messages = [] |
| for node in (conversation.get("mapping") or {}).values(): |
| message = (node or {}).get("message") or {} |
| if ((message.get("author") or {}).get("role") or "") == "assistant": |
| messages.append(message) |
| message = max(messages, key=lambda item: float(item.get("create_time") or 0.0)) if messages else {} |
| metadata = message.get("metadata") if isinstance(message.get("metadata"), dict) else {} |
| finish_details = metadata.get("finish_details") if isinstance(metadata.get("finish_details"), dict) else {} |
| answer = self._search_message_text(message) |
| sources = self._extract_search_sources(message) |
| images = self._extract_search_images(message) |
| for url in SEARCH_URL_RE.findall(answer): |
| url = self._clean_search_url(url) |
| if url and all(item["url"] != url for item in sources): |
| sources.append({"title": "", "url": url, "snippet": "", "source_type": ""}) |
| return { |
| "conversation_id": conversation_id, |
| "status": str(finish_details.get("type") or metadata.get("status") or self._find_search_value(message, "status") or "").strip(), |
| "answer": answer, |
| "sources": sources, |
| "images": images, |
| "assistant_message_id": str(message.get("id") or ""), |
| "create_time": float(message.get("create_time") or 0.0), |
| } |
|
|
| def _extract_search_sources(self, payload: Any) -> list[Dict[str, str]]: |
| sources: list[Dict[str, str]] = [] |
| for obj in self._walk_search_dicts(payload): |
| metadata = obj.get("metadata") if isinstance(obj.get("metadata"), dict) else {} |
| url = self._clean_search_url(obj.get("url") or obj.get("link") or obj.get("source_url") or metadata.get("url")) |
| if url and all(item["url"] != url for item in sources): |
| sources.append({ |
| "title": str(obj.get("title") or obj.get("name") or obj.get("source") or "").strip(), |
| "url": url, |
| "snippet": str(obj.get("snippet") or obj.get("text") or obj.get("description") or "").strip(), |
| "source_type": str(obj.get("type") or obj.get("source_type") or "").strip(), |
| }) |
| return sources |
|
|
| def _extract_search_images(self, payload: Any) -> list[Dict[str, Any]]: |
| """Extract ChatGPT image-search cards from assistant metadata.""" |
| images: list[Dict[str, Any]] = [] |
| seen: set[str] = set() |
| for obj in self._walk_search_dicts(payload): |
| image_result = obj.get("image_result") |
| if not isinstance(image_result, dict): |
| continue |
| content_url = self._clean_search_url(image_result.get("content_url")) |
| thumbnail_url = self._clean_search_url(image_result.get("thumbnail_url")) |
| image_url = content_url or thumbnail_url |
| if not image_url or image_url in seen: |
| continue |
| seen.add(image_url) |
| size = image_result.get("content_size") if isinstance(image_result.get("content_size"), dict) else {} |
| images.append({ |
| "id": str(obj.get("image_ref_id") or f"search-image-{len(images) + 1}"), |
| "title": str(image_result.get("title") or "搜索图片").strip() or "搜索图片", |
| "content_url": image_url, |
| "thumbnail_url": thumbnail_url or image_url, |
| "source_url": self._clean_search_url(image_result.get("url")), |
| "query": str(obj.get("image_search_query") or "").strip(), |
| "width": int(size.get("width") or 0), |
| "height": int(size.get("height") or 0), |
| }) |
| return images |
|
|
| def _search_message_text(self, message: Any) -> str: |
| content = message.get("content") if isinstance(message, dict) else {} |
| parts = [] |
| if isinstance(content, dict): |
| if isinstance(content.get("text"), str): |
| parts.append(content["text"]) |
| for part in content.get("parts") or []: |
| if isinstance(part, str): |
| parts.append(part) |
| elif isinstance(part, dict): |
| parts.extend(str(part.get(key) or "") for key in ("text", "summary", "content") if part.get(key)) |
| elif isinstance(content, str): |
| parts.append(content) |
| return "\n".join(part.strip() for part in parts if str(part).strip()).strip() |
|
|
| def _find_search_value(self, payload: Any, key: str) -> str: |
| if isinstance(payload, str): |
| match = SEARCH_CONVERSATION_ID_RE.search(payload) if key == "conversation_id" else None |
| if match: |
| return match.group(1) |
| try: |
| payload = json.loads(payload) |
| except json.JSONDecodeError: |
| return "" |
| if isinstance(payload, dict): |
| value = payload.get(key) |
| if isinstance(value, str) and value: |
| return value |
| return next((found for item in payload.values() if (found := self._find_search_value(item, key))), "") |
| if isinstance(payload, list): |
| return next((found for item in payload if (found := self._find_search_value(item, key))), "") |
| return "" |
|
|
| def _walk_search_dicts(self, payload: Any) -> list[Dict[str, Any]]: |
| if isinstance(payload, dict): |
| return [payload, *(item for value in payload.values() for item in self._walk_search_dicts(value))] |
| if isinstance(payload, list): |
| return [item for value in payload for item in self._walk_search_dicts(value)] |
| return [] |
|
|
| def _clean_search_url(self, value: Any) -> str: |
| return str(value or "").strip().rstrip(".,;，。；") |
|
|
| @staticmethod |
| def _editable_prompt(fixed_prompt: str, user_prompt_text: str) -> str: |
| extra = str(user_prompt_text or "").strip() |
| return fixed_prompt if not extra else fixed_prompt + "\n\n以下是用户补充需求，请直接结合执行：\n" + extra |
|
|
| def export_ppt_zip( |
| self, |
| base64_images: list[str] | None, |
| prompt: str, |
| output_dir: str | Path = EDITABLE_FILE_PPT_OUTPUT_DIR, |
| timeout_secs: float = EDITABLE_FILE_TIMEOUT_SECS, |
| poll_interval_secs: float = EDITABLE_FILE_POLL_INTERVAL_SECS, |
| ) -> EditableFileExportResult: |
| return self._export_editable_file_zip( |
| base64_images or [], |
| self._editable_prompt(EDITABLE_FILE_PPT_PROMPT, prompt), |
| output_dir, |
| primary_label="ppt", |
| primary_suffixes=(".ppt", ".pptx"), |
| primary_mime_types=EDITABLE_PPT_MIME_TYPES, |
| primary_mime_keywords=("presentationml.presentation", "ms-powerpoint"), |
| primary_default_extension=".pptx", |
| export_file_re=EDITABLE_PPT_EXPORT_FILE_RE, |
| timeout_secs=timeout_secs, |
| poll_interval_secs=poll_interval_secs, |
| ) |
|
|
| def export_psd_zip( |
| self, |
| base64_images: list[str], |
| prompt: str, |
| output_dir: str | Path = EDITABLE_FILE_PSD_OUTPUT_DIR, |
| timeout_secs: float = EDITABLE_FILE_TIMEOUT_SECS, |
| poll_interval_secs: float = EDITABLE_FILE_POLL_INTERVAL_SECS, |
| ) -> EditableFileExportResult: |
| if not base64_images: |
| raise ValueError("base64_images is empty") |
| return self._export_editable_file_zip( |
| base64_images, |
| self._editable_prompt(EDITABLE_FILE_PSD_PROMPT, prompt), |
| output_dir, |
| primary_label="psd", |
| primary_suffixes=(".psd",), |
| primary_mime_types=EDITABLE_PSD_MIME_TYPES, |
| primary_mime_keywords=("photoshop",), |
| primary_default_extension=".psd", |
| export_file_re=EDITABLE_PSD_EXPORT_FILE_RE, |
| timeout_secs=timeout_secs, |
| poll_interval_secs=poll_interval_secs, |
| ) |
|
|
| def _export_editable_file_zip( |
| self, |
| base64_images: list[str], |
| prompt: str, |
| output_dir: str | Path, |
| *, |
| primary_label: str, |
| primary_suffixes: tuple[str, ...], |
| primary_mime_types: set[str], |
| primary_mime_keywords: tuple[str, ...], |
| primary_default_extension: str, |
| export_file_re: re.Pattern[str], |
| timeout_secs: float, |
| poll_interval_secs: float, |
| ) -> EditableFileExportResult: |
| if not self.access_token: |
| raise RuntimeError("access_token is required for editable file export") |
| self.client_version = EDITABLE_FILE_CLIENT_VERSION |
| self.client_build_number = EDITABLE_FILE_CLIENT_BUILD_NUMBER |
| self.session.headers["OAI-Client-Version"] = EDITABLE_FILE_CLIENT_VERSION |
| self.session.headers["OAI-Client-Build-Number"] = EDITABLE_FILE_CLIENT_BUILD_NUMBER |
| output_path = Path(output_dir).expanduser().resolve() |
| output_path.mkdir(parents=True, exist_ok=True) |
| uploaded = [self._upload_editable_base64_image(item, index) for index, item in enumerate(base64_images, start=1)] |
| conduit_token = self._prepare_editable_conversation(prompt, [item["mime_type"] for item in uploaded]) |
| conversation_id = self._run_editable_conversation(prompt, uploaded, conduit_token) |
| artifacts = self._wait_editable_output_artifacts( |
| conversation_id, |
| primary_label, |
| primary_suffixes, |
| primary_mime_types, |
| primary_mime_keywords, |
| export_file_re, |
| timeout_secs, |
| poll_interval_secs, |
| ) |
| downloaded = [self._download_editable_artifact( |
| conversation_id, |
| item, |
| output_path, |
| primary_mime_types, |
| primary_mime_keywords, |
| primary_default_extension, |
| ) for item in artifacts] |
| primary_path = next((item for item in downloaded if item.suffix.lower() in primary_suffixes), None) |
| zip_path = next((item for item in downloaded if item.suffix.lower() == ".zip"), None) |
| if not primary_path or not zip_path: |
| raise RuntimeError(f"download finished but did not get both {primary_label} and zip files: {downloaded}") |
| return EditableFileExportResult(conversation_id=conversation_id, primary_path=primary_path, zip_path=zip_path) |
|
|
| def _upload_editable_base64_image(self, base64_image: str, index: int) -> Dict[str, Any]: |
| data, file_name, mime_type, width, height = self._decode_editable_base64_image(base64_image, index) |
| path = "/backend-api/files" |
| response = self.session.post( |
| self.base_url + path, |
| headers=self._headers(path, {"Accept": "*/*", "Content-Type": "application/json"}), |
| json={ |
| "file_name": file_name, |
| "file_size": len(data), |
| "use_case": "multimodal", |
| "timezone_offset_min": -480, |
| "reset_rate_limits": False, |
| "store_in_library": True, |
| "library_persistence_mode": "opportunistic", |
| }, |
| timeout=60, |
| ) |
| ensure_ok(response, path) |
| payload = response.json() |
| upload_url = str(payload.get("upload_url") or "") |
| file_id = str(payload.get("file_id") or "") |
| if not upload_url or not file_id: |
| raise RuntimeError(f"invalid upload response: {payload}") |
| response = self.session.put( |
| upload_url, |
| headers={ |
| "Content-Type": mime_type, |
| "x-ms-blob-type": "BlockBlob", |
| "x-ms-version": "2020-04-08", |
| "Origin": self.base_url, |
| "Referer": self.base_url + "/", |
| "User-Agent": self.user_agent, |
| "Accept": "application/json, text/plain, */*", |
| "Accept-Language": "en-US,en;q=0.8", |
| }, |
| data=data, |
| timeout=120, |
| ) |
| ensure_ok(response, "image_upload") |
| path = f"/backend-api/files/{file_id}/uploaded" |
| response = self.session.post( |
| self.base_url + path, |
| headers=self._headers(path, {"Accept": "*/*", "Content-Type": "application/json"}), |
| data="{}", |
| timeout=60, |
| ) |
| ensure_ok(response, path) |
| return { |
| "file_id": file_id, |
| "library_file_id": str(payload.get("library_file_id") or ""), |
| "file_name": file_name, |
| "file_size": len(data), |
| "mime_type": mime_type, |
| "width": width, |
| "height": height, |
| } |
|
|
| def _decode_editable_base64_image(self, base64_image: str, index: int) -> tuple[bytes, str, str, int, int]: |
| raw = str(base64_image or "").strip() |
| if not raw: |
| raise ValueError("base64 image is empty") |
| mime_type = "" |
| payload = raw |
| match = re.match(r"^data:([^;]+);base64,(.*)$", raw, re.IGNORECASE | re.DOTALL) |
| if match: |
| mime_type = str(match.group(1) or "").strip().lower() |
| payload = str(match.group(2) or "").strip() |
| data = base64.b64decode(payload) |
| image = Image.open(BytesIO(data)) |
| image.load() |
| width, height = image.size |
| mime_type = Image.MIME.get(image.format, mime_type or "image/png") |
| extension = mimetypes.guess_extension(mime_type) or ".png" |
| return data, f"image_{index}{extension}", mime_type, width, height |
|
|
| def _prepare_editable_conversation(self, prompt: str, attachment_mime_types: list[str]) -> str: |
| path = "/backend-api/f/conversation/prepare" |
| payload: Dict[str, Any] = { |
| "action": "next", |
| "fork_from_shared_post": False, |
| "parent_message_id": "client-created-root", |
| "model": EDITABLE_FILE_MODEL, |
| "client_prepare_state": "success", |
| "timezone_offset_min": -480, |
| "timezone": "Asia/Shanghai", |
| "conversation_mode": {"kind": "primary_assistant"}, |
| "system_hints": [], |
| "partial_query": {"id": new_uuid(), "author": {"role": "user"}, "content": {"content_type": "text", "parts": [prompt]}}, |
| "supports_buffering": True, |
| "supported_encodings": ["v1"], |
| "client_contextual_info": {"app_name": "chatgpt.com"}, |
| "thinking_effort": EDITABLE_FILE_THINKING_EFFORT, |
| } |
| if attachment_mime_types: |
| payload["attachment_mime_types"] = attachment_mime_types |
| response = self.session.post( |
| self.base_url + path, |
| headers=self._headers(path, {"Accept": "*/*", "Content-Type": "application/json", "X-Conduit-Token": "no-token"}), |
| json=payload, |
| timeout=60, |
| ) |
| ensure_ok(response, path) |
| conduit_token = str(response.json().get("conduit_token") or "") |
| if not conduit_token: |
| raise RuntimeError(f"missing conduit_token: {response.text}") |
| return conduit_token |
|
|
| def _run_editable_conversation(self, prompt: str, uploaded: list[Dict[str, Any]], conduit_token: str) -> str: |
| self._bootstrap() |
| requirements = self._get_chat_requirements() |
| message: Dict[str, Any] = {"id": new_uuid(), "author": {"role": "user"}, "create_time": time.time()} |
| if uploaded: |
| parts = [{ |
| "content_type": "image_asset_pointer", |
| "asset_pointer": f"sediment://{item['file_id']}", |
| "size_bytes": item["file_size"], |
| "width": item["width"], |
| "height": item["height"], |
| } for item in uploaded] |
| parts.append(prompt) |
| message["content"] = {"content_type": "multimodal_text", "parts": parts} |
| message["metadata"] = { |
| "attachments": [{ |
| "id": item["file_id"], |
| "size": item["file_size"], |
| "name": item["file_name"], |
| "mime_type": item["mime_type"], |
| "width": item["width"], |
| "height": item["height"], |
| "source": "library", |
| "library_file_id": item["library_file_id"], |
| "is_big_paste": False, |
| } for item in uploaded], |
| "developer_mode_connector_ids": [], |
| "selected_sources": [], |
| "selected_github_repos": [], |
| "selected_all_github_repos": False, |
| "serialization_metadata": {"custom_symbol_offsets": []}, |
| } |
| else: |
| message["content"] = {"content_type": "text", "parts": [prompt]} |
| path = "/backend-api/f/conversation" |
| response = self.session.post( |
| self.base_url + path, |
| headers=self._image_headers(path, requirements, conduit_token, "text/event-stream"), |
| json={ |
| "action": "next", |
| "messages": [message], |
| "parent_message_id": "client-created-root", |
| "model": EDITABLE_FILE_MODEL, |
| "client_prepare_state": "sent", |
| "timezone_offset_min": -480, |
| "timezone": "Asia/Shanghai", |
| "conversation_mode": {"kind": "primary_assistant"}, |
| "enable_message_followups": True, |
| "system_hints": [], |
| "supports_buffering": True, |
| "supported_encodings": ["v1"], |
| "client_contextual_info": { |
| "is_dark_mode": False, |
| "time_since_loaded": 401, |
| "page_height": 1138, |
| "page_width": 803, |
| "pixel_ratio": 2, |
| "screen_height": 1440, |
| "screen_width": 2560, |
| "app_name": "chatgpt.com", |
| }, |
| "paragen_cot_summary_display_override": "allow", |
| "force_parallel_switch": "auto", |
| "thinking_effort": EDITABLE_FILE_THINKING_EFFORT, |
| }, |
| timeout=300, |
| stream=True, |
| ) |
| ensure_ok(response, path) |
| conversation_id = "" |
| try: |
| for payload in iter_sse_payloads(response): |
| if payload == "[DONE]": |
| break |
| conversation_id = conversation_id or self._find_editable_value(payload, "conversation_id") |
| finally: |
| response.close() |
| if not conversation_id: |
| raise RuntimeError("conversation_id not found in stream") |
| return conversation_id |
|
|
| def _wait_editable_output_artifacts( |
| self, |
| conversation_id: str, |
| primary_label: str, |
| primary_suffixes: tuple[str, ...], |
| primary_mime_types: set[str], |
| primary_mime_keywords: tuple[str, ...], |
| export_file_re: re.Pattern[str], |
| timeout_secs: float, |
| poll_interval_secs: float, |
| ) -> list[EditableFileArtifact]: |
| deadline = time.time() + timeout_secs |
| while time.time() < deadline: |
| try: |
| conversation = self._get_editable_conversation_detail(conversation_id) |
| except UpstreamHTTPError as exc: |
| if exc.status_code in {404, 409, 423, 429, 500, 502, 503, 504}: |
| time.sleep(poll_interval_secs) |
| continue |
| raise |
| targeted = self._pick_editable_target_artifacts( |
| self._extract_editable_artifacts(conversation, export_file_re), |
| primary_suffixes, |
| primary_mime_types, |
| primary_mime_keywords, |
| ) |
| if targeted: |
| return targeted |
| time.sleep(poll_interval_secs) |
| raise RuntimeError(f"timed out waiting for {primary_label}/zip outputs") |
|
|
| def _get_editable_conversation_detail(self, conversation_id: str) -> Dict[str, Any]: |
| path = f"/backend-api/conversation/{conversation_id}" |
| response = self.session.get(self.base_url + path, headers=self._editable_conversation_document_headers(path, conversation_id), timeout=60) |
| ensure_ok(response, path) |
| return response.json() |
|
|
| def _editable_browser_headers(self, path: str, conversation_id: str) -> Dict[str, str]: |
| headers = self._headers(path, {"Accept": "*/*"}) |
| headers["Referer"] = f"{self.base_url}/c/{conversation_id}" |
| return headers |
|
|
| def _editable_conversation_document_headers(self, path: str, conversation_id: str) -> Dict[str, str]: |
| headers = self._editable_browser_headers(path, conversation_id) |
| headers["X-OpenAI-Target-Route"] = "/backend-api/conversation/{conversation_id}" |
| return headers |
|
|
| def _extract_editable_artifacts(self, conversation: Dict[str, Any], export_file_re: re.Pattern[str]) -> list[EditableFileArtifact]: |
| artifacts: dict[str, EditableFileArtifact] = {} |
| for node in sorted((conversation.get("mapping") or {}).values(), key=lambda item: float(((item or {}).get("message") or {}).get("create_time") or 0.0)): |
| message = (node or {}).get("message") or {} |
| message_id = str(message.get("id") or "") |
| author_role = str(((message.get("author") or {}).get("role") or "")).strip() |
| if author_role not in {"assistant", "tool"}: |
| continue |
| create_time = float(message.get("create_time") or 0.0) |
| message_text = self._editable_message_text(message) |
| for artifact in self._extract_editable_message_artifacts(message, message_id, author_role, create_time, export_file_re): |
| key = artifact.attachment_id or artifact.file_id or artifact.name or artifact.sandbox_path |
| if key: |
| artifacts[key] = self._merge_editable_artifact(artifacts.get(key), artifact) |
| for export_path in self._extract_editable_export_paths(message_text, export_file_re): |
| inferred = EditableFileArtifact(name=Path(export_path).name, create_time=create_time, author_role=author_role, sandbox_path=export_path, message_id=message_id) |
| artifacts[export_path] = self._merge_editable_artifact(artifacts.get(export_path), inferred) |
| return sorted(artifacts.values(), key=lambda item: item.create_time) |
|
|
| def _extract_editable_message_artifacts( |
| self, |
| message: Dict[str, Any], |
| message_id: str, |
| author_role: str, |
| create_time: float, |
| export_file_re: re.Pattern[str], |
| ) -> list[EditableFileArtifact]: |
| artifacts: list[EditableFileArtifact] = [] |
| for item in (message.get("metadata") or {}).get("attachments") or []: |
| artifact = self._editable_artifact_from_dict(item, message_id, author_role, create_time, export_file_re) |
| if artifact: |
| artifacts.append(artifact) |
| for obj in self._walk_search_dicts(message): |
| artifact = self._editable_artifact_from_dict(obj, message_id, author_role, create_time, export_file_re) |
| if artifact: |
| artifacts.append(artifact) |
| return artifacts |
|
|
| def _editable_artifact_from_dict( |
| self, |
| payload: Dict[str, Any], |
| message_id: str, |
| author_role: str, |
| create_time: float, |
| export_file_re: re.Pattern[str], |
| ) -> EditableFileArtifact | None: |
| if not ({"id", "file_id", "asset_pointer", "name", "file_name", "filename", "mime_type", "mimeType"} & set(payload.keys())): |
| return None |
| attachment_id = self._match_editable_file_id(str(payload.get("id") or "")) |
| file_id = self._match_editable_file_id(str(payload.get("file_id") or "")) |
| name = self._sanitize_editable_filename(str(payload.get("name") or payload.get("file_name") or payload.get("filename") or payload.get("title") or "").strip()) |
| mime_type = self._clean_editable_mime_type(payload.get("mime_type") or payload.get("mimeType") or "") |
| for asset_id in EDITABLE_ASSET_POINTER_RE.findall(str(payload.get("asset_pointer") or "")): |
| attachment_id = attachment_id or asset_id |
| file_id = file_id or asset_id |
| if not attachment_id or not file_id: |
| ids = self._extract_editable_file_ids(json.dumps(payload, ensure_ascii=False)) |
| attachment_id = attachment_id or (ids[0] if ids else "") |
| file_id = file_id or (ids[0] if ids else "") |
| if not attachment_id and not file_id: |
| return None |
| return EditableFileArtifact( |
| attachment_id=attachment_id, |
| file_id=file_id, |
| name=name, |
| mime_type=mime_type, |
| create_time=create_time, |
| author_role=author_role, |
| sandbox_path=(self._extract_editable_export_paths(payload, export_file_re) or [""])[0], |
| message_id=message_id, |
| ) |
|
|
| def _pick_editable_target_artifacts( |
| self, |
| artifacts: list[EditableFileArtifact], |
| primary_suffixes: tuple[str, ...], |
| primary_mime_types: set[str], |
| primary_mime_keywords: tuple[str, ...], |
| ) -> list[EditableFileArtifact]: |
| primary = next((item for item in reversed(artifacts) if self._looks_like_editable_primary(item, primary_suffixes, primary_mime_types, primary_mime_keywords)), None) |
| zip_item = next((item for item in reversed(artifacts) if self._looks_like_editable_zip(item)), None) |
| return [primary, zip_item] if primary and zip_item else [] |
|
|
| def _download_editable_artifact( |
| self, |
| conversation_id: str, |
| artifact: EditableFileArtifact, |
| output_dir: Path, |
| primary_mime_types: set[str], |
| primary_mime_keywords: tuple[str, ...], |
| primary_default_extension: str, |
| ) -> Path: |
| download_url = self._resolve_editable_download_url(conversation_id, artifact) |
| if not download_url: |
| raise RuntimeError(f"download url not found for artifact: {artifact}") |
| response = self.session.get(download_url, timeout=300) |
| ensure_ok(response, "artifact_download") |
| content_type = self._clean_editable_mime_type(response.headers.get("Content-Type") or artifact.mime_type) |
| file_name = self._resolve_editable_output_name(artifact, response.url, response.headers.get("Content-Disposition"), content_type, primary_mime_types, primary_mime_keywords, primary_default_extension) |
| target_path = self._unique_editable_path(output_dir / file_name) |
| target_path.write_bytes(response.content) |
| return target_path |
|
|
| def download_conversation_artifact( |
| self, |
| conversation_id: str, |
| message_id: str, |
| sandbox_path: str, |
| output_dir: Path, |
| ) -> Path: |
| """Download a generic file produced by ChatGPT's interpreter. |
| |
| This deliberately reuses the same resolver used by PPT/PSD exports: |
| interpreter download first, followed by attachment/file-id fallbacks. |
| Only files explicitly referenced from ``/mnt/data`` are accepted. |
| """ |
| normalized_path = unquote(str(sandbox_path or "").strip()) |
| if not normalized_path.startswith("/mnt/data/"): |
| raise ValueError("invalid sandbox artifact path") |
| output_dir.mkdir(parents=True, exist_ok=True) |
| artifact = EditableFileArtifact( |
| name=self._sanitize_editable_filename(Path(normalized_path).name), |
| sandbox_path=normalized_path, |
| message_id=str(message_id or "").strip(), |
| author_role="assistant", |
| ) |
| return self.download_conversation_artifact_record(conversation_id, artifact, output_dir) |
|
|
| def list_conversation_artifacts( |
| self, |
| conversation_id: str, |
| min_create_time: float = 0.0, |
| ) -> list[EditableFileArtifact]: |
| """Return assistant/tool artifacts from a conversation document.""" |
| conversation = self._get_editable_conversation_detail(conversation_id) |
| artifacts = self._extract_editable_artifacts(conversation, GENERIC_SANDBOX_EXPORT_FILE_RE) |
| return [item for item in artifacts if item.create_time >= float(min_create_time or 0.0)] |
|
|
| def download_conversation_artifact_record( |
| self, |
| conversation_id: str, |
| artifact: EditableFileArtifact, |
| output_dir: Path, |
| ) -> Path: |
| output_dir.mkdir(parents=True, exist_ok=True) |
| return self._download_editable_artifact( |
| conversation_id, |
| artifact, |
| output_dir, |
| set(), |
| (), |
| ".bin", |
| ) |
|
|
| def _resolve_editable_download_url(self, conversation_id: str, artifact: EditableFileArtifact) -> str: |
| ids: list[str] = [] |
| for item in (artifact.attachment_id, artifact.file_id): |
| if item and item not in ids: |
| ids.append(item) |
| if artifact.sandbox_path and artifact.message_id: |
| path = f"/backend-api/conversation/{conversation_id}/interpreter/download" |
| response = self.session.get( |
| self.base_url + path, |
| headers=self._editable_download_headers(path, conversation_id, "/backend-api/conversation/{conversation_id}/interpreter/download"), |
| params={"message_id": artifact.message_id, "sandbox_path": artifact.sandbox_path}, |
| timeout=60, |
| ) |
| if 200 <= response.status_code < 300: |
| url = self._download_url_from_response(response) |
| if url: |
| return url |
| for attachment_id in ids: |
| path = f"/backend-api/conversation/{conversation_id}/attachment/{attachment_id}/download" |
| response = self.session.get( |
| self.base_url + path, |
| headers=self._editable_download_headers(path, conversation_id, "/backend-api/conversation/{conversation_id}/attachment/{attachment_id}/download"), |
| timeout=60, |
| ) |
| if 200 <= response.status_code < 300: |
| url = self._download_url_from_response(response) |
| if url: |
| return url |
| for file_id in ids: |
| path = f"/backend-api/files/download/{file_id}" |
| response = self.session.get( |
| self.base_url + path, |
| headers=self._editable_download_headers(path, conversation_id, "/backend-api/files/download/{file_id}"), |
| params={"post_id": "", "inline": "false"}, |
| timeout=60, |
| ) |
| if 200 <= response.status_code < 300: |
| url = self._download_url_from_response(response) |
| if url: |
| return url |
| for file_id in ids: |
| path = f"/backend-api/files/{file_id}/download" |
| response = self.session.get( |
| self.base_url + path, |
| headers=self._editable_download_headers(path, conversation_id, "/backend-api/files/download/{file_id}"), |
| timeout=60, |
| ) |
| if 200 <= response.status_code < 300: |
| url = self._download_url_from_response(response) |
| if url: |
| return url |
| return "" |
|
|
| def _editable_download_headers(self, path: str, conversation_id: str, route: str) -> Dict[str, str]: |
| headers = self._editable_browser_headers(path, conversation_id) |
| headers["X-OpenAI-Target-Route"] = route |
| return headers |
|
|
| @staticmethod |
| def _download_url_from_response(response: Any) -> str: |
| try: |
| payload = response.json() |
| except Exception: |
| payload = {} |
| return str(payload.get("download_url") or payload.get("url") or "") |
|
|
| def _resolve_editable_output_name( |
| self, |
| artifact: EditableFileArtifact, |
| final_url: str, |
| content_disposition: str | None, |
| content_type: str, |
| primary_mime_types: set[str], |
| primary_mime_keywords: tuple[str, ...], |
| primary_default_extension: str, |
| ) -> str: |
| file_name = self._sanitize_editable_filename(artifact.name) |
| if not file_name and artifact.sandbox_path: |
| file_name = self._sanitize_editable_filename(Path(artifact.sandbox_path).name) |
| if not file_name: |
| file_name = self._sanitize_editable_filename(self._editable_filename_from_content_disposition(content_disposition or "")) |
| if not file_name: |
| file_name = self._sanitize_editable_filename(Path(urlparse(final_url).path).name) |
| extension = self._editable_extension_from_mime_type(content_type, primary_mime_types, primary_mime_keywords, primary_default_extension) |
| return f"artifact{extension}" if not file_name else (file_name if Path(file_name).suffix else file_name + extension) |
|
|
| def _find_editable_value(self, payload: Any, key: str) -> str: |
| if isinstance(payload, str): |
| match = SEARCH_CONVERSATION_ID_RE.search(payload) if key == "conversation_id" else None |
| if match: |
| return match.group(1) |
| try: |
| payload = json.loads(payload) |
| except json.JSONDecodeError: |
| return "" |
| if isinstance(payload, dict): |
| value = payload.get(key) |
| if isinstance(value, str) and value: |
| return value |
| return next((found for item in payload.values() if (found := self._find_editable_value(item, key))), "") |
| if isinstance(payload, list): |
| return next((found for item in payload if (found := self._find_editable_value(item, key))), "") |
| return "" |
|
|
| def _extract_editable_file_ids(self, text: str) -> list[str]: |
| values: list[str] = [] |
| for item in EDITABLE_ASSET_POINTER_RE.findall(text): |
| if item not in values: |
| values.append(item) |
| for item in FILE_ID_RE.findall(text): |
| if item not in values: |
| values.append(item) |
| return values |
|
|
| @staticmethod |
| def _match_editable_file_id(value: str) -> str: |
| match = FILE_ID_RE.search(value) |
| return match.group(1) if match else "" |
|
|
| @staticmethod |
| def _clean_editable_mime_type(value: Any) -> str: |
| text = str(value or "").strip().lower() |
| return text.split(";", 1)[0] if "/" in text else "" |
|
|
| def _looks_like_editable_primary( |
| self, |
| artifact: EditableFileArtifact, |
| primary_suffixes: tuple[str, ...], |
| primary_mime_types: set[str], |
| primary_mime_keywords: tuple[str, ...], |
| ) -> bool: |
| path, name, mime = artifact.sandbox_path.lower(), artifact.name.lower(), artifact.mime_type |
| return name.endswith(primary_suffixes) or path.endswith(primary_suffixes) or mime in primary_mime_types or any(keyword in mime for keyword in primary_mime_keywords) |
|
|
| @staticmethod |
| def _looks_like_editable_zip(artifact: EditableFileArtifact) -> bool: |
| path, name, mime = artifact.sandbox_path.lower(), artifact.name.lower(), artifact.mime_type |
| return name.endswith(".zip") or path.endswith(".zip") or mime in EDITABLE_ZIP_MIME_TYPES or mime.endswith("/zip") |
|
|
| @staticmethod |
| def _editable_extension_from_mime_type( |
| mime_type: str, |
| primary_mime_types: set[str], |
| primary_mime_keywords: tuple[str, ...], |
| primary_default_extension: str, |
| ) -> str: |
| if mime_type in primary_mime_types or any(keyword in mime_type for keyword in primary_mime_keywords): |
| return primary_default_extension |
| if mime_type in EDITABLE_ZIP_MIME_TYPES or mime_type.endswith("/zip"): |
| return ".zip" |
| return mimetypes.guess_extension(mime_type) or "" |
|
|
| @staticmethod |
| def _editable_filename_from_content_disposition(content_disposition: str) -> str: |
| extended_match = re.search(r"filename\*=UTF-8''([^;]+)", content_disposition, re.IGNORECASE) |
| if extended_match: |
| return unquote(extended_match.group(1)).strip() |
| plain_match = re.search(r'filename="([^"]+)"', content_disposition, re.IGNORECASE) |
| return plain_match.group(1).strip() if plain_match else "" |
|
|
| @staticmethod |
| def _sanitize_editable_filename(value: str) -> str: |
| return Path(str(value or "").strip()).name.replace("\x00", "").strip() |
|
|
| @staticmethod |
| def _unique_editable_path(path: Path) -> Path: |
| if not path.exists(): |
| return path |
| for index in range(1, 1000): |
| candidate = path.with_name(f"{path.stem}_{index}{path.suffix}") |
| if not candidate.exists(): |
| return candidate |
| raise RuntimeError(f"failed to allocate output path for {path}") |
|
|
| @staticmethod |
| def _merge_editable_artifact(current: EditableFileArtifact | None, latest: EditableFileArtifact) -> EditableFileArtifact: |
| if current is None: |
| return latest |
| return EditableFileArtifact( |
| attachment_id=latest.attachment_id or current.attachment_id, |
| file_id=latest.file_id or current.file_id, |
| name=latest.name or current.name, |
| mime_type=latest.mime_type or current.mime_type, |
| create_time=max(current.create_time, latest.create_time), |
| author_role=latest.author_role or current.author_role, |
| sandbox_path=latest.sandbox_path or current.sandbox_path, |
| message_id=latest.message_id or current.message_id, |
| ) |
|
|
| @staticmethod |
| def _editable_message_text(message: Any) -> str: |
| if not isinstance(message, dict): |
| return "" |
| content = message.get("content") or {} |
| parts: list[str] = [] |
| if isinstance(content, dict): |
| if isinstance(content.get("text"), str): |
| parts.append(content["text"]) |
| for part in content.get("parts") or []: |
| if isinstance(part, str): |
| parts.append(part) |
| elif isinstance(part, dict): |
| parts.extend(str(part.get(key) or "") for key in ("text", "asset_pointer", "model_set_context") if part.get(key)) |
| if isinstance(message.get("content"), str): |
| parts.append(str(message["content"])) |
| return "\n".join(part for part in parts if part).strip() |
|
|
| @staticmethod |
| def _extract_editable_export_paths(payload: Any, export_file_re: re.Pattern[str]) -> list[str]: |
| if isinstance(payload, str): |
| text = payload |
| else: |
| try: |
| text = json.dumps(payload, ensure_ascii=False) |
| except Exception: |
| text = str(payload) |
| values: list[str] = [] |
| for item in export_file_re.findall(text): |
| path = str(item or "").strip() |
| if path and path not in values: |
| values.append(path) |
| return values |
|
|
| @staticmethod |
| def _find_content_policy_error_in_conversation(data: Dict[str, Any]) -> str: |
| """从对话文档中查找内容政策违规错误消息。 |
| |
| 上游拒绝生成图片时，错误消息会出现在 assistant 消息的文本中。 |
| 本方法遍历所有 assistant/tool 消息，检查是否包含内容政策违规关键词， |
| 如果匹配则返回该消息文本（截断至 500 字符），否则返回空字符串。 |
| """ |
| mapping = data.get("mapping") or {} |
| for node in mapping.values(): |
| message = (node or {}).get("message") or {} |
| author = message.get("author") or {} |
| role = str(author.get("role") or "").strip().lower() |
| if role not in {"assistant", "tool"}: |
| continue |
| content = message.get("content") or {} |
| |
| text_parts: list[str] = [] |
| if isinstance(content, dict): |
| msg_parts = content.get("parts") or [] |
| if isinstance(msg_parts, list): |
| for part in msg_parts: |
| if isinstance(part, str) and part.strip(): |
| text_parts.append(part.strip()) |
| text_field = str(content.get("text") or "") |
| if text_field.strip(): |
| text_parts.append(text_field.strip()) |
| elif isinstance(content, str) and content.strip(): |
| text_parts.append(content.strip()) |
| msg_text = "\n".join(text_parts) |
| if msg_text and _is_content_policy_error(msg_text): |
| return msg_text[:500] |
| return "" |
|
|
| def _query_backend_tasks( |
| self, |
| conversation_id: str = "", |
| task_id: str = "", |
| timeout_secs: float = 30.0, |
| ) -> list[Dict[str, Any]]: |
| """查询 /backend-api/tasks/ 接口获取异步任务状态和错误信息。 |
| |
| 参数： |
| - `conversation_id`：可选。按 conversation_id 过滤任务。 |
| - `task_id`：可选。按 task_id 过滤任务。 |
| - `timeout_secs`：请求超时秒数。 |
| |
| 返回： |
| - 任务列表，每个任务包含 image_gen_message 等字段。 |
| """ |
| path = "/backend-api/tasks" |
| response = self.session.get( |
| self.base_url + path, |
| headers=self._headers(path, {"Accept": "application/json"}), |
| timeout=timeout_secs, |
| ) |
| ensure_ok(response, path) |
| data = response.json() |
| tasks = data.get("tasks", []) |
| if not isinstance(tasks, list): |
| return [] |
|
|
| |
| if conversation_id: |
| tasks = [ |
| t for t in tasks |
| if isinstance(t, dict) and ( |
| t.get("conversation_id") == conversation_id |
| or t.get("original_conversation_id") == conversation_id |
| ) |
| ] |
| if task_id: |
| tasks = [t for t in tasks if isinstance(t, dict) and t.get("task_id") == task_id] |
| return tasks |
|
|
| def check_task_error(self, task: Dict[str, Any]) -> tuple[bool, str, Dict[str, Any]]: |
| """检查单个任务是否包含结构化错误。 |
| |
| 通过以下字段判断（不依赖文本匹配）： |
| - image_gen_message.metadata.is_error == True |
| - image_gen_message.author.role == "assistant" (而非 "tool") |
| - image_gen_message.content.content_type == "text" (而非 "multimodal_text") |
| |
| 返回： |
| - (is_error, error_msg, metadata) |
| """ |
| img_msg = task.get("image_gen_message") or {} |
| if not img_msg: |
| return False, "", {} |
|
|
| metadata = img_msg.get("metadata") or {} |
| content = img_msg.get("content") or {} |
| author = img_msg.get("author") or {} |
|
|
| is_error = metadata.get("is_error", False) |
| is_text_only = content.get("content_type") == "text" |
| is_assistant_role = author.get("role") == "assistant" |
|
|
| |
| error_msg = "" |
| if is_error and is_text_only: |
| parts = content.get("parts", []) |
| error_msg = "".join(p for p in parts if isinstance(p, str)) |
|
|
| return is_error, error_msg, metadata |
|
|
| def _bootstrap(self) -> None: |
| """预热首页，并提取 PoW 相关脚本引用。""" |
| response = self.session.get( |
| self.base_url + "/", |
| headers=self._bootstrap_headers(), |
| timeout=30, |
| ) |
| ensure_ok(response, "bootstrap") |
| self.pow_script_sources, self.pow_data_build = parse_pow_resources(response.text) |
| if not self.pow_script_sources: |
| self.pow_script_sources = [DEFAULT_POW_SCRIPT] |
|
|
| def _get_chat_requirements(self) -> ChatRequirements: |
| """获取当前模式对话所需的 sentinel token（prepare + finalize 两步流程）。 |
| |
| 2026-10-14 起 chatgpt.com 把 chat-requirements 拆成 prepare/finalize 两步：prepare |
| 先返回 challenge 参数（proofofwork / turnstile 等），必须先在真实页面里解出 Turnstile， |
| 再 finalize 换回真正可用的 token，否则对话接口会返回伪装成 404 的 "Model not found"。 |
| """ |
| base = "/backend-api/sentinel/chat-requirements" if self.access_token else "/backend-anon/sentinel/chat-requirements" |
| sentinel_user_agent = self._sentinel_user_agent() |
| p_token = build_legacy_requirements_token(sentinel_user_agent, self.pow_script_sources, self.pow_data_build) |
|
|
| prepare_path = base + "/prepare" |
| response = self.session.post( |
| self.base_url + prepare_path, |
| headers=self._headers(prepare_path, {"Content-Type": "application/json"}), |
| json={"p": p_token}, |
| timeout=30, |
| ) |
| ensure_ok(response, "chat_requirements_prepare") |
| prepare_data = response.json() |
|
|
| prepare_token = prepare_data.get("prepare_token", "") |
| req = self._build_requirements(prepare_data, p_token) |
| proof_token = req.proof_token |
| turnstile_token = req.turnstile_token |
| |
| so_token = req.so_token |
| if not so_token: |
| so_info = prepare_data.get("so") or {} |
| if so_info.get("required") and so_info.get("collector_dx") and so_info.get("snapshot_dx"): |
| so_token = solve_so(so_info["collector_dx"], so_info["snapshot_dx"], p_token) |
|
|
| finalize_path = base + "/finalize" |
| finalize_payload = { |
| "prepare_token": prepare_token, |
| "proofofwork": proof_token, |
| "turnstile": turnstile_token, |
| } |
| if so_token: |
| finalize_payload["so"] = so_token |
| response = self.session.post( |
| self.base_url + finalize_path, |
| headers=self._headers(finalize_path, {"Content-Type": "application/json"}), |
| json=finalize_payload, |
| timeout=30, |
| ) |
| ensure_ok(response, "chat_requirements_finalize") |
| data = response.json() |
|
|
| token = data.get("token", "") |
| if not token: |
| message = "missing auth chat requirements token" if self.access_token else "missing chat requirements token" |
| raise RuntimeError(f"{message}: {data}") |
|
|
| return ChatRequirements( |
| token=token, |
| proof_token=proof_token, |
| turnstile_token=turnstile_token, |
| so_token=data.get("so_token", "") or so_token, |
| raw_finalize=data, |
| ) |
|
|
| def _chat_target(self) -> tuple[str, str]: |
| if self.access_token: |
| return "/backend-api/conversation", "Asia/Shanghai" |
| return "/backend-anon/conversation", "America/Los_Angeles" |
|
|
| def list_models(self) -> Dict[str, Any]: |
| """返回当前模式下可用模型，格式对齐 OpenAI `/v1/models`。""" |
| self._bootstrap() |
| path = "/backend-api/models?history_and_training_disabled=false" if self.access_token else ( |
| "/backend-anon/models?iim=false&is_gizmo=false" |
| ) |
| route = "/backend-api/models" if self.access_token else "/backend-anon/models" |
| context = "auth_models" if self.access_token else "anon_models" |
| response = self.session.get( |
| self.base_url + path, |
| headers=self._headers(route), |
| timeout=30, |
| ) |
| ensure_ok(response, context) |
| data = [] |
| seen = set() |
| for item in response.json().get("models", []): |
| if not isinstance(item, dict): |
| continue |
| slug = str(item.get("slug", "")).strip() |
| if not slug or slug in seen: |
| continue |
| seen.add(slug) |
| entry: Dict[str, Any] = { |
| "id": slug, |
| "object": "model", |
| "created": int(item.get("created") or 0), |
| "owned_by": str(item.get("owned_by") or "chatgpt"), |
| "permission": [], |
| "root": slug, |
| "parent": None, |
| } |
| |
| |
| |
| title = str(item.get("title") or "").strip() |
| if title: |
| entry["title"] = title |
| if "is_work_mode_model" in item: |
| entry["is_work_mode_model"] = bool(item.get("is_work_mode_model")) |
| reasoning_type = str(item.get("reasoning_type") or "").strip() |
| if reasoning_type: |
| entry["reasoning_type"] = reasoning_type |
| enabled_tools = item.get("enabled_tools") |
| if isinstance(enabled_tools, list): |
| entry["enabled_tools"] = [str(tool) for tool in enabled_tools if str(tool).strip()] |
| data.append(entry) |
| data.sort(key=lambda item: item["id"]) |
| return {"object": "list", "data": data} |
|
|
