scanner.py

"""Detection of unannounced and embedded quotations (no quotation marks, no introductory phrase needed).

The scanner reads the text as a stream of phonetic-skeleton words and looks for stretches that follow the Quran or a
Hadith closely:

1. *Seeds*      a sliding window of 3 words (plus two gapped variants that survive one changed word) is looked up in the
                pre-built phonetic n-gram index of the Quran.
2. *Chaining*   seeds of the same ayah on a consistent diagonal are chained into a candidate region; regions of adjacent
                ayahs that touch in the text are merged.
3. *Boundaries* the region grows word by word (tolerating one substituted word) while the text keeps following the
                source, and never crosses a sentence boundary on its own; this is the contextual boundary step.
4. *Evidence*   a region is kept only if enough of its words match and the matched words are rare enough (summed IDF), so
                everyday phrases that merely occur in the Quran are not reported.
5. *Hadith*     clause-sized windows are sent to BM25, and the best record is aligned word by word with the same rules.

Phonetic keys make the search tolerant to sound-alike spelling, but the verifier still reports every real difference.
"""
from __future__ import annotations

import re
from collections import defaultdict
from dataclasses import dataclass
from difflib import SequenceMatcher
from typing import Dict, List, Optional, Sequence, Tuple

from alignment import aligned_words
from detector import DetectedSpan, RuleDetector, trim_span
from index_builder import anchor_keys
from normalization import content_words, normalize_for_matching, phonetic_key
from retrieval import SourceRetriever

_ARABIC = re.compile(r"[\u0621-\u064A]")
_STRONG_BOUNDARY = re.compile(r"[.؟?!؛;:\n…]")


@dataclass
class Token:
    skeleton: str
    key: str
    start: int
    end: int
    boundary_after: bool   # a sentence-level punctuation mark or line break follows this word


def tokenize_with_offsets(text: str) -> List[Token]:
    matches = list(re.finditer(r"\S+", text))
    tokens: List[Token] = []
    for i, match in enumerate(matches):
        skeleton = normalize_for_matching(match.group()).replace(" ", "")
        if not skeleton or not _ARABIC.search(skeleton):
            if tokens and _STRONG_BOUNDARY.search(match.group()):
                tokens[-1].boundary_after = True
            continue
        gap_end = matches[i + 1].start() if i + 1 < len(matches) else len(text)
        trailing = text[match.end():gap_end] + match.group()[-2:]
        tokens.append(Token(skeleton, phonetic_key(skeleton), match.start(), match.end(),
                            bool(_STRONG_BOUNDARY.search(trailing))))
    return tokens


@dataclass
class _Region:
    start: int          # token index in the run (inclusive)
    end: int            # exclusive
    label: str
    matched: int
    ratio: float
    idf: float
    surah: Optional[int] = None
    first_ayah: Optional[int] = None
    last_ayah: Optional[int] = None


class CorpusScanner:
    """Finds Quran / Hadith stretches in free text. Thresholds are deliberately conservative."""

    def __init__(self, retriever: SourceRetriever, min_tokens: int = 4, min_ratio: float = 0.7, min_idf: float = 8.0,
                 hadith_min_tokens: int = 6, hadith_min_idf: float = 14.0, scan_hadith: bool = True) -> None:
        self.kb = retriever
        self.min_tokens, self.min_ratio, self.min_idf = min_tokens, min_ratio, min_idf
        self.hadith_min_tokens, self.hadith_min_idf = hadith_min_tokens, hadith_min_idf
        self.scan_hadith = scan_hadith

    # ---- public -----------------------------------------------------------------------------------------------
    def scan(self, text: str, exclude: Sequence[Tuple[int, int]] = ()) -> List[DetectedSpan]:
        tokens = tokenize_with_offsets(text)
        runs = self._runs(tokens, exclude)
        spans: List[DetectedSpan] = []
        for run in runs:
            for region in self._merge(self._scan_quran(run)):
                spans.append(self._to_span(text, run, region))
        if self.scan_hadith:
            for run in runs:
                taken = [(s.start, s.end) for s in spans]
                for region in self._scan_hadith(run, taken):
                    spans.append(self._to_span(text, run, region))
        return sorted(spans, key=lambda s: s.start)

    # ---- helpers ----------------------------------------------------------------------------------------------
    @staticmethod
    def _runs(tokens: List[Token], exclude: Sequence[Tuple[int, int]]) -> List[List[Token]]:
        runs, current = [], []
        for token in tokens:
            if any(token.start < e and token.end > s for s, e in exclude):
                if current:
                    runs.append(current)
                current = []
            else:
                current.append(token)
        if current:
            runs.append(current)
        return runs

    @staticmethod
    def _to_span(text: str, run: List[Token], region: _Region) -> DetectedSpan:
        start, end = trim_span(text, run[region.start].start, run[region.end - 1].end)
        return DetectedSpan(start, end, region.label, round(region.ratio, 3), "scan", text[start:end])

    # ---- Quran ------------------------------------------------------------------------------------------------
    def _scan_quran(self, run: List[Token]) -> List[_Region]:
        kb = self.kb
        if len(run) < self.min_tokens:
            return []
        hits: Dict[int, List[Tuple[int, int]]] = defaultdict(list)
        for key_hash, pos in anchor_keys([t.skeleton for t in run]):
            postings = kb.quran_anchors.get(key_hash)
            if not postings or len(postings) > 60:   # skip phrases that occur everywhere
                continue
            for ayah, source_pos in postings:
                hits[ayah].append((pos, source_pos))

        regions: List[_Region] = self._whole_ayahs(run)
        for ayah, found in hits.items():
            found.sort()
            chains: List[dict] = []
            for pos, source_pos in found:
                diagonal = source_pos - pos
                for chain in chains:
                    if abs(diagonal - chain["d"]) <= 2 and pos - chain["last"] <= 5:
                        chain["hits"].append((pos, source_pos))
                        chain["last"], chain["d"] = pos, diagonal
                        break
                else:
                    chains.append({"d": diagonal, "last": pos, "hits": [(pos, source_pos)]})
            for chain in chains:
                region = self._grow(run, ayah, chain)
                if region is not None:
                    regions.append(region)
        return self._non_overlapping(regions)

    def _whole_ayah_index(self) -> Dict[str, list]:
        """first word -> [(words, ayah index)] for every ayah of at least ``min_tokens`` words (built once)."""
        if getattr(self, "_whole", None) is None:
            index: Dict[str, list] = defaultdict(list)
            for i, norm in enumerate(self.kb.q_norm_match):
                words = tuple(w for w in norm.split() if w)
                if len(words) >= self.min_tokens:
                    index[words[0]].append((words, i))
            for entries in index.values():
                entries.sort(key=lambda e: -len(e[0]))   # longest first
            self._whole = index
        return self._whole

    def _whole_ayahs(self, run: List[Token]) -> List[_Region]:
        """A complete ayah typed as it is: found by exact word sequence, whatever the rarity of its words (so a short
        ayah made of common words, like ``قل هو الله احد``, is not missed)."""
        index, skeletons, found, i = self._whole_ayah_index(), [t.skeleton for t in run], [], 0
        while i < len(run):
            for words, ayah in index.get(skeletons[i], ()):
                n = len(words)
                if tuple(skeletons[i:i + n]) == words and not any(t.boundary_after for t in run[i:i + n - 1]):
                    record = self.kb.quran[ayah]
                    found.append(_Region(i, i + n, "Ayah", n, 1.0, 99.0, record["surah_id"], record["ayah_id"], record["ayah_id"]))
                    i += n - 1
                    break
            i += 1
        return found

    def _grow(self, run: List[Token], ayah: int, chain: dict) -> Optional[_Region]:
        kb = self.kb
        source = kb.q_norm_match[ayah].split()
        source_keys = [phonetic_key(w) for w in source]
        diagonal = sorted(source_pos - pos for pos, source_pos in chain["hits"])[len(chain["hits"]) // 2]
        start = min(pos for pos, _ in chain["hits"])
        end = min(len(run), max(pos for pos, _ in chain["hits"]) + 3)

        def key_at(i: int) -> Optional[str]:
            return source_keys[i + diagonal] if 0 <= i + diagonal < len(source_keys) else None

        while start > 0 and not run[start - 1].boundary_after:   # grow left while the text keeps following the source
            if run[start - 1].key == key_at(start - 1):
                start -= 1
            elif start >= 2 and not run[start - 2].boundary_after and run[start - 2].key == key_at(start - 2):
                start -= 2   # one substituted word
            else:
                break
        while end < len(run) and not run[end - 1].boundary_after:
            if run[end].key == key_at(end):
                end += 1
            elif end + 1 < len(run) and run[end + 1].key == key_at(end + 1):
                end += 2
            else:
                break

        start = self._soft_left(run, start, diagonal, source_keys)
        end = self._soft_right(run, end, diagonal, source_keys)
        positions = range(start, end)
        matched_idx = [i for i in positions if run[i].key == key_at(i)]
        matched = len(matched_idx)
        n = end - start
        idf = sum(kb.quran_bm25.idf.get(run[i].skeleton, 0.0) for i in matched_idx)
        ratio = matched / n if n else 0.0
        needed_idf = self.min_idf if n >= 5 else self.min_idf + 6.0   # very short stretches must be rare phrases
        if n < self.min_tokens or matched < self.min_tokens or ratio < self.min_ratio or idf < needed_idf:
            return None
        record = kb.quran[ayah]
        return _Region(start, end, "Ayah", matched, ratio, idf, record["surah_id"], record["ayah_id"], record["ayah_id"])

    @staticmethod
    def _closest(run: List[Token], candidates, target: str):
        """Among candidate token ranges, the one whose joined phonetic key best resembles ``target``; near-ties prefer the
        longer range (words next to a changed word usually belong to the same altered quotation)."""
        scored = [(SequenceMatcher(None, "".join(t.key for t in run[a:b]), target).ratio(), a, b) for a, b in candidates]
        if not scored:
            return None
        top = max(score for score, _, _ in scored)
        if top < 0.6:
            return None
        return max((c for c in scored if c[0] >= top - 0.2), key=lambda c: c[2] - c[1])

    def _soft_left(self, run: List[Token], start: int, diagonal: int, source_keys: List[str]) -> int:
        """Pull in up to two words before the region that look like the source words missing at its beginning."""
        missing = min(start + diagonal, 2)
        if missing <= 0:
            return start
        target = "".join(source_keys[start + diagonal - missing : start + diagonal])
        candidates = [(start - n, start) for n in range(max(1, missing - 1), missing + 2)
                      if start - n >= 0 and not any(run[i].boundary_after for i in range(start - n, start))]
        best = self._closest(run, candidates, target)
        return best[1] if best else start

    def _soft_right(self, run: List[Token], end: int, diagonal: int, source_keys: List[str]) -> int:
        missing = min(len(source_keys) - (end + diagonal), 2)
        if missing <= 0 or end >= len(run) or run[end - 1].boundary_after:
            return end
        target = "".join(source_keys[end + diagonal : end + diagonal + missing])
        candidates = [(end, end + n) for n in range(max(1, missing - 1), missing + 2)
                      if end + n <= len(run) and not any(run[i].boundary_after for i in range(end, end + n - 1))]
        best = self._closest(run, candidates, target)
        return best[2] if best else end

    @staticmethod
    def _non_overlapping(regions: List[_Region]) -> List[_Region]:
        chosen: List[_Region] = []
        for region in sorted(regions, key=lambda r: (r.matched, r.ratio), reverse=True):
            if all(region.end <= c.start or region.start >= c.end for c in chosen):
                chosen.append(region)
        return sorted(chosen, key=lambda r: r.start)

    @staticmethod
    def _merge(regions: List[_Region]) -> List[_Region]:
        """Join regions of consecutive ayahs that follow each other in the text (a quotation spanning several ayahs)."""
        merged: List[_Region] = []
        for region in regions:
            last = merged[-1] if merged else None
            if (last and last.surah == region.surah and region.first_ayah == last.last_ayah + 1
                    and region.start - last.end <= 1):
                last.end, last.matched = region.end, last.matched + region.matched
                last.ratio = last.matched / (last.end - last.start)
                last.idf += region.idf
                last.last_ayah = region.last_ayah
            else:
                merged.append(region)
        return merged

    # ---- Hadith -----------------------------------------------------------------------------------------------
    def _scan_hadith(self, run: List[Token], taken: Sequence[Tuple[int, int]]) -> List[_Region]:
        kb = self.kb
        regions: List[_Region] = []
        segment_start = 0
        for i, token in enumerate(run):
            if token.boundary_after or i == len(run) - 1:
                segment = (segment_start, i + 1)
                segment_start = i + 1
                if segment[1] - segment[0] < self.hadith_min_tokens:
                    continue
                for a, b in self._windows(*segment):
                    region = self._match_hadith(run, a, b)
                    if region is not None:
                        regions.append(region)
        return self._non_overlapping(regions)

    @staticmethod
    def _windows(start: int, end: int, size: int = 24, stride: int = 12):
        if end - start <= 40:
            yield start, end
        else:
            for a in range(start, end - 6, stride):
                yield a, min(end, a + size)

    def _match_hadith(self, run: List[Token], a: int, b: int) -> Optional[_Region]:
        kb = self.kb
        window = run[a:b]
        words = content_words(t.skeleton for t in window)
        if len(words) < 4:
            return None
        window_keys = [t.key for t in window]
        best = None
        for idx in kb.hadith_candidates(words, 3):
            record = kb.hadith[idx]
            source = [p[1] for p in aligned_words(record["matn"] or record["full"])]
            matcher = SequenceMatcher(None, window_keys, [phonetic_key(w) for w in source], autojunk=False)
            blocks = [blk for blk in matcher.get_matching_blocks() if blk.size >= 3]
            matched = sum(blk.size for blk in blocks)
            if matched >= self.hadith_min_tokens and (best is None or matched > best[0]):
                best = (matched, blocks)
        if best is None:
            return None
        matched, blocks = best
        start, end = blocks[0].a, blocks[-1].a + blocks[-1].size
        ratio = matched / (end - start)
        idf = sum(kb.hadith_data["bm25"].idf.get(window[i].skeleton, 0.0) for blk in blocks for i in range(blk.a, blk.a + blk.size))
        if ratio < self.min_ratio or idf < self.hadith_min_idf:
            return None
        return _Region(a + start, a + end, "Hadith", matched, ratio, idf)


class HybridDetector:
    """Rule-based detection first (quotation marks / brackets, typed by the corpora; phrases are only a hint), then the corpus scanner on the rest of the
    text to find unannounced and embedded quotations. Rule spans always win overlaps: an author-delimited quotation is
    verified exactly as written."""

    def __init__(self, retriever: SourceRetriever, use_scanner: bool = True, rules_use_corpus: bool = True,
                 decouple_triggers: bool = True) -> None:
        """``rules_use_corpus`` / ``decouple_triggers``: type delimited quotations from the corpora instead of from
        introductory phrases (set both to False to reproduce the earlier trigger-driven behaviour for ablations)."""
        self.rules = RuleDetector(retriever if rules_use_corpus else None, decouple_triggers=decouple_triggers)
        self.scanner = CorpusScanner(retriever) if use_scanner else None

    def detect(self, text: str) -> List[DetectedSpan]:
        spans = self.rules.detect(text)
        if self.scanner is not None:
            spans += self.scanner.scan(text, [(s.start, s.end) for s in spans])
        return sorted(spans, key=lambda s: s.start)