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>
71 lines
2.7 KiB
Python
71 lines
2.7 KiB
Python
"""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
|