#!/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] 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']}. " "The station exit must remain visibly open. The man stands to one side and never blocks, grabs or corners the woman. " "Use only an inactive fictional station platform with an old glass canopy; no moving train, danger, readable timetable, railway logo or real transit branding. " "The canonical title MUST end exactly with the words Một Chiều and MUST NOT add a period, full stop, dot or terminal punctuation after Chiều. " ) 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']}. " "Keep the exit open and the man clearly off the woman's path. No moving train, danger, readable signage or real transit branding. " ) 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, station, train, ticket, contract, clock or whistle motifs. " "The call-to-action MUST end exactly with the word ký and MUST NOT add a period, full stop, dot or 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 freely choosing to leave with a blank one-way ticket while a man respects her open path; 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(): 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)