Upload phase_b/tools/broll_randomizer.py

phase_b/tools/broll_randomizer.py

@@ -0,0 +1,180 @@
+"""
+方案 B v2 落地工具 #2：B-Roll 随机组合 + CapCut 操作指令生成
+
+输入：批量生成的 ScriptVariant JSON
+输出：
+  1. 每条视频的 B-Roll 随机分配（24段素材，8动作×3角度）
+  2. 防查重参数（翻转/滤镜/裁剪）
+  3. CapCut 逐条操作手册
+  4. 20天发布排期表
+
+使用方法：
+  python broll_randomizer.py --input batch_result.json --output ./production/
+"""
+
+import json
+import random
+import os
+import csv
+import argparse
+from pathlib import Path
+from datetime import datetime, timedelta
+
+# 24 段素材：8 个动作 × 3 个角度
+BROLL_POOL = {
+    "product_show":    {"angle_A":"broll_01_product_front.mp4", "angle_B":"broll_02_product_side.mp4", "angle_C":"broll_03_product_topdown.mp4"},
+    "texture_squeeze": {"angle_A":"broll_04_squeeze_front.mp4", "angle_B":"broll_05_squeeze_side.mp4", "angle_C":"broll_06_squeeze_45deg.mp4"},
+    "texture_spread":  {"angle_A":"broll_07_spread_front.mp4", "angle_B":"broll_08_spread_side_slowmo.mp4", "angle_C":"broll_09_spread_topdown.mp4"},
+    "face_apply":      {"angle_A":"broll_10_apply_front.mp4", "angle_B":"broll_11_apply_side.mp4", "angle_C":None},
+    "absorption":      {"angle_A":"broll_12_absorb_front.mp4", "angle_B":"broll_13_absorb_sidecloseup.mp4", "angle_C":None},
+    "paper_test":      {"angle_A":"broll_14_paper_front.mp4", "angle_B":"broll_15_paper_side.mp4", "angle_C":"broll_16_paper_topdown.mp4"},
+    "result_glow":     {"angle_A":"broll_17_glow_front.mp4", "angle_B":"broll_18_glow_sidelight.mp4", "angle_C":None},
+    "cta":             {"angle_A":"broll_19_cta_front.mp4", "angle_B":"broll_20_cta_side.mp4", "angle_C":"broll_21_cta_topdown.mp4"},
+}
+
+FORMAT_BROLL_ACTIONS = {
+    "before_after": ["product_show","face_apply","result_glow","texture_spread","cta"],
+    "day_in_life": ["product_show","face_apply","texture_spread","absorption","result_glow","cta"],
+    "unboxing_test": ["product_show","texture_squeeze","texture_spread","face_apply","result_glow","cta"],
+    "reaction_duet": ["product_show","texture_spread","face_apply","result_glow","cta"],
+    "viral_challenge": ["product_show","face_apply","result_glow","cta"],
+    "educational_hack": ["texture_squeeze","texture_spread","absorption","result_glow","cta"],
+    "comparison_test": ["product_show","texture_spread","face_apply","result_glow","cta"],
+    "texture_asmr": ["texture_squeeze","texture_spread","absorption","product_show","cta"],
+    "storytelling": ["product_show","face_apply","texture_spread","result_glow","cta"],
+    "trend_setter": ["product_show","face_apply","result_glow","cta"],
+    "ingredient_deep_dive": ["texture_squeeze","texture_spread","absorption","result_glow","cta"],
+    "transformation_journey": ["product_show","face_apply","absorption","result_glow","cta"],
+    "myth_busting": ["product_show","texture_spread","absorption","result_glow","cta"],
+}
+
+FLIP_OPTIONS = [False, True]
+FILTER_OPTIONS = ["原色","暖调","冷调","褪色","日系","复古","清新","美食","电影","自然"]
+CROP_OPTIONS = ["无裁剪","放大105%","放大110%","居中裁剪","上移10%"]
+
+
+def assign_broll(video_format: str, seed: int = None) -> list[dict]:
+    if seed is not None: random.seed(seed)
+    actions = FORMAT_BROLL_ACTIONS.get(video_format, FORMAT_BROLL_ACTIONS["before_after"])
+    assigned = []
+    for action in actions:
+        pool = BROLL_POOL.get(action, {})
+        available = [k for k, v in pool.items() if v is not None]
+        if not available: continue
+        chosen = random.choice(available)
+        assigned.append({
+            "action": action, "angle": chosen, "file": pool[chosen],
+            "flip": random.choice(FLIP_OPTIONS),
+            "filter": random.choice(FILTER_OPTIONS),
+            "crop": random.choice(CROP_OPTIONS),
+        })
+    return assigned
+
+
+def generate_production_plan(variants_raw: list, output_dir: str) -> dict:
+    os.makedirs(output_dir, exist_ok=True)
+    random.seed(42)
+    valid = [v for v in variants_raw if v.get("video_format") in FORMAT_BROLL_ACTIONS]
+    plan = []
+    used = set()
+    for i, v in enumerate(valid):
+        broll = assign_broll(v["video_format"], seed=42 + i * 7)
+        fp = "|".join(b["file"] for b in broll) + str(broll[0]["flip"])
+        if fp in used:
+            broll = assign_broll(v["video_format"], seed=42 + i * 7 + 999)
+            fp = "|".join(b["file"] for b in broll) + str(broll[0]["flip"])
+        used.add(fp)
+        capcut_steps = _capcut_steps(v, broll, i+1)
+        plan.append({"sequence":i+1,"variant_id":v["variant_id"],"video_format":v["video_format"],
+                     "hook_text":v.get("hook_style",""),"caption":v.get("caption_local",""),
+                     "hashtags":v.get("hashtags",[]),"broll_assignment":broll,"fingerprint":fp,
+                     "capcut_steps":capcut_steps})
+
+    schedule = _make_schedule(plan)
+    plan_path = os.path.join(output_dir, "production_plan.json")
+    with open(plan_path,"w",encoding="utf-8") as f:
+        json.dump({"total_videos":len(plan),"generated_at":datetime.now().isoformat(),"production_plan":plan},f,ensure_ascii=False,indent=2)
+    md_path = os.path.join(output_dir, "CAPCUT_操作手册.md")
+    _make_md(plan, schedule, md_path)
+    csv_path = os.path.join(output_dir, "发布排期表.csv")
+    _make_csv(schedule, csv_path)
+    return {"plan_file":plan_path,"manual_file":md_path,"schedule_csv":csv_path,"total":len(plan)}
+
+
+def _capcut_steps(variant: dict, broll: list, seq: int) -> list[str]:
+    fn = f"output_{seq:03d}_{variant['variant_id']}.mp4"
+    steps = [
+        f"1. 打开模板「beauty_sunscreen_template」→ 另存为「{variant['variant_id']}」",
+        f"2. 口播音轨：拖入 /tts/{variant['variant_id']}.mp3",
+        f"3. 前置 Hook 自拍：录 5 秒手机自拍，说：「{variant.get('hook_style','')[:50]}...」",
+    ]
+    for j, b in enumerate(broll):
+        fn = " → 水平翻转" if b["flip"] else ""
+        steps.append(f"4.{j+1}. 视频轨-素材{j+1}：拖入 /broll/{b['file']}{fn}")
+    steps.append(f"5. 滤镜：{broll[0]['filter']}")
+    steps.append("6. 自动字幕 → 语言选越南语 → 确认")
+    steps.append(f"7. 文案：{variant.get('caption_local','')[:80]}")
+    steps.append(f"8. 导出 → /output/{fn}")
+    steps.append(f"9. 发布时粘贴 Hashtags：{' '.join(variant.get('hashtags',[])[:5])}")
+    return steps
+
+
+def _make_schedule(plan: list) -> list:
+    slots = ["07:30","12:30","19:00","20:00","21:00"]
+    schedule = []
+    start = datetime.now().replace(hour=0,minute=0,second=0,microsecond=0)+timedelta(days=1)
+    for i, item in enumerate(plan):
+        day = start + timedelta(days=i//5)
+        h,m = int(slots[i%5].split(":")[0]), int(slots[i%5].split(":")[1])
+        pt = day.replace(hour=h, minute=m)
+        schedule.append({
+            "video_seq":item["sequence"],"variant_id":item["variant_id"],
+            "video_format":item["video_format"],
+            "publish_date":pt.strftime("%Y-%m-%d"),"publish_time":pt.strftime("%H:%M"),
+            "publish_datetime":pt.isoformat(),
+            "hashtags":" ".join(item.get("hashtags",[])[:5]),
+        })
+    return schedule
+
+
+def _make_md(plan, schedule, path):
+    lines = ["# CapCut 批量视频生产 — 操作手册","",f"总视频数: {len(plan)}","",
+             "## 准备工作","",
+             "□ /broll/ 有 24 个素材文件 (broll_01 ~ broll_21)",
+             "□ /tts/ 有所有口播 .mp3",
+             "□ CapCut 模板「beauty_sunscreen_template」已创建",
+             "□ 滤镜已安装（暖调/冷调/日系/复古/清新/自然/电影）","",
+             "## 逐条操作（前 5 条示例）",""]
+    for item in plan[:5]:
+        lines.append(f"### {item['sequence']}. `{item['variant_id']}` ({item['video_format']})")
+        for s in item["capcut_steps"]: lines.append(f"- {s}")
+        lines.append("")
+    lines.append(f"*(共 {len(plan)} 条，完整见 production_plan.json)*\n\n## 发布排期（前 3 天）\n")
+    lines.append("| 日期 | 时间 | 视频 | 格式 |")
+    lines.append("|---|---|---|---|")
+    for s in schedule[:15]:
+        lines.append(f"| {s['publish_date']} | {s['publish_time']} | #{s['video_seq']} | {s['video_format']} |")
+    with open(path,"w",encoding="utf-8") as f: f.write("\n".join(lines))
+
+
+def _make_csv(schedule, path):
+    with open(path,"w",newline="",encoding="utf-8") as f:
+        w = csv.DictWriter(f,fieldnames=["publish_date","publish_time","video_seq","variant_id","video_format","hashtags"])
+        w.writeheader()
+        for s in schedule:
+            w.writerow({"publish_date":s["publish_date"],"publish_time":s["publish_time"],
+                        "video_seq":s["video_seq"],"variant_id":s["variant_id"],
+                        "video_format":s["video_format"],"hashtags":s["hashtags"]})
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--input", required=True)
+    parser.add_argument("--output", default="./production")
+    args = parser.parse_args()
+    with open(args.input,"r",encoding="utf-8") as f: data = json.load(f)
+    variants = data.get("variants",[])
+    result = generate_production_plan(variants, args.output)
+    print(f"\n✅ 生产计划已生成！{result['total']} 条视频")
+    print(f"   手册: {result['manual_file']}")
+    print(f"   排期: {result['schedule_csv']}")