momentum-system/youtube_claims/recap/quotes.py

"""Anti-hallucination: every quote the LLM returns must be findable in the transcript.

Extends youtube_claims.extraction._quote_supported (whitespace-normalized 40-char prefix match,
which accepts a quote whose tail is invented) with a full-length, punctuation-insensitive token
match plus a small fuzzy tolerance for a mis-copied word,
and SNAPS an accepted quote to the transcript's own words -- so what we store is verbatim
source text, never the model's paraphrase of it. A quote that can't be located is rejected
and the caller drops the item it supported."""

from __future__ import annotations

import re
from difflib import SequenceMatcher

from youtube_claims.extraction import _norm

_TOKEN = re.compile(r"[a-z0-9]+(?:[.,'][a-z0-9]+)*")


def _tok(s: str) -> list[str]:
    """Lowercased word tokens; '$5.50' -> '5.50', "don't" -> "don't", 'up-to' -> 'up','to'."""
    return _TOKEN.findall((s or "").lower())


class Transcript:
    """A transcript indexed for quote lookup. Keeps the original words so a matched quote can
    be returned exactly as transcribed (ASR noise included)."""

    def __init__(self, text: str):
        self.text = text or ""
        self.ntext = _norm(self.text)
        self.words = self.text.split()
        self.wtoks = [_tok(w) for w in self.words]          # tokens per original word
        self.toks: list[str] = []
        self.tok_word: list[int] = []                         # token index -> word index
        for wi, ts in enumerate(self.wtoks):
            for t in ts:
                self.toks.append(t)
                self.tok_word.append(wi)
        self.joined = " " + " ".join(self.toks) + " "
        # number tokens also without thousands separators / leading '$' for price lookups
        self.token_set = set(self.toks) | {t.replace(",", "") for t in self.toks}
        self._first: dict[str, list[int]] = {}
        for i, t in enumerate(self.toks):
            self._first.setdefault(t, []).append(i)

    def _span(self, i: int, n: int) -> str:
        a, b = self.tok_word[i], self.tok_word[min(i + n, len(self.toks)) - 1]
        return " ".join(self.words[a:b + 1])

    def find(self, quote: str, min_ratio: float = 0.9) -> str | None:
        """The transcript text matching `quote`, or None if it isn't there.
        Exact (token-level, punctuation-blind) first; then a fuzzy window of the same length
        whose tokens agree >= min_ratio (tolerates one mis-copied word in ten)."""
        q = _tok(quote)
        if len(q) < 3:                    # too short to be evidence of anything
            return None
        needle = " " + " ".join(q) + " "
        pos = self.joined.find(needle)
        if pos >= 0:
            start = self.joined[:pos + 1].count(" ") - 1
            return self._span(start, len(q))
        if len(q) < 6:
            return None
        n = len(q)
        cands: set[int] = set()
        for k in range(min(4, n)):                            # windows anchored on any of the
            for i in self._first.get(q[k], ()):               # quote's first four tokens
                if i - k >= 0:
                    cands.add(i - k)
        best, best_i = 0.0, None
        for i in sorted(cands):
            win = self.toks[i:i + n]
            if len(win) < n * 0.8:
                continue
            r = SequenceMatcher(None, q, win, autojunk=False).ratio()
            if r > best:
                best, best_i = r, i
        if best_i is None or best < min_ratio:
            return None
        # a dropped/added word shifts the true end: pick the window length that fits best
        best_n = max(range(max(1, n - 2), n + 3),
                     key=lambda m: (SequenceMatcher(None, q, self.toks[best_i:best_i + m], autojunk=False).ratio(), -abs(m - n)))
        return self._span(best_i, best_n)

    def supports(self, quote: str) -> bool:
        return self.find(quote) is not None


def transcript_text_from_bundle(bundle: str) -> str:
    """The transcript body of a bundle, without the [mm:ss] prefixes (segments joined by
    spaces so a quote spanning two segments still matches)."""
    body = bundle.split("TRANSCRIPT (mm:ss text):", 1)[-1]
    return " ".join(re.sub(r"^\[\d+:\d+\]\s?", "", ln) for ln in body.splitlines() if ln.strip())