Spaces:
Sleeping
Sleeping
File size: 4,514 Bytes
6b64d63 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | """
backend/services/text_chunker.py
==================================
Splits long transcripts into overlapping chunks for LLM processing.
Preserves segment-level metadata (timestamps) per chunk.
"""
from typing import Dict, List
from backend.utils.config import settings
from backend.utils.logger import get_logger
logger = get_logger(__name__)
class TextChunker:
"""
Splits a Whisper transcript into overlapping text chunks
while preserving start/end timestamps for each chunk.
"""
def __init__(
self,
max_chunk_size: int = None,
overlap: int = None,
):
self.max_chunk_size = max_chunk_size or settings.MAX_CHUNK_SIZE
self.overlap = overlap or settings.CHUNK_OVERLAP
# ── Public API ────────────────────────────────────────────
def chunk_transcript(self, transcript: Dict) -> List[Dict]:
"""
Chunk a transcript dict (as returned by WhisperTranscriber).
Returns:
List of chunk dicts, each with:
- chunk_id: int index
- text: chunk text
- start: float seconds (start of chunk)
- end: float seconds (end of chunk)
- start_ts: HH:MM:SS string
- end_ts: HH:MM:SS string
- segments: original Whisper segment IDs in this chunk
"""
segments = transcript.get("segments", [])
if not segments:
logger.warning("Empty transcript — nothing to chunk.")
return []
chunks = []
current_words: List[str] = []
current_start: float = segments[0]["start"]
current_end: float = 0.0
current_seg_ids: List[int] = []
for seg in segments:
seg_words = seg["text"].split()
current_words.extend(seg_words)
current_end = seg["end"]
current_seg_ids.append(seg["id"])
if len(current_words) >= self.max_chunk_size:
chunk_text = " ".join(current_words[: self.max_chunk_size])
chunks.append(
self._make_chunk(
len(chunks),
chunk_text,
current_start,
current_end,
current_seg_ids[:],
)
)
# Overlap: keep last N words as context for next chunk
overlap_words = current_words[-self.overlap:] if self.overlap else []
current_words = overlap_words
current_start = seg["end"]
current_seg_ids = []
# Flush remaining words
if current_words:
chunk_text = " ".join(current_words)
chunks.append(
self._make_chunk(
len(chunks),
chunk_text,
current_start,
current_end,
current_seg_ids,
)
)
logger.info(
f"Transcript chunked into {len(chunks)} chunks "
f"(max_size={self.max_chunk_size}, overlap={self.overlap})"
)
return chunks
def chunk_text(self, text: str) -> List[str]:
"""
Simple text-only chunking (no timestamp preservation).
Useful for plain string input.
"""
words = text.split()
chunks = []
step = self.max_chunk_size - self.overlap
for i in range(0, len(words), step):
chunk = " ".join(words[i: i + self.max_chunk_size])
if chunk:
chunks.append(chunk)
logger.debug(f"Text split into {len(chunks)} plain chunks.")
return chunks
# ── Private ───────────────────────────────────────────────
@staticmethod
def _make_chunk(
idx: int,
text: str,
start: float,
end: float,
seg_ids: List[int],
) -> Dict:
from backend.utils.helper import seconds_to_timestamp
return {
"chunk_id": idx,
"text": text,
"start": start,
"end": end,
"start_ts": seconds_to_timestamp(start),
"end_ts": seconds_to_timestamp(end),
"segments": seg_ids,
}
|