AI/gpt-image-2-generator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(project): init

Browse Source
This commit is contained in:
shay7sev
2026-04-27 14:33:50 +08:00
commit 3d74b28f26
12 changed files with 2789 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
scripts/__pycache__/generate_image.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
scripts/__pycache__/runtime_config.cpython-312.pyc Normal file
View File

Binary file not shown.

BIN
scripts/__pycache__/web_app.cpython-312.pyc Normal file
View File

Binary file not shown.

114
scripts/export_share_package.ps1 Normal file
View File

@@ -0,0 +1,114 @@
param(
[string]$OutputDir,
[string]$PackageName
)
$ErrorActionPreference = 'Stop'
$scriptDir = Split-Path -Parent $MyInvocation.MyCommand.Path
$projectRoot = (Resolve-Path (Join-Path $scriptDir '..')).Path
$projectName = Split-Path $projectRoot -Leaf
$shareBaseName = "gpt-image-2-generator"
$timestamp = Get-Date -Format 'yyyyMMdd-HHmmss'
if ([string]::IsNullOrWhiteSpace($OutputDir)) {
$OutputDir = Join-Path $projectRoot 'dist'
}
$OutputDir = [System.IO.Path]::GetFullPath($OutputDir)
if ([string]::IsNullOrWhiteSpace($PackageName)) {
$PackageName = "$shareBaseName-share-$timestamp"
}
$zipPath = Join-Path $OutputDir ($PackageName + '.zip')
$tempRoot = Join-Path ([System.IO.Path]::GetTempPath()) ("$shareBaseName-share-export-" + [System.Guid]::NewGuid().ToString('N'))
$stageDir = Join-Path $tempRoot $PackageName
$excludedTopDirs = @('outputs', 'dist', '.git', '.idea', '.vscode')
$excludedFileNames = @('history.jsonl')
$excludedExtensions = @('.pyc', '.pyo')
function Should-SkipFile {
param([string]$RelativePath)
$normalized = $RelativePath -replace '/', '\\'
$parts = $normalized.Split('\\', [System.StringSplitOptions]::RemoveEmptyEntries)
if ($parts.Count -eq 0) {
return $true
}
if ($excludedTopDirs -contains $parts[0]) {
return $true
}
if ($parts -contains '__pycache__') {
return $true
}
if ($normalized -ieq 'config\token.txt' -or $normalized -ieq 'config\base_url.txt') {
return $true
}
$fileName = [System.IO.Path]::GetFileName($normalized)
if ($excludedFileNames -contains $fileName) {
return $true
}
$extension = [System.IO.Path]::GetExtension($normalized)
if ($excludedExtensions -contains $extension) {
return $true
}
return $false
}
function Ensure-ParentDirectory {
param([string]$FilePath)
$parent = Split-Path -Parent $FilePath
if (-not [string]::IsNullOrWhiteSpace($parent) -and -not (Test-Path $parent)) {
New-Item -ItemType Directory -Path $parent -Force | Out-Null
}
}
try {
New-Item -ItemType Directory -Path $OutputDir -Force | Out-Null
New-Item -ItemType Directory -Path $stageDir -Force | Out-Null
$files = Get-ChildItem -Path $projectRoot -Recurse -File
foreach ($file in $files) {
$relativePath = $file.FullName.Substring($projectRoot.Length).TrimStart('\')
if (Should-SkipFile -RelativePath $relativePath) {
continue
}
$destination = Join-Path $stageDir $relativePath
Ensure-ParentDirectory -FilePath $destination
Copy-Item -Path $file.FullName -Destination $destination -Force
}
$placeholderTokenPath = Join-Path $stageDir 'config\token.txt'
Ensure-ParentDirectory -FilePath $placeholderTokenPath
@(
'# 在下一行填入 apicodex.xyz 的 API token，不要带 Bearer 前缀。',
'YOUR_API_TOKEN_HERE'
) | Set-Content -Path $placeholderTokenPath -Encoding UTF8
$placeholderBaseUrlPath = Join-Path $stageDir 'config\base_url.txt'
Ensure-ParentDirectory -FilePath $placeholderBaseUrlPath
@(
'# 在下一行填入 API base URL，可写根地址或完整 /v1 地址。',
'https://apicodex.xyz/v1'
) | Set-Content -Path $placeholderBaseUrlPath -Encoding UTF8
if (Test-Path $zipPath) {
Remove-Item -Path $zipPath -Force
}
Compress-Archive -Path $stageDir -DestinationPath $zipPath -Force
Write-Host '可分享版导出完成：' $zipPath
Write-Host '已自动排除：config\token.txt（真实内容）、config\base_url.txt（真实内容）、outputs、history.jsonl、__pycache__、*.pyc、dist。'
}
finally {
if (Test-Path $tempRoot) {
Remove-Item -Path $tempRoot -Recurse -Force
}
}

992
scripts/generate_image.py Normal file
View File

@@ -0,0 +1,992 @@
from __future__ import annotations
import argparse
import base64
import binascii
from dataclasses import dataclass
import json
import mimetypes
import os
import re
import socket
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any
from runtime_config import (
ConfigError,
DEFAULT_BASE_URL,
DEFAULT_MODEL,
ROOT_DIR,
build_api_url,
infer_request_style,
load_runtime_config,
normalize_base_url,
normalize_token,
parse_env_file,
runtime_config_summary,
save_base_url as save_runtime_base_url,
save_token as save_runtime_token,
)
SUPPORTED_ASPECT_RATIOS = (
"1:1",
"16:9",
"9:16",
"4:3",
"3:4",
"3:2",
"2:3",
"5:4",
"4:5",
"2:1",
"1:2",
"21:9",
"9:21",
)
PIXEL_SIZE_PATTERN = re.compile(r"^(?P<width>\d{2,5})x(?P<height>\d{2,5})$")
OPENAI_RATIO_SIZE_MAP = {
"1:1": "1024x1024",
"16:9": "1536x864",
"9:16": "864x1536",
"4:3": "1536x1152",
"3:4": "1152x1536",
"3:2": "1536x1024",
"2:3": "1024x1536",
"5:4": "1280x1024",
"4:5": "1024x1280",
"2:1": "1536x768",
"1:2": "768x1536",
"21:9": "1680x720",
"9:21": "720x1680",
}
SIZE_ENV_VARS = (
"GPT_IMAGE_2_SIZE",
"IMAGE_SIZE",
"OPENAI_IMAGE_SIZE",
)
EDIT_SIZE_ENV_VARS = (
"GPT_IMAGE_2_EDIT_SIZE",
"IMAGE_EDIT_SIZE",
"OPENAI_IMAGE_EDIT_SIZE",
)
EDIT_QUALITY_ENV_VARS = (
"GPT_IMAGE_2_EDIT_QUALITY",
"IMAGE_EDIT_QUALITY",
"OPENAI_IMAGE_EDIT_QUALITY",
)
EDIT_BACKGROUND_ENV_VARS = (
"GPT_IMAGE_2_EDIT_BACKGROUND",
"IMAGE_EDIT_BACKGROUND",
"OPENAI_IMAGE_EDIT_BACKGROUND",
)
EDIT_OUTPUT_FORMAT_ENV_VARS = (
"GPT_IMAGE_2_EDIT_OUTPUT_FORMAT",
"IMAGE_EDIT_OUTPUT_FORMAT",
"OPENAI_IMAGE_EDIT_OUTPUT_FORMAT",
)
EDIT_INPUT_FIDELITY_ENV_VARS = (
"GPT_IMAGE_2_EDIT_INPUT_FIDELITY",
"IMAGE_EDIT_INPUT_FIDELITY",
"OPENAI_IMAGE_EDIT_INPUT_FIDELITY",
)
REQUEST_STYLE_CHOICES = ("auto", "apicodex", "openai")
REQUEST_RETRY_ATTEMPTS = max(1, int(os.environ.get("IMAGE_RETRY_ATTEMPTS", "2")))
REQUEST_RETRY_DELAY_SECONDS = max(0.0, float(os.environ.get("IMAGE_RETRY_DELAY_SECONDS", "2")))
class ImageGenerationError(RuntimeError):
pass
class ImageRequestError(ImageGenerationError):
def __init__(self, message: str, *, status: int | None = None, timeout: bool = False):
super().__init__(message)
self.status = status
self.timeout = timeout
@dataclass(frozen=True)
class ImageInputFile:
filename: str
mime_type: str
data: bytes
@dataclass(frozen=True)
class RequestTarget:
name: str
base_url: str
token: str
request_style: str
source: str
def normalize_size_value(raw_size: str) -> str:
value = raw_size.strip().lower().replace(" ", "")
match = PIXEL_SIZE_PATTERN.fullmatch(value)
if match:
width = int(match.group("width"))
height = int(match.group("height"))
if width < 64 or height < 64:
raise ImageGenerationError(f"Unsupported size: {raw_size}")
return f"{width}x{height}"
return value
def is_supported_size(size: str) -> bool:
return size in SUPPORTED_ASPECT_RATIOS or PIXEL_SIZE_PATTERN.fullmatch(size) is not None
def resolve_api_size(requested_size: str, request_style: str) -> str:
if request_style == "openai" and requested_size in OPENAI_RATIO_SIZE_MAP:
return OPENAI_RATIO_SIZE_MAP[requested_size]
return requested_size
def lookup_env_setting(env_map: dict[str, str], env_vars: tuple[str, ...]) -> str | None:
for env_name in env_vars:
value = os.environ.get(env_name, "").strip()
if value:
return value
for env_name in env_vars:
value = env_map.get(env_name, "").strip()
if value:
return value
return None
def resolve_requested_size(
size: str | None,
request_style: str,
has_refs: bool,
env_map: dict[str, str],
) -> tuple[str, str]:
if size and size.strip():
requested_size = normalize_size_value(size)
else:
env_vars = EDIT_SIZE_ENV_VARS if has_refs else SIZE_ENV_VARS
env_size = lookup_env_setting(env_map, env_vars)
requested_size = normalize_size_value(env_size) if env_size else "1:1"
if not is_supported_size(requested_size):
raise ImageGenerationError(f"Unsupported size: {requested_size}")
return requested_size, resolve_api_size(requested_size, request_style)
def resolve_optional_edit_field(
explicit_value: str | None,
env_map: dict[str, str],
env_vars: tuple[str, ...],
) -> str | None:
if explicit_value is not None and explicit_value.strip():
return explicit_value.strip()
value = lookup_env_setting(env_map, env_vars)
return value.strip() if value else None
def build_env_template() -> str:
return "\n".join(
[
"# Required: apicodex-compatible image API token",
"IMAGE_API_KEY=your_apicodex_token_here",
"",
"# Optional: host root or full /v1 path are both accepted",
f"IMAGE_BASE_URL={DEFAULT_BASE_URL}/v1",
"",
"# Optional: defaults shown below",
f"IMAGE_MODEL={DEFAULT_MODEL}",
"IMAGE_REQUEST_STYLE=auto",
"IMAGE_OUTPUT_DIR=./output",
"",
]
)
def write_env_template(
*,
env_file: str | Path | None = None,
overwrite: bool = False,
) -> Path:
execution_root = Path.cwd()
if env_file:
target_path = Path(env_file).expanduser()
if not target_path.is_absolute():
target_path = (execution_root / target_path).resolve()
else:
target_path = execution_root / ".env.example"
if target_path.exists() and not overwrite:
raise ConfigError(f"Config template already exists: {target_path}. Use --force to overwrite.")
target_path.parent.mkdir(parents=True, exist_ok=True)
target_path.write_text(build_env_template(), encoding="utf-8")
return target_path
def read_json_response(request: urllib.request.Request, timeout: int) -> dict[str, Any]:
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
body = response.read().decode("utf-8")
except urllib.error.HTTPError as error:
body = error.read().decode("utf-8", errors="replace")
raise ImageRequestError(f"HTTP {error.code}: {body}", status=error.code) from error
except urllib.error.URLError as error:
if is_timeout_error(error):
raise ImageRequestError(f"Timeout: {error}", timeout=True) from error
raise ImageRequestError(f"Network error: {error}") from error
except (TimeoutError, socket.timeout) as error:
raise ImageRequestError(f"Timeout: {error}", timeout=True) from error
if not body:
return {}
try:
return json.loads(body)
except json.JSONDecodeError as error:
raise ImageGenerationError(f"API returned non-JSON response: {body[:500]}") from error
def request_json(method: str, url: str, token: str, payload: dict[str, Any] | None = None, timeout: int = 300) -> dict[str, Any]:
data = None
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"User-Agent": "gpt-image-2-generator/1.0",
}
if payload is not None:
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
request = urllib.request.Request(url, data=data, headers=headers, method=method)
return read_json_response(request, timeout)
def request_multipart_json(
url: str,
token: str,
fields: dict[str, Any],
files: list[ImageInputFile],
image_field_name: str = "image[]",
timeout: int = 300,
) -> dict[str, Any]:
boundary = f"----gpt-image-2-{uuid.uuid4().hex}"
body = bytearray()
def add(value: str | bytes) -> None:
if isinstance(value, str):
body.extend(value.encode("utf-8"))
else:
body.extend(value)
for name, value in fields.items():
add(f"--{boundary}\r\n")
add(f'Content-Disposition: form-data; name="{name}"\r\n\r\n')
add(str(value))
add("\r\n")
for image_file in files:
add(f"--{boundary}\r\n")
add(
f'Content-Disposition: form-data; name="{image_field_name}"; '
f'filename="{image_file.filename}"\r\n'
)
add(f"Content-Type: {image_file.mime_type}\r\n\r\n")
add(image_file.data)
add("\r\n")
add(f"--{boundary}--\r\n")
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": f"multipart/form-data; boundary={boundary}",
"Content-Length": str(len(body)),
"User-Agent": "gpt-image-2-generator/1.0",
}
request = urllib.request.Request(url, data=bytes(body), headers=headers, method="POST")
return read_json_response(request, timeout)
def is_timeout_error(error: urllib.error.URLError) -> bool:
reason = getattr(error, "reason", None)
if isinstance(reason, (TimeoutError, socket.timeout)):
return True
return "timed out" in str(reason).lower() or "timeout" in str(error).lower()
def is_failover_eligible_error(error: BaseException) -> bool:
if isinstance(error, ImageRequestError):
if error.status in {502, 503, 504, 524}:
return True
if error.timeout:
return True
message = str(error).lower()
return "timed out" in message or "timeout" in message
def read_first_config_line(path: Path) -> str:
if not path.exists():
return ""
for raw_line in path.read_text(encoding="utf-8-sig").splitlines():
line = raw_line.strip()
if line and not line.startswith("#"):
return line
return ""
def load_skill_fallback_target(runtime_config: Any) -> RequestTarget | None:
skill_config_dir = ROOT_DIR / "config"
base_url = normalize_base_url(read_first_config_line(skill_config_dir / "base_url.txt"))
token = normalize_token(read_first_config_line(skill_config_dir / "token.txt"))
request_style_raw = read_first_config_line(skill_config_dir / "request_style.txt").strip().lower()
if not base_url or not token:
return None
request_style = request_style_raw if request_style_raw in {"apicodex", "openai"} else infer_request_style(base_url)
primary_token = normalize_token(runtime_config.token or "")
if runtime_config.base_url == base_url and primary_token == token:
return None
return RequestTarget(
name="skill-fallback",
base_url=base_url,
token=token,
request_style=request_style,
source=str(skill_config_dir),
)
def build_request_targets(runtime_config: Any, request_style: str) -> list[RequestTarget]:
targets = [
RequestTarget(
name="primary",
base_url=runtime_config.base_url,
token=runtime_config.token or "",
request_style=request_style,
source=runtime_config.base_url_source,
)
]
fallback_target = load_skill_fallback_target(runtime_config)
if fallback_target is not None:
targets.append(fallback_target)
return targets
def format_request_error(error: BaseException) -> str:
if isinstance(error, ImageRequestError):
if error.status is not None:
return f"HTTP {error.status}"
if error.timeout:
return "timeout"
return error.__class__.__name__
def should_retry_request(error: BaseException) -> bool:
if isinstance(error, ImageRequestError):
if error.status in {429, 500, 502, 503, 504, 524}:
return True
if error.timeout:
return True
message = str(error).lower()
return "timed out" in message or "timeout" in message
def submit_request_for_target(
target: RequestTarget,
*,
runtime_config: Any,
env_map: dict[str, str],
prompt: str,
size: str | None,
n: int,
has_refs: bool,
image_files: list[ImageInputFile],
timeout: int,
quality: str | None,
background: str | None,
output_format: str | None,
input_fidelity: str | None,
) -> tuple[dict[str, Any], str, str, str]:
generations_api_url = build_api_url(target.base_url, "images/generations")
edits_api_url = build_api_url(target.base_url, "images/edits")
payload: dict[str, Any] = {
"model": runtime_config.model,
"prompt": prompt,
}
requested_size, api_size = resolve_requested_size(size, target.request_style, has_refs, env_map)
payload["size"] = api_size
if target.request_style == "apicodex" or n != 1:
payload["n"] = n
if has_refs:
edit_quality = resolve_optional_edit_field(quality, env_map, EDIT_QUALITY_ENV_VARS)
edit_background = resolve_optional_edit_field(background, env_map, EDIT_BACKGROUND_ENV_VARS)
edit_output_format = resolve_optional_edit_field(output_format, env_map, EDIT_OUTPUT_FORMAT_ENV_VARS)
edit_input_fidelity = resolve_optional_edit_field(input_fidelity, env_map, EDIT_INPUT_FIDELITY_ENV_VARS)
if edit_quality:
payload["quality"] = edit_quality
if edit_background:
payload["background"] = edit_background
if edit_output_format:
payload["output_format"] = edit_output_format
if edit_input_fidelity:
payload["input_fidelity"] = edit_input_fidelity
image_field_name = "image[]" if target.request_style == "apicodex" or len(image_files) != 1 else "image"
response = request_multipart_json(
edits_api_url,
target.token,
payload,
image_files,
image_field_name=image_field_name,
timeout=timeout,
)
else:
response = request_json("POST", generations_api_url, target.token, payload, timeout=timeout)
return response, target.request_style, requested_size, api_size
def safe_filename(filename: str, fallback: str, mime_type: str) -> str:
candidate = re.sub(r"[^A-Za-z0-9._-]+", "_", Path(filename).name).strip("._")
if not candidate:
candidate = fallback
if not Path(candidate).suffix:
candidate += mimetypes.guess_extension(mime_type) or ".png"
return candidate
def read_data_uri_image(value: str, index: int) -> ImageInputFile:
match = re.match(r"^data:(?P<mime>image/[^;,]+)(?P<params>(?:;[^,]+)*),(?P<data>.+)$", value, flags=re.DOTALL)
if not match or ";base64" not in match.group("params").lower():
raise ImageGenerationError("Reference image data URI must be a base64 image")
mime_type = match.group("mime")
try:
data = base64.b64decode(match.group("data"), validate=True)
except (ValueError, binascii.Error) as error:
raise ImageGenerationError("Reference image data URI contains invalid base64") from error
return ImageInputFile(
filename=safe_filename("", f"reference-{index}", mime_type),
mime_type=mime_type,
data=data,
)
def download_reference_image(url: str, index: int, timeout: int) -> ImageInputFile:
request = urllib.request.Request(url, headers={"User-Agent": "gpt-image-2-generator/1.0"})
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
data = response.read()
content_type = response.headers.get("Content-Type")
except urllib.error.HTTPError as error:
body = error.read().decode("utf-8", errors="replace")
raise ImageGenerationError(f"Reference image URL returned HTTP {error.code}: {body[:500]}") from error
except urllib.error.URLError as error:
raise ImageGenerationError(f"Reference image URL could not be downloaded: {error}") from error
mime_type = (content_type or "").split(";")[0].strip().lower()
if not mime_type.startswith("image/"):
mime_type = mimetypes.guess_type(urllib.parse.urlparse(url).path)[0] or "image/png"
filename = safe_filename(Path(urllib.parse.urlparse(url).path).name, f"reference-{index}", mime_type)
return ImageInputFile(filename=filename, mime_type=mime_type, data=data)
def read_local_reference_image(image_ref: str, index: int) -> ImageInputFile:
value = image_ref.strip()
path = Path(value).expanduser()
if not path.is_absolute():
cwd_path = Path.cwd() / path
root_path = ROOT_DIR / path
path = cwd_path if cwd_path.exists() else root_path
if not path.exists():
raise ImageGenerationError(f"Reference image not found: {image_ref}")
mime_type = mimetypes.guess_type(str(path))[0] or "image/png"
return ImageInputFile(
filename=safe_filename(path.name, f"reference-{index}", mime_type),
mime_type=mime_type,
data=path.read_bytes(),
)
def image_ref_to_input_file(image_ref: str, index: int, timeout: int) -> ImageInputFile:
value = image_ref.strip()
lowered = value.lower()
if lowered.startswith("data:"):
return read_data_uri_image(value, index)
if lowered.startswith(("http://", "https://")):
return download_reference_image(value, index, timeout)
return read_local_reference_image(value, index)
def save_json(path: Path, data: dict[str, Any]) -> None:
path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
def unique_path(path: Path) -> Path:
if not path.exists():
return path
for index in range(2, 1000):
candidate = path.with_name(f"{path.stem}-{index}{path.suffix}")
if not candidate.exists():
return candidate
raise ImageGenerationError(f"Cannot create unique output path for {path}")
def save_b64_image(image_b64: str, output_dir: Path, stem: str, index: int, mime: str | None = None) -> Path:
extension = mimetypes.guess_extension(mime or "image/png") or ".png"
output_path = unique_path(output_dir / f"{stem}-{index}{extension}")
output_path.write_bytes(base64.b64decode(image_b64))
return output_path
def save_data_uri(data_uri: str, output_dir: Path, stem: str, index: int) -> Path:
match = re.match(r"^data:(?P<mime>[^;]+);base64,(?P<data>.+)$", data_uri, flags=re.DOTALL)
if not match:
raise ImageGenerationError("Unsupported data URI image format")
return save_b64_image(match.group("data"), output_dir, stem, index, match.group("mime"))
def extract_b64_items(data: Any) -> list[dict[str, Any]]:
items: list[dict[str, Any]] = []
def visit(value: Any) -> None:
if isinstance(value, dict):
if isinstance(value.get("b64_json"), str):
items.append(value)
return
for child in value.values():
visit(child)
elif isinstance(value, list):
for child in value:
visit(child)
visit(data)
return items
def extract_image_values(data: Any) -> list[str]:
values: list[str] = []
def add(value: Any) -> None:
if isinstance(value, str) and value.startswith(("http://", "https://", "data:image/")):
values.append(value)
elif isinstance(value, list):
for item in value:
add(item)
candidates = []
if isinstance(data, dict):
candidates.append(data.get("url"))
nested_data = data.get("data")
if isinstance(nested_data, list):
candidates.extend(item.get("url") for item in nested_data if isinstance(item, dict))
if isinstance(nested_data, dict):
result = nested_data.get("result")
if isinstance(result, dict):
images = result.get("images")
if isinstance(images, list):
candidates.extend(item.get("url") for item in images if isinstance(item, dict))
candidates.append(result.get("url"))
result = data.get("result")
if isinstance(result, dict):
images = result.get("images")
if isinstance(images, list):
candidates.extend(item.get("url") for item in images if isinstance(item, dict))
for candidate in candidates:
add(candidate)
return values
def extract_task_id(data: dict[str, Any]) -> str | None:
for key in ("task_id", "id"):
value = data.get(key)
if isinstance(value, str) and value:
return value
nested_data = data.get("data")
if isinstance(nested_data, dict):
for key in ("task_id", "id"):
value = nested_data.get(key)
if isinstance(value, str) and value:
return value
return None
def get_task_status(task: dict[str, Any]) -> str | None:
nested_data = task.get("data")
if isinstance(nested_data, dict) and isinstance(nested_data.get("status"), str):
return nested_data["status"]
status = task.get("status")
return status if isinstance(status, str) else None
def get_task_error(task: dict[str, Any]) -> str:
nested_data = task.get("data")
error = nested_data.get("error") if isinstance(nested_data, dict) else task.get("error")
if isinstance(error, dict):
return str(error.get("message") or error)
if error:
return str(error)
return "task failed"
def guess_download_extension(url: str, content_type: str | None) -> str:
suffix = Path(urllib.parse.urlparse(url).path).suffix.lower()
if suffix in {".png", ".jpg", ".jpeg", ".webp", ".gif"}:
return suffix
if content_type:
return mimetypes.guess_extension(content_type.split(";")[0].strip()) or ".png"
return ".png"
def download_image(url: str, output_dir: Path, stem: str, index: int, timeout: int) -> Path:
request = urllib.request.Request(url, headers={"User-Agent": "gpt-image-2-generator/1.0"})
with urllib.request.urlopen(request, timeout=timeout) as response:
content_type = response.headers.get("Content-Type")
extension = guess_download_extension(url, content_type)
output_path = unique_path(output_dir / f"{stem}-{index}{extension}")
output_path.write_bytes(response.read())
return output_path
def save_image_values(values: list[str], output_dir: Path, stem: str, timeout: int) -> list[Path]:
paths: list[Path] = []
for index, value in enumerate(values, start=1):
if value.startswith("data:image/"):
paths.append(save_data_uri(value, output_dir, stem, index))
else:
paths.append(download_image(value, output_dir, stem, index, timeout))
return paths
def generate_image(
prompt: str,
size: str | None = None,
image_refs: list[str] | None = None,
n: int = 1,
output_dir: str | Path | None = None,
timeout: int = 300,
initial_poll_delay: float = 10,
poll_interval: float = 5,
max_wait: float = 300,
config_dir: str | Path | None = None,
env_file: str | Path | None = None,
base_url: str | None = None,
model: str | None = None,
request_style: str | None = None,
quality: str | None = None,
background: str | None = None,
output_format: str | None = None,
input_fidelity: str | None = None,
) -> dict[str, Any]:
prompt = prompt.strip()
if not prompt:
raise ImageGenerationError("Prompt is required")
if n < 1:
raise ImageGenerationError("n must be >= 1")
runtime_config = load_runtime_config(
execution_root=Path.cwd(),
config_dir=config_dir,
env_file=env_file,
base_url_override=base_url,
model_override=model,
output_dir_override=output_dir,
require_token=True,
)
env_map = parse_env_file(runtime_config.env_file)
resolved_request_style = request_style.strip().lower() if request_style else runtime_config.request_style
if resolved_request_style == "auto":
resolved_request_style = runtime_config.request_style
if resolved_request_style not in {"apicodex", "openai"}:
raise ImageGenerationError(
f"Unsupported request style: {resolved_request_style}. Use one of: {', '.join(REQUEST_STYLE_CHOICES)}"
)
refs = [ref.strip() for ref in (image_refs or []) if ref.strip()]
output_path = runtime_config.output_dir
output_path.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
stem = f"gpt-image-2-{timestamp}"
request_type = "edit" if refs else "generation"
image_files = [image_ref_to_input_file(ref, index, timeout) for index, ref in enumerate(refs, start=1)] if refs else []
targets = build_request_targets(runtime_config, resolved_request_style)
response: dict[str, Any] | None = None
selected_target: RequestTarget | None = None
requested_size = ""
api_size = ""
last_error: BaseException | None = None
for target_index, target in enumerate(targets):
for attempt in range(1, REQUEST_RETRY_ATTEMPTS + 1):
try:
response, resolved_request_style, requested_size, api_size = submit_request_for_target(
target,
runtime_config=runtime_config,
env_map=env_map,
prompt=prompt,
size=size,
n=n,
has_refs=bool(refs),
image_files=image_files,
timeout=timeout,
quality=quality,
background=background,
output_format=output_format,
input_fidelity=input_fidelity,
)
selected_target = target
break
except BaseException as error:
last_error = error
if attempt < REQUEST_RETRY_ATTEMPTS and should_retry_request(error):
print(
f"request attempt {attempt}/{REQUEST_RETRY_ATTEMPTS} to {target.base_url} failed with {format_request_error(error)}; retrying in {REQUEST_RETRY_DELAY_SECONDS:g}s",
file=sys.stderr,
)
time.sleep(REQUEST_RETRY_DELAY_SECONDS)
continue
next_target = targets[target_index + 1] if target_index + 1 < len(targets) else None
if next_target is not None and is_failover_eligible_error(error):
print(
f"request to {target.base_url} failed with {format_request_error(error)}; switching to fallback {next_target.base_url}",
file=sys.stderr,
)
break
raise
if response is not None and selected_target is not None:
break
if response is None or selected_target is None:
raise ImageGenerationError(f"Image request failed: {last_error}")
task_api_base = build_api_url(selected_target.base_url, "tasks/")
response_path = output_path / f"{stem}-response.json"
save_json(response_path, response)
image_paths: list[Path] = []
revised_prompts: list[str] = []
b64_items = extract_b64_items(response)
if b64_items:
for index, item in enumerate(b64_items, start=1):
image_paths.append(save_b64_image(item["b64_json"], output_path, stem, index))
if isinstance(item.get("revised_prompt"), str):
revised_prompts.append(item["revised_prompt"])
return {
"images": [str(path) for path in image_paths],
"response_path": str(response_path),
"revised_prompts": revised_prompts,
"request_type": request_type,
"mode": "sync-b64",
"base_url": selected_target.base_url,
"model": runtime_config.model,
"request_style": resolved_request_style,
"requested_size": requested_size,
"api_size": api_size,
"output_dir": str(output_path),
}
image_values = extract_image_values(response)
if image_values:
image_paths = save_image_values(image_values, output_path, stem, timeout)
return {
"images": [str(path) for path in image_paths],
"response_path": str(response_path),
"revised_prompts": revised_prompts,
"request_type": request_type,
"mode": "sync-url",
"base_url": selected_target.base_url,
"model": runtime_config.model,
"request_style": resolved_request_style,
"requested_size": requested_size,
"api_size": api_size,
"output_dir": str(output_path),
}
task_id = extract_task_id(response)
if not task_id:
raise ImageGenerationError(f"Unexpected response. Saved raw response to {response_path}")
time.sleep(initial_poll_delay)
deadline = time.time() + max_wait
last_task: dict[str, Any] | None = None
while time.time() <= deadline:
last_task = request_json(
"GET",
urllib.parse.urljoin(task_api_base, urllib.parse.quote(task_id)),
runtime_config.token or "",
timeout=timeout,
)
status = get_task_status(last_task)
if status == "completed":
task_path = output_path / f"{stem}-task.json"
save_json(task_path, last_task)
b64_items = extract_b64_items(last_task)
if b64_items:
for index, item in enumerate(b64_items, start=1):
image_paths.append(save_b64_image(item["b64_json"], output_path, stem, index))
if isinstance(item.get("revised_prompt"), str):
revised_prompts.append(item["revised_prompt"])
else:
image_paths = save_image_values(extract_image_values(last_task), output_path, stem, timeout)
if not image_paths:
raise ImageGenerationError(f"Task completed but no image was found. Saved task response to {task_path}")
return {
"images": [str(path) for path in image_paths],
"response_path": str(response_path),
"task_response_path": str(task_path),
"revised_prompts": revised_prompts,
"request_type": request_type,
"mode": "async-task",
"base_url": selected_target.base_url,
"model": runtime_config.model,
"request_style": resolved_request_style,
"requested_size": requested_size,
"api_size": api_size,
"output_dir": str(output_path),
}
if status == "failed":
raise ImageGenerationError(get_task_error(last_task))
time.sleep(poll_interval)
if last_task is not None:
task_path = output_path / f"{stem}-task-timeout.json"
save_json(task_path, last_task)
raise ImageGenerationError(f"Task polling timed out. Saved last task response to {task_path}")
raise ImageGenerationError("Task polling timed out before the first task response")
def save_token(token: str, config_dir: str | Path | None = None) -> Path:
return save_runtime_token(token, execution_root=Path.cwd(), config_dir=config_dir)
def save_base_url(base_url: str, config_dir: str | Path | None = None) -> Path:
return save_runtime_base_url(base_url, execution_root=Path.cwd(), config_dir=config_dir)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Generate or edit images with GPT Image 2 via apicodex-style or OpenAI-compatible APIs")
parser.add_argument("--save-token", metavar="TOKEN", help="Save an API token to <cwd>/config/token.txt and exit")
parser.add_argument("--save-base-url", metavar="URL", help="Save an API base URL to <cwd>/config/base_url.txt and exit")
parser.add_argument("--print-env-template", action="store_true", help="Print a recommended .env template and exit")
parser.add_argument("--init-config", action="store_true", help="Write a recommended env template file and exit")
parser.add_argument("--prompt", required=False, default=None, help="Text prompt")
parser.add_argument("--size", default=None, help="Size such as 2:3 or 1024x1536. If omitted, tries env defaults before falling back to 1:1.")
parser.add_argument("--n", default=1, type=int, help="Number of images")
parser.add_argument("--image-ref", action="append", default=[], help="Reference image URL, data URI, or local path")
parser.add_argument("--output-dir", default=None, help="Output directory. Defaults to <cwd>/outputs.")
parser.add_argument("--timeout", default=300, type=int, help="HTTP timeout in seconds")
parser.add_argument("--initial-poll-delay", default=10, type=float, help="Seconds to wait before polling async tasks")
parser.add_argument("--poll-interval", default=5, type=float, help="Async task poll interval in seconds")
parser.add_argument("--max-wait", default=300, type=float, help="Maximum async task wait time in seconds")
parser.add_argument("--config-dir", help="Config directory. Defaults to <cwd>/config.")
parser.add_argument("--env-file", help="Env file. Defaults to <cwd>/.env.")
parser.add_argument("--base-url", help="Override the API base URL for this request.")
parser.add_argument("--model", help="Override the model for this request.")
parser.add_argument("--request-style", choices=REQUEST_STYLE_CHOICES, default=None, help="Force request style. Defaults to auto/inferred from base URL.")
parser.add_argument("--quality", default=None, help="Optional image edit quality value.")
parser.add_argument("--background", default=None, help="Optional image edit background value.")
parser.add_argument("--output-format", default=None, help="Optional image edit output format value.")
parser.add_argument("--input-fidelity", default=None, help="Optional image edit input fidelity value.")
parser.add_argument("--show-config", action="store_true", help="Print the resolved runtime configuration and exit")
parser.add_argument("--force", action="store_true", help="Allow overwriting files when used with --init-config")
return parser
def main() -> int:
args = build_parser().parse_args()
if args.save_token:
try:
path = save_token(args.save_token, config_dir=args.config_dir)
print(f"Token saved to {path}")
return 0
except (ConfigError, Exception) as error:
print(f"ERROR: {error}", file=sys.stderr)
return 1
if args.save_base_url:
try:
path = save_base_url(args.save_base_url, config_dir=args.config_dir)
print(f"Base URL saved to {path}")
return 0
except (ConfigError, Exception) as error:
print(f"ERROR: {error}", file=sys.stderr)
return 1
if args.print_env_template:
print(build_env_template(), end="")
return 0
if args.init_config:
try:
path = write_env_template(env_file=args.env_file or ".env.example", overwrite=args.force)
print(f"Config template written to {path}")
return 0
except (ConfigError, Exception) as error:
print(f"ERROR: {error}", file=sys.stderr)
return 1
if args.show_config:
try:
config = load_runtime_config(
execution_root=Path.cwd(),
config_dir=args.config_dir,
env_file=args.env_file,
base_url_override=args.base_url,
model_override=args.model,
output_dir_override=args.output_dir,
require_token=False,
)
print(json.dumps(runtime_config_summary(config), ensure_ascii=False, indent=2))
return 0
except (ConfigError, Exception) as error:
print(f"ERROR: {error}", file=sys.stderr)
return 1
if not args.prompt:
print(
"ERROR: --prompt is required (unless using --save-token / --save-base-url / --print-env-template / --init-config / --show-config)",
file=sys.stderr,
)
return 1
try:
result = generate_image(
prompt=args.prompt,
size=args.size,
image_refs=args.image_ref,
n=args.n,
output_dir=args.output_dir,
timeout=args.timeout,
initial_poll_delay=args.initial_poll_delay,
poll_interval=args.poll_interval,
max_wait=args.max_wait,
config_dir=args.config_dir,
env_file=args.env_file,
base_url=args.base_url,
model=args.model,
request_style=args.request_style,
quality=args.quality,
background=args.background,
output_format=args.output_format,
input_fidelity=args.input_fidelity,
)
except (ConfigError, Exception) as error:
print(f"ERROR: {error}", file=sys.stderr)
return 1
print(json.dumps(result, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())

399
scripts/runtime_config.py Normal file
View File

@@ -0,0 +1,399 @@
from __future__ import annotations
from dataclasses import dataclass
import os
from pathlib import Path
from typing import Callable
import urllib.parse
ROOT_DIR = Path(__file__).resolve().parents[1]
DEFAULT_BASE_URL = "https://apicodex.xyz"
DEFAULT_MODEL = "gpt-image-2"
DEFAULT_OUTPUT_DIR_NAME = "outputs"
DEFAULT_REQUEST_STYLE = "apicodex"
TOKEN_PLACEHOLDERS = {"YOUR_API_TOKEN_HERE", "<token>", "Bearer <token>"}
BASE_URL_PLACEHOLDERS = {"<base_url>", "https://example.com", "https://example.com/v1"}
MODEL_PLACEHOLDERS = {"<model>"}
REQUEST_STYLE_VALUES = {"auto", "apicodex", "openai"}
TOKEN_ENV_VARS = (
"GPT_IMAGE_2_TOKEN",
"APICODEX_API_KEY",
"APICODEX_TOKEN",
"IMAGE_API_KEY",
"CODEX_API_KEY",
"OPENAI_API_KEY",
)
BASE_URL_ENV_VARS = (
"GPT_IMAGE_2_BASE_URL",
"APICODEX_BASE_URL",
"IMAGE_BASE_URL",
"OPENAI_BASE_URL",
"OPENAI_API_BASE",
)
MODEL_ENV_VARS = (
"GPT_IMAGE_2_MODEL",
"IMAGE_MODEL",
)
REQUEST_STYLE_ENV_VARS = (
"GPT_IMAGE_2_REQUEST_STYLE",
"IMAGE_REQUEST_STYLE",
"IMAGE_API_STYLE",
)
OUTPUT_DIR_ENV_VARS = (
"GPT_IMAGE_2_OUTPUT_DIR",
"IMAGE_OUTPUT_DIR",
)
class ConfigError(RuntimeError):
pass
@dataclass(frozen=True)
class ConfigValue:
value: str
source: str
@dataclass(frozen=True)
class RuntimeConfig:
execution_root: Path
env_file: Path
project_config_dir: Path
skill_root: Path
skill_config_dir: Path
token: str | None
token_source: str | None
base_url: str
base_url_source: str
model: str
model_source: str
request_style: str
request_style_source: str
output_dir: Path
output_dir_source: str
def normalize_token(raw_token: str) -> str:
return raw_token.strip().removeprefix("Bearer ").strip()
def normalize_base_url(raw_value: str) -> str:
value = raw_value.strip().strip('"').strip("'").rstrip("/")
if not value:
return ""
parsed = urllib.parse.urlparse(value)
if not parsed.scheme or not parsed.netloc:
return value
path = parsed.path.rstrip("/")
if not path:
path = "/v1"
normalized = parsed._replace(path=path)
return urllib.parse.urlunparse(normalized).rstrip("/")
def normalize_model(raw_value: str) -> str:
return raw_value.strip()
def normalize_request_style(raw_value: str) -> str:
value = raw_value.strip().strip('"').strip("'").lower()
if not value:
return ""
if value not in REQUEST_STYLE_VALUES:
raise ConfigError(
f"Unsupported request style: {raw_value}. Use one of: {', '.join(sorted(REQUEST_STYLE_VALUES))}"
)
return value
def infer_request_style(base_url: str) -> str:
parsed = urllib.parse.urlparse(normalize_base_url(base_url))
host = (parsed.netloc or "").lower()
if "apicodex" in host:
return "apicodex"
if "openai" in host:
return "openai"
return "openai"
def missing_token_message() -> str:
env_list = ", ".join(TOKEN_ENV_VARS)
return (
f"No usable token found. Set one of these environment variables: {env_list}, "
"or configure the current working directory with .env / config/token.txt. "
f"Skill fallback config is {ROOT_DIR / 'config' / 'token.txt'}."
)
def build_api_url(base_url: str, path_fragment: str) -> str:
base = normalize_base_url(base_url).rstrip("/") + "/"
return urllib.parse.urljoin(base, path_fragment.lstrip("/"))
def load_runtime_config(
*,
execution_root: str | Path | None = None,
config_dir: str | Path | None = None,
env_file: str | Path | None = None,
base_url_override: str | None = None,
model_override: str | None = None,
output_dir_override: str | Path | None = None,
require_token: bool = True,
) -> RuntimeConfig:
resolved_execution_root = Path(execution_root or os.getcwd()).expanduser().resolve()
resolved_config_dir = _resolve_project_config_dir(resolved_execution_root, config_dir)
resolved_env_file = _resolve_env_file(resolved_execution_root, env_file)
skill_config_dir = ROOT_DIR / "config"
env_map = parse_env_file(resolved_env_file)
token_value = _select_text_value(
env_vars=TOKEN_ENV_VARS,
env_map=env_map,
file_candidates=(resolved_config_dir / "token.txt", skill_config_dir / "token.txt"),
normalizer=normalize_token,
placeholders=TOKEN_PLACEHOLDERS,
)
if require_token and token_value is None:
raise ConfigError(missing_token_message())
base_url_value = _select_text_value(
env_vars=BASE_URL_ENV_VARS,
env_map=env_map,
file_candidates=(
resolved_config_dir / "base_url.txt",
resolved_config_dir / "baseurl.txt",
skill_config_dir / "base_url.txt",
skill_config_dir / "baseurl.txt",
),
normalizer=normalize_base_url,
placeholders=BASE_URL_PLACEHOLDERS,
override=(base_url_override, "cli:--base-url") if base_url_override else None,
) or ConfigValue(normalize_base_url(DEFAULT_BASE_URL), "default")
model_value = _select_text_value(
env_vars=MODEL_ENV_VARS,
env_map=env_map,
file_candidates=(resolved_config_dir / "model.txt", skill_config_dir / "model.txt"),
normalizer=normalize_model,
placeholders=MODEL_PLACEHOLDERS,
override=(model_override, "cli:--model") if model_override else None,
) or ConfigValue(DEFAULT_MODEL, "default")
request_style_value = _select_text_value(
env_vars=REQUEST_STYLE_ENV_VARS,
env_map=env_map,
file_candidates=(resolved_config_dir / "request_style.txt", skill_config_dir / "request_style.txt"),
normalizer=normalize_request_style,
placeholders=set(),
)
if request_style_value is None or request_style_value.value == "auto":
resolved_request_style = infer_request_style(base_url_value.value)
request_style_source = (
f"{request_style_value.source}->inferred:base-url"
if request_style_value is not None
else "inferred:base-url"
)
else:
resolved_request_style = request_style_value.value
request_style_source = request_style_value.source
output_dir_value = _select_text_value(
env_vars=OUTPUT_DIR_ENV_VARS,
env_map=env_map,
file_candidates=(resolved_config_dir / "output_dir.txt", skill_config_dir / "output_dir.txt"),
normalizer=lambda value: value.strip(),
placeholders=set(),
override=(str(output_dir_override), "cli:--output-dir") if output_dir_override else None,
)
if output_dir_value is None:
resolved_output_dir = resolved_execution_root / DEFAULT_OUTPUT_DIR_NAME
output_dir_source = "default"
else:
resolved_output_dir = resolve_path_from_root(output_dir_value.value, resolved_execution_root)
output_dir_source = output_dir_value.source
return RuntimeConfig(
execution_root=resolved_execution_root,
env_file=resolved_env_file,
project_config_dir=resolved_config_dir,
skill_root=ROOT_DIR,
skill_config_dir=skill_config_dir,
token=token_value.value if token_value else None,
token_source=token_value.source if token_value else None,
base_url=base_url_value.value,
base_url_source=base_url_value.source,
model=model_value.value,
model_source=model_value.source,
request_style=resolved_request_style,
request_style_source=request_style_source,
output_dir=resolved_output_dir,
output_dir_source=output_dir_source,
)
def runtime_config_summary(config: RuntimeConfig) -> dict[str, str | bool | None]:
return {
"execution_root": str(config.execution_root),
"env_file": str(config.env_file),
"env_file_exists": config.env_file.exists(),
"project_config_dir": str(config.project_config_dir),
"skill_config_dir": str(config.skill_config_dir),
"token_configured": bool(config.token),
"token_source": config.token_source,
"base_url": config.base_url,
"base_url_source": config.base_url_source,
"model": config.model,
"model_source": config.model_source,
"request_style": config.request_style,
"request_style_source": config.request_style_source,
"output_dir": str(config.output_dir),
"output_dir_source": config.output_dir_source,
}
def save_token(token: str, *, execution_root: str | Path | None = None, config_dir: str | Path | None = None) -> Path:
normalized = normalize_token(token)
if not normalized or normalized in TOKEN_PLACEHOLDERS:
raise ConfigError("Token cannot be empty")
target_dir = _resolve_project_config_dir(Path(execution_root or os.getcwd()).expanduser().resolve(), config_dir)
target_dir.mkdir(parents=True, exist_ok=True)
target_path = target_dir / "token.txt"
target_path.write_text(
f"# apicodex.xyz API token\n{normalized}\n",
encoding="utf-8",
)
return target_path
def save_base_url(
base_url: str,
*,
execution_root: str | Path | None = None,
config_dir: str | Path | None = None,
) -> Path:
normalized = normalize_base_url(base_url)
if not normalized or normalized in BASE_URL_PLACEHOLDERS:
raise ConfigError("Base URL cannot be empty")
target_dir = _resolve_project_config_dir(Path(execution_root or os.getcwd()).expanduser().resolve(), config_dir)
target_dir.mkdir(parents=True, exist_ok=True)
target_path = target_dir / "base_url.txt"
target_path.write_text(
f"# API base URL. Either host root or full /v1 path is accepted.\n{normalized}\n",
encoding="utf-8",
)
return target_path
def resolve_path_from_root(value: str, root: Path) -> Path:
path = Path(value).expanduser()
if path.is_absolute():
return path
return (root / path).resolve()
def parse_env_file(path: Path) -> dict[str, str]:
if not path.exists():
return {}
values: dict[str, str] = {}
for raw_line in path.read_text(encoding="utf-8-sig").splitlines():
line = raw_line.strip()
if not line or line.startswith("#"):
continue
if line.startswith("export "):
line = line[7:].lstrip()
if "=" not in line:
continue
key, raw_value = line.split("=", 1)
key = key.strip()
if not key:
continue
values[key] = _parse_env_value(raw_value.strip())
return values
def _parse_env_value(raw_value: str) -> str:
if not raw_value:
return ""
if raw_value[0] in {'"', "'"} and raw_value[-1] == raw_value[0]:
return raw_value[1:-1]
if " #" in raw_value:
raw_value = raw_value.split(" #", 1)[0].rstrip()
return raw_value.strip()
def _resolve_project_config_dir(execution_root: Path, config_dir: str | Path | None) -> Path:
raw_value = config_dir or os.environ.get("GPT_IMAGE_2_CONFIG_DIR")
if not raw_value:
return execution_root / "config"
return resolve_path_from_root(str(raw_value), execution_root)
def _resolve_env_file(execution_root: Path, env_file: str | Path | None) -> Path:
raw_value = env_file or os.environ.get("GPT_IMAGE_2_ENV_FILE")
if not raw_value:
return execution_root / ".env"
return resolve_path_from_root(str(raw_value), execution_root)
def _select_text_value(
*,
env_vars: tuple[str, ...],
env_map: dict[str, str],
file_candidates: tuple[Path, ...],
normalizer: Callable[[str], str],
placeholders: set[str],
override: tuple[str, str] | None = None,
) -> ConfigValue | None:
if override is not None:
resolved = _coerce_text_value(override[0], override[1], normalizer, placeholders)
if resolved is not None:
return resolved
for env_name in env_vars:
resolved = _coerce_text_value(os.environ.get(env_name, ""), f"env:{env_name}", normalizer, placeholders)
if resolved is not None:
return resolved
for env_name in env_vars:
if env_name not in env_map:
continue
resolved = _coerce_text_value(env_map[env_name], f"env-file:{env_name}", normalizer, placeholders)
if resolved is not None:
return resolved
for candidate in file_candidates:
resolved = _coerce_text_value(_read_config_text(candidate), f"file:{candidate}", normalizer, placeholders)
if resolved is not None:
return resolved
return None
def _coerce_text_value(
raw_value: str,
source: str,
normalizer: Callable[[str], str],
placeholders: set[str],
) -> ConfigValue | None:
value = normalizer(raw_value)
if not value or value in placeholders:
return None
return ConfigValue(value=value, source=source)
def _read_config_text(path: Path) -> str:
if not path.exists():
return ""
for line in path.read_text(encoding="utf-8-sig").splitlines():
stripped = line.strip()
if stripped and not stripped.startswith("#"):
return stripped
return ""

775
scripts/web_app.py Normal file
View File

@@ -0,0 +1,775 @@
from __future__ import annotations
import argparse
import html
import io
import json
import mimetypes
import sys
import threading
import urllib.parse
import webbrowser
import zipfile
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
SCRIPT_DIR = Path(__file__).resolve().parent
ROOT_DIR = SCRIPT_DIR.parent
MAX_PREVIEW_STRING = 240
MAX_HISTORY_ITEMS = 20
if str(SCRIPT_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPT_DIR))
import generate_image # noqa: E402
import runtime_config # noqa: E402
RUNTIME_CONFIG = runtime_config.load_runtime_config(
execution_root=Path.cwd(),
require_token=False,
)
OUTPUT_DIR = RUNTIME_CONFIG.output_dir
HISTORY_FILE = OUTPUT_DIR / "history.jsonl"
HTML_PAGE = r"""<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<title>GPT Image 2 生图工具</title>
<style>
:root { color-scheme: dark; font-family: Inter, "Segoe UI", sans-serif; }
* { box-sizing: border-box; }
body { margin: 0; background: #0f172a; color: #e5e7eb; }
main { max-width: 1180px; margin: 0 auto; padding: 28px 16px 56px; }
.panel { background: rgba(15, 23, 42, .82); border: 1px solid #334155; border-radius: 18px; padding: 22px; box-shadow: 0 18px 60px rgba(0, 0, 0, .22); }
.panel + .panel { margin-top: 18px; }
h1, h2, h3 { margin: 0; }
h1 { font-size: 30px; margin-bottom: 8px; }
h2 { font-size: 22px; margin-bottom: 12px; }
p { margin: 0; color: #94a3b8; line-height: 1.6; }
label { display: block; margin: 16px 0 8px; font-weight: 700; }
textarea, select, input { width: 100%; border: 1px solid #475569; border-radius: 12px; background: #020617; color: #f8fafc; padding: 12px; font: inherit; }
textarea { min-height: 132px; resize: vertical; }
.row { display: grid; grid-template-columns: 1fr 170px 120px; gap: 14px; align-items: start; }
.advanced-grid { display: grid; grid-template-columns: repeat(4, minmax(120px, 1fr)); gap: 14px; }
.chips { display: flex; flex-wrap: wrap; gap: 10px; margin-top: 12px; }
.chip { margin: 0; padding: 8px 14px; border-radius: 999px; border: 1px solid #475569; background: #111827; color: #cbd5e1; cursor: pointer; }
.toolbar { display: flex; flex-wrap: wrap; gap: 12px; align-items: center; margin-top: 16px; }
.button, button { display: inline-flex; align-items: center; justify-content: center; gap: 8px; border: 0; border-radius: 999px; padding: 13px 22px; font-weight: 800; font-size: 14px; background: linear-gradient(135deg, #38bdf8, #818cf8); color: #020617; cursor: pointer; text-decoration: none; }
.button.secondary, button.secondary { background: #1e293b; color: #e2e8f0; border: 1px solid #475569; }
.button.ghost, button.ghost { background: transparent; color: #cbd5e1; border: 1px dashed #475569; }
button:disabled { opacity: .55; cursor: wait; }
.hint { margin-top: 8px; font-size: 13px; color: #94a3b8; }
.status { white-space: pre-wrap; margin-top: 16px; color: #cbd5e1; min-height: 48px; }
.grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(240px, 1fr)); gap: 16px; margin-top: 20px; }
.card, .history-item { border: 1px solid #334155; border-radius: 14px; padding: 12px; background: #020617; }
.history-item + .history-item { margin-top: 14px; }
.thumbs { display: grid; grid-template-columns: repeat(auto-fit, minmax(150px, 1fr)); gap: 12px; margin-top: 14px; }
img { width: 100%; height: auto; border-radius: 10px; display: block; background: #111827; }
a { color: #7dd3fc; word-break: break-all; }
code, pre { background: #111827; border-radius: 12px; }
code { padding: 2px 6px; }
pre { margin: 12px 0 0; padding: 14px; overflow: auto; white-space: pre-wrap; word-break: break-word; border: 1px solid #334155; }
details { margin-top: 16px; border: 1px solid #334155; border-radius: 14px; padding: 12px 14px; background: #020617; }
summary { cursor: pointer; font-weight: 750; color: #bfdbfe; }
.section-title { display: flex; justify-content: space-between; gap: 12px; align-items: center; margin-bottom: 12px; }
.meta { margin-top: 8px; color: #94a3b8; font-size: 13px; }
.json-grid { display: grid; grid-template-columns: 1fr 1fr; gap: 16px; }
.muted { color: #94a3b8; }
.empty { color: #94a3b8; padding: 10px 0; }
.inline-links { display: flex; flex-wrap: wrap; gap: 12px; margin-top: 12px; }
@media (max-width: 900px) {
.row, .advanced-grid, .json-grid { grid-template-columns: 1fr; }
}
</style>
</head>
<body>
<main>
<section class="panel">
<h1>GPT Image 2 生图工具</h1>
<p>优先读取当前执行目录的 <code>.env</code> 和 <code>config/</code>，再回退到 skill 自带 <code>config/</code>。输入 prompt，可选添加一张或多张参考图，服务端会自动选择纯文本生成或带图编辑，并兼容 apicodex 风格与 OpenAI-compatible 网关。</p>
<div id="runtimeConfig" class="hint">正在读取运行时配置…</div>
<label for="prompt">Prompt</label>
<textarea id="prompt" placeholder="例如：一只橘猫坐在窗台看夕阳，水彩画风格"></textarea>
<div class="chips" id="exampleChips">
<button type="button" class="chip" data-size="16:9" data-prompt="一只橘猫坐在窗台看夕阳，水彩画风格，电影感，暖色调">示例：橘猫夕阳</button>
<button type="button" class="chip" data-size="9:16" data-prompt="中国古风少女站在竹林中，轻雾，插画风，高清细节">示例：古风竖图</button>
<button type="button" class="chip" data-size="1:1" data-prompt="把参考图改成水彩插画风格，保留主体构图">示例：参考图改风格</button>
<button type="button" class="chip" data-size="4:3" data-prompt="把参考图融合成一张复古海报，统一色调与光影">示例：多参考图海报</button>
</div>
<div class="row">
<div>
<label for="refs">参考图 URL / data URI（每行一个，可选）</label>
<textarea id="refs" placeholder="https://example.com/photo.jpg"></textarea>
</div>
<div>
<label for="size">尺寸</label>
<input id="size" list="sizeOptions" value="16:9" placeholder="例如 2:3 或 1024x1536" />
<datalist id="sizeOptions">
<option value="1:1">1:1 正方形</option>
<option value="16:9">16:9 横屏</option>
<option value="9:16">9:16 竖屏</option>
<option value="4:3">4:3 横图</option>
<option value="3:4">3:4 竖图</option>
<option value="3:2">3:2 横图</option>
<option value="2:3">2:3 竖图</option>
<option value="5:4">5:4 横图</option>
<option value="4:5">4:5 竖图</option>
<option value="2:1">2:1 宽屏</option>
<option value="1:2">1:2 长竖图</option>
<option value="21:9">21:9 超宽屏</option>
<option value="9:21">9:21 超长竖图</option>
<option value="1024x1024">1024x1024</option>
<option value="1024x1536">1024x1536</option>
<option value="1536x1024">1536x1024</option>
<option value="864x1536">864x1536</option>
<option value="1536x864">1536x864</option>
<option value="2048x3584">2048x3584</option>
</datalist>
</div>
<div>
<label for="count">数量</label>
<input id="count" type="number" min="1" max="4" value="1" />
</div>
</div>
<label for="files">本地参考图（可多选，可选）</label>
<input id="files" type="file" accept="image/*" multiple />
<div class="hint">本地图片会在浏览器中转换为 data URI 后提交给本机服务。尺寸既可填比例，如 <code>2:3</code>，也可填像素，如 <code>1024x1536</code>。</div>
<details>
<summary>高级参数：超时和异步轮询</summary>
<div class="advanced-grid">
<div>
<label for="timeout">请求超时（秒）</label>
<input id="timeout" type="number" min="30" max="600" value="300" />
</div>
<div>
<label for="initialPollDelay">首次轮询等待（秒）</label>
<input id="initialPollDelay" type="number" min="0" max="60" value="10" />
</div>
<div>
<label for="pollInterval">轮询间隔（秒）</label>
<input id="pollInterval" type="number" min="1" max="30" value="5" />
</div>
<div>
<label for="maxWait">最长等待（秒）</label>
<input id="maxWait" type="number" min="30" max="1800" value="300" />
</div>
</div>
</details>
<div class="toolbar">
<button id="generate">开始生成</button>
<a id="downloadAllCurrent" class="button secondary" href="#" hidden>下载本次全部图片</a>
<button id="refreshHistory" type="button" class="secondary">刷新历史记录</button>
</div>
<div id="status" class="status"></div>
<div id="result" class="grid"></div>
</section>
<section class="panel">
<div class="section-title">
<h2>JSON 详情面板</h2>
<div class="inline-links">
<a id="rawResponseLink" href="#" target="_blank" hidden>打开原始响应 JSON</a>
<a id="rawTaskLink" href="#" target="_blank" hidden>打开任务 JSON</a>
</div>
</div>
<p id="jsonTitle" class="muted">生成完成后，可在这里查看接口返回的 JSON 预览。</p>
<div class="json-grid">
<div>
<h3>响应 JSON 预览</h3>
<pre id="responsePreview">暂无</pre>
</div>
<div>
<h3>任务 JSON 预览</h3>
<pre id="taskPreview">暂无</pre>
</div>
</div>
</section>
<section class="panel">
<div class="section-title">
<h2>历史生成记录</h2>
<span class="muted">显示最近 20 条</span>
</div>
<div id="historyList" class="empty">暂无历史记录</div>
</section>
</main>
<script>
const state = { historyItems: [] };
function escapeHtml(value) {
return String(value || "")
.replace(/&/g, "&amp;")
.replace(/</g, "&lt;")
.replace(/>/g, "&gt;")
.replace(/"/g, "&quot;")
.replace(/'/g, "&#39;");
}
function fileToDataURL(file) {
return new Promise((resolve, reject) => {
const reader = new FileReader();
reader.onload = () => resolve(reader.result);
reader.onerror = () => reject(reader.error);
reader.readAsDataURL(file);
});
}
function toJsonText(value) {
if (!value) return "暂无";
return JSON.stringify(value, null, 2);
}
function buildDownloadAllUrl(images) {
const names = (images || []).map(item => item.name).filter(Boolean);
return names.length ? `/api/download-all?names=${encodeURIComponent(names.join(","))}` : "#";
}
function updateDownloadButton(images) {
const button = document.querySelector("#downloadAllCurrent");
const url = buildDownloadAllUrl(images);
if (url === "#") {
button.hidden = true;
button.removeAttribute("href");
return;
}
button.hidden = false;
button.href = url;
}
function showJsonDetails(title, responsePreview, taskPreview, responseFile, taskFile) {
document.querySelector("#jsonTitle").textContent = title || "JSON 详情";
document.querySelector("#responsePreview").textContent = toJsonText(responsePreview);
document.querySelector("#taskPreview").textContent = toJsonText(taskPreview);
const responseLink = document.querySelector("#rawResponseLink");
if (responseFile && responseFile.url) {
responseLink.hidden = false;
responseLink.href = responseFile.url;
} else {
responseLink.hidden = true;
responseLink.removeAttribute("href");
}
const taskLink = document.querySelector("#rawTaskLink");
if (taskFile && taskFile.url) {
taskLink.hidden = false;
taskLink.href = taskFile.url;
} else {
taskLink.hidden = true;
taskLink.removeAttribute("href");
}
}
function renderRuntimeConfig(data) {
const el = document.querySelector("#runtimeConfig");
if (!data) {
el.textContent = "未获取到运行时配置。";
return;
}
const tokenLabel = data.token_configured
? `已配置（${data.token_source || "unknown"}）`
: "未配置";
const envLabel = data.env_file_exists ? data.env_file : `${data.env_file}（不存在）`;
el.textContent =
`执行目录：${data.execution_root} | .env：${envLabel} | token：${tokenLabel} | base URL：${data.base_url} | 请求风格：${data.request_style || "unknown"} | 输出目录：${data.output_dir}`;
}
function renderResult(data) {
const result = document.querySelector("#result");
const images = data.images || [];
if (!images.length) {
result.innerHTML = '<div class="empty">本次没有返回可展示的图片。</div>';
} else {
result.innerHTML = images.map(image => `
<div class="card">
<a href="${image.url}" target="_blank"><img src="${image.url}" alt="generated image"></a>
<div class="meta">${escapeHtml(image.name)}</div>
<div class="inline-links">
<a href="${image.url}" download>下载图片</a>
<a href="${image.url}" target="_blank">新窗口打开</a>
</div>
</div>
`).join("");
}
updateDownloadButton(images);
const sizeLabel = data.api_size && data.api_size !== data.size ? `${data.size} -> ${data.api_size}` : data.size;
showJsonDetails(
`本次生成：${data.request_mode} / ${data.request_type || "unknown"} / ${data.mode} / ${sizeLabel}`,
data.response_preview,
data.task_response_preview,
data.response_file,
data.task_response_file,
);
}
function renderHistory(items) {
state.historyItems = items || [];
const list = document.querySelector("#historyList");
if (!state.historyItems.length) {
list.className = "empty";
list.textContent = "暂无历史记录";
return;
}
list.className = "";
list.innerHTML = state.historyItems.map((item, index) => {
const thumbs = (item.images || []).map(image => `
<a href="${image.url}" target="_blank">
<img src="${image.url}" alt="history image">
</a>
`).join("");
const downloadLink = buildDownloadAllUrl(item.images || []);
return `
<div class="history-item">
<h3>${escapeHtml(item.prompt || "未命名 prompt")}</h3>
<div class="meta">${escapeHtml(item.created_at || "")} · 输入 ${escapeHtml(item.request_mode || "")} · 接口 ${escapeHtml(item.request_type || "")} · 返回模式 ${escapeHtml(item.result_mode || "")} · 尺寸 ${escapeHtml(item.api_size && item.api_size !== item.size ? `${item.size} -> ${item.api_size}` : item.size || "")}</div>
<div class="meta">模型 ${escapeHtml(item.model || "")} · Base URL ${escapeHtml(item.base_url || "")} · 风格 ${escapeHtml(item.request_style || "")}</div>
<div class="meta">数量 ${escapeHtml(item.n || 0)} · 参考图 ${escapeHtml(item.ref_count || 0)} 张</div>
<div class="toolbar">
<button type="button" class="secondary history-json" data-index="${index}">查看 JSON</button>
${downloadLink === "#" ? "" : `<a class="button secondary" href="${downloadLink}">下载该次全部图片</a>`}
</div>
${thumbs ? `<div class="thumbs">${thumbs}</div>` : '<div class="empty">该记录暂无可展示图片</div>'}
</div>
`;
}).join("");
list.querySelectorAll(".history-json").forEach(button => {
button.addEventListener("click", () => {
const index = Number(button.dataset.index);
const item = state.historyItems[index];
showJsonDetails(
`历史记录：${item.prompt || "未命名 prompt"}`,
item.response_preview,
item.task_response_preview,
item.response_file,
item.task_response_file,
);
});
});
}
async function refreshHistory() {
const response = await fetch("/api/history");
const data = await response.json();
if (!response.ok || !data.ok) throw new Error(data.error || "读取历史失败");
renderHistory(data.items || []);
}
async function refreshRuntimeConfig() {
const response = await fetch("/api/runtime-config");
const data = await response.json();
if (!response.ok || !data.ok) throw new Error(data.error || "读取运行时配置失败");
renderRuntimeConfig(data.config || null);
}
async function generate() {
const button = document.querySelector("#generate");
const status = document.querySelector("#status");
button.disabled = true;
document.querySelector("#result").innerHTML = "";
status.textContent = "正在提交请求，请等待……";
try {
const prompt = document.querySelector("#prompt").value.trim();
const textRefs = document.querySelector("#refs").value.split(/\r?\n/).map(v => v.trim()).filter(Boolean);
const fileRefs = await Promise.all([...document.querySelector("#files").files].map(fileToDataURL));
const allRefs = [...textRefs, ...fileRefs];
if (!prompt) throw new Error("请输入 prompt");
const response = await fetch("/api/generate", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
prompt,
size: document.querySelector("#size").value,
n: Number(document.querySelector("#count").value || 1),
image_refs: allRefs,
timeout: Number(document.querySelector("#timeout").value || 300),
initial_poll_delay: Number(document.querySelector("#initialPollDelay").value || 10),
poll_interval: Number(document.querySelector("#pollInterval").value || 5),
max_wait: Number(document.querySelector("#maxWait").value || 300)
})
});
const data = await response.json();
if (!response.ok || !data.ok) throw new Error(data.error || "生成失败");
const sizeLabel = data.api_size && data.api_size !== data.size ? `${data.size} -> ${data.api_size}` : data.size;
status.textContent = `完成：${data.mode}\n输入：${data.request_mode}\n接口：${data.request_type || "unknown"}\n模型：${data.model || "unknown"}\nBase URL：${data.base_url || "unknown"}\n风格：${data.request_style || "unknown"}\n尺寸：${sizeLabel}\nJSON：${data.response_file ? data.response_file.name : '无'}`;
if (data.task_response_file) {
status.textContent += `\nTask JSON：${data.task_response_file.name}`;
}
if (data.revised_prompts && data.revised_prompts.length) {
status.textContent += `\nRevised prompt：${data.revised_prompts.join("\n")}`;
}
renderResult(data);
await refreshHistory();
} catch (error) {
updateDownloadButton([]);
status.textContent = `错误：${error.message}`;
} finally {
button.disabled = false;
}
}
document.querySelector("#generate").addEventListener("click", generate);
document.querySelector("#refreshHistory").addEventListener("click", async () => {
try {
await refreshHistory();
} catch (error) {
document.querySelector("#status").textContent = `错误：${error.message}`;
}
});
document.querySelectorAll("#exampleChips .chip").forEach(button => {
button.addEventListener("click", () => {
document.querySelector("#prompt").value = button.dataset.prompt || "";
document.querySelector("#size").value = button.dataset.size || "16:9";
});
});
refreshHistory().catch(error => {
document.querySelector("#status").textContent = `历史记录加载失败：${error.message}`;
});
refreshRuntimeConfig().catch(error => {
document.querySelector("#runtimeConfig").textContent = `运行时配置加载失败：${error.message}`;
});
</script>
</body>
</html>
"""
def json_bytes(data: dict[str, Any]) -> bytes:
return json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8")
def ensure_output_dir() -> None:
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
def safe_output_path(name: str) -> Path:
candidate = OUTPUT_DIR / Path(name).name
resolved = candidate.resolve()
output_root = OUTPUT_DIR.resolve()
if resolved.parent != output_root:
raise ValueError("invalid file name")
return resolved
def build_file_meta(path_value: str | Path | None) -> dict[str, str] | None:
if not path_value:
return None
path = Path(path_value)
name = path.name
return {
"name": name,
"path": str(path),
"url": f"/outputs/{urllib.parse.quote(name)}",
}
def sanitize_for_preview(value: Any) -> Any:
if isinstance(value, dict):
return {key: sanitize_for_preview(item) for key, item in value.items()}
if isinstance(value, list):
if len(value) > 20:
trimmed = value[:20]
return [sanitize_for_preview(item) for item in trimmed] + [f"... truncated {len(value) - 20} more items"]
return [sanitize_for_preview(item) for item in value]
if isinstance(value, str) and len(value) > MAX_PREVIEW_STRING:
return f"{value[:MAX_PREVIEW_STRING]}... <truncated {len(value) - MAX_PREVIEW_STRING} chars>"
return value
def load_json_preview(path_value: str | Path | None) -> Any | None:
if not path_value:
return None
path = Path(path_value)
if not path.exists():
return None
try:
return sanitize_for_preview(json.loads(path.read_text(encoding="utf-8")))
except Exception:
return None
def append_history(item: dict[str, Any]) -> None:
ensure_output_dir()
with HISTORY_FILE.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(item, ensure_ascii=False) + "\n")
def load_history(limit: int = MAX_HISTORY_ITEMS) -> list[dict[str, Any]]:
if not HISTORY_FILE.exists():
return []
items: list[dict[str, Any]] = []
for line in HISTORY_FILE.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line:
continue
try:
items.append(json.loads(line))
except json.JSONDecodeError:
continue
return list(reversed(items[-limit:]))
def build_history_entry(
request_mode: str,
prompt: str,
size: str,
n: int,
ref_count: int,
images: list[dict[str, str]],
result: dict[str, Any],
response_preview: Any | None,
task_response_preview: Any | None,
) -> dict[str, Any]:
return {
"created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"request_mode": request_mode,
"request_type": result.get("request_type"),
"result_mode": result.get("mode"),
"base_url": result.get("base_url"),
"model": result.get("model"),
"request_style": result.get("request_style"),
"prompt": prompt,
"size": result.get("requested_size", size),
"api_size": result.get("api_size"),
"n": n,
"ref_count": ref_count,
"images": images,
"response_file": build_file_meta(result.get("response_path")),
"task_response_file": build_file_meta(result.get("task_response_path")),
"response_preview": response_preview,
"task_response_preview": task_response_preview,
}
class WebHandler(BaseHTTPRequestHandler):
server_version = "GPTImage2Web/2.0"
def log_message(self, format: str, *args: Any) -> None:
print(f"{self.address_string()} - {format % args}")
def send_body(self, status: int, body: bytes, content_type: str, extra_headers: dict[str, str] | None = None) -> None:
self.send_response(status)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
if extra_headers:
for key, value in extra_headers.items():
self.send_header(key, value)
self.end_headers()
self.wfile.write(body)
def send_json(self, status: int, data: dict[str, Any]) -> None:
self.send_body(status, json_bytes(data), "application/json; charset=utf-8")
def do_GET(self) -> None:
parsed = urllib.parse.urlparse(self.path)
if parsed.path in {"/", "/index.html"}:
self.send_body(200, HTML_PAGE.encode("utf-8"), "text/html; charset=utf-8")
return
if parsed.path == "/health":
self.send_json(200, {"ok": True})
return
if parsed.path == "/api/runtime-config":
self.handle_runtime_config()
return
if parsed.path == "/api/history":
self.handle_history(parsed.query)
return
if parsed.path == "/api/download-all":
self.handle_download_all(parsed.query)
return
if parsed.path.startswith("/outputs/"):
self.serve_output(parsed.path)
return
self.send_json(404, {"ok": False, "error": "not found"})
def do_POST(self) -> None:
parsed = urllib.parse.urlparse(self.path)
if parsed.path != "/api/generate":
self.send_json(404, {"ok": False, "error": "not found"})
return
try:
length = int(self.headers.get("Content-Length", "0"))
if length > 80 * 1024 * 1024:
self.send_json(413, {"ok": False, "error": "request too large"})
return
payload = json.loads(self.rfile.read(length).decode("utf-8"))
prompt = str(payload.get("prompt", "")).strip()
size = str(payload.get("size", "16:9"))
n = int(payload.get("n", 1))
image_refs = [str(item) for item in payload.get("image_refs", []) if str(item).strip()]
request_mode = "prompt+images" if image_refs else "prompt-only"
if not prompt:
raise ValueError("请输入 prompt")
result = generate_image.generate_image(
prompt=prompt,
size=size,
n=n,
image_refs=image_refs,
output_dir=str(OUTPUT_DIR),
timeout=int(payload.get("timeout", 300)),
initial_poll_delay=float(payload.get("initial_poll_delay", 10)),
poll_interval=float(payload.get("poll_interval", 5)),
max_wait=float(payload.get("max_wait", 300)),
)
except (ValueError, generate_image.ImageGenerationError) as error:
self.send_json(400, {"ok": False, "error": str(error)})
return
except Exception as error:
self.send_json(500, {"ok": False, "error": str(error)})
return
images = [build_file_meta(path) for path in result.get("images", [])]
images = [item for item in images if item is not None]
response_preview = load_json_preview(result.get("response_path"))
task_response_preview = load_json_preview(result.get("task_response_path"))
history_entry = build_history_entry(
request_mode=request_mode,
prompt=prompt,
size=size,
n=n,
ref_count=len(image_refs),
images=images,
result=result,
response_preview=response_preview,
task_response_preview=task_response_preview,
)
append_history(history_entry)
self.send_json(200, {
"ok": True,
"request_mode": request_mode,
"request_type": result.get("request_type"),
"mode": result.get("mode"),
"base_url": result.get("base_url"),
"model": result.get("model"),
"request_style": result.get("request_style"),
"size": result.get("requested_size", size),
"api_size": result.get("api_size"),
"n": n,
"images": images,
"response_file": build_file_meta(result.get("response_path")),
"task_response_file": build_file_meta(result.get("task_response_path")),
"response_preview": response_preview,
"task_response_preview": task_response_preview,
"revised_prompts": result.get("revised_prompts", []),
})
def handle_history(self, query: str) -> None:
params = urllib.parse.parse_qs(query)
try:
limit = int(params.get("limit", [str(MAX_HISTORY_ITEMS)])[0])
except ValueError:
limit = MAX_HISTORY_ITEMS
limit = max(1, min(limit, 100))
self.send_json(200, {"ok": True, "items": load_history(limit)})
def handle_runtime_config(self) -> None:
self.send_json(200, {"ok": True, "config": runtime_config.runtime_config_summary(RUNTIME_CONFIG)})
def handle_download_all(self, query: str) -> None:
params = urllib.parse.parse_qs(query)
raw_names = params.get("names", [""])[0]
names = [Path(name).name for name in raw_names.split(",") if name.strip()]
if not names:
self.send_json(400, {"ok": False, "error": "no image names provided"})
return
try:
paths = [safe_output_path(name) for name in names]
except ValueError as error:
self.send_json(400, {"ok": False, "error": str(error)})
return
existing = [path for path in paths if path.exists() and path.is_file()]
if not existing:
self.send_json(404, {"ok": False, "error": "no output files found"})
return
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
for path in existing:
archive.write(path, arcname=path.name)
body = buffer.getvalue()
filename = f"gpt-image-2-bundle-{datetime.now().strftime('%Y%m%d-%H%M%S')}.zip"
self.send_body(
200,
body,
"application/zip",
{
"Content-Disposition": f'attachment; filename="{filename}"',
},
)
def serve_output(self, request_path: str) -> None:
name = Path(urllib.parse.unquote(request_path.removeprefix("/outputs/"))).name
try:
resolved = safe_output_path(name)
if not resolved.exists() or not resolved.is_file():
self.send_json(404, {"ok": False, "error": "output not found"})
return
content_type = mimetypes.guess_type(str(resolved))[0] or "application/octet-stream"
self.send_body(200, resolved.read_bytes(), content_type)
except Exception as error:
self.send_json(500, {"ok": False, "error": html.escape(str(error))})
def main() -> int:
parser = argparse.ArgumentParser(description="Start the GPT Image 2 local web UI")
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", default=7862, type=int)
parser.add_argument("--open", action="store_true", help="Open browser after startup")
args = parser.parse_args()
ensure_output_dir()
server = ThreadingHTTPServer((args.host, args.port), WebHandler)
url = f"http://{args.host}:{args.port}/"
print(f"GPT Image 2 web tool is running: {url}")
print(f"Execution root: {RUNTIME_CONFIG.execution_root}")
print(f"Project config dir: {RUNTIME_CONFIG.project_config_dir}")
print(f"Skill fallback config dir: {RUNTIME_CONFIG.skill_config_dir}")
print(f"Base URL: {RUNTIME_CONFIG.base_url} ({RUNTIME_CONFIG.base_url_source})")
print(f"Output dir: {OUTPUT_DIR}")
print("Press Ctrl+C to stop.")
if args.open:
threading.Timer(1.0, lambda: webbrowser.open(url)).start()
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nStopped.")
finally:
server.server_close()
return 0
if __name__ == "__main__":
raise SystemExit(main())