"""The up-front interview, the brief gate, approvals and voice auditions. fishvideo interview open questions, as JSON grouped by round (ask them with AskUserQuestion) fishvideo interview --set id=value ... record answers (value may be JSON: '["shorts","youtube"]') fishvideo interview --answers a.json record many answers {id: value} fishvideo interview --summary the brief as the user will hear it read back fishvideo interview --confirm the user said "yes, go": freezes the interview fishvideo gate [--stage script|render] fail while required answers, confirmation or copy are missing fishvideo approve storyboard|stills|final fishvideo voices --audition render the audition line in every shortlisted voice The interview is asked ONCE, before creation. Every answer is mapped into brief.yaml (palette -> brand.colors, music mood -> score, voice pick + delivery -> voice config, style -> defaults), so the rest of the pipeline never needs to ask again. """ from __future__ import annotations import argparse import json import re from pathlib import Path from fvcore import SKILL, FishError, ensure_project, load_json, load_yaml, log, main_guard, save_json, save_yaml, secret, warn CFG = SKILL / "config" PLACEHOLDER = re.compile(r"TODO|<[^<>\n]{2,80}>|lorem ipsum", re.I) # phrases the v1 scaffold produced; they must never ship STOCK_PHRASES = ["explained.", "in 60 seconds", "Here is ", "in three parts", "Step by step:", "Thanks for watching", "This is "] def bank() -> dict: return load_yaml(CFG / "interview.yaml") def styles() -> dict: return load_yaml(CFG / "styles.yaml") def voices_cfg() -> dict: return load_yaml(CFG / "voices.yaml") def _get(d: dict, path: str): cur = d for k in path.split("."): if not isinstance(cur, dict) or k not in cur: return None cur = cur[k] return cur def _set(d: dict, path: str, value) -> None: keys = path.split(".") cur = d for k in keys[:-1]: if not isinstance(cur.get(k), dict): cur[k] = {} cur = cur[k] cur[keys[-1]] = value def _options(q: dict, answers: dict) -> list[dict]: src = q.get("options_from") if not src: return q.get("options") or [] st = styles() if src == "styles": return [{"label": v["label"][:40], "value": k, "description": f"{v['look']}. Best for {v['best_for']}"} for k, v in st["styles"].items()] if src == "palettes": rec = (st["styles"].get(answers.get("style")) or {}).get("defaults", {}).get("palette") out = [] for k, v in st["palettes"].items(): c = v.get("colors") or {} desc = "Give your hex codes (bg, ink, accent)" if k == "brand" else f"bg {c.get('bg')} ink {c.get('ink')} accent {c.get('accent')}" out.append({"label": v["label"] + (" (Recommended)" if k == rec else ""), "value": k, "description": desc}) out.sort(key=lambda o: "(Recommended)" not in o["label"]) return out if src == "music_moods": rec = (st["styles"].get(answers.get("style")) or {}).get("defaults", {}).get("music") out = [{"label": v["label"] + (" (Recommended)" if k == rec else ""), "value": k, "description": f"library search: '{v['search']}', or an original synthesised score"} for k, v in st["music_moods"].items()] out.sort(key=lambda o: "(Recommended)" not in o["label"]) return out if src == "voices": lang = answers.get("language") or "en-US" delivery = answers.get("voice_energy") or "calm" vs = voices_cfg()["voices"].get(lang) or voices_cfg()["voices"]["en-US"] vs = sorted(vs, key=lambda v: delivery not in v.get("fits", [])) return [{"label": v["label"].split(" - ")[0], "value": v["id"], "description": v["label"] + ("" if v["provider"] == "edge" else f" [{v['provider']}]")} for v in vs] return [] def _applies(q: dict, answers: dict) -> bool: w = q.get("when") if not w: return True for k, allowed in w.items(): a = answers.get(k) vals = a if isinstance(a, list) else [a] if not any(str(x) in [str(y) for y in allowed] for x in vals): return False return True def state(project: Path) -> tuple[dict, dict]: brief = load_yaml(project / "brief.yaml") or {} iv = brief.get("interview") or {} return brief, iv def open_questions(project: Path) -> list[dict]: brief, iv = state(project) answers = iv.get("answers") or {} template = load_yaml(SKILL / "templates" / "brief.yaml") or {} rounds = [] for r in bank()["rounds"]: qs = [] for q in r["questions"]: if q["id"] in answers or not _applies(q, answers): continue item = {k: q[k] for k in ("id", "header", "question", "required") if k in q} item["multiSelect"] = bool(q.get("multi")) item["free_text"] = bool(q.get("free_text")) opts = _options(q, answers) if opts: item["options"] = opts prefill = _get(brief, q["maps_to"]) if q.get("maps_to") else None if prefill not in (None, "", [], {}) and prefill != _get(template, q["maps_to"]): # template defaults are not answers item["from_prompt"] = prefill # the prompt seemed to answer it: confirm, don't re-ask blindly qs.append(item) if qs: rounds.append({"round": r["id"], "title": r["title"], "note": r.get("note"), "questions": qs}) return rounds def _coerce(v): if isinstance(v, str): s = v.strip() if s[:1] in "[{" or s in ("true", "false", "null") or re.fullmatch(r"-?\d+(\.\d+)?", s): try: return json.loads(s) except json.JSONDecodeError: return v return v def record(project: Path, answers: dict) -> dict: brief, iv = state(project) if iv.get("confirmed"): warn("the interview was already confirmed; recording a change the user asked for") qmap = {q["id"]: q for r in bank()["rounds"] for q in r["questions"]} store = iv.setdefault("answers", {}) for k, v in answers.items(): if k not in qmap: raise FishError(f"unknown interview question '{k}' (see config/interview.yaml)") q = qmap[k] v = _coerce(v) if q.get("multi") and isinstance(v, str): v = [x.strip() for x in v.split(",") if x.strip()] if q.get("secret"): _store_keys(str(v)) store[k] = "(stored in .env)" continue store[k] = v brief["interview"] = iv apply(brief) save_yaml(project / "brief.yaml", brief) return store def _store_keys(text: str) -> None: env = SKILL / ".env" have = env.read_text(encoding="utf-8").splitlines() if env.exists() else [] keep = {ln.split("=", 1)[0]: ln for ln in have if "=" in ln} n = 0 for ln in text.splitlines(): m = re.match(r"\s*([A-Z][A-Z0-9_]{2,})\s*=\s*(\S+)\s*$", ln) if m: keep[m.group(1)] = f"{m.group(1)}={m.group(2)}" n += 1 env.write_text("\n".join(keep.values()) + "\n", encoding="utf-8") log(f"stored {n} key(s) in {env} (gitignored; never printed)") def apply(brief: dict) -> dict: """Map answers into the brief + style defaults. Idempotent.""" a = (brief.get("interview") or {}).get("answers") or {} st = styles() qmap = {q["id"]: q for r in bank()["rounds"] for q in r["questions"]} for k, v in a.items(): m = qmap.get(k, {}).get("maps_to") if isinstance(v, str) and qmap.get(k, {}).get("free_text") and v.strip().lower() in ("none", "no", "n/a", "skip", "-"): v = None # "no channel" must never become a badge that says "none" if k == "facts" and v: _apply_facts(brief, str(v)) if m and v not in (None, ""): if k == "platforms": v = [p for x in (v if isinstance(v, list) else [v]) for p in str(x).split(",")] if k == "points" and isinstance(v, str): v = [p.strip(" -•\t") for p in v.splitlines() if p.strip(" -•\t")] if k == "references" and isinstance(v, str): v = [] if v.strip().lower() in ("none", "no", "") else re.findall(r"\S+", v) _set(brief, m, v) if a.get("topic") and not brief.get("title_set"): brief["title"] = re.split(r"[:.\n]", str(a["topic"]))[0].strip()[:70] cta_text = {"follow": "Follow for more", "comment": "Comment below"}.get(a.get("cta")) if cta_text and not (brief.get("cta") or {}).get("text"): brief.setdefault("cta", {})["text"] = cta_text style = st["styles"].get(a.get("style") or brief.get("style") or "") d = (style or {}).get("defaults", {}) if style: brief.setdefault("texture", d.get("texture")) brief["pacing"] = {**(brief.get("pacing") or {}), **(style.get("pacing") or {})} brief["motif"] = style.get("motif") pal = st["palettes"].get(a.get("palette") or d.get("palette") or "") if pal and pal.get("colors"): brief.setdefault("brand", {})["colors"] = dict(pal["colors"]) mode = a.get("mode") brief["brand"]["mode"] = pal["mode"] if mode in (None, "auto") else mode elif isinstance(a.get("palette"), str) and "#" in a["palette"]: hexes = re.findall(r"#[0-9a-fA-F]{6}", a["palette"]) if hexes: c = brief.setdefault("brand", {}).setdefault("colors", {}) for key, h in zip(["bg", "ink", "accent", "accent2"], hexes): c[key] = h mood = st["music_moods"].get(a.get("music") or d.get("music") or "") if mood: brief["score"] = {**(brief.get("score") or {}), **{k: v for k, v in mood["score"].items() if k != "bpm"}} brief["tempo"] = {**(brief.get("tempo") or {}), "bpm": mood["score"].get("bpm", 120)} brief.setdefault("music", {})["search"] = mood["search"] if a.get("sfx"): brief.setdefault("sfx", {})["density"] = a["sfx"] if a.get("captions"): brief.setdefault("captions", {})["style"] = a["captions"] _apply_voice(brief, a) # Short vertical cuts: cut on beats, no dead air around or inside lines plats = brief.get("platforms") or [] if plats and all(p in ("shorts", "reels", "tiktok") for p in plats) or (brief.get("length") or 999) <= 90: pac = brief.setdefault("pacing", {}) for k, v in {"snap": "beat", "lead_in": 0.1, "tail": 0.2, "max_word_gap": 0.3}.items(): pac.setdefault(k, v) return brief def _apply_facts(brief: dict, text: str) -> None: """One fact per line: 'claim ... https://source'. Numbered facts become brief.facts; every link joins sources.""" facts = brief.get("facts") or [] sources = brief.get("sources") or [] for line in text.splitlines(): url = re.search(r"https?://\S+", line) if not url: continue src = url.group(0).rstrip(").,") claim = line.replace(url.group(0), "").strip(" -•()|\t") if src not in sources: sources.append(src) num = re.search(r"\d[\d,.]*\s?(%|[KMB]\\b|million|billion|trillion)?", claim) if num and not any(f.get("text") == claim for f in facts): facts.append({"text": claim, "value": num.group(0).strip(), "source": src}) brief["facts"], brief["sources"] = facts, sources def _apply_voice(brief: dict, a: dict) -> None: vc = voices_cfg() delivery = a.get("voice_energy") if delivery == "none" or (brief.get("style") and ((styles()["styles"].get(brief["style"]) or {}).get("defaults", {}).get("voice") == "none") and not a.get("voice")): brief.setdefault("voice", {})["provider"] = "none" return pick = a.get("voice") if not pick: return lang = a.get("language") or brief.get("language") or "en-US" allv = [v for vs in vc["voices"].values() for v in vs] v = next((x for x in allv if x["id"] == pick), None) pre = vc["delivery"].get(delivery or "calm", vc["delivery"]["calm"]) voice = brief.setdefault("voice", {}) voice["language"] = lang voice["post"] = pre["post"] voice["delivery"] = delivery or "calm" if v is None: # "Other": a raw edge voice name or an ElevenLabs id v = {"provider": "elevenlabs" if re.fullmatch(r"[A-Za-z0-9]{20}", str(pick)) else "edge", "voice": pick} if v["provider"] == "elevenlabs": voice["provider"] = "elevenlabs" voice["elevenlabs"] = {**(voice.get("elevenlabs") or {}), "voice_id": v["voice"], "voice_settings": {"stability": pre["stability"], "similarity_boost": 0.8, "style": pre["eleven_style"], "use_speaker_boost": True, "speed": pre["eleven_speed"]}} near = next((x for x in vc["voices"].get(lang, []) if x["provider"] == "edge"), None) if near: voice["fallback"] = {"provider": "edge-tts", "voice": near["voice"], "rate": pre["edge_rate"], "pitch": pre["edge_pitch"]} elif v["provider"] == "kokoro": voice["provider"] = "kokoro" voice["kokoro"] = {"voice": v["voice"], "speed": round(pre["wps"] / 2.9, 2)} near = next((x for x in vc["voices"].get(lang, []) if x["provider"] == "edge"), None) voice["fallback"] = {"provider": "edge-tts", "voice": (near or {}).get("voice", "en-US-AndrewMultilingualNeural"), "rate": pre["edge_rate"], "pitch": pre["edge_pitch"]} else: voice["provider"] = "edge-tts" voice["fallback"] = {"provider": "edge-tts", "voice": v["voice"], "rate": pre["edge_rate"], "pitch": pre["edge_pitch"]} brief.setdefault("pacing", {})["line_gap"] = pre["line_gap"] def summary(project: Path) -> str: brief, iv = state(project) a = iv.get("answers") or {} st = styles() s = st["styles"].get(a.get("style"), {}) lines = [f"# Brief summary: {brief.get('title') or a.get('topic', '')[:60]}", ""] for r in bank()["rounds"]: rows = [(q["header"], a[q["id"]]) for q in r["questions"] if q["id"] in a] if rows: lines.append(f"**{r['title']}**") lines += [f"- {h}: {json.dumps(v, ensure_ascii=False) if not isinstance(v, str) else v}" for h, v in rows] lines.append("") if s: lines += [f"Style bible: {s['look']}. Motif: {s['motif']}. A visual change every ~{s['pacing']['change_every']} s.", ""] c = (brief.get("brand") or {}).get("colors") or {} if c: lines.append("Colours: " + ", ".join(f"{k} {v}" for k, v in c.items() if k in ("bg", "ink", "accent", "accent2"))) v = brief.get("voice") or {} lines.append(f"Voice: {v.get('provider')} {((v.get('fallback') or {}).get('voice') if v.get('provider') != 'elevenlabs' else (v.get('elevenlabs') or {}).get('voice_id'))} ({v.get('delivery', '')})") open_req = [q["id"] for rr in open_questions(project) for q in rr["questions"] if q.get("required")] lines.append("") lines.append("Open required questions: " + (", ".join(open_req) if open_req else "none")) return "\n".join(lines) def confirm(project: Path) -> None: open_req = [q["id"] for rr in open_questions(project) for q in rr["questions"] if q.get("required")] if open_req: raise FishError(f"cannot confirm: required questions still open: {open_req}") brief, iv = state(project) iv["confirmed"] = True brief["interview"] = iv save_yaml(project / "brief.yaml", brief) log("interview confirmed: creation may start; no further questions until the final review") def copy_problems(project: Path) -> list[str]: """Placeholder / stock / copied-example text in the storyboard.""" sbf = project / "storyboard.yaml" if not sbf.exists(): return ["no storyboard.yaml"] sb = load_yaml(sbf) or {} probs = [] ex_lines = set() for exf in (SKILL / "examples").glob("*/storyboard.yaml"): for sc in (load_yaml(exf) or {}).get("scenes", []): for ln in (sc.get("say") if isinstance(sc.get("say"), list) else [sc.get("say")]): if ln and len(str(ln)) > 25: ex_lines.add(re.sub(r"\W+", " ", str(ln)).strip().lower()) for sc in sb.get("scenes", []): blob = json.dumps({"p": sc.get("params"), "say": sc.get("say")}, ensure_ascii=False) if PLACEHOLDER.search(blob): probs.append(f"{sc.get('id')}: placeholder text left ({PLACEHOLDER.search(blob).group(0)})") says = sc.get("say") if isinstance(sc.get("say"), list) else [sc.get("say")] for ln in says: if not ln: continue norm = re.sub(r"\W+", " ", str(ln)).strip().lower() if norm in ex_lines: probs.append(f"{sc.get('id')}: narration copied from an example storyboard: '{ln[:50]}'") for ph in STOCK_PHRASES: if ph.lower() in str(ln).lower() and len(str(ln)) < 60: probs.append(f"{sc.get('id')}: stock scaffold phrase '{ph.strip()}' in narration: rewrite it") if not ((load_yaml(project / "brief.yaml") or {}).get("publish") or {}).get("titles"): probs.append("publishing copy: write brief.publish.titles (5 options), description, tags and hashtags") th = sb.get("thumbnail") or {} if PLACEHOLDER.search(json.dumps(th)): probs.append("thumbnail: placeholder text left") return probs def gate(project: Path, stage: str = "render", force: bool = False) -> list[str]: project = ensure_project(project) brief, iv = state(project) errs = [] open_req = [q["id"] for rr in open_questions(project) for q in rr["questions"] if q.get("required")] if open_req: errs.append(f"interview: required questions still open: {open_req} (run `fishvideo interview {project}` and ask them)") if not iv.get("confirmed"): errs.append("interview: not confirmed: read `fishvideo interview --summary` to the user, then `--confirm`") if stage == "render": errs += copy_problems(project) need = set((brief.get("approvals") or {}).get("required") or []) got = load_json(project / "approvals.json", default={}) if (project / "approvals.json").exists() else {} for n in ("storyboard", "stills"): if n in need and not got.get(n): errs.append(f"approval: the user asked to approve the {n} first (fishvideo approve {n})") if errs and force: for e in errs: warn(f"gate (forced): {e}") return [] return errs def approve(project: Path, what: str) -> None: f = project / "approvals.json" got = load_json(f, default={}) if f.exists() else {} import time got[what] = time.strftime("%Y-%m-%d %H:%M") save_json(f, got) log(f"approved: {what}") def audition(project: Path) -> list[dict]: """The audition line in each shortlisted voice for the chosen language + delivery.""" import voice as voicemod brief, iv = state(project) a = iv.get("answers") or {} lang = a.get("language") or brief.get("language") or "en-US" vc = voices_cfg() pre = vc["delivery"].get(a.get("voice_energy") or "calm", vc["delivery"]["calm"]) line = vc["audition_line"]["hi" if lang.startswith("hi") else "en"] out = project / "build" / "auditions" out.mkdir(parents=True, exist_ok=True) res = [] for v in vc["voices"].get(lang, vc["voices"]["en-US"]): f = out / f"{v['id']}.mp3" try: if v["provider"] == "edge": audio, _ = voicemod.edge(line, {"voice": v["voice"], "rate": pre["edge_rate"], "pitch": pre["edge_pitch"]}) elif v["provider"] == "elevenlabs": key = secret("ELEVENLABS_API_KEY", project) if not key: res.append({"id": v["id"], "skipped": "no ELEVENLABS_API_KEY"}) continue audio, _ = voicemod.elevenlabs(line, {"elevenlabs": {"voice_id": v["voice"], "model_id": "eleven_multilingual_v2", "voice_settings": {"stability": pre["stability"], "similarity_boost": 0.8, "style": pre["eleven_style"], "use_speaker_boost": True}}}, key) elif v["provider"] == "kokoro": audio = voicemod.kokoro_audio(line, {"voice": v["voice"], "speed": round(pre["wps"] / 2.9, 2)}) else: continue f.write_bytes(audio) res.append({"id": v["id"], "label": v["label"], "file": str(f)}) except Exception as e: # noqa: BLE001 - an audition failing must not stop the others res.append({"id": v["id"], "skipped": str(e)[:120]}) save_json(out / "auditions.json", res) for r in res: log(f" {r['id']:<16} {r.get('file') or 'skipped: ' + r.get('skipped', '')}") return res def cli(argv=None) -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("project") ap.add_argument("--set", action="append", default=[]) ap.add_argument("--answers") ap.add_argument("--summary", action="store_true") ap.add_argument("--confirm", action="store_true") a = ap.parse_args(argv) project = ensure_project(a.project) ans = {} if a.answers: ans.update(json.loads(Path(a.answers).read_text(encoding="utf-8"))) for kv in a.set: k, _, v = kv.partition("=") ans[k.strip()] = v if ans: record(project, ans) if a.confirm: confirm(project) if a.summary: print(summary(project)) elif not a.confirm: print(json.dumps(open_questions(project), indent=1, ensure_ascii=False)) return 0 if __name__ == "__main__": main_guard(cli)