""" gemini_factory.py — 每任务构建独立的 AI client 实例 解决全局单例问题（扒 banana gemini_service.py __init__，改造为每任务局部实例） 铁律（基石B）： - 调用方只传 task_id，不传 key - 本模块在 worker 内部查库 → Fernet 解密 → 构建 client - 解密结果只活在局部变量，函数返回后即销毁 - 绝不打印 / 记录 / 传递明文 key """ from __future__ import annotations import asyncio import logging import os from dataclasses import dataclass, field from typing import Any import httpx logger = logging.getLogger(__name__) @dataclass class AIClients: """ 一个任务专用的 AI client 集合。 worker 在任务开始时构建，任务结束后释放（局部变量，不存 Redis/DB）。 """ # httpx AsyncClient 懒加载 + 按事件循环缓存：Celery 每任务多次 asyncio.run， # 持久 client 会绑死到首个已关闭的 loop → "Event loop is closed"。 # 故只存 token/base，按当前运行 loop 缓存 client，loop 变了就重建。 _gpt_token: str | None = field(default=None, repr=False) _gpt_base: str | None = field(default=None, repr=False) _gpt_client: httpx.AsyncClient | None = field(default=None, repr=False) _gpt_client_loop_id: int | None = field(default=None, repr=False) # 备用 OpenAI 兼容中转站（codeproxy）：apiports 503 时真正切过去（独立 base+key） _alt_token: str | None = field(default=None, repr=False) _alt_base: str | None = field(default=None, repr=False) # 多 base/token 的 client 池：key=(base,token的id)，按 loop 失效重建 _client_pool: dict = field(default_factory=dict, repr=False) _pool_loop_id: int | None = field(default=None, repr=False) _gemini_key: str | None = field(default=None, repr=False) # 局部变量不打印 _model_image: str = "gpt-image-2" _model_text: str = "claude-opus-4-8" # 最强档(倩倩姐红线):Claude系一律4.8,绝不降级 def has_alt_channel(self) -> bool: """是否真正配置了 codeproxy 备用通道（有独立 key 才算）。 🔴 没配 codeproxy key 时（用户未录 + env 未设），编排层据此跳过 codeproxy， 避免每张图都白白尝试 codeproxy → 撞"绝不静默降级"红线 → 空转重试 3 次。 录了 key 就自动启用真双通道互备，逻辑不变。 """ return bool(self._alt_token) def _client(self) -> httpx.AsyncClient: """主通道(apiports) client，按当前事件循环缓存""" return self._client_for(self._gpt_base, self._gpt_token) def _client_for(self, base: str | None, token: str | None) -> httpx.AsyncClient: """按 (base, token) 返回可用 client；loop 变化则整池重建（避免跨 loop 复用）""" if not token: raise RuntimeError("GPT client 未初始化（缺 token）") loop_id = id(asyncio.get_running_loop()) if self._pool_loop_id != loop_id: self._client_pool = {} self._pool_loop_id = loop_id ck = (base or "", token) if ck not in self._client_pool: self._client_pool[ck] = httpx.AsyncClient( headers={"Authorization": f"Bearer {token}"}, base_url=base or None, timeout=120.0, ) return self._client_pool[ck] # ── ImageClient 协议实现（供 image_gen.py 使用）──────── def _gpt_target(self, provider: str | None) -> tuple[str, str | None, httpx.AsyncClient]: """按 provider 选 (base, token, client)；codeproxy 走备用站独立 base+key。 🔴 绝不静默降级（红线）：显式点名 codeproxy 却没有 codeproxy 独立 key 时, 必须抛错让该通道明确失败(由上层 generate_one_image 记录并切下一通道), 绝不静默用 apiports 主 key 冒充 codeproxy——那等于"假双通道"且无声退化, 既骗了主备设计也违反红线。用户要真备份就得录入第二把 key。 """ if provider == "codeproxy": if not self._alt_token: raise RuntimeError( "codeproxy 通道未配置独立 key,拒绝用 apiports 主 key 冒充" "(红线:绝不静默降级)。请为该用户录入 codeproxy key 以启用真双通道互备。" ) base = (self._alt_base or os.environ.get("CODEPROXY_BASE_URL") or "").rstrip("/") return base, self._alt_token, self._client_for(base, self._alt_token) base = (os.environ.get("IMAGE_API_BASE") or os.environ.get("APIPORTS_BASE_URL") or "").rstrip("/") return base, self._gpt_token, self._client_for(self._gpt_base, self._gpt_token) async def gpt_edits(self, prompt: str, reference_images: list[bytes], size: str, provider: str | None = None) -> bytes: """GPT edits endpoint（带产品参考图，禁纯文生图）。 codeproxy 走 OpenAI Images `/images/edits`；apiports 实测图片编辑注册在 `/chat/completions` 多模态通道，不能复用 `/images/edits`。 """ if provider == "apiports": return await self._apiports_chat_image_edit(prompt, reference_images) import io files: list[tuple] = [("prompt", (None, prompt))] for i, img in enumerate(reference_images): files.append(("image[]", (f"ref_{i}.png", io.BytesIO(img), "image/png"))) files.append(("size", (None, size))) files.append(("model", (None, self._model_image))) base, _, client = self._gpt_target(provider) resp = await client.post(f"{base}/images/edits", files=files, timeout=120.0) resp.raise_for_status() return _extract_image_bytes(resp.json()) async def _apiports_chat_image_edit(self, prompt: str, reference_images: list[bytes]) -> bytes: """apiports 图片编辑：走 /chat/completions 多模态，参考图用 data URL。""" if not reference_images: raise ValueError("apiports 图生图缺少参考图，拒绝纯文生图") import base64 base, _, client = self._gpt_target("apiports") content: list[dict] = [{"type": "text", "text": prompt}] for img in reference_images: b64 = base64.b64encode(img).decode() content.append({ "type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}, }) payload = { "model": self._model_image, "messages": [{"role": "user", "content": content}], } resp = await client.post(f"{base}/chat/completions", json=payload, timeout=300.0) resp.raise_for_status() return _extract_chat_image_bytes(resp.json()) async def gpt_generate(self, prompt: str, size: str, provider: str | None = None) -> bytes: """GPT 纯文生图（仅 ALLOW_TEXT_ONLY_IMAGE=true 时用）""" base, _, client = self._gpt_target(provider) payload = {"model": self._model_image, "prompt": prompt, "n": 1, "size": size} resp = await client.post(f"{base}/images/generations", json=payload, timeout=120.0) resp.raise_for_status() return _extract_image_bytes(resp.json()) async def gemini_generate(self, prompt: str, reference_images: list[bytes], model: str) -> bytes: """Gemini 生图（备用通道）""" if not self._gemini_key: raise RuntimeError("Gemini key 未初始化") gemini_base = os.environ.get("GEMINI_API_URL", "https://generativelanguage.googleapis.com/v1beta") # 安全红线：key 走 header(x-goog-api-key)不进 URL query， # 否则 httpx 异常的 __str__ 含完整 URL，会把明文 key 带进 WARNING 日志。 url = f"{gemini_base}/models/{model}:generateContent" parts: list[dict] = [{"text": prompt}] for img in reference_images: import base64 parts.append({"inline_data": {"mime_type": "image/png", "data": base64.b64encode(img).decode()}}) payload = {"contents": [{"role": "user", "parts": parts}], "generationConfig": {"responseModalities": ["IMAGE", "TEXT"]}} async with httpx.AsyncClient() as client: resp = await client.post(url, json=payload, headers={"x-goog-api-key": self._gemini_key}, timeout=120.0) resp.raise_for_status() return _extract_gemini_image(resp.json()) async def _chat_post_failover(self, payload: dict, timeout: float) -> dict: """ chat/completions 发送器，带 codeproxy 回落。 主通道(apiports, claude-opus-4-8)若 5xx/超时/连接错，自动切 codeproxy 重试一次。 ⚠ codeproxy 账号池只支持 gpt 系(gpt-5.5)，无 claude，故回落时模型换成 gpt-5.5。 这是强档↔强档切换：红线"Claude系4.8/GPT系5.5"本就是两个平级最强档、互为兜底， 非降级(降级=落 sonnet/弱档)。codeproxy 回落档由 CODEPROXY_CHAT_MODEL 配置(默认 gpt-5.5)。 codeproxy 未配置(_alt_token 为空)时不回落，原样抛错。 """ main_base = (os.environ.get("IMAGE_API_BASE") or os.environ.get("APIPORTS_BASE_URL") or "").rstrip("/") try: resp = await self._client().post(f"{main_base}/chat/completions", json=payload, timeout=timeout) resp.raise_for_status() return resp.json() except (httpx.HTTPStatusError, httpx.TransportError, httpx.TimeoutException) as exc: # 回落判定：网络层错误、5xx(过载)、以及 402(欠费)/429(限流) 都该切通道—— # 这几种都是"本通道暂时不行、换通道有救"。401/403(鉴权)/400(参数) 回落也没用，直接抛。 status = getattr(getattr(exc, "response", None), "status_code", None) retryable = status is None or status >= 500 or status in (402, 429) if not (retryable and self._alt_token): raise alt_base = (self._alt_base or os.environ.get("CODEPROXY_BASE_URL") or "").rstrip("/") alt_payload = dict(payload) alt_payload["model"] = os.environ.get("CODEPROXY_CHAT_MODEL", "gpt-5.5") logger.warning("apiports chat 失败(%s)，回落 codeproxy 用 %s 重试（强档兜底，非降级）", status or type(exc).__name__, alt_payload["model"]) client = self._client_for(alt_base, self._alt_token) resp = await client.post(f"{alt_base}/chat/completions", json=alt_payload, timeout=timeout) resp.raise_for_status() return resp.json() async def chat_complete(self, messages: list[dict], model: str | None = None, max_tokens: int = 4096, temperature: float = 0.75) -> str: """文字生成（文案生成用）。apiports 503 时自动回落 codeproxy。""" payload = {"model": model or self._model_text, "messages": messages, "max_tokens": max_tokens, "temperature": temperature} # 单批≤4条文案正常 40-55s 返回；apiports 网关 ~60s 上限。客户端超时设 75s： # 略高于网关上限即可，过长（如180s）会在 apiports 卡顿时干等，拖慢整体。 timeout = float(os.environ.get("TEXT_LLM_TIMEOUT", "75")) data = await self._chat_post_failover(payload, timeout) return data["choices"][0]["message"]["content"] or "" async def gpt_vision_analyze(self, prompt: str, images: list[bytes], model: str | None = None) -> str: """ GPT/Claude vision 读产品图，返回 JSON 字符串。 messages content 混合 text + image_url(base64)，OpenAI vision 格式。 model 默认最强档（claude-opus-4-8），绝不偷降级。 最多传 4 张图，避免超 token。 """ import base64 content: list[dict] = [{"type": "text", "text": prompt}] for img in images[:4]: b64 = base64.b64encode(img).decode() content.append({ "type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}, }) used_model = model or os.environ.get("MODEL_TEXT", "claude-opus-4-8") messages = [{"role": "user", "content": content}] payload = { "model": used_model, "messages": messages, "max_tokens": 2048, "temperature": 0.2, } # 评图也走 codeproxy 回落：apiports 503 时切备用站，模型档不变（守红线）。 data = await self._chat_post_failover(payload, 90.0) return data["choices"][0]["message"]["content"] or "" # duck-type: text_variants._call_llm 用的属性 @property def _model(self) -> str: return self._model_text async def aclose(self) -> None: # client 可能绑在已关闭的 loop（Celery 多次 asyncio.run），aclose 也可能报 # "Event loop is closed"，吞掉即可——进程级连接随 loop 关闭自然释放。 for c in list(self._client_pool.values()): try: await c.aclose() except Exception: pass self._client_pool = {} self._pool_loop_id = None self._gpt_client = None self._gpt_client_loop_id = None def build_ai_clients(plain_key: str, gemini_key: str | None = None, alt_key: str | None = None) -> AIClients: """ 用解密后的明文 key 构建 AIClients。 只在 Celery worker 函数体内调用，plain_key 是局部变量。 httpx client 不在此预创建（避免绑死到调用方 loop），首次 await 时按 loop 懒建。 调用完成后 caller 负责 await clients.aclose()。 alt_key：用户录入的 codeproxy 备用站 key（基石B：worker 内查库解密后传入）。 倩倩姐红线「全做成自己的，不要埋进系统」——codeproxy key 优先用用户录入的， 未录入才回落 env CODEPROXY_KEY（向后兼容，避免没录时备用通道直接失效）。 base 地址（中转站 URL）非秘密，仍走 env 配置。 """ gpt_base = ( os.environ.get("IMAGE_API_BASE") # 旧变量名 or os.environ.get("APIPORTS_BASE_URL") # .env 实际变量名 or "" ).rstrip("/") # 备用站 codeproxy：优先用户录入 key，回落 env；apiports 503 时切过去保生图成功 alt_base = (os.environ.get("CODEPROXY_BASE_URL") or "").rstrip("/") alt_token = alt_key or os.environ.get("CODEPROXY_KEY") or None return AIClients( _gpt_token=plain_key, _gpt_base=gpt_base or None, _alt_base=alt_base or None, _alt_token=alt_token, _gemini_key=gemini_key, _model_image=os.environ.get("IMAGE_MODEL") or os.environ.get("MODEL_IMAGE", "gpt-image-2"), _model_text=os.environ.get("MODEL_TEXT", "claude-opus-4-8"), ) # ── 图片响应解析工具 ───────────────────────────────────── def _extract_image_bytes(resp_json: dict) -> bytes: """从 OpenAI images API 响应提取图片 bytes（b64 或 url）""" import base64 data = resp_json.get("data", [{}]) if not data: raise ValueError("图片 API 返回空 data") item = data[0] if "b64_json" in item: return base64.b64decode(item["b64_json"]) if "url" in item: resp = httpx.get(item["url"], timeout=30.0) resp.raise_for_status() return resp.content raise ValueError(f"无法解析图片响应：{list(item.keys())}") def _decode_b64_image(value: str) -> bytes: """解 data URL 或裸 base64 图片。""" import base64 if value.startswith("data:"): value = value.split(",", 1)[1] return base64.b64decode(value) def _bytes_from_image_item(item: dict) -> bytes | None: """兼容 chat 图片 item 里的常见 url/base64 字段。""" if not isinstance(item, dict): return None for key in ("b64_json", "base64", "data"): val = item.get(key) if isinstance(val, str) and val: return _decode_b64_image(val) url_val = None if isinstance(item.get("image_url"), dict): url_val = item["image_url"].get("url") elif isinstance(item.get("image_url"), str): url_val = item.get("image_url") elif isinstance(item.get("url"), str): url_val = item.get("url") if isinstance(url_val, str) and url_val: if url_val.startswith("data:"): return _decode_b64_image(url_val) resp = httpx.get(url_val, timeout=30.0) resp.raise_for_status() return resp.content return None def _bytes_from_content_string(text: str) -> bytes | None: """从 chat 返回的正文字符串里抠图：data URL / markdown 图链 / 裸图片URL。 很多中转站把 gpt-image 结果以 ![](url) 或 data URL 嵌在 content 文本里， 不走 images[]/parts 结构化字段，这里兜住这几种常见形态（前面带说明文字也能抠出）。 """ import re m = re.search(r"data:image/[\w.+-]+;base64,[A-Za-z0-9+/=]+", text) if m: return _decode_b64_image(m.group(0)) m = re.search(r"!\[[^\]]*\]\((https?://[^\s)]+)\)", text) # markdown 图链 if m: resp = httpx.get(m.group(1), timeout=30.0) resp.raise_for_status() return resp.content m = re.search( # 裸图片URL（限图片后缀，避免误抓非图链接） r"https?://[^\s)\]\"']+\.(?:png|jpe?g|webp|gif)(?:\?[^\s)\]\"']*)?", text, re.IGNORECASE, ) if m: resp = httpx.get(m.group(0), timeout=30.0) resp.raise_for_status() return resp.content return None def _extract_chat_image_bytes(resp_json: dict) -> bytes: """从 /chat/completions 图片响应中提取图片 bytes。""" choices = resp_json.get("choices") or [] if not choices: raise ValueError("chat 图片响应缺少 choices") message = choices[0].get("message") or {} images = message.get("images") if isinstance(images, list): for item in images: found = _bytes_from_image_item(item) if found: return found content = message.get("content") if isinstance(content, list): for part in content: found = _bytes_from_image_item(part) if found: return found elif isinstance(content, str) and content: found = _bytes_from_content_string(content) if found: return found raise ValueError( "无法解析 chat 图片响应：" f"top={list(resp_json.keys())}, message={list(message.keys())}" ) def _extract_gemini_image(resp_json: dict) -> bytes: """从 Gemini generateContent 响应提取图片 bytes""" import base64 candidates = resp_json.get("candidates", []) for cand in candidates: parts = cand.get("content", {}).get("parts", []) for part in parts: if "inlineData" in part: return base64.b64decode(part["inlineData"]["data"]) raise ValueError("Gemini 响应中未找到图片数据")