houhou/wechat-robot-skills
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: houhou:20ea36a34083a39cacc73fdd41d9000833bad00f
Branches Tags
houhou:main
..
compare: houhou:352a1dab1bf67c7c0a303a97bbdb7a26383498fe
Branches Tags
houhou:main

No commits in common. "20ea36a34083a39cacc73fdd41d9000833bad00f" and "352a1dab1bf67c7c0a303a97bbdb7a26383498fe" have entirely different histories.
20ea36a340 ... 352a1dab1b

7 changed files with 19 additions and 454 deletions
Show Stats Expand all files Collapse all files

16
README.md
View File

@ -55,8 +55,6 @@
**发送图片的时候也可以调用 Agent 接口** **发送图片的时候也可以调用 Agent 接口**
1. 发送远程图片地址
``` ```
[POST] http://127.0.0.1:{ROBOT_WECHAT_CLIENT_PORT}/api/v1//robot/message/send/image/url [POST] http://127.0.0.1:{ROBOT_WECHAT_CLIENT_PORT}/api/v1//robot/message/send/image/url
@ -69,20 +67,6 @@
``` ```
2. 发送本地图片路径
```
[POST] http://127.0.0.1:{ROBOT_WECHAT_CLIENT_PORT}/api/v1//robot/message/send/image/local
请求体 Body:
{
"to_wxid": "{{ROBOT_FROM_WX_ID}}",
"file_path": ["{{file_path}}"]
}
```
**发送视频的时候也可以调用 Agent 接口** **发送视频的时候也可以调用 Agent 接口**
``` ```

7
skills/image-to-image/SKILL.md
View File

@ -10,7 +10,7 @@ argument-hint: "需要 prompt（提示词）和 images（图片链接列表）
这是一个 AI 图生图技能，基于输入的一张或多张图片，结合文本提示词生成新的图片。支持图片混合、风格转换、内容合成等多种创作模式。 这是一个 AI 图生图技能，基于输入的一张或多张图片，结合文本提示词生成新的图片。支持图片混合、风格转换、内容合成等多种创作模式。
支持多个绘图模型：即梦（JiMeng）、豆包（DouBao）、造相（Z-Image）、OpenAI GPT Image。 支持多个绘图模型：即梦（JiMeng）、豆包（DouBao）、造相（Z-Image）。
从数据库中读取绘图配置（API 密钥、Base URL 等），根据用户选择的模型调用对应的绘图 API，返回生成的图片 URL。 从数据库中读取绘图配置（API 密钥、Base URL 等），根据用户选择的模型调用对应的绘图 API，返回生成的图片 URL。
@ -37,7 +37,7 @@ argument-hint: "需要 prompt（提示词）和 images（图片链接列表）
}, },
"model": { "model": {
"type": "string", "type": "string",
"description": "画图模型选择（可选）：即梦4.5(jimeng-4.5) / 即梦4.6(jimeng-4.6) / 即梦5.0(jimeng-5.0) / 豆包图生图(doubao-seededit-3.0-i2i) / 造相基础版(Z-Image) / 造相蒸馏版(Z-Image-Turbo) / 造相图片编辑(Qwen-Image-Edit-2511) / OpenAI GPT Image(gpt-image-2)，默认: 空(none)。", "description": "画图模型选择（可选）：即梦4.5(jimeng-4.5) / 即梦4.6(jimeng-4.6) / 即梦5.0(jimeng-5.0) / 豆包图生图(doubao-seededit-3.0-i2i) / 造相基础版(Z-Image) / 造相蒸馏版(Z-Image-Turbo) / 造相图片编辑(Qwen-Image-Edit-2511)，默认: 空(none)。",
"enum": [ "enum": [
"none", "none",
"jimeng-4.5", "jimeng-4.5",
@ -46,8 +46,7 @@ argument-hint: "需要 prompt（提示词）和 images（图片链接列表）
"doubao-seededit-3.0-i2i", "doubao-seededit-3.0-i2i",
"Z-Image", "Z-Image",
"Z-Image-Turbo", "Z-Image-Turbo",
"Qwen-Image-Edit-2511", "Qwen-Image-Edit-2511"
"gpt-image-2"
], ],
"default": "none" "default": "none"
}, },

243
skills/image-to-image/scripts/image_to_image.py
View File

@ -3,17 +3,13 @@
from __future__ import annotations from __future__ import annotations
import argparse import argparse
import base64
import json import json
import mimetypes
import os import os
import re import re
import subprocess import subprocess
import sys import sys
import tempfile
import time import time
import traceback import traceback
import urllib.parse
import urllib.request import urllib.request
from pathlib import Path from pathlib import Path
@ -71,7 +67,6 @@ _ensure_skill_venv_python()
try: try:
import pymysql # type: ignore # noqa: E402 import pymysql # type: ignore # noqa: E402
from openai import OpenAI # type: ignore # noqa: E402
except ModuleNotFoundError: except ModuleNotFoundError:
_run_bootstrap() _run_bootstrap()
_py = _get_python_executable() _py = _get_python_executable()
@ -172,188 +167,6 @@ def _http_get_json(url: str, headers: dict, timeout: int = 30) -> dict:
return json.loads(resp.read().decode("utf-8")) return json.loads(resp.read().decode("utf-8"))
def _coerce_int(value, default: int, minimum: int, maximum: int) -> int:
try:
parsed = int(value)
except (TypeError, ValueError):
parsed = default
return min(max(parsed, minimum), maximum)
def _openai_output_format(config: dict) -> str:
output_format = str(config.get("output_format", "png") or "png").lower()
if output_format not in {"png", "jpeg", "webp"}:
return "png"
return output_format
def _openai_size(config: dict, ratio: str, resolution: str) -> str:
configured = str(config.get("size", "") or "").strip()
if configured:
return configured
normalized_ratio = (ratio or "").replace(" ", "").lower()
normalized_resolution = (resolution or "").replace(" ", "").lower()
if normalized_resolution in {"4k", "2160p", "3840x2160"}:
sizes = {
"16:9": "3840x2160",
"9:16": "2160x3840",
"1:1": "2048x2048",
"3:2": "3072x2048",
"2:3": "2048x3072",
}
elif normalized_resolution in {"2k", "1440p", "2048"}:
sizes = {
"16:9": "2048x1152",
"9:16": "1152x2048",
"1:1": "2048x2048",
"3:2": "2048x1360",
"2:3": "1360x2048",
}
elif normalized_resolution in {"1k", "1024", "1024p"}:
sizes = {
"16:9": "1536x864",
"9:16": "864x1536",
"1:1": "1024x1024",
"3:2": "1536x1024",
"2:3": "1024x1536",
}
else:
return "auto"
return sizes.get(normalized_ratio, "auto")
def _openai_prompt(prompt: str, negative_prompt: str) -> str:
if not negative_prompt:
return prompt
return f"{prompt}\n\n不要包含: {negative_prompt}"
def _openai_client(config: dict) -> OpenAI:
api_key = str(config.get("api_key", "")).strip()
if not api_key:
raise RuntimeError("OpenAI 绘图配置缺少 api_key")
kwargs: dict[str, str | float] = {"api_key": api_key}
base_url = str(config.get("base_url", "") or "").strip()
if base_url:
kwargs["base_url"] = base_url
organization = str(config.get("organization", "") or "").strip()
if organization:
kwargs["organization"] = organization
project = str(config.get("project", "") or "").strip()
if project:
kwargs["project"] = project
timeout_value = config.get("timeout")
if timeout_value not in (None, ""):
kwargs["timeout"] = float(timeout_value)
return OpenAI(**kwargs)
def _openai_image_suffix(output_format: str) -> str:
if output_format == "jpeg":
return ".jpg"
return f".{output_format}"
def _write_openai_image_file(b64_json: str, output_format: str) -> str:
image_bytes = base64.b64decode(b64_json)
with tempfile.NamedTemporaryFile(
prefix="wechat-openai-image-",
suffix=_openai_image_suffix(output_format),
delete=False,
) as image_file:
image_file.write(image_bytes)
return image_file.name
def _openai_images_from_response(response, output_format: str) -> list[str]:
outputs: list[str] = []
for item in getattr(response, "data", []) or []:
b64_json = getattr(item, "b64_json", None)
if b64_json:
outputs.append(_write_openai_image_file(b64_json, output_format))
continue
url = getattr(item, "url", None)
if url:
outputs.append(url)
return outputs
def _is_remote_image_url(value: str) -> bool:
return urllib.parse.urlparse(value).scheme in {"http", "https"}
def _send_image_outputs(client_port: str, from_wx_id: str, image_outputs: list[str]) -> None:
remote_urls = [value for value in image_outputs if value and _is_remote_image_url(value)]
local_paths = [value for value in image_outputs if value and not _is_remote_image_url(value)]
if remote_urls:
send_url = f"http://127.0.0.1:{client_port}/api/v1/robot/message/send/image/url"
send_body = {
"to_wxid": from_wx_id,
"image_urls": remote_urls,
}
_http_post_json(send_url, send_body, {"Content-Type": "application/json"}, timeout=60)
if local_paths:
send_url = f"http://127.0.0.1:{client_port}/api/v1/robot/message/send/image/local"
send_body = {
"to_wxid": from_wx_id,
"file_path": local_paths,
}
_http_post_json(send_url, send_body, {"Content-Type": "application/json"}, timeout=60)
def _cleanup_openai_temp_files(image_outputs: list[str]) -> None:
for value in image_outputs:
path = Path(value)
if path.name.startswith("wechat-openai-image-") and path.is_file():
try:
path.unlink()
except OSError:
pass
def _extension_from_mime(mime_type: str) -> str:
if mime_type == "image/jpeg":
return ".jpg"
guessed = mimetypes.guess_extension(mime_type)
if guessed in {".png", ".jpg", ".jpeg", ".webp"}:
return guessed
return ".png"
def _download_openai_input_image(image: str, directory: str, index: int) -> Path:
stripped = image.strip()
if stripped.startswith("data:"):
header, encoded = stripped.split(",", 1)
mime_type = header[5:].split(";", 1)[0] or "image/png"
path = Path(directory) / f"input-{index}{_extension_from_mime(mime_type)}"
path.write_bytes(base64.b64decode(encoded))
return path
parsed = urllib.parse.urlparse(stripped)
if parsed.scheme in {"http", "https"}:
request = urllib.request.Request(stripped, headers={"User-Agent": "wechat-robot-skills/1.0"})
with urllib.request.urlopen(request, timeout=60) as response:
content_type = response.headers.get("Content-Type", "image/png").split(";", 1)[0].strip()
suffix = Path(parsed.path).suffix.lower()
if suffix not in {".png", ".jpg", ".jpeg", ".webp"}:
suffix = _extension_from_mime(content_type)
path = Path(directory) / f"input-{index}{suffix}"
path.write_bytes(response.read())
return path
path = Path(stripped).expanduser()
if path.is_file():
return path
raise RuntimeError(f"无法读取图片: {image}")
def call_jimeng(config: dict, prompt: str, model: str, images: list[str], def call_jimeng(config: dict, prompt: str, model: str, images: list[str],
negative_prompt: str, ratio: str, resolution: str) -> list[str]: negative_prompt: str, ratio: str, resolution: str) -> list[str]:
"""Call JiMeng (即梦) image compositions API (图生图).""" """Call JiMeng (即梦) image compositions API (图生图)."""
@ -496,44 +309,6 @@ def call_zimage(config: dict, prompt: str, model: str, images: list[str]) -> lis
raise RuntimeError("造相绘图任务超时") raise RuntimeError("造相绘图任务超时")
def call_openai(config: dict, prompt: str, model: str, images: list[str],
negative_prompt: str, ratio: str, resolution: str) -> list[str]:
"""Call OpenAI GPT Image API for image editing."""
client = _openai_client(config)
output_format = _openai_output_format(config)
quality = str(config.get("quality", "auto") or "auto")
background = str(config.get("background", "auto") or "auto")
if background == "transparent":
background = "auto"
with tempfile.TemporaryDirectory() as temp_dir:
input_paths = [
_download_openai_input_image(image, temp_dir, index)
for index, image in enumerate(images[:16], start=1)
]
input_files = [path.open("rb") for path in input_paths]
try:
kwargs = {
"model": model or "gpt-image-2",
"prompt": _openai_prompt(prompt, negative_prompt),
"image": input_files,
"n": _coerce_int(config.get("n"), 1, 1, 10),
"size": _openai_size(config, ratio, resolution),
"quality": quality,
"background": background,
"output_format": output_format,
}
if output_format in {"jpeg", "webp"} and config.get("output_compression") is not None:
kwargs["output_compression"] = _coerce_int(config.get("output_compression"), 100, 0, 100)
response = client.images.edit(**kwargs)
finally:
for input_file in input_files:
input_file.close()
return _openai_images_from_response(response, output_format)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Main # Main
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@ -541,7 +316,6 @@ def call_openai(config: dict, prompt: str, model: str, images: list[str],
JIMENG_MODELS = {"jimeng-4.5", "jimeng-4.6", "jimeng-5.0"} JIMENG_MODELS = {"jimeng-4.5", "jimeng-4.6", "jimeng-5.0"}
DOUBAO_MODELS = {"doubao-seededit-3.0-i2i"} DOUBAO_MODELS = {"doubao-seededit-3.0-i2i"}
ZIMAGE_MODELS = {"Z-Image", "Z-Image-Turbo", "Qwen-Image-Edit-2511"} ZIMAGE_MODELS = {"Z-Image", "Z-Image-Turbo", "Qwen-Image-Edit-2511"}
OPENAI_MODELS = {"gpt-image-2"}
def _parse_cli_params(argv: list[str]) -> dict: def _parse_cli_params(argv: list[str]) -> dict:
@ -650,13 +424,6 @@ def main() -> int:
return 0 return 0
image_urls = call_zimage(zimage_config, prompt, model, images) image_urls = call_zimage(zimage_config, prompt, model, images)
elif model in OPENAI_MODELS:
openai_config = settings_json.get("OpenAI", {})
if not openai_config.get("enabled", False):
sys.stdout.write("OpenAI 绘图未开启\n")
return 0
image_urls = call_openai(openai_config, prompt, model, images, negative_prompt, ratio, resolution)
else: else:
sys.stdout.write("不支持的 AI 图像模型\n") sys.stdout.write("不支持的 AI 图像模型\n")
return 1 return 1
@ -672,18 +439,20 @@ def main() -> int:
# 通过客户端接口发送图片 # 通过客户端接口发送图片
client_port = os.environ.get("ROBOT_WECHAT_CLIENT_PORT", "").strip() client_port = os.environ.get("ROBOT_WECHAT_CLIENT_PORT", "").strip()
if not client_port: if not client_port:
_cleanup_openai_temp_files(image_urls)
sys.stdout.write("环境变量 ROBOT_WECHAT_CLIENT_PORT 未配置\n") sys.stdout.write("环境变量 ROBOT_WECHAT_CLIENT_PORT 未配置\n")
return 1 return 1
send_url = f"http://127.0.0.1:{client_port}/api/v1/robot/message/send/image/url"
send_body = {
"to_wxid": from_wx_id,
"image_urls": [u for u in image_urls if u],
}
try: try:
_send_image_outputs(client_port, from_wx_id, image_urls) _http_post_json(send_url, send_body, {"Content-Type": "application/json"}, timeout=60)
sys.stdout.write("图片发送成功\n") sys.stdout.write("图片发送成功\n")
except Exception as exc: except Exception as exc:
sys.stdout.write(f"发送图片失败: {exc}\n") sys.stdout.write(f"发送图片失败: {exc}\n")
return 1 return 1
finally:
_cleanup_openai_temp_files(image_urls)
return 0 return 0

1
skills/image-to-image/scripts/requirements.txt
View File

@ -1,3 +1,2 @@
cryptography cryptography
openai>=2.34.0
pymysql>=1.1,<2 pymysql>=1.1,<2

7
skills/text-to-image/SKILL.md
View File

@ -8,7 +8,7 @@ argument-hint: "需要 prompt 参数（画图提示词），可选 model（模
## 描述 ## 描述
这是一个 AI 文生图技能，当用户想通过文本描述生成图像时触发。支持多个绘图模型：即梦（JiMeng）、豆包（DouBao）、造相（Z-Image）、OpenAI GPT Image。 这是一个 AI 文生图技能，当用户想通过文本描述生成图像时触发。支持多个绘图模型：即梦（JiMeng）、豆包（DouBao）、造相（Z-Image）。
从数据库中读取绘图配置（API 密钥、Base URL 等），根据用户选择的模型调用对应的绘图 API，返回生成的图片 URL。 从数据库中读取绘图配置（API 密钥、Base URL 等），根据用户选择的模型调用对应的绘图 API，返回生成的图片 URL。
@ -35,7 +35,7 @@ argument-hint: "需要 prompt 参数（画图提示词），可选 model（模
}, },
"model": { "model": {
"type": "string", "type": "string",
"description": "画图模型选择（可选）：即梦4.5(jimeng-4.5) / 即梦4.6(jimeng-4.6) / 即梦5.0(jimeng-5.0) / 豆包4.5(doubao-seedream-4.5) / 豆包4.0(doubao-seedream-4.0) / 豆包文生图(doubao-seedream-3.0-t2i) / 豆包图生图(doubao-seededit-3.0-i2i) / 造相基础版(Z-Image) / 造相蒸馏版(Z-Image-Turbo) / 造相图片编辑(Qwen-Image-Edit-2511) / OpenAI GPT Image(gpt-image-2)，默认: 空(none)。", "description": "画图模型选择（可选）：即梦4.5(jimeng-4.5) / 即梦4.6(jimeng-4.6) / 即梦5.0(jimeng-5.0) / 豆包4.5(doubao-seedream-4.5) / 豆包4.0(doubao-seedream-4.0) / 豆包文生图(doubao-seedream-3.0-t2i) / 豆包图生图(doubao-seededit-3.0-i2i) / 造相基础版(Z-Image) / 造相蒸馏版(Z-Image-Turbo) / 造相图片编辑(Qwen-Image-Edit-2511)，默认: 空(none)。",
"enum": [ "enum": [
"none", "none",
"jimeng-4.5", "jimeng-4.5",
@ -47,8 +47,7 @@ argument-hint: "需要 prompt 参数（画图提示词），可选 model（模
"doubao-seededit-3.0-i2i", "doubao-seededit-3.0-i2i",
"Z-Image", "Z-Image",
"Z-Image-Turbo", "Z-Image-Turbo",
"Qwen-Image-Edit-2511", "Qwen-Image-Edit-2511"
"gpt-image-2"
], ],
"default": "none" "default": "none"
}, },

3
skills/text-to-image/scripts/requirements.txt
View File

@ -1,3 +1,2 @@
cryptography cryptography
openai>=2.34.0 pymysql>=1.1,<2
pymysql>=1.1,<2

196
skills/text-to-image/scripts/text_to_image.py
View File

@ -3,16 +3,13 @@
from __future__ import annotations from __future__ import annotations
import argparse import argparse
import base64
import json import json
import os import os
import re import re
import subprocess import subprocess
import sys import sys
import tempfile
import time import time
import traceback import traceback
import urllib.parse
import urllib.request import urllib.request
from pathlib import Path from pathlib import Path
@ -70,7 +67,6 @@ _ensure_skill_venv_python()
try: try:
import pymysql # type: ignore # noqa: E402 import pymysql # type: ignore # noqa: E402
from openai import OpenAI # type: ignore # noqa: E402
except ModuleNotFoundError: except ModuleNotFoundError:
_run_bootstrap() _run_bootstrap()
_py = _get_python_executable() _py = _get_python_executable()
@ -173,152 +169,6 @@ def _http_get_json(url: str, headers: dict, timeout: int = 30) -> dict:
return json.loads(resp.read().decode("utf-8")) return json.loads(resp.read().decode("utf-8"))
def _coerce_int(value, default: int, minimum: int, maximum: int) -> int:
try:
parsed = int(value)
except (TypeError, ValueError):
parsed = default
return min(max(parsed, minimum), maximum)
def _openai_output_format(config: dict) -> str:
output_format = str(config.get("output_format", "png") or "png").lower()
if output_format not in {"png", "jpeg", "webp"}:
return "png"
return output_format
def _openai_size(config: dict, ratio: str, resolution: str) -> str:
configured = str(config.get("size", "") or "").strip()
if configured:
return configured
normalized_ratio = (ratio or "").replace(" ", "").lower()
normalized_resolution = (resolution or "").replace(" ", "").lower()
if normalized_resolution in {"4k", "2160p", "3840x2160"}:
sizes = {
"16:9": "3840x2160",
"9:16": "2160x3840",
"1:1": "2048x2048",
"3:2": "3072x2048",
"2:3": "2048x3072",
}
elif normalized_resolution in {"2k", "1440p", "2048"}:
sizes = {
"16:9": "2048x1152",
"9:16": "1152x2048",
"1:1": "2048x2048",
"3:2": "2048x1360",
"2:3": "1360x2048",
}
elif normalized_resolution in {"1k", "1024", "1024p"}:
sizes = {
"16:9": "1536x864",
"9:16": "864x1536",
"1:1": "1024x1024",
"3:2": "1536x1024",
"2:3": "1024x1536",
}
else:
return "auto"
return sizes.get(normalized_ratio, "auto")
def _openai_prompt(prompt: str, negative_prompt: str) -> str:
if not negative_prompt:
return prompt
return f"{prompt}\n\n不要包含: {negative_prompt}"
def _openai_client(config: dict) -> OpenAI:
api_key = str(config.get("api_key", "")).strip()
if not api_key:
raise RuntimeError("OpenAI 绘图配置缺少 api_key")
kwargs: dict[str, str | float] = {"api_key": api_key}
base_url = str(config.get("base_url", "") or "").strip()
if base_url:
kwargs["base_url"] = base_url
organization = str(config.get("organization", "") or "").strip()
if organization:
kwargs["organization"] = organization
project = str(config.get("project", "") or "").strip()
if project:
kwargs["project"] = project
timeout_value = config.get("timeout")
if timeout_value not in (None, ""):
kwargs["timeout"] = float(timeout_value)
return OpenAI(**kwargs)
def _openai_image_suffix(output_format: str) -> str:
if output_format == "jpeg":
return ".jpg"
return f".{output_format}"
def _write_openai_image_file(b64_json: str, output_format: str) -> str:
image_bytes = base64.b64decode(b64_json)
with tempfile.NamedTemporaryFile(
prefix="wechat-openai-image-",
suffix=_openai_image_suffix(output_format),
delete=False,
) as image_file:
image_file.write(image_bytes)
return image_file.name
def _openai_images_from_response(response, output_format: str) -> list[str]:
outputs: list[str] = []
for item in getattr(response, "data", []) or []:
b64_json = getattr(item, "b64_json", None)
if b64_json:
outputs.append(_write_openai_image_file(b64_json, output_format))
continue
url = getattr(item, "url", None)
if url:
outputs.append(url)
return outputs
def _is_remote_image_url(value: str) -> bool:
return urllib.parse.urlparse(value).scheme in {"http", "https"}
def _send_image_outputs(client_port: str, from_wx_id: str, image_outputs: list[str]) -> None:
remote_urls = [value for value in image_outputs if value and _is_remote_image_url(value)]
local_paths = [value for value in image_outputs if value and not _is_remote_image_url(value)]
if remote_urls:
send_url = f"http://127.0.0.1:{client_port}/api/v1/robot/message/send/image/url"
send_body = {
"to_wxid": from_wx_id,
"image_urls": remote_urls,
}
_http_post_json(send_url, send_body, {"Content-Type": "application/json"}, timeout=60)
if local_paths:
send_url = f"http://127.0.0.1:{client_port}/api/v1/robot/message/send/image/local"
send_body = {
"to_wxid": from_wx_id,
"file_path": local_paths,
}
_http_post_json(send_url, send_body, {"Content-Type": "application/json"}, timeout=60)
def _cleanup_openai_temp_files(image_outputs: list[str]) -> None:
for value in image_outputs:
path = Path(value)
if path.name.startswith("wechat-openai-image-") and path.is_file():
try:
path.unlink()
except OSError:
pass
def call_jimeng(config: dict, prompt: str, model: str, def call_jimeng(config: dict, prompt: str, model: str,
negative_prompt: str, ratio: str, resolution: str) -> list[str]: negative_prompt: str, ratio: str, resolution: str) -> list[str]:
"""Call JiMeng (即梦) image generation API.""" """Call JiMeng (即梦) image generation API."""
@ -465,34 +315,6 @@ def call_zimage(config: dict, prompt: str, model: str) -> list[str]:
raise RuntimeError("造相绘图任务超时") raise RuntimeError("造相绘图任务超时")
def call_openai(config: dict, prompt: str, model: str,
negative_prompt: str, ratio: str, resolution: str) -> list[str]:
"""Call OpenAI GPT Image API for text-to-image generation."""
client = _openai_client(config)
output_format = _openai_output_format(config)
quality = str(config.get("quality", "auto") or "auto")
moderation = str(config.get("moderation", "auto") or "auto")
background = str(config.get("background", "auto") or "auto")
if background == "transparent":
background = "auto"
kwargs = {
"model": model or "gpt-image-2",
"prompt": _openai_prompt(prompt, negative_prompt),
"n": _coerce_int(config.get("n"), 1, 1, 10),
"size": _openai_size(config, ratio, resolution),
"quality": quality,
"background": background,
"moderation": moderation,
"output_format": output_format,
}
if output_format in {"jpeg", "webp"} and config.get("output_compression") is not None:
kwargs["output_compression"] = _coerce_int(config.get("output_compression"), 100, 0, 100)
response = client.images.generate(**kwargs)
return _openai_images_from_response(response, output_format)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Main # Main
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@ -500,7 +322,6 @@ def call_openai(config: dict, prompt: str, model: str,
JIMENG_MODELS = {"jimeng-4.5", "jimeng-4.6", "jimeng-5.0"} JIMENG_MODELS = {"jimeng-4.5", "jimeng-4.6", "jimeng-5.0"}
DOUBAO_MODELS = {"doubao-seedream-4.5", "doubao-seedream-4.0", "doubao-seedream-3.0-t2i", "doubao-seededit-3.0-i2i"} DOUBAO_MODELS = {"doubao-seedream-4.5", "doubao-seedream-4.0", "doubao-seedream-3.0-t2i", "doubao-seededit-3.0-i2i"}
ZIMAGE_MODELS = {"Z-Image", "Z-Image-Turbo", "Qwen-Image-Edit-2511"} ZIMAGE_MODELS = {"Z-Image", "Z-Image-Turbo", "Qwen-Image-Edit-2511"}
OPENAI_MODELS = {"gpt-image-2"}
def _parse_cli_params(argv: list[str]) -> dict[str, str]: def _parse_cli_params(argv: list[str]) -> dict[str, str]:
@ -602,13 +423,6 @@ def main() -> int:
return 0 return 0
image_urls = call_zimage(zimage_config, prompt, model) image_urls = call_zimage(zimage_config, prompt, model)
elif model in OPENAI_MODELS:
openai_config = settings_json.get("OpenAI", {})
if not openai_config.get("enabled", False):
sys.stdout.write("OpenAI 绘图未开启\n")
return 0
image_urls = call_openai(openai_config, prompt, model, negative_prompt, ratio, resolution)
else: else:
sys.stdout.write("不支持的 AI 图像模型\n") sys.stdout.write("不支持的 AI 图像模型\n")
return 1 return 1
@ -624,18 +438,20 @@ def main() -> int:
# 通过客户端接口发送图片 # 通过客户端接口发送图片
client_port = os.environ.get("ROBOT_WECHAT_CLIENT_PORT", "").strip() client_port = os.environ.get("ROBOT_WECHAT_CLIENT_PORT", "").strip()
if not client_port: if not client_port:
_cleanup_openai_temp_files(image_urls)
sys.stdout.write("环境变量 ROBOT_WECHAT_CLIENT_PORT 未配置\n") sys.stdout.write("环境变量 ROBOT_WECHAT_CLIENT_PORT 未配置\n")
return 1 return 1
send_url = f"http://127.0.0.1:{client_port}/api/v1/robot/message/send/image/url"
send_body = {
"to_wxid": from_wx_id,
"image_urls": [u for u in image_urls if u],
}
try: try:
_send_image_outputs(client_port, from_wx_id, image_urls) _http_post_json(send_url, send_body, {"Content-Type": "application/json"}, timeout=60)
sys.stdout.write("图片发送成功\n") sys.stdout.write("图片发送成功\n")
except Exception as exc: except Exception as exc:
sys.stdout.write(f"发送图片失败: {exc}\n") sys.stdout.write(f"发送图片失败: {exc}\n")
return 1 return 1
finally:
_cleanup_openai_temp_files(image_urls)
return 0 return 0