app/models.py

"""Shared data structures passed between pipeline stages.

A ``Passage`` is one searchable unit: a sentence-packed window of transcript with a
precise [start_ms, end_ms] span inside one recording. It is what we embed, index, and
return to the UI (so the player can seek straight to ``start_ms``).
"""
from __future__ import annotations

import hashlib
from dataclasses import asdict, dataclass, field
from typing import Any, Dict, List, Optional


def make_recording_id(content_hash: str) -> str:
    """Short, stable id for a recording, derived from its content hash."""
    return content_hash[:16]


def make_passage_id(recording_id: str, start_ms: int) -> str:
    """Stable id for a passage (unique within the corpus, deterministic on re-ingest)."""
    return f"{recording_id}:{start_ms:09d}"


@dataclass
class Passage:
    id: str
    recording_id: str
    source_file: str          # path relative to the audio_dir (used to build the audio URL)
    start_ms: int
    end_ms: int
    hindi_text: str
    n_words: int = 0
    english_gloss: str = ""    # filled only when translation is enabled

    def to_row(self) -> Dict[str, Any]:
        """Row dict for LanceDB (vector is added separately at index time)."""
        return asdict(self)

    @staticmethod
    def from_words(
        recording_id: str,
        source_file: str,
        words: List[Dict[str, Any]],
        text: str,
    ) -> "Passage":
        start_ms = int(round(words[0]["start"] * 1000))
        end_ms = int(round(words[-1]["end"] * 1000))
        return Passage(
            id=make_passage_id(recording_id, start_ms),
            recording_id=recording_id,
            source_file=source_file,
            start_ms=start_ms,
            end_ms=end_ms,
            hindi_text=text.strip(),
            n_words=len(words),
        )


@dataclass
class SearchResult:
    recording_id: str
    source_file: str
    start_ms: int
    end_ms: int
    hindi_text: str
    english_gloss: str
    score: float
    rerank_score: Optional[float] = None

    @property
    def start_seconds(self) -> float:
        return self.start_ms / 1000.0

    def to_dict(self) -> Dict[str, Any]:
        d = asdict(self)
        d["start_seconds"] = round(self.start_seconds, 2)
        return d


def file_content_hash(path: str, chunk_size: int = 1 << 20) -> str:
    """SHA-256 of the file's bytes. Keys the manifest so renames don't re-transcribe."""
    h = hashlib.sha256()
    with open(path, "rb") as fh:
        for block in iter(lambda: fh.read(chunk_size), b""):
            h.update(block)
    return h.hexdigest()