diff --git a/SKILL.md b/SKILL.md index 52cbcbd..15f241a 100644 --- a/SKILL.md +++ b/SKILL.md @@ -107,7 +107,7 @@ python3 -c "import markdown, bs4, cssutils, requests, yaml, pygments, PIL" 2>&1 | `config.yaml` 存在 | 静默 | 引导创建，或设 `skip_publish = true` | | Python 依赖 | 静默 | 提供 `pip install -r requirements.txt` | | `wechat.appid` + `secret` | 静默 | 设 `skip_publish = true` | -| `image.api_key` | 静默 | 设 `skip_image_gen = true` | +| `image.api_key` 或 `image.providers` 至少一项有效 | 静默 | 设 `skip_image_gen = true` | | `references/exemplars/index.yaml` | 静默 | 提示："范文库为空。如果你有已发布的文章（markdown），可以说**'导入范文'**建立风格库，写出来的文章会更像你。没有也不影响使用。" | **1.2 版本检查**（静默通过或提醒）： @@ -386,9 +386,11 @@ python3 {skill_dir}/scripts/humanness_score.py {article_path} --json --tier3 {ag - **交互模式**：展示封面，问用户"封面效果如何？"。用户 OK → 继续；不满意 → 调整提示词重新生成。 - **全自动模式**：agent 自检——提示词中的实体是否在画面描述中可识别？如果提示词过于泛化（仅含"科技感""未来感"等抽象词，无具体实体），换一组提示词重试 1 次。 -**6.4 内文配图**：分析文章结构，生成 3-6 张内文配图提示词（按 visual-prompts.md）。风格、色调、画风沿用封面，保持视觉一致。批量调用 image_gen.py，替换 Markdown 占位符。 +**6.3b 风格锚定**：封面确认后，提取视觉锚点（色板 hex、风格关键词、画面调性），后续所有内文配图的提示词必须引用这组锚点，保证全文视觉一致。 -**降级**：生图失败 → 输出提示词 + 备选图库关键词，继续。 +**6.4 内文配图**：分析文章结构，为每个需要配图的段落选择图片类型（infographic/scene/flowchart/comparison/framework/timeline），使用对应的结构化提示词模板生成 3-6 张配图提示词（按 visual-prompts.md）。批量调用 image_gen.py，替换 Markdown 占位符。 + +**降级**：image_gen.py 支持多 provider 自动 fallback（按 config.yaml 中 providers 列表顺序尝试）。全部失败 → 输出提示词 + 备选图库关键词，继续。 --- diff --git a/config.example.yaml b/config.example.yaml index 9562d9f..194c979 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -8,27 +8,63 @@ wechat: author: "" # 默认署名（可选） # AI 图片生成 +# 支持 9 个 provider，配一个就能用，配多个自动 fallback。 +# +# ┌─────────────────┬────────────────────────────────────────────────┬────────────────────┐ +# │ Provider │ 获取 API Key │ 特点 │ +# ├─────────────────┼────────────────────────────────────────────────┼────────────────────┤ +# │ doubao │ https://console.volcengine.com/ark │ 中文提示词最优 │ +# │ dashscope │ https://dashscope.console.aliyun.com/ │ 阿里通义万相 │ +# │ jimeng │ https://console.volcengine.com/iam │ 字节即梦，中文强 │ +# │ minimax │ https://platform.minimaxi.com/ │ 国内 provider │ +# │ openai │ https://platform.openai.com/api-keys │ DALL-E，通用性强 │ +# │ azure_openai │ Azure Portal │ 国内可访问的 OpenAI│ +# │ gemini │ https://aistudio.google.com/apikey │ 免费额度较多 │ +# │ openrouter │ https://openrouter.ai/settings/keys │ 多模型代理 │ +# │ replicate │ https://replicate.com/account/api-tokens │ 开源模型丰富 │ +# └─────────────────┴────────────────────────────────────────────────┴────────────────────┘ +# +# 支持两种配置方式： + +# 方式一：单 provider（简单用法，填一个就行） image: - # 可选 provider: doubao | openai | gemini - provider: "doubao" + provider: "doubao" # 见上表 Provider 列 api_key: "your_api_key" + # model: "doubao-seedream-5-0-260128" # 可选，各 provider 有默认值 + # base_url: "https://ark.cn-beijing.volces.com/api/v3" # 可选 - # doubao-seedream（默认） - # 获取 API key: https://console.volcengine.com/ark - # model: "doubao-seedream-5-0-260128" - # base_url: "https://ark.cn-beijing.volces.com/api/v3" - - # OpenAI DALL-E 3 - # provider: "openai" - # api_key: "sk-..." - # model: "dall-e-3" - # base_url: "https://api.openai.com/v1" - - # Google Gemini Imagen - # provider: "gemini" - # api_key: "AIza..." - # 获取 API key: https://aistudio.google.com/apikey - # model: "gemini-3.1-flash-image-preview" +# 方式二：多 provider 自动 fallback（推荐） +# 按顺序尝试，第一个失败自动切换下一个，不需要全部填写 +# image: +# providers: +# - provider: doubao +# api_key: "your_volcengine_key" +# - provider: dashscope +# api_key: "your_dashscope_key" +# # model: "qwen-image-2.0-pro" +# - provider: jimeng +# api_key: "your_access_key_id" # 即梦需要 access_key_id + secret_key +# secret_key: "your_secret_access_key" +# # model: "jimeng_t2i_v40" +# - provider: minimax +# api_key: "your_minimax_key" +# # model: "image-01" +# - provider: openai +# api_key: "sk-..." +# # model: "dall-e-3" +# - provider: azure_openai +# api_key: "your_azure_key" +# base_url: "https://YOUR-RESOURCE.openai.azure.com/openai" # 必填 +# # deployment: "dall-e-3" +# - provider: gemini +# api_key: "AIza..." +# # model: "gemini-3.1-flash-image-preview" +# - provider: openrouter +# api_key: "sk-or-..." +# # model: "google/gemini-3.1-flash-image-preview" +# - provider: replicate +# api_key: "r8_..." +# # model: "google/nano-banana-pro" # 默认排版主题 theme: "professional-clean" diff --git a/references/visual-prompts.md b/references/visual-prompts.md index f212b57..8431f11 100644 --- a/references/visual-prompts.md +++ b/references/visual-prompts.md @@ -73,6 +73,24 @@ --- +## 风格锚定 + +封面确认后，**立即提取视觉锚点**，后续所有内文配图必须复用： + +``` +视觉锚点： +- 色板：{封面的主色 hex + 辅色 hex，如 #2563EB + #F97316} +- 风格关键词：{封面的风格描述，如 "flat illustration, minimalist, bold outlines"} +- 画面调性：{冷调/暖调/中性} +``` + +**规则**： +- 每条内文配图提示词的末尾，必须附加视觉锚点中的色板和风格关键词 +- 如果封面是暖调，内文配图不能突然切换为冷调科技风（反之亦然） +- 视觉锚点在整篇文章的所有配图中保持一致 + +--- + ## 二、内文配图（3-6 张） ### 分析流程 @@ -94,7 +112,20 @@ | 转折/高潮处 → 视觉冲击 | 紧接着另一张配图（间距不足300字） | | 长段落后（>400字无图） → 节奏调节 | 结尾 CTA 段落 | -**第三步：确定位置** +**第三步：确定图片类型** + +根据段落内容，为每张配图选择最匹配的类型： + +| 类型 | 适用内容 | 核心构图 | +|------|---------|---------| +| infographic | 数据、统计、指标对比 | 区域分块 + 标签标注 | +| scene | 叙事场景、情绪渲染、人物故事 | 焦点主体 + 氛围光影 | +| flowchart | 流程、步骤、工作流 | 步骤节点 + 连接箭头 | +| comparison | 两个方案/观点对比 | 左右分栏 + 分隔线 | +| framework | 概念模型、架构关系 | 层级节点 + 关系连线 | +| timeline | 时间线、发展历程 | 时间轴 + 里程碑标记 | + +**第四步：确定位置** - 配图插入在对应段落**之后**（不是之前） - 具体到"H2 XX 下的第 N 段之后" @@ -104,24 +135,132 @@ - 不要在文章第一段之前放配图 - 不要在结尾 CTA 段落放配图 -### 提示词格式 +### 结构化提示词模板 -每张输出： +根据图片类型，使用对应的结构化模板生成提示词。**禁止自由文本描述**——所有提示词必须填写模板的每个字段。 + +#### infographic（信息图） ``` ### 配图 {序号}: 位于「{H2标题}」第{N}段后 -- 配图目的：{信息强化/场景还原/节奏调节} -- 对应内容：{这段讲了什么，1句话概括} -- 画面描述：{具体的画面内容，80-120字} -- AI 绘图提示词： - "{中文提示词，给 doubao-seedream 用}" +- 类型：infographic +- 对应内容：{1句话概括} + +Layout: {grid / radial / hierarchical} +Zones: + - Zone 1: {具体数据点，用文章真实数字} + - Zone 2: {对比/趋势，用文章真实数字} + - Zone 3: {结论/要点} +Labels: {文章中的真实数字、术语、指标名} +Colors: {视觉锚点色板} +Style: {视觉锚点风格关键词}, clean infographic, no text +Aspect: 16:9 + - 备选方案：{Unsplash/Pexels 搜索关键词} ``` -### 内文配图的特殊要求 +#### scene（场景） + +``` +### 配图 {序号}: 位于「{H2标题}」第{N}段后 +- 类型：scene +- 对应内容：{1句话概括} + +Focal Point: {画面主体，必须是文章实体} +Atmosphere: {光影、环境、时间} +Mood: {情绪基调} +Color Temperature: {warm / cool / neutral，与视觉锚点一致} +Style: {视觉锚点风格关键词}, no text no letters +Aspect: 16:9 + +- 备选方案：{Unsplash/Pexels 搜索关键词} +``` + +#### flowchart（流程图） + +``` +### 配图 {序号}: 位于「{H2标题}」第{N}段后 +- 类型：flowchart +- 对应内容：{1句话概括} + +Layout: {left-right / top-down / circular} +Steps: + 1. {步骤名} — {简述} + 2. {步骤名} — {简述} + 3. {步骤名} — {简述} +Connections: {箭头方向、决策分支} +Colors: {视觉锚点色板} +Style: {视觉锚点风格关键词}, clean diagram, no text +Aspect: 16:9 + +- 备选方案：{Unsplash/Pexels 搜索关键词} +``` + +#### comparison（对比图） + +``` +### 配图 {序号}: 位于「{H2标题}」第{N}段后 +- 类型：comparison +- 对应内容：{1句话概括} + +Left Side — {选项A名称}: + - {要点1} + - {要点2} +Right Side — {选项B名称}: + - {要点1} + - {要点2} +Divider: {分隔线样式} +Colors: {视觉锚点色板，左右各用一个主色} +Style: {视觉锚点风格关键词}, split layout, no text +Aspect: 16:9 + +- 备选方案：{Unsplash/Pexels 搜索关键词} +``` + +#### framework（架构图） + +``` +### 配图 {序号}: 位于「{H2标题}」第{N}段后 +- 类型：framework +- 对应内容：{1句话概括} + +Structure: {hierarchical / network / matrix} +Nodes: + - {概念1} — {角色} + - {概念2} — {角色} + - {概念3} — {角色} +Relationships: {节点间如何连接} +Colors: {视觉锚点色板} +Style: {视觉锚点风格关键词}, clean diagram, no text +Aspect: 16:9 + +- 备选方案：{Unsplash/Pexels 搜索关键词} +``` + +#### timeline（时间线） + +``` +### 配图 {序号}: 位于「{H2标题}」第{N}段后 +- 类型：timeline +- 对应内容：{1句话概括} + +Direction: {horizontal / vertical} +Events: + - {时间点1}: {里程碑} + - {时间点2}: {里程碑} + - {时间点3}: {里程碑} +Markers: {视觉标记样式} +Colors: {视觉锚点色板} +Style: {视觉锚点风格关键词}, clean timeline, no text +Aspect: 16:9 + +- 备选方案：{Unsplash/Pexels 搜索关键词} +``` + +### 内文配图通用要求 - 尺寸统一 **16:9 横版**（image_gen.py --size article） -- **风格一致性**：沿用封面确定的色调、画风、视觉语言。在每条提示词中显式复用封面的风格描述（如 "flat illustration, blue-orange palette, minimalist"） +- **视觉锚定**：每条提示词的 Colors 和 Style 字段必须引用封面提取的视觉锚点 - 实体锚定规则同封面——每条提示词至少包含 2 个文章实体 - 不要太复杂——手机屏幕上看，简洁的图比复杂的图好 - 提示词用中文（seedream 中文理解强） diff --git a/toolkit/image_gen.py b/toolkit/image_gen.py index a59b317..7b3a69e 100644 --- a/toolkit/image_gen.py +++ b/toolkit/image_gen.py @@ -6,6 +6,12 @@ Supports multiple providers via a simple abstraction: - doubao-seedream (Volcengine Ark) — default, good for Chinese prompts - openai (DALL-E 3) — broad availability - gemini (Google Gemini Imagen) — multimodal image generation + - dashscope (Alibaba Tongyi Wanxiang) — good for Chinese prompts + - minimax — Chinese provider + - replicate — open-source models + - azure_openai — Azure-hosted DALL-E + - openrouter — multi-model proxy + - jimeng (ByteDance) — good for Chinese prompts - Custom providers via ImageProvider base class Usage as CLI: @@ -21,8 +27,12 @@ Usage as module: import abc import argparse import base64 +import hashlib +import hmac import json import sys +import time +from datetime import datetime, timezone from pathlib import Path import requests @@ -51,11 +61,31 @@ def _load_config() -> dict: # Cover: 2.35:1 微信封面比例 # Article: 16:9 横版内文配图 # Vertical: 9:16 竖版 +_DEFAULT = "1792x1024" +_DEFAULT_V = "1024x1792" +_DEFAULT_SQ = "1024x1024" + SIZE_PRESETS = { - "cover": {"doubao": "2952x1256", "openai": "1792x1024", "gemini": "1792x1024"}, - "article": {"doubao": "2560x1440", "openai": "1792x1024", "gemini": "1792x1024"}, - "vertical": {"doubao": "1088x2560", "openai": "1024x1792", "gemini": "1024x1792"}, - "square": {"doubao": "2048x2048", "openai": "1024x1024", "gemini": "1024x1024"}, + "cover": { + "doubao": "2952x1256", "openai": _DEFAULT, "gemini": _DEFAULT, + "dashscope": _DEFAULT, "minimax": _DEFAULT, "replicate": _DEFAULT, + "azure_openai": _DEFAULT, "openrouter": _DEFAULT, "jimeng": _DEFAULT, + }, + "article": { + "doubao": "2560x1440", "openai": _DEFAULT, "gemini": _DEFAULT, + "dashscope": _DEFAULT, "minimax": _DEFAULT, "replicate": _DEFAULT, + "azure_openai": _DEFAULT, "openrouter": _DEFAULT, "jimeng": _DEFAULT, + }, + "vertical": { + "doubao": "1088x2560", "openai": _DEFAULT_V, "gemini": _DEFAULT_V, + "dashscope": _DEFAULT_V, "minimax": _DEFAULT_V, "replicate": _DEFAULT_V, + "azure_openai": _DEFAULT_V, "openrouter": _DEFAULT_V, "jimeng": _DEFAULT_V, + }, + "square": { + "doubao": "2048x2048", "openai": _DEFAULT_SQ, "gemini": _DEFAULT_SQ, + "dashscope": _DEFAULT_SQ, "minimax": _DEFAULT_SQ, "replicate": _DEFAULT_SQ, + "azure_openai": _DEFAULT_SQ, "openrouter": _DEFAULT_SQ, "jimeng": _DEFAULT_SQ, + }, } MAX_FILE_SIZE = 5 * 1024 * 1024 # 5MB @@ -79,6 +109,29 @@ def _compress_image(raw_bytes: bytes, max_size: int) -> bytes: return buf.getvalue() +def _size_to_aspect(size: str) -> str: + """Convert 'WxH' to nearest standard aspect ratio string.""" + if ":" in size: + return size + try: + w, h = (int(x) for x in size.split("x", 1)) + except ValueError: + return "16:9" + ratio = w / h + for ar, val in [("1:1", 1.0), ("16:9", 16/9), ("9:16", 9/16), + ("4:3", 4/3), ("3:4", 3/4), ("3:2", 3/2), ("2:3", 2/3)]: + if abs(ratio - val) < 0.15: + return ar + return "16:9" + + +def _download_image(url: str) -> bytes: + """Download image bytes from URL.""" + resp = requests.get(url, timeout=60) + resp.raise_for_status() + return resp.content + + # --- Provider abstraction --- class ImageProvider(abc.ABC): @@ -86,15 +139,7 @@ class ImageProvider(abc.ABC): @abc.abstractmethod def generate(self, prompt: str, size: str) -> bytes: - """Generate an image and return raw bytes. - - Args: - prompt: Image description (Chinese or English). - size: Resolved size string (e.g. "1792x1024"). - - Returns: - Raw image bytes. - """ + """Generate an image and return raw bytes.""" ... def resolve_size(self, preset: str) -> str: @@ -102,63 +147,45 @@ class ImageProvider(abc.ABC): provider_key = self.provider_key if preset in SIZE_PRESETS: return SIZE_PRESETS[preset].get(provider_key, list(SIZE_PRESETS[preset].values())[0]) - return preset # assume explicit WxH + return preset @property @abc.abstractmethod def provider_key(self) -> str: - """Short identifier used for size preset lookup.""" ... +# --- Providers --- + class DoubaoProvider(ImageProvider): """doubao-seedream via Volcengine Ark API.""" provider_key = "doubao" def __init__(self, api_key: str, model: str = "doubao-seedream-5-0-260128", - base_url: str = "https://ark.cn-beijing.volces.com/api/v3"): + base_url: str = "https://ark.cn-beijing.volces.com/api/v3", **_kw): self._api_key = api_key self._model = model self._base_url = base_url def generate(self, prompt: str, size: str) -> bytes: - body = { - "model": self._model, - "prompt": prompt, - "response_format": "url", - "size": size, - "stream": False, - "watermark": False, - } - resp = requests.post( f"{self._base_url}/images/generations", - headers={ - "Content-Type": "application/json", - "Authorization": f"Bearer {self._api_key}", - }, - json=body, + headers={"Content-Type": "application/json", + "Authorization": f"Bearer {self._api_key}"}, + json={"model": self._model, "prompt": prompt, + "response_format": "url", "size": size, + "stream": False, "watermark": False}, timeout=120, ) - data = resp.json() if resp.status_code != 200: - error = data.get("error", {}) - msg = error.get("message", json.dumps(data, ensure_ascii=False)) - raise ValueError(f"Doubao API error ({resp.status_code}): {msg}") - - image_data = data.get("data", []) - if not image_data: - raise ValueError(f"No image returned: {json.dumps(data, ensure_ascii=False)}") - - image_url = image_data[0].get("url") - if not image_url: - raise ValueError(f"No image URL in response: {json.dumps(data, ensure_ascii=False)}") - - img_resp = requests.get(image_url, timeout=60) - img_resp.raise_for_status() - return img_resp.content + raise ValueError(f"Doubao error ({resp.status_code}): " + f"{data.get('error', {}).get('message', str(data))}") + url = data.get("data", [{}])[0].get("url") + if not url: + raise ValueError(f"No image URL: {data}") + return _download_image(url) class OpenAIProvider(ImageProvider): @@ -167,50 +194,28 @@ class OpenAIProvider(ImageProvider): provider_key = "openai" def __init__(self, api_key: str, model: str = "dall-e-3", - base_url: str = "https://api.openai.com/v1"): + base_url: str = "https://api.openai.com/v1", **_kw): self._api_key = api_key self._model = model self._base_url = base_url def generate(self, prompt: str, size: str) -> bytes: - # DALL-E 3 expects size as "WxH" format - dall_e_size = size.replace("x", "x") # normalize - - body = { - "model": self._model, - "prompt": prompt, - "n": 1, - "size": dall_e_size, - "response_format": "url", - } - resp = requests.post( f"{self._base_url}/images/generations", - headers={ - "Content-Type": "application/json", - "Authorization": f"Bearer {self._api_key}", - }, - json=body, + headers={"Content-Type": "application/json", + "Authorization": f"Bearer {self._api_key}"}, + json={"model": self._model, "prompt": prompt, + "n": 1, "size": size, "response_format": "url"}, timeout=120, ) - data = resp.json() if resp.status_code != 200: - error = data.get("error", {}) - msg = error.get("message", json.dumps(data, ensure_ascii=False)) - raise ValueError(f"OpenAI API error ({resp.status_code}): {msg}") - - image_data = data.get("data", []) - if not image_data: - raise ValueError(f"No image returned: {json.dumps(data, ensure_ascii=False)}") - - image_url = image_data[0].get("url") - if not image_url: - raise ValueError(f"No image URL in response: {json.dumps(data, ensure_ascii=False)}") - - img_resp = requests.get(image_url, timeout=60) - img_resp.raise_for_status() - return img_resp.content + raise ValueError(f"OpenAI error ({resp.status_code}): " + f"{data.get('error', {}).get('message', str(data))}") + url = data.get("data", [{}])[0].get("url") + if not url: + raise ValueError(f"No image URL: {data}") + return _download_image(url) class GeminiProvider(ImageProvider): @@ -219,47 +224,371 @@ class GeminiProvider(ImageProvider): provider_key = "gemini" def __init__(self, api_key: str, model: str = "gemini-3.1-flash-image-preview", - base_url: str = "https://generativelanguage.googleapis.com/v1beta"): + base_url: str = "https://generativelanguage.googleapis.com/v1beta", **_kw): self._api_key = api_key self._model = model self._base_url = base_url def generate(self, prompt: str, size: str) -> bytes: - # Append size instruction to prompt (Gemini doesn't have a native size param) if "x" in size: w, h = size.split("x", 1) prompt = f"{prompt}\n\nGenerate this image at {w}x{h} resolution." - - body = { - "contents": [{"parts": [{"text": prompt}]}], - "generationConfig": {"responseModalities": ["TEXT", "IMAGE"]}, - } resp = requests.post( f"{self._base_url}/models/{self._model}:generateContent", - headers={ - "Content-Type": "application/json", - "x-goog-api-key": self._api_key, - }, - json=body, + headers={"Content-Type": "application/json", + "x-goog-api-key": self._api_key}, + json={"contents": [{"parts": [{"text": prompt}]}], + "generationConfig": {"responseModalities": ["TEXT", "IMAGE"]}}, timeout=120, ) if resp.status_code != 200: + msg = resp.text[:200] try: - error = resp.json().get("error", {}) - msg = error.get("message", resp.text[:200]) - except (ValueError, KeyError): - msg = resp.text[:200] - raise ValueError(f"Gemini API error ({resp.status_code}): {msg}") + msg = resp.json().get("error", {}).get("message", msg) + except Exception: + pass + raise ValueError(f"Gemini error ({resp.status_code}): {msg}") + for part in resp.json().get("candidates", [{}])[0].get("content", {}).get("parts", []): + inline = part.get("inlineData") + if inline and inline.get("mimeType", "").startswith("image/"): + return base64.b64decode(inline["data"]) + raise ValueError("No image in Gemini response") + + +class DashScopeProvider(ImageProvider): + """Alibaba Tongyi Wanxiang (通义万相) via DashScope API.""" + + provider_key = "dashscope" + + def __init__(self, api_key: str, model: str = "qwen-image-2.0-pro", + base_url: str = "https://dashscope.aliyuncs.com/api/v1", **_kw): + self._api_key = api_key + self._model = model + self._base_url = base_url + + def generate(self, prompt: str, size: str) -> bytes: + ds_size = size.replace("x", "*") # DashScope uses "W*H" + resp = requests.post( + f"{self._base_url}/services/aigc/multimodal-generation/generation", + headers={"Content-Type": "application/json", + "Authorization": f"Bearer {self._api_key}"}, + json={ + "model": self._model, + "input": {"messages": [{"role": "user", "content": [{"text": prompt}]}]}, + "parameters": {"prompt_extend": False, "size": ds_size, "watermark": False}, + }, + timeout=120, + ) data = resp.json() - candidates = data.get("candidates", []) - if not candidates: - raise ValueError("No candidates in Gemini response") - parts = candidates[0].get("content", {}).get("parts", []) - for part in parts: - inline_data = part.get("inlineData") - if inline_data and inline_data.get("mimeType", "").startswith("image/"): - return base64.b64decode(inline_data["data"]) - raise ValueError("No image found in Gemini response parts") + if resp.status_code != 200: + raise ValueError(f"DashScope error ({resp.status_code}): " + f"{data.get('message', str(data))}") + # Try output.result_image first, then output.choices + output = data.get("output", {}) + img = output.get("result_image") + if not img: + choices = output.get("choices", []) + if choices: + for c in choices[0].get("message", {}).get("content", []): + if "image" in c: + img = c["image"] + break + if not img: + raise ValueError(f"No image in DashScope response: {data}") + if img.startswith("http"): + return _download_image(img) + return base64.b64decode(img) + + +class MiniMaxProvider(ImageProvider): + """MiniMax image generation.""" + + provider_key = "minimax" + + def __init__(self, api_key: str, model: str = "image-01", + base_url: str = "https://api.minimax.io/v1", **_kw): + self._api_key = api_key + self._model = model + self._base_url = base_url + + def generate(self, prompt: str, size: str) -> bytes: + w, h = 1024, 1024 + try: + w, h = (int(x) for x in size.split("x", 1)) + except ValueError: + pass + resp = requests.post( + f"{self._base_url}/image_generation", + headers={"Content-Type": "application/json", + "Authorization": f"Bearer {self._api_key}"}, + json={"model": self._model, "prompt": prompt, + "response_format": "base64", + "width": w, "height": h, "n": 1}, + timeout=120, + ) + data = resp.json() + if resp.status_code != 200: + raise ValueError(f"MiniMax error ({resp.status_code}): {data}") + b64_list = data.get("data", {}).get("image_base64", []) + if not b64_list: + raise ValueError(f"No image in MiniMax response: {data}") + return base64.b64decode(b64_list[0]) + + +class ReplicateProvider(ImageProvider): + """Replicate API — supports many open-source image models.""" + + provider_key = "replicate" + _POLL_INTERVAL = 2 + _POLL_TIMEOUT = 300 + + def __init__(self, api_key: str, model: str = "google/nano-banana-pro", + base_url: str = "https://api.replicate.com/v1", **_kw): + self._api_key = api_key + self._model = model + self._base_url = base_url + + def generate(self, prompt: str, size: str) -> bytes: + aspect = _size_to_aspect(size) + headers = {"Content-Type": "application/json", + "Authorization": f"Bearer {self._api_key}", + "Prefer": "wait=60"} + resp = requests.post( + f"{self._base_url}/models/{self._model}/predictions", + headers=headers, + json={"input": {"prompt": prompt, "aspect_ratio": aspect, + "number_of_images": 1, "output_format": "png"}}, + timeout=120, + ) + data = resp.json() + if resp.status_code not in (200, 201): + raise ValueError(f"Replicate error ({resp.status_code}): {data}") + + # Poll if not completed yet + poll_url = data.get("urls", {}).get("get") + deadline = time.monotonic() + self._POLL_TIMEOUT + while data.get("status") not in ("succeeded", "failed", "canceled"): + if time.monotonic() > deadline: + raise ValueError("Replicate polling timeout") + time.sleep(self._POLL_INTERVAL) + data = requests.get(poll_url, headers=headers, timeout=30).json() + + if data.get("status") != "succeeded": + raise ValueError(f"Replicate failed: {data.get('error')}") + + output = data.get("output") + if isinstance(output, list): + output = output[0] + if isinstance(output, dict): + output = output.get("url", output.get("uri")) + if not output or not isinstance(output, str): + raise ValueError(f"No image URL in Replicate output: {data}") + return _download_image(output) + + +class AzureOpenAIProvider(ImageProvider): + """Azure-hosted OpenAI DALL-E.""" + + provider_key = "azure_openai" + + def __init__(self, api_key: str, model: str = "dall-e-3", + base_url: str = "", deployment: str = "", **_kw): + self._api_key = api_key + self._deployment = deployment or model + self._base_url = base_url.rstrip("/") + + def generate(self, prompt: str, size: str) -> bytes: + if not self._base_url: + raise ValueError("Azure OpenAI requires base_url " + "(e.g. https://YOUR-RESOURCE.openai.azure.com/openai)") + resp = requests.post( + f"{self._base_url}/deployments/{self._deployment}" + f"/images/generations?api-version=2025-04-01-preview", + headers={"Content-Type": "application/json", + "api-key": self._api_key}, + json={"prompt": prompt, "size": size, "n": 1, "quality": "medium"}, + timeout=120, + ) + data = resp.json() + if resp.status_code != 200: + raise ValueError(f"Azure OpenAI error ({resp.status_code}): {data}") + item = data.get("data", [{}])[0] + if item.get("url"): + return _download_image(item["url"]) + if item.get("b64_json"): + return base64.b64decode(item["b64_json"]) + raise ValueError(f"No image in Azure response: {data}") + + +class OpenRouterProvider(ImageProvider): + """OpenRouter — multi-model proxy using chat completions format.""" + + provider_key = "openrouter" + + def __init__(self, api_key: str, model: str = "google/gemini-3.1-flash-image-preview", + base_url: str = "https://openrouter.ai/api/v1", **_kw): + self._api_key = api_key + self._model = model + self._base_url = base_url + + def generate(self, prompt: str, size: str) -> bytes: + aspect = _size_to_aspect(size) + resp = requests.post( + f"{self._base_url}/chat/completions", + headers={"Content-Type": "application/json", + "Authorization": f"Bearer {self._api_key}"}, + json={ + "model": self._model, + "messages": [{"role": "user", "content": prompt}], + "modalities": ["image"], + "stream": False, + "image_config": {"aspect_ratio": aspect}, + "provider": {"require_parameters": True}, + }, + timeout=120, + ) + data = resp.json() + if resp.status_code != 200: + raise ValueError(f"OpenRouter error ({resp.status_code}): {data}") + + # Extract image from multiple possible locations + choice = data.get("choices", [{}])[0].get("message", {}) + # Path 1: images array + images = choice.get("images", []) + if images: + img = images[0] + if img.startswith("http"): + return _download_image(img) + if img.startswith("data:"): + _, b64 = img.split(",", 1) + return base64.b64decode(b64) + # Path 2: content array with image items + content = choice.get("content", []) + if isinstance(content, list): + for item in content: + if isinstance(item, dict) and item.get("type") == "image": + url = item.get("url") or item.get("image_url", {}).get("url") + if url: + if url.startswith("data:"): + _, b64 = url.split(",", 1) + return base64.b64decode(b64) + return _download_image(url) + raise ValueError(f"No image in OpenRouter response: {data}") + + +class JimengProvider(ImageProvider): + """ByteDance Jimeng (即梦) — async submit + poll with HMAC-SHA256 auth.""" + + provider_key = "jimeng" + _POLL_INTERVAL = 2 + _POLL_MAX_ATTEMPTS = 60 + + def __init__(self, api_key: str, secret_key: str = "", + model: str = "jimeng_t2i_v40", + base_url: str = "https://visual.volcengineapi.com", **_kw): + self._access_key = api_key + self._secret_key = secret_key + self._model = model + self._base_url = base_url + + def _sign(self, method: str, path: str, query: str, + headers: dict, payload: bytes) -> dict: + """Generate Volcengine HMAC-SHA256 signed headers.""" + now = datetime.now(timezone.utc) + date_stamp = now.strftime("%Y%m%d") + amz_date = now.strftime("%Y%m%dT%H%M%SZ") + + signed_headers_list = sorted(k.lower() for k in headers) + signed_headers_str = ";".join(signed_headers_list) + + canonical = "\n".join([ + method, path, query, + "".join(f"{k.lower()}:{headers[k]}\n" for k in sorted(headers)), + signed_headers_str, + hashlib.sha256(payload).hexdigest(), + ]) + + region = "cn-north-1" + service = "cv" + scope = f"{date_stamp}/{region}/{service}/request" + string_to_sign = "\n".join([ + "HMAC-SHA256", amz_date, scope, + hashlib.sha256(canonical.encode()).hexdigest(), + ]) + + def _hmac(key: bytes, msg: str) -> bytes: + return hmac.new(key, msg.encode(), hashlib.sha256).digest() + + k_date = _hmac(self._secret_key.encode(), date_stamp) + k_region = _hmac(k_date, region) + k_service = _hmac(k_region, service) + k_signing = _hmac(k_service, "request") + signature = hmac.new(k_signing, string_to_sign.encode(), + hashlib.sha256).hexdigest() + + auth = (f"HMAC-SHA256 Credential={self._access_key}/{scope}, " + f"SignedHeaders={signed_headers_str}, Signature={signature}") + return {**headers, "Authorization": auth, "X-Date": amz_date} + + def _request(self, action: str, body: dict) -> dict: + payload = json.dumps(body).encode() + path = "/" + query = f"Action={action}&Version=2022-08-31" + headers = { + "Content-Type": "application/json", + "Host": self._base_url.replace("https://", "").replace("http://", ""), + } + signed = self._sign("POST", path, query, headers, payload) + resp = requests.post( + f"{self._base_url}/?{query}", + headers=signed, data=payload, timeout=120, + ) + data = resp.json() + if resp.status_code != 200: + raise ValueError(f"Jimeng error ({resp.status_code}): {data}") + return data + + def generate(self, prompt: str, size: str) -> bytes: + if not self._secret_key: + raise ValueError("Jimeng requires both api_key (access_key_id) " + "and secret_key (secret_access_key)") + w, h = 1024, 1024 + try: + w, h = (int(x) for x in size.split("x", 1)) + except ValueError: + pass + + # Submit task + submit = self._request("CVSync2AsyncSubmitTask", { + "req_key": self._model, "prompt": prompt, + "width": w, "height": h, + }) + task_id = submit.get("data", {}).get("task_id") + if not task_id: + raise ValueError(f"No task_id from Jimeng: {submit}") + + # Poll for result + for _ in range(self._POLL_MAX_ATTEMPTS): + time.sleep(self._POLL_INTERVAL) + result = self._request("CVSync2AsyncGetResult", { + "req_key": self._model, "task_id": task_id, + }) + code = result.get("code") + if code == 10000: + data = result.get("data", {}) + b64_list = data.get("binary_data_base64", []) + if b64_list: + return base64.b64decode(b64_list[0]) + urls = data.get("image_urls", []) + if urls: + return _download_image(urls[0]) + raise ValueError(f"No image data in Jimeng result: {result}") + if code and code != 10000: + status = result.get("data", {}).get("status") + if status in ("failed", "canceled"): + raise ValueError(f"Jimeng task failed: {result}") + + raise ValueError("Jimeng polling timeout") # --- Provider registry --- @@ -268,37 +597,82 @@ PROVIDERS = { "doubao": DoubaoProvider, "openai": OpenAIProvider, "gemini": GeminiProvider, + "dashscope": DashScopeProvider, + "minimax": MiniMaxProvider, + "replicate": ReplicateProvider, + "azure_openai": AzureOpenAIProvider, + "openrouter": OpenRouterProvider, + "jimeng": JimengProvider, } -def _build_provider(config: dict) -> ImageProvider: - """Build an ImageProvider from config.yaml's image section.""" - img_cfg = config.get("image", {}) - provider_name = img_cfg.get("provider", "doubao") - api_key = img_cfg.get("api_key") +def _build_provider_from_entry(entry: dict) -> ImageProvider: + """Build a single ImageProvider from a provider config entry.""" + provider_name = entry.get("provider", "doubao") + api_key = entry.get("api_key") if not api_key: - raise ValueError( - f"image.api_key not set in config.yaml. " - f"Configure your {provider_name} API key to enable image generation." - ) + raise ValueError(f"No api_key for provider '{provider_name}'") provider_cls = PROVIDERS.get(provider_name) if not provider_cls: raise ValueError( - f"Unknown image provider: '{provider_name}'. " + f"Unknown provider: '{provider_name}'. " f"Available: {', '.join(PROVIDERS.keys())}" ) kwargs = {"api_key": api_key} - if img_cfg.get("model"): - kwargs["model"] = img_cfg["model"] - if img_cfg.get("base_url"): - kwargs["base_url"] = img_cfg["base_url"] + if entry.get("model"): + kwargs["model"] = entry["model"] + if entry.get("base_url"): + kwargs["base_url"] = entry["base_url"] + if entry.get("secret_key"): + kwargs["secret_key"] = entry["secret_key"] + if entry.get("deployment"): + kwargs["deployment"] = entry["deployment"] return provider_cls(**kwargs) +def _build_provider_chain(config: dict) -> list[ImageProvider]: + """Build an ordered list of providers to try. + + Supports two config formats: + - Legacy: image.provider + image.api_key (single provider) + - New: image.providers (list, tried in order with auto-fallback) + """ + img_cfg = config.get("image", {}) + providers_list = img_cfg.get("providers") + + if providers_list and isinstance(providers_list, list): + chain = [] + for entry in providers_list: + try: + chain.append(_build_provider_from_entry(entry)) + except ValueError: + continue # skip misconfigured entries + if not chain: + raise ValueError( + "No valid providers in image.providers list. " + "Each entry needs 'provider' and 'api_key'." + ) + return chain + + # Legacy single-provider format + api_key = img_cfg.get("api_key") + if not api_key: + raise ValueError( + "image.api_key not set in config.yaml. " + "Configure your API key to enable image generation." + ) + return [_build_provider_from_entry(img_cfg)] + + +def _build_provider(config: dict) -> ImageProvider: + """Build an ImageProvider from config.yaml (backward-compatible entry point).""" + return _build_provider_chain(config)[0] + + # --- Public API --- def generate_image( @@ -308,7 +682,10 @@ def generate_image( config: dict = None, ) -> str: """ - Generate an image using the configured provider. + Generate an image using configured providers with auto-fallback. + + Tries each provider in order. If one fails, falls back to the next. + Supports both single-provider (legacy) and multi-provider config. Args: prompt: Image generation prompt (Chinese or English). @@ -322,38 +699,45 @@ def generate_image( if config is None: config = _load_config() - provider = _build_provider(config) - resolved_size = provider.resolve_size(size) + chain = _build_provider_chain(config) + last_error = None - raw_bytes = provider.generate(prompt, resolved_size) + for provider in chain: + resolved_size = provider.resolve_size(size) + try: + raw_bytes = provider.generate(prompt, resolved_size) + except Exception as e: + last_error = e + print( + f"Provider '{provider.provider_key}' failed: {e}. " + f"Trying next...", + file=sys.stderr, + ) + continue - # Compress if over 5MB (WeChat upload limit) - if len(raw_bytes) > MAX_FILE_SIZE: - raw_bytes = _compress_image(raw_bytes, MAX_FILE_SIZE) + # Compress if over 5MB (WeChat upload limit) + if len(raw_bytes) > MAX_FILE_SIZE: + raw_bytes = _compress_image(raw_bytes, MAX_FILE_SIZE) - output = Path(output_path) - output.parent.mkdir(parents=True, exist_ok=True) - output.write_bytes(raw_bytes) - return str(output) + output = Path(output_path) + output.parent.mkdir(parents=True, exist_ok=True) + output.write_bytes(raw_bytes) + return str(output) + + raise ValueError( + f"All providers failed. Last error: {last_error}" + ) def main(): - parser = argparse.ArgumentParser( - description="Generate images using AI (doubao-seedream, OpenAI DALL-E, Gemini Imagen, etc.)" - ) - parser.add_argument("--prompt", required=True, help="Image generation prompt") - parser.add_argument("--output", required=True, help="Output file path") - parser.add_argument( - "--size", - default="cover", - help="Size: cover, article, vertical, square, or WxH", - ) - parser.add_argument( - "--provider", - default=None, - help="Override provider (doubao, openai, gemini). Default: from config.yaml", - ) - args = parser.parse_args() + ap = argparse.ArgumentParser(description="Generate images using AI") + ap.add_argument("--prompt", required=True, help="Image generation prompt") + ap.add_argument("--output", required=True, help="Output file path") + ap.add_argument("--size", default="cover", + help="Size: cover, article, vertical, square, or WxH") + ap.add_argument("--provider", default=None, + help=f"Override provider ({', '.join(PROVIDERS)})") + args = ap.parse_args() try: config = _load_config()