#!/usr/bin/env python3 """ 剪映草稿生成脚本 — 调用 ab-api POST /file/generateJianYing 通过 PrivToken（X-Priv-Token）认证。 将素材 URL（视频/图片/音频）+ 时长描述打包为剪映可导入的草稿 ZIP。 默认行为（非 --download）: 生成完成后直接输出下载 URL，供调用方直接展示，无需落盘。 用法: python gen_jianying_draft.py --title "我的视频" --scenes '[{"videoUrl":"...","duration":5}]' python gen_jianying_draft.py --title "测试" --scenes scenes.json --download -o draft.zip 环境变量: MM_API_BASE_URL - 后端 API 地址（默认: https://api-agent.remixmate.ai/api） PRIV_TOKEN - PrivToken（优先读取；未配置时提示手动输入） """ from __future__ import annotations import argparse import json import os import sys import time import urllib.error import urllib.parse import urllib.request import uuid import zipfile API_BASE_URL = os.environ.get("MM_API_BASE_URL", "https://api-agent.remixmate.ai/api") PRIVATE_TOKEN = "" # 在 main() 中通过 resolve_token() 初始化 SKILL_NAME = "export-jianying" AGENT_NAME = os.environ.get("AGENT_NAME", "") # 任务状态常量 STATUS_PENDING = "pending" STATUS_PROCESSING = "processing" STATUS_COMPLETED = "completed" STATUS_FAILED = "failed" # textLayer 角色预设（fontSize/color/bold/position/shadow） ROLE_PRESETS = { "headline": {"fontSize": 15, "color": "#FFFFFF", "bold": True, "position": {"x": 0, "y": -0.35}, "shadow": True}, "subheadline": {"fontSize": 8, "color": "#F0C040", "bold": False, "position": {"x": 0, "y": 0.38}, "shadow": False}, "badge": {"fontSize": 6, "color": "#FFFFFF", "bold": True, "position": {"x": -0.3, "y": -0.40}, "backgroundColor": "#D4A017"}, "cta": {"fontSize": 10, "color": "#F0C040", "bold": True, "position": {"x": 0, "y": 0.42}, "backgroundColor": "#2563EB"}, "caption": {"fontSize": 6, "color": "#CCCCCC", "bold": False, "position": {"x": 0, "y": 0.44}}, } # Pill 字幕样式（匹配 Remotion SubtitleBar "pill" 风格） PILL_SUBTITLE_STYLE = { "font_size": 8.0, "background_style": 1, "background_color": "#000000", "background_alpha": 0.65, "background_round_radius": 0.15, "has_shadow": False, "shadow_alpha": 0, "shadow_color": "", "shadow_distance": 0, "shadow_point": {"x": 0, "y": 0}, "shadow_smoothing": 0, "letter_spacing": 0.0, "line_max_width": 0.85, "line_spacing": 0.05, "force_apply_line_max_width": True, } PILL_SUBTITLE_POSITION_Y = -0.458 # --------------------------------------------------------------------------- # 剪映草稿根目录解析 # --------------------------------------------------------------------------- DRAFT_ROOT_PRESETS = { "mac": "~/Movies/JianyingPro/User Data/Projects/com.lveditor.draft", # Windows 预设保留字面量 %LOCALAPPDATA%：在 Mac 上执行时不展开， # 原样传给 API；用户在 Windows 解压时目录会自然匹配。 "windows": "%LOCALAPPDATA%/JianyingPro/User Data/Projects/com.lveditor.draft", } def resolve_draft_root_path(explicit: str, system: str) -> str: """ 解析剪映草稿根目录，按优先级： 1. 显式 --draft-root-path（做 expanduser） 2. --system 预设（mac → expanduser，windows → 字面量） 返回值始终作为 draftRootPath 传给 API。 """ if explicit: return os.path.expanduser(explicit) preset = DRAFT_ROOT_PRESETS.get(system) if not preset: print(f"❌ unsupported --system value: {system} (supported: mac / windows)", file=sys.stderr) sys.exit(1) if system == "mac": return os.path.expanduser(preset) return preset def resolve_token() -> str: """优先从 PRIV_TOKEN 环境变量读取 token，取不到则报错退出""" token = os.environ.get("PRIV_TOKEN", "") if not token: if sys.stdin.isatty(): token = input("请输入 PrivToken: ").strip() else: print("❌ PRIV_TOKEN env var not configured; cannot authenticate", file=sys.stderr) sys.exit(1) return token def gen_jianying_uuid() -> str: """生成剪映风格的大写 UUID""" return str(uuid.uuid4()).upper() def hex_to_rgb_float(hex_color: str) -> list: """'#RRGGBB' → [r, g, b]（0.0~1.0）""" hex_color = hex_color.lstrip("#") r = int(hex_color[0:2], 16) / 255.0 g = int(hex_color[2:4], 16) / 255.0 b = int(hex_color[4:6], 16) / 255.0 return [round(r, 6), round(g, 6), round(b, 6)] def build_text_content_json(text: str, style: dict) -> str: """构建剪映 text material 的 content JSON 字符串（双重编码格式）""" rgb = hex_to_rgb_float(style.get("color", "#FFFFFF")) font_size = style.get("fontSize", 8) char_count = len(text) content_obj = { "styles": [ { "fill": { "alpha": 1, "content": { "render_type": "solid", "solid": {"alpha": 1, "color": rgb}, }, }, "font": {"id": "", "path": ""}, "range": [0, char_count], "size": font_size, } ], "text": text, } return json.dumps(content_obj, ensure_ascii=False) def _build_headers(content_type: str = "application/json") -> dict: """构建统一请求头（X-Priv-Token 认证头）""" headers = { "X-Priv-Token": PRIVATE_TOKEN, "x-invoke-skill": SKILL_NAME, } if content_type: headers["Content-Type"] = content_type if AGENT_NAME: headers["x-invoke-agent"] = AGENT_NAME return headers def api_request(path: str, payload: dict, *, exit_on_error: bool = True) -> dict: """ 通用 API 请求函数（统一 POST 方法） exit_on_error=False 时网络/连接错误会抛出异常而非直接退出（供轮询重试使用） """ if not PRIVATE_TOKEN: print("❌ PrivToken not configured: set the PRIV_TOKEN env var", file=sys.stderr) sys.exit(1) url = f"{API_BASE_URL}{path}" data = json.dumps(payload).encode("utf-8") headers = _build_headers() req = urllib.request.Request(url, data=data, headers=headers, method="POST") try: with urllib.request.urlopen(req, timeout=120) as resp: return json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: body = e.read().decode("utf-8") print(f"❌ API request failed (HTTP {e.code}): {body}", file=sys.stderr) sys.exit(1) except (urllib.error.URLError, OSError, ConnectionError) as e: if exit_on_error: print(f"❌ network error: {e}", file=sys.stderr) sys.exit(1) raise def build_text_material(material_id: str, text: str, style: dict) -> dict: """ 构建完整的剪映 text material dict（对应 materials.texts[] 中一项）。 style 包含 fontSize / color / bold / shadow / backgroundColor 等。 """ content_json = build_text_content_json(text, style) font_size = style.get("fontSize", 8) color_hex = style.get("color", "#FFFFFF") bold = style.get("bold", False) has_shadow = style.get("shadow", False) bg_color = style.get("backgroundColor") or "" bg_style = 1 if bg_color else 0 return { "add_type": 0, "alignment": 1, "background_alpha": 1, "background_color": bg_color, "background_height": 0.14, "background_horizontal_offset": 0, "background_round_radius": 0, "background_style": bg_style, "background_vertical_offset": 0, "background_width": 0.14, "base_content": "", "bold_width": 0.04 if bold else 0, "border_alpha": 1, "border_color": "", "border_width": 0.08, "caption_template_info": { "category_id": "", "category_name": "", "effect_id": "", "is_new": False, "path": "", "request_id": "", "resource_id": "", "resource_name": "", "source_platform": 0, }, "check_flag": 7, "combo_info": {"text_templates": []}, "content": content_json, "fixed_height": -1, "fixed_width": -1, "font_category_id": "", "font_category_name": "", "font_id": "", "font_name": "", "font_path": "", "font_resource_id": "", "font_size": font_size, "font_source_platform": 0, "font_team_id": "", "font_title": "none", "font_url": "", "fonts": [], "force_apply_line_max_width": False, "global_alpha": 1, "group_id": "", "has_shadow": has_shadow, "id": material_id, "initial_scale": 1, "inner_padding": -1, "is_rich_text": False, "italic_degree": 0, "ktv_color": "", "language": "", "layer_weight": 1, "letter_spacing": 0, "line_feed": 1, "line_max_width": 0.82, "line_spacing": 0.02, "multi_language_current": "none", "name": "", "original_size": [], "preset_category": "", "preset_category_id": "", "preset_has_set_alignment": False, "preset_id": "", "preset_index": 0, "preset_name": "", "recognize_task_id": "", "recognize_type": 0, "relevance_segment": [], "shadow_alpha": 0.9, "shadow_angle": -45, "shadow_color": "", "shadow_distance": 8, "shadow_point": {"x": 0.6363961030678928, "y": -0.6363961030678928}, "shadow_smoothing": 1, "shape_clip_x": False, "shape_clip_y": False, "style_name": "", "sub_type": 0, "subtitle_keywords": None, "subtitle_template_original_fontsize": 0, "text_alpha": 1, "text_color": color_hex, "text_curve": None, "text_preset_resource_id": "", "text_size": 30, "text_to_audio_ids": [], "tts_auto_update": False, "type": "text", "typesetting": 0, "underline": False, "underline_offset": 0.22, "underline_width": 0.05, "use_effect_default_color": True, "words": {"end_time": [], "start_time": [], "text": []}, } def build_text_segment( segment_id: str, material_id: str, start_us: int, duration_us: int, position: dict, ) -> dict: """ 构建完整的剪映 text track segment dict。 position: {"x": float, "y": float} 归一化坐标。 """ return { "caption_info": None, "check_flag": 7, "clip": { "alpha": 1, "flip": {"horizontal": False, "vertical": False}, "rotation": 0, "scale": {"x": 1, "y": 1}, "transform": { "x": position.get("x", 0), "y": position.get("y", 0), }, }, "common_keyframes": [], "enable_adjust": False, "enable_color_correct_adjust": False, "enable_color_curves": True, "enable_color_match_adjust": False, "enable_color_wheels": True, "enable_lut": False, "enable_smart_color_adjust": False, "extra_material_refs": [], "group_id": "", "hdr_settings": None, "id": segment_id, "intensifies_audio": False, "is_placeholder": False, "is_tone_modify": False, "keyframe_refs": [], "last_nonzero_volume": 1, "material_id": material_id, "render_index": 11000, "responsive_layout": { "enable": False, "horizontal_pos_layout": 0, "size_layout": 0, "target_follow": "", "vertical_pos_layout": 0, }, "reverse": False, "source_timerange": {"duration": duration_us, "start": 0}, "speed": 1, "target_timerange": {"duration": duration_us, "start": start_us}, "template_id": "", "template_scene": "default", "track_attribute": 0, "track_render_index": 0, "uniform_scale": {"on": True, "value": 1}, "visible": True, "volume": 1, } def convert_render_plan_to_scenes(render_plan: dict) -> tuple: """ 将 RenderPlan JSON 转换为剪映草稿的 scenes 列表。 返回 (scenes, width, height)。 从 render-plan.json 中提取每个 scene 的素材 URL、时长、字幕文本， 自动映射为 gen_jianying_draft 所需的 scenes JSON 格式。 """ timeline = render_plan.get("timeline", []) resolved_assets = {a["assetId"]: a for a in render_plan.get("resolvedAssets", [])} render_config = render_plan.get("renderConfig", {}) fps = render_config.get("fps", 30) width = render_config.get("width", 1080) height = render_config.get("height", 1920) scenes = [] for entry in timeline: scene: dict = {} duration_frames = entry.get("durationFrames", 0) duration_sec = round(duration_frames / fps, 2) if fps else 5.0 scene["duration"] = duration_sec # 提取视觉素材（图片或视频） for layer in entry.get("layers", []): if layer.get("type") == "visual": asset = resolved_assets.get(layer.get("assetId", "")) if asset and asset.get("url"): asset_type = asset.get("type", "image") if asset_type == "video": scene["videoUrl"] = asset["url"] else: scene["imageUrl"] = asset["url"] break # 提取音频素材（TTS 旁白） for layer in entry.get("layers", []): if layer.get("type") == "audio": asset = resolved_assets.get(layer.get("assetId", "")) if asset and asset.get("url"): scene["audioUrl"] = asset["url"] if asset.get("duration"): scene["audioDuration"] = round(asset["duration"] / 1000.0, 2) break # 提取字幕文本（从 subtitleSegments 拼接） subtitle_segments = entry.get("subtitleSegments", []) if subtitle_segments: subtitle_text = " ".join(seg.get("text", "") for seg in subtitle_segments if seg.get("text")) if subtitle_text.strip(): scene["subtitleText"] = subtitle_text.strip() # 提取 textLayers（从 props 中的 textLayers 或 scene 级别的文本信息） props = entry.get("props", {}) title_text = props.get("titleText", "") if title_text: text_layers = [] text_layers.append({ "content": title_text, "role": "headline", }) # 如果有 subtitleText 也加入 if scene.get("subtitleText"): text_layers.append({ "content": scene["subtitleText"], "role": "subheadline", }) del scene["subtitleText"] # 避免重复 scene["textLayers"] = text_layers # 如果没有视觉素材，跳过该场景（剪映需要每个场景有 videoUrl 或 imageUrl） if not scene.get("videoUrl") and not scene.get("imageUrl"): # 尝试用占位图 print(f" ⚠️ scene {entry.get('sceneId', '?')} has no visual asset, skipping", file=sys.stderr) continue scenes.append(scene) if not scenes: print("❌ no convertible scenes in the RenderPlan (all scenes lack a visual asset)", file=sys.stderr) sys.exit(1) return scenes, width, height def parse_scenes(scenes_arg: str) -> list: """解析 --scenes 参数：支持 inline JSON 字符串或文件路径""" # 尝试作为文件路径读取 if os.path.isfile(scenes_arg): with open(scenes_arg, "r", encoding="utf-8") as f: content = f.read() try: scenes = json.loads(content) except json.JSONDecodeError as e: print(f"❌ failed to parse scenes JSON file: {e}", file=sys.stderr) sys.exit(1) else: # 尝试作为 inline JSON 解析 try: scenes = json.loads(scenes_arg) except json.JSONDecodeError as e: print(f"❌ failed to parse scenes JSON: {e}", file=sys.stderr) print(" hint: --scenes accepts an inline JSON string or a JSON file path", file=sys.stderr) sys.exit(1) if not isinstance(scenes, list): print("❌ scenes JSON must be an array", file=sys.stderr) sys.exit(1) if len(scenes) == 0: print("❌ scenes list cannot be empty", file=sys.stderr) sys.exit(1) # 基本校验 for i, scene in enumerate(scenes): if not isinstance(scene, dict): print(f"❌ scene {i + 1} is malformed: must be an object", file=sys.stderr) sys.exit(1) if not scene.get("videoUrl") and not scene.get("imageUrl"): print(f"❌ scene {i + 1} is missing videoUrl or imageUrl", file=sys.stderr) sys.exit(1) if not scene.get("duration") or scene["duration"] <= 0: print(f"❌ scene {i + 1} is missing a valid duration (must be > 0)", file=sys.stderr) sys.exit(1) # textLayers 校验 text_layers = scene.get("textLayers") if text_layers is not None: if not isinstance(text_layers, list): print(f"❌ scene {i + 1} textLayers must be an array", file=sys.stderr) sys.exit(1) for j, layer in enumerate(text_layers): if not isinstance(layer, dict): print(f"❌ scene {i + 1} textLayers[{j}] must be an object", file=sys.stderr) sys.exit(1) if not layer.get("content"): print(f"❌ scene {i + 1} textLayers[{j}] missing 'content'", file=sys.stderr) sys.exit(1) return scenes def create_jianying_task( title: str, scenes: list, draft_root_path: str, width: int = 1080, height: int = 1920, draft_name: str = "", ) -> str: """创建剪映草稿生成任务，返回 taskId""" print("📦 Submitting Jianying draft generation task...") print(f" title: {title}") print(f" scenes: {len(scenes)}") print(f" canvas: {width}x{height}") print(f" draft root: {draft_root_path}") if draft_name: print(f" draft name: {draft_name}") payload: dict = { "title": title, "scenes": scenes, "width": width, "height": height, "draftRootPath": draft_root_path, } if draft_name: payload["draftName"] = draft_name result = api_request("/file/generateJianYing", payload) # 检查业务状态码 if result.get("code") != 0: msg = result.get("msg", "未知错误") print(f"❌ API returned an error: {msg}", file=sys.stderr) sys.exit(1) result_data = result.get("data") or {} task_id = result_data.get("taskId") if not task_id: print(f"❌ could not parse task id; API returned: {json.dumps(result, ensure_ascii=False)}", file=sys.stderr) sys.exit(1) print(f"✅ task submitted, task id: {task_id}") return task_id def poll_task_status(task_id: str, poll_interval: int = 5, max_wait: int = 300) -> dict: """ 轮询剪映草稿生成状态，返回最终结果。 网络瞬断时自动重试（最多连续 5 次），不会因单次网络抖动而中断。 """ print(f"\n⏳ Waiting for draft generation (up to {max_wait} s)...") start_time = time.time() consecutive_errors = 0 max_consecutive_errors = 5 while True: elapsed = time.time() - start_time if elapsed > max_wait: print(f"\n❌ wait timed out (waited {elapsed:.0f} s)", file=sys.stderr) print(f" task id: {task_id}", file=sys.stderr) sys.exit(1) try: result = api_request("/file/generateJianYingQuery", {"taskId": task_id}, exit_on_error=False) except (urllib.error.URLError, OSError, ConnectionError) as e: consecutive_errors += 1 print(f"\n ⚠️ polling network error ({consecutive_errors}/{max_consecutive_errors}): {e}", flush=True) if consecutive_errors >= max_consecutive_errors: print(f"❌ {max_consecutive_errors} consecutive network errors, giving up on polling", file=sys.stderr) print(f" task id: {task_id} (the job may still be running on the backend)", file=sys.stderr) sys.exit(1) time.sleep(poll_interval) continue # 请求成功，重置连续错误计数 consecutive_errors = 0 # 检查业务状态码 if result.get("code") != 0: msg = result.get("msg", "未知错误") print(f"\n❌ status query failed: {msg}", file=sys.stderr) sys.exit(1) result_data = result.get("data") or {} status = result_data.get("status", "unknown") # 打印状态 status_icons = { STATUS_PENDING: "⏳", STATUS_PROCESSING: "🔄", STATUS_COMPLETED: "✅", STATUS_FAILED: "❌", } icon = status_icons.get(status, "❓") print(f"\r {icon} status: {status} | elapsed: {elapsed:.0f}s", end="", flush=True) if status == STATUS_COMPLETED: print() # 换行 return result_data if status == STATUS_FAILED: print() error_msg = result_data.get("errorMsg") or "未知错误" print(f"❌ draft generation failed: {error_msg}", file=sys.stderr) sys.exit(1) time.sleep(poll_interval) def _encode_url(raw_url: str) -> str: """对 URL 中的非 ASCII 字符做 percent-encoding（保留已编码部分和合法字符）""" parsed = urllib.parse.urlparse(raw_url) encoded_path = urllib.parse.quote(parsed.path, safe="/:@!$&'()*+,;=-._~") return urllib.parse.urlunparse(parsed._replace(path=encoded_path)) def download_zip(download_url: str, output_path: str) -> None: """下载 ZIP 文件到本地""" print(f"\n⬇️ Downloading draft ZIP: {download_url[:80]}...") safe_url = _encode_url(download_url) try: urllib.request.urlretrieve(safe_url, output_path) except OSError as e: print(f"❌ download failed: {e}", file=sys.stderr) sys.exit(1) if os.path.exists(output_path): file_size = os.path.getsize(output_path) print(f"✅ draft ZIP saved: {output_path} ({file_size / 1024 / 1024:.1f} MB)") else: print(f"❌ file not found: {output_path}", file=sys.stderr) sys.exit(1) def _resolve_layer_style(layer: dict) -> dict: """将 textLayer 字段与角色预设合并，返回最终有效样式""" role = layer.get("role", "subheadline") preset = ROLE_PRESETS.get(role, ROLE_PRESETS["subheadline"]) style = dict(preset) # 浅拷贝预设 # 用户显式字段覆盖预设 for key in ("fontSize", "color", "bold", "shadow", "backgroundColor"): if key in layer: style[key] = layer[key] if "position" in layer: style["position"] = layer["position"] return style def apply_text_layers(draft: dict, scenes: list) -> None: """ 核心后处理：遍历场景 textLayers，修改/新增 draft 中的文本素材与轨道片段。 直接原地修改 draft dict。 """ # 1. 找到 video track，提取每个片段的时间范围 video_track = None for track in draft.get("tracks", []): if track.get("type") == "video": video_track = track break if not video_track: print(" ⚠️ video track not found; skipping textLayers post-processing", file=sys.stderr) return video_segments = video_track.get("segments", []) if len(video_segments) != len(scenes): print( f" ⚠️ video segments ({len(video_segments)}) 与 scenes ({len(scenes)}) 数量不一致，" "按最小数量匹配", file=sys.stderr, ) # 2. 找到 text track，不存在则创建 text_track = None for track in draft.get("tracks", []): if track.get("type") == "text": text_track = track break if not text_track: text_track = { "attribute": 0, "flag": 0, "id": gen_jianying_uuid(), "is_default_name": True, "name": "", "segments": [], "type": "text", } draft["tracks"].append(text_track) text_segments = text_track.get("segments", []) texts_materials = draft.get("materials", {}).get("texts", []) # 建立 target_timerange.start → text_segment 索引映射 seg_by_start: dict[int, int] = {} for idx, seg in enumerate(text_segments): start = seg.get("target_timerange", {}).get("start", -1) seg_by_start[start] = idx # 建立 material_id → texts 索引映射 mat_by_id: dict[str, int] = {} for idx, mat in enumerate(texts_materials): mat_by_id[mat["id"]] = idx # 额外的 text track 列表（用于放置第 2+ 个 textLayer 的片段） extra_text_tracks: list[dict] = [] # 3. 遍历每个场景 match_count = min(len(video_segments), len(scenes)) for i in range(match_count): scene = scenes[i] text_layers = scene.get("textLayers") if not text_layers: continue v_seg = video_segments[i] start_us = v_seg["target_timerange"]["start"] duration_us = v_seg["target_timerange"]["duration"] for layer_idx, layer in enumerate(text_layers): content = layer.get("content", "") if not content: continue style = _resolve_layer_style(layer) position = style.get("position", {"x": 0, "y": 0}) if layer_idx == 0 and start_us in seg_by_start: # 替换已有文本素材 seg_idx = seg_by_start[start_us] existing_seg = text_segments[seg_idx] mat_id = existing_seg.get("material_id", "") # 更新素材 if mat_id in mat_by_id: mat = texts_materials[mat_by_id[mat_id]] mat["content"] = build_text_content_json(content, style) mat["font_size"] = style.get("fontSize", 8) mat["text_color"] = style.get("color", "#FFFFFF") mat["has_shadow"] = style.get("shadow", False) mat["bold_width"] = 0.04 if style.get("bold", False) else 0 bg_color = style.get("backgroundColor") or "" mat["background_color"] = bg_color mat["background_style"] = 1 if bg_color else 0 # 更新片段位置 existing_seg["clip"]["transform"]["x"] = position.get("x", 0) existing_seg["clip"]["transform"]["y"] = position.get("y", 0) else: # 新增素材 + 片段 new_mat_id = gen_jianying_uuid() new_seg_id = gen_jianying_uuid() new_material = build_text_material(new_mat_id, content, style) texts_materials.append(new_material) mat_by_id[new_mat_id] = len(texts_materials) - 1 new_segment = build_text_segment( new_seg_id, new_mat_id, start_us, duration_us, position ) # 第 2+ 个 textLayer 放入额外 track（每个 layer_idx 一条 track） extra_idx = layer_idx - 1 if (layer_idx > 0 and start_us in seg_by_start) else layer_idx while extra_idx >= len(extra_text_tracks): new_track = { "attribute": 0, "flag": 0, "id": gen_jianying_uuid(), "is_default_name": True, "name": "", "segments": [], "type": "text", } extra_text_tracks.append(new_track) extra_text_tracks[extra_idx]["segments"].append(new_segment) # 将额外 text tracks 追加到 draft for et in extra_text_tracks: if et["segments"]: draft["tracks"].append(et) # 更新 materials.texts 引用（因为可能通过 append 改变了列表内容） draft["materials"]["texts"] = texts_materials print(f" ✅ textLayers post-processing finished") def apply_subtitle_pill_style(draft: dict) -> None: """ 将 Pill 字幕样式（匹配 Remotion SubtitleBar "pill" 风格）应用到 draft 中的 所有 text materials 和 text segments。直接原地修改 draft dict。 """ texts_materials = draft.get("materials", {}).get("texts", []) or [] # 1. 覆盖每个 text material 的样式属性 for mat in texts_materials: for key, value in PILL_SUBTITLE_STYLE.items(): mat[key] = value # 更新 material 内嵌 content JSON 的 styles[0].size content_raw = mat.get("content", "") if content_raw: try: content_obj = json.loads(content_raw) styles = content_obj.get("styles", []) if styles: styles[0]["size"] = PILL_SUBTITLE_STYLE["font_size"] mat["content"] = json.dumps(content_obj, ensure_ascii=False) except (json.JSONDecodeError, TypeError): # 若 content 非 JSON（极少数情况），跳过内嵌更新 pass # 2. 将每个 text segment 的 clip.transform.y 设为 PILL_SUBTITLE_POSITION_Y for track in draft.get("tracks", []): if track.get("type") != "text": continue for seg in track.get("segments", []): clip = seg.setdefault("clip", {}) transform = clip.setdefault("transform", {"x": 0, "y": 0}) transform["y"] = PILL_SUBTITLE_POSITION_Y print(f" ✅ Pill subtitle style post-processing finished") def postprocess_draft_zip(zip_path: str, scenes: list) -> None: """ 后处理剪映草稿 ZIP：读取 draft_content.json，修改文本层，重新打包。 同时处理 draft_content.json 和 draft_info.json（两者内容相同）。 """ print(f"\n🔧 Post-processing draft ZIP (applying textLayers)...") tmp_path = zip_path + ".tmp" with zipfile.ZipFile(zip_path, "r") as zin: # 读取 draft_content.json draft_json_name = None draft_info_name = None for name in zin.namelist(): if name.endswith("draft_content.json"): draft_json_name = name elif name.endswith("draft_info.json"): draft_info_name = name if not draft_json_name: print(" ⚠️ draft_content.json not found in the ZIP; skipping post-processing", file=sys.stderr) return draft_data = json.loads(zin.read(draft_json_name).decode("utf-8")) # 应用 textLayers 修改 apply_text_layers(draft_data, scenes) # 应用 Pill 字幕样式（匹配 Remotion SubtitleBar） apply_subtitle_pill_style(draft_data) modified_json = json.dumps(draft_data, ensure_ascii=False, indent=4).encode("utf-8") # 写入新 ZIP with zipfile.ZipFile(tmp_path, "w", zipfile.ZIP_DEFLATED) as zout: for item in zin.infolist(): if item.filename == draft_json_name: zout.writestr(item, modified_json) elif item.filename == draft_info_name: # draft_info.json 内容同 draft_content.json zout.writestr(item, modified_json) else: zout.writestr(item, zin.read(item.filename)) # 原子替换 os.replace(tmp_path, zip_path) print(f" ✅ draft ZIP post-processing finished: {zip_path}") def prepare_api_scenes(scenes: list) -> list: """ 为 API 调用准备场景列表：剥离 textLayers，自动派生 subtitleText。 返回新列表（不修改原始 scenes）。 """ api_scenes = [] for scene in scenes: api_scene = {k: v for k, v in scene.items() if k != "textLayers"} text_layers = scene.get("textLayers") if text_layers and not api_scene.get("subtitleText"): # 从第一个 textLayer 派生 subtitleText api_scene["subtitleText"] = text_layers[0].get("content", "") api_scenes.append(api_scene) return api_scenes def has_text_layers(scenes: list) -> bool: """检查场景列表中是否有任何场景包含 textLayers""" return any(scene.get("textLayers") for scene in scenes) def main() -> None: parser = argparse.ArgumentParser( description="Jianying draft generator — packages asset URLs into a Jianying-importable draft ZIP", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Scene JSON example: [ {"videoUrl": "https://example.com/video.mp4", "duration": 5}, {"imageUrl": "https://example.com/img.jpg", "duration": 3, "subtitleText": "Product spotlight"}, {"videoUrl": "https://example.com/v2.mp4", "duration": 8, "audioUrl": "https://example.com/bgm.mp3", "audioDuration": 8} ] Scene fields: videoUrl Video material URL (mutually exclusive with imageUrl) imageUrl Image material URL (mutually exclusive with videoUrl) duration Scene duration in seconds, required width/height Material width/height (optional) audioUrl Audio URL (optional) audioDuration Audio duration (optional, defaults to duration) subtitleText Subtitle text (optional) textLayers Multi-layer text style array (optional, supports custom font size / color / position) """, ) parser.add_argument("--scenes", required=False, help="Scenes JSON (inline string or file path)") parser.add_argument("--title", required=False, help="Draft title") parser.add_argument( "--from-render-plan", default="", help="Auto-convert a RenderPlan JSON file into a Jianying draft (replaces --scenes; pulls material URL / duration / subtitle automatically)", ) parser.add_argument( "--from-job-id", type=int, default=0, help="Load the RenderPlan from the database (pass a jobId, replaces --from-render-plan; requires PRIV_TOKEN)", ) parser.add_argument("--width", type=int, default=1080, help="Canvas width (default 1080)") parser.add_argument("--height", type=int, default=1920, help="Canvas height (default 1920)") parser.add_argument("--draft-name", default="", help="Jianying draft name (defaults to title)") parser.add_argument( "--system", choices=["mac", "windows"], default="mac", help="Pick the draft-root preset by OS (default: mac)", ) parser.add_argument( "--draft-root-path", default="", help="Explicitly specify the Jianying draft root path (overrides the --system preset)", ) parser.add_argument( "--no-download", action="store_true", help="Skip the ZIP download; only print the download URL (ignored when scenes include textLayers / subtitleText)", ) parser.add_argument("-o", "--output", default=None, help="Download path (default jianying_draft_{timestamp}.zip, prevents multi-user collisions)") parser.add_argument("--poll-interval", type=int, default=5, help="Polling interval (seconds, default 5)") parser.add_argument("--max-wait", type=int, default=300, help="Maximum wait time (seconds, default 300)") parser.add_argument("--priv-token", default="", help="Override the auth token") args = parser.parse_args() # 初始化 token global PRIVATE_TOKEN if args.priv_token: PRIVATE_TOKEN = args.priv_token else: PRIVATE_TOKEN = resolve_token() if not PRIVATE_TOKEN: print("❌ PrivToken not configured", file=sys.stderr) sys.exit(1) # --from-job-id 模式：从数据库加载 RenderPlan if args.from_job_id: print(f"📋 loading RenderPlan from database (jobId={args.from_job_id})...") if not PRIVATE_TOKEN: print("❌ --from-job-id mode requires PRIV_TOKEN", file=sys.stderr) sys.exit(1) # render_job_client lives in skills/template-registry/scripts/ — the # shared location for cross-skill Python helpers (matches the pattern # render-video uses for the same import). _template_registry_scripts = os.path.join( os.path.dirname(__file__), "..", "..", "template-registry", "scripts" ) sys.path.insert(0, _template_registry_scripts) from render_job_client import get_plan as rjc_get_plan try: render_plan_str = rjc_get_plan(args.from_job_id, PRIVATE_TOKEN) except RuntimeError as e: print(f"❌ failed to fetch RenderPlan: {e}", file=sys.stderr) sys.exit(1) render_plan = json.loads(render_plan_str) scenes, rp_width, rp_height = convert_render_plan_to_scenes(render_plan) if args.width == 1080 and args.height == 1920: args.width = rp_width args.height = rp_height if not args.title: dsl_meta = render_plan.get("dslMeta", {}) args.title = dsl_meta.get("title", "") or "RenderPlan 草稿" print(f" ✅ converted {len(scenes)} scene(s), canvas {args.width}x{args.height}") # --from-render-plan 模式：从 RenderPlan 自动转换 elif args.from_render_plan: if not os.path.isfile(args.from_render_plan): print(f"❌ RenderPlan file does not exist: {args.from_render_plan}", file=sys.stderr) sys.exit(1) print(f"📋 converting from RenderPlan: {args.from_render_plan}") with open(args.from_render_plan, "r", encoding="utf-8") as f: render_plan = json.load(f) scenes, rp_width, rp_height = convert_render_plan_to_scenes(render_plan) # 使用 RenderPlan 中的画布尺寸（除非用户显式指定） if args.width == 1080 and args.height == 1920: args.width = rp_width args.height = rp_height # 自动推导标题（优先从 dslMeta.title 读取） if not args.title: dsl_meta = render_plan.get("dslMeta", {}) args.title = dsl_meta.get("title", "") or "RenderPlan 草稿" print(f" ✅ converted {len(scenes)} scene(s), canvas {args.width}x{args.height}") else: # 传统模式：必须提供 --scenes 和 --title if not args.scenes: print("❌ provide --scenes or --from-render-plan", file=sys.stderr) parser.print_help() sys.exit(1) if not args.title: print("❌ provide --title", file=sys.stderr) parser.print_help() sys.exit(1) scenes = parse_scenes(args.scenes) # 解析草稿根目录（始终作为 draftRootPath 传给 API） draft_root_path = resolve_draft_root_path(args.draft_root_path, args.system) # 检测是否需要后处理（textLayers 或任意字幕文本都需要应用 pill 样式） needs_postprocess = has_text_layers(scenes) or any(s.get("subtitleText") for s in scenes) if needs_postprocess: print("📝 textLayers or subtitleText detected; post-processing after generation (incl. Pill subtitle styling)") # 为 API 调用准备场景（剥离 textLayers，派生 subtitleText） api_scenes = prepare_api_scenes(scenes) # 提交任务 task_id = create_jianying_task( title=args.title, scenes=api_scenes, draft_root_path=draft_root_path, width=args.width, height=args.height, draft_name=args.draft_name, ) # 轮询状态 result_data = poll_task_status(task_id, args.poll_interval, args.max_wait) download_url = result_data.get("downloadUrl", "") file_name = result_data.get("fileName", "") if not download_url: print("❌ generation finished but no download URL was returned", file=sys.stderr) sys.exit(1) print(f"\n🎉 Jianying draft generated successfully!") if file_name: print(f" filename: {file_name}") print(f" download URL: {download_url}") # 默认下载 ZIP；含 textLayers/subtitleText 时强制下载以便后处理 if args.no_download and not needs_postprocess: print(f"\n💡 To download the ZIP, drop --no-download (or open the URL above)") return if args.no_download and needs_postprocess: print("⚠️ textLayers/subtitleText detected; --no-download is ignored (post-processing is local)") output_path = args.output or f"jianying_draft_{int(time.time())}.zip" download_zip(download_url, output_path) # 后处理 textLayers / Pill 字幕样式 if needs_postprocess: postprocess_draft_zip(output_path, scenes) print(f"\n💡 How to import into Jianying:") print(f" 1. Extract {output_path} into the Jianying draft directory:") print(f" {draft_root_path}") print(f" 2. Start or restart Jianying; the draft will appear in your project list") if __name__ == "__main__": main()