le-gpt-image v1.1：curl_cffi 可选传输层（Chrome TLS 指纹）修复 Cloudflare 质询——chatgpt.com 按指纹放行浏览器/Codex App、拦 urllib/curl；curl_cffi 路径不覆盖 Chrome UA（矛盾组合同样被掐）；排障表改写（此前「切节点」结论证伪）；实测功夫女足海报 61.5s 成功

skills/le-gpt-image/scripts/gpt_image.py

@@ -42,6 +42,16 @@ try:
 except ImportError:  # Python < 3.11
     tomllib = None  # type: ignore[assignment]
 
+# 可选传输层：curl_cffi 以 Chrome TLS 指纹发请求。
+# chatgpt.com 的 Cloudflare 会质询 python-urllib/curl 的握手指纹（403 cf-mitigated: challenge），
+# Chrome 指纹（及 Codex/ChatGPT.app 的网络栈）不受影响。未安装时回退 urllib。
+try:
+    from curl_cffi import requests as _cffi_requests
+
+    _HAS_CFFI = True
+except ImportError:
+    _HAS_CFFI = False
+
 __version__ = "1.0.0"
 
 DEFAULT_BASE_URL = "https://chatgpt.com/backend-api/codex"
@@ -248,19 +258,52 @@ def _refresh_lock(auth_file: Path, timeout: float):
             fcntl.flock(lock.fileno(), fcntl.LOCK_UN)
 
 
+def _browser_headers(headers: dict[str, str]) -> dict[str, str]:
+    """curl_cffi 路径的请求头：剥掉会破坏 Chrome 模拟的自定义 UA/version。
+
+    实测「Chrome TLS 指纹 + codex_cli_rs UA」的矛盾组合会被边缘直接掐断连接；
+    纯浏览器头（Authorization/originator/ChatGPT-Account-ID）可完整走通生图。
+    """
+    trimmed = dict(headers)
+    trimmed.pop("User-Agent", None)
+    trimmed.pop("version", None)
+    return trimmed
+
+
 def _post_json(
     url: str, *, headers: dict[str, str], payload: dict[str, Any], timeout: float
 ) -> dict[str, Any]:
-    body = json.dumps(payload).encode("utf-8")
-    req = request.Request(url, data=body, headers=headers, method="POST")
-    try:
-        with request.urlopen(req, timeout=timeout) as response:
-            raw = response.read().decode("utf-8")
-    except error.HTTPError as exc:
-        raw = exc.read().decode("utf-8", errors="replace")
-        raise HttpError(exc.code, raw, exc.headers.get("Retry-After") if exc.headers else None) from exc
-    except (error.URLError, OSError) as exc:
-        raise CliError(f"网络请求失败: {exc}") from exc
+    if _HAS_CFFI:
+        try:
+            response = _cffi_requests.post(
+                url,
+                json=payload,
+                headers=_browser_headers(headers),
+                timeout=timeout,
+                impersonate="chrome",
+            )
+        except Exception as exc:  # curl_cffi 的各类传输异常
+            raise CliError(f"网络请求失败: {exc}") from exc
+        if response.status_code >= 400:
+            raise HttpError(
+                response.status_code,
+                response.text or "",
+                response.headers.get("Retry-After"),
+            )
+        raw = response.text
+    else:
+        body = json.dumps(payload).encode("utf-8")
+        req = request.Request(url, data=body, headers=headers, method="POST")
+        try:
+            with request.urlopen(req, timeout=timeout) as response:
+                raw = response.read().decode("utf-8")
+        except error.HTTPError as exc:
+            raw = exc.read().decode("utf-8", errors="replace")
+            raise HttpError(
+                exc.code, raw, exc.headers.get("Retry-After") if exc.headers else None
+            ) from exc
+        except (error.URLError, OSError) as exc:
+            raise CliError(f"网络请求失败: {exc}") from exc
     if not raw:
         return {}
     try:
@@ -474,6 +517,38 @@ def _native_result(
 
 def _sse_data_lines(url: str, headers: dict[str, str], payload: dict[str, Any], timeout: float):
     req_headers = {**headers, "Accept": "text/event-stream", "Content-Type": "application/json"}
+    if _HAS_CFFI:
+        req_headers = _browser_headers(req_headers)
+        try:
+            response = _cffi_requests.post(
+                url,
+                json=payload,
+                headers=req_headers,
+                timeout=timeout,
+                impersonate="chrome",
+                stream=True,
+            )
+            if response.status_code >= 400:
+                raise HttpError(
+                    response.status_code,
+                    response.text or "",
+                    response.headers.get("Retry-After"),
+                )
+            for raw_line in response.iter_lines():
+                if isinstance(raw_line, bytes):
+                    line = raw_line.decode("utf-8", "replace")
+                else:
+                    line = str(raw_line)
+                stripped = line.strip()
+                if stripped.startswith("data:"):
+                    data = stripped[5:].strip()
+                    if data and data != "[DONE]":
+                        yield data
+        except HttpError:
+            raise
+        except Exception as exc:
+            raise CliError(f"网络请求失败: {exc}") from exc
+        return
     req = request.Request(
         url, data=json.dumps(payload).encode("utf-8"), headers=req_headers, method="POST"
     )
@@ -849,6 +924,12 @@ def build_parser() -> argparse.ArgumentParser:
 
 def main(argv: Sequence[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
+    if not getattr(args, "dry_run", False) and args.func is not _cmd_check:
+        _log(
+            "传输层: curl_cffi (Chrome 指纹)"
+            if _HAS_CFFI
+            else "传输层: urllib（若被 Cloudflare 质询，pip install curl_cffi 后重试）"
+        )
     try:
         return int(args.func(args))
     except CliError as exc: