le-gpt-image v1.0:无 Key GPT Image 生图工坊——复用本机 Codex ChatGPT 登录直连内部图像通道(token 自动刷新/原子写回),generate+edit+check 零依赖 CLI + 提示词手艺参考;协议参考 codex-imagegen-cli(Apache-2.0)、方法论参考 GPT-Image2-Skill(MIT);本机实测生图 38.8s、改图 45.4s 通过
This commit is contained in:
@@ -0,0 +1,58 @@
|
||||
# le-gpt-image —— 无 Key 的 GPT Image 生图工坊
|
||||
|
||||
用 GPT Image 生成与编辑图片,**不需要 OpenAI API Key**:凭证复用本机 Codex CLI 的 ChatGPT 登录,消耗 ChatGPT 订阅额度。ZCode / Claude Code / Codex 通用。
|
||||
|
||||
## 原理
|
||||
|
||||
```
|
||||
~/.codex/auth.json(codex login 的 ChatGPT 凭证)
|
||||
│ access_token 临过期 → refresh_token 到 auth.openai.com 换新(原子写回 + 文件锁)
|
||||
▼
|
||||
POST https://chatgpt.com/backend-api/codex/images/generations|edits
|
||||
│ 请求头带 Codex 客户端身份(originator: codex_cli_rs 等)
|
||||
▼
|
||||
data[0].b64_json → 校验 PNG → 原子落盘
|
||||
```
|
||||
|
||||
这就是 Codex CLI 自己生图走的内部通道。`model` 字段固定为 `gpt-image-2`(请求格式匹配值,实际图像模型由服务端决定);备用 `--backend responses` 走推理模型强制 `image_generation` 工具。
|
||||
|
||||
脚本零第三方依赖(python3 标准库);Pillow 为可选增强——在位时改图输入自动压成最长边 1536 的 WebP。
|
||||
|
||||
## 使用
|
||||
|
||||
```bash
|
||||
# 1. 确认凭证(应显示 chatgpt / plus / token 余量)
|
||||
python3 scripts/gpt_image.py check
|
||||
|
||||
# 2. 生图
|
||||
python3 scripts/gpt_image.py generate \
|
||||
--prompt 'A minimalist poster, the text "好茶慢慢喝" centered' \
|
||||
--size portrait --quality high --out poster.png
|
||||
|
||||
# 3. 改图 / 风格迁移
|
||||
python3 scripts/gpt_image.py edit --image photo.png \
|
||||
--prompt "Change the background to snowy mountains, keep the person unchanged" --out new.png
|
||||
python3 scripts/gpt_image.py edit --image content.png --style-image ink.png --out styled.png
|
||||
```
|
||||
|
||||
前置条件:本机已 `codex login`(选 ChatGPT 登录)且 Codex 使用文件凭证存储(`~/.codex/config.toml` 中 `cli_auth_credentials_store = "file"`,CLI 登录默认即文件存储)。
|
||||
|
||||
## 实测记录(2026-10-06,本机 plus 账号)
|
||||
|
||||
| 操作 | 结果 |
|
||||
|---|---|
|
||||
| `check` | chatgpt / plus / token 剩 82.7h / refresh_token 在位 |
|
||||
| generate low 1024x1024 | 38.8s 成功;请求 1024x1024 实际返回 1374x1145(**通道不严格遵循请求尺寸**) |
|
||||
| edit 单图 low | 45.4s 成功,主体保持、背景按指令替换 |
|
||||
|
||||
## 限制
|
||||
|
||||
- 内部接口,无稳定性契约;native 失效可试 `--backend responses`。
|
||||
- 不支持透明背景;质量只有 auto/low/medium/high(xhigh/max 是 API 通道档位)。
|
||||
- `gpt-image-2.5-flare/sunburst` 为 API 通道专属,本通道不可选。
|
||||
- 单次响应一张图,`--n` 为逐张请求;每张消耗订阅额度。
|
||||
|
||||
## 来源与许可
|
||||
|
||||
- 协议实现参考 [jdmnk/codex-imagegen-cli](https://github.com/jdmnk/codex-imagegen-cli)(Apache-2.0):auth 读取/刷新/锁、请求头与端点、尺寸约束、重试策略。本仓库为独立精简实现,未逐行复制;如后续大幅借用其代码需按 Apache-2.0 附带许可声明。
|
||||
- 提示词方法论参考 [wuyoscar/GPT-Image2-Skill](https://github.com/wuyoscar/GPT-Image2-Skill)(MIT):prompt 结构公式、文字渲染、尺寸策略、反模式——只借机制,重新表述。
|
||||
@@ -0,0 +1,103 @@
|
||||
---
|
||||
name: le-gpt-image
|
||||
description: >
|
||||
用 GPT Image 生成或编辑图片:海报、封面、插画、UI mockup、信息图配图、文章插图、表情包、风格迁移改图。
|
||||
复用本机 Codex CLI 的 ChatGPT 登录凭证(~/.codex/auth.json),不需要 OpenAI API Key,
|
||||
走 Codex 内部图像通道,消耗的是 ChatGPT 订阅额度。
|
||||
当用户说「画一张图」「生成一张…的图」「做张海报/封面/头图」「帮文章配图」「把这张图的背景换成…」
|
||||
「按这张图的风格重画」「GPT Image 生图」时使用;拿到图像需求但没指明工具、而本机已登录 Codex 时也触发。
|
||||
不要用于:视频生成、需要 API 计费与发票的正式生产管线、要求 gpt-image-2.5-flare/sunburst 专属能力的任务
|
||||
(那是 API 通道专属模型,无 Key 走不了)、未登录 Codex 的机器。
|
||||
---
|
||||
|
||||
# le-gpt-image
|
||||
|
||||
用 GPT Image 生图与改图,凭证复用本机 Codex 登录,全程无 API Key。额度来自 ChatGPT 订阅——把每次生成当作在花用户的订阅用量,高成本操作先确认。
|
||||
|
||||
## 核心原则
|
||||
|
||||
- **无 Key 通道。** 凭证只读 `~/.codex/auth.json`(ChatGPT 登录态),token 临过期由脚本自动刷新并原子写回;任何情况下不打印、不外发 token。
|
||||
- **执行只走脚本。** 生图一律调 `scripts/gpt_image.py`,不现场写 HTTP/SDK 代码;判断放参考(提示词手艺读 [references/prompt-craft.md](references/prompt-craft.md)),调用走 CLI。
|
||||
- **尺寸是建议值。** 实测 Codex 通道不严格遵循请求尺寸(请求 1024x1024 可能返回 1374x1145);要精确画布时在交付说明里报告实际尺寸,必要时用改图/裁剪兜底。
|
||||
- **失败即停。** 401 刷新失败、配额不足、内容策略拒绝:如实报告并停,不换参数偷跑重试。
|
||||
|
||||
## 能力边界
|
||||
|
||||
| 能力 | 说明 |
|
||||
|---|---|
|
||||
| 文生图 | `generate`,prompt + 尺寸 + 质量 |
|
||||
| 改图 | `edit`,1–5 张参考图 + 指令;多图可做融合/参考绘制 |
|
||||
| 风格迁移 | `edit --style-image`:内容取 `--image`,画风取 `--style-image` |
|
||||
| 通道 | native(默认,GPT Image 直连,即 Codex 内部同款);responses(备用,走推理模型强制图像工具) |
|
||||
| 质量 | `auto / low / medium / high`(无 xhigh/max——那是 API 通道档位) |
|
||||
| 背景 | `auto / opaque`;**不支持透明背景**(当前通道限制) |
|
||||
| 张数 | `--n` 逐张请求,输出文件名加 `-1..-n` 后缀 |
|
||||
|
||||
单张耗时实测 40–60 秒;脚本默认超时 300 秒,429 限流自动退避重试(最多 4 次)。
|
||||
|
||||
## 工作流
|
||||
|
||||
1. **接单**:明确用途(海报/封面/插画/配图…)、主体、需出现的文字(逐字)、比例倾向。据 [references/prompt-craft.md](references/prompt-craft.md) 对应分节起草英文 prompt(图像模型对英文 prompt 更稳;图内文字保持用户原文)。
|
||||
2. **停点确认**(回合终止契约):出现以下任一情况,展示「最终 prompt + 尺寸/质量/张数 + 输出路径」后**立即结束回合**,不得再发起任何工具调用;用户确认或给修改意见后再继续——
|
||||
- 实质改写或补写了用户给的描述(用户只说「画只猫」而你扩写了一整段场景);
|
||||
- `--n > 1` 或 4k 尺寸(成倍消耗额度);
|
||||
- 要编辑不可再生的原图(用户仅有的一份照片/扫描件)——先确认输出另存、不覆盖。
|
||||
用户明确说「直接生成」、或需求已完整(明确描述 + 默认参数 + 单张)时跳过停点,直接生成。
|
||||
3. **生成与交付**:调 CLI,stdout 末行即输出路径。交付时报告:路径 + 实际尺寸 + 关键参数,另附一条可选的改进建议(换质量档/调 prompt),不自动追加生成变体。
|
||||
|
||||
## CLI 速查
|
||||
|
||||
```bash
|
||||
SKILL/scripts/gpt_image.py check # 看凭证:模式/套餐/token 余量
|
||||
SKILL/scripts/gpt_image.py generate \
|
||||
--prompt "A minimalist poster for a tea shop, warm beige palette, the text \"好茶慢慢喝\" centered" \
|
||||
--size portrait --quality high --out poster.png
|
||||
SKILL/scripts/gpt_image.py edit \
|
||||
--image photo.png --prompt "Change the background to snowy mountains, keep the person unchanged" \
|
||||
--out photo-new.png
|
||||
SKILL/scripts/gpt_image.py edit \
|
||||
--image content.png --style-image ink.png --out styled.png # 风格迁移
|
||||
```
|
||||
|
||||
| 参数 | 取值 | 说明 |
|
||||
|---|---|---|
|
||||
| `--prompt` / `--prompt-file` | 文本 / 路径 | 生图或改图指令(edit 配 `--style-image` 时可选) |
|
||||
| `--out` | 路径 | 缺省 `fig/YYYYMMDD-HHMMSS-<slug>.png`;存在须加 `--force` 才覆盖 |
|
||||
| `--size` | 见下表 | 缺省 `auto` |
|
||||
| `--quality` | `auto/low/medium/high` | 草稿 low,成品 high |
|
||||
| `--n` | 整数 | 逐张请求,文件名加后缀 |
|
||||
| `--backend` | `native/responses` | native 默认;responses 为备用通道 |
|
||||
| `--image` | 路径,可重复 | edit 输入图,1–5 张(Pillow 在位时自动压至最长边 1536 的 WebP) |
|
||||
| `--style-image` | 路径 | edit 的风格参考图 |
|
||||
| `--dry-run` | 开关 | 只打印请求(凭证打码)不联网,排障首选 |
|
||||
|
||||
### 尺寸速查
|
||||
|
||||
| 别名 | 像素 | 适用 |
|
||||
|---|---|---|
|
||||
| `1k` / `square` | 1024x1024 | 默认方图、社交媒体 |
|
||||
| `portrait` | 1024x1536 | 海报、封面、手机壁纸 |
|
||||
| `landscape` | 1536x1024 | 文章头图、横幅 |
|
||||
| `wide` | 1536x640 | 超宽 banner |
|
||||
| `tall` | 1024x2048 | 长图、竖版故事 |
|
||||
| `2k` / `4k` | 2048x2048 / 3840x2160 | 印刷、大屏(高额度,先停点确认) |
|
||||
|
||||
也接受 `WIDTHxHEIGHT`(须 16 倍数、最长边 ≤3840、宽高比 1:3–3:1)。**记住服务端可能自行调整尺寸**,交付时以实际输出为准。
|
||||
|
||||
## 故障排查
|
||||
|
||||
| 现象 | 处置 |
|
||||
|---|---|
|
||||
| 未找到凭证文件 / auth_mode 不是 chatgpt | 指引用户在本机跑 `codex login` 选 ChatGPT 登录;Codex 需文件凭证存储(config.toml `cli_auth_credentials_store = "file"`) |
|
||||
| 刷新令牌失败(HTTP 4xx) | refresh_token 已失效,指引用户重新 `codex login`;不要反复重试 |
|
||||
| HTTP 429 | 脚本自动退避重试;重试仍失败则告知用户订阅图像额度已到限,建议稍后再试或降 quality/尺寸 |
|
||||
| 尺寸不符警告 | 属正常(服务端建议值行为),按实际尺寸交付 |
|
||||
| 网络请求失败 | 检查代理(chatgpt.com 需可达);脚本遵循 `https_proxy` 等环境变量 |
|
||||
| native 通道 404/失效 | 内部接口无稳定性契约;切 `--backend responses` 试试,并在交付时说明 |
|
||||
|
||||
## 安全边界
|
||||
|
||||
- 凭证只读 `auth.json`,刷新写回仅限该文件;绝不打印 token、不把凭证带进生成参数。
|
||||
- 输出只落在用户指定目录或当前目录 `fig/`;覆盖已存在文件必须显式 `--force`。
|
||||
- 生成内容遵守平台内容政策;收到拒绝时如实转达,不换措辞绕过。
|
||||
- 每张图都消耗 ChatGPT 订阅额度:不擅自生成变体对比、不批量跑图,多张先停点确认。
|
||||
@@ -0,0 +1,65 @@
|
||||
# 提示词手艺(prompt-craft)
|
||||
|
||||
生图 prompt 的判断手册。方法论提炼自 [GPT-Image2-Skill](https://github.com/wuyoscar/GPT-Image2-Skill)(MIT,只借机制与思路),结合本通道实测校订。prompt 正文用英文写更稳,**图内要出现的文字保持用户原文**(中文、引号内逐字)。
|
||||
|
||||
## 通用结构公式
|
||||
|
||||
按此顺序组织,缺哪段跳哪段:
|
||||
|
||||
```
|
||||
[用途] + [场景/背景] + [主体] + [关键细节] + [约束(比例/风格/留白)]
|
||||
```
|
||||
|
||||
- **开头点明用途**:"A poster for…" "A UI mockup of…" "An infographic showing…"——用途决定构图逻辑。
|
||||
- **一个主角**:一张图一个视觉中心,配角围着他转;什么都想要 = 什么都糊。
|
||||
- **具体到材质与光**:说 "matte ceramic" 不说 "nice cup";说 "soft morning light from the left" 不说 "beautiful lighting"。
|
||||
- **格式一致性 > 格式本身**:短描述、指令式、JSON 式、标签式都能用,关键是通篇统一,别混。
|
||||
- **比例早定**:先想清楚这张图去哪用(见 SKILL.md 尺寸速查),prompt 里的构图描述与 `--size` 匹配——说了 "wide banner" 就别传 portrait。
|
||||
|
||||
## 文字与排版类(海报 / 封面 / 标题图)
|
||||
|
||||
- 图内文字**用直引号逐字引用**,如 `the text "好茶慢慢喝" centered`;文字要短(≤8 字最佳),长文案分图做或后期加。
|
||||
- 指定字体的**风格**而非具体字体名:"bold sans-serif", "elegant serif", "hand-brushed calligraphy"。
|
||||
- 版式约束写清楚:居中/左对齐、留白方向、文字与图形的层级("text on the upper third, illustration below")。
|
||||
- 例:`A vertical poster for a jazz bar, deep navy background, a saxophone silhouette in gold line art, the text "MIDNIGHT JAZZ" in bold condensed sans-serif at the center, generous margins, 1950s print aesthetic`
|
||||
|
||||
## 摄影类
|
||||
|
||||
- 想要真实感就直接说 "photorealistic",或用设备暗示:"iPhone photo", "shot on 35mm film", "telephoto wildlife shot"。
|
||||
- 摄影三件套:光线(golden hour / overcast / neon)、镜头(close-up / wide angle / top-down)、氛围(haze / bokeh / long exposure)。
|
||||
- 例:`Photorealistic iPhone photo of a rainy convenience store at 10pm, warm interior light spilling onto wet asphalt, reflections, a single customer under the awning`
|
||||
|
||||
## 插画 / 艺术类
|
||||
|
||||
- 点媒介与技法:watercolor / ink wash / flat vector / risograph / oil impasto / cel shading / pixel art / isometric。
|
||||
- 点参照系(艺术家、年代、地区画风)比堆形容词有效:"1960s Soviet cartoon style", "Japanese ukiyo-e composition"。
|
||||
- 控色用色板表述:"limited palette of teal, cream and burnt orange"。
|
||||
|
||||
## 信息图 / UI 类
|
||||
|
||||
- 信息图先给信息结构:"a 3-step flow diagram", "a comparison table of 4 plans"——模型按结构排,不按你脑补排。
|
||||
- 数据必须真实可核:图里的数字来自用户素材,编数字 = 报废。
|
||||
- UI mockup 说明平台与状态:"iOS settings screen, light mode, a toggle switched on"。
|
||||
- 例:`A clean infographic explaining how cold brew coffee is made, 4 numbered steps arranged vertically, minimal line icons, muted earth tones, generous whitespace`
|
||||
|
||||
## 中文与东方风格(实测有效向)
|
||||
|
||||
- 水墨:`Chinese ink-and-wash painting, negative space, a lone fisherman on a small boat, mist over the river`
|
||||
- 宋画长卷:`horizontal handscroll in Song dynasty style, riverside night market, fine brushwork, muted colors`
|
||||
- 新中式海报:竖版 + 大量留白 + 单件器物 + 书法字:"a celadon teapot on the lower third, the calligraphy text \"静\" in the upper left, rice-paper texture"
|
||||
- 国潮:饱和撞色 + 传统纹样(cloud patterns, wave motifs)+ 粗描边。
|
||||
|
||||
## 改图指令写法(edit)
|
||||
|
||||
- **说清不变量**:"keep the person, pose and lighting unchanged"——只说改什么,模型会连不该动的也动。
|
||||
- 局部改写指名区域:"replace only the background behind the subject", "remove the watermark in the bottom-right corner"。
|
||||
- 风格迁移优先用 `--style-image`(脚本内置了保内容换画风的指令模板),别自己描述"像第二张图那样"。
|
||||
- 多参考图融合时给角色分工:"the first image provides the layout, the second provides the color palette"。
|
||||
|
||||
## 反模式
|
||||
|
||||
- 堆形容词不给结构("amazing stunning beautiful…")——噪声。
|
||||
- 一图多主角、一 prompt 多场景——出拼接怪。
|
||||
- 图内塞长段落文字——当前模型长文本渲染不可靠,≤8 字最优。
|
||||
- 负面指令("no hands")基本无效,改用正面描述绕开。
|
||||
- 拿尺寸当精确裁剪工具——通道对尺寸只算建议值,精确裁剪后期做。
|
||||
Executable
+866
@@ -0,0 +1,866 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""le-gpt-image 生图 CLI:复用本机 Codex 的 ChatGPT 登录调 GPT Image,无需 OpenAI API Key。
|
||||
|
||||
用法:
|
||||
gpt_image.py generate --prompt "..." --out cover.png --size portrait --quality high
|
||||
gpt_image.py edit --image in.png --prompt "把背景换成雪山" --out out.png
|
||||
gpt_image.py check # 查看凭证状态(不生图、不打印任何密钥)
|
||||
|
||||
协议参考 jdmnk/codex-imagegen-cli(Apache-2.0)与 OpenAI Codex CLI 的请求格式:
|
||||
读取 ~/.codex/auth.json → access_token 临近过期时用 refresh_token 换新并原子写回 →
|
||||
带 Codex 客户端请求头 POST chatgpt.com/backend-api/codex/images/generations|edits。
|
||||
仅用 python3 标准库;Pillow 为可选增强(存在时把改图输入压缩成 WebP,否则原样内嵌)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import platform
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
from contextlib import contextmanager
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Sequence
|
||||
from urllib import error, request
|
||||
|
||||
try:
|
||||
import fcntl
|
||||
except ImportError: # Windows:无 flock,退化为无锁刷新
|
||||
fcntl = None # type: ignore[assignment]
|
||||
|
||||
try:
|
||||
import tomllib
|
||||
except ImportError: # Python < 3.11
|
||||
tomllib = None # type: ignore[assignment]
|
||||
|
||||
__version__ = "1.0.0"
|
||||
|
||||
DEFAULT_BASE_URL = "https://chatgpt.com/backend-api/codex"
|
||||
REFRESH_URL = "https://auth.openai.com/oauth/token"
|
||||
CODEX_CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann"
|
||||
NATIVE_MODEL = "gpt-image-2" # 请求格式匹配值,实际图像模型由服务端决定
|
||||
DEFAULT_REASONING_MODEL = "gpt-5.5"
|
||||
ORIGINATOR = "codex_cli_rs"
|
||||
MAX_EDIT_IMAGES = 5
|
||||
MAX_RETRIES = 4
|
||||
TOKEN_LEEWAY_SECONDS = 60
|
||||
INPUT_MAX_EDGE = 1536
|
||||
|
||||
# 尺寸别名 → 具体像素(全部满足 16 倍数 / 边≤3840 / 1:3–3:1 / 总像素上下限)
|
||||
SIZE_ALIASES = {
|
||||
"1k": "1024x1024",
|
||||
"square": "1024x1024",
|
||||
"2k": "2048x2048",
|
||||
"4k": "3840x2160",
|
||||
"portrait": "1024x1536",
|
||||
"landscape": "1536x1024",
|
||||
"wide": "1536x640",
|
||||
"tall": "1024x2048",
|
||||
}
|
||||
|
||||
|
||||
class CliError(Exception):
|
||||
"""预期内的用户可见失败。"""
|
||||
|
||||
|
||||
class HttpError(CliError):
|
||||
def __init__(self, status: int, body: str, retry_after: str | None = None) -> None:
|
||||
self.status = status
|
||||
self.body = body
|
||||
self.retry_after = retry_after
|
||||
super().__init__(f"HTTP {status}: {body[:800]}")
|
||||
|
||||
|
||||
def _log(msg: str) -> None:
|
||||
print(msg, file=sys.stderr)
|
||||
|
||||
|
||||
def _warn(msg: str) -> None:
|
||||
print(f"警告: {msg}", file=sys.stderr)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 凭证与令牌
|
||||
|
||||
def _jwt_payload(token: str) -> dict[str, Any]:
|
||||
parts = token.split(".")
|
||||
if len(parts) < 2:
|
||||
return {}
|
||||
try:
|
||||
padded = parts[1] + "=" * (-len(parts[1]) % 4)
|
||||
payload = json.loads(base64.urlsafe_b64decode(padded.encode("ascii")).decode("utf-8"))
|
||||
except (ValueError, UnicodeDecodeError):
|
||||
return {}
|
||||
return payload if isinstance(payload, dict) else {}
|
||||
|
||||
|
||||
def _token_is_expiring(access_token: str, leeway: int = TOKEN_LEEWAY_SECONDS) -> bool:
|
||||
exp = _jwt_payload(access_token).get("exp")
|
||||
if not isinstance(exp, (int, float)):
|
||||
return False
|
||||
return exp <= datetime.now(timezone.utc).timestamp() + leeway
|
||||
|
||||
|
||||
def _codex_home(args: argparse.Namespace) -> Path:
|
||||
if getattr(args, "codex_home", None):
|
||||
return Path(args.codex_home).expanduser().resolve()
|
||||
if os.environ.get("CODEX_HOME"):
|
||||
return Path(os.environ["CODEX_HOME"]).expanduser().resolve()
|
||||
return Path.home() / ".codex"
|
||||
|
||||
|
||||
def _auth_file(args: argparse.Namespace) -> Path:
|
||||
if getattr(args, "auth_file", None):
|
||||
return Path(args.auth_file).expanduser().resolve()
|
||||
return _codex_home(args) / "auth.json"
|
||||
|
||||
|
||||
def _load_auth(auth_file: Path) -> dict[str, Any]:
|
||||
if not auth_file.exists():
|
||||
raise CliError(
|
||||
f"未找到 Codex 凭证文件: {auth_file}\n"
|
||||
"本 skill 走 ChatGPT 订阅通道,需要先在本机登录 Codex CLI(`codex login` 选 ChatGPT 账号),\n"
|
||||
'且 Codex 需使用文件凭证存储(config.toml 里 cli_auth_credentials_store = "file")。'
|
||||
)
|
||||
try:
|
||||
data = json.loads(auth_file.read_text(encoding="utf-8"))
|
||||
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
||||
raise CliError(f"凭证文件不可读或不是合法 JSON: {auth_file} ({exc})") from exc
|
||||
if not isinstance(data, dict):
|
||||
raise CliError(f"凭证文件内容应为 JSON 对象: {auth_file}")
|
||||
return data
|
||||
|
||||
|
||||
def _tokens(auth: dict[str, Any]) -> dict[str, Any]:
|
||||
tokens = auth.get("tokens")
|
||||
if not isinstance(tokens, dict):
|
||||
raise CliError("凭证里没有 ChatGPT tokens,请重新 `codex login`(选 ChatGPT 登录)。")
|
||||
return tokens
|
||||
|
||||
|
||||
def _access_token(auth: dict[str, Any]) -> str:
|
||||
token = _tokens(auth).get("access_token")
|
||||
if not isinstance(token, str) or not token:
|
||||
raise CliError("缺少 access_token,请重新 `codex login`。")
|
||||
return token
|
||||
|
||||
|
||||
def _id_token_claims(auth: dict[str, Any]) -> dict[str, Any]:
|
||||
id_token = _tokens(auth).get("id_token")
|
||||
if isinstance(id_token, str) and id_token:
|
||||
payload = _jwt_payload(id_token)
|
||||
claim = payload.get("https://api.openai.com/auth")
|
||||
return claim if isinstance(claim, dict) else {}
|
||||
return {}
|
||||
|
||||
|
||||
def _account_id(auth: dict[str, Any]) -> str | None:
|
||||
account_id = _tokens(auth).get("account_id")
|
||||
if isinstance(account_id, str) and account_id:
|
||||
return account_id
|
||||
account_id = _id_token_claims(auth).get("chatgpt_account_id")
|
||||
return account_id if isinstance(account_id, str) and account_id else None
|
||||
|
||||
|
||||
def _auth_headers(auth: dict[str, Any]) -> dict[str, str]:
|
||||
headers = {
|
||||
"Authorization": f"Bearer {_access_token(auth)}",
|
||||
"Accept": "application/json",
|
||||
"Content-Type": "application/json",
|
||||
"originator": ORIGINATOR,
|
||||
"User-Agent": (
|
||||
f"{ORIGINATOR}/{_codex_version()} ({platform.system()} {platform.machine()}) "
|
||||
f"le-gpt-image/{__version__}"
|
||||
),
|
||||
"version": _codex_version(),
|
||||
}
|
||||
account_id = _account_id(auth)
|
||||
if account_id:
|
||||
headers["ChatGPT-Account-ID"] = account_id
|
||||
if _id_token_claims(auth).get("chatgpt_account_is_fedramp"):
|
||||
headers["X-OpenAI-Fedramp"] = "true"
|
||||
return headers
|
||||
|
||||
|
||||
def _codex_version() -> str:
|
||||
codex_bin = shutil.which("codex")
|
||||
if not codex_bin:
|
||||
return __version__
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[codex_bin, "--version"], check=False, text=True, capture_output=True, timeout=5
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
return __version__
|
||||
match = re.search(r"(\d+\.\d+\.\d+)", f"{result.stdout}\n{result.stderr}")
|
||||
return match.group(1) if match else __version__
|
||||
|
||||
|
||||
def _atomic_write(path: Path, data: bytes, *, replace: bool) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent)
|
||||
try:
|
||||
with os.fdopen(fd, "wb") as output:
|
||||
output.write(data)
|
||||
output.flush()
|
||||
os.fsync(output.fileno())
|
||||
if replace:
|
||||
os.replace(temporary, path)
|
||||
else:
|
||||
# 原子防覆盖:生成中途出现的输出文件绝不能被悄悄覆盖
|
||||
try:
|
||||
os.link(temporary, path)
|
||||
except FileExistsError as exc:
|
||||
raise CliError(f"输出文件已存在: {path}(加 --force 才允许覆盖)") from exc
|
||||
finally:
|
||||
Path(temporary).unlink(missing_ok=True)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _refresh_lock(auth_file: Path, timeout: float):
|
||||
"""与并发实例互斥的刷新锁;锁文件独立于 auth.json,才能扛住原子替换。"""
|
||||
lock_path = auth_file.with_name(auth_file.name + ".gptimg.lock")
|
||||
fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
|
||||
with os.fdopen(fd, "r+b") as lock:
|
||||
if fcntl is None:
|
||||
yield
|
||||
return
|
||||
deadline = time.monotonic() + min(timeout, 30.0)
|
||||
while True:
|
||||
try:
|
||||
fcntl.flock(lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
break
|
||||
except (BlockingIOError, OSError) as exc:
|
||||
if time.monotonic() >= deadline:
|
||||
raise CliError("等待另一进程刷新凭证超时,请重试。") from exc
|
||||
time.sleep(0.05)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
fcntl.flock(lock.fileno(), fcntl.LOCK_UN)
|
||||
|
||||
|
||||
def _post_json(
|
||||
url: str, *, headers: dict[str, str], payload: dict[str, Any], timeout: float
|
||||
) -> dict[str, Any]:
|
||||
body = json.dumps(payload).encode("utf-8")
|
||||
req = request.Request(url, data=body, headers=headers, method="POST")
|
||||
try:
|
||||
with request.urlopen(req, timeout=timeout) as response:
|
||||
raw = response.read().decode("utf-8")
|
||||
except error.HTTPError as exc:
|
||||
raw = exc.read().decode("utf-8", errors="replace")
|
||||
raise HttpError(exc.code, raw, exc.headers.get("Retry-After") if exc.headers else None) from exc
|
||||
except (error.URLError, OSError) as exc:
|
||||
raise CliError(f"网络请求失败: {exc}") from exc
|
||||
if not raw:
|
||||
return {}
|
||||
try:
|
||||
parsed = json.loads(raw)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise CliError("响应不是合法 JSON。") from exc
|
||||
return parsed if isinstance(parsed, dict) else {}
|
||||
|
||||
|
||||
def _refresh_auth(auth: dict[str, Any], auth_file: Path, *, timeout: float) -> dict[str, Any]:
|
||||
import copy
|
||||
|
||||
with _refresh_lock(auth_file, timeout):
|
||||
current = _load_auth(auth_file) if auth_file.exists() else auth
|
||||
# 别的进程刚刷新过就直接用新的
|
||||
if (
|
||||
current is not auth
|
||||
and current.get("tokens") != auth.get("tokens")
|
||||
and not _token_is_expiring(_access_token(current))
|
||||
):
|
||||
return current
|
||||
updated = copy.deepcopy(current)
|
||||
tokens = _tokens(updated)
|
||||
refresh_token = tokens.get("refresh_token")
|
||||
if not isinstance(refresh_token, str) or not refresh_token:
|
||||
raise CliError("缺少 refresh_token,请重新 `codex login`。")
|
||||
try:
|
||||
refreshed = _post_json(
|
||||
REFRESH_URL,
|
||||
headers={"Accept": "application/json", "Content-Type": "application/json"},
|
||||
payload={
|
||||
"client_id": CODEX_CLIENT_ID,
|
||||
"grant_type": "refresh_token",
|
||||
"refresh_token": refresh_token,
|
||||
},
|
||||
timeout=timeout,
|
||||
)
|
||||
except HttpError as exc:
|
||||
# OAuth 错误体可能带凭证内容,绝不回显
|
||||
raise CliError(f"刷新令牌失败(HTTP {exc.status}),请重新 `codex login`。") from exc
|
||||
access = refreshed.get("access_token")
|
||||
if not isinstance(access, str) or not access:
|
||||
raise CliError("刷新响应里没有 access_token,凭证文件未改动。")
|
||||
for key in ("access_token", "refresh_token", "id_token"):
|
||||
value = refreshed.get(key)
|
||||
if isinstance(value, str) and value:
|
||||
tokens[key] = value
|
||||
if not isinstance(tokens.get("id_token"), str) or not tokens["id_token"]:
|
||||
raise CliError("刷新后 id_token 无效,请重新 `codex login`。")
|
||||
updated["last_refresh"] = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
||||
_atomic_write(auth_file, (json.dumps(updated, indent=2) + "\n").encode(), replace=True)
|
||||
return updated
|
||||
|
||||
|
||||
def _load_ready_auth(args: argparse.Namespace) -> tuple[dict[str, Any], Path]:
|
||||
auth_file = _auth_file(args)
|
||||
auth = _load_auth(auth_file)
|
||||
if _token_is_expiring(_access_token(auth)):
|
||||
auth = _refresh_auth(auth, auth_file, timeout=args.timeout)
|
||||
return auth, auth_file
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 请求体
|
||||
|
||||
def _mime_and_bytes(path: Path) -> tuple[bytes, str]:
|
||||
raw = path.read_bytes()
|
||||
if raw.startswith(b"\x89PNG\r\n\x1a\n"):
|
||||
mime = "image/png"
|
||||
elif raw.startswith(b"\xff\xd8\xff"):
|
||||
mime = "image/jpeg"
|
||||
elif raw[0:4] == b"RIFF" and raw[8:12] == b"WEBP":
|
||||
mime = "image/webp"
|
||||
elif raw.startswith((b"GIF87a", b"GIF89a")):
|
||||
mime = "image/gif"
|
||||
else:
|
||||
raise CliError(f"无法识别图片格式(支持 PNG/JPEG/WebP/GIF): {path}")
|
||||
return raw, mime
|
||||
|
||||
|
||||
def _data_url_for_image(path: Path, max_edge: int) -> str:
|
||||
if not path.is_file():
|
||||
raise CliError(f"输入图片不存在: {path}")
|
||||
try: # Pillow 可选增强:压到最长边 max_edge 的 WebP,省流量
|
||||
from io import BytesIO
|
||||
|
||||
from PIL import Image, ImageOps # type: ignore[import-not-found]
|
||||
|
||||
with Image.open(path) as source:
|
||||
source.load()
|
||||
image = ImageOps.exif_transpose(source)
|
||||
width, height = image.size
|
||||
edge = max(width, height)
|
||||
if 0 < max_edge < edge:
|
||||
scale = max_edge / float(edge)
|
||||
image = image.resize(
|
||||
(max(1, round(width * scale)), max(1, round(height * scale))),
|
||||
Image.Resampling.LANCZOS,
|
||||
)
|
||||
if image.mode not in {"RGB", "RGBA"}:
|
||||
image = image.convert("RGBA" if "A" in image.getbands() else "RGB")
|
||||
output = BytesIO()
|
||||
image.save(output, "WEBP", quality=90)
|
||||
raw, mime = output.getvalue(), "image/webp"
|
||||
except ImportError:
|
||||
raw, mime = _mime_and_bytes(path)
|
||||
except OSError as exc:
|
||||
raise CliError(f"读取图片失败 {path}: {exc}") from exc
|
||||
encoded = base64.b64encode(raw).decode("ascii")
|
||||
return f"data:{mime};base64,{encoded}"
|
||||
|
||||
|
||||
STYLE_TRANSFER_PROMPT = (
|
||||
"Create a new version of the first input image. Preserve its main subject, identity, "
|
||||
"composition, proportions, and important details. Use the final input image only as a style "
|
||||
"reference: apply its visual medium, rendering approach, color palette, lighting, texture, "
|
||||
"finish, and mood. Do not copy the style reference's subject or scene content."
|
||||
)
|
||||
|
||||
|
||||
def _native_payload(
|
||||
*, prompt: str, args: argparse.Namespace, mode: str, images: Sequence[Path] | None
|
||||
) -> dict[str, Any]:
|
||||
payload: dict[str, Any] = {
|
||||
"model": NATIVE_MODEL,
|
||||
"prompt": prompt,
|
||||
"size": args.size,
|
||||
"quality": args.quality,
|
||||
"background": args.background,
|
||||
}
|
||||
if mode == "edit":
|
||||
if not images or len(images) > MAX_EDIT_IMAGES:
|
||||
raise CliError(f"改图需要 1–{MAX_EDIT_IMAGES} 张输入图。")
|
||||
payload["images"] = [
|
||||
{"image_url": _data_url_for_image(path, args.input_max_edge)} for path in images
|
||||
]
|
||||
return payload
|
||||
|
||||
|
||||
def _reasoning_model(args: argparse.Namespace) -> str:
|
||||
if args.model:
|
||||
return args.model
|
||||
if os.environ.get("CODEX_IMAGEGEN_MODEL"):
|
||||
return os.environ["CODEX_IMAGEGEN_MODEL"]
|
||||
config_file = _codex_home(args) / "config.toml"
|
||||
if tomllib is not None and config_file.exists():
|
||||
try:
|
||||
config = tomllib.loads(config_file.read_text(encoding="utf-8"))
|
||||
model = config.get("model")
|
||||
if isinstance(model, str) and model.strip():
|
||||
return model
|
||||
except (OSError, UnicodeError, Exception):
|
||||
pass
|
||||
return DEFAULT_REASONING_MODEL
|
||||
|
||||
|
||||
def _responses_payload(
|
||||
*, prompt: str, args: argparse.Namespace, mode: str, images: Sequence[Path] | None
|
||||
) -> dict[str, Any]:
|
||||
content: list[dict[str, Any]] = [{"type": "input_text", "text": prompt}]
|
||||
if images:
|
||||
if len(images) > MAX_EDIT_IMAGES:
|
||||
raise CliError(f"改图最多 {MAX_EDIT_IMAGES} 张输入图。")
|
||||
for image in images:
|
||||
content.append(
|
||||
{"type": "input_image", "image_url": _data_url_for_image(image, args.input_max_edge)}
|
||||
)
|
||||
instructions = (
|
||||
"Use the available image generation tool to generate exactly one PNG image for the "
|
||||
"user request. Do not use any other tool."
|
||||
)
|
||||
if mode == "edit":
|
||||
instructions += " Treat the provided input images as edit/reference images for the request."
|
||||
return {
|
||||
"model": _reasoning_model(args),
|
||||
"instructions": instructions,
|
||||
"input": [{"type": "message", "role": "user", "content": content}],
|
||||
"tools": [
|
||||
{
|
||||
"type": "image_generation",
|
||||
"output_format": "png",
|
||||
"size": args.size,
|
||||
"quality": args.quality,
|
||||
"background": args.background,
|
||||
}
|
||||
],
|
||||
"tool_choice": {"type": "image_generation"},
|
||||
"parallel_tool_calls": False,
|
||||
"store": False,
|
||||
"stream": True,
|
||||
"prompt_cache_key": "le-gpt-image",
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 调用与落盘
|
||||
|
||||
def _native_result(
|
||||
url: str, *, headers: dict[str, str], payload: dict[str, Any], timeout: float
|
||||
) -> bytes:
|
||||
response = _post_json(url, headers=headers, payload=payload, timeout=timeout)
|
||||
data = response.get("data")
|
||||
if not isinstance(data, list) or len(data) != 1 or not isinstance(data[0], dict):
|
||||
error_obj = response.get("error")
|
||||
if isinstance(error_obj, dict):
|
||||
raise CliError(f"生图失败: {error_obj.get('message') or error_obj}")
|
||||
raise CliError("响应里没有图像数据。")
|
||||
encoded = data[0].get("b64_json")
|
||||
if not isinstance(encoded, str) or not encoded:
|
||||
raise CliError("响应里的图像数据为空。")
|
||||
return base64.b64decode(encoded)
|
||||
|
||||
|
||||
def _sse_data_lines(url: str, headers: dict[str, str], payload: dict[str, Any], timeout: float):
|
||||
req_headers = {**headers, "Accept": "text/event-stream", "Content-Type": "application/json"}
|
||||
req = request.Request(
|
||||
url, data=json.dumps(payload).encode("utf-8"), headers=req_headers, method="POST"
|
||||
)
|
||||
try:
|
||||
with request.urlopen(req, timeout=timeout) as response:
|
||||
buf = ""
|
||||
while True:
|
||||
chunk = response.read(4096)
|
||||
if not chunk:
|
||||
break
|
||||
buf += chunk.decode("utf-8", "replace")
|
||||
while "\n" in buf:
|
||||
line, buf = buf.split("\n", 1)
|
||||
stripped = line.strip()
|
||||
if stripped.startswith("data:"):
|
||||
data = stripped[5:].strip()
|
||||
if data and data != "[DONE]":
|
||||
yield data
|
||||
tail = buf.strip()
|
||||
if tail.startswith("data:"):
|
||||
data = tail[5:].strip()
|
||||
if data and data != "[DONE]":
|
||||
yield data
|
||||
except error.HTTPError as exc:
|
||||
raw = exc.read().decode("utf-8", errors="replace")
|
||||
raise HttpError(exc.code, raw, exc.headers.get("Retry-After") if exc.headers else None) from exc
|
||||
except (error.URLError, OSError) as exc:
|
||||
raise CliError(f"网络请求失败: {exc}") from exc
|
||||
|
||||
|
||||
def _responses_result(
|
||||
url: str, *, headers: dict[str, str], payload: dict[str, Any], timeout: float
|
||||
) -> bytes:
|
||||
last_status: str | None = None
|
||||
for data in _sse_data_lines(url, headers, payload, timeout):
|
||||
try:
|
||||
event = json.loads(data)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise CliError("响应流里包含非法 JSON。") from exc
|
||||
if not isinstance(event, dict):
|
||||
continue
|
||||
if event.get("type") in {"response.failed", "response.incomplete", "error"}:
|
||||
error_obj = event.get("error") or (event.get("response") or {}).get("error")
|
||||
message = error_obj.get("message") if isinstance(error_obj, dict) else None
|
||||
raise CliError(f"responses 通道生图失败: {message or event}")
|
||||
items: list[Any] = [event.get("item")]
|
||||
if event.get("type") == "response.completed":
|
||||
response_obj = event.get("response")
|
||||
if isinstance(response_obj, dict) and isinstance(response_obj.get("output"), list):
|
||||
items.extend(response_obj["output"])
|
||||
for item in items:
|
||||
if isinstance(item, dict) and item.get("type") == "image_generation_call":
|
||||
last_status = item.get("status")
|
||||
result = item.get("result")
|
||||
if isinstance(result, str) and result and last_status == "completed":
|
||||
return base64.b64decode(result)
|
||||
if last_status:
|
||||
raise CliError(f"响应流结束但没有拿到图像(最后状态: {last_status})。")
|
||||
raise CliError("响应流里没有图像生成结果。")
|
||||
|
||||
|
||||
def _retry_decision(exc: HttpError, attempt: int) -> float | None:
|
||||
"""只在明确限流时重试;传输类/5xx 故意不重放——图像可能已消耗额度。"""
|
||||
if exc.status != 429:
|
||||
return None
|
||||
delay = min(2.0**attempt, 16.0)
|
||||
if exc.retry_after:
|
||||
try:
|
||||
wait = float(exc.retry_after)
|
||||
except ValueError:
|
||||
wait = 0.0
|
||||
delay = max(delay, wait)
|
||||
return delay if math.isfinite(delay) and delay <= 300 else None
|
||||
|
||||
|
||||
def _png_size(image_bytes: bytes) -> tuple[int, int] | None:
|
||||
if image_bytes[:8] != b"\x89PNG\r\n\x1a\n" or len(image_bytes) < 24:
|
||||
return None
|
||||
width = int.from_bytes(image_bytes[16:20], "big")
|
||||
height = int.from_bytes(image_bytes[20:24], "big")
|
||||
return width, height
|
||||
|
||||
|
||||
def _write_image(image_bytes: bytes, out_path: Path, *, force: bool, requested_size: str) -> None:
|
||||
actual = _png_size(image_bytes)
|
||||
if actual is None:
|
||||
raise CliError("服务端返回的不是有效 PNG 图像。")
|
||||
_atomic_write(out_path, image_bytes, replace=force)
|
||||
if requested_size != "auto":
|
||||
requested = requested_size.lower().replace("x", "×")
|
||||
actual_text = f"{actual[0]}x{actual[1]}"
|
||||
if requested_size != actual_text:
|
||||
_warn(f"请求尺寸 {requested_size},服务端实际返回 {actual_text},已按实际尺寸保存。")
|
||||
_log(f" 已保存 {actual[0]}x{actual[1]} PNG")
|
||||
|
||||
|
||||
def _call_and_save(
|
||||
*, args: argparse.Namespace, url: str, payload: dict[str, Any], native: bool, out_path: Path
|
||||
) -> None:
|
||||
if args.dry_run:
|
||||
preview = json.dumps(
|
||||
{
|
||||
"method": "POST",
|
||||
"url": url,
|
||||
"headers": {"Authorization": "Bearer <redacted>"},
|
||||
"payload": _redact_payload(payload),
|
||||
"outputs": [str(out_path)],
|
||||
},
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
)
|
||||
print(preview)
|
||||
return
|
||||
auth, auth_file = _load_ready_auth(args)
|
||||
headers = _auth_headers(auth)
|
||||
call = _native_result if native else _responses_result
|
||||
attempt = 0
|
||||
refreshed_401 = False
|
||||
t0 = time.monotonic()
|
||||
while True:
|
||||
try:
|
||||
image_bytes = call(url, headers=headers, payload=payload, timeout=args.timeout)
|
||||
break
|
||||
except HttpError as exc:
|
||||
if exc.status == 401 and not refreshed_401:
|
||||
auth = _refresh_auth(auth, auth_file, timeout=args.timeout)
|
||||
headers = _auth_headers(auth)
|
||||
refreshed_401 = True
|
||||
continue
|
||||
delay = _retry_decision(exc, attempt)
|
||||
if delay is None or attempt >= MAX_RETRIES:
|
||||
raise
|
||||
attempt += 1
|
||||
_warn(f"被限流,{delay:.0f}s 后重试({attempt}/{MAX_RETRIES})")
|
||||
time.sleep(delay)
|
||||
_write_image(image_bytes, out_path, force=args.force, requested_size=args.size)
|
||||
_log(f" 完成({time.monotonic() - t0:.1f}s)")
|
||||
|
||||
|
||||
def _redact_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||||
def redact(value: Any) -> Any:
|
||||
if isinstance(value, dict):
|
||||
return {
|
||||
key: (f"{str(item).split(';base64,', 1)[0]};base64,<redacted>"
|
||||
if key == "image_url" and isinstance(item, str)
|
||||
else redact(item))
|
||||
for key, item in value.items()
|
||||
}
|
||||
if isinstance(value, list):
|
||||
return [redact(item) for item in value]
|
||||
return value
|
||||
|
||||
return redact(payload)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 参数与入口
|
||||
|
||||
def _parse_size(value: str) -> str:
|
||||
value = value.strip().lower()
|
||||
if value in SIZE_ALIASES:
|
||||
return SIZE_ALIASES[value]
|
||||
if value == "auto":
|
||||
return value
|
||||
match = re.fullmatch(r"([1-9][0-9]{0,3})x([1-9][0-9]{0,3})", value)
|
||||
if not match:
|
||||
raise argparse.ArgumentTypeError(
|
||||
f"尺寸须为 auto、别名({'/'.join(SIZE_ALIASES)})或 WIDTHxHEIGHT,收到: {value}"
|
||||
)
|
||||
width, height = map(int, match.groups())
|
||||
if not (
|
||||
width % 16 == height % 16 == 0
|
||||
and max(width, height) <= 3840
|
||||
and max(width, height) <= 3 * min(width, height)
|
||||
and 655360 <= width * height <= 8294400
|
||||
):
|
||||
raise argparse.ArgumentTypeError(
|
||||
"尺寸约束不满足: 宽高为 16 的倍数、最长边 ≤3840、宽高比 1:3–3:1、"
|
||||
"总像素 655360–8294400。"
|
||||
)
|
||||
return value
|
||||
|
||||
|
||||
def _slugify(text: str) -> str:
|
||||
value = re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-")
|
||||
return value[:60] or "image"
|
||||
|
||||
|
||||
def _default_out(prompt: str) -> Path:
|
||||
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
|
||||
return Path("fig") / f"{stamp}-{_slugify(prompt)}.png"
|
||||
|
||||
|
||||
def _resolve_path(raw: str, cd: Path) -> Path:
|
||||
path = Path(raw).expanduser()
|
||||
if not path.is_absolute():
|
||||
path = cd / path
|
||||
return path.resolve()
|
||||
|
||||
|
||||
def _read_prompt(args: argparse.Namespace) -> str | None:
|
||||
if args.prompt is not None and args.prompt_file is not None:
|
||||
raise CliError("--prompt 与 --prompt-file 只能二选一。")
|
||||
if args.prompt_file is not None:
|
||||
text = Path(args.prompt_file).read_text(encoding="utf-8").strip()
|
||||
elif args.prompt is not None:
|
||||
text = args.prompt.strip()
|
||||
else:
|
||||
return None
|
||||
if not text:
|
||||
raise CliError("prompt 不能为空。")
|
||||
return text
|
||||
|
||||
|
||||
def _validate_common(args: argparse.Namespace) -> Path:
|
||||
if not math.isfinite(args.timeout) or args.timeout <= 0:
|
||||
raise CliError("--timeout 须为正数。")
|
||||
if args.n < 1:
|
||||
raise CliError("--n 至少为 1。")
|
||||
if args.input_max_edge < 0:
|
||||
raise CliError("--input-max-edge 须 ≥0。")
|
||||
if args.background == "transparent":
|
||||
raise CliError("当前 Codex 图像通道不支持透明背景,请用 auto 或 opaque。")
|
||||
if args.backend == "native" and getattr(args, "model", None):
|
||||
raise CliError("--model 只在 --backend responses 下有效(它选的是推理模型,不是图像模型)。")
|
||||
cd = Path(getattr(args, "cd", None) or os.getcwd()).expanduser().resolve()
|
||||
if not cd.is_dir():
|
||||
raise CliError(f"--cd 目录不存在: {cd}")
|
||||
return cd
|
||||
|
||||
|
||||
def _output_paths(out_path: Path, count: int) -> list[Path]:
|
||||
if count <= 1:
|
||||
return [out_path]
|
||||
return [
|
||||
out_path.with_name(f"{out_path.stem}-{idx}{out_path.suffix or '.png'}")
|
||||
for idx in range(1, count + 1)
|
||||
]
|
||||
|
||||
|
||||
def _run_jobs(
|
||||
*, args: argparse.Namespace, mode: str, prompt: str, out_path: Path,
|
||||
images: Sequence[Path] | None,
|
||||
) -> None:
|
||||
paths = _output_paths(out_path, args.n)
|
||||
if not args.dry_run:
|
||||
for path in paths:
|
||||
if path.exists() and not args.force:
|
||||
raise CliError(f"输出文件已存在: {path}(加 --force 才允许覆盖)")
|
||||
count_text = f"{args.n}× " if args.n > 1 else ""
|
||||
_log(f"{mode} {count_text}{out_path}")
|
||||
if args.backend == "native":
|
||||
url = args.base_url.rstrip("/") + ("/images/edits" if mode == "edit" else "/images/generations")
|
||||
payload = _native_payload(prompt=prompt, args=args, mode=mode, images=images)
|
||||
native = True
|
||||
else:
|
||||
url = args.base_url.rstrip("/") + "/responses"
|
||||
payload = _responses_payload(prompt=prompt, args=args, mode=mode, images=images)
|
||||
native = False
|
||||
for path in paths:
|
||||
_call_and_save(args=args, url=url, payload=payload, native=native, out_path=path)
|
||||
print(str(path), flush=True) # stdout 输出最终路径,供 agent 解析
|
||||
|
||||
|
||||
def _cmd_generate(args: argparse.Namespace) -> int:
|
||||
cd = _validate_common(args)
|
||||
prompt = _read_prompt(args)
|
||||
if prompt is None:
|
||||
raise CliError("缺少 prompt,用 --prompt 或 --prompt-file。")
|
||||
out_path = _resolve_path(args.out, cd) if args.out else (cd / _default_out(prompt))
|
||||
_run_jobs(args=args, mode="generate", prompt=prompt, out_path=out_path, images=None)
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_edit(args: argparse.Namespace) -> int:
|
||||
cd = _validate_common(args)
|
||||
prompt = _read_prompt(args)
|
||||
if args.style_image:
|
||||
extra = f"\n\nAdditional instruction: {prompt}" if prompt else ""
|
||||
prompt = STYLE_TRANSFER_PROMPT + extra
|
||||
elif prompt is None:
|
||||
raise CliError("缺少 prompt,用 --prompt、--prompt-file 或 --style-image。")
|
||||
out_path = _resolve_path(args.out, cd) if args.out else (cd / _default_out(prompt))
|
||||
images = [_resolve_path(raw, cd) for raw in args.image]
|
||||
if args.style_image:
|
||||
images.append(_resolve_path(args.style_image, cd))
|
||||
_run_jobs(args=args, mode="edit", prompt=prompt, out_path=out_path, images=images)
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_check(args: argparse.Namespace) -> int:
|
||||
auth_file = _auth_file(args)
|
||||
auth = _load_auth(auth_file)
|
||||
print(f"凭证文件: {auth_file}")
|
||||
print(f"登录方式: {auth.get('auth_mode', '未知')}")
|
||||
if auth.get("auth_mode") and auth["auth_mode"] != "chatgpt":
|
||||
raise CliError("当前不是 ChatGPT 登录态(可能是 API Key 模式),请 `codex login` 选 ChatGPT。")
|
||||
tokens = _tokens(auth)
|
||||
account_id = _account_id(auth)
|
||||
print(f"账号 ID: {account_id[:8]}…(已截断)" if account_id else "账号 ID: 未找到")
|
||||
plan = _id_token_claims(auth).get("chatgpt_plan_type")
|
||||
print(f"套餐类型: {plan or '未知'}")
|
||||
exp = _jwt_payload(_access_token(auth)).get("exp")
|
||||
if isinstance(exp, (int, float)):
|
||||
hours = (exp - datetime.now(timezone.utc).timestamp()) / 3600
|
||||
print(f"access_token: 剩余 {hours:.1f} 小时")
|
||||
else:
|
||||
print("access_token: 无法解析有效期")
|
||||
print(f"refresh_token: {'有' if tokens.get('refresh_token') else '无'}")
|
||||
if args.refresh:
|
||||
_refresh_auth(auth, auth_file, timeout=args.timeout)
|
||||
print("已刷新令牌并写回凭证文件。")
|
||||
else:
|
||||
print("提示: 加 --refresh 可强制刷新令牌。")
|
||||
return 0
|
||||
|
||||
|
||||
def _add_common_args(parser: argparse.ArgumentParser, *, require_out: bool) -> None:
|
||||
parser.add_argument("--prompt")
|
||||
parser.add_argument("--prompt-file")
|
||||
parser.add_argument("--out", required=require_out, help="输出路径;缺省为 fig/时间戳-slug.png")
|
||||
parser.add_argument("--force", action="store_true", help="允许覆盖已存在的输出文件")
|
||||
parser.add_argument(
|
||||
"--size", type=_parse_size, default="auto",
|
||||
help="尺寸: auto / 别名(1k 2k 4k square portrait landscape wide tall) / WIDTHxHEIGHT",
|
||||
)
|
||||
parser.add_argument("--quality", choices=["auto", "low", "medium", "high"], default="auto")
|
||||
parser.add_argument("--background", choices=["auto", "opaque"], default="auto")
|
||||
parser.add_argument("--n", type=int, default=1, help="生成张数(逐张请求,文件名加 -1..-n)")
|
||||
parser.add_argument(
|
||||
"--backend", choices=["native", "responses"], default="native",
|
||||
help="native=GPT Image 直连通道(默认);responses=走推理模型的备用通道",
|
||||
)
|
||||
parser.add_argument("--model", help="responses 通道的推理模型(默认读 Codex 配置或 gpt-5.5)")
|
||||
parser.add_argument("--base-url", default=os.environ.get("CODEX_IMAGEGEN_BASE_URL", DEFAULT_BASE_URL))
|
||||
parser.add_argument("--timeout", type=float, default=300.0)
|
||||
parser.add_argument("--auth-file", help="自定义 auth.json 路径")
|
||||
parser.add_argument("--codex-home", help="自定义 Codex 目录(默认 $CODEX_HOME 或 ~/.codex)")
|
||||
parser.add_argument("--cd", help="相对路径的基准目录")
|
||||
parser.add_argument("--input-max-edge", type=int, default=INPUT_MAX_EDGE,
|
||||
help="改图输入压缩的最长边像素,0 为不压缩")
|
||||
parser.add_argument("--dry-run", action="store_true", help="只打印请求(凭证打码),不联网不生图")
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="gpt_image.py",
|
||||
description="用本机 Codex ChatGPT 登录调 GPT Image 生图/改图(无需 API Key)",
|
||||
)
|
||||
parser.add_argument("--version", "-v", action="version", version=f"%(prog)s {__version__}")
|
||||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||||
|
||||
generate = subparsers.add_parser("generate", help="文生图")
|
||||
_add_common_args(generate, require_out=False)
|
||||
generate.set_defaults(func=_cmd_generate)
|
||||
|
||||
edit = subparsers.add_parser("edit", help="参考图改图 / 风格迁移")
|
||||
_add_common_args(edit, require_out=False)
|
||||
edit.add_argument("--image", action="append", required=True,
|
||||
help="输入图片路径,可重复(1–5 张)")
|
||||
edit.add_argument("--style-image",
|
||||
help="风格参考图:内容取 --image,画风取此图(此时 --prompt 变为可选补充)")
|
||||
edit.set_defaults(func=_cmd_edit)
|
||||
|
||||
check = subparsers.add_parser("check", help="检查/刷新本机凭证状态")
|
||||
check.add_argument("--refresh", action="store_true", help="强制刷新令牌")
|
||||
check.add_argument("--timeout", type=float, default=60.0)
|
||||
check.add_argument("--auth-file")
|
||||
check.add_argument("--codex-home")
|
||||
check.set_defaults(func=_cmd_check)
|
||||
|
||||
return parser
|
||||
|
||||
|
||||
def main(argv: Sequence[str] | None = None) -> int:
|
||||
args = build_parser().parse_args(argv)
|
||||
try:
|
||||
return int(args.func(args))
|
||||
except CliError as exc:
|
||||
print(f"错误: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
except (OSError, UnicodeError) as exc:
|
||||
print(f"错误: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
except KeyboardInterrupt:
|
||||
print("已中断。", file=sys.stderr)
|
||||
return 130
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user