From 4a54b2adccd7ad28b42aa8e7f272dd86033a9c26 Mon Sep 17 00:00:00 2001 From: momentum dev Date: Wed, 1 Jul 2026 15:00:09 -0400 Subject: [PATCH] YouTube -> transcript -> claim-extraction pipeline (youtube_claims/) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- youtube_claims/__init__.py | 7 ++ youtube_claims/config.py | 64 ++++++++++++ youtube_claims/db.py | 166 ++++++++++++++++++++++++++++++++ youtube_claims/detection.py | 70 ++++++++++++++ youtube_claims/extraction.py | 110 +++++++++++++++++++++ youtube_claims/run.py | 86 +++++++++++++++++ youtube_claims/schema.sql | 60 ++++++++++++ youtube_claims/transcription.py | 71 ++++++++++++++ youtube_claims/worker.py | 44 +++++++++ 9 files changed, 678 insertions(+) create mode 100644 youtube_claims/__init__.py create mode 100644 youtube_claims/config.py create mode 100644 youtube_claims/db.py create mode 100644 youtube_claims/detection.py create mode 100644 youtube_claims/extraction.py create mode 100644 youtube_claims/run.py create mode 100644 youtube_claims/schema.sql create mode 100644 youtube_claims/transcription.py create mode 100644 youtube_claims/worker.py diff --git a/youtube_claims/__init__.py b/youtube_claims/__init__.py new file mode 100644 index 0000000..b206086 --- /dev/null +++ b/youtube_claims/__init__.py @@ -0,0 +1,7 @@ +"""YouTube -> transcript -> claim-extraction pipeline. + +A DESCRIPTIVE ingestion layer (not a decision layer): it records WHAT was claimed in +monitored YouTube playlists, with provenance + a verbatim source quote, into Postgres for a +downstream reasoning model. It never judges whether a claim is correct or tradeable -- all +judgment lives downstream, in one place. See the build spec in docs / the handoff. +""" diff --git a/youtube_claims/config.py b/youtube_claims/config.py new file mode 100644 index 0000000..bce0f01 --- /dev/null +++ b/youtube_claims/config.py @@ -0,0 +1,64 @@ +"""Config — env-driven (§10 defaults). Reuses the app's DATABASE_URL and Ollama host; +secrets (YOUTUBE_API_KEY) come from .env only, never hard-coded.""" + +import os + +_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + +# load .env once (the same file the trading app uses; gitignored) +try: + from dotenv import load_dotenv + load_dotenv(os.path.join(_ROOT, ".env")) +except Exception: # noqa: BLE001 + pass + +# --- §10 defaults (env-overridable) --- +POLL_INTERVAL_MINUTES = int(os.getenv("YT_POLL_INTERVAL_MINUTES", "30")) +PLAYLIST_PAGE_SIZE = 50 +WORKER_CONCURRENCY = 1 # deliberate — §13 (yt-dlp politeness) +INTER_VIDEO_DELAY_SECONDS = int(os.getenv("YT_INTER_VIDEO_DELAY_SECONDS", "60")) +MAX_RETRIES = int(os.getenv("YT_MAX_RETRIES", "3")) +RETRY_BACKOFF_BASE_SEC = 30 # 30, 60, 120... +YT_DLP_FORMAT = "bestaudio" +YOUTUBE_API_QUOTA_DAILY = 10000 +TRANSCRIPT_DIR = os.getenv("YT_TRANSCRIPT_DIR", os.path.join(_ROOT, "data", "transcripts")) +PG_SCHEMA = os.getenv("YT_PG_SCHEMA", "transcripts") + +# Whisper — option A (CPU on this WSL box; no local GPU). Flip to large-v3/cuda/float16 +# via env if we later move the worker to the LAN GPU host (option B). +WHISPER_MODEL = os.getenv("YT_WHISPER_MODEL", "medium") +WHISPER_DEVICE = os.getenv("YT_WHISPER_DEVICE", "cpu") +WHISPER_COMPUTE_TYPE = os.getenv("YT_WHISPER_COMPUTE_TYPE", "int8") + +# Extraction LLM — local Ollama (free, on-LAN, private; transcripts never leave the network). +OLLAMA_MODEL = os.getenv("YT_EXTRACT_MODEL", "qwen2.5:7b-instruct") + + +def youtube_api_key() -> str: + k = os.getenv("YOUTUBE_API_KEY") + if not k: + raise RuntimeError("YOUTUBE_API_KEY not set — put it in .env") + return k + + +def database_url() -> str: + u = os.getenv("DATABASE_URL") + if not u: + raise RuntimeError("DATABASE_URL not set — the pipeline shares the app's Postgres") + return u + + +def ollama_host() -> str: + """Reuse the app's resolver so the pipeline self-heals across the WSL host-flip.""" + default = os.getenv("OLLAMA_HOST", "http://192.168.1.5:30068") + try: + from config import _resolve_ollama_host + return _resolve_ollama_host(default) + except Exception: # noqa: BLE001 + return default + + +def watchlist() -> list[str]: + """Tickers/names to bias Whisper's initial_prompt + focus extraction (§6, §12.4).""" + raw = os.getenv("YT_WATCHLIST", "") + return [w.strip() for w in raw.split(",") if w.strip()] diff --git a/youtube_claims/db.py b/youtube_claims/db.py new file mode 100644 index 0000000..bc36245 --- /dev/null +++ b/youtube_claims/db.py @@ -0,0 +1,166 @@ +"""Postgres layer — shares the app's database (DATABASE_URL), isolated in the `transcripts` +schema. psycopg2 (consistent with the trading app). Videos are claimed by an atomic +status-flip under FOR UPDATE SKIP LOCKED, so the long transcription never holds a row lock.""" + +import os +from contextlib import contextmanager +from datetime import date, datetime, timezone + +import psycopg2 +import psycopg2.extras + +from youtube_claims import config + + +def _dsn() -> str: + return config.database_url() + + +@contextmanager +def connect(): + conn = psycopg2.connect(_dsn()) + try: + with conn.cursor() as cur: + cur.execute(f'SET search_path TO {config.PG_SCHEMA}, public') + yield conn + finally: + conn.close() + + +def apply_schema() -> None: + ddl = open(os.path.join(os.path.dirname(__file__), "schema.sql")).read() + ddl = ddl.replace("{SCHEMA}", config.PG_SCHEMA) + with psycopg2.connect(_dsn()) as conn: + with conn.cursor() as cur: + cur.execute(ddl) + conn.commit() + + +# --- playlists --- +def upsert_playlist(playlist_id: str, label: str, content_type: str = "unknown") -> None: + with connect() as conn: + with conn.cursor() as cur: + cur.execute( + "INSERT INTO playlists (playlist_id, label, content_type) VALUES (%s,%s,%s) " + "ON CONFLICT (playlist_id) DO UPDATE SET label=EXCLUDED.label, " + "content_type=EXCLUDED.content_type", + (playlist_id, label, content_type)) + conn.commit() + + +def enabled_playlists() -> list[dict]: + with connect() as conn: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute("SELECT * FROM playlists WHERE enabled ORDER BY added_at") + return [dict(r) for r in cur.fetchall()] + + +# --- videos --- +def seen_video_ids() -> set[str]: + with connect() as conn: + with conn.cursor() as cur: + cur.execute("SELECT video_id FROM videos") + return {r[0] for r in cur.fetchall()} + + +def insert_pending_video(v: dict) -> bool: + """Insert a newly-detected video as pending. Returns False if it already existed.""" + pub = v.get("published_at") + lat = None + if pub: + lat = int((v["detected_at"] - pub).total_seconds()) + with connect() as conn: + with conn.cursor() as cur: + cur.execute( + "INSERT INTO videos (video_id, playlist_id, channel_id, channel_name, title, " + "content_type, published_at, detected_at, poll_latency_seconds, status) " + "VALUES (%(video_id)s,%(playlist_id)s,%(channel_id)s,%(channel_name)s,%(title)s," + "%(content_type)s,%(published_at)s,%(detected_at)s,%(lat)s,'pending') " + "ON CONFLICT (video_id) DO NOTHING", + {**v, "lat": lat}) + inserted = cur.rowcount == 1 + conn.commit() + return inserted + + +def claim_next_pending() -> dict | None: + """Atomically pick one pending video and flip it to `transcribing`. The lock is held + only for the flip (SKIP LOCKED prevents two workers grabbing the same row); the long + transcription runs AFTER the commit, holding no lock.""" + with connect() as conn: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute( + "SELECT * FROM videos WHERE status='pending' " + "ORDER BY published_at NULLS LAST, detected_at " + "FOR UPDATE SKIP LOCKED LIMIT 1") + row = cur.fetchone() + if row is None: + conn.commit() + return None + cur.execute("UPDATE videos SET status='transcribing' WHERE video_id=%s", + (row["video_id"],)) + conn.commit() + return dict(row) + + +def set_status(video_id: str, status: str, **fields) -> None: + cols = ", ".join(f"{k}=%s" for k in fields) + params = list(fields.values()) + [video_id] + sql = f"UPDATE videos SET status=%s{', ' + cols if cols else ''} WHERE video_id=%s" + with connect() as conn: + with conn.cursor() as cur: + cur.execute(sql, [status, *params]) + conn.commit() + + +def record_failure(video_id: str, stage: str, error: str) -> str: + """Bump retry_count and set last_error; return the resulting status + (`pending` to retry, or `failed` once retries are exhausted).""" + with connect() as conn: + with conn.cursor() as cur: + cur.execute("SELECT retry_count FROM videos WHERE video_id=%s", (video_id,)) + rc = (cur.fetchone() or [0])[0] + 1 + status = "pending" if rc <= config.MAX_RETRIES else "failed" + cur.execute( + "UPDATE videos SET status=%s, retry_count=%s, last_error=%s WHERE video_id=%s", + (status, rc, f"[{stage}] {error}"[:2000], video_id)) + conn.commit() + return status + + +def mark_done(video_id: str, transcript_source: str, whisper_model: str | None, + transcript_path: str) -> None: + set_status(video_id, "done", transcript_source=transcript_source, + whisper_model=whisper_model, transcript_path=transcript_path, + processed_at=datetime.now(timezone.utc)) + + +# --- claims --- +def insert_claims(video_id: str, claims: list[dict]) -> int: + if not claims: + return 0 + fields = ["asset_ticker", "asset_name", "asset_class", "direction", "claim_text", + "verbatim_quote", "timestamp_start", "timestamp_end", "stated_rationale", + "stated_horizon", "extraction_confidence"] + with connect() as conn: + with conn.cursor() as cur: + for c in claims: + cur.execute( + f"INSERT INTO claims (video_id, {', '.join(fields)}) " + f"VALUES (%s, {', '.join(['%s'] * len(fields))})", + [video_id, *[c.get(f) for f in fields]]) + conn.commit() + return len(claims) + + +# --- quota (§5) --- +def add_quota(units: int) -> int: + with connect() as conn: + with conn.cursor() as cur: + cur.execute( + "INSERT INTO api_quota (day, units_used) VALUES (%s,%s) " + "ON CONFLICT (day) DO UPDATE SET units_used = api_quota.units_used + EXCLUDED.units_used " + "RETURNING units_used", (date.today(), units)) + used = cur.fetchone()[0] + conn.commit() + return used diff --git a/youtube_claims/detection.py b/youtube_claims/detection.py new file mode 100644 index 0000000..19b7fbf --- /dev/null +++ b/youtube_claims/detection.py @@ -0,0 +1,70 @@ +"""Detection layer (§5): poll each enabled playlist, dedupe by video_id, insert `pending` +rows. Uses the YouTube Data API `playlistItems.list` over urllib (no extra SDK dep).""" + +import json +import urllib.parse +import urllib.request +from datetime import datetime, timezone + +from youtube_claims import config, db + + +def _get(url: str) -> dict: + with urllib.request.urlopen(url, timeout=20) as r: + return json.load(r) + + +def _iso(s: str | None): + return datetime.fromisoformat(s.replace("Z", "+00:00")) if s else None + + +def fetch_playlist_items(playlist_id: str, key: str) -> tuple[list[dict], int]: + """All items of a playlist, paginated. Returns (items, api_units_used ~= #pages).""" + items, token, units = [], None, 0 + while True: + q = {"part": "snippet,contentDetails", "maxResults": config.PLAYLIST_PAGE_SIZE, + "playlistId": playlist_id, "key": key} + if token: + q["pageToken"] = token + data = _get("https://www.googleapis.com/youtube/v3/playlistItems?" + urllib.parse.urlencode(q)) + units += 1 + items += data.get("items", []) + token = data.get("nextPageToken") + if not token: + break + return items, units + + +def poll_once() -> tuple[int, int]: + """One detection pass over all enabled playlists. Returns (#new videos, quota used today).""" + key = config.youtube_api_key() + seen = db.seen_video_ids() + new_count, units = 0, 0 + for pl in db.enabled_playlists(): + try: + items, u = fetch_playlist_items(pl["playlist_id"], key) + except Exception as exc: # noqa: BLE001 — one bad playlist can't stop the poll + print(f"[detect] playlist {pl['playlist_id']} poll failed: {exc}") + continue + units += u + now = datetime.now(timezone.utc) + for it in items: + sn = it["snippet"] + vid = it["contentDetails"]["videoId"] + if vid in seen: + continue + v = dict( + video_id=vid, playlist_id=pl["playlist_id"], + channel_id=sn.get("videoOwnerChannelId") or sn.get("channelId"), + channel_name=sn.get("videoOwnerChannelTitle") or sn.get("channelTitle"), + title=sn.get("title"), + content_type=pl.get("content_type", "unknown"), + published_at=_iso(it["contentDetails"].get("videoPublishedAt") or sn.get("publishedAt")), + detected_at=now) + if db.insert_pending_video(v): + new_count += 1 + seen.add(vid) + used = db.add_quota(units) + if used > config.YOUTUBE_API_QUOTA_DAILY * 0.8: + print(f"[detect] WARNING: {used}/{config.YOUTUBE_API_QUOTA_DAILY} daily API units used") + return new_count, used diff --git a/youtube_claims/extraction.py b/youtube_claims/extraction.py new file mode 100644 index 0000000..e8523a6 --- /dev/null +++ b/youtube_claims/extraction.py @@ -0,0 +1,110 @@ +"""Extraction (§8): local Ollama turns a transcript into DESCRIPTIVE claim rows. Never +evaluative — it records WHAT was claimed + an EXACT verbatim quote; it does not judge whether +the claim is correct or tradeable (all judgment lives downstream). A claim without a +verbatim_quote is dropped: that quote is the chain of custody the reasoning model verifies +against ASR error.""" + +import json +import urllib.request + +from youtube_claims import config + +_PROMPT = """You are a precise information extractor. From the TRANSCRIPT below, extract every +distinct market/trading CLAIM about a specific asset (stock, crypto, ETF, index, commodity, fx). + +STRICT RULES: +- DESCRIPTIVE ONLY. Record what the speaker CLAIMED. Never judge if it is correct, good, or + tradeable. `direction` describes the claim (bullish/bearish/neutral/mixed), not a recommendation. +- Every claim MUST include `verbatim_quote`: text copied EXACTLY from the transcript (word-for-word) + that supports the claim. If you cannot copy an exact supporting quote, DO NOT emit the claim. +- `extraction_confidence` (0-1) is how sure you are you PARSED it correctly — NOT that it is true. +- Prefer claims about these watchlist assets if present, but include others too: {watchlist} +- If the transcript contains no asset-specific claims, return {{"claims": []}}. + +Return ONLY JSON: {{"claims": [ {{ + "asset_ticker": "AAPL or null", "asset_name": "as spoken or null", + "asset_class": "equity|crypto|etf|index|commodity|fx|other", + "direction": "bullish|bearish|neutral|mixed", + "claim_text": "short paraphrase of what was asserted", + "verbatim_quote": "EXACT transcript text", + "stated_rationale": "reason given or null", "stated_horizon": "e.g. next quarter or null", + "extraction_confidence": 0.0 +}} ] }} + +TRANSCRIPT: +{transcript} +""" + + +def _ollama_generate(prompt: str, model: str, host: str) -> str: + payload = {"model": model, "prompt": prompt, "stream": False, "format": "json", + "options": {"temperature": 0.1, "num_predict": 2048}} + req = urllib.request.Request(host.rstrip("/") + "/api/generate", + data=json.dumps(payload).encode(), + headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=240) as r: + return json.load(r).get("response", "") + + +def _chunks(segments: list[dict], max_chars: int = 8000): + """Group whisper segments into ~max_chars windows, carrying their time span so extracted + claims can be time-located and their quotes checked against the exact window text.""" + buf, start, n = [], None, 0 + for seg in segments: + t = (seg.get("text") or "").strip() + if not t: + continue + if start is None: + start = seg.get("start") + buf.append((seg, t)) + n += len(t) + if n >= max_chars: + yield buf, start, seg.get("end") + buf, start, n = [], None, 0 + if buf: + yield buf, start, buf[-1][0].get("end") + + +def _normalize(c: dict, win_start, win_end) -> dict | None: + vq = (c.get("verbatim_quote") or "").strip() + if not vq: + return None # mandatory — drop unsupported claims + tkr = (c.get("asset_ticker") or "").strip().upper() or None + return { + "asset_ticker": tkr, + "asset_name": (c.get("asset_name") or None), + "asset_class": (c.get("asset_class") or None), + "direction": (c.get("direction") or None), + "claim_text": (c.get("claim_text") or None), + "verbatim_quote": vq, + "timestamp_start": win_start, + "timestamp_end": win_end, + "stated_rationale": (c.get("stated_rationale") or None), + "stated_horizon": (c.get("stated_horizon") or None), + "extraction_confidence": c.get("extraction_confidence"), + } + + +def extract_claims(segments: list[dict], watchlist: list[str] | None = None, + *, model: str | None = None, host: str | None = None) -> list[dict]: + """Extract descriptive claims from whisper segments. Chunks long transcripts so nothing is + dropped, and attaches each claim's window time span. Pure w.r.t. the DB (caller inserts).""" + model = model or config.OLLAMA_MODEL + host = host or config.ollama_host() + wl = ", ".join(watchlist if watchlist is not None else config.watchlist()) or "(none specified)" + out: list[dict] = [] + for win, w_start, w_end in _chunks(segments): + text = "\n".join(t for _, t in win) + try: + raw = _ollama_generate(_PROMPT.format(watchlist=wl, transcript=text), model, host) + data = json.loads(raw) + except Exception: # noqa: BLE001 — a bad chunk shouldn't kill the whole video + continue + for c in (data.get("claims") or []): + vq = (c.get("verbatim_quote") or "").strip() + # keep only claims whose quote is actually present in THIS window (anti-hallucination) + if vq and vq[:60].lower() in text.lower(): + nc = _normalize(c, w_start, w_end) + if nc: + out.append(nc) + return out diff --git a/youtube_claims/run.py b/youtube_claims/run.py new file mode 100644 index 0000000..5f577eb --- /dev/null +++ b/youtube_claims/run.py @@ -0,0 +1,86 @@ +"""Orchestration (§2): one long-lived process. APScheduler fires the 30-min poll; a worker +loop drains pending videos alongside it. On restart nothing is lost — pending rows resume from +Postgres. The same CLI exposes each piece for setup + manual runs. + + python -m youtube_claims.run apply-schema + python -m youtube_claims.run add-playlist --label "..." --content-type analysis + python -m youtube_claims.run poll-once # detection pass only + python -m youtube_claims.run drain --max 1 # process N pending videos then stop + python -m youtube_claims.run run # the full service (poll + worker) +""" + +import argparse +import threading +import time + +from youtube_claims import config, db, detection, worker + + +def _worker_loop(stop: threading.Event) -> None: + while not stop.is_set(): + try: + if worker.drain() == 0: + stop.wait(30) # nothing pending — idle until the next poll fills it + except Exception as exc: # noqa: BLE001 — the loop must survive one bad video + print(f"[worker-loop] {exc}") + stop.wait(30) + + +def run_service() -> None: + from apscheduler.schedulers.background import BackgroundScheduler + db.apply_schema() + + def _poll(): + try: + new, used = detection.poll_once() + print(f"[poll] {new} new videos, {used} api units today") + except Exception as exc: # noqa: BLE001 + print(f"[poll] failed: {exc}") + + sched = BackgroundScheduler() + sched.add_job(_poll, "interval", minutes=config.POLL_INTERVAL_MINUTES) + sched.start() + _poll() # once at boot + stop = threading.Event() + threading.Thread(target=_worker_loop, args=(stop,), daemon=True).start() + print(f"[service] running: poll every {config.POLL_INTERVAL_MINUTES}m, worker concurrency 1") + try: + while True: + time.sleep(3600) + except (KeyboardInterrupt, SystemExit): + stop.set() + sched.shutdown() + + +def main() -> None: + ap = argparse.ArgumentParser(description="YouTube -> transcript -> claims pipeline") + sub = ap.add_subparsers(dest="cmd", required=True) + sub.add_parser("apply-schema") + a = sub.add_parser("add-playlist") + a.add_argument("playlist_id") + a.add_argument("--label", default="") + a.add_argument("--content-type", default="unknown") + sub.add_parser("poll-once") + d = sub.add_parser("drain") + d.add_argument("--max", type=int) + sub.add_parser("run") + args = ap.parse_args() + + if args.cmd == "apply-schema": + db.apply_schema() + print(f"schema '{config.PG_SCHEMA}' applied") + elif args.cmd == "add-playlist": + db.apply_schema() + db.upsert_playlist(args.playlist_id, args.label, args.content_type) + print(f"playlist {args.playlist_id} added ({args.label!r}, {args.content_type})") + elif args.cmd == "poll-once": + new, used = detection.poll_once() + print(f"{new} new videos inserted, {used} api units used today") + elif args.cmd == "drain": + print("processed", worker.drain(max_videos=args.max)) + elif args.cmd == "run": + run_service() + + +if __name__ == "__main__": + main() diff --git a/youtube_claims/schema.sql b/youtube_claims/schema.sql new file mode 100644 index 0000000..c028364 --- /dev/null +++ b/youtube_claims/schema.sql @@ -0,0 +1,60 @@ +-- YouTube claim-extraction pipeline schema (§7-8). Isolated in its own schema so it never +-- collides with the trading tables. {SCHEMA} is substituted at apply time (default: transcripts). + +CREATE SCHEMA IF NOT EXISTS {SCHEMA}; + +-- config: which playlists to monitor +CREATE TABLE IF NOT EXISTS {SCHEMA}.playlists ( + playlist_id TEXT PRIMARY KEY, + label TEXT, + content_type TEXT DEFAULT 'unknown', -- analysis / news / opinion / promo / unknown + enabled BOOLEAN NOT NULL DEFAULT TRUE, + added_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- videos: one row per detected upload; dedupe key is video_id (§5) +CREATE TABLE IF NOT EXISTS {SCHEMA}.videos ( + video_id TEXT PRIMARY KEY, + playlist_id TEXT REFERENCES {SCHEMA}.playlists(playlist_id), + channel_id TEXT, + channel_name TEXT, + title TEXT, + content_type TEXT DEFAULT 'unknown', + published_at TIMESTAMPTZ, + detected_at TIMESTAMPTZ NOT NULL DEFAULT now(), + poll_latency_seconds INT, + status TEXT NOT NULL DEFAULT 'pending', -- pending|transcribing|extracting|done|failed + transcript_source TEXT, -- whisper|captions|NULL + whisper_model TEXT, + transcript_path TEXT, + processed_at TIMESTAMPTZ, + retry_count INT NOT NULL DEFAULT 0, + last_error TEXT +); +CREATE INDEX IF NOT EXISTS videos_status_idx ON {SCHEMA}.videos(status); + +-- claims: the core deliverable (§8) — DESCRIPTIVE only, verbatim_quote mandatory +CREATE TABLE IF NOT EXISTS {SCHEMA}.claims ( + claim_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + video_id TEXT NOT NULL REFERENCES {SCHEMA}.videos(video_id), + asset_ticker TEXT, + asset_name TEXT, + asset_class TEXT, -- equity|crypto|etf|index|commodity|fx|other + direction TEXT, -- bullish|bearish|neutral|mixed (of the CLAIM, not advice) + claim_text TEXT, + verbatim_quote TEXT NOT NULL, -- mandatory chain of custody back to source + timestamp_start NUMERIC, + timestamp_end NUMERIC, + stated_rationale TEXT, + stated_horizon TEXT, + extraction_confidence NUMERIC, -- 0-1 PARSE confidence, NOT truth confidence + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +CREATE INDEX IF NOT EXISTS claims_video_idx ON {SCHEMA}.claims(video_id); +CREATE INDEX IF NOT EXISTS claims_ticker_idx ON {SCHEMA}.claims(asset_ticker); + +-- daily YouTube API quota counter (§5) +CREATE TABLE IF NOT EXISTS {SCHEMA}.api_quota ( + day DATE PRIMARY KEY, + units_used INT NOT NULL DEFAULT 0 +); diff --git a/youtube_claims/transcription.py b/youtube_claims/transcription.py new file mode 100644 index 0000000..9b4cab0 --- /dev/null +++ b/youtube_claims/transcription.py @@ -0,0 +1,71 @@ +"""Transcription (§6): yt-dlp bestaudio -> faster-whisper (initial_prompt seeded with the +watchlist) with a youtube-transcript-api captions FALLBACK. Whisper is PRIMARY because +auto-captions mangle the load-bearing tokens (tickers/numbers/prices); captions are a +last resort so a blocked/failed download still yields something. Writes transcript JSON.""" + +import json +import os +import tempfile + +from youtube_claims import config + +_model = None + + +def _whisper(): + global _model + if _model is None: + from faster_whisper import WhisperModel + _model = WhisperModel(config.WHISPER_MODEL, device=config.WHISPER_DEVICE, + compute_type=config.WHISPER_COMPUTE_TYPE) + return _model + + +def _download_audio(video_id: str, out_dir: str) -> str | None: + import yt_dlp + tmpl = os.path.join(out_dir, f"{video_id}.%(ext)s") + opts = {"format": config.YT_DLP_FORMAT, "outtmpl": tmpl, "quiet": True, "no_warnings": True, + "postprocessors": [{"key": "FFmpegExtractAudio", "preferredcodec": "wav"}]} + with yt_dlp.YoutubeDL(opts) as ydl: + ydl.download([f"https://www.youtube.com/watch?v={video_id}"]) + wav = os.path.join(out_dir, f"{video_id}.wav") + return wav if os.path.exists(wav) else None + + +def _transcribe_whisper(wav: str, watchlist: list[str]) -> list[dict]: + prompt = ("Tickers and companies: " + ", ".join(watchlist)) if watchlist else None + segments, _info = _whisper().transcribe(wav, initial_prompt=prompt, vad_filter=True) + return [{"start": round(s.start, 2), "end": round(s.end, 2), "text": s.text} for s in segments] + + +def _captions_fallback(video_id: str) -> list[dict]: + from youtube_transcript_api import YouTubeTranscriptApi + try: # newer + older API shapes + rows = YouTubeTranscriptApi().fetch(video_id).to_raw_data() + except Exception: # noqa: BLE001 + rows = YouTubeTranscriptApi.get_transcript(video_id) + return [{"start": round(x["start"], 2), "end": round(x["start"] + x.get("duration", 0), 2), + "text": x["text"]} for x in rows] + + +def transcribe(video_id: str, watchlist: list[str] | None = None) -> tuple[list[dict], str, str | None, str]: + """Returns (segments, source, whisper_model_or_None, transcript_path). Whisper primary, + captions fallback. Raises only if BOTH paths fail (worker handles the retry).""" + watchlist = watchlist if watchlist is not None else config.watchlist() + os.makedirs(config.TRANSCRIPT_DIR, exist_ok=True) + segs, source, model = None, None, None + try: + with tempfile.TemporaryDirectory() as td: + wav = _download_audio(video_id, td) + if wav: + segs = _transcribe_whisper(wav, watchlist) + source, model = "whisper", config.WHISPER_MODEL + except Exception as exc: # noqa: BLE001 — fall through to captions + print(f"[transcribe] whisper path failed for {video_id}: {exc}") + if not segs: + segs = _captions_fallback(video_id) # may raise -> worker retry/backoff + source = "captions" + path = os.path.join(config.TRANSCRIPT_DIR, f"{video_id}.json") + with open(path, "w") as f: + json.dump({"video_id": video_id, "source": source, "whisper_model": model, "segments": segs}, f) + return segs, source, model, path diff --git a/youtube_claims/worker.py b/youtube_claims/worker.py new file mode 100644 index 0000000..63cddc5 --- /dev/null +++ b/youtube_claims/worker.py @@ -0,0 +1,44 @@ +"""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