| import re |
| from datetime import datetime |
|
|
| def chunk_text_splitter(documents, separator="\n", chunk_size=1000, chunk_overlap=100): |
| """ |
| Splits the 'text' of each document into chunks of specified size, with optional overlap. |
| Returns a list of dicts with the same metadata as the input. |
| """ |
| chunks = [] |
| for doc in documents: |
| text = doc["text"] |
| meta = {k: v for k, v in doc.items() if k != "text"} |
| |
| split_text = text.split(separator) |
| joined_text = separator.join(split_text) |
| start = 0 |
| while start < len(joined_text): |
| end = min(start + chunk_size, len(joined_text)) |
| chunk = joined_text[start:end] |
| if chunk.strip(): |
| chunks.append({"text": chunk, **meta}) |
| start += chunk_size - chunk_overlap |
| return chunks |
|
|
| def paragraph_text_splitter(documents, separator="\n\n"): |
| """ |
| Splits the 'text' of each document into paragraphs. |
| Returns a list of dicts with the same metadata as the input. |
| """ |
| chunks = [] |
| for doc in documents: |
| text = doc["text"] |
| meta = {k: v for k, v in doc.items() if k != "text"} |
| for para in text.split(separator): |
| if para.strip(): |
| chunks.append({"text": para.strip(), **meta}) |
| return chunks |
|
|
| def transcript_chunker(documents): |
| """ |
| Splits each transcript document by lines and attaches speaker/timestamp if present. |
| Returns a list of dicts with the same metadata as the input. |
| """ |
| pattern = r"^\[(\d{2}:\d{2}:\d{2})\]\s+([A-Za-z]+):\s*(.*)$" |
| chunks = [] |
| for doc in documents: |
| meta = {k: v for k, v in doc.items() if k != "text"} |
| for line in doc["text"].splitlines(): |
| match = re.match(pattern, line) |
| if match: |
| timestamp = match.group(1) |
| speaker = match.group(2) |
| content = match.group(3) |
| timestamp_seconds = sum([int(x) * 60 ** i for i, x in enumerate(reversed(timestamp.split(":")))]) |
| chunk = { |
| "text": content, |
| "timestamp": timestamp, |
| "timestamp_seconds": timestamp_seconds, |
| "speaker": speaker, |
| **meta |
| } |
| chunks.append(chunk) |
| elif line.strip(): |
| |
| chunks.append({"text": line.strip(), **meta}) |
| return chunks |
|
|
| def srt_chunker(documents): |
| """ |
| Splits SRT transcript documents into chunks with start/end time, speaker, and content. |
| Returns a list of dicts with the same metadata as the input. |
| """ |
| pattern = r"(\d{2}:\d{2}:\d{2}),\d{3}\s-->\s(\d{2}:\d{2}:\d{2}),\d{3}\n\[(.*?)\]:\s(.+)" |
| chunks = [] |
| for doc in documents: |
| meta = {k: v for k, v in doc.items() if k != "text"} |
| matches = re.finditer(pattern, doc["text"], re.MULTILINE) |
| for match in matches: |
| start_time = match.group(1) |
| end_time = match.group(2) |
| speaker = match.group(3) |
| content = match.group(4).replace("\n", " ") |
| start_time_seconds = sum([int(x) * 60 ** i for i, x in enumerate(reversed(start_time.split(":")))]) |
| end_time_seconds = sum([int(x) * 60 ** i for i, x in enumerate(reversed(end_time.split(":")))]) |
| chunk = { |
| "text": content, |
| "start_time": start_time, |
| "end_time": end_time, |
| "start_time_seconds": start_time_seconds, |
| "end_time_seconds": end_time_seconds, |
| "speaker": speaker, |
| **meta |
| } |
| chunks.append(chunk) |
| return chunks |
|
|
