momentum-system/youtube_claims/recap/channel.py

"""Channel listing, audio download, whisper transcription and bundle building.

Runs on the owner's Windows workstation: YouTube blocks the TrueNAS server's IP, and the caption
API rate-limits (IpBlocked after ~20 requests), so we pull AUDIO with yt-dlp -- one polite
request per new video -- and transcribe locally with faster-whisper."""

from __future__ import annotations

import datetime as dt
import json
import time
from pathlib import Path

from youtube_claims.recap import settings

_whisper = None


def list_uploads(limit: int | None = None) -> list[dict]:
    """Newest-first [{id, title, duration}] from the channel's /videos tab (flat, one request)."""
    import yt_dlp
    opts = {"extract_flat": "in_playlist", "quiet": True, "no_warnings": True,
            "playlistend": limit or settings.LIST_LIMIT}
    with yt_dlp.YoutubeDL(opts) as ydl:
        info = ydl.extract_info(settings.CHANNEL_URL, download=False)
    out = []
    for e in info.get("entries") or []:
        if e and e.get("id"):
            out.append({"id": e["id"], "title": e.get("title") or "", "duration": e.get("duration")})
    return out


def video_meta(vid: str, title: str = "", duration=None) -> dict:
    """{id, title, duration, upload_date} -- cached in transcripts/<id>.meta.json (the flat
    listing has no upload date, so new ids cost one metadata request each)."""
    f = settings.dirs()["transcripts"] / f"{vid}.meta.json"
    if f.exists():
        m = json.loads(f.read_text(encoding="utf-8"))
        if m.get("upload_date"):
            return m
    import yt_dlp
    with yt_dlp.YoutubeDL({"quiet": True, "no_warnings": True, "skip_download": True}) as ydl:
        info = ydl.extract_info(f"https://www.youtube.com/watch?v={vid}", download=False)
    m = {"id": vid, "title": info.get("title") or title, "duration": str(info.get("duration") or duration or ""),
         "upload_date": info.get("upload_date") or ""}
    f.write_text(json.dumps(m), encoding="utf-8")
    return m


def download_audio(vid: str) -> Path:
    """Best m4a audio (no ffmpeg re-encode needed; faster-whisper decodes it directly)."""
    ad = settings.dirs()["audio"]
    have = next(iter(ad.glob(f"{vid}.*")), None)
    if have and not have.name.endswith(".part"):
        return have
    import yt_dlp
    opts = {"format": "ba[ext=m4a]/ba", "outtmpl": str(ad / "%(id)s.%(ext)s"), "quiet": True,
            "no_warnings": True, "noprogress": True}
    with yt_dlp.YoutubeDL(opts) as ydl:
        ydl.download([f"https://www.youtube.com/watch?v={vid}"])
    got = next((p for p in ad.glob(f"{vid}.*") if not p.name.endswith(".part")), None)
    if got is None:
        raise RuntimeError(f"yt-dlp produced no audio file for {vid}")
    return got


def transcribe(vid: str, audio: Path) -> list[dict]:
    """faster-whisper segments, cached in transcripts/<id>.whisper.json (the pilot's format)."""
    out = settings.dirs()["transcripts"] / f"{vid}.whisper.json"
    if out.exists():
        return json.loads(out.read_text(encoding="utf-8"))["segments"]
    global _whisper
    if _whisper is None:
        from faster_whisper import WhisperModel
        _whisper = WhisperModel(settings.WHISPER_MODEL, device="cpu", compute_type="int8",
                                cpu_threads=settings.WHISPER_THREADS)
    t0 = time.time()
    segs, _ = _whisper.transcribe(str(audio), initial_prompt=settings.WHISPER_PROMPT, vad_filter=True,
                                  language="en")
    rows = [{"start": round(s.start, 2), "end": round(s.end, 2), "text": s.text} for s in segs]
    out.write_text(json.dumps({"video_id": vid, "source": "whisper", "whisper_model": settings.WHISPER_MODEL,
                               "seconds": round(time.time() - t0, 1), "segments": rows}), encoding="utf-8")
    return rows


def best_transcript(vid: str) -> tuple[str | None, list[dict] | None]:
    """Local whisper (small.en) > repo whisper (medium) > YouTube captions -- as the pilot."""
    td = settings.dirs()["transcripts"]
    repo_tx = settings.DATA_DIR.parent / "transcripts" / f"{vid}.json"   # youtube_claims whisper-medium
    for path, src in ((td / f"{vid}.whisper.json", "whisper-small"), (repo_tx, "whisper-medium")):
        if path.exists():
            return src, json.loads(path.read_text(encoding="utf-8"))["segments"]
    cap = td / f"{vid}.captions.json"
    if cap.exists():
        rows = json.loads(cap.read_text(encoding="utf-8"))
        return "captions", [{"start": r["start"], "text": r["text"]} for r in rows]
    return None, None


def prior_weekdays(upload_date: str, n: int = 1) -> list[str]:
    """The upload date (if a weekday) and the n weekdays before it, as YYYY-MM-DD."""
    d = dt.date(int(upload_date[:4]), int(upload_date[4:6]), int(upload_date[6:]))
    out = [d.isoformat()] if d.weekday() < 5 else []
    while len(out) < (n + 1 if d.weekday() < 5 else n):
        d -= dt.timedelta(days=1)
        if d.weekday() < 5:
            out.append(d.isoformat())
    return out


def scanner_for(upload_date: str, ourside=None) -> dict:
    """{session date: [(symbol, gap %, price)]} for the upload day and the weekday before --
    cached in scanner/<upload_date>.json (the pilot's format)."""
    f = settings.dirs()["scanner"] / f"{upload_date}.json"
    if f.exists():
        return json.loads(f.read_text())
    if ourside is None:
        return {}
    scan = {d: [list(r) for r in ourside.day_universe(d)] for d in prior_weekdays(upload_date)}
    if any(scan.values()):
        f.write_text(json.dumps(scan))
    return scan


def bundle_source(bundle: str) -> str | None:
    for ln in bundle.splitlines()[:6]:
        if ln.startswith("transcript_source: "):
            return ln.split(": ", 1)[1].strip()
    return None


def make_bundle(meta: dict, source: str, segments: list[dict], scanner: dict) -> str:
    """One plain-text bundle: meta + our scanner list + timestamped transcript (pilot format)."""
    lines = [f"video_id: {meta['id']}", f"upload_date: {meta['upload_date']}", f"title: {meta.get('title', '')}",
             f"transcript_source: {source}", "", "OUR SCANNER LIST (symbol, gap %, price) by session date:"]
    for d, rows in scanner.items():
        lines.append(f"  {d}: " + ", ".join(f"{s} {g}% ${p}" if p else f"{s} {g}%" for s, g, p in rows))
    lines += ["", "TRANSCRIPT (mm:ss text):"]
    for s in segments:
        t = int(s["start"])
        lines.append(f"[{t // 60:02d}:{t % 60:02d}] {s['text'].strip()}")
    return "\n".join(lines)