n3wk
311 lines
12 KiB
Python
311 lines
12 KiB
Python
"""Full-backlog driver: Ep 1 -> latest. One pass, fully resumable, verbose.
|
|
|
|
Per episode:
|
|
1. skip if transcript + verified analysis already exist
|
|
2. download mp3 (skipped if present)
|
|
3. transcribe (re-does partial transcripts, e.g. old smoke tests)
|
|
4. analyze via opencode (fresh session, WITH HARD TIMEOUT) + dossier fact-check
|
|
5. every 10 episodes: rebuild FTS index + master JSON
|
|
|
|
Run: python pipeline_all.py [--eps 84 87 89 ...] [--start-ep N] [--dry-run]
|
|
Log: stdout (use with redirection)
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import traceback
|
|
import subprocess
|
|
|
|
try:
|
|
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
|
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
|
|
except Exception:
|
|
pass
|
|
|
|
from config import TRANSCRIPT_DIR, KEEP_AUDIO
|
|
from feed import load_episodes
|
|
from download import audio_path, slug as ep_slug, download_episode
|
|
from preprocess import prepare_audio, prepared_path
|
|
from transcribe import transcript_path
|
|
from lexicon import build_lexicon, tags_for_text
|
|
from export_txt import export_txt
|
|
|
|
# seconds between opencode calls - provider rate limits are real
|
|
ANALYSIS_PAUSE = int(os.environ.get("DS_ANALYZE_PAUSE", "45"))
|
|
# hard timeout for entire analysis (seconds) - opencode can hang
|
|
ANALYSIS_TIMEOUT = int(os.environ.get("DS_ANALYSIS_TIMEOUT", "600")) # 10 minutes
|
|
|
|
# memory guard threshold (bytes) - pause if available RAM drops below this
|
|
MIN_FREE_RAM = int(os.environ.get("DS_MIN_FREE_RAM", str(1024 * 1024 * 1024))) # 1 GB
|
|
|
|
|
|
def has_partial(transcript_file: str) -> bool:
|
|
try:
|
|
with open(transcript_file, encoding="utf-8") as f:
|
|
return json.load(f).get("partial_minutes") is not None
|
|
except (OSError, json.JSONDecodeError):
|
|
return False
|
|
|
|
|
|
def check_memory() -> bool:
|
|
"""Check if we have enough free RAM. Return True if OK, False if low."""
|
|
try:
|
|
import psutil
|
|
avail = psutil.virtual_memory().available
|
|
if avail < MIN_FREE_RAM:
|
|
return False
|
|
except ImportError:
|
|
pass # psutil not installed, skip check
|
|
except Exception:
|
|
pass
|
|
return True
|
|
|
|
|
|
def wait_for_memory() -> None:
|
|
"""Block until enough RAM is available."""
|
|
print(f" [MEM] Low memory (< {MIN_FREE_RAM/1e9:.1f} GB free), waiting...", flush=True)
|
|
while not check_memory():
|
|
time.sleep(30)
|
|
print(f" [MEM] Memory recovered, continuing...", flush=True)
|
|
|
|
|
|
def do_transcribe(model, ep: dict) -> None:
|
|
from transcribe import transcribe_episode
|
|
from preprocess import load_pcm
|
|
src = audio_path(ep)
|
|
if not os.path.exists(src):
|
|
print(f" [{time.strftime('%H:%M:%S')}] downloading {ep['audio_url'][-40:]} ...", flush=True)
|
|
download_episode(ep)
|
|
prep = prepared_path(ep)
|
|
ok = False
|
|
t0 = time.time()
|
|
try:
|
|
# Check memory before loading episode
|
|
wait_for_memory()
|
|
print(f" [{time.strftime('%H:%M:%S')}] transcribing Ep {ep['ep_num']} ({ep['title'][:50]})", flush=True)
|
|
result = transcribe_episode(model, ep, None)
|
|
outp = transcript_path(ep)
|
|
with open(outp + ".tmp", "w", encoding="utf-8") as f:
|
|
json.dump(result, f, ensure_ascii=False)
|
|
os.replace(outp + ".tmp", outp)
|
|
dur = result["duration_sec"] or 1
|
|
rate = dur / max(time.time() - t0, 0.001)
|
|
lex = build_lexicon()
|
|
tags = set()
|
|
for seg in result["segments"]:
|
|
tags.update(tags_for_text(seg["text"], lex))
|
|
print(f" [{time.strftime('%H:%M:%S')}] TRANSCRIBED: {len(result['segments'])} segments, "
|
|
f"{dur/60:.0f} min, {rate:.1f}x realtime", flush=True)
|
|
print(f" TAGS: {' '.join(sorted('#' + t for t in tags)) or '(none)'}", flush=True)
|
|
ok = True
|
|
finally:
|
|
if os.path.exists(prep):
|
|
try:
|
|
os.remove(prep)
|
|
except OSError:
|
|
pass
|
|
if not KEEP_AUDIO and ok and os.path.exists(src):
|
|
try:
|
|
os.remove(src)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
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)
|
|
"""
|
|
from analyze import ANALYSIS_DIR, OPENCODE, build_prompt, strip_to_json, validate
|
|
from verify import verify_episode
|
|
from download import slug as ep_slug
|
|
from export_txt import txt_path
|
|
from config import BASE_DIR
|
|
|
|
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 main() -> None:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--eps", nargs="*", type=int, default=None,
|
|
help="Specific episode numbers to process (default: all from start-ep)")
|
|
ap.add_argument("--start-ep", type=int, default=0)
|
|
ap.add_argument("--dry-run", action="store_true")
|
|
args = ap.parse_args()
|
|
|
|
from faster_whisper import WhisperModel
|
|
from config import WHISPER_MODEL, DEVICE, COMPUTE_TYPE
|
|
from analyze import analyze_episode, OPENCODE
|
|
from verify import verify_episode
|
|
|
|
eps = load_episodes()
|
|
if args.eps:
|
|
eps = [e for e in eps if e["ep_num"] in args.eps]
|
|
else:
|
|
eps = [e for e in eps if (e["ep_num"] or 0) >= args.start_ep]
|
|
|
|
print(f"[{time.strftime('%H:%M:%S')}] pipeline over {len(eps)} episodes "
|
|
f"(model={WHISPER_MODEL}/{DEVICE}, analyzer={os.path.basename(OPENCODE)})", flush=True)
|
|
|
|
if args.dry_run:
|
|
for e in eps[:20]:
|
|
tp = transcript_path(e)
|
|
t_exists = os.path.exists(tp) and not has_partial(tp)
|
|
a_exists = os.path.exists(os.path.join(
|
|
os.path.dirname(TRANSCRIPT_DIR), "analysis", f"{e['ep_num']:04d}.json")) if e["ep_num"] else False
|
|
print(f" ep {e['ep_num']}: transcript={'Y' if t_exists else 'N'} analysis={'Y' if a_exists else 'N'}", flush=True)
|
|
return
|
|
|
|
model = None
|
|
done_t = done_a = 0
|
|
t_start = time.time()
|
|
|
|
for i, ep in enumerate(eps, 1):
|
|
num = ep["ep_num"] if ep["ep_num"] else ep["title"][:30]
|
|
tp = transcript_path(ep)
|
|
|
|
need_transcript = (not os.path.exists(tp)) or has_partial(tp)
|
|
if need_transcript and model is None:
|
|
print(f"[{time.strftime('%H:%M:%S')}] loading whisper model '{WHISPER_MODEL}'...", flush=True)
|
|
model = WhisperModel(WHISPER_MODEL, device=DEVICE, compute_type=COMPUTE_TYPE)
|
|
|
|
if has_partial(tp):
|
|
print(f"[{time.strftime('%H:%M:%S')}] [{i}/{len(eps)}] Ep {num}: redo partial transcript", flush=True)
|
|
os.remove(tp)
|
|
elif need_transcript:
|
|
print(f"[{time.strftime('%H:%M:%S')}] [{i}/{len(eps)}] Ep {num}: transcribing", flush=True)
|
|
|
|
if need_transcript:
|
|
try:
|
|
do_transcribe(model, ep)
|
|
done_t += 1
|
|
# keep the searchable index fresh-ish
|
|
if done_t % 10 == 0:
|
|
import index as index_mod
|
|
print(f" [{time.strftime('%H:%M:%S')}] rebuilding search index...", flush=True)
|
|
index_mod.index_transcripts()
|
|
except Exception as ex:
|
|
print(f" [{time.strftime('%H:%M:%S')}] TRANSCRIBE FAILED, skipping episode: {ex}", flush=True)
|
|
traceback.print_exc()
|
|
continue
|
|
|
|
# analysis + fact-check (with hard timeout)
|
|
label = f"Ep {num}"
|
|
print(f" [{time.strftime('%H:%M:%S')}] analyzing {label} ...", flush=True)
|
|
try:
|
|
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)
|
|
else:
|
|
print(f" [{time.strftime('%H:%M:%S')}] ANALYSIS FAILED {label}: {error}", flush=True)
|
|
except Exception as ex:
|
|
print(f" [{time.strftime('%H:%M:%S')}] ANALYZE FAILED {label}: {ex}", flush=True)
|
|
traceback.print_exc()
|
|
|
|
# pacing: keep the provider happy
|
|
print(f" [{time.strftime('%H:%M:%S')}] pacing pause {ANALYSIS_PAUSE}s ...", flush=True)
|
|
time.sleep(ANALYSIS_PAUSE)
|
|
|
|
if i % 10 == 0:
|
|
print(f" [{time.strftime('%H:%M:%S')}] rebuilding search index (periodic)...", flush=True)
|
|
import index as index_mod
|
|
index_mod.index_transcripts()
|
|
from aggregate import aggregate as _agg
|
|
_agg()
|
|
hrs = (time.time() - t_start) / 3600
|
|
rate_eps = i / hrs
|
|
remaining = (len(eps) - i) / max(rate_eps, 0.01)
|
|
print(f"[{time.strftime('%H:%M:%S')}] == progress: {i}/{len(eps)} eps "
|
|
f"({rate_eps:.1f} eps/hr, ~{remaining:.0f} hr left) ==", flush=True)
|
|
|
|
print(f"[{time.strftime('%H:%M:%S')}] final index rebuild...", flush=True)
|
|
import index as index_mod
|
|
index_mod.index_transcripts()
|
|
from aggregate import aggregate as _agg
|
|
_agg()
|
|
print(f"[{time.strftime('%H:%M:%S')}] PIPELINE COMPLETE: transcribed={done_t} analyzed={done_a}", flush=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |