"""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/.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/.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/.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)