From 43d511f5ea17b12fa3a9d354590d7e6b74d2febf Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 6 Jul 2026 09:40:47 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20transcript=20search=20=E2=80=94=20inver?= =?UTF-8?q?ted=20index=20over=20timestamped=20transcripts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add transcript_search.py, a stdlib-only module that answers "find where we discussed X" across an archive of recordings: - TranscriptIndex: add(recording_id, segments) with duck-typed segments (attributes or mapping keys), search(query) with case-insensitive, punctuation-insensitive AND semantics, scored by term frequency and sorted by score desc then recording/time. - Match dataclass carrying recording_id, start, end, text, score. - load_transcript_json(path): reads the transcription sidecar shape {"segments": [{"start", "end", "text"}]} so it composes with the transcription pipeline output without importing it. - Honest documentation of the CJK whitespace-tokenization limitation. Tests cover timestamp fidelity, AND semantics, ranking, tie-breaking, case-insensitivity, empty-query ValueError, no-match empty list, multi-recording search, duck typing, and sidecar JSON roundtrip. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CJRVbDrp1vGYkJgNHMGPpG --- tests/test_transcript_search.py | 206 +++++++++++++++++++++ transcript_search.py | 306 ++++++++++++++++++++++++++++++++ 2 files changed, 512 insertions(+) create mode 100644 tests/test_transcript_search.py create mode 100644 transcript_search.py diff --git a/tests/test_transcript_search.py b/tests/test_transcript_search.py new file mode 100644 index 0000000..7e5d27e --- /dev/null +++ b/tests/test_transcript_search.py @@ -0,0 +1,206 @@ +"""Tests for transcript_search: inverted index over timestamped transcripts.""" + +import json +import tempfile +import unittest +from pathlib import Path + +from transcript_search import ( + Match, + Segment, + TranscriptIndex, + load_transcript_json, + tokenize, +) + + +def _make_index(): + """Build a small two-recording index used by several tests.""" + idx = TranscriptIndex() + idx.add( + "standup-monday", + [ + Segment(0.0, 4.5, "Good morning, let's start the standup."), + Segment(4.5, 12.0, "The codec budget is over by two megabytes."), + Segment(12.0, 20.0, "Codec, codec, codec — we keep saying codec."), + ], + ) + idx.add( + "retro-friday", + [ + Segment(0.0, 6.0, "Retro time: what went well this sprint?"), + Segment(6.0, 15.0, "The budget discussion about the codec dragged."), + ], + ) + return idx + + +class TokenizeTest(unittest.TestCase): + """Behavior of the shared tokenizer.""" + + def test_lowercases_and_strips_punctuation(self): + """Tokens are case-folded and punctuation is dropped.""" + self.assertEqual(tokenize("Hello, WORLD! It's fine."), ["hello", "world", "it", "s", "fine"]) + + def test_empty_text_yields_no_tokens(self): + """Empty and punctuation-only strings produce no tokens.""" + self.assertEqual(tokenize(""), []) + self.assertEqual(tokenize("... !!! ???"), []) + + +class SearchTest(unittest.TestCase): + """Indexing and search semantics.""" + + def test_hit_carries_correct_timestamps_and_recording(self): + """A match reports the recording id and the segment's start/end.""" + idx = _make_index() + matches = idx.search("budget megabytes") + self.assertEqual(len(matches), 1) + match = matches[0] + self.assertIsInstance(match, Match) + self.assertEqual(match.recording_id, "standup-monday") + self.assertEqual(match.start, 4.5) + self.assertEqual(match.end, 12.0) + self.assertIn("codec budget", match.text) + + def test_multi_word_query_uses_and_semantics(self): + """Only segments containing every query term match.""" + idx = _make_index() + # "budget" appears in two recordings, "dragged" in only one. + matches = idx.search("budget dragged") + self.assertEqual([m.recording_id for m in matches], ["retro-friday"]) + + def test_ranking_by_term_frequency(self): + """Segments with more query-term occurrences rank first.""" + idx = _make_index() + matches = idx.search("codec") + self.assertEqual(len(matches), 3) + # The codec-codec-codec segment (tf=4) must outrank single mentions. + self.assertEqual(matches[0].start, 12.0) + self.assertEqual(matches[0].recording_id, "standup-monday") + self.assertEqual(matches[0].score, 4) + self.assertGreaterEqual(matches[0].score, matches[1].score) + self.assertGreaterEqual(matches[1].score, matches[2].score) + + def test_ties_break_by_recording_then_time(self): + """Equal scores sort by recording id, then start time ascending.""" + idx = TranscriptIndex() + idx.add("b-rec", [Segment(5.0, 6.0, "alpha"), Segment(1.0, 2.0, "alpha")]) + idx.add("a-rec", [Segment(9.0, 10.0, "alpha")]) + matches = idx.search("alpha") + self.assertEqual( + [(m.recording_id, m.start) for m in matches], + [("a-rec", 9.0), ("b-rec", 1.0), ("b-rec", 5.0)], + ) + + def test_search_is_case_insensitive(self): + """Query case and transcript case are both irrelevant.""" + idx = _make_index() + upper = idx.search("CODEC BUDGET") + lower = idx.search("codec budget") + self.assertEqual(upper, lower) + self.assertEqual(len(lower), 2) + + def test_punctuation_in_query_is_ignored(self): + """Punctuation around query words does not affect matching.""" + idx = _make_index() + self.assertEqual(idx.search("codec!!!"), idx.search("codec")) + + def test_empty_query_raises_value_error(self): + """Empty or punctuation-only queries raise ValueError.""" + idx = _make_index() + with self.assertRaises(ValueError): + idx.search("") + with self.assertRaises(ValueError): + idx.search(" ... ") + + def test_no_match_returns_empty_list(self): + """Unknown terms yield an empty result, not an error.""" + idx = _make_index() + self.assertEqual(idx.search("zeppelin"), []) + # AND semantics: one known + one unknown term also yields nothing. + self.assertEqual(idx.search("codec zeppelin"), []) + + def test_search_spans_multiple_recordings(self): + """A query can hit segments across several recordings.""" + idx = _make_index() + matches = idx.search("budget") + self.assertEqual( + sorted({m.recording_id for m in matches}), + ["retro-friday", "standup-monday"], + ) + + def test_search_on_empty_index_returns_empty(self): + """Searching before any add() returns no matches.""" + self.assertEqual(TranscriptIndex().search("codec"), []) + + def test_add_accepts_mapping_segments(self): + """Duck typing: dict-shaped segments index the same as objects.""" + idx = TranscriptIndex() + added = idx.add( + "dict-rec", [{"start": 1.0, "end": 2.0, "text": "mapping works"}] + ) + self.assertEqual(added, 1) + self.assertEqual(len(idx), 1) + self.assertEqual(idx.search("mapping")[0].start, 1.0) + + def test_add_rejects_segment_missing_field(self): + """Segments lacking a required field raise TypeError.""" + idx = TranscriptIndex() + with self.assertRaises(TypeError): + idx.add("bad", [{"start": 0.0, "end": 1.0}]) + + +class LoadTranscriptJsonTest(unittest.TestCase): + """Reading the transcription sidecar JSON shape.""" + + def test_roundtrip_from_tmp_file(self): + """Sidecar JSON loads into Segments and is searchable end to end.""" + payload = { + "segments": [ + {"start": 0.0, "end": 3.2, "text": "Welcome to the meeting."}, + {"start": 3.2, "end": 9.9, "text": "We discussed the codec roadmap."}, + ] + } + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "recording.json" + path.write_text(json.dumps(payload), encoding="utf-8") + segments = load_transcript_json(path) + + self.assertEqual( + segments, + [ + Segment(0.0, 3.2, "Welcome to the meeting."), + Segment(3.2, 9.9, "We discussed the codec roadmap."), + ], + ) + idx = TranscriptIndex() + idx.add("recording", segments) + matches = idx.search("codec roadmap") + self.assertEqual(len(matches), 1) + self.assertEqual(matches[0].start, 3.2) + self.assertEqual(matches[0].end, 9.9) + + def test_rejects_non_sidecar_shape(self): + """A JSON file without a 'segments' list raises ValueError.""" + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "bad.json" + path.write_text(json.dumps(["not", "a", "sidecar"]), encoding="utf-8") + with self.assertRaises(ValueError): + load_transcript_json(path) + + def test_rejects_segment_missing_key(self): + """A segment entry missing 'text' raises ValueError naming the key.""" + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "partial.json" + path.write_text( + json.dumps({"segments": [{"start": 0.0, "end": 1.0}]}), + encoding="utf-8", + ) + with self.assertRaises(ValueError) as ctx: + load_transcript_json(path) + self.assertIn("text", str(ctx.exception)) + + +if __name__ == "__main__": + unittest.main() diff --git a/transcript_search.py b/transcript_search.py new file mode 100644 index 0000000..5bb2fac --- /dev/null +++ b/transcript_search.py @@ -0,0 +1,306 @@ +"""Search across timestamped transcripts with a lightweight inverted index. + +This module answers the archival question "find where we discussed X": +given one or many transcripts made of timestamped segments (for example the +``.json`` sidecars produced by the transcription pipeline), it builds an +in-memory inverted index and returns every segment that matches a query, +each with its recording id and start/end timestamps. + +Design notes +------------ +* **Stdlib only.** No third-party dependencies; safe to vendor anywhere. +* **Duck-typed segments.** :meth:`TranscriptIndex.add` accepts any object + exposing ``start``, ``end``, and ``text`` attributes, or an equivalent + mapping with those keys — so it composes with the transcription sidecar's + JSON output (via :func:`load_transcript_json`) without importing it. +* **AND semantics.** A multi-word query matches only segments containing + *every* query term; results are scored by summed term frequency and + sorted by score (descending), then by recording id and start time. + +Known limitation (documented honestly) +-------------------------------------- +Tokenization uses a Unicode word regex split on non-word boundaries, which +works well for whitespace-delimited languages. CJK text (Korean, Japanese, +Chinese) is only split on whitespace/punctuation, **not** morphologically +segmented — so a Korean query term matches only when the same +space-delimited token appears in the transcript. Proper CJK support would +require a morphological analyzer, which is out of scope for a +stdlib-only module. +""" + +from __future__ import annotations + +import json +import re +from collections import Counter +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Iterable, Mapping + +__all__ = [ + "Match", + "Segment", + "TranscriptIndex", + "load_transcript_json", + "tokenize", +] + +# Unicode-aware word tokenizer: runs of word characters (letters, digits, +# underscore). Case is folded by the caller; punctuation is stripped by +# construction because it never matches ``\w+``. +_WORD_RE = re.compile(r"\w+", re.UNICODE) + + +def tokenize(text: str) -> list[str]: + """Split *text* into lowercase word tokens. + + Uses a Unicode ``\\w+`` regex, so punctuation is dropped and tokens are + case-folded (``"Hello, World!"`` -> ``["hello", "world"]``). + + Note: + CJK text is split only on whitespace/punctuation boundaries — see + the module docstring for the honest limitation statement. + + Args: + text: Arbitrary text to tokenize. + + Returns: + List of lowercase tokens (possibly empty). + """ + return _WORD_RE.findall(text.lower()) + + +@dataclass(frozen=True) +class Segment: + """One timestamped chunk of a transcript. + + Attributes: + start: Segment start time in seconds. + end: Segment end time in seconds. + text: Spoken text of the segment. + """ + + start: float + end: float + text: str + + +@dataclass(frozen=True) +class Match: + """A single search hit inside a recording. + + Attributes: + recording_id: Identifier of the recording the hit belongs to + (as passed to :meth:`TranscriptIndex.add`). + start: Start timestamp (seconds) of the matching segment. + end: End timestamp (seconds) of the matching segment. + text: Original (unnormalized) text of the matching segment. + score: Relevance score — the summed term frequency of all query + terms within the segment. Higher is more relevant. + """ + + recording_id: str + start: float + end: float + text: str + score: int + + +@dataclass(frozen=True) +class _Entry: + """Internal indexed segment: original fields plus its token counts.""" + + recording_id: str + start: float + end: float + text: str + counts: Counter = field(compare=False) + + +def _read_attr(segment: Any, name: str) -> Any: + """Fetch *name* from a duck-typed segment (attribute or mapping key). + + Args: + segment: Object with ``start``/``end``/``text`` attributes, or a + mapping with those keys. + name: Field name to read. + + Returns: + The field value. + + Raises: + TypeError: If the segment exposes the field neither as an + attribute nor as a mapping key. + """ + if isinstance(segment, Mapping): + try: + return segment[name] + except KeyError: + raise TypeError( + f"segment mapping is missing required key {name!r}" + ) from None + try: + return getattr(segment, name) + except AttributeError: + raise TypeError( + f"segment object is missing required attribute {name!r}" + ) from None + + +class TranscriptIndex: + """Inverted index over timestamped transcript segments. + + Add one or many recordings with :meth:`add`, then query with + :meth:`search`. The index is in-memory and append-only; re-adding a + recording id simply indexes more segments under the same id. + + Example: + >>> idx = TranscriptIndex() + >>> idx.add("standup-01", [Segment(0.0, 4.0, "codec budget review")]) + >>> [m.recording_id for m in idx.search("codec")] + ['standup-01'] + """ + + def __init__(self) -> None: + """Create an empty index.""" + # token -> set of entry positions in self._entries + self._postings: dict[str, set[int]] = {} + self._entries: list[_Entry] = [] + + def __len__(self) -> int: + """Return the number of indexed segments.""" + return len(self._entries) + + def add(self, recording_id: str, segments: Iterable[Any]) -> int: + """Index the *segments* of one recording. + + Args: + recording_id: Stable identifier for the recording (e.g. the + source filename); echoed back on every :class:`Match`. + segments: Iterable of duck-typed segments — each must expose + ``start``, ``end`` and ``text`` as attributes (e.g. + :class:`Segment` or the transcription sidecar's segment + objects) or as mapping keys. + + Returns: + The number of segments indexed from this call. + + Raises: + TypeError: If a segment lacks one of the required fields. + """ + added = 0 + for segment in segments: + entry = _Entry( + recording_id=recording_id, + start=float(_read_attr(segment, "start")), + end=float(_read_attr(segment, "end")), + text=str(_read_attr(segment, "text")), + counts=Counter(tokenize(str(_read_attr(segment, "text")))), + ) + position = len(self._entries) + self._entries.append(entry) + for token in entry.counts: + self._postings.setdefault(token, set()).add(position) + added += 1 + return added + + def search(self, query: str) -> list[Match]: + """Find segments containing **all** words of *query*. + + Matching is case-insensitive and punctuation-insensitive (both the + query and the indexed text pass through :func:`tokenize`). + Multi-word queries use AND semantics: only segments containing + every query term are returned. + + Args: + query: One or more words to look for. + + Returns: + Matches sorted by ``score`` descending, then by + ``recording_id`` and ``start`` ascending. Empty list when + nothing matches. + + Raises: + ValueError: If the query is empty or contains no indexable + words (e.g. punctuation only). + """ + terms = tokenize(query) + if not terms: + raise ValueError("query must contain at least one word") + + # Intersect postings lists (AND semantics), rarest term first so + # the working set shrinks as fast as possible. + unique_terms = sorted( + set(terms), key=lambda t: len(self._postings.get(t, ())) + ) + candidates: set[int] | None = None + for term in unique_terms: + postings = self._postings.get(term) + if not postings: + return [] + candidates = ( + set(postings) if candidates is None else candidates & postings + ) + if not candidates: + return [] + + matches = [] + for position in candidates or (): + entry = self._entries[position] + score = sum(entry.counts[term] for term in unique_terms) + matches.append( + Match( + recording_id=entry.recording_id, + start=entry.start, + end=entry.end, + text=entry.text, + score=score, + ) + ) + matches.sort(key=lambda m: (-m.score, m.recording_id, m.start)) + return matches + + +def load_transcript_json(path: str | Path) -> list[Segment]: + """Load segments from a transcription sidecar JSON file. + + Reads the sidecar shape ``{"segments": [{"start": .., "end": .., + "text": ..}, ...]}`` and returns :class:`Segment` objects ready for + :meth:`TranscriptIndex.add`. This mirrors the transcription + pipeline's output format without importing that module, so the two + features compose while remaining independent. + + Args: + path: Path to the ``.json`` sidecar file. + + Returns: + List of :class:`Segment` in file order. + + Raises: + ValueError: If the file is not a JSON object with a ``"segments"`` + list, or a segment entry is missing ``start``/``end``/``text``. + OSError: If the file cannot be read. + json.JSONDecodeError: If the file is not valid JSON. + """ + raw = json.loads(Path(path).read_text(encoding="utf-8")) + if not isinstance(raw, dict) or not isinstance(raw.get("segments"), list): + raise ValueError( + f"{path}: expected a JSON object with a 'segments' list" + ) + segments = [] + for i, item in enumerate(raw["segments"]): + if not isinstance(item, dict): + raise ValueError(f"{path}: segments[{i}] is not an object") + try: + segments.append( + Segment( + start=float(item["start"]), + end=float(item["end"]), + text=str(item["text"]), + ) + ) + except KeyError as exc: + raise ValueError( + f"{path}: segments[{i}] is missing key {exc.args[0]!r}" + ) from None + return segments