#!/usr/bin/env python3 """Follow-up for pending 06 and failed 09 clean character jobs.""" import base64, json, mimetypes, os, pathlib, re, time, urllib.error, urllib.request from PIL import Image, ImageDraw BASE="https://mcp.mxai.cn" PROJECT=pathlib.Path('/Users/bot1/Volumes/root_for_ai/AI工作区/通用_产品宣传视频_古钱币杜邦纸钱袋包_20260530_1702') OUTDIR=PROJECT/'outputs/coin-character-clean-set/from_user_sheet_original_artifacts_20260605' MANIFEST=OUTDIR/'manifest.json' STYLE=PROJECT/'assets/character_style_refs/from_user_sheet_20260605/character_crops/09_textured_bar_character_crop.png' ART=PROJECT/'assets/reference-coins/raw/coin-04-flat-spade-green-patina.jpg' def read_key(): key=os.environ.get('MX_AI_API_KEY') if key: return key.strip() for p in [pathlib.Path.home()/'.hermes/secret-vault/shared-services.env', pathlib.Path.home()/'.hermes/profiles/video/.env']: if p.exists(): m=re.search(r"^\s*MX_AI_API_KEY\s*=\s*['\"]?([^'\"\n]+)",p.read_text(encoding='utf-8',errors='ignore'),re.M) if m: return m.group(1).strip() raise RuntimeError('MX_AI_API_KEY missing') def redact(s,key=''): s=str(s) if key: s=s.replace(key,'********') s=re.sub(r'nb_[A-Za-z0-9_\-]+','nb_********',s) return s def req(method,path,key,body=None,timeout=120): headers={'Content-Type':'application/json; charset=utf-8','Authorization':'Bearer '+key,'User-Agent':'HermesVideo/1.0'} data=None if body is None else json.dumps(body,ensure_ascii=False).encode('utf-8') r=urllib.request.Request(BASE+path,data=data,method=method,headers=headers) try: with urllib.request.urlopen(r,timeout=timeout) as resp: return json.loads(resp.read().decode('utf-8',errors='replace')) except urllib.error.HTTPError as e: raise RuntimeError(f"HTTP {e.code}: {redact(e.read().decode('utf-8',errors='replace'),key)}") def nested(x): return x.get('data') if isinstance(x,dict) and isinstance(x.get('data'),dict) else x def urls(obj): u=obj.get('image_urls') or obj.get('images') or obj.get('urls') or obj.get('image') or [] return [u] if isinstance(u,str) else u def data_url(p): return 'data:'+(mimetypes.guess_type(p.name)[0] or 'image/png')+';base64,'+base64.b64encode(p.read_bytes()).decode('ascii') def serial_from(resp): o=nested(resp) for k in ('serial_no','serialNo','serial'): if isinstance(o,dict) and o.get(k): return o[k] return None def download(url,dest): r=urllib.request.Request(url,headers={'User-Agent':'HermesVideo/1.0'}) with urllib.request.urlopen(r,timeout=180) as resp: dest.write_bytes(resp.read()) def make_contact(files,outpath): thumbs=[] for f in files: im=Image.open(f).convert('RGB'); im.thumbnail((260,300)) c=Image.new('RGB',(300,360),'white'); c.paste(im,((300-im.width)//2,15)) d=ImageDraw.Draw(c); d.text((10,330),pathlib.Path(f).stem[:34],fill=(0,0,0)); thumbs.append(c) sheet=Image.new('RGB',(900,((len(thumbs)+2)//3)*360),(245,242,235)) for i,t in enumerate(thumbs): sheet.paste(t,((i%3)*300,(i//3)*360)) sheet.save(outpath,quality=92) def main(): key=read_key(); data=json.loads(MANIFEST.read_text(encoding='utf-8')); results=data['results'] byid={r['id']:r for r in results} # Poll pending 06 a few times r6=byid.get('06_bronze_script_hollow') if r6 and r6.get('status')!='completed' and r6.get('serial_no'): for _ in range(24): obj=nested(req('GET',f"/mcp/api/task/{r6['serial_no']}",key,timeout=90)); st=str(obj.get('status')) print(json.dumps({'event':'poll06','status':st,'text':obj.get('status_text') or obj.get('message')},ensure_ascii=False),flush=True) if st=='2': files=[] for i,u in enumerate(urls(obj),1): dest=OUTDIR/f"06_bronze_script_hollow_clean_character_{i}.png"; download(u,dest); files.append(str(dest)) r6.update({'status':'completed','files':files}); break if st in {'3','4'}: r6.update({'status':'failed','fail_msg':redact(obj.get('fail_msg') or obj,key)}); break time.sleep(5) # Retry 09 with cleaned wording r9=byid.get('09_textured_bar') if r9 and r9.get('status')!='completed': prompt=""" Create a clean character design image for one anthropomorphic ancient coin artifact character. Input reference use: - Image 1 is only a style reference for short rounded arms and legs, premium toy-like 3D rendering, and friendly standing proportions. Ignore all card text, borders, icons, inset pictures, labels, and pedestal from image 1. - Image 2 is the strict artifact reference for the central torso shape. Preserve the flat spade-shaped green-patinated ancient money silhouette, long handle, broad flat base section, mottled mineral texture, copper-green oxidation, surface marks and proportions from image 2. Final image: One full-body character centered on a clean warm off-white background. The artifact itself is the torso. Add only simple short rounded arms and simple short rounded legs in a consistent premium toy-like 3D style. The character stands in a neutral slightly lively pose suitable for later video animation. No face, no eyes, no mouth, no costume, no accessories, no pedestal, no card frame, no text, no labels, no UI, no watermark. Keep the silhouette readable and the original artifact identity obvious. Style: premium cultural-creative product character design, realistic aged metal and mineral surface, soft studio lighting, subtle contact shadow, high detail, clean reference sheet. """.strip() ppath=OUTDIR/'prompts/09_textured_bar_prompt_retry_clean.md'; ppath.write_text(prompt+'\n',encoding='utf-8') body={'prompt':prompt,'model':'gpt-image-2','aspect_ratio':'3:4','resolution':'1K','quality':'high','count':1,'input_images':[data_url(STYLE),data_url(ART)]} resp=req('POST','/mcp/api/generate/image',key,body,timeout=180); serial=serial_from(resp) r9.update({'status':'submitted_retry','serial_no':serial,'prompt_path':str(ppath)}) print(json.dumps({'event':'submit09_retry','serial_no':serial},ensure_ascii=False),flush=True) if serial: for _ in range(60): obj=nested(req('GET',f'/mcp/api/task/{serial}',key,timeout=90)); st=str(obj.get('status')) print(json.dumps({'event':'poll09','status':st,'text':obj.get('status_text') or obj.get('message')},ensure_ascii=False),flush=True) if st=='2': files=[] for i,u in enumerate(urls(obj),1): dest=OUTDIR/f"09_textured_bar_clean_character_{i}.png"; download(u,dest); files.append(str(dest)) r9.update({'status':'completed','files':files}); break if st in {'3','4'}: r9.update({'status':'failed','fail_msg':redact(obj.get('fail_msg') or obj,key)}); break time.sleep(5) completed=[f for r in results if r.get('status')=='completed' for f in r.get('files',[])] contact=str(OUTDIR/'clean_character_contact_sheet.jpg') if completed else None if completed: make_contact(completed,OUTDIR/'clean_character_contact_sheet.jpg') data.update({'results':results,'contact_sheet':contact}) MANIFEST.write_text(json.dumps(data,ensure_ascii=False,indent=2),encoding='utf-8') print('FINAL_JSON_START') print(json.dumps({'completed':len(completed),'contact_sheet':contact,'manifest':str(MANIFEST),'results':results},ensure_ascii=False,indent=2)) print('FINAL_JSON_END') if __name__=='__main__': main()