#!/usr/bin/env python3 import base64 import hashlib import json import os import sys import time import urllib.error 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): h=hashlib.sha256() with path.open('rb') as f: for block in iter(lambda:f.read(8*1024*1024),b''):h.update(block) return h.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('"\'') hermes_config=Path.home()/'.hermes/config.yaml' if hermes_config.exists(): model={};in_model=False for raw in hermes_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 CHANNEL BRANDING PANEL, not a story poster. No people, no faces, no silhouettes, no bridge disaster scene and no readable warning sign. " "Rain-darkened structural steel textures, subtle amber work lights and one abstract hairline crack as restrained decoration, cinematic navy-charcoal and steel-blue palette. " "Minimal clean typography with exactly two separated text blocks and no other readable words anywhere. " "Top text block, render exactly with Vietnamese accents: Huyền An Audio. " "Lower text block, render exactly with Vietnamese accents and bullet separators: Like • Chia sẻ • Đăng ký. " f"Absolutely forbidden text: {title}; Truyện được phát độc quyền; CẦU LAM GIANG; SẬP CẦU. " "Do not depict Lâm Tĩnh Nghi or any character. Safe margin 10 percent on all sides. No watermark, no logo, no pseudo headline." ) if asset == 'right_panel': exact_brand = "Huyền An Audio" exact_exclusive = "Truyện được phát độc quyền tại Huyền An Audio, nghiêm cấm sao chép dưới mọi hình thức" return ( f"Standalone vertical story poster for a completely fictional Chinese-style audio drama. {spec['composition']} " f"Setting: {brief['world']} Protagonist: {brief.get('protagonist','')} Supporting cast: {'; '.join(brief.get('supporting_cast',[]))}. " f"Motifs: {', '.join(brief['motifs'])}. Palette: {brief['palette']}. Cinematic semi-realistic key art, no real logo, watermark or QR. " f"Reveal limit: {brief['reveal_limit']} Negative: {'; '.join(brief['negative_constraints'])}. " f"Render exactly three readable Vietnamese text blocks: 1) {exact_brand} 2) {title} 3) {exact_exclusive}. " "The brand block must end immediately after the word Audio: absolutely no pipe, slash, dash, colon, bullet or decorative character after it. " "Hard composition rule: reserve the top 12 percent and bottom 15 percent of the canvas as completely text-free breathing room. Place the brand below y=15 percent, the title inside the central band, and the complete exclusivity footer above y=82 percent. Keep left and right text edges at least 10 percent from the canvas edges. Do not place any border or decorative line close to the canvas edge. " "The footer must end at the word 'thức' with no terminal period or extra character. Do not add any other headline or pseudo-text. Preserve all Vietnamese accents exactly." ) common=(f"Ảnh {asset} cho truyện audio drama Trung Quốc hoàn toàn hư cấu. {spec['composition']} " f"Bối cảnh: {brief['world']} Nhân vật và continuity: {brief.get('protagonist','')} {'; '.join(brief.get('supporting_cast',[]))}. " f"Motif: {', '.join(brief['motifs'])}. Bảng màu: {brief['palette']}. Phong cách poster phim điện ảnh bán hiện thực, ánh sáng có chiều sâu, chất lượng cao, không người thật nổi tiếng, không watermark, không logo thật. " f"Reveal limit bắt buộc: {brief['reveal_limit']} Negative: {'; '.join(brief['negative_constraints'])}. ") required=' | '.join(spec['required_text']) return common+f"Provider phải render trực tiếp, nguyên văn, đúng toàn bộ dấu tiếng Việt và dễ đọc các khối chữ chính sau: {required}. Không thêm hoặc đổi chữ chính. Title authority: {title}." 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: req=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(req,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(r['verified_file'] for r 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())