diff --git a/.gitignore b/.gitignore index 10a5b8d..b5a0e82 100644 --- a/.gitignore +++ b/.gitignore @@ -18,6 +18,9 @@ data/analysis/ data/index.db data/index.db-wal data/index.db-shm +data/episodes.jsonl +data/dickshow_analysis.json +data/lm_blacklist.json # IDE .idea/ diff --git a/aggregate.py b/aggregate.py index 922e029..ec97122 100644 --- a/aggregate.py +++ b/aggregate.py @@ -9,20 +9,23 @@ Contents: """ import json import os +import re from datetime import datetime, timezone from config import DATA_DIR from analyze import ANALYSIS_DIR, MASTER_JSON from feed import load_episodes +_EP_FILE_RE = re.compile(r"^ep\d{4}\.json$") + def load_analyses() -> list[dict]: out = [] if not os.path.isdir(ANALYSIS_DIR): return out for f in sorted(os.listdir(ANALYSIS_DIR)): - if not f.endswith(".json"): - continue + if not _EP_FILE_RE.match(f): + continue # ignore stray files like epNNNN_corrected.json try: with open(os.path.join(ANALYSIS_DIR, f), encoding="utf-8") as fh: out.append(json.load(fh)) diff --git a/analyze.py b/analyze.py index 5da3d8e..30cc866 100644 --- a/analyze.py +++ b/analyze.py @@ -1,14 +1,15 @@ -"""Analyze episodes via the local `opencode run` CLI. +"""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 "" - 3. parse/validate the returned JSON (retry up to 3x on bad output) + 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) + after each new analysis in this script) Resumable: existing per-episode analyses are skipped unless --force. """ @@ -16,11 +17,12 @@ import argparse import json import os import re -import subprocess 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 @@ -36,7 +38,7 @@ 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", "900")) # 15 min +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]). @@ -69,14 +71,26 @@ def build_prompt() -> str: 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: fences, trailing commas, stray text, missing commas.""" - text = FENCES_RE.sub("", text.strip()).strip() + """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 or end < len(text) - 1: - if start != -1 and end > start: - text = text[start:end + 1] + 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 @@ -141,101 +155,69 @@ def validate(obj: dict) -> None: def run_opencode(prompt: str, attach: str) -> str: - # NOTE: deliberately NO `-c`/`-s` session reuse: `run -c` continues the - # globally-last session, which under concurrent pipelines cross-pollinates - # analyses between episodes. Every call is a fresh session. - # --auto: headless runs must not stall on permission prompts (the model - # sometimes Reads attached files for more context); scope is our own data. - cmd = [OPENCODE, "run", "--auto", "-f", attach, - "--dir", BASE_DIR] - if OPENCODE_MODEL: - cmd += ["-m", OPENCODE_MODEL] - cmd.append(prompt) - proc = subprocess.run(cmd, capture_output=True, text=True, - encoding="utf-8", errors="replace", - cwd=BASE_DIR, timeout=OPENCODE_TIMEOUT) - out = proc.stdout or "" - err = proc.stderr or "" - combined = (out + err).lower() - if "rate limit" in combined or "429" in combined: - raise RateLimitError("provider rate limit exceeded") - if not out.strip(): - raise RuntimeError(f"opencode produced no stdout: {err[-400:]}") - return out + # 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") +# legacy alias kept for imports class RateLimitError(RuntimeError): pass # When the provider quota dries up, stop *attempting* analyses pipeline-wide -# for a while; transcription (local) keeps running. Probe again after cooldown. +# 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 = 3) -> dict | None: +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 time.time() < _RATE_COOLDOWN["until"]: - print(f" (analysis cooldown {int(_RATE_COOLDOWN['until'] - time.time())}s " - f"- will catch up later)", flush=True) - 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() - last_err = None - # escalating waits; rate limits need minutes, not seconds - waits = [10, 60, 180, 420, 600, 900] - attempt = 0 - while attempt < len(waits): - attempt += 1 - t0 = time.time() - try: - raw = run_opencode( - (REPAIR_PREAMBLE + prompt) if attempt > 1 else prompt, attach) - obj = strip_to_json(raw) - validate(obj) - ensure_content_match(obj, attach) - 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) - dt = time.time() - t0 - 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)} in {dt:.0f}s " - f"(TI={ti}, {n_tp} points, {n_q} quotes)", flush=True) - return obj - except RateLimitError as e: - last_err = e - wait = waits[min(attempt, len(waits) - 1)] - print(f" rate-limited (attempt {attempt}); " - f"sleeping {wait}s ...", flush=True) - time.sleep(wait) - except Exception as e: # noqa: BLE001 - last_err = e - wait = waits[min(attempt - 1, len(waits) - 1)] - print(f" attempt {attempt} failed: {e}; retry in {wait}s", - flush=True) - time.sleep(wait) - print(f" GIVING UP on {ep['title']} after {attempt} attempts: " - f"{last_err}", flush=True) - if isinstance(last_err, RateLimitError): - _RATE_COOLDOWN["until"] = time.time() + COOLDOWN_SECONDS - print(f" entering analysis cooldown for {COOLDOWN_SECONDS}s " - f"(transcription continues; analyses catch up later)", - flush=True) - return None + + 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: diff --git a/llm.py b/llm.py new file mode 100644 index 0000000..86f5db4 --- /dev/null +++ b/llm.py @@ -0,0 +1,284 @@ +"""Resilient opencode runner for analysis/verify passes. + +Why this exists (after a string of outages): + * The opencode "Console" provider models are unstable: some get retired or + break upstream and then return "[404] Provider returned error" while still + appearing in `opencode models` (nemotron-3.5-lightning-free, 2026-08-25). + Retrying the SAME broken model loops for hours and always fails. + * Rate limits are account-wide (shared by all opencode/* models), so a 429 + should wait out the cooldown, NOT flip models. + * Provider-dead errors should flip to the next WORKING model immediately. + * A model that returns text but garbage/non-JSON output should be skipped + in favor of the next candidate, not sent back through an outer retry loop. + +This module is the SINGLE retry/fallback/cooldown authority. Callers +(analyze.py, verify.py, pipeline_all.py) just call run() and handle the +outcome or the raised RateLimitError/ProviderError. +""" +import json +import os +import re +import subprocess +import sys +import time + +from config import BASE_DIR, DATA_DIR + +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() +ANALYZE_TIMEOUT = int(os.environ.get("DS_ANALYZE_TIMEOUT", "1800")) # 30 min +VERIFY_TIMEOUT = int(os.environ.get("DS_VERIFY_TIMEOUT", "900")) # 15 min + +# Preferred model first; everything else is a fallback tried in order until +# one returns usable output. Updated from `opencode models` 2026-08-25. +MODEL_CHAIN = [ + "opencode/big-pickle", + "opencode/nemotron-3-ultra-free", + "opencode/hy3-free", + "opencode/ling-3.0-flash-fin-free", + "opencode/mimo-v2.5-free", + "opencode/muse-spark-1.2-contributor-free", + "opencode/nemotron-3.5-lightning-free", +] + +# Models that hard-failed with provider 404 / model-not-found at runtime. +# Persisted so later runs skip them immediately instead of burning retries. +BLACKLIST_JSON = os.path.join(DATA_DIR, "lm_blacklist.json") + +# Cooldown after consecutive rate limits, shared by every caller. +COOLDOWN_SECONDS = int(os.environ.get("DS_RATE_COOLDOWN", "900")) +_COOLDOWN = {"until": 0.0} + +# Escalating wait between attempts *for the same model* on any failure. +WAITS = [10, 60, 180, 420, 600, 900] +# Per-model retry budget for blips before moving to the next model. +BLIP_RETRIES = 2 +# Rate-limit turns on the same model before bailing to the cooldown. +RATE_ESCALATION = 3 + + +class RateLimitError(RuntimeError): + pass + + +class ProviderError(RuntimeError): + pass + + +# --- error classification ------------------------------------------------- +# ORDER MATTERS: rate first, then 404/model-death, then generic transient. +_RATE_RE = re.compile( + r"rate limit|too many requests|\b429\b|quota exceeded", re.IGNORECASE) +_DEAD_RE = re.compile( + r"\[404\]|provider returned error[^\n]*\b4\d\d\b|model not found|" + r"model_not_found|model does not exist|not available", + re.IGNORECASE) +_TRANSIENT_RE = re.compile( + r"\b5\d\d\b|502|503|504|bad gateway|temporar|timed? ?out|connection|" + r"\bEOF\b|upstream request failed|http 500|500", + re.IGNORECASE) + + +def _load_blacklist() -> list[str]: + try: + with open(BLACKLIST_JSON, encoding="utf-8") as f: + data = json.load(f) + return [m for m in data.get("models", []) if isinstance(m, str)] + except (OSError, json.JSONDecodeError): + return [] + + +def _save_blacklist(models: list[str]) -> None: + os.makedirs(os.path.dirname(BLACKLIST_JSON), exist_ok=True) + tmp = BLACKLIST_JSON + ".tmp" + with open(tmp, "w", encoding="utf-8") as f: + json.dump({"models": sorted(set(models))}, f, indent=1) + os.replace(tmp, BLACKLIST_JSON) + + +def _blacklist_model(model: str) -> None: + models = _load_blacklist() + if model not in models: + models.append(model) + _save_blacklist(models) + print(f" [lm] BLACKLISTED {model} (provider-dead)", + file=sys.stderr, flush=True) + + +def active_models(env_override: str = "") -> list[str]: + """Ordered fallback chain, minus blacklisted. Env override goes first.""" + blacklist = set(_load_blacklist()) + chain: list[str] = [] + if env_override and env_override not in blacklist: + chain.append(env_override) + for m in MODEL_CHAIN: + if m not in chain and m not in blacklist: + chain.append(m) + if not chain: + # everything blacklisted - reset and warn + print(" [lm] ALL models blacklisted; resetting blacklist", + file=sys.stderr, flush=True) + _save_blacklist([]) + chain = [m for m in MODEL_CHAIN if m != env_override][:1] + return chain + + +def classify_combined(out: str, err: str) -> str | None: + """Return 'rate' | 'dead' | 'transient' | None based on provider output.""" + combined = (out + "\n" + err) + if _RATE_RE.search(combined): + return "rate" + if _DEAD_RE.search(combined): + return "dead" + if _TRANSIENT_RE.search(combined): + return "transient" + return None + + +def run(prompt: str, attachments: list[str], *, + timeout: int | None = None, label: str = "opencode", + parse=None, env_override: str | None = None, + print_progress: bool = True): + """Resilient `opencode run` against the model chain. + + Returns parse(raw_stdout) if `parse` is given (and it must return/raise), + else the raw stdout text. Raises RateLimitError or ProviderError once the + chain and escalation budgets are exhausted. + """ + if env_override is None: + env_override = os.environ.get("OPENCODE_ANALYZE_MODEL", "").strip() + timeout = timeout or ANALYZE_TIMEOUT + + if time.time() < _COOLDOWN["until"]: + raise RateLimitError( + f"in cooldown for {int(_COOLDOWN['until'] - time.time())}s") + + models = active_models(env_override) + if not models: + raise ProviderError("no usable models after blacklist") + + last_transient = None + last_rate = None + saw_rate = False + bad_output = [] # models that answered but failed `parse` + + for model in models: + blips_left = BLIP_RETRIES + for round_no in range(RATE_ESCALATION + 1): + cmd = [OPENCODE, "run", "--auto"] + for a in attachments: + cmd += ["-f", a] + cmd += ["--dir", BASE_DIR, "-m", model, prompt] + if print_progress: + print(f" [{label}] running analysis " + f"(attempt {round_no + 1} ({model}, {timeout}s))...", + flush=True) + t0 = time.time() + try: + proc = subprocess.run( + cmd, capture_output=True, text=True, + encoding="utf-8", errors="replace", + cwd=BASE_DIR, timeout=timeout) + except subprocess.TimeoutExpired: + blips_left -= 1 + last_transient = f"{model} hit timeout {timeout}s" + if blips_left <= 0: + break # next model + wait = min(60, 2 ** blips_left * 15) + if print_progress: + print(f" [{label}] timed out after {timeout}s; " + f"retry in {wait}s", flush=True) + time.sleep(wait) + continue + + out = proc.stdout or "" + err = proc.stderr or "" + cls = classify_combined(out, err) + + if cls == "rate": + saw_rate = True + last_rate = (model, err[-200:]) + if round_no >= RATE_ESCALATION: + _COOLDOWN["until"] = time.time() + COOLDOWN_SECONDS + raise RateLimitError( + f"{model} rate-limited (entering " + f"{COOLDOWN_SECONDS}s cooldown)") + wait = WAITS[min(round_no, len(WAITS) - 1)] + if print_progress: + print(f" [{label}] rate-limited ({model}); " + f"sleeping {wait}s ...", flush=True) + time.sleep(wait) + continue # same model, longer wait + + if cls == "dead": + _blacklist_model(model) + break # next model + + if cls == "transient": + blips_left -= 1 + last_transient = f"{model}: {err[-200:]}" + if blips_left <= 0: + break # next model + wait = min(60, 2 ** blips_left * 15) + if print_progress: + print(f" [{label}] transient error ({model}); " + f"retry in {wait}s", flush=True) + time.sleep(wait) + continue + + if not out.strip(): + blips_left -= 1 + last_transient = f"{model}: no stdout ({err[-200:].strip()})" + if blips_left <= 0: + break # next model + wait = min(60, 2 ** blips_left * 15) + if print_progress: + print(f" [{label}] no output from {model}; " + f"retry in {wait}s", flush=True) + time.sleep(wait) + continue + + elapsed = time.time() - t0 + if parse is not None: + try: + result = parse(out) + except Exception as e: # bad output: skip this model + bad_output.append((model, str(e)[:120])) + if print_progress: + print(f" [{label}] {model} gave unparsable output " + f"({str(e)[:90]}); next model", flush=True) + break + if print_progress: + print(f" [{label}] {model} OK in {elapsed:.0f}s", + flush=True) + return result + if print_progress: + print(f" [{label}] {model} returned " + f"{len(out)} chars in {elapsed:.0f}s", flush=True) + return out + + if saw_rate: + _COOLDOWN["until"] = time.time() + COOLDOWN_SECONDS + raise RateLimitError("all models rate-limited: " + f"{last_rate[1] if last_rate else 'unknown'}") + raise ProviderError(( + f"no model produced valid output; models_with_bad_output=" + f"{bad_output}; last_transient={last_transient}" + )) + + +if __name__ == "__main__": + # smoke test: pick the first live model and ask it to say hello + t0 = time.time() + try: + res = run("Reply with exactly: OK", [], timeout=90, label="smoke") + print(f"SMOKE OK: {res.strip()[:80]} in {time.time()-t0:.0f}s") + print(f"active models: {active_models()}") + except RateLimitError as e: + print(f"SMOKE RATE-LIMITED: {e}") + sys.exit(3) + except ProviderError as e: + print(f"SMOKE EXHAUSTED: {e}") + sys.exit(4) \ No newline at end of file diff --git a/pipeline_all.py b/pipeline_all.py index d9fda11..60a1db1 100644 --- a/pipeline_all.py +++ b/pipeline_all.py @@ -17,6 +17,7 @@ import sys import time import traceback import subprocess +from pathlib import Path try: sys.stdout.reconfigure(encoding="utf-8", errors="replace") @@ -143,94 +144,67 @@ def do_transcribe(model, ep: dict) -> None: pass +from llm import run as llm_run, RateLimitError, ProviderError +from llm import ANALYZE_TIMEOUT + + def run_analysis_with_timeout(ep: dict) -> tuple: """ - Run analysis + verification using opencode subprocess with hard timeout. - Returns (success: bool, result: dict | None, error: str | None) + Run analysis + verification using the resilient opencode runner (model + fallback chain + provider-death blacklist). Returns (success, result, error). """ - from analyze import ANALYSIS_DIR, OPENCODE, build_prompt, strip_to_json, validate + from analyze import ANALYSIS_DIR, build_prompt, validate, strip_to_json + from analyze import ensure_content_match from verify import verify_episode from download import slug as ep_slug - from export_txt import txt_path - from config import BASE_DIR - + from export_txt import export_txt + label = f"Ep {ep['ep_num']}" apath = os.path.join(ANALYSIS_DIR, f"{ep_slug(ep)}.json") - + # Check if already analyzed if os.path.exists(apath): with open(apath, encoding="utf-8") as f: existing = json.load(f) return (True, existing, None) - + # Build prompt prompt = build_prompt() attach = export_txt(ep) - - # Run analysis with timeout + rate-limit retries - cmd = [OPENCODE, "run", "--auto", "-f", attach, "--dir", BASE_DIR, prompt] - - # Rate limit retry waits (seconds) - rate_limit_waits = [10, 60, 180, 420, 600] - - for attempt, wait_time in enumerate(rate_limit_waits, 1): - try: - print(f" [{time.strftime('%H:%M:%S')}] running analysis (attempt {attempt}, timeout {ANALYSIS_TIMEOUT}s)...", flush=True) - proc = subprocess.run(cmd, capture_output=True, text=True, - encoding="utf-8", errors="replace", - cwd=BASE_DIR, timeout=ANALYSIS_TIMEOUT) - out = proc.stdout or "" - err = proc.stderr or "" - combined = (out + err).lower() - - if "rate limit" in combined or "429" in combined: - raise RuntimeError("provider rate limit exceeded") - - if not out.strip() and err: - raise RuntimeError(f"opencode produced no stdout: {err[-400:]}") - - # Parse result - from analyze import _repair - obj = json.loads(_repair(out), strict=False) - validate(obj) - - # Add metadata - 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"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) - obj["_source_transcript"] = os.path.basename(transcript_path(ep)) - - # Save analysis - from analyze import ANALYSIS_DIR as ANALYSIS_DIR2 - from download import slug as ep_slug - apath = os.path.join(ANALYSIS_DIR2, f"{ep_slug(ep)}.json") - with open(apath + ".tmp", "w", encoding="utf-8") as f: - json.dump(obj, f, ensure_ascii=False, indent=1) - os.replace(apath + ".tmp", apath) - - # Now run verification - print(f" [{time.strftime('%H:%M:%S')}] fact-checking...", flush=True) - from verify import verify_episode - verify_episode(ep) - - return (True, obj, None) - - except RuntimeError as e: - if "rate limit" in str(e).lower(): - if attempt < len(rate_limit_waits): - wait = rate_limit_waits[attempt] - print(f" [{time.strftime('%H:%M:%S')}] rate limited (attempt {attempt}), sleeping {wait}s...", flush=True) - time.sleep(wait) - continue - else: - return (False, None, f"Rate limit exceeded after {attempt} attempts") - else: - return (False, None, str(e)) - except Exception as e: - return (False, None, str(e)) - - return (False, None, "Max retries exceeded") + + def parse_obj(raw: str) -> dict: + obj = strip_to_json(raw) + validate(obj) + ensure_content_match(obj, attach) + return obj + + # Resilient runner handles model fallback + retries internally. + try: + print(f" [{time.strftime('%H:%M:%S')}] running analysis " + f"(timeout {ANALYZE_TIMEOUT}s)...", flush=True) + obj = llm_run(prompt, [attach], timeout=ANALYZE_TIMEOUT, + label="analyze", parse=parse_obj) + except (RateLimitError, ProviderError) as e: + return (False, None, str(e)) + + obj["title"] = obj.get("title") or ep["title"] + obj["pub_date"] = ep["pub_date"] + obj["_analyzed_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + obj["_source_transcript"] = os.path.basename(transcript_path(ep)) + obj["ep_num"] = ep["ep_num"] if ep["ep_num"] else obj.get("ep_num") + + # Save analysis + apath = os.path.join(ANALYSIS_DIR, f"{ep_slug(ep)}.json") + with open(apath + ".tmp", "w", encoding="utf-8") as f: + json.dump(obj, f, ensure_ascii=False, indent=1) + os.replace(apath + ".tmp", apath) + + # Now run verification + print(f" [{time.strftime('%H:%M:%S')}] fact-checking...", flush=True) + from verify import verify_episode + verified = verify_episode(ep) + + return (True, obj, None if verified is not None else "verification failed") def main() -> None: @@ -304,9 +278,14 @@ def main() -> None: success, result, error = run_analysis_with_timeout(ep) if success: done_a += 1 - print(f" [{time.strftime('%H:%M:%S')}] analyzed {label} (TI={result['transgression']['overall_ti']}, " - f"{len(result['talking_points'])} points, {len(result['notable_quotes'])} quotes)", flush=True) - print(f" [{time.strftime('%H:%M:%S')}] verified {label}", flush=True) + ti = result.get("transgression", {}).get("overall_ti", "?") + print(f" [{time.strftime('%H:%M:%S')}] analyzed {label} (TI={ti}, " + f"{len(result.get('talking_points', []))} points, " + f"{len(result.get('notable_quotes', []))} quotes)", flush=True) + if error: + print(f" [{time.strftime('%H:%M:%S')}] NOT verified {label}: {error}", flush=True) + else: + print(f" [{time.strftime('%H:%M:%S')}] verified {label}", flush=True) else: print(f" [{time.strftime('%H:%M:%S')}] ANALYSIS FAILED {label}: {error}", flush=True) except Exception as ex: diff --git a/verify.py b/verify.py index 5c85c21..6b7bf01 100644 --- a/verify.py +++ b/verify.py @@ -5,6 +5,8 @@ Spots "strange assumptions" like attributing a 2023 cohost seat to Maddox. Usage: python verify.py # verify ALL existing analyses python verify.py 361 373 # verify specific episodes + +Retries, model fallback and provider cooldown live in llm.py (shared). """ import json import os @@ -18,12 +20,14 @@ try: except Exception: pass +from llm import run as llm_run +from llm import RateLimitError, ProviderError +from llm import VERIFY_TIMEOUT from config import BASE_DIR from feed import load_episodes from download import slug as ep_slug from facts import DOSSIER_PATH, CAST_SHEET -from analyze import (ANALYSIS_DIR, OPENCODE, run_opencode, strip_to_json, - validate) +from analyze import (ANALYSIS_DIR, OPENCODE, strip_to_json, validate) from export_txt import txt_path VERIFY_PROMPT = """You are the fact-checking archivist for The Dick Show project. Three files are attached: @@ -77,72 +81,54 @@ def verify_episode(ep: dict, force: bool = False) -> dict | None: return None prompt = build_verify_prompt() - from analyze import RateLimitError - waits = [30, 120, 300, 600] - attempt = 0 - last_err = None - while attempt < len(waits): - attempt += 1 - try: - raw = run_opencode_multi(prompt, [attach_txt, DOSSIER_PATH, apath]) - verdict = strip_to_json(raw) - v = verdict.get("verdict") - if v not in ("pass", "fix"): - raise ValueError(f"bad verdict: {v!r}") - if v == "pass": - current.setdefault("_verification", { - "verdict": "pass", "checked_at": _now(), "issues": []}) - _save(apath, current) - print(f" verified OK (no issues)", flush=True) - return current - fixed = verdict.get("corrected_analysis") + + def parse_verdict(raw: str) -> dict: + v = strip_to_json(raw) + verdict = v.get("verdict") + if verdict not in ("pass", "fix"): + raise ValueError(f"bad verdict: {verdict!r}") + if verdict == "fix": + fixed = v.get("corrected_analysis") if not isinstance(fixed, dict): raise ValueError("verdict=fix but no corrected_analysis") validate(fixed) - issues = verdict.get("issues", []) - fixed["_verification"] = {"verdict": "fix", "checked_at": _now(), - "issues": issues} - fixed["ep_num"] = ep["ep_num"] if ep["ep_num"] else fixed.get("ep_num") - _save(apath, fixed) - print(f" FIXED {len(issues)} issue(s):", flush=True) - for i in issues[:6]: - print(f" - [{i.get('field')}] {str(i.get('issue'))[:90]}", - flush=True) - return fixed - except RateLimitError as e: - last_err = e - wait = waits[min(attempt, len(waits) - 1)] - print(f" verify rate-limited (attempt {attempt}); " - f"sleeping {wait}s ...", flush=True) - time.sleep(wait) - except Exception as e: # noqa: BLE001 - last_err = e - wait = waits[min(attempt - 1, len(waits) - 1)] - print(f" verify attempt {attempt} failed: {e}; " - f"retry in {wait}s", flush=True) - time.sleep(wait) - print(f" verification FAILED for {ep['title']}: {last_err}", flush=True) - return None + return v + + from datetime import datetime, timezone as _tz + try: + verdict = llm_run(prompt, [attach_txt, DOSSIER_PATH, apath], + timeout=VERIFY_TIMEOUT, label="verify", + parse=parse_verdict) + except (RateLimitError, ProviderError) as e: + print(f" verification NOT RUN for {ep['title']}: {e}", flush=True) + return None + + v = verdict.get("verdict") + if v == "pass": + current.setdefault("_verification", { + "verdict": "pass", + "checked_at": datetime.now(_tz.utc).isoformat(), "issues": []}) + _save(apath, current) + print(f" verified OK (no issues)", flush=True) + return current + fixed = verdict["corrected_analysis"] + issues = verdict.get("issues", []) + fixed["_verification"] = {"verdict": "fix", + "checked_at": datetime.now(_tz.utc).isoformat(), + "issues": issues} + fixed["ep_num"] = ep["ep_num"] if ep["ep_num"] else fixed.get("ep_num") + _save(apath, fixed) + print(f" FIXED {len(issues)} issue(s):", flush=True) + for i in issues[:6]: + print(f" - [{i.get('field')}] {str(i.get('issue'))[:90]}", + flush=True) + return fixed def run_opencode_multi(prompt: str, attachments: list[str]) -> str: - from analyze import OPENCODE_MODEL - cmd = [OPENCODE, "run", "--auto"] - for a in attachments: - cmd += ["-f", a] - cmd += ["--dir", BASE_DIR] - if OPENCODE_MODEL: - cmd += ["-m", OPENCODE_MODEL] - cmd.append(prompt) - import subprocess - proc = subprocess.run(cmd, capture_output=True, text=True, - encoding="utf-8", errors="replace", - cwd=BASE_DIR, timeout=900) - out = proc.stdout or "" - if not out.strip() and proc.stderr: - raise RuntimeError(f"opencode produced no stdout: " - f"{proc.stderr[-400:]}") - return out + # compat shim: raw output only, JSON parsed by the caller + return llm_run(prompt, attachments, timeout=VERIFY_TIMEOUT, + label="verify") def _now() -> str: @@ -174,4 +160,4 @@ def main() -> None: if __name__ == "__main__": - main() + main() \ No newline at end of file