#!/usr/bin/env python3 """nex_topic_to_deck.py — a generic source → NotebookLM slide deck for "The Nex Brief" tech videos. The tech sibling of bunty_match_to_deck.py: instead of a cricket URL → Playwright scorecard PDF, it takes any tech source (article URL(s), a topic sentence, or a text/PDF file) and drives the SAME NotebookLM flow (notebook create → add sources → slides create → poll → download → extract). Output layout (matches the presenter pipeline): projects/{slug}/ reference/topic.txt (the source text, used as narrator "facts") reference/notebook_id.txt slides/deck.pdf slides/slide_001.jpg ... (via extract_pdf_slides.py) analysis/slides.json analysis/deck_meta.json ({num_slides, style: "nex"}) """ from __future__ import annotations import argparse import json import os import subprocess import sys import time from pathlib import Path _RP = Path(__file__).resolve() REPO_ROOT = _RP.parents[4] if _RP.parents[3].name == ".claude" else _RP.parents[3] if Path.cwd() != REPO_ROOT: os.chdir(REPO_ROOT) NLM = str(Path.home() / ".local/bin/nlm") # Tech-briefing narrative + aesthetic focus for the NotebookLM deck. Generic across # any tech topic; "The Nex Brief" hosted by Nex, a young tech commentator. NEX_FOCUS = ( "Create a punchy tech-briefing slide deck for 'The Nex Brief', a short video hosted by " "Nex, an energetic young tech commentator. Tell the story of the topic in ~9-11 beats: " "(1) TITLE CARD — the topic/headline + 'The Nex Brief'; (2) THE HEADLINE — what happened, " "in one line; (3) WHY IT MATTERS — the significance in plain terms; (4-6) KEY DETAILS — the " "specs, numbers, features, or how-it-works, one clear point per slide; (7) THE CATCH / " "TRADE-OFFS — limitations, risks, or the counter-take; (8) REACTIONS / CONTEXT — how it " "compares or what people are saying; (9) WHAT'S NEXT — implications or the road ahead; " "(10) TAKEAWAY — the one-line bottom line. " "AESTHETIC: modern clean tech-broadcast / product-keynote look — bold sans-serif typography, " "hero-sized numbers and spec callouts, dark UI panels with neon/RGB accent glows, subtle grid " "and dashboard motifs. ONE headline plus ONE key point or big number per slide; minimal prose, " "no bullet lists longer than 3 items, no walls of text. Each slide plays under ~10 seconds of " "spoken commentary, so prioritise visual clarity and readability at video resolution. Spell " "product names, versions and numbers exactly as in the sources. End on the takeaway slide — " "render each beat EXACTLY ONCE, do not duplicate slides." ) def run(cmd, *, check=True, quiet=False, retries=1, backoff=(10, 60)): if not quiet: print(f"[nex-deck] $ {' '.join(str(c) for c in cmd[:6])}…", file=sys.stderr) # On a non-zero exit print the CLI's OWN stderr/stdout tail before raising — the # 2026-08-27 `notebook create` deaths left no diagnostic bytes at all. `retries` # re-runs the call with backoff for the sites that have no other guard. for attempt in range(1, max(1, int(retries)) + 1): p = subprocess.run(cmd, capture_output=True, text=True) if p.returncode == 0 or not check: return p print(f"[nex-deck] FAILED rc={p.returncode} (attempt {attempt}/{retries}) " f"cmd={' '.join(str(c) for c in cmd[:4])}\n" f"[nex-deck] stderr: {(p.stderr or '').strip()[-800:]}\n" f"[nex-deck] stdout: {(p.stdout or '').strip()[-400:]}", file=sys.stderr, flush=True) if "authentication expired" in ((p.stdout or "") + (p.stderr or "")).lower(): print("[nex-deck] NotebookLM answered 'Authentication expired' while nlm-refresh keeps " "the session valid every 10 min: its write-endpoint throttle (2026-08-18), not " "auth — not retrying in-process, every refused call feeds the penalty", file=sys.stderr, flush=True) break if attempt < retries: time.sleep(backoff[min(attempt, len(backoff)) - 1]) # same schedule; anchor must differ raise subprocess.CalledProcessError(p.returncode, cmd, p.stdout, p.stderr) def nlm_create_notebook(title: str) -> str: res = run([NLM, "notebook", "create", title], retries=3) for line in (res.stdout or "").splitlines(): line = line.strip() if line and " " not in line and len(line) > 8: # a bare id return line # fallback: newest notebook with this title res = run([NLM, "list", "notebooks", "--json"], quiet=True) nbs = json.loads(res.stdout) hits = [n for n in nbs if (n.get("title") or n.get("name")) == title] if not hits: raise RuntimeError(f"Could not locate notebook '{title}'") hits.sort(key=lambda n: n.get("created_at") or n.get("updated_at") or "", reverse=True) return hits[0].get("id") or hits[0].get("notebook_id") def nlm_add_source(nb_id: str, *, url=None, text=None, file=None) -> None: """Add a URL / text / file source, with one inline retry (nlm source add is occasionally flaky on the first call in a fresh notebook).""" cmd = [NLM, "source", "add", nb_id, "--wait", "--wait-timeout", "300"] if url: cmd += ["--url", url] elif text: cmd += ["--text", text] elif file: cmd += ["--file", str(file)] for attempt in (1, 2): try: run(cmd) return except subprocess.CalledProcessError as e: print(f"[nex-deck] source add failed (exit {e.returncode}); " f"{'retrying' if attempt == 1 else 'giving up'}…", file=sys.stderr) if attempt == 1: time.sleep(10) else: raise def find_deck(nb_id): try: data = json.loads(run([NLM, "studio", "status", nb_id, "--json"], quiet=True).stdout) except (json.JSONDecodeError, subprocess.CalledProcessError): return None decks = [] for a in data: kind = (a.get("type") or a.get("artifact_type") or a.get("kind") or "").lower() if ("slide" in kind or "deck" in kind) and (a.get("status") or a.get("state") or "").lower() not in {"failed", "error"}: decks.append(a) decks.sort(key=lambda a: a.get("created_at") or a.get("updated_at") or "", reverse=True) return decks[0] if decks else None def wait_for_deck(nb_id, timeout=3900, poll=30): deadline = time.time() + timeout last = None while time.time() < deadline: art = find_deck(nb_id) if art: st = (art.get("status") or art.get("state") or "ready").lower() if st != last: print(f"[nex-deck] deck status={st}", file=sys.stderr); last = st if st in {"ready", "completed", "done", "succeeded"}: return if st in {"failed", "error"}: raise RuntimeError(f"NotebookLM deck failed: {art}") time.sleep(poll) raise TimeoutError(f"Deck not ready within {timeout}s") def download_deck(nb_id, dest: Path, retries=4, delay=10): dest.parent.mkdir(parents=True, exist_ok=True) for attempt in range(1, retries + 1): try: run([NLM, "download", "slide-deck", nb_id, "--format", "pdf", "--no-progress", "-o", str(dest)]) if dest.exists() and dest.stat().st_size > 0: return except subprocess.CalledProcessError as e: print(f"[nex-deck] download {attempt}/{retries} failed (exit {e.returncode})", file=sys.stderr) if attempt < retries: time.sleep(delay) raise RuntimeError("deck download failed") def extract(deck_pdf: Path, proj: Path): slides = proj / "slides"; sj = proj / "analysis" / "slides.json" slides.mkdir(parents=True, exist_ok=True); sj.parent.mkdir(parents=True, exist_ok=True) run([sys.executable, str(_RP.with_name("extract_pdf_slides.py")), "--pdf", str(deck_pdf), "--output-dir", str(slides), "--output-json", str(sj), "--dpi", "200"]) def main() -> int: ap = argparse.ArgumentParser(description="Topic/URL → NotebookLM tech deck for The Nex Brief") ap.add_argument("--slug", required=True) ap.add_argument("--title", default="The Nex Brief") ap.add_argument("--url", action="append", default=[], help="source URL (repeatable — e.g. story links for the brief)") ap.add_argument("--topic", help="a topic sentence / text source") ap.add_argument("--file", help="a local PDF/text source file") ap.add_argument("--focus", help="override the NotebookLM focus prompt") a = ap.parse_args() if not (a.url or a.topic or a.file): print("need at least one of --url / --topic / --file", file=sys.stderr); return 2 proj = REPO_ROOT / "projects" / a.slug (proj / "reference").mkdir(parents=True, exist_ok=True) (proj / "analysis").mkdir(parents=True, exist_ok=True) # narrator "facts" = the topic/URLs (analogous to match_facts.txt) facts = (a.topic or "") + "\n" + "\n".join(a.url) (proj / "reference" / "topic.txt").write_text(facts.strip() + "\n") print("① creating NotebookLM notebook …", file=sys.stderr) try: nb = nlm_create_notebook(f"{a.title} — {a.slug}") except subprocess.CalledProcessError as e: if "authentication expired" in f"{e.output or ''}{e.stderr or ''}".lower(): print("[nex-deck] refused by NotebookLM's write throttle — exit 75 (temporary; " "the render rescue re-fires this later)", file=sys.stderr, flush=True) return 75 raise (proj / "reference" / "notebook_id.txt").write_text(nb + "\n") print("② adding sources …", file=sys.stderr) for u in a.url: nlm_add_source(nb, url=u) if a.topic: nlm_add_source(nb, text=a.topic) if a.file: nlm_add_source(nb, file=a.file) print("③ generating the tech deck …", file=sys.stderr) # NEX_DECK_FORMAT=detailed_deck yields ~2x the slides from the same sources # (presenter_slides caps ~6; detailed_deck hit 14 on identical content, 2026-07-11) run([NLM, "slides", "create", nb, "--format", os.environ.get("NEX_DECK_FORMAT", "presenter_slides"), "--length", "default", "--focus", a.focus or NEX_FOCUS, "--confirm"], retries=3) wait_for_deck(nb) print("④ downloading + extracting slides …", file=sys.stderr) deck_pdf = proj / "slides" / "deck.pdf" download_deck(nb, deck_pdf) extract(deck_pdf, proj) sdata = json.load(open(proj / "analysis" / "slides.json")) n = len(sdata["slides"]) if isinstance(sdata, dict) else len(sdata) json.dump({"num_slides": n, "style": "nex"}, open(proj / "analysis" / "deck_meta.json", "w")) print(f"✓ deck ready: {n} slides → {proj}", file=sys.stderr) print(json.dumps({"slug": a.slug, "num_slides": n, "notebook": nb})) return 0 if __name__ == "__main__": sys.exit(main())