momentum devandClaude Fable 5
92488b72bc
One playlist per add-playlist CLI call made adding a second playlist a chore. Now YT_PLAYLIST_URLS takes a comma-separated list of playlist URLs (list= param extracted) or bare ids; the transcripts service seeds the playlists table from it at boot (insert-if-missing — hand-set labels are never clobbered), and `python -m youtube_claims.run sync-playlists` does the same on demand. Documented in both compose files; 3 tests. Also: queue the two executor-surfaced defects (OrderFilledEvent side, /api/attribution env hardcode) + the flaky nightly regime build-features step in the plans index; regime_features 07-16 backfilled supervised. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
98 lines
3.7 KiB
Python
98 lines
3.7 KiB
Python
"""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
|
|
import shutil
|
|
|
|
_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
# yt-dlp's audio postprocessor needs the ffmpeg BINARY. The systemd user service runs with a
|
|
# minimal PATH that omits /usr/bin, so pass the location explicitly instead of relying on PATH.
|
|
_ff = shutil.which("ffmpeg")
|
|
FFMPEG_LOCATION = os.getenv("YT_FFMPEG_LOCATION") or (os.path.dirname(_ff) if _ff else "/usr/bin")
|
|
|
|
# load .env once (the same file the trading app uses; gitignored)
|
|
try:
|
|
from dotenv import load_dotenv
|
|
load_dotenv(os.path.join(_ROOT, ".env"))
|
|
try:
|
|
from runtime.env_overrides import load as _load_overrides
|
|
_load_overrides()
|
|
except Exception:
|
|
pass
|
|
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()]
|
|
|
|
|
|
def playlist_urls() -> list[str]:
|
|
"""Playlist IDs from YT_PLAYLIST_URLS — a comma-separated list of playlist
|
|
URLs or bare ids, so adding another playlist is a one-line .env edit
|
|
instead of an add-playlist CLI call per id. A URL yields its ``list=``
|
|
query param; a token without ``list=`` and without ``://`` is taken as a
|
|
bare id; a URL lacking ``list=`` is not a playlist and is dropped."""
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
out = []
|
|
for tok in os.getenv("YT_PLAYLIST_URLS", "").split(","):
|
|
tok = tok.strip()
|
|
if not tok:
|
|
continue
|
|
if "://" in tok or tok.startswith("www."):
|
|
pid = (parse_qs(urlparse(tok).query).get("list") or [None])[0]
|
|
if pid:
|
|
out.append(pid)
|
|
else:
|
|
out.append(tok)
|
|
return out
|