dickshow-index/analyze.py

"""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()