n3wk
267 lines
11 KiB
Python
267 lines
11 KiB
Python
"""Analyze episodes via the local `opencode run` CLI, routed through the
|
|
resilient runner in llm.py (model fallback chain + provider-death blacklist).
|
|
|
|
For each transcript:
|
|
1. export readable timestamped txt (export_txt.py)
|
|
2. call: opencode run -f <txt> "<prompt with rubric + schema>"
|
|
3. parse/validate the returned JSON (model fallback chain handles retries)
|
|
4. save data/analysis/epXXXX.json
|
|
|
|
Then aggregate all analyses into one master file for routine opencode access:
|
|
data/dickshow_analysis.json (run aggregate.py, or it runs automatically
|
|
after each new analysis in this script)
|
|
|
|
Resumable: existing per-episode analyses are skipped unless --force.
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timezone
|
|
|
|
from llm import run as llm_run
|
|
from llm import RateLimitError, ProviderError # noqa: F401 (compat)
|
|
from config import DATA_DIR, BASE_DIR
|
|
from feed import load_episodes
|
|
from download import slug as ep_slug, audio_path
|
|
from transcribe import transcript_path as tpath
|
|
from export_txt import export_txt, txt_path
|
|
from rubric import RUBRIC, SCHEMA_EXAMPLE
|
|
from facts import CAST_SHEET
|
|
|
|
ANALYSIS_DIR = os.path.join(DATA_DIR, "analysis")
|
|
MASTER_JSON = os.path.join(DATA_DIR, "dickshow_analysis.json")
|
|
|
|
OPENCODE = os.environ.get(
|
|
"OPENCODE_BIN", os.path.expanduser(r"~\.opencode\bin\opencode.exe"))
|
|
# analyzer model override, e.g. opencode/x-preview-f-free
|
|
OPENCODE_MODEL = os.environ.get("OPENCODE_ANALYZE_MODEL", "").strip()
|
|
OPENCODE_TIMEOUT = int(os.environ.get("DS_ANALYZE_TIMEOUT", "1800")) # 30 min
|
|
|
|
PROMPT_TEMPLATE = """You are the show archivist for The Dick Show podcast. Attached is the FULL transcript of one episode (timestamped lines like [HH:MM:SS]).
|
|
|
|
Your job: produce a structured analyst file for this episode. Read the ENTIRE transcript carefully, then reconstruct what was talked about, the arguments made, ideas/bits coined, and rate how edgy/transgressive the content is using the rubric below. The show contains extreme adult language; analyze it dispassionately like a content researcher - do not sanitize quotes, do not editorialize morally.
|
|
|
|
IMPORTANT - speaker attribution rules:
|
|
{cast}
|
|
Base attributions on what voices actually say (self-identifications, direct address) - never on famous-name pattern matching.
|
|
|
|
{rubric}
|
|
|
|
OUTPUT RULES:
|
|
- Output ONLY one valid JSON object. No markdown fences, no commentary before or after.
|
|
- Match exactly these keys (extra keys not allowed): {schema}
|
|
- All timestamps are SECONDS (number), converted from the [HH:MM:SS] markers.
|
|
- talking_points: cover every distinct topic IN ORDER of appearance
|
|
(5-15 items; summaries max ~40 words each).
|
|
- notable_quotes: the most characteristic/extreme/funny verbatim lines (5-10 items).
|
|
- If a field would be empty, use [] or "".
|
|
- ep_num and title must be taken from the transcript header comments.
|
|
|
|
Begin now."""
|
|
|
|
FENCES_RE = re.compile(r"^```(?:json)?\s*|\s*```$", re.MULTILINE)
|
|
|
|
|
|
def build_prompt() -> str:
|
|
return PROMPT_TEMPLATE.format(cast=CAST_SHEET.strip(),
|
|
rubric=RUBRIC.strip(),
|
|
schema=SCHEMA_EXAMPLE.strip())
|
|
|
|
|
|
_ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")
|
|
# "tool call" display lines opencode may echo before the model's reply:
|
|
# `→ Read file ...` / spinner glyphs / "Press Enter..." prompts.
|
|
_TOOLCALL_RE = re.compile(
|
|
r"(?m)^.*\x1b?\[?→.*$|^\s*(Press Enter|•|→)\s*$")
|
|
|
|
|
|
def _repair(text: str) -> str:
|
|
"""Common LLM-JSON repairs: ANSI codes, fences, trailing commas, missing commas."""
|
|
text = _ANSI_RE.sub("", text)
|
|
text = FENCES_RE.sub("", text).strip()
|
|
# extract just the outermost {...} JSON object if the reply is noisy
|
|
start = text.find("{")
|
|
end = text.rfind("}")
|
|
if start >= 0 and end > start:
|
|
text = text[start:end + 1]
|
|
elif text.strip():
|
|
return text.strip()
|
|
else:
|
|
return ""
|
|
# trailing commas before } or ]
|
|
text = re.sub(r",(\s*[}\]])", r"\1", text)
|
|
# missing commas between object members: "key": "value" "key2": "value2" -> add comma
|
|
text = re.sub(r'"\s+"', '", "', text)
|
|
# missing commas between array elements: "value1" "value2" -> add comma
|
|
text = re.sub(r'"\s*\n\s*"', '",\n"', text)
|
|
# missing commas after closing brace/bracket followed by opening brace/bracket
|
|
text = re.sub(r'(\]|\})\s*(\[|\{)', r'\1,\2', text)
|
|
return text
|
|
|
|
|
|
def strip_to_json(text: str):
|
|
# strict=False: tolerate raw newlines/tabs inside strings (common LLM slip)
|
|
return json.loads(_repair(text), strict=False)
|
|
|
|
|
|
REPAIR_PREAMBLE = (
|
|
"IMPORTANT: Your reply must be ONE valid strict JSON object matching the "
|
|
"schema below. Double-quoted keys/strings, escape quotes inside strings "
|
|
"(\\\"), no trailing commas, no markdown fences, no commentary.\n\n"
|
|
)
|
|
|
|
|
|
def ensure_content_match(obj: dict, attach_txt: str) -> None:
|
|
"""Guard against cross-episode contamination: the analysis must reference
|
|
words that actually occur in this episode's transcript."""
|
|
import json as _json
|
|
with open(attach_txt, encoding="utf-8") as f:
|
|
lines = [ln for ln in f if ln.startswith("[")]
|
|
if not lines:
|
|
return
|
|
sample = lines[:40] + lines[len(lines)//2:len(lines)//2+40] + lines[-20:]
|
|
stop = {"there", "about", "would", "because", "really", "going", "gonna",
|
|
"know", "like", "that", "this", "with", "what", "when", "they",
|
|
"them", "then", "than", "have", "been", "were", "just", "yeah"}
|
|
words = {w.strip(".,!?\"':;()[]").lower()
|
|
for ln in sample for w in ln.split()
|
|
if len(w) > 4 and w.strip(".,!?\"':;()[]").lower() not in stop}
|
|
blob = _json.dumps(obj, ensure_ascii=False).lower()
|
|
hits = {w for w in words if w in blob}
|
|
if len(hits) < 5:
|
|
raise ValueError(
|
|
f"content mismatch: analysis shares only {len(hits)} vocabulary "
|
|
f"items with this transcript - likely wrong-session output")
|
|
|
|
|
|
def validate(obj: dict) -> None:
|
|
required_top = ["ep_num", "title", "one_line_summary", "guests",
|
|
"talking_points", "arguments", "ideas", "notable_quotes",
|
|
"transgression"]
|
|
missing = [k for k in required_top if k not in obj]
|
|
if missing:
|
|
raise ValueError(f"missing keys: {missing}")
|
|
tg = obj["transgression"]
|
|
for k in ["overall_ti", "profanity", "slur_usage",
|
|
"harassment_or_cruelty", "sincerity"]:
|
|
if k not in tg or not isinstance(tg[k], int) or not 0 <= tg[k] <= 10:
|
|
raise ValueError(f"transgression.{k} must be int 0-10")
|
|
for q in obj.get("notable_quotes", []):
|
|
if not isinstance(q.get("ti"), int) or not 0 <= q["ti"] <= 10:
|
|
raise ValueError("quote ti must be int 0-10")
|
|
|
|
|
|
def run_opencode(prompt: str, attach: str) -> str:
|
|
# NOTE: deliberately NO `-c`/`-s` session reuse. The resilient runner
|
|
# (llm.py) owns fallback/retry/cooldown; this is the compat shim used by
|
|
# verify.py for single-file analysis (JSON parsed by the caller).
|
|
return llm_run(prompt, [attach], timeout=OPENCODE_TIMEOUT,
|
|
label="analyze")
|
|
|
|
|
|
# When the provider quota dries up, stop *attempting* analyses pipeline-wide
|
|
# for a while; transcription (local) keeps running. Cooldown state lives in
|
|
# llm.py and is shared with verify + pipeline_all.
|
|
_RATE_COOLDOWN = {"until": 0.0}
|
|
COOLDOWN_SECONDS = int(os.environ.get("DS_RATE_COOLDOWN", "900"))
|
|
|
|
|
|
def analyze_episode(ep: dict, force: bool = False, retries: int = 1) -> dict | None:
|
|
"""Analyze one episode. Retries/fallback/cooldown handled by llm.run.
|
|
|
|
Returns the analysis dict, or None if the episode was already done or the
|
|
provider is unavailable (callers catch up on a later pass).
|
|
"""
|
|
os.makedirs(ANALYSIS_DIR, exist_ok=True)
|
|
apath = os.path.join(ANALYSIS_DIR, f"{ep_slug(ep)}.json")
|
|
if os.path.exists(apath) and not force:
|
|
return None
|
|
if not os.path.exists(tpath(ep)):
|
|
print(f" no transcript for {ep['title']}; skipping", flush=True)
|
|
return None
|
|
|
|
attach = export_txt(ep)
|
|
prompt = build_prompt()
|
|
|
|
def parse_obj(raw: str) -> dict:
|
|
obj = strip_to_json(raw)
|
|
validate(obj)
|
|
ensure_content_match(obj, attach)
|
|
return obj
|
|
|
|
try:
|
|
obj = llm_run(prompt, [attach], timeout=OPENCODE_TIMEOUT,
|
|
label="analyze", parse=parse_obj)
|
|
except (RateLimitError, ProviderError) as e:
|
|
print(f" ANALYSIS NOT RUN {ep['title']}: {e}", flush=True)
|
|
return None
|
|
|
|
obj["ep_num"] = ep["ep_num"] if ep["ep_num"] else obj.get("ep_num")
|
|
obj["title"] = obj.get("title") or ep["title"]
|
|
obj["pub_date"] = ep["pub_date"]
|
|
obj["_analyzed_at"] = datetime.now(timezone.utc).isoformat()
|
|
obj["_source_transcript"] = os.path.basename(tpath(ep))
|
|
with open(apath + ".tmp", "w", encoding="utf-8") as f:
|
|
json.dump(obj, f, ensure_ascii=False, indent=1)
|
|
os.replace(apath + ".tmp", apath)
|
|
ti = obj["transgression"]["overall_ti"]
|
|
n_tp = len(obj.get("talking_points", []))
|
|
n_q = len(obj.get("notable_quotes", []))
|
|
print(f" OK {os.path.basename(apath)} "
|
|
f"(TI={ti}, {n_tp} points, {n_q} quotes)", flush=True)
|
|
return obj
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("eps", nargs="*", type=int)
|
|
ap.add_argument("--force", action="store_true")
|
|
ap.add_argument("--no-verify", action="store_true",
|
|
help="skip the dossier fact-check second pass")
|
|
args = ap.parse_args()
|
|
eps = load_episodes()
|
|
todo = [e for e in eps
|
|
if (not args.eps or e["ep_num"] in args.eps)
|
|
and os.path.exists(tpath(e))]
|
|
print(f"analyzing {len(todo)} episode(s) with {os.path.basename(OPENCODE)}",
|
|
flush=True)
|
|
|
|
# lazy import: verify.py imports this module at its own top level
|
|
from verify import verify_episode
|
|
|
|
done = 0
|
|
for i, ep in enumerate(todo, 1):
|
|
label = f"Ep {ep['ep_num']}" if ep["ep_num"] else ep["title"][:40]
|
|
existing = os.path.exists(os.path.join(
|
|
ANALYSIS_DIR, f"{ep_slug(ep)}.json"))
|
|
if not existing:
|
|
print(f"[{i}/{len(todo)}] analyzing {label} ...", flush=True)
|
|
if analyze_episode(ep, force=args.force) is not None:
|
|
done += 1
|
|
# ALWAYS fact-check fresh analyses against the dossier
|
|
if not args.no_verify:
|
|
print(f" fact-checking {label} ...", flush=True)
|
|
verify_episode(ep)
|
|
elif not args.no_verify:
|
|
# already analyzed -> run/refresh the dossier fact-check
|
|
vpath = os.path.join(ANALYSIS_DIR, f"{ep_slug(ep)}.json")
|
|
with open(vpath, encoding="utf-8") as f:
|
|
already = "_verification" in json.load(f)
|
|
if args.force or not already:
|
|
print(f"[{i}/{len(todo)}] verifying (existing) {label} ...",
|
|
flush=True)
|
|
verify_episode(ep)
|
|
# keep the master file fresh after every run
|
|
from aggregate import aggregate
|
|
aggregate()
|
|
print(f"analyzed={done} total_files="
|
|
f"{len([f for f in os.listdir(ANALYSIS_DIR) if f.endswith('.json')])}",
|
|
flush=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|