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""" GPT Image 2 生图工具

GPT Image 2 生图工具

优先读取当前执行目录的 .env 和 config/，再回退到 skill 自带 config/。输入 prompt，可选添加一张或多张参考图，服务端会自动选择纯文本生成或带图编辑，并兼容 apicodex 风格与 OpenAI-compatible 网关。

正在读取运行时配置…

Prompt

参考图 URL / data URI（每行一个，可选）

尺寸

数量

本地参考图（可多选，可选）

本地图片会在浏览器中转换为 data URI 后提交给本机服务。尺寸既可填比例，如 2:3，也可填像素，如 1024x1536。

高级参数：超时和异步轮询

请求超时（秒）

首次轮询等待（秒）

轮询间隔（秒）

最长等待（秒）

JSON 详情面板

生成完成后，可在这里查看接口返回的 JSON 预览。

响应 JSON 预览

暂无

任务 JSON 预览

暂无

历史生成记录

显示最近 20 条

暂无历史记录

""" 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]}... " 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())