services/transcript_service.py

"""
services/transcript_service.py   [v2 — cascade + diagnostics]

"No transcript extracted" is never one bug. It's a chain, and you need to know
WHICH link broke. So this service tries six sources in order and RECORDS what
happened at each one:

  1. client_override   caption segments the browser already fetched (best: the
                       browser holds the OAuth token AND has a residential IP)
  2. srt_upload        creator pasted / uploaded an .srt or .vtt
  3. captions_api      official captions.download with the owner's OAuth token
                       ← IP-agnostic. THIS is the one that works on HF Spaces.
  4. scrape            youtube-transcript-api (+ optional proxy). Works on your
                       laptop, fails on AWS/HF: YouTube blocks datacenter IPs.
  5. whisper           yt-dlp → ffmpeg → Groq whisper-large-v3 (verbose_json,
                       segment timestamps). ENABLE_WHISPER=1. Note yt-dlp hits
                       the SAME IP block, so it needs a proxy/cookies on a Space.
  6. chapters          description chapters — not speech, but a real time→text
                       map. Flagged kind="chapters" so nothing ever quotes it
                       as words the creator said.

Every attempt lands in `Transcript.attempts` and is surfaced by
GET /api/coach/transcript-check/{video_id} — so you see the exact reason in one
call instead of guessing.

env:
  TRANSCRIPT_PROXY   http(s) proxy for tiers 4/5
  TRANSCRIPT_LANGS   comma list (default "en,en-US,en-GB,hi,es,pt,fr,de,id")
  ENABLE_WHISPER     "1" to allow the audio-transcription fallback
"""

from __future__ import annotations

import asyncio
import os
import re
import subprocess
import tempfile
import time
from dataclasses import dataclass
from typing import Optional

import httpx

from services import captions_service as caps

TRANSCRIPT_PROXY = os.getenv("TRANSCRIPT_PROXY", "").strip()
PREFERRED_LANGS = [l.strip() for l in os.getenv(
    "TRANSCRIPT_LANGS", "en,en-US,en-GB,hi,es,pt,fr,de,id").split(",") if l.strip()]
ENABLE_WHISPER = os.getenv("ENABLE_WHISPER", "").strip() in ("1", "true", "yes")
GROQ_API_KEY = os.getenv("GROQ_API_KEY", "")
WHISPER_MODEL = os.getenv("WHISPER_MODEL", "whisper-large-v3")

_CACHE: dict[str, tuple[float, "Transcript"]] = {}
_CACHE_TTL = 60 * 60 * 6


# ── Model ───────────────────────────────────────────────────────────────────

@dataclass
class Segment:
    start: float
    duration: float
    text: str

    @property
    def end(self) -> float:
        return self.start + self.duration


class Transcript:
    def __init__(self, segments: list[Segment], language: str = "", generated: bool = True,
                 source: str = "none", kind: str = "speech"):
        self.segments = sorted(segments, key=lambda s: s.start)
        self.language = language
        self.generated = generated
        self.source = source
        self.kind = kind                     # "speech" | "chapters" | "none"
        self.attempts: list[dict] = []

    @property
    def available(self) -> bool:
        return len(self.segments) > 0

    @property
    def is_speech(self) -> bool:
        return self.available and self.kind == "speech"

    @property
    def word_count(self) -> int:
        return sum(len(s.text.split()) for s in self.segments)

    @property
    def duration(self) -> float:
        return self.segments[-1].end if self.segments else 0.0

    def window(self, start: float, end: float, max_chars: int = 420) -> str:
        if not self.segments:
            return ""
        hits = [s.text for s in self.segments if s.end > start and s.start < end]
        text = _clean(" ".join(hits))
        return text[:max_chars].rstrip() + ("…" if len(text) > max_chars else "")

    def at(self, t: float, before: float = 6.0, after: float = 3.0) -> str:
        return self.window(max(0.0, t - before), t + after)

    def section_at(self, t: float) -> str:
        """Chapter mode: which section was playing at second t."""
        cur = ""
        for s in self.segments:
            if s.start <= t:
                cur = s.text
        return cur

    def timeline(self, step: float = 30.0, max_chars_per_row: int = 150, limit: int = 40) -> list[dict]:
        if not self.segments:
            return []
        out, t = [], 0.0
        while t < self.duration and len(out) < limit:
            txt = self.window(t, t + step, max_chars=max_chars_per_row)
            if txt:
                out.append({"t": int(t), "timestamp": fmt_ts(t), "text": txt})
            t += step
        return out

    def to_dict(self) -> dict:
        return {
            "available": self.available,
            "kind": self.kind,
            "source": self.source,
            "language": self.language,
            "auto_generated": self.generated,
            "word_count": self.word_count,
            "segments": len(self.segments),
            "attempts": self.attempts,
        }


EMPTY = Transcript([], source="none", kind="none")


def fmt_ts(seconds: float) -> str:
    s = int(max(0, seconds))
    return f"{s // 60}:{s % 60:02d}"


def _clean(text: str) -> str:
    text = re.sub(r"\[[^\]]{0,30}\]", " ", text)
    text = (text.replace("\n", " ").replace("&#39;", "'")
                .replace("&amp;", "&").replace("&quot;", '"'))
    return re.sub(r"\s+", " ", text).strip()


def _segs(raw: list[dict]) -> list[Segment]:
    return [Segment(float(r.get("start", 0)), float(r.get("duration") or 0), _clean(r.get("text", "")))
            for r in raw if _clean(r.get("text", ""))]


# ── Tier 4: scrape ──────────────────────────────────────────────────────────

def _proxies() -> Optional[dict]:
    if not TRANSCRIPT_PROXY:
        return None
    return {"http": TRANSCRIPT_PROXY, "https": TRANSCRIPT_PROXY}


def _scrape_sync(video_id: str) -> list[dict]:
    from youtube_transcript_api import YouTubeTranscriptApi

    def pick(listing):
        for finder in ("find_manually_created_transcript", "find_generated_transcript"):
            try:
                return getattr(listing, finder)(PREFERRED_LANGS)
            except Exception:
                pass
        for t in listing:
            return t
        raise ValueError("no caption tracks")

    try:                                              # new instance API (>= 1.0)
        api = YouTubeTranscriptApi(proxies=_proxies()) if _proxies() else YouTubeTranscriptApi()
        tr = pick(api.list(video_id))
        return [{"start": f.start, "duration": f.duration, "text": f.text} for f in tr.fetch()]
    except (AttributeError, TypeError):
        pass                                          # old static API (<= 0.6)

    tr = pick(YouTubeTranscriptApi.list_transcripts(video_id, proxies=_proxies()))
    return tr.fetch()


# ── Tier 5: audio → Groq Whisper ────────────────────────────────────────────

async def _whisper(video_id: str) -> list[dict]:
    """
    yt-dlp pulls the audio, ffmpeg downsamples to 16kHz mono, Groq
    whisper-large-v3 returns segment-level timestamps. Chunked at 20 minutes to
    stay under the upload limit; offsets stitched back on.

    Needs: pip install yt-dlp  +  ffmpeg on PATH (packages.txt: ffmpeg)
    Caveat: yt-dlp hits the SAME YouTube IP block on a Space — give it
    TRANSCRIPT_PROXY, or run this tier somewhere with a clean IP.
    """
    if not GROQ_API_KEY:
        raise RuntimeError("GROQ_API_KEY not set")

    with tempfile.TemporaryDirectory() as tmp:
        audio = os.path.join(tmp, "a.m4a")
        cmd = ["yt-dlp", "-f", "bestaudio", "-o", audio,
               f"https://www.youtube.com/watch?v={video_id}", "--quiet", "--no-warnings"]
        if TRANSCRIPT_PROXY:
            cmd += ["--proxy", TRANSCRIPT_PROXY]
        p = await asyncio.to_thread(subprocess.run, cmd, capture_output=True, text=True, timeout=300)
        if p.returncode != 0 or not os.path.exists(audio):
            raise RuntimeError(f"yt-dlp failed: {(p.stderr or '')[:160]}")

        pattern = os.path.join(tmp, "c%03d.wav")
        await asyncio.to_thread(
            subprocess.run,
            ["ffmpeg", "-i", audio, "-ac", "1", "-ar", "16000", "-f", "segment",
             "-segment_time", "1200", pattern, "-y", "-loglevel", "error"],
            capture_output=True, timeout=600)

        chunks = sorted(f for f in os.listdir(tmp) if f.startswith("c") and f.endswith(".wav"))
        if not chunks:
            raise RuntimeError("ffmpeg produced no audio")

        out: list[dict] = []
        async with httpx.AsyncClient(timeout=180.0) as client:
            for i, name in enumerate(chunks):
                offset = i * 1200
                with open(os.path.join(tmp, name), "rb") as fh:
                    r = await client.post(
                        "https://api.groq.com/openai/v1/audio/transcriptions",
                        headers={"Authorization": f"Bearer {GROQ_API_KEY}"},
                        files={"file": (name, fh, "audio/wav")},
                        data={"model": WHISPER_MODEL, "response_format": "verbose_json",
                              "timestamp_granularities[]": "segment"},
                    )
                if r.status_code != 200:
                    raise RuntimeError(f"Groq whisper {r.status_code}: {r.text[:140]}")
                for s in r.json().get("segments", []):
                    out.append({"start": float(s["start"]) + offset,
                                "duration": float(s["end"]) - float(s["start"]),
                                "text": s.get("text", "")})
        return out


# ── The cascade ─────────────────────────────────────────────────────────────

async def get_transcript(video_id: str, *,
                         override: Optional[list[dict]] = None,
                         srt_text: Optional[str] = None,
                         access_token: Optional[str] = None,
                         description: str = "") -> Transcript:
    """Never raises. Always explains itself through .attempts."""
    if not (override or srt_text):
        hit = _CACHE.get(video_id)
        if hit and time.time() - hit[0] < _CACHE_TTL and hit[1].is_speech:
            return hit[1]

    attempts: list[dict] = []

    def ok(t: Transcript, tier: str, note: str = "") -> Transcript:
        attempts.append({"tier": tier, "ok": True, "note": note or f"{len(t.segments)} segments"})
        t.attempts = attempts
        _CACHE[video_id] = (time.time(), t)
        return t

    def fail(tier: str, why: str) -> None:
        attempts.append({"tier": tier, "ok": False, "note": why[:220]})
        print(f"[transcript] {video_id} · {tier}: {why[:220]}")

    # 1 ── client-supplied segments (browser already has the token + a clean IP)
    if override:
        segs = _segs(override)
        if segs:
            return ok(Transcript(segs, "supplied", False, "client_override"), "client_override")
        fail("client_override", "supplied but empty")

    # 2 ── pasted / uploaded SRT or VTT
    if srt_text and srt_text.strip():
        try:
            segs = _segs(caps.parse_srt(srt_text))
            if segs:
                return ok(Transcript(segs, "supplied", False, "srt_upload"), "srt_upload")
            fail("srt_upload", "parsed to 0 cues — is the file really SRT/VTT?")
        except Exception as e:
            fail("srt_upload", f"{type(e).__name__}: {e}")

    # 3 ── OFFICIAL captions API — the route that works on Hugging Face
    if access_token:
        try:
            srt, snip = await caps.download(video_id, access_token, PREFERRED_LANGS)
            segs = _segs(caps.parse_srt(srt))
            if segs:
                t = Transcript(segs, snip.get("language", ""),
                               snip.get("trackKind") == "ASR", "captions_api")
                return ok(t, "captions_api", f"{len(segs)} cues · {snip.get('trackKind')}")
            fail("captions_api", "downloaded file had no cues")
        except caps.CaptionsError as e:
            fail("captions_api", str(e))
        except Exception as e:
            fail("captions_api", f"{type(e).__name__}: {e}")
    else:
        fail("captions_api", "no OAuth access_token supplied — the official, IP-proof route is unusable")

    # 4 ── scrape (expected to fail on AWS/HF without a proxy)
    try:
        segs = _segs(await asyncio.to_thread(_scrape_sync, video_id))
        if segs:
            return ok(Transcript(segs, "", True, "scrape"), "scrape")
        fail("scrape", "no cues returned")
    except Exception as e:
        msg = f"{type(e).__name__}: {e}"
        if any(k in msg.lower() for k in ("block", "ip", "too many", "429", "captcha", "bot", "consent")):
            msg += "  ← YouTube is blocking this server's IP. Expected on Hugging Face/AWS. Use the captions API or set TRANSCRIPT_PROXY."
        fail("scrape", msg)

    # 5 ── whisper on the audio
    if ENABLE_WHISPER:
        try:
            segs = _segs(await _whisper(video_id))
            if segs:
                return ok(Transcript(segs, "", True, "whisper"), "whisper")
            fail("whisper", "no segments")
        except Exception as e:
            fail("whisper", f"{type(e).__name__}: {e}")
    else:
        fail("whisper", "disabled (ENABLE_WHISPER=1; needs yt-dlp + ffmpeg + a clean IP/proxy)")

    # 6 ── chapters: not speech, but a real time→text map
    try:
        ch = _segs(caps.parse_chapters(description))
        if ch:
            t = Transcript(ch, "", False, "chapters", kind="chapters")
            return ok(t, "chapters", f"{len(ch)} chapters — LOW FIDELITY, not speech")
        fail("chapters", "no timestamped chapters in the description")
    except Exception as e:
        fail("chapters", str(e))

    empty = Transcript([], source="none", kind="none")
    empty.attempts = attempts
    return empty


def unavailable_reason(t: Transcript) -> str:
    if t.is_speech:
        return ""
    if t.kind == "chapters":
        return ("No captions available — the Coach is using your description chapters as a coarse time "
                "map. It can say WHICH SECTION viewers left in, but not the exact words. Turn on "
                "auto-captions in YouTube Studio for word-level diagnosis.")
    return ("No transcript could be extracted. Most likely: the 'youtube.force-ssl' scope isn't granted "
            "(so the official captions API is closed to us) AND YouTube is blocking this server's IP for "
            "scraping. Reconnecting the channel to grant that scope fixes it with no proxy needed.")