#!/usr/bin/env python3 import base64 import hashlib import json import os import subprocess import sys import time import urllib.request from datetime import datetime, timezone from pathlib import Path PROJECT = Path(__file__).resolve().parents[1] BRIEF = PROJECT / "image/image-brief.json" RECEIPT = PROJECT / "log/image-generation.json" ENDPOINT = "http://192.168.40.11:20128/v1/images/generations" MODEL = "cx/gpt-5.5-image" def sha_bytes(data): return hashlib.sha256(data).hexdigest() def sha_file(path): digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(8 * 1024 * 1024), b""): digest.update(block) return digest.hexdigest() def find_key(): if os.environ.get("IMAGE_API_KEY"): return os.environ["IMAGE_API_KEY"] candidates = [Path.home() / ".config/huyen-an-audio/image.env", Path.home() / ".config/image-api.env", Path.home() / ".env"] for path in candidates: if not path.exists(): continue for raw in path.read_text(encoding="utf-8", errors="ignore").splitlines(): if raw.strip().startswith("IMAGE_API_KEY="): return raw.split("=", 1)[1].strip().strip("\"'") config = Path.home() / ".hermes/config.yaml" if config.exists(): model = {} in_model = False for raw in config.read_text(encoding="utf-8", errors="ignore").splitlines(): if raw and not raw[0].isspace(): in_model = raw.strip() == "model:" continue if in_model and ":" in raw: key, value = raw.strip().split(":", 1) model[key] = value.strip().strip('"').strip("'") if model.get("base_url", "").rstrip("/") == "http://192.168.40.11:20128/v1" and model.get("api_key"): return model["api_key"] raise RuntimeError("IMAGE_API_KEY is unavailable") def collect(obj, found): if isinstance(obj, dict): for key, value in obj.items(): if key in {"b64_json", "image"} and isinstance(value, str) and len(value) > 1000: found.append(("base64", value)) elif key == "url" and isinstance(value, str) and value.startswith(("http://", "https://")): found.append(("url", value)) else: collect(value, found) elif isinstance(obj, list): for value in obj: collect(value, found) def parse_response(data): found = [] events = 0 for raw in data.decode("utf-8", errors="replace").splitlines(): if not raw.startswith("data:"): continue payload = raw[5:].strip() if not payload or payload == "[DONE]": continue events += 1 try: collect(json.loads(payload), found) except json.JSONDecodeError: pass if not found: try: collect(json.loads(data), found) except Exception: pass if not found: raise RuntimeError("Image provider response contains no image") kind, value = found[-1] if kind == "base64": return base64.b64decode(value), events with urllib.request.urlopen(value, timeout=180) as response: return response.read(), events def prompt_for(asset, brief): spec = brief["assets"][asset] prompt = spec.get("prompt") required = spec.get("required_text") if not isinstance(prompt, str) or not prompt.strip(): raise RuntimeError(f"image brief prompt missing for {asset}") if not isinstance(required, list) or not required or not all(isinstance(x, str) and x for x in required): raise RuntimeError(f"required text contract invalid for {asset}") exact = " | ".join(required) return ( prompt.strip() + " " "Typography must be rendered directly by the image provider in this same asset; no separate text layer or postprocessing. " f"Required text contract, byte-for-byte and in this order: {exact}. " "These required blocks are the only readable characters anywhere in the image. " "Keep every required block fully visible, readable and uncropped. No watermark, QR code, real logo or extra caption." ) def main(): gate = subprocess.run([sys.executable, str(PROJECT / "script/validate-workflow-copy.py"), "pre-provider"], cwd=PROJECT) if gate.returncode: raise RuntimeError("workflow-copy authority gate failed") if not BRIEF.is_file(): raise RuntimeError("verified image brief missing") brief = json.loads(BRIEF.read_text(encoding="utf-8")) if brief.get("verified") is not True or not brief.get("source_canon_sha256"): raise RuntimeError("image brief not verified") key = find_key() mapping = {"intro_poster": "intro-poster-master.png", "right_panel": "right-panel-master.png", "left_panel": "left-panel-master.png"} prior_rows = {} if RECEIPT.exists(): prior = json.loads(RECEIPT.read_text(encoding="utf-8")) if prior.get("source_canon_sha256") == brief["source_canon_sha256"] and prior.get("workflow_copy_authority_sha256") == brief.get("workflow_copy_authority_sha256"): prior_rows = {row.get("asset"): row for row in prior.get("assets", [])} rows = [] for asset, filename in mapping.items(): output = PROJECT / "image" / filename prompt = prompt_for(asset, brief) prompt_sha = sha_bytes(prompt.encode("utf-8")) old = prior_rows.get(asset, {}) if output.exists() and old.get("verified_file") is True and old.get("prompt_sha256") == prompt_sha and old.get("sha256") == sha_file(output): rows.append(old) continue payload = {"model": MODEL, "prompt": prompt, "n": 1, "size": "auto", "quality": "auto", "background": "auto", "image_detail": "high", "output_format": "png"} last = None for attempt in range(1, 4): try: request = urllib.request.Request(ENDPOINT, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), headers={"Content-Type": "application/json", "Accept": "text/event-stream", "Authorization": "Bearer " + key}, method="POST") with urllib.request.urlopen(request, timeout=900) as response: data = response.read() image, events = parse_response(data) if len(image) < 100000 or not image.startswith(b"\x89PNG\r\n\x1a\n"): raise RuntimeError("generated asset failed PNG verification") temp = output.with_suffix(".part.png") temp.write_bytes(image) os.replace(temp, output) row = {"asset": asset, "source_canon_sha256": brief["source_canon_sha256"], "workflow_copy_authority_sha256": brief["workflow_copy_authority_sha256"], "required_text": brief["assets"][asset]["required_text"], "model": MODEL, "endpoint": ENDPOINT, "prompt_sha256": prompt_sha, "path": str(output.relative_to(PROJECT)), "bytes": output.stat().st_size, "sha256": sha_file(output), "events": events, "attempts": attempt, "verified_file": True, "text_rendered_by_provider": True, "postprocessed_text": False} rows.append(row) break except Exception as exc: last = str(exc) if attempt < 3: time.sleep(2 ** attempt) else: raise RuntimeError(f"{asset} failed: {last}") RECEIPT.parent.mkdir(parents=True, exist_ok=True) RECEIPT.write_text(json.dumps({"version": 2, "verified": False, "source_canon_sha256": brief["source_canon_sha256"], "workflow_copy_authority_sha256": brief["workflow_copy_authority_sha256"], "assets": rows, "updated_at": datetime.now(timezone.utc).isoformat()}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps({"asset": asset, "verified_file": True, "path": str(output)}, ensure_ascii=False), flush=True) receipt = {"version": 2, "verified": len(rows) == 3 and all(row["verified_file"] for row in rows), "source_canon_sha256": brief["source_canon_sha256"], "workflow_copy_authority_sha256": brief["workflow_copy_authority_sha256"], "provider_rendered_text_only": True, "assets": rows, "created_at": datetime.now(timezone.utc).isoformat()} RECEIPT.write_text(json.dumps(receipt, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps({"verified": receipt["verified"], "assets": len(rows)}, ensure_ascii=False)) return 0 if receipt["verified"] else 1 if __name__ == "__main__": try: sys.exit(main()) except Exception as exc: print(f"Image generation blocked: {exc}", file=sys.stderr) sys.exit(1)