YouTube -> transcript -> claim-extraction pipeline (youtube_claims/)
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>
This commit is contained in:
momentum devandClaude Opus 4.8 committed
1 parent
a7f63a5b11
commit
4a54b2adcc
9 files changed
+678
No files matched your search
@@ -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.
|
||||
"""
|
||||
@@ -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()]
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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 <playlist_id> --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()
|
||||
@@ -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
|
||||
);
|
||||
@@ -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
|
||||
@@ -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
|
||||
Reference in new issue
Block a user