diff --git a/alembic/versions/d4a72f51c8e3_v4_sources_transcript_segments.py b/alembic/versions/d4a72f51c8e3_v4_sources_transcript_segments.py new file mode 100644 index 0000000..fe1f6b5 --- /dev/null +++ b/alembic/versions/d4a72f51c8e3_v4_sources_transcript_segments.py @@ -0,0 +1,42 @@ +"""v4 second_brain: sources.transcript_segments JSONB + +Revision ID: d4a72f51c8e3 +Revises: c8f1d9e34b7a +Create Date: 2026-05-25 13:00:00.000000 + +Adds a JSONB column to `sources` for the tower-side worker to persist +segment-level transcription output (start/end/text per segment + word- +level timing when faster-whisper returns it). Additive: `transcript_text` +remains the canonical full-text the extractor consumes; this column is a +secondary index for future use (chunked retrieval, timestamp-anchored +quoting, etc.). + +JSONB rather than JSON so equality/containment ops are indexable later +without a re-migration. No index added yet — Travis flagged "no +gold-plating" and the column may sit unused for a while. + +Same lovebug-no-CREATE-on-petalbrain guard as the prior migrations: +env.py has already bootstrapped the schema, this just SETs search_path +and emits DDL inside it. +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op + +revision: str = "d4a72f51c8e3" +down_revision: str | Sequence[str] | None = "c8f1d9e34b7a" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.execute("SET search_path TO second_brain") + op.execute("ALTER TABLE sources ADD COLUMN transcript_segments JSONB") + + +def downgrade() -> None: + op.execute("SET search_path TO second_brain") + op.execute("ALTER TABLE sources DROP COLUMN IF EXISTS transcript_segments") diff --git a/src/second_brain/models.py b/src/second_brain/models.py index daa42b1..1f67131 100644 --- a/src/second_brain/models.py +++ b/src/second_brain/models.py @@ -29,6 +29,7 @@ from sqlalchemy import ( Text, Time, ) +from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship @@ -113,6 +114,11 @@ class Source(Base): media_path: Mapped[Optional[str]] = mapped_column(String(1000), nullable=True) transcript_path: Mapped[Optional[str]] = mapped_column(String(1000), nullable=True) transcript_text: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + # Segment-level output from faster-whisper. List of dicts + # `{id, start, end, text, words: [{start, end, word, probability}, …]}`. + # transcript_text remains the canonical full text the extractor reads; + # this is an additive index for future timestamp-anchored use cases. + transcript_segments: Mapped[Optional[list]] = mapped_column(JSONB, nullable=True) # Metadata from the source published_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) diff --git a/src/second_brain/scheduler/transcribe_worker.py b/src/second_brain/scheduler/transcribe_worker.py index 57a2527..d54469d 100644 --- a/src/second_brain/scheduler/transcribe_worker.py +++ b/src/second_brain/scheduler/transcribe_worker.py @@ -293,10 +293,28 @@ class TranscribeWorker: write_srt(segments, srt_path) source.transcript_path = str(srt_path) source.transcript_text = text + # transcript_segments is additive — if the JSONB write blows up + # (oversize row, malformed dict, etc.) we still want the + # canonical text + status transition to land. Worst case: this + # row's segment payload is missing and a backfill picks it up + # later; we never crash the pipeline over it. + try: + source.transcript_segments = segments + except Exception as exc: + logger.warning( + "transcript_segments persist failed for source %s: %s", + source.id, + exc, + ) source.status = SourceStatus.TRANSCRIBED source.updated_at = utcnow() release_claim(sess, source.id) - logger.info("transcribed source %s — %d chars", source.id, len(text)) + logger.info( + "transcribed source %s — %d chars, %d segment(s)", + source.id, + len(text), + len(segments), + ) # ------------------------------------------------------------------ # Helpers diff --git a/src/second_brain/transcribe.py b/src/second_brain/transcribe.py index 8c7f201..943cd5c 100644 --- a/src/second_brain/transcribe.py +++ b/src/second_brain/transcribe.py @@ -21,7 +21,7 @@ from __future__ import annotations import logging import os from pathlib import Path -from typing import Iterable, Optional +from typing import Any, Iterable, Optional logger = logging.getLogger(__name__) @@ -116,8 +116,12 @@ class Transcriber: ) -> tuple[str, list[dict]]: """Transcribe `media_file`. Returns (full_text, segments). - Each segment dict has start/end (float seconds) and text (str) — - same shape the SRT writer needs. + Segments are converted via `segment_to_dict` — each one carries + `id/start/end/text` plus a `words` list when faster-whisper + returns word-level timing (we always pass `word_timestamps=True`). + + `transcript_text` is the joined segment text — same contract the + extractor has consumed since before segment capture existed. """ self._load() assert self._impl is not None # for the type-checker @@ -126,6 +130,7 @@ class Transcriber: str(media_file), language=language, beam_size=beam_size, + word_timestamps=True, ) logger.info( "transcribe start: %s language=%s duration=%.1fs", @@ -137,14 +142,71 @@ class Transcriber: segments: list[dict] = [] text_parts: list[str] = [] for seg in segments_iter: - t = (seg.text or "").strip() - segments.append({"start": float(seg.start), "end": float(seg.end), "text": t}) + d = segment_to_dict(seg) + segments.append(d) + t = d["text"] if t: text_parts.append(t) return " ".join(text_parts), segments +# --------------------------------------------------------------------------- +# Segment serialisation +# --------------------------------------------------------------------------- + + +def _word_to_dict(w: Any) -> dict | None: + """Convert a faster-whisper `Word` to a plain dict, or None on failure. + + Done per-word so one malformed item can't drop the whole segment. + """ + try: + return { + "start": float(w.start), + "end": float(w.end), + "word": getattr(w, "word", "") or "", + "probability": float(getattr(w, "probability", 0.0) or 0.0), + } + except Exception: # noqa: BLE001 — defensive against partial outputs + return None + + +def segment_to_dict(seg: Any) -> dict: + """Convert a faster-whisper `Segment` to a JSON-safe plain dict. + + Shape: + { + "id": int, + "start": float, "end": float, + "text": str, # stripped + "words": [ # empty list when word_timestamps was off + {"start": float, "end": float, "word": str, "probability": float}, + … + ], + } + + The faster-whisper objects are NamedTuple-shaped (CTranslate2 output) + and aren't JSON-serialisable directly; flattening to dicts also lets + us drop the upstream dependency from any consumer that later reads + the JSONB column. + """ + text = (getattr(seg, "text", "") or "").strip() + words_raw = getattr(seg, "words", None) or [] + words: list[dict] = [] + for w in words_raw: + d = _word_to_dict(w) + if d is not None: + words.append(d) + return { + "id": int(getattr(seg, "id", 0) or 0), + "start": float(getattr(seg, "start", 0.0) or 0.0), + "end": float(getattr(seg, "end", 0.0) or 0.0), + "text": text, + "words": words, + } + + # --------------------------------------------------------------------------- # SRT writer # --------------------------------------------------------------------------- @@ -168,4 +230,4 @@ def _fmt_timestamp(seconds: float) -> str: return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}" -__all__ = ["Transcriber", "resolve_settings", "write_srt"] +__all__ = ["Transcriber", "resolve_settings", "segment_to_dict", "write_srt"] diff --git a/tests/test_transcribe.py b/tests/test_transcribe.py index d75a624..ba239d3 100644 --- a/tests/test_transcribe.py +++ b/tests/test_transcribe.py @@ -63,6 +63,95 @@ def test_write_srt_roundtrip(tmp_path): assert "2\n00:00:01,500 --> 00:00:03,250\nworld" in body +# --------------------------------------------------------------------------- +# Segment / word conversion — exercised on faked faster-whisper objects so +# we don't need a real model + audio to validate the JSONB-bound shape. +# --------------------------------------------------------------------------- + + +class _FakeWord: + def __init__(self, start, end, word, probability): + self.start = start + self.end = end + self.word = word + self.probability = probability + + +class _FakeSegment: + def __init__(self, id, start, end, text, words=None): + self.id = id + self.start = start + self.end = end + self.text = text + self.words = words + + +def test_segment_to_dict_with_words(): + """Happy path: faster-whisper segment + word_timestamps → JSON-safe dict.""" + import json + + from second_brain.transcribe import segment_to_dict + + seg = _FakeSegment( + id=3, + start=1.25, + end=3.5, + text=" hello world ", + words=[ + _FakeWord(1.25, 1.6, " hello", 0.92), + _FakeWord(1.6, 3.5, " world", 0.87), + ], + ) + d = segment_to_dict(seg) + + assert d == { + "id": 3, + "start": 1.25, + "end": 3.5, + "text": "hello world", # stripped + "words": [ + {"start": 1.25, "end": 1.6, "word": " hello", "probability": 0.92}, + {"start": 1.6, "end": 3.5, "word": " world", "probability": 0.87}, + ], + } + # Must be JSON-serialisable for the JSONB column. + json.dumps(d) + + +def test_segment_to_dict_without_words(): + """word_timestamps=False (or older faster-whisper) → segment.words is None.""" + from second_brain.transcribe import segment_to_dict + + seg = _FakeSegment(id=0, start=0.0, end=1.0, text="just text", words=None) + d = segment_to_dict(seg) + assert d["text"] == "just text" + assert d["words"] == [] + + +def test_segment_to_dict_skips_malformed_words(): + """A single bad word entry must not drop the whole segment.""" + from second_brain.transcribe import segment_to_dict + + class _BadWord: + # Missing start → float(None) raises in the helper, must be skipped. + start = None + end = 1.0 + word = "broken" + probability = 0.5 + + seg = _FakeSegment( + id=1, + start=0.0, + end=1.0, + text="hi", + words=[_FakeWord(0.0, 0.5, "hi", 0.99), _BadWord()], + ) + d = segment_to_dict(seg) + # Good word survives; bad word is filtered. + assert len(d["words"]) == 1 + assert d["words"][0]["word"] == "hi" + + # --------------------------------------------------------------------------- # Integration: synthesized audio → CPU/int8 faster-whisper # ---------------------------------------------------------------------------