feat(api): 集成ChatGPT服务并支持图像生成功能

- 添加ChatGPTService类用于处理图像生成请求
- 实现图像生成相关API端点(/v1/images/generations, /v1/chat/completions, /v1/responses)
- 实现图像生成结果的处理和返回
- 添加图像服务模块处理底层API交互
- 新增测试文件验证图像生成功能

README.md

@@ -6,7 +6,10 @@ ChatGPT 图片生成代理与账号池管理面板，提供账号维护、额度
 
 ## 功能
 
-- 批量导入和管理 `access_token`
+- 兼容 OpenAI `Chat Completions` 图片响应
+- 兼容 OpenAI `Responses API` 图片生成接口
+- 支持导入 CPA 格式文件
+- 支持多种方式导入 `access_token`
 - 自动刷新账号邮箱、类型、图片额度、恢复时间
 - 轮询可用账号进行图片生成
 - 失效 Token 自动剔除
@@ -15,9 +18,15 @@ ChatGPT 图片生成代理与账号池管理面板，提供账号维护、额度
 > 目前仅实现了生图效果，编辑图片以及gpt-image-2模型尚未实现，需要后续更新。
 
 生图界面：
+
 ![image](assets/image.png)
 
+Chery Studio 中使用：
+
+![image](assets/chery_studio.png)
+
 号池管理：
+
 ![image](assets/account_pool.png)
 
 ## 接口
@@ -34,6 +43,14 @@ Authorization: Bearer <auth-key>
 POST /v1/images/generations
 ```
 
+```http
+POST /v1/chat/completions
+```
+
+```http
+POST /v1/responses
+```
+
 请求体示例：
 
 ```json

services/account_service.py

@@ -221,22 +221,59 @@ class AccountService:
         with self._lock:
             return [token for item in self._accounts if (token := self._clean_token(item.get("access_token")))]
 
-    def next_token(self, excluded_tokens: set[str] | None = None) -> str:
+    def _list_available_candidate_tokens(self, excluded_tokens: set[str] | None = None) -> list[str]:
+        excluded = {self._clean_token(token) for token in (excluded_tokens or set()) if self._clean_token(token)}
+        return [
+            token
+            for item in self._accounts
+            if self._is_image_account_available(item)
+            and (token := self._clean_token(item.get("access_token")))
+            and token not in excluded
+        ]
+
+    def _pick_next_candidate_token(self, excluded_tokens: set[str] | None = None) -> str:
         with self._lock:
-            excluded = {self._clean_token(token) for token in (excluded_tokens or set()) if self._clean_token(token)}
-            tokens = [
-                token
-                for item in self._accounts
-                if self._is_image_account_available(item)
-                and (token := self._clean_token(item.get("access_token")))
-                and token not in excluded
-            ]
+            tokens = self._list_available_candidate_tokens(excluded_tokens)
             if not tokens:
                 raise RuntimeError(f"No available tokens found in {self.store_file}")
             access_token = tokens[self._index % len(tokens)]
             self._index += 1
             return access_token
 
+    def refresh_account_state(self, access_token: str) -> dict | None:
+        try:
+            remote_info = self.fetch_remote_info(access_token)
+        except Exception as exc:
+            message = str(exc)
+            print(f"[account-available] refresh token={access_token[:12]}... fail {message}")
+            if "/backend-api/me failed: HTTP 401" in message:
+                return self.update_account(
+                    access_token,
+                    {
+                        "status": "异常",
+                        "quota": 0,
+                    },
+                )
+            return None
+        return self.update_account(access_token, remote_info)
+
+    def get_available_access_token(self) -> str:
+        attempted_tokens: set[str] = set()
+        while True:
+            access_token = self._pick_next_candidate_token(excluded_tokens=attempted_tokens)
+            attempted_tokens.add(access_token)
+            account = self.refresh_account_state(access_token)
+            if self._is_image_account_available(account or {}):
+                return access_token
+            print(
+                f"[account-available] skip token={access_token[:12]}... "
+                f"quota={account.get('quota') if account else 'unknown'} "
+                f"status={account.get('status') if account else 'unknown'}"
+            )
+
+    def next_token(self) -> str:
+        return self.get_available_access_token()
+
     def get_account(self, access_token: str) -> dict | None:
         access_token = self._clean_token(access_token)
         if not access_token:

services/api.py

@@ -8,16 +8,14 @@ from fastapi import APIRouter, FastAPI, Header, HTTPException
 from fastapi.concurrency import run_in_threadpool
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import FileResponse
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 from services.account_service import account_service
+from services.chatgpt_service import ChatGPTService
 from services.config import config
-from services.backend_service import BackendService
-from services.chat_image_service import ChatImageService
 from services.image_service import ImageGenerationError
 from services.version import get_app_version
 
-
 BASE_DIR = Path(__file__).resolve().parents[1]
 WEB_DIST_DIR = BASE_DIR / "web_dist"
 
@@ -49,6 +47,27 @@ class AccountUpdateRequest(BaseModel):
     quota: int | None = None
 
 
+class ChatCompletionRequest(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    model: str | None = None
+    prompt: str | None = None
+    n: int | None = None
+    stream: bool | None = None
+    modalities: list[str] | None = None
+    messages: list[dict[str, object]] | None = None
+
+
+class ResponseCreateRequest(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    model: str | None = None
+    input: object | None = None
+    tools: list[dict[str, object]] | None = None
+    tool_choice: object | None = None
+    stream: bool | None = None
+
+
 def build_model_item(model_id: str) -> dict[str, object]:
     return {
         "id": model_id,
@@ -98,11 +117,7 @@ def resolve_web_asset(requested_path: str) -> Path | None:
         candidates = [WEB_DIST_DIR / "index.html"]
     else:
         relative_path = Path(clean_path)
-        candidates = [
-            WEB_DIST_DIR / relative_path,
-            WEB_DIST_DIR / relative_path / "index.html",
-            WEB_DIST_DIR / f"{clean_path}.html",
-        ]
+        candidates = [WEB_DIST_DIR / relative_path, WEB_DIST_DIR / relative_path / "index.html", WEB_DIST_DIR / f"{clean_path}.html"]
 
     for candidate in candidates:
         try:
@@ -116,8 +131,7 @@ def resolve_web_asset(requested_path: str) -> Path | None:
 
 
 def create_app() -> FastAPI:
-    service = BackendService(account_service)
-    chat_image_service = ChatImageService(service)
+    chatgpt_service = ChatGPTService(account_service)
     app_version = get_app_version()
 
     @asynccontextmanager
@@ -142,13 +156,7 @@ def create_app() -> FastAPI:
 
     @router.get("/v1/models")
     async def list_models():
-        return {
-            "object": "list",
-            "data": [
-                build_model_item("gpt-image-1"),
-                build_model_item("gpt-image-2"),
-            ],
-        }
+        return {"object": "list", "data": [build_model_item("gpt-image-1"), build_model_item("gpt-image-2")]}
 
     @router.post("/auth/login")
     async def login(authorization: str | None = Header(default=None)):
@@ -165,28 +173,17 @@ def create_app() -> FastAPI:
         return {"items": account_service.list_accounts()}
 
     @router.post("/api/accounts")
-    async def create_accounts(
-            body: AccountCreateRequest,
-            authorization: str | None = Header(default=None),
-    ):
+    async def create_accounts(body: AccountCreateRequest, authorization: str | None = Header(default=None)):
         require_auth_key(authorization)
         tokens = [str(token or "").strip() for token in body.tokens if str(token or "").strip()]
         if not tokens:
             raise HTTPException(status_code=400, detail={"error": "tokens is required"})
         result = account_service.add_accounts(tokens)
         refresh_result = account_service.refresh_accounts(tokens)
-        return {
-            **result,
-            "refreshed": refresh_result.get("refreshed", 0),
-            "errors": refresh_result.get("errors", []),
-            "items": refresh_result.get("items", result.get("items", [])),
-        }
+        return {**result, "refreshed": refresh_result.get("refreshed", 0), "errors": refresh_result.get("errors", []), "items": refresh_result.get("items", result.get("items", []))}
 
     @router.delete("/api/accounts")
-    async def delete_accounts(
-            body: AccountDeleteRequest,
-            authorization: str | None = Header(default=None),
-    ):
+    async def delete_accounts(body: AccountDeleteRequest, authorization: str | None = Header(default=None)):
         require_auth_key(authorization)
         tokens = [str(token or "").strip() for token in body.tokens if str(token or "").strip()]
         if not tokens:
@@ -194,10 +191,7 @@ def create_app() -> FastAPI:
         return account_service.delete_accounts(tokens)
 
     @router.post("/api/accounts/refresh")
-    async def refresh_accounts(
-            body: AccountRefreshRequest,
-            authorization: str | None = Header(default=None),
-    ):
+    async def refresh_accounts(body: AccountRefreshRequest, authorization: str | None = Header(default=None)):
         require_auth_key(authorization)
         access_tokens = [str(token or "").strip() for token in body.access_tokens if str(token or "").strip()]
         if not access_tokens:
@@ -207,24 +201,13 @@ def create_app() -> FastAPI:
         return account_service.refresh_accounts(access_tokens)
 
     @router.post("/api/accounts/update")
-    async def update_account(
-            body: AccountUpdateRequest,
-            authorization: str | None = Header(default=None),
-    ):
+    async def update_account(body: AccountUpdateRequest, authorization: str | None = Header(default=None)):
         require_auth_key(authorization)
         access_token = str(body.access_token or "").strip()
         if not access_token:
             raise HTTPException(status_code=400, detail={"error": "access_token is required"})
 
-        updates = {
-            key: value
-            for key, value in {
-                "type": body.type,
-                "status": body.status,
-                "quota": body.quota,
-            }.items()
-            if value is not None
-        }
+        updates = {key: value for key, value in {"type": body.type, "status": body.status, "quota": body.quota}.items() if value is not None}
         if not updates:
             raise HTTPException(status_code=400, detail={"error": "no updates provided"})
 
@@ -234,28 +217,22 @@ def create_app() -> FastAPI:
         return {"item": account, "items": account_service.list_accounts()}
 
     @router.post("/v1/images/generations")
-    async def generate_images(
-            body: ImageGenerationRequest,
-            authorization: str | None = Header(default=None),
-    ):
+    async def generate_images(body: ImageGenerationRequest, authorization: str | None = Header(default=None)):
         require_auth_key(authorization)
         try:
-            return await run_in_threadpool(
-                service.generate_with_pool,
-                body.prompt,
-                body.model,
-                body.n,
-            )
+            return await run_in_threadpool(chatgpt_service.generate_with_pool, body.prompt, body.model, body.n)
         except ImageGenerationError as exc:
             raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc
 
     @router.post("/v1/chat/completions")
-    async def create_chat_completion(
-            body: dict[str, object],
-            authorization: str | None = Header(default=None),
-    ):
+    async def create_chat_completion(body: ChatCompletionRequest, authorization: str | None = Header(default=None)):
+        require_auth_key(authorization)
+        return await run_in_threadpool(chatgpt_service.create_image_completion, body.model_dump(mode="python"))
+
+    @router.post("/v1/responses")
+    async def create_response(body: ResponseCreateRequest, authorization: str | None = Header(default=None)):
         require_auth_key(authorization)
-        return await run_in_threadpool(chat_image_service.create_image_completion, body)
+        return await run_in_threadpool(chatgpt_service.create_response, body.model_dump(mode="python"))
 
     app.include_router(router)
 

services/backend_service.py

@@ -1,82 +0,0 @@
-from __future__ import annotations
-
-from fastapi import HTTPException
-
-from services.account_service import AccountService
-from services.image_service import ImageGenerationError, generate_image_result, is_token_invalid_error
-
-
-class BackendService:
-    def __init__(self, account_service: AccountService):
-        self.account_service = account_service
-
-    @staticmethod
-    def _is_account_ready_for_image(account: dict | None) -> bool:
-        if not isinstance(account, dict):
-            return False
-        if account.get("status") in {"禁用", "异常"}:
-            return False
-        return int(account.get("quota") or 0) > 0
-
-    def _refresh_request_token(self, access_token: str) -> dict | None:
-        try:
-            remote_info = self.account_service.fetch_remote_info(access_token)
-        except Exception as exc:
-            message = str(exc)
-            print(f"[image-generate] refresh token={access_token[:12]}... fail {message}")
-            if "/backend-api/me failed: HTTP 401" in message:
-                return self.account_service.update_account(
-                    access_token,
-                    {
-                        "status": "异常",
-                        "quota": 0,
-                    },
-                )
-            return None
-        return self.account_service.update_account(access_token, remote_info)
-
-    def resolve_request_token(self, excluded_tokens: set[str] | None = None) -> str:
-        try:
-            return self.account_service.next_token(excluded_tokens=excluded_tokens)
-        except RuntimeError as exc:
-            raise HTTPException(status_code=503, detail={"error": str(exc)}) from exc
-
-    def generate_with_pool(self, prompt: str, model: str, n: int):
-        attempted_tokens: set[str] = set()
-
-        while True:
-            try:
-                request_token = self.resolve_request_token(excluded_tokens=attempted_tokens)
-            except HTTPException:
-                raise
-
-            attempted_tokens.add(request_token)
-            refreshed_account = self._refresh_request_token(request_token)
-            if not self._is_account_ready_for_image(refreshed_account):
-                print(
-                    f"[image-generate] skip token={request_token[:12]}... "
-                    f"quota={refreshed_account.get('quota') if refreshed_account else 'unknown'} "
-                    f"status={refreshed_account.get('status') if refreshed_account else 'unknown'}"
-                )
-                continue
-
-            print(f"[image-generate] start pooled token={request_token[:12]}... model={model} n={n}")
-            try:
-                result = generate_image_result(request_token, prompt, model, n)
-                account = self.account_service.mark_image_result(request_token, success=True)
-                print(
-                    f"[image-generate] success pooled token={request_token[:12]}... "
-                    f"quota={account.get('quota') if account else 'unknown'} status={account.get('status') if account else 'unknown'}"
-                )
-                return result
-            except ImageGenerationError as exc:
-                account = self.account_service.mark_image_result(request_token, success=False)
-                print(
-                    f"[image-generate] fail pooled token={request_token[:12]}... "
-                    f"error={exc} quota={account.get('quota') if account else 'unknown'} status={account.get('status') if account else 'unknown'}"
-                )
-                if is_token_invalid_error(str(exc)):
-                    self.account_service.remove_token(request_token)
-                    print(f"[image-generate] remove invalid token={request_token[:12]}...")
-                    continue
-                raise

services/chat_image_service.py

@@ -1,174 +0,0 @@
-from __future__ import annotations
-
-import time
-import uuid
-
-from fastapi import HTTPException
-
-from services.backend_service import BackendService
-from services.image_service import ImageGenerationError
-
-
-IMAGE_MODELS = {"gpt-image-1", "gpt-image-2"}
-
-
-class ChatImageService:
-    def __init__(self, backend_service: BackendService):
-        self.backend_service = backend_service
-
-    def is_image_chat_request(self, body: dict[str, object]) -> bool:
-        model = str(body.get("model") or "").strip()
-        modalities = body.get("modalities")
-        if model in IMAGE_MODELS:
-            return True
-        if isinstance(modalities, list):
-            normalized = {str(item or "").strip().lower() for item in modalities}
-            return "image" in normalized
-        return False
-
-    def create_image_completion(self, body: dict[str, object]) -> dict[str, object]:
-        if not self.is_image_chat_request(body):
-            raise HTTPException(
-                status_code=400,
-                detail={"error": "only image generation requests are supported on this endpoint"},
-            )
-
-        if bool(body.get("stream")):
-            raise HTTPException(status_code=400, detail={"error": "stream is not supported for image generation"})
-
-        model = str(body.get("model") or "gpt-image-1").strip() or "gpt-image-1"
-        n = self._parse_image_count(body.get("n"))
-        prompt = self._extract_chat_prompt(body)
-        if not prompt:
-            raise HTTPException(status_code=400, detail={"error": "prompt is required"})
-
-        try:
-            image_result = self.backend_service.generate_with_pool(prompt, model, n)
-        except ImageGenerationError as exc:
-            raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc
-
-        return self._build_chat_image_completion(model, prompt, image_result)
-
-    def _extract_prompt_from_message_content(self, content: object) -> str:
-        if isinstance(content, str):
-            return content.strip()
-        if not isinstance(content, list):
-            return ""
-
-        parts: list[str] = []
-        for item in content:
-            if not isinstance(item, dict):
-                continue
-            item_type = str(item.get("type") or "").strip()
-            if item_type == "text":
-                text = str(item.get("text") or "").strip()
-                if text:
-                    parts.append(text)
-                continue
-            if item_type == "input_text":
-                text = str(item.get("input_text") or "").strip()
-                if text:
-                    parts.append(text)
-        return "\n".join(parts).strip()
-
-    def _extract_chat_prompt(self, body: dict[str, object]) -> str:
-        direct_prompt = str(body.get("prompt") or "").strip()
-        if direct_prompt:
-            return direct_prompt
-
-        messages = body.get("messages")
-        if not isinstance(messages, list):
-            return ""
-
-        prompt_parts: list[str] = []
-        for message in messages:
-            if not isinstance(message, dict):
-                continue
-            role = str(message.get("role") or "").strip().lower()
-            if role != "user":
-                continue
-            prompt = self._extract_prompt_from_message_content(message.get("content"))
-            if prompt:
-                prompt_parts.append(prompt)
-
-        return "\n".join(prompt_parts).strip()
-
-    def _parse_image_count(self, raw_value: object) -> int:
-        try:
-            value = int(raw_value or 1)
-        except (TypeError, ValueError) as exc:
-            raise HTTPException(status_code=400, detail={"error": "n must be an integer"}) from exc
-        if value < 1 or value > 4:
-            raise HTTPException(status_code=400, detail={"error": "n must be between 1 and 4"})
-        return value
-
-    def _build_chat_image_completion(
-            self,
-            model: str,
-            prompt: str,
-            image_result: dict[str, object],
-    ) -> dict[str, object]:
-        created = int(image_result.get("created") or time.time())
-        image_items = image_result.get("data") if isinstance(image_result.get("data"), list) else []
-
-        markdown_images = []
-        images_payload = []
-        content_parts = []
-
-        for index, item in enumerate(image_items, start=1):
-            if not isinstance(item, dict):
-                continue
-            b64_json = str(item.get("b64_json") or "").strip()
-            revised_prompt = str(item.get("revised_prompt") or prompt).strip()
-            if not b64_json:
-                continue
-            image_data_url = f"data:image/png;base64,{b64_json}"
-            markdown_images.append(f"![image_{index}]({image_data_url})")
-            images_payload.append(
-                {
-                    "b64_json": b64_json,
-                    "revised_prompt": revised_prompt,
-                    "url": image_data_url,
-                }
-            )
-            content_parts.append(
-                {
-                    "type": "image_url",
-                    "image_url": {
-                        "url": image_data_url,
-                    },
-                }
-            )
-
-        text_content = "\n\n".join(markdown_images) if markdown_images else "Image generation completed."
-        content_parts.insert(
-            0,
-            {
-                "type": "text",
-                "text": text_content,
-            },
-        )
-
-        return {
-            "id": f"chatcmpl-{uuid.uuid4().hex}",
-            "object": "chat.completion",
-            "created": created,
-            "model": model,
-            "choices": [
-                {
-                    "index": 0,
-                    "message": {
-                        "role": "assistant",
-                        "content": text_content,
-                        "content_parts": content_parts,
-                        "images": images_payload,
-                    },
-                    "finish_reason": "stop",
-                }
-            ],
-            "usage": {
-                "prompt_tokens": 0,
-                "completion_tokens": 0,
-                "total_tokens": 0,
-            },
-        }

services/chatgpt_service.py

@@ -0,0 +1,143 @@
+from __future__ import annotations
+
+from fastapi import HTTPException
+
+from services.account_service import AccountService
+from services.image_service import ImageGenerationError, generate_image_result, is_token_invalid_error
+from services.utils import (
+    build_chat_image_completion,
+    extract_chat_prompt,
+    extract_response_prompt,
+    has_response_image_generation_tool,
+    is_image_chat_request,
+    parse_image_count,
+)
+
+
+class ChatGPTService:
+    def __init__(self, account_service: AccountService):
+        self.account_service = account_service
+
+    def generate_with_pool(self, prompt: str, model: str, n: int):
+        created = None
+        image_items: list[dict[str, object]] = []
+
+        for index in range(1, n + 1):
+            while True:
+                try:
+                    request_token = self.account_service.get_available_access_token()
+                except RuntimeError as exc:
+                    print(f"[image-generate] stop index={index}/{n} error={exc}")
+                    break
+
+                print(f"[image-generate] start pooled token={request_token[:12]}... model={model} index={index}/{n}")
+                try:
+                    result = generate_image_result(request_token, prompt, model)
+                    account = self.account_service.mark_image_result(request_token, success=True)
+                    if created is None:
+                        created = result.get("created")
+                    data = result.get("data")
+                    if isinstance(data, list):
+                        image_items.extend(item for item in data if isinstance(item, dict))
+                    print(
+                        f"[image-generate] success pooled token={request_token[:12]}... "
+                        f"quota={account.get('quota') if account else 'unknown'} status={account.get('status') if account else 'unknown'}"
+                    )
+                    break
+                except ImageGenerationError as exc:
+                    account = self.account_service.mark_image_result(request_token, success=False)
+                    message = str(exc)
+                    print(
+                        f"[image-generate] fail pooled token={request_token[:12]}... "
+                        f"error={message} quota={account.get('quota') if account else 'unknown'} status={account.get('status') if account else 'unknown'}"
+                    )
+                    if is_token_invalid_error(message):
+                        self.account_service.remove_token(request_token)
+                        print(f"[image-generate] remove invalid token={request_token[:12]}...")
+                        continue
+                    break
+
+        if not image_items:
+            raise ImageGenerationError("image generation failed")
+
+        return {
+            "created": created,
+            "data": image_items,
+        }
+
+    def create_image_completion(self, body: dict[str, object]) -> dict[str, object]:
+        if not is_image_chat_request(body):
+            raise HTTPException(
+                status_code=400,
+                detail={"error": "only image generation requests are supported on this endpoint"},
+            )
+
+        if bool(body.get("stream")):
+            raise HTTPException(status_code=400, detail={"error": "stream is not supported for image generation"})
+
+        model = str(body.get("model") or "gpt-image-1").strip() or "gpt-image-1"
+        n = parse_image_count(body.get("n"))
+        prompt = extract_chat_prompt(body)
+        if not prompt:
+            raise HTTPException(status_code=400, detail={"error": "prompt is required"})
+
+        try:
+            image_result = self.generate_with_pool(prompt, model, n)
+        except ImageGenerationError as exc:
+            raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc
+
+        return build_chat_image_completion(model, prompt, image_result)
+
+    def create_response(self, body: dict[str, object]) -> dict[str, object]:
+        if bool(body.get("stream")):
+            raise HTTPException(status_code=400, detail={"error": "stream is not supported"})
+
+        if not has_response_image_generation_tool(body):
+            raise HTTPException(
+                status_code=400,
+                detail={"error": "only image_generation tool requests are supported on this endpoint"},
+            )
+
+        prompt = extract_response_prompt(body.get("input"))
+        if not prompt:
+            raise HTTPException(status_code=400, detail={"error": "input text is required"})
+
+        model = str(body.get("model") or "gpt-5").strip() or "gpt-5"
+        try:
+            image_result = self.generate_with_pool(prompt, "gpt-image-1", 1)
+        except ImageGenerationError as exc:
+            raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc
+
+        image_items = image_result.get("data") if isinstance(image_result.get("data"), list) else []
+        output = []
+        for item in image_items:
+            if not isinstance(item, dict):
+                continue
+            b64_json = str(item.get("b64_json") or "").strip()
+            if not b64_json:
+                continue
+            output.append(
+                {
+                    "id": f"ig_{len(output) + 1}",
+                    "type": "image_generation_call",
+                    "status": "completed",
+                    "result": b64_json,
+                    "revised_prompt": str(item.get("revised_prompt") or prompt).strip(),
+                }
+            )
+
+        if not output:
+            raise HTTPException(status_code=502, detail={"error": "image generation failed"})
+
+        created = int(image_result.get("created") or 0)
+        return {
+            "id": f"resp_{created}",
+            "object": "response",
+            "created_at": created,
+            "status": "completed",
+            "error": None,
+            "incomplete_details": None,
+            "model": model,
+            "output": output,
+            "parallel_tool_calls": False,
+        }

services/image_service.py

@@ -442,71 +442,65 @@ def _resolve_upstream_model(access_token: str, requested_model: str) -> str:
     return str(requested_model or DEFAULT_MODEL).strip() or DEFAULT_MODEL
 
 
-def generate_image_result(access_token: str, prompt: str, model: str = DEFAULT_MODEL, n: int = 1) -> dict:
+def generate_image_result(access_token: str, prompt: str, model: str = DEFAULT_MODEL) -> dict:
     prompt = str(prompt or "").strip()
     access_token = str(access_token or "").strip()
     if not prompt:
         raise ImageGenerationError("prompt is required")
     if not access_token:
         raise ImageGenerationError("token is required")
-    if n < 1:
-        raise ImageGenerationError("n must be >= 1")
 
     session, fp = _new_session(access_token)
     try:
         upstream_model = _resolve_upstream_model(access_token, model)
         print(
             f"[image-upstream] start token={access_token[:12]}... "
-            f"requested_model={model} upstream_model={upstream_model} n={n}"
+            f"requested_model={model} upstream_model={upstream_model}"
         )
-        results: list[GeneratedImage] = []
-        for _ in range(n):
-            device_id = _bootstrap(session, fp)
-            chat_token, pow_info = _chat_requirements(session, access_token, device_id)
-            proof_token = None
-            if pow_info.get("required"):
-                proof_token = _generate_proof_token(
-                    seed=str(pow_info["seed"]),
-                    difficulty=str(pow_info["difficulty"]),
-                    user_agent=USER_AGENT,
-                    proof_config=_pow_config(USER_AGENT),
-                )
-            parent_message_id = str(uuid.uuid4())
-            response = _send_conversation(
-                session,
-                access_token,
-                device_id,
-                chat_token,
-                proof_token,
-                parent_message_id,
-                prompt,
-                upstream_model,
+        device_id = _bootstrap(session, fp)
+        chat_token, pow_info = _chat_requirements(session, access_token, device_id)
+        proof_token = None
+        if pow_info.get("required"):
+            proof_token = _generate_proof_token(
+                seed=str(pow_info["seed"]),
+                difficulty=str(pow_info["difficulty"]),
+                user_agent=USER_AGENT,
+                proof_config=_pow_config(USER_AGENT),
             )
-            parsed = _parse_sse(response)
-            actual_conversation_id = parsed.get("conversation_id") or ""
-            file_ids = parsed.get("file_ids") or []
-            response_text = str(parsed.get("text") or "").strip()
-            if actual_conversation_id and not file_ids:
-                file_ids = _poll_image_ids(session, access_token, device_id, actual_conversation_id)
-            if not file_ids:
-                if response_text:
-                    raise ImageGenerationError(response_text)
-                raise ImageGenerationError("no image returned from upstream")
-            first_file_id = str(file_ids[0])
-            download_url = _fetch_download_url(session, access_token, device_id, actual_conversation_id, first_file_id)
-            if not download_url:
-                raise ImageGenerationError("failed to get download url")
-            results.append(
-                GeneratedImage(
-                    b64_json=_download_as_base64(session, download_url),
-                    revised_prompt=prompt,
-                    url=download_url,
-                )
-            )
-        print(f"[image-upstream] success token={access_token[:12]}... images={len(results)}")
+        parent_message_id = str(uuid.uuid4())
+        response = _send_conversation(
+            session,
+            access_token,
+            device_id,
+            chat_token,
+            proof_token,
+            parent_message_id,
+            prompt,
+            upstream_model,
+        )
+        parsed = _parse_sse(response)
+        actual_conversation_id = parsed.get("conversation_id") or ""
+        file_ids = parsed.get("file_ids") or []
+        response_text = str(parsed.get("text") or "").strip()
+        if actual_conversation_id and not file_ids:
+            file_ids = _poll_image_ids(session, access_token, device_id, actual_conversation_id)
+        if not file_ids:
+            if response_text:
+                raise ImageGenerationError(response_text)
+            raise ImageGenerationError("no image returned from upstream")
+        first_file_id = str(file_ids[0])
+        download_url = _fetch_download_url(session, access_token, device_id, actual_conversation_id, first_file_id)
+        if not download_url:
+            raise ImageGenerationError("failed to get download url")
+        result = GeneratedImage(
+            b64_json=_download_as_base64(session, download_url),
+            revised_prompt=prompt,
+            url=download_url,
+        )
+        print(f"[image-upstream] success token={access_token[:12]}... images=1")
         return {
             "created": time.time_ns() // 1_000_000_000,
-            "data": [{"b64_json": item.b64_json, "revised_prompt": item.revised_prompt} for item in results],
+            "data": [{"b64_json": result.b64_json, "revised_prompt": result.revised_prompt}],
         }
     except Exception as exc:
         print(f"[image-upstream] fail token={access_token[:12]}... error={exc}")

services/proof_of_work.py

@@ -9,13 +9,11 @@ from html.parser import HTMLParser
 
 import pybase64
 
-
-class _Logger:
-    def info(self, *args, **kwargs):
-        return None
-
-
-logger = _Logger()
+"""
+这个文件用于生成上游 ChatGPT 网页接口需要的 PoW 参数。
+它会解析首页 HTML 并缓存构建信息，组装浏览器指纹风格的配置数据，
+然后生成图片请求流程里会用到的 requirements token 和 proof token。
+"""
 
 cores = [8, 16, 24, 32]
 timeLayout = "%a %b %d %Y %H:%M:%S"
@@ -402,7 +400,6 @@ def get_data_build_from_html(html_content):
             data_build = match.group(1)
             cached_dpl = data_build
             cached_time = int(time.time())
-            logger.info(f"Found dpl: {cached_dpl}")
 
 
 async def get_dpl(service):
@@ -421,7 +418,6 @@ async def get_dpl(service):
         else:
             return True
     except Exception as e:
-        logger.info(f"Failed to get dpl: {e}")
         cached_dpl = None
         cached_time = int(time.time())
         return False
@@ -460,7 +456,6 @@ def get_answer_token(seed, diff, config):
     start = time.time()
     answer, solved = generate_answer(seed, diff, config)
     end = time.time()
-    logger.info(f'diff: {diff}, time: {int((end - start) * 1e6) / 1e3}ms, solved: {solved}')
     return "gAAAAAB" + answer, solved
 
 
@@ -488,20 +483,3 @@ def generate_answer(seed, diff, config):
 def get_requirements_token(config):
     require, solved = generate_answer(format(random.random()), "0fffff", config)
     return 'gAAAAAC' + require
-
-
-if __name__ == "__main__":
-    # cached_scripts.append(
-    #     "https://cdn.oaistatic.com/_next/static/cXh69klOLzS0Gy2joLDRS/_ssgManifest.js?dpl=453ebaec0d44c2decab71692e1bfe39be35a24b3")
-    # cached_dpl = "453ebaec0d44c2decab71692e1bfe39be35a24b3"
-    # cached_time = int(time.time())
-    # for i in range(10):
-    #     seed = format(random.random())
-    #     diff = "000032"
-    #     config = get_config("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome")
-    #     answer = get_answer_token(seed, diff, config)
-    cached_scripts.append(
-        "https://cdn.oaistatic.com/_next/static/cXh69klOLzS0Gy2joLDRS/_ssgManifest.js?dpl=453ebaec0d44c2decab71692e1bfe39be35a24b3")
-    cached_dpl = "prod-f501fe933b3edf57aea882da888e1a544df99840"
-    config = get_config("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36")
-    get_requirements_token(config)

services/test_image.py

@@ -1,98 +0,0 @@
-from __future__ import annotations
-
-import base64
-import json
-import sys
-import time
-from pathlib import Path
-
-ROOT_DIR = Path(__file__).resolve().parents[2]
-if str(ROOT_DIR) not in sys.path:
-    sys.path.insert(0, str(ROOT_DIR))
-
-from services.account_service import account_service
-from services.image_service import ImageGenerationError, generate_image_result, is_token_invalid_error
-
-
-OUTPUT_DIR = ROOT_DIR / "data" / "output"
-
-
-def detect_ext(image_bytes: bytes) -> str:
-    if image_bytes.startswith(b"\xff\xd8\xff"):
-        return ".jpg"
-    if image_bytes.startswith(b"RIFF") and image_bytes[8:12] == b"WEBP":
-        return ".webp"
-    if image_bytes.startswith(b"GIF87a") or image_bytes.startswith(b"GIF89a"):
-        return ".gif"
-    return ".png"
-
-
-def save_image(image_b64: str, index: int) -> Path:
-    image_bytes = base64.b64decode(image_b64)
-    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
-    path = OUTPUT_DIR / f"image_{index}_{int(time.time())}{detect_ext(image_bytes)}"
-    path.write_bytes(image_bytes)
-    return path
-
-
-def generate_image_direct(prompt: str, model: str = "gpt-5-3", n: int = 1) -> dict:
-    tokens = account_service.list_tokens()
-    if not tokens:
-        raise ImageGenerationError(f"No tokens found in {account_service.store_file}")
-
-    last_error: Exception | None = None
-    for index, token in enumerate(tokens, start=1):
-        try:
-            print(f"try token {index}/{len(tokens)}")
-            return generate_image_result(token, prompt, model, n)
-        except ImageGenerationError as exc:
-            last_error = exc
-            if is_token_invalid_error(str(exc)):
-                account_service.remove_token(token)
-                print(f"skip invalid token {index}/{len(tokens)}")
-                continue
-            raise
-
-    if last_error is not None:
-        raise last_error
-    raise ImageGenerationError("No usable token")
-
-
-def main() -> int:
-    prompt = " ".join(sys.argv[1:]).strip()
-    if not prompt:
-        prompt = "一只橘猫坐在窗台上，午后阳光，写实摄影"
-
-    print(f"prompt: {prompt}")
-    try:
-        print("request: direct image_service")
-        data = generate_image_direct(prompt)
-    except ImageGenerationError as exc:
-        print(f"image generation error: {exc}")
-        return 1
-
-    items = data.get("data") or []
-    if not items:
-        print("error: response data is empty")
-        print(json.dumps(data, ensure_ascii=False, indent=2))
-        return 1
-
-    saved = []
-    for index, item in enumerate(items, start=1):
-        image_b64 = str((item or {}).get("b64_json") or "").strip()
-        if not image_b64:
-            continue
-        path = save_image(image_b64, index)
-        saved.append(path)
-        print(f"saved: {path}")
-
-    if not saved:
-        print("error: no b64_json returned")
-        print(json.dumps(data, ensure_ascii=False, indent=2))
-        return 1
-
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

services/utils.py

@@ -0,0 +1,164 @@
+from __future__ import annotations
+
+import time
+import uuid
+
+from fastapi import HTTPException
+
+
+IMAGE_MODELS = {"gpt-image-1", "gpt-image-2"}
+
+
+def is_image_chat_request(body: dict[str, object]) -> bool:
+    model = str(body.get("model") or "").strip()
+    modalities = body.get("modalities")
+    if model in IMAGE_MODELS:
+        return True
+    if isinstance(modalities, list):
+        normalized = {str(item or "").strip().lower() for item in modalities}
+        return "image" in normalized
+    return False
+
+
+def extract_response_prompt(input_value: object) -> str:
+    if isinstance(input_value, str):
+        return input_value.strip()
+
+    if isinstance(input_value, dict):
+        role = str(input_value.get("role") or "").strip().lower()
+        if role and role != "user":
+            return ""
+        return extract_prompt_from_message_content(input_value.get("content"))
+
+    if not isinstance(input_value, list):
+        return ""
+
+    prompt_parts: list[str] = []
+    for item in input_value:
+        if isinstance(item, dict) and str(item.get("type") or "").strip() == "input_text":
+            text = str(item.get("text") or "").strip()
+            if text:
+                prompt_parts.append(text)
+            continue
+        if not isinstance(item, dict):
+            continue
+        role = str(item.get("role") or "").strip().lower()
+        if role and role != "user":
+            continue
+        prompt = extract_prompt_from_message_content(item.get("content"))
+        if prompt:
+            prompt_parts.append(prompt)
+    return "\n".join(prompt_parts).strip()
+
+
+def has_response_image_generation_tool(body: dict[str, object]) -> bool:
+    tools = body.get("tools")
+    if isinstance(tools, list):
+        for tool in tools:
+            if isinstance(tool, dict) and str(tool.get("type") or "").strip() == "image_generation":
+                return True
+
+    tool_choice = body.get("tool_choice")
+    if isinstance(tool_choice, dict) and str(tool_choice.get("type") or "").strip() == "image_generation":
+        return True
+    return False
+
+
+def extract_prompt_from_message_content(content: object) -> str:
+    if isinstance(content, str):
+        return content.strip()
+    if not isinstance(content, list):
+        return ""
+
+    parts: list[str] = []
+    for item in content:
+        if not isinstance(item, dict):
+            continue
+        item_type = str(item.get("type") or "").strip()
+        if item_type == "text":
+            text = str(item.get("text") or "").strip()
+            if text:
+                parts.append(text)
+            continue
+        if item_type == "input_text":
+            text = str(item.get("text") or item.get("input_text") or "").strip()
+            if text:
+                parts.append(text)
+    return "\n".join(parts).strip()
+
+
+def extract_chat_prompt(body: dict[str, object]) -> str:
+    direct_prompt = str(body.get("prompt") or "").strip()
+    if direct_prompt:
+        return direct_prompt
+
+    messages = body.get("messages")
+    if not isinstance(messages, list):
+        return ""
+
+    prompt_parts: list[str] = []
+    for message in messages:
+        if not isinstance(message, dict):
+            continue
+        role = str(message.get("role") or "").strip().lower()
+        if role != "user":
+            continue
+        prompt = extract_prompt_from_message_content(message.get("content"))
+        if prompt:
+            prompt_parts.append(prompt)
+
+    return "\n".join(prompt_parts).strip()
+
+
+def parse_image_count(raw_value: object) -> int:
+    try:
+        value = int(raw_value or 1)
+    except (TypeError, ValueError) as exc:
+        raise HTTPException(status_code=400, detail={"error": "n must be an integer"}) from exc
+    if value < 1 or value > 4:
+        raise HTTPException(status_code=400, detail={"error": "n must be between 1 and 4"})
+    return value
+
+
+def build_chat_image_completion(
+    model: str,
+    prompt: str,
+    image_result: dict[str, object],
+) -> dict[str, object]:
+    created = int(image_result.get("created") or time.time())
+    image_items = image_result.get("data") if isinstance(image_result.get("data"), list) else []
+
+    markdown_images = []
+
+    for index, item in enumerate(image_items, start=1):
+        if not isinstance(item, dict):
+            continue
+        b64_json = str(item.get("b64_json") or "").strip()
+        if not b64_json:
+            continue
+        image_data_url = f"data:image/png;base64,{b64_json}"
+        markdown_images.append(f"![image_{index}]({image_data_url})")
+
+    text_content = "\n\n".join(markdown_images) if markdown_images else "Image generation completed."
+
+    return {
+        "id": f"chatcmpl-{uuid.uuid4().hex}",
+        "object": "chat.completion",
+        "created": created,
+        "model": model,
+        "choices": [
+            {
+                "index": 0,
+                "message": {
+                    "role": "assistant",
+                    "content": text_content,
+                },
+                "finish_reason": "stop",
+            }
+        ],
+        "usage": {
+            "prompt_tokens": 0,
+            "completion_tokens": 0,
+            "total_tokens": 0,
+        },
+    }

test/test_completions.py

@@ -0,0 +1,12 @@
+from test.utils import post_json, save_image
+
+
+def main() -> None:
+    prompt = "A cute orange cat sitting on a chair"
+    result = post_json("/v1/chat/completions", {"model": "gpt-image-1", "stream": False, "messages": [{"role": "user", "content": prompt}]})
+    for index, item in enumerate(result["choices"][0]["message"]["images"], start=1):
+        print(save_image(item["b64_json"], f"completions_{index}"))
+
+
+if __name__ == "__main__":
+    main()

test/test_generations.py

@@ -0,0 +1,12 @@
+from test.utils import post_json, save_image
+
+
+def main() -> None:
+    prompt = "A cute orange cat sitting on a chair"
+    result = post_json("/v1/images/generations", {"prompt": prompt, "model": "gpt-image-1", "n": 1})
+    for index, item in enumerate(result["data"], start=1):
+        print(save_image(item["b64_json"], f"generations_{index}"))
+
+
+if __name__ == "__main__":
+    main()

test/test_image.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+from services.account_service import account_service
+from services.chatgpt_service import ChatGPTService
+from test.utils import save_image
+
+
+def main() -> None:
+    prompt = "一只橘猫坐在窗台上，午后阳光，写实摄影"
+    data = ChatGPTService(account_service).generate_with_pool(prompt, "gpt-5-3", 1)
+    for index, item in enumerate(data["data"], start=1):
+        print(save_image(item["b64_json"], f"image_{index}"))
+
+
+if __name__ == "__main__":
+    main()

test/utils.py

@@ -0,0 +1,47 @@
+import base64
+import json
+import sys
+import time
+import urllib.request
+from pathlib import Path
+
+
+ROOT_DIR = Path(__file__).resolve().parents[1]
+OUTPUT_DIR = ROOT_DIR / "data" / "output"
+BASE_URL = "http://127.0.0.1:8000"
+
+if str(ROOT_DIR) not in sys.path:
+    sys.path.insert(0, str(ROOT_DIR))
+
+
+def load_auth_key() -> str:
+    return json.loads((ROOT_DIR / "config.json").read_text(encoding="utf-8"))["auth-key"]
+
+
+def post_json(path: str, payload: dict) -> dict:
+    request = urllib.request.Request(
+        BASE_URL + path,
+        data=json.dumps(payload).encode(),
+        headers={"Content-Type": "application/json", "Authorization": f"Bearer {load_auth_key()}"},
+        method="POST",
+    )
+    with urllib.request.urlopen(request) as response:
+        return json.loads(response.read().decode())
+
+
+def detect_ext(image_bytes: bytes) -> str:
+    if image_bytes.startswith(b"\xff\xd8\xff"):
+        return ".jpg"
+    if image_bytes.startswith(b"RIFF") and image_bytes[8:12] == b"WEBP":
+        return ".webp"
+    if image_bytes.startswith((b"GIF87a", b"GIF89a")):
+        return ".gif"
+    return ".png"
+
+
+def save_image(image_b64: str, name: str) -> Path:
+    image_bytes = base64.b64decode(image_b64)
+    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
+    path = OUTPUT_DIR / f"{name}_{int(time.time())}{detect_ext(image_bytes)}"
+    path.write_bytes(image_bytes)
+    return path

web/next.config.ts

@@ -1,6 +1,25 @@
+import { readFileSync } from 'node:fs'
+import { dirname, join } from 'node:path'
+import { fileURLToPath } from 'node:url'
 import type { NextConfig } from 'next'
 
+const projectRoot = join(dirname(fileURLToPath(import.meta.url)), '..')
+
+function readAppVersion() {
+    try {
+        const version = readFileSync(join(projectRoot, 'VERSION'), 'utf-8').trim()
+        return version || '0.0.0'
+    } catch {
+        return '0.0.0'
+    }
+}
+
+const appVersion = process.env.NEXT_PUBLIC_APP_VERSION || readAppVersion()
+
 const nextConfig: NextConfig = {
+    env: {
+        NEXT_PUBLIC_APP_VERSION: appVersion,
+    },
     output: 'export',
     images: {
         unoptimized: true,

web/package.json

@@ -6,12 +6,7 @@
   "scripts": {
     "dev": "next dev --webpack -H 0.0.0.0",
     "build": "next build",
-    "start": "next start",
-    "lint": "eslint .",
-    "build:dev": "cp .dev .env.local && next build",
-    "build:test": "cp .test .env.local && next build",
-    "build:prod": "cp .prod .env.local && next build",
-    "openapi": "node openapi.config.js"
+    "start": "next start"
   },
   "dependencies": {
     "@radix-ui/react-checkbox": "^1.3.3",