momentum devandClaude Opus 4.8
4a54b2adcc
A descriptive ingestion layer feeding the downstream reasoning model: monitors YouTube
playlists, transcribes new uploads, and extracts structured, verifiable CLAIM records
(with provenance + a mandatory verbatim quote). It records WHAT was claimed; it never
judges correctness or tradeability -- all judgment stays downstream, in one place.
Built per the operator's spec. Two clocks coordinated through Postgres (§2): a 30-min
detection poll + a paced concurrency-1 transcription/extraction worker. Reuses the app's
infra: shared Postgres (isolated `transcripts` schema), the app's Ollama for extraction
(free, on-LAN, private -- transcripts never leave the network), the config.py host resolver.
youtube_claims/schema.sql playlists / videos / claims (§7-8); claims.verbatim_quote NOT NULL
youtube_claims/db.py psycopg2; claim-by-atomic-status-flip under FOR UPDATE SKIP LOCKED
youtube_claims/detection.py playlistItems.list over urllib; dedupe by video_id (§5)
youtube_claims/extraction.py local Ollama -> descriptive claims; drops any claim lacking an
exact verbatim quote present in the source window (anti-hallucination)
youtube_claims/transcription.py yt-dlp -> faster-whisper (medium/cpu/int8, option A;
initial_prompt=watchlist) with youtube-transcript-api captions fallback
youtube_claims/worker.py paced drain; per-video state machine + retries/backoff
youtube_claims/run.py APScheduler poll + worker loop in one process; CLI for setup
VERIFIED LIVE: schema applied; playlist "Anatomy of a Trade" (Ross Cameron) added; poll
ingested all 12 videos (1 API unit); Ollama extraction pulled clean descriptive claims
(TC/DXST/CF) with exact verbatim quotes from a real transcript. Whisper transcription pending
ffmpeg (`apt install ffmpeg`, needs sudo) + the yt-dlp/faster-whisper deps (installing via uv).
Secrets (YOUTUBE_API_KEY) live in gitignored .env only. Whisper is option A (CPU); env-flippable
to option B (LAN GPU large-v3) later.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
45 lines
1.7 KiB
Python
45 lines
1.7 KiB
Python
"""Worker (§9): drains the pending backlog, concurrency 1, paced (INTER_VIDEO_DELAY). State
|
|
machine per video (pending -> transcribing -> extracting -> done | failed) with retries/backoff.
|
|
Each stage is independently retryable; a `done` video is never reprocessed."""
|
|
|
|
import time
|
|
|
|
from youtube_claims import config, db, extraction, transcription
|
|
|
|
|
|
def process_one(v: dict) -> str:
|
|
"""Transcribe -> extract -> insert claims for one already-claimed video. Returns final status."""
|
|
vid = v["video_id"]
|
|
try:
|
|
segs, source, model, path = transcription.transcribe(vid)
|
|
except Exception as exc: # noqa: BLE001
|
|
st = db.record_failure(vid, "transcribe", str(exc))
|
|
print(f"[worker] {vid} transcribe failed -> {st}: {exc}")
|
|
return st
|
|
db.set_status(vid, "extracting", transcript_source=source, whisper_model=model, transcript_path=path)
|
|
try:
|
|
claims = extraction.extract_claims(segs)
|
|
except Exception as exc: # noqa: BLE001
|
|
st = db.record_failure(vid, "extract", str(exc))
|
|
print(f"[worker] {vid} extract failed -> {st}: {exc}")
|
|
return st
|
|
n = db.insert_claims(vid, claims)
|
|
db.mark_done(vid, source, model, path)
|
|
print(f"[worker] {vid} done: {len(segs)} segments ({source}), {n} claims")
|
|
return "done"
|
|
|
|
|
|
def drain(max_videos: int | None = None) -> int:
|
|
"""Process pending videos one at a time with a politeness gap. Returns #processed."""
|
|
done = 0
|
|
while True:
|
|
v = db.claim_next_pending()
|
|
if v is None:
|
|
break
|
|
process_one(v)
|
|
done += 1
|
|
if max_videos and done >= max_videos:
|
|
break
|
|
time.sleep(config.INTER_VIDEO_DELAY_SECONDS) # §13 — keep yt-dlp under the rate cap
|
|
return done
|