#!/usr/bin/env python3 import base64 import hashlib import json import os 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 digest_bytes(data): return hashlib.sha256(data).hexdigest() def digest_file(path): hasher = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(8 * 1024 * 1024), b""): hasher.update(block) return hasher.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().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().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_images(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_images(value, found) elif isinstance(obj, list): for value in obj: collect_images(value, found) def parse_sse(data): found = [] events = 0 for raw in data.decode(errors="replace").splitlines(): if not raw.startswith("data:"): continue payload = raw[5:].strip() if not payload or payload == "[DONE]": continue events += 1 try: collect_images(json.loads(payload), found) except json.JSONDecodeError: pass if not found: try: collect_images(json.loads(data), found) except Exception: pass if not found: raise RuntimeError("Image response contained no usable image payload") 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] title = brief["canonical_title"] if asset == "left_panel": return ( "Standalone vertical cinematic CHANNEL BRANDING PANEL, not a story poster. No people, faces, silhouettes, story title or exclusivity sentence. " "Dark premium cold-storage and frosted-glass textures, charcoal clean-room surfaces, restrained amber warning lights, subtle cold haze and abstract pharmaceutical shelf highlights. " "Let the design and typography flow naturally with the artwork; do not use rigid boxes, coordinate instructions or fixed margins. " "Render exactly two clear Vietnamese text blocks and no other main headline: Huyền An Audio; Like • Chia sẻ • Đăng ký. " "Keep both blocks readable, balanced, not cropped and not overlapping. No medicine brand, watermark, QR, real logo or pseudo headline." ) exact = " | ".join(spec["required_text"]) return ( f"Create a premium cinematic {'vertical' if asset == 'right_panel' else 'horizontal'} key art poster for a completely fictional Chinese-style Vietnamese audio drama. " f"Composition: {spec['composition']} Setting: {brief['world']} Protagonist: {brief['protagonist']} Supporting cast: {'; '.join(brief['supporting_cast'])}. " f"Motifs: {', '.join(brief['motifs'])}. Palette: {brief['palette']}. Semi-realistic Asian fictional cast, layered cinematic depth, controlled low-key lighting, detailed faces, fabric, atmospheric haze and reflections. " f"Reveal limit: {brief['reveal_limit']} Negative constraints: {'; '.join(brief['negative_constraints'])}. " "Typography must be integrated naturally into the key art. Let the provider choose attractive positions and line breaks; do not force text into fixed coordinates, percentage bands, boxes or numerical safe margins. " "Keep the protagonist's eyes and face unobstructed, and keep all main text readable and uncropped. " f"Render directly and exactly these three Vietnamese text blocks with all accents preserved, without rewriting or adding a slogan: {exact}. " f"Canonical title authority: {title}. No watermark, QR, real brand or additional main headline." ) def verify_png(data): return len(data) > 100000 and data.startswith(b"\x89PNG\r\n\x1a\n") def main(): if not BRIEF.exists(): raise RuntimeError("Image brief missing; provider spend blocked") brief = json.loads(BRIEF.read_text()) if brief.get("verified") is not True or not brief.get("source_canon_sha256"): raise RuntimeError("Image brief is not verified") key = find_key() rows = [] mapping = {"intro_poster": "intro-poster-master.png", "right_panel": "right-panel-master.png", "left_panel": "left-panel-master.png"} existing = {row.get("asset"): row for row in json.loads(RECEIPT.read_text()).get("assets", [])} if RECEIPT.exists() else {} for asset, filename in mapping.items(): output = PROJECT / "image" / filename prompt = prompt_for(asset, brief) prompt_hash = digest_bytes(prompt.encode()) prior = existing.get(asset, {}) if output.exists() and prior.get("verified_file") is True and prior.get("source_canon_sha256") == brief["source_canon_sha256"] and prior.get("prompt_sha256") == prompt_hash and prior.get("sha256") == digest_file(output): rows.append(prior) 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(), 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_sse(data) if not verify_png(image): raise RuntimeError("Generated image failed PNG verification") temp = output.with_suffix(".png.part") temp.write_bytes(image) temp.replace(output) row = {"asset": asset, "source_canon_sha256": brief["source_canon_sha256"], "model": MODEL, "endpoint": ENDPOINT, "prompt_sha256": prompt_hash, "path": str(output.relative_to(PROJECT)), "bytes": output.stat().st_size, "events": events, "sha256": digest_file(output), "verified_file": True, "attempts": attempt} 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": 1, "verified": len(rows) == 3, "source_canon_sha256": brief["source_canon_sha256"], "assets": rows, "updated_at": datetime.now(timezone.utc).isoformat()}, ensure_ascii=False, indent=2) + "\n") print(json.dumps({"asset": asset, "verified": True, "path": str(output)}, ensure_ascii=False), flush=True) receipt = {"version": 1, "verified": len(rows) == 3 and all(row["verified_file"] for row in rows), "source_canon_sha256": brief["source_canon_sha256"], "assets": rows, "created_at": datetime.now(timezone.utc).isoformat()} RECEIPT.write_text(json.dumps(receipt, ensure_ascii=False, indent=2) + "\n") print(json.dumps({"verified": receipt["verified"], "assets": len(rows), "receipt": str(RECEIPT)}, ensure_ascii=False)) return 0 if receipt["verified"] else 1 if __name__ == "__main__": sys.exit(main())