imagegen Skill

1#!/usr/bin/env python32"""Fallback CLI for explicit image generation or editing with GPT Image models.3 4Used only when the user explicitly opts into CLI fallback mode, or when explicit5transparent output requires the `gpt-image-1.5` fallback path.6 7Defaults to gpt-image-2 and a structured prompt augmentation workflow.8"""9 10from __future__ import annotations11 12import argparse13import asyncio14import base6415import json16import os17from pathlib import Path18import re19import sys20import time21from typing import Any, Dict, Iterable, List, Optional, Tuple22 23from io import BytesIO24 25DEFAULT_MODEL = "gpt-image-2"26DEFAULT_SIZE = "auto"27DEFAULT_QUALITY = "medium"28DEFAULT_OUTPUT_FORMAT = "png"29DEFAULT_CONCURRENCY = 530DEFAULT_DOWNSCALE_SUFFIX = "-web"31DEFAULT_OUTPUT_PATH = "output/imagegen/output.png"32GPT_IMAGE_MODEL_PREFIX = "gpt-image-"33 34ALLOWED_LEGACY_SIZES = {"1024x1024", "1536x1024", "1024x1536", "auto"}35ALLOWED_QUALITIES = {"low", "medium", "high", "auto"}36ALLOWED_BACKGROUNDS = {"transparent", "opaque", "auto", None}37ALLOWED_INPUT_FIDELITIES = {"low", "high", None}38 39GPT_IMAGE_2_MODEL = "gpt-image-2"40GPT_IMAGE_2_MIN_PIXELS = 655_36041GPT_IMAGE_2_MAX_PIXELS = 8_294_40042GPT_IMAGE_2_MAX_EDGE = 384043GPT_IMAGE_2_MAX_RATIO = 3.044 45MAX_IMAGE_BYTES = 50 * 1024 * 102446MAX_BATCH_JOBS = 50047 48 49def _die(message: str, code: int = 1) -> None:50    print(f"Error: {message}", file=sys.stderr)51    raise SystemExit(code)52 53 54def _warn(message: str) -> None:55    print(f"Warning: {message}", file=sys.stderr)56 57 58def _dependency_hint(package: str, *, upgrade: bool = False) -> str:59    command = f"uv pip install {'-U ' if upgrade else ''}{package}"60    return (61        "Activate the repo-selected environment first, then install it with "62        f"`{command}`. If this repo uses a local virtualenv, start with "63        "`source .venv/bin/activate`; otherwise use this repo's configured shared fallback "64        "environment. If your project declares dependencies, prefer that project's normal "65        "`uv sync` flow."66    )67 68 69def _ensure_api_key(dry_run: bool) -> None:70    if os.getenv("OPENAI_API_KEY"):71        print("OPENAI_API_KEY is set.", file=sys.stderr)72        return73    if dry_run:74        _warn("OPENAI_API_KEY is not set; dry-run only.")75        return76    _die("OPENAI_API_KEY is not set. Export it before running.")77 78 79def _read_prompt(prompt: Optional[str], prompt_file: Optional[str]) -> str:80    if prompt and prompt_file:81        _die("Use --prompt or --prompt-file, not both.")82    if prompt_file:83        path = Path(prompt_file)84        if not path.exists():85            _die(f"Prompt file not found: {path}")86        return path.read_text(encoding="utf-8").strip()87    if prompt:88        return prompt.strip()89    _die("Missing prompt. Use --prompt or --prompt-file.")90    return ""  # unreachable91 92 93def _check_image_paths(paths: Iterable[str]) -> List[Path]:94    resolved: List[Path] = []95    for raw in paths:96        path = Path(raw)97        if not path.exists():98            _die(f"Image file not found: {path}")99        if path.stat().st_size > MAX_IMAGE_BYTES:100            _warn(f"Image exceeds 50MB limit: {path}")101        resolved.append(path)102    return resolved103 104 105def _normalize_output_format(fmt: Optional[str]) -> str:106    if not fmt:107        return DEFAULT_OUTPUT_FORMAT108    fmt = fmt.lower()109    if fmt not in {"png", "jpeg", "jpg", "webp"}:110        _die("output-format must be png, jpeg, jpg, or webp.")111    return "jpeg" if fmt == "jpg" else fmt112 113 114def _parse_size(size: str) -> Optional[Tuple[int, int]]:115    match = re.fullmatch(r"([1-9][0-9]*)x([1-9][0-9]*)", size)116    if not match:117        return None118    return int(match.group(1)), int(match.group(2))119 120 121def _validate_gpt_image_2_size(size: str) -> None:122    if size == "auto":123        return124 125    parsed = _parse_size(size)126    if parsed is None:127        _die("size must be auto or WIDTHxHEIGHT, for example 1024x1024.")128 129    width, height = parsed130    max_edge = max(width, height)131    min_edge = min(width, height)132    total_pixels = width * height133 134    if max_edge > GPT_IMAGE_2_MAX_EDGE:135        _die(136            "gpt-image-2 size maximum edge length must be less than or equal to 3840px."137        )138    if width % 16 != 0 or height % 16 != 0:139        _die("gpt-image-2 size width and height must be multiples of 16px.")140    if max_edge / min_edge > GPT_IMAGE_2_MAX_RATIO:141        _die("gpt-image-2 size long edge to short edge ratio must not exceed 3:1.")142    if total_pixels < GPT_IMAGE_2_MIN_PIXELS or total_pixels > GPT_IMAGE_2_MAX_PIXELS:143        _die(144            "gpt-image-2 size total pixels must be at least 655,360 and no more than 8,294,400."145        )146 147 148def _validate_size(size: str, model: str) -> None:149    if model == GPT_IMAGE_2_MODEL:150        _validate_gpt_image_2_size(size)151        return152 153    if size not in ALLOWED_LEGACY_SIZES:154        _die(155            "size must be one of 1024x1024, 1536x1024, 1024x1536, or auto for this GPT Image model."156        )157 158 159def _validate_quality(quality: str) -> None:160    if quality not in ALLOWED_QUALITIES:161        _die("quality must be one of low, medium, high, or auto.")162 163 164def _validate_background(background: Optional[str]) -> None:165    if background not in ALLOWED_BACKGROUNDS:166        _die("background must be one of transparent, opaque, or auto.")167 168 169def _validate_input_fidelity(input_fidelity: Optional[str]) -> None:170    if input_fidelity not in ALLOWED_INPUT_FIDELITIES:171        _die("input-fidelity must be one of low or high.")172 173 174def _validate_model(model: str) -> None:175    if not model.startswith(GPT_IMAGE_MODEL_PREFIX):176        _die(177            "model must be a GPT Image model (for example gpt-image-1.5, gpt-image-1, or gpt-image-1-mini)."178        )179 180 181def _validate_transparency(background: Optional[str], output_format: str) -> None:182    if background == "transparent" and output_format not in {"png", "webp"}:183        _die("transparent background requires output-format png or webp.")184 185 186def _validate_model_specific_options(187    *,188    model: str,189    background: Optional[str],190    input_fidelity: Optional[str] = None,191) -> None:192    if model != GPT_IMAGE_2_MODEL:193        return194    if background == "transparent":195        _die(196            "transparent backgrounds are not supported in gpt-image-2, the latest model. "197            "Use --model gpt-image-1.5 --background transparent --output-format png instead."198        )199    if input_fidelity is not None:200        _die(201            "input_fidelity is not supported in gpt-image-2 because image inputs always use high fidelity for this model."202        )203 204 205def _validate_generate_payload(payload: Dict[str, Any]) -> None:206    model = str(payload.get("model", DEFAULT_MODEL))207    _validate_model(model)208    n = int(payload.get("n", 1))209    if n < 1 or n > 10:210        _die("n must be between 1 and 10")211    size = str(payload.get("size", DEFAULT_SIZE))212    quality = str(payload.get("quality", DEFAULT_QUALITY))213    background = payload.get("background")214    _validate_size(size, model)215    _validate_quality(quality)216    _validate_background(background)217    _validate_model_specific_options(model=model, background=background)218    oc = payload.get("output_compression")219    if oc is not None and not (0 <= int(oc) <= 100):220        _die("output_compression must be between 0 and 100")221 222 223def _build_output_paths(224    out: str,225    output_format: str,226    count: int,227    out_dir: Optional[str],228) -> List[Path]:229    ext = "." + output_format230 231    if out_dir:232        out_base = Path(out_dir)233        out_base.mkdir(parents=True, exist_ok=True)234        return [out_base / f"image_{i}{ext}" for i in range(1, count + 1)]235 236    out_path = Path(out)237    if out_path.exists() and out_path.is_dir():238        out_path.mkdir(parents=True, exist_ok=True)239        return [out_path / f"image_{i}{ext}" for i in range(1, count + 1)]240 241    if out_path.suffix == "":242        out_path = out_path.with_suffix(ext)243    elif output_format and out_path.suffix.lstrip(".").lower() != output_format:244        _warn(245            f"Output extension {out_path.suffix} does not match output-format {output_format}."246        )247 248    if count == 1:249        return [out_path]250 251    return [252        out_path.with_name(f"{out_path.stem}-{i}{out_path.suffix}")253        for i in range(1, count + 1)254    ]255 256 257def _augment_prompt(args: argparse.Namespace, prompt: str) -> str:258    fields = _fields_from_args(args)259    return _augment_prompt_fields(args.augment, prompt, fields)260 261 262def _augment_prompt_fields(263    augment: bool, prompt: str, fields: Dict[str, Optional[str]]264) -> str:265    if not augment:266        return prompt267 268    sections: List[str] = []269    if fields.get("use_case"):270        sections.append(f"Use case: {fields['use_case']}")271    sections.append(f"Primary request: {prompt}")272    if fields.get("scene"):273        sections.append(f"Scene/background: {fields['scene']}")274    if fields.get("subject"):275        sections.append(f"Subject: {fields['subject']}")276    if fields.get("style"):277        sections.append(f"Style/medium: {fields['style']}")278    if fields.get("composition"):279        sections.append(f"Composition/framing: {fields['composition']}")280    if fields.get("lighting"):281        sections.append(f"Lighting/mood: {fields['lighting']}")282    if fields.get("palette"):283        sections.append(f"Color palette: {fields['palette']}")284    if fields.get("materials"):285        sections.append(f"Materials/textures: {fields['materials']}")286    if fields.get("text"):287        sections.append(f'Text (verbatim): "{fields["text"]}"')288    if fields.get("constraints"):289        sections.append(f"Constraints: {fields['constraints']}")290    if fields.get("negative"):291        sections.append(f"Avoid: {fields['negative']}")292 293    return "\n".join(sections)294 295 296def _fields_from_args(args: argparse.Namespace) -> Dict[str, Optional[str]]:297    return {298        "use_case": getattr(args, "use_case", None),299        "scene": getattr(args, "scene", None),300        "subject": getattr(args, "subject", None),301        "style": getattr(args, "style", None),302        "composition": getattr(args, "composition", None),303        "lighting": getattr(args, "lighting", None),304        "palette": getattr(args, "palette", None),305        "materials": getattr(args, "materials", None),306        "text": getattr(args, "text", None),307        "constraints": getattr(args, "constraints", None),308        "negative": getattr(args, "negative", None),309    }310 311 312def _print_request(payload: dict) -> None:313    print(json.dumps(payload, indent=2, sort_keys=True))314 315 316def _decode_and_write(images: List[str], outputs: List[Path], force: bool) -> None:317    for idx, image_b64 in enumerate(images):318        if idx >= len(outputs):319            break320        out_path = outputs[idx]321        if out_path.exists() and not force:322            _die(f"Output already exists: {out_path} (use --force to overwrite)")323        out_path.parent.mkdir(parents=True, exist_ok=True)324        out_path.write_bytes(base64.b64decode(image_b64))325        print(f"Wrote {out_path}")326 327 328def _derive_downscale_path(path: Path, suffix: str) -> Path:329    if suffix and not suffix.startswith("-") and not suffix.startswith("_"):330        suffix = "-" + suffix331    return path.with_name(f"{path.stem}{suffix}{path.suffix}")332 333 334def _downscale_image_bytes(335    image_bytes: bytes, *, max_dim: int, output_format: str336) -> bytes:337    try:338        from PIL import Image339    except Exception:340        _die(f"Downscaling requires Pillow. {_dependency_hint('pillow')}")341 342    if max_dim < 1:343        _die("--downscale-max-dim must be >= 1")344 345    with Image.open(BytesIO(image_bytes)) as img:346        img.load()347        w, h = img.size348        scale = min(1.0, float(max_dim) / float(max(w, h)))349        target = (max(1, int(round(w * scale))), max(1, int(round(h * scale))))350 351        resized = (352            img if target == (w, h) else img.resize(target, Image.Resampling.LANCZOS)353        )354 355        fmt = output_format.lower()356        if fmt == "jpg":357            fmt = "jpeg"358 359        if fmt == "jpeg":360            if resized.mode in ("RGBA", "LA") or (361                "transparency" in getattr(resized, "info", {})362            ):363                bg = Image.new("RGB", resized.size, (255, 255, 255))364                bg.paste(365                    resized.convert("RGBA"), mask=resized.convert("RGBA").split()[-1]366                )367                resized = bg368            else:369                resized = resized.convert("RGB")370 371        out = BytesIO()372        resized.save(out, format=fmt.upper())373        return out.getvalue()374 375 376def _decode_write_and_downscale(377    images: List[str],378    outputs: List[Path],379    *,380    force: bool,381    downscale_max_dim: Optional[int],382    downscale_suffix: str,383    output_format: str,384) -> None:385    for idx, image_b64 in enumerate(images):386        if idx >= len(outputs):387            break388        out_path = outputs[idx]389        if out_path.exists() and not force:390            _die(f"Output already exists: {out_path} (use --force to overwrite)")391        out_path.parent.mkdir(parents=True, exist_ok=True)392 393        raw = base64.b64decode(image_b64)394        out_path.write_bytes(raw)395        print(f"Wrote {out_path}")396 397        if downscale_max_dim is None:398            continue399 400        derived = _derive_downscale_path(out_path, downscale_suffix)401        if derived.exists() and not force:402            _die(f"Output already exists: {derived} (use --force to overwrite)")403        derived.parent.mkdir(parents=True, exist_ok=True)404        resized = _downscale_image_bytes(405            raw, max_dim=downscale_max_dim, output_format=output_format406        )407        derived.write_bytes(resized)408        print(f"Wrote {derived}")409 410 411def _create_client():412    try:413        from openai import OpenAI414    except ImportError:415        _die(416            f"openai SDK not installed in the active environment. {_dependency_hint('openai')}"417        )418    return OpenAI()419 420 421def _create_async_client():422    try:423        from openai import AsyncOpenAI424    except ImportError:425        try:426            import openai as _openai  # noqa: F401427        except ImportError:428            _die(429                f"openai SDK not installed in the active environment. {_dependency_hint('openai')}"430            )431        _die(432            "AsyncOpenAI not available in this openai SDK version. "433            f"{_dependency_hint('openai', upgrade=True)}"434        )435    return AsyncOpenAI()436 437 438def _slugify(value: str) -> str:439    value = value.strip().lower()440    value = re.sub(r"[^a-z0-9]+", "-", value)441    value = re.sub(r"-{2,}", "-", value).strip("-")442    return value[:60] if value else "job"443 444 445def _normalize_job(job: Any, idx: int) -> Dict[str, Any]:446    if isinstance(job, str):447        prompt = job.strip()448        if not prompt:449            _die(f"Empty prompt at job {idx}")450        return {"prompt": prompt}451    if isinstance(job, dict):452        if "prompt" not in job or not str(job["prompt"]).strip():453            _die(f"Missing prompt for job {idx}")454        return job455    _die(f"Invalid job at index {idx}: expected string or object.")456    return {}  # unreachable457 458 459def _read_jobs_jsonl(path: str) -> List[Dict[str, Any]]:460    p = Path(path)461    if not p.exists():462        _die(f"Input file not found: {p}")463    jobs: List[Dict[str, Any]] = []464    for line_no, raw in enumerate(p.read_text(encoding="utf-8").splitlines(), start=1):465        line = raw.strip()466        if not line or line.startswith("#"):467            continue468        try:469            item: Any470            if line.startswith("{"):471                item = json.loads(line)472            else:473                item = line474            jobs.append(_normalize_job(item, idx=line_no))475        except json.JSONDecodeError as exc:476            _die(f"Invalid JSON on line {line_no}: {exc}")477    if not jobs:478        _die("No jobs found in input file.")479    if len(jobs) > MAX_BATCH_JOBS:480        _die(f"Too many jobs ({len(jobs)}). Max is {MAX_BATCH_JOBS}.")481    return jobs482 483 484def _merge_non_null(dst: Dict[str, Any], src: Dict[str, Any]) -> Dict[str, Any]:485    merged = dict(dst)486    for k, v in src.items():487        if v is not None:488            merged[k] = v489    return merged490 491 492def _job_output_paths(493    *,494    out_dir: Path,495    output_format: str,496    idx: int,497    prompt: str,498    n: int,499    explicit_out: Optional[str],500) -> List[Path]:501    out_dir.mkdir(parents=True, exist_ok=True)502    ext = "." + output_format503 504    if explicit_out:505        base = Path(explicit_out)506        if base.suffix == "":507            base = base.with_suffix(ext)508        elif base.suffix.lstrip(".").lower() != output_format:509            _warn(510                f"Job {idx}: output extension {base.suffix} does not match output-format {output_format}."511            )512        base = out_dir / base.name513    else:514        slug = _slugify(prompt[:80])515        base = out_dir / f"{idx:03d}-{slug}{ext}"516 517    if n == 1:518        return [base]519    return [base.with_name(f"{base.stem}-{i}{base.suffix}") for i in range(1, n + 1)]520 521 522def _extract_retry_after_seconds(exc: Exception) -> Optional[float]:523    # Best-effort: openai SDK errors vary by version. Prefer a conservative fallback.524    for attr in ("retry_after", "retry_after_seconds"):525        val = getattr(exc, attr, None)526        if isinstance(val, (int, float)) and val >= 0:527            return float(val)528    msg = str(exc)529    m = re.search(r"retry[- ]after[:= ]+([0-9]+(?:\\.[0-9]+)?)", msg, re.IGNORECASE)530    if m:531        try:532            return float(m.group(1))533        except Exception:534            return None535    return None536 537 538def _is_rate_limit_error(exc: Exception) -> bool:539    name = exc.__class__.__name__.lower()540    if "ratelimit" in name or "rate_limit" in name:541        return True542    msg = str(exc).lower()543    return "429" in msg or "rate limit" in msg or "too many requests" in msg544 545 546def _is_transient_error(exc: Exception) -> bool:547    if _is_rate_limit_error(exc):548        return True549    name = exc.__class__.__name__.lower()550    if "timeout" in name or "timedout" in name or "tempor" in name:551        return True552    msg = str(exc).lower()553    return "timeout" in msg or "timed out" in msg or "connection reset" in msg554 555 556async def _generate_one_with_retries(557    client: Any,558    payload: Dict[str, Any],559    *,560    attempts: int,561    job_label: str,562) -> Any:563    last_exc: Optional[Exception] = None564    for attempt in range(1, attempts + 1):565        try:566            return await client.images.generate(**payload)567        except Exception as exc:568            last_exc = exc569            if not _is_transient_error(exc):570                raise571            if attempt == attempts:572                raise573            sleep_s = _extract_retry_after_seconds(exc)574            if sleep_s is None:575                sleep_s = min(60.0, 2.0**attempt)576            print(577                f"{job_label} attempt {attempt}/{attempts} failed ({exc.__class__.__name__}); retrying in {sleep_s:.1f}s",578                file=sys.stderr,579            )580            await asyncio.sleep(sleep_s)581    raise last_exc or RuntimeError("unknown error")582 583 584async def _run_generate_batch(args: argparse.Namespace) -> int:585    jobs = _read_jobs_jsonl(args.input)586    out_dir = Path(args.out_dir)587 588    base_fields = _fields_from_args(args)589    base_payload = {590        "model": args.model,591        "n": args.n,592        "size": args.size,593        "quality": args.quality,594        "background": args.background,595        "output_format": args.output_format,596        "output_compression": args.output_compression,597        "moderation": args.moderation,598    }599 600    if args.dry_run:601        for i, job in enumerate(jobs, start=1):602            prompt = str(job["prompt"]).strip()603            fields = _merge_non_null(base_fields, job.get("fields", {}))604            # Allow flat job keys as well (use_case, scene, etc.)605            fields = _merge_non_null(606                fields, {k: job.get(k) for k in base_fields.keys()}607            )608            augmented = _augment_prompt_fields(args.augment, prompt, fields)609 610            job_payload = dict(base_payload)611            job_payload["prompt"] = augmented612            job_payload = _merge_non_null(613                job_payload, {k: job.get(k) for k in base_payload.keys()}614            )615            job_payload = {k: v for k, v in job_payload.items() if v is not None}616 617            _validate_generate_payload(job_payload)618            effective_output_format = _normalize_output_format(619                job_payload.get("output_format")620            )621            _validate_transparency(622                job_payload.get("background"), effective_output_format623            )624            job_payload["output_format"] = effective_output_format625 626            n = int(job_payload.get("n", 1))627            outputs = _job_output_paths(628                out_dir=out_dir,629                output_format=effective_output_format,630                idx=i,631                prompt=prompt,632                n=n,633                explicit_out=job.get("out"),634            )635            downscaled = None636            if args.downscale_max_dim is not None:637                downscaled = [638                    str(_derive_downscale_path(p, args.downscale_suffix))639                    for p in outputs640                ]641            _print_request(642                {643                    "endpoint": "/v1/images/generations",644                    "job": i,645                    "outputs": [str(p) for p in outputs],646                    "outputs_downscaled": downscaled,647                    **job_payload,648                }649            )650        return 0651 652    client = _create_async_client()653    sem = asyncio.Semaphore(args.concurrency)654 655    any_failed = False656 657    async def run_job(i: int, job: Dict[str, Any]) -> Tuple[int, Optional[str]]:658        nonlocal any_failed659        prompt = str(job["prompt"]).strip()660        job_label = f"[job {i}/{len(jobs)}]"661 662        fields = _merge_non_null(base_fields, job.get("fields", {}))663        fields = _merge_non_null(fields, {k: job.get(k) for k in base_fields.keys()})664        augmented = _augment_prompt_fields(args.augment, prompt, fields)665 666        payload = dict(base_payload)667        payload["prompt"] = augmented668        payload = _merge_non_null(payload, {k: job.get(k) for k in base_payload.keys()})669        payload = {k: v for k, v in payload.items() if v is not None}670 671        n = int(payload.get("n", 1))672        _validate_generate_payload(payload)673        effective_output_format = _normalize_output_format(payload.get("output_format"))674        _validate_transparency(payload.get("background"), effective_output_format)675        payload["output_format"] = effective_output_format676        outputs = _job_output_paths(677            out_dir=out_dir,678            output_format=effective_output_format,679            idx=i,680            prompt=prompt,681            n=n,682            explicit_out=job.get("out"),683        )684        try:685            async with sem:686                print(f"{job_label} starting", file=sys.stderr)687                started = time.time()688                result = await _generate_one_with_retries(689                    client,690                    payload,691                    attempts=args.max_attempts,692                    job_label=job_label,693                )694                elapsed = time.time() - started695                print(f"{job_label} completed in {elapsed:.1f}s", file=sys.stderr)696            images = [item.b64_json for item in result.data]697            _decode_write_and_downscale(698                images,699                outputs,700                force=args.force,701                downscale_max_dim=args.downscale_max_dim,702                downscale_suffix=args.downscale_suffix,703                output_format=effective_output_format,704            )705            return i, None706        except Exception as exc:707            any_failed = True708            print(f"{job_label} failed: {exc}", file=sys.stderr)709            if args.fail_fast:710                raise711            return i, str(exc)712 713    tasks = [714        asyncio.create_task(run_job(i, job)) for i, job in enumerate(jobs, start=1)715    ]716 717    try:718        await asyncio.gather(*tasks)719    except Exception:720        for t in tasks:721            if not t.done():722                t.cancel()723        raise724 725    return 1 if any_failed else 0726 727 728def _generate_batch(args: argparse.Namespace) -> None:729    exit_code = asyncio.run(_run_generate_batch(args))730    if exit_code:731        raise SystemExit(exit_code)732 733 734def _generate(args: argparse.Namespace) -> None:735    prompt = _read_prompt(args.prompt, args.prompt_file)736    prompt = _augment_prompt(args, prompt)737 738    payload = {739        "model": args.model,740        "prompt": prompt,741        "n": args.n,742        "size": args.size,743        "quality": args.quality,744        "background": args.background,745        "output_format": args.output_format,746        "output_compression": args.output_compression,747        "moderation": args.moderation,748    }749    payload = {k: v for k, v in payload.items() if v is not None}750 751    output_format = _normalize_output_format(args.output_format)752    _validate_transparency(args.background, output_format)753    payload["output_format"] = output_format754    output_paths = _build_output_paths(args.out, output_format, args.n, args.out_dir)755    downscaled = None756    if args.downscale_max_dim is not None:757        downscaled = [758            str(_derive_downscale_path(p, args.downscale_suffix)) for p in output_paths759        ]760 761    if args.dry_run:762        _print_request(763            {764                "endpoint": "/v1/images/generations",765                "outputs": [str(p) for p in output_paths],766                "outputs_downscaled": downscaled,767                **payload,768            }769        )770        return771 772    print(773        "Calling Image API (generation). This can take up to a couple of minutes.",774        file=sys.stderr,775    )776    started = time.time()777    client = _create_client()778    result = client.images.generate(**payload)779    elapsed = time.time() - started780    print(f"Generation completed in {elapsed:.1f}s.", file=sys.stderr)781 782    images = [item.b64_json for item in result.data]783    _decode_write_and_downscale(784        images,785        output_paths,786        force=args.force,787        downscale_max_dim=args.downscale_max_dim,788        downscale_suffix=args.downscale_suffix,789        output_format=output_format,790    )791 792 793def _edit(args: argparse.Namespace) -> None:794    prompt = _read_prompt(args.prompt, args.prompt_file)795    prompt = _augment_prompt(args, prompt)796 797    image_paths = _check_image_paths(args.image)798    mask_path = Path(args.mask) if args.mask else None799    if mask_path:800        if not mask_path.exists():801            _die(f"Mask file not found: {mask_path}")802        if mask_path.suffix.lower() != ".png":803            _warn(f"Mask should be a PNG with an alpha channel: {mask_path}")804        if mask_path.stat().st_size > MAX_IMAGE_BYTES:805            _warn(f"Mask exceeds 50MB limit: {mask_path}")806 807    payload = {808        "model": args.model,809        "prompt": prompt,810        "n": args.n,811        "size": args.size,812        "quality": args.quality,813        "background": args.background,814        "output_format": args.output_format,815        "output_compression": args.output_compression,816        "input_fidelity": args.input_fidelity,817        "moderation": args.moderation,818    }819    payload = {k: v for k, v in payload.items() if v is not None}820 821    output_format = _normalize_output_format(args.output_format)822    _validate_transparency(args.background, output_format)823    payload["output_format"] = output_format824    _validate_input_fidelity(args.input_fidelity)825    output_paths = _build_output_paths(args.out, output_format, args.n, args.out_dir)826    downscaled = None827    if args.downscale_max_dim is not None:828        downscaled = [829            str(_derive_downscale_path(p, args.downscale_suffix)) for p in output_paths830        ]831 832    if args.dry_run:833        payload_preview = dict(payload)834        payload_preview["image"] = [str(p) for p in image_paths]835        if mask_path:836            payload_preview["mask"] = str(mask_path)837        _print_request(838            {839                "endpoint": "/v1/images/edits",840                "outputs": [str(p) for p in output_paths],841                "outputs_downscaled": downscaled,842                **payload_preview,843            }844        )845        return846 847    print(848        f"Calling Image API (edit) with {len(image_paths)} image(s).",849        file=sys.stderr,850    )851    started = time.time()852    client = _create_client()853 854    with _open_files(image_paths) as image_files, _open_mask(mask_path) as mask_file:855        request = dict(payload)856        request["image"] = image_files if len(image_files) > 1 else image_files[0]857        if mask_file is not None:858            request["mask"] = mask_file859        result = client.images.edit(**request)860 861    elapsed = time.time() - started862    print(f"Edit completed in {elapsed:.1f}s.", file=sys.stderr)863    images = [item.b64_json for item in result.data]864    _decode_write_and_downscale(865        images,866        output_paths,867        force=args.force,868        downscale_max_dim=args.downscale_max_dim,869        downscale_suffix=args.downscale_suffix,870        output_format=output_format,871    )872 873 874def _open_files(paths: List[Path]):875    return _FileBundle(paths)876 877 878def _open_mask(mask_path: Optional[Path]):879    if mask_path is None:880        return _NullContext()881    return _SingleFile(mask_path)882 883 884class _NullContext:885    def __enter__(self):886        return None887 888    def __exit__(self, exc_type, exc, tb):889        return False890 891 892class _SingleFile:893    def __init__(self, path: Path):894        self._path = path895        self._handle = None896 897    def __enter__(self):898        self._handle = self._path.open("rb")899        return self._handle900 901    def __exit__(self, exc_type, exc, tb):902        if self._handle:903            try:904                self._handle.close()905            except Exception:906                pass907        return False908 909 910class _FileBundle:911    def __init__(self, paths: List[Path]):912        self._paths = paths913        self._handles: List[object] = []914 915    def __enter__(self):916        self._handles = [p.open("rb") for p in self._paths]917        return self._handles918 919    def __exit__(self, exc_type, exc, tb):920        for handle in self._handles:921            try:922                handle.close()923            except Exception:924                pass925        return False926 927 928def _add_shared_args(parser: argparse.ArgumentParser) -> None:929    parser.add_argument("--model", default=DEFAULT_MODEL)930    parser.add_argument("--prompt")931    parser.add_argument("--prompt-file")932    parser.add_argument("--n", type=int, default=1)933    parser.add_argument("--size", default=DEFAULT_SIZE)934    parser.add_argument("--quality", default=DEFAULT_QUALITY)935    parser.add_argument("--background")936    parser.add_argument("--output-format")937    parser.add_argument("--output-compression", type=int)938    parser.add_argument("--moderation")939    parser.add_argument("--out", default=DEFAULT_OUTPUT_PATH)940    parser.add_argument("--out-dir")941    parser.add_argument("--force", action="store_true")942    parser.add_argument("--dry-run", action="store_true")943    parser.add_argument("--augment", dest="augment", action="store_true")944    parser.add_argument("--no-augment", dest="augment", action="store_false")945    parser.set_defaults(augment=True)946 947    # Prompt augmentation hints948    parser.add_argument("--use-case")949    parser.add_argument("--scene")950    parser.add_argument("--subject")951    parser.add_argument("--style")952    parser.add_argument("--composition")953    parser.add_argument("--lighting")954    parser.add_argument("--palette")955    parser.add_argument("--materials")956    parser.add_argument("--text")957    parser.add_argument("--constraints")958    parser.add_argument("--negative")959 960    # Post-processing (optional): generate an additional downscaled copy for fast web loading.961    parser.add_argument("--downscale-max-dim", type=int)962    parser.add_argument("--downscale-suffix", default=DEFAULT_DOWNSCALE_SUFFIX)963 964 965def main() -> int:966    parser = argparse.ArgumentParser(967        description="Fallback CLI for explicit image generation or editing via GPT Image models"968    )969    subparsers = parser.add_subparsers(dest="command", required=True)970 971    gen_parser = subparsers.add_parser("generate", help="Create a new image")972    _add_shared_args(gen_parser)973    gen_parser.set_defaults(func=_generate)974 975    batch_parser = subparsers.add_parser(976        "generate-batch",977        help="Generate multiple prompts concurrently (JSONL input)",978    )979    _add_shared_args(batch_parser)980    batch_parser.add_argument(981        "--input", required=True, help="Path to JSONL file (one job per line)"982    )983    batch_parser.add_argument("--concurrency", type=int, default=DEFAULT_CONCURRENCY)984    batch_parser.add_argument("--max-attempts", type=int, default=3)985    batch_parser.add_argument("--fail-fast", action="store_true")986    batch_parser.set_defaults(func=_generate_batch)987 988    edit_parser = subparsers.add_parser("edit", help="Edit an existing image")989    _add_shared_args(edit_parser)990    edit_parser.add_argument("--image", action="append", required=True)991    edit_parser.add_argument("--mask")992    edit_parser.add_argument("--input-fidelity")993    edit_parser.set_defaults(func=_edit)994 995    args = parser.parse_args()996    if args.n < 1 or args.n > 10:997        _die("--n must be between 1 and 10")998    if getattr(args, "concurrency", 1) < 1 or getattr(args, "concurrency", 1) > 25:999        _die("--concurrency must be between 1 and 25")1000    if getattr(args, "max_attempts", 3) < 1 or getattr(args, "max_attempts", 3) > 10:1001        _die("--max-attempts must be between 1 and 10")1002    if args.output_compression is not None and not (1003        0 <= args.output_compression <= 1001004    ):1005        _die("--output-compression must be between 0 and 100")1006    if args.command == "generate-batch" and not args.out_dir:1007        _die("generate-batch requires --out-dir")1008    if (1009        getattr(args, "downscale_max_dim", None) is not None1010        and args.downscale_max_dim < 11011    ):1012        _die("--downscale-max-dim must be >= 1")1013 1014    _validate_model(args.model)1015    _validate_size(args.size, args.model)1016    _validate_quality(args.quality)1017    _validate_background(args.background)1018    _validate_model_specific_options(1019        model=args.model,1020        background=args.background,1021        input_fidelity=getattr(args, "input_fidelity", None),1022    )1023    _ensure_api_key(args.dry_run)1024 1025    args.func(args)1026    return 01027 1028 1029if __name__ == "__main__":1030    raise SystemExit(main())1031