#!/usr/bin/env python3 from __future__ import annotations import base64 import datetime as dt import json import mimetypes import os from pathlib import Path import re import subprocess import sys import time import urllib.error import urllib.request BASE = "https://mcp.mxai.cn" MODEL = "gpt-image-2" ASPECT_RATIO = "9:16" RESOLUTION = "1K" QUALITY = "high" COUNT = 1 PROJECT = Path("/Users/bot1/Volumes/root_for_ai/AI工作区/国博古钱币盲盒套装_商品主视频_20260625") SRC_BASE = Path("/Users/bot1/Volumes/root_for_ai/public-assets/product-archive-library/archive/2026/PA-2026-0006__国博古钱币盲盒套装/01_拍摄原片") OUT_FRAMES = PROJECT / "02_storyboard_frames_mxai" OUT_SHEET = PROJECT / "03_storyboard_sheet" OUT_PROMPTS = PROJECT / "04_prompts" REF_CACHE = PROJECT / "01_refs_mxai_cache" for d in [OUT_FRAMES, OUT_SHEET, OUT_PROMPTS, REF_CACHE]: d.mkdir(parents=True, exist_ok=True) STYLE_COMMON = ( "商品主视频分镜图，9:16竖版单镜头画面。基于参考图中的真实产品生成：保持毛绒挂件的正面造型、颜色、刺绣纹样、圆润轮廓、薄挂件比例和毛绒纤维质感。" "所有毛绒尽量正面对镜头，像薄薄的毛绒挂件/正面纸偶小人轻轻活动；动作只表现为正面轻弹、轻晃、点头、靠近、轻碰或轻柔漂浮。" "不要生成厚重3D公仔，不要侧面、背面、转身、奔跑、翻滚、真实开盒动作、复杂肢体拖拽。" "背景为暖白奶油色微型展台，淡金色古钱币光点，干净电商产品宣传片质感。" "保留产品自身已有纹样和刺绣细节；不要新增文字、额外logo、水印、标签、边框。" ) ref_hidden = SRC_BASE / "20260625-140632.jpg" ref_group = SRC_BASE / "20260625-140637.jpg" SHOTS = [ { "id": "shot01_front_closeup", "name": "镜头1_正面可爱特写", "ref": ref_group, "prompt": STYLE_COMMON + "镜头1：正面可爱特写。选择参考图中的2到3只普通款古钱币毛绒小挂件组成近景小合影，角色从画面下方轻轻弹入后停住，正面对镜头，轻微上下错位；画面像拍大头贴，主体占画面中部，留出柔和空间。", }, { "id": "shot02_three_friends", "name": "镜头2_三只小伙伴嬉戏合影", "ref": ref_group, "prompt": STYLE_COMMON + "镜头2：三只毛绒小伙伴嬉戏合影。参考图中的三只不同颜色普通款毛绒靠在一起，正面对镜头，一个稍微向前，两个在后方左右错开，表现轻轻点头、微微摆动、软软碰一下的瞬间；动作幅度很小，保持挂件正面比例。", }, { "id": "shot03_single_regular_closeup", "name": "镜头3_单个普通款萌态特写", "ref": ref_group, "prompt": STYLE_COMMON + "镜头3：单个普通款毛绒正面萌态特写。画面只突出一只颜色鲜明的普通款古钱币毛绒，正面对镜头，像被轻轻提起后小幅晃动，周围有另外两只毛绒在边缘虚化陪衬；保持真实毛绒挂件平面感和产品刺绣细节。", }, { "id": "shot04_gathering_to_center", "name": "镜头4_普通款向中心聚拢", "ref": ref_group, "prompt": STYLE_COMMON + "镜头4：六只普通款向中心聚拢。多个普通款毛绒分布在画面左右和下方，正面朝向观众，沿柔和弧线轻轻漂移到中间附近，中间保留空位；它们像小队集合，不排得太死，不做侧身和转身，淡金色光点开始向中央汇聚。", }, { "id": "shot05_hidden_conductor", "name": "镜头5_海晏河清尊隐藏款C位登场", "ref": ref_hidden, "prompt": STYLE_COMMON + "镜头5：海晏河清尊隐藏款C位登场。以参考图中的浅青灰色海晏河清尊原型毛绒挂件为唯一核心主体，正面位于画面中央，略高、略大，像小指挥家一样温柔抬起小手或轻轻点头；周围用淡金色光晕和少量柔和光点表现其他毛绒正在环绕，但不要把主体变成瓷器或厚重3D模型。", }, { "id": "shot06_final_group_c_position", "name": "镜头6_最终全员商品主视觉", "ref": ref_group, "prompt": STYLE_COMMON + "镜头6：最终全员商品主视觉。七只毛绒全部以正面合影出现，浅青灰色海晏河清尊隐藏款站在正中央C位，略高于其他角色；其他六只分布在左右和下方，像用一圈柔和金色光带托起C位主角，形成完整套装合影。画面干净、可爱、适合电商商品主视频结尾定格。", }, ] def redact(text: object, key: str = "") -> str: s = str(text) if key: s = s.replace(key, "********") s = re.sub(r"nb_[A-Za-z0-9_\-]+", "nb_********", s) s = re.sub(r"(auth[_-]?key=)[^&\s]+", r"\1********", s, flags=re.I) return s def request_json(method: str, path: str, key: str, body=None, timeout=180): 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") req = urllib.request.Request(BASE + path, data=data, method=method, headers=headers) try: with urllib.request.urlopen(req, timeout=timeout) as resp: raw = resp.read().decode("utf-8", errors="replace") return json.loads(raw) except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", errors="replace") raise RuntimeError(f"HTTP {e.code} {path}: {redact(raw, key)}") except Exception as e: raise RuntimeError(f"{type(e).__name__} {path}: {redact(e, key)}") def nested_dict(resp): if isinstance(resp, dict) and isinstance(resp.get("data"), dict): return resp["data"] return resp if isinstance(resp, dict) else {} def serial_from(resp): obj = nested_dict(resp) for k in ("serial_no", "serialNo", "serial"): if obj.get(k): return obj[k] for k in ("serial_no", "serialNo", "serial"): if isinstance(resp, dict) and resp.get(k): return resp[k] return None def make_reference(src: Path) -> Path: # 只复制/压缩到项目工作区，不修改产品中枢原片。 ref = REF_CACHE / f"{src.stem}__ref1800.jpg" if ref.exists() and ref.stat().st_size > 0: return ref subprocess.check_call([ "sips", "-Z", "1800", "-s", "format", "jpeg", "-s", "formatOptions", "88", str(src), "--out", str(ref) ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return ref def data_url(path: Path) -> str: mime = mimetypes.guess_type(path.name)[0] or "image/jpeg" data = base64.b64encode(path.read_bytes()).decode("ascii") return f"data:{mime};base64,{data}" def download(url: str, dest: Path): req = urllib.request.Request(url, headers={"User-Agent": "HermesVideo/1.0"}) with urllib.request.urlopen(req, timeout=240) as resp: dest.write_bytes(resp.read()) def poll_task(serial: str, key: str, timeout_s: int = 720): deadline = time.time() + timeout_s last = None while time.time() < deadline: obj = nested_dict(request_json("GET", f"/mcp/api/task/{serial}", key, timeout=90)) status = obj.get("status") last = obj print(json.dumps({"event": "poll", "serial_no": serial, "status": status, "message": obj.get("message") or obj.get("status_text")}, ensure_ascii=False), flush=True) if str(status) == "2": return obj if str(status) in {"3", "4"}: raise RuntimeError("MXAI task failed: " + redact(obj.get("fail_msg") or obj.get("message") or obj, key)) time.sleep(5) raise TimeoutError("MXAI task timeout: " + serial + " last=" + redact(last, key)) def get_dimensions(path: Path): out = subprocess.check_output(["sips", "-g", "pixelWidth", "-g", "pixelHeight", str(path)], text=True, stderr=subprocess.STDOUT) w = int(re.search(r"pixelWidth: (\d+)", out).group(1)) h = int(re.search(r"pixelHeight: (\d+)", out).group(1)) return w, h def main(): key = os.environ.get("MX_AI_API_KEY") if not key: raise SystemExit("MX_AI_API_KEY missing") run_id = dt.datetime.now().strftime("%Y%m%d_%H%M%S") manifest = { "run_id": run_id, "project": str(PROJECT), "model": MODEL, "aspect_ratio": ASPECT_RATIO, "resolution": RESOLUTION, "quality": QUALITY, "count_per_shot": COUNT, "shots": [], } for idx, shot in enumerate(SHOTS, 1): src = Path(shot["ref"]) ref = make_reference(src) rec = { "index": idx, "id": shot["id"], "name": shot["name"], "source_ref": str(src), "reference_cache_file": str(ref), "prompt": shot["prompt"], "model": MODEL, "aspect_ratio": ASPECT_RATIO, "resolution": RESOLUTION, "quality": QUALITY, "count": COUNT, } body = { "prompt": shot["prompt"], "model": MODEL, "aspect_ratio": ASPECT_RATIO, "resolution": RESOLUTION, "quality": QUALITY, "count": COUNT, "input_images": [data_url(ref)], } print(json.dumps({"event": "submit", "index": idx, "id": shot["id"], "ref": src.name}, ensure_ascii=False), flush=True) try: resp = request_json("POST", "/mcp/api/generate/image", key, body, timeout=180) serial = serial_from(resp) if not serial: raise RuntimeError("submit response missing serial_no: " + redact(resp, key)) rec["serial_no"] = serial print(json.dumps({"event": "submitted", "index": idx, "serial_no": serial}, ensure_ascii=False), flush=True) done = poll_task(serial, key) urls = done.get("image_urls") or done.get("images") or done.get("urls") or [] if isinstance(urls, str): urls = [urls] if not urls: raise RuntimeError("completed task has no image url") files = [] for j, url in enumerate(urls, 1): ext = ".png" m = re.search(r"\.(png|jpg|jpeg|webp)(?:\?|$)", url, re.I) if m: ext = "." + m.group(1).lower().replace("jpeg", "jpg") dest = OUT_FRAMES / f"{idx:02d}_{shot['id']}_{run_id}_{j}{ext}" download(url, dest) w, h = get_dimensions(dest) files.append({"local_file": str(dest), "width": w, "height": h, "size_bytes": dest.stat().st_size}) print(json.dumps({"event": "saved", "index": idx, "file": str(dest), "width": w, "height": h}, ensure_ascii=False), flush=True) rec["status"] = "completed" rec["files"] = files rec["completed_at"] = dt.datetime.now().isoformat(timespec="seconds") except Exception as e: rec["status"] = "failed" rec["error"] = redact(e, key) print(json.dumps({"event": "failed", "index": idx, "id": shot["id"], "error": rec["error"]}, ensure_ascii=False), flush=True) manifest["shots"].append(rec) (OUT_PROMPTS / "mxai_storyboard_generation_manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") # 顺序提交，避免一次并发打满 time.sleep(3) completed = sum(1 for s in manifest["shots"] if s.get("status") == "completed") failed = sum(1 for s in manifest["shots"] if s.get("status") == "failed") manifest["completed"] = completed manifest["failed"] = failed (OUT_PROMPTS / "mxai_storyboard_generation_manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") print("FINAL_JSON_START") print(json.dumps({"run_id": run_id, "completed": completed, "failed": failed, "manifest": str(OUT_PROMPTS / "mxai_storyboard_generation_manifest.json")}, ensure_ascii=False, indent=2)) print("FINAL_JSON_END") if failed: sys.exit(2) if __name__ == "__main__": main()