#!/usr/bin/env python3 """Learning brief -> NotebookLM Audio Overview -> n8n (Slack). NotebookLM has no API, so this runs on the Mac with the `notebooklm` CLI (one-time `notebooklm login`). Your n8n workflow receives the mp3 and posts it to Slack, or posts the error when something failed. Usage: python3 podcast.py latest brief in briefs/ python3 podcast.py 2026-09-18 a specific day python3 podcast.py --dry-run show what would happen, touch nothing python3 podcast.py --force redo a day that already has an episode python3 podcast.py --repost send the existing mp3 to n8n again (no regeneration) Settings via env: LB_PODCAST_LANG (default nl_NL, English = en), LB_PODCAST_LENGTH (short|default|long, default default), LB_PODCAST_WEBHOOK, LB_PODCAST_OUT. """ import json import os import re import shutil import subprocess import sys import tempfile from pathlib import Path DIR = Path(__file__).resolve().parent BRIEFS = DIR / "briefs" LOG = DIR / "podcasts.json" OUT = Path(os.environ.get("LB_PODCAST_OUT", Path.home() / "Podcasts" / "learning-brief")) WEBHOOK = os.environ.get( "LB_PODCAST_WEBHOOK", "https://YOUR-N8N-HOST/webhook/YOUR-WEBHOOK-PATH", ) LANG = os.environ.get("LB_PODCAST_LANG", "nl_NL") LENGTH = os.environ.get("LB_PODCAST_LENGTH", "default") MAX_SOURCES = 8 MAX_UPLOAD_MB = 15 # n8n payload limit is 16 MB by default NLM = shutil.which("notebooklm") or str(Path.home() / ".local/bin/notebooklm") FFMPEG = shutil.which("ffmpeg") or "/opt/homebrew/bin/ffmpeg" FFPROBE = shutil.which("ffprobe") or "/opt/homebrew/bin/ffprobe" HOST_PROMPT = """One listener: [who you are and what you want to achieve this quarter]. They listen to this instead of reading their daily learning brief, so cover the brief fully and in its order. Open with the single action to apply today. Then take each numbered insight: explain the idea, say any script word for word, and say how it fits their situation as the brief describes it. Use the linked sources only to add concrete examples and detail for those insights, not for new topics. If the brief recommends a full episode, say which one, how long, and which minutes matter. Close by repeating the one action. No filler, no generic advice, no hype. Speak Dutch. Keep English marketing and sales jargon in English and say English scripts and quotes in English first, then give the Dutch version.""" class Failed(Exception): pass def run(cmd, timeout=180, check=True): p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) if check and p.returncode != 0: tail = (p.stderr or p.stdout or "").strip().splitlines()[-3:] raise Failed(f"{' '.join(cmd[1:3])}: {' | '.join(tail)}"[:400]) return p def find_id(obj): """First 'id' value anywhere in a CLI --json payload.""" if isinstance(obj, dict): if isinstance(obj.get("id"), str): return obj["id"] for v in obj.values(): found = find_id(v) if found: return found if isinstance(obj, list): for v in obj: found = find_id(v) if found: return found return None def parse_json(text): start = text.find("{") return json.loads(text[start:]) if start >= 0 else {} def pick_brief(arg): if arg: path = BRIEFS / f"{arg}.md" if not path.exists(): sys.exit(f"Geen brief voor {arg}") return path briefs = sorted(BRIEFS.glob("20*.md")) if not briefs: sys.exit("Geen briefs gevonden") return briefs[-1] def brief_meta(text): title = next((l[2:].strip() for l in text.splitlines() if l.startswith("# ")), "Learning brief") m = re.search(r"\*\*(?:Vandaag toepassen|Apply today):\*\*\s*(.+)", text) summary = m.group(1).strip() if m else "" urls = [] for u in re.findall(r"https?://[^\s)>\]]+", text): u = u.rstrip(".,;") if u not in urls: urls.append(u) has_insights = bool(re.search(r"^## 1\.", text, re.M)) return title, summary, urls[:MAX_SOURCES], has_insights def post(fields, audio=None): cmd = ["curl", "-sS", "-m", "120", "-X", "POST", WEBHOOK] for k, v in fields.items(): cmd += ["-F", f"{k}={v}"] if audio: cmd += ["-F", f"audio=@{audio};type=audio/mpeg"] p = subprocess.run(cmd, capture_output=True, text=True) print("n8n:", (p.stdout or p.stderr).strip()[:300]) def main(): args = [a for a in sys.argv[1:] if not a.startswith("--")] dry = "--dry-run" in sys.argv force = "--force" in sys.argv brief = pick_brief(args[0] if args else None) date = brief.stem title, summary, urls, has_insights = brief_meta(brief.read_text()) log = json.loads(LOG.read_text()) if LOG.exists() else {} if "--repost" in sys.argv: ep = log.get(date) if not ep or not Path(ep["mp3"]).exists(): sys.exit(f"Geen bestaande aflevering voor {date}") post({"status": "ok", "date": date, "title": title, "summary": summary, "notebook_url": ep["notebook_url"], "minutes": ep.get("minutes", "")}, audio=ep["mp3"]) return if date in log and not force: print(f"{date} heeft al een aflevering ({log[date].get('notebook_url')}). Gebruik --force.") return if not has_insights: print(f"{date}: stille dag, geen inzichten, geen podcast.") return if dry: print(json.dumps({"brief": str(brief), "title": title, "summary": summary, "sources": urls, "lang": LANG, "length": LENGTH, "webhook": WEBHOOK}, indent=2)) return notebook_url = "" try: prev = log.get(date, {}).get("notebook_url", "") skipped = log.get(date, {}).get("skipped_sources", []) if prev: # --force on a day that already has a notebook: reuse it, only make new audio nb_id = prev.rsplit("/", 1)[-1] notebook_url = prev else: nb = parse_json(run([NLM, "create", f"Learning brief {date}", "--json"]).stdout) nb_id = find_id(nb) if not nb_id: raise Failed("notebook aangemaakt maar geen id in de output") notebook_url = f"https://notebooklm.google.com/notebook/{nb_id}" added = [] src = parse_json(run([NLM, "source", "add", str(brief), "-n", nb_id, "--title", title, "--json"]).stdout) added.append(find_id(src)) skipped = [] for u in urls: p = run([NLM, "source", "add", u, "-n", nb_id, "--json", "--timeout", "90"], check=False) sid = find_id(parse_json(p.stdout)) if p.returncode == 0 else None (added if sid else skipped).append(sid or u) for sid in filter(None, added): run([NLM, "source", "wait", sid, "-n", nb_id, "--timeout", "300"], timeout=330, check=False) with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as f: f.write(HOST_PROMPT) run([NLM, "generate", "audio", "--prompt-file", f.name, "-n", nb_id, "--format", "deep-dive", "--length", LENGTH, "--language", LANG, "--wait", "--timeout", "1500", "--retry", "2", "--json"], timeout=1600) OUT.mkdir(parents=True, exist_ok=True) raw = OUT / f"learning-brief-{date}.m4a" mp3 = OUT / f"learning-brief-{date}.mp3" run([NLM, "download", "audio", str(raw), "-n", nb_id, "--force"], timeout=300) run([FFMPEG, "-y", "-loglevel", "error", "-i", str(raw), "-ac", "1", "-b:a", "64k", str(mp3)], timeout=300) raw.unlink(missing_ok=True) dur = run([FFPROBE, "-v", "error", "-show_entries", "format=duration", "-of", "csv=p=0", str(mp3)], check=False).stdout.strip() minutes = str(round(float(dur) / 60)) if dur else "" if mp3.stat().st_size > MAX_UPLOAD_MB * 1024 * 1024: raise Failed(f"mp3 is groter dan {MAX_UPLOAD_MB} MB; ze staat lokaal in {mp3}") log[date] = {"notebook_url": notebook_url, "mp3": str(mp3), "minutes": minutes, "skipped_sources": skipped} LOG.write_text(json.dumps(log, indent=2)) post({"status": "ok", "date": date, "title": title, "summary": summary, "notebook_url": notebook_url, "minutes": minutes}, audio=mp3) print(f"Klaar: {mp3} ({minutes} min), {notebook_url}") except (Failed, subprocess.TimeoutExpired) as e: print("MISLUKT:", e) post({"status": "failed", "date": date, "title": title, "notebook_url": notebook_url, "error": str(e)[:400]}) sys.exit(1) if __name__ == "__main__": main()