#!/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 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] exact = " | ".join(spec["required_text"]) orientation = "horizontal 16:9" if asset == "intro_poster" else "vertical" text_count = "two" if asset == "left_panel" else "three" if asset == "intro_poster": cast_rule = "Use exactly one fictional adult Asian woman and one fictional adult Asian man. No second woman, crowd or romantic rival. " context = ( f"World: {brief['world']} Protagonist: {brief['protagonist']} Love interest: {brief['love_interest']}. " "Set the ENTIRE visible scene only in a narrow service corridor and an enclosed cinema projection control booth containing visible unlabelled projector machinery, reels and control equipment. " "ABSOLUTELY NO auditorium, theater hall, cinema seating, red seats, seat rows, audience area, stage, screen or auditorium doorway may appear ANYWHERE in the image, background, foreground, doorway, reflection or side area. " "The canonical title MUST end exactly with the word Năm and MUST NOT add a period, full stop, dot or any terminal punctuation after Năm. " ) elif asset == "right_panel": cast_rule = "Use exactly one fictional adult Asian woman and one fictional adult Asian man. No second woman, crowd or romantic rival. " context = f"World: {brief['world']} Protagonist: {brief['protagonist']} Love interest: {brief['love_interest']}. " else: cast_rule = "No people, faces, silhouettes, characters, props, setting or narrative scene. " context = ( "This asset is abstract channel branding only; do not inherit story world, cast, cinema, key, ticket, chair or projector motifs. " "The call-to-action MUST end exactly with the word ký and MUST NOT add a period, full stop, dot or any terminal punctuation after ký. " ) return ( f"Premium cinematic {orientation} artwork for a completely fictional Chinese-style Vietnamese romance audio drama. " f"Composition: {spec['composition']} {context}Palette: {brief['palette']}. Reveal limit: {brief['reveal_limit']}. " f"Negative constraints: {'; '.join(brief['negative_constraints'])}. " + cast_rule + "Keep the visible emotional focus on an adult woman setting a boundary and a man respecting her space; do not imply coercion, a triangle, investigation or mystery. " "Typography must be rendered directly by the image provider as part of this same artwork; no blank text area and no separate text layer. " "Let typography flow naturally, never cover eyes or faces, and keep every required block fully visible, readable and uncropped. " f"Render exactly these {text_count} Vietnamese text blocks with every accent preserved and without rewriting: {exact}. " "ABSOLUTE TEXT RULE: the required blocks are the only readable or letter-like characters anywhere. " "Every prop, surface and decorative element must contain no other letters, numbers, signatures, pseudo-text or glyphs. " "No watermark, QR code, real logo or extra caption." ) def main(): 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"]: 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"], "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": 1, "verified": False, "source_canon_sha256": brief["source_canon_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": 1, "verified": len(rows) == 3 and all(row["verified_file"] for row in rows), "source_canon_sha256": brief["source_canon_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)