Aryan2301/YouTube_RAG_Intelligence

1from langchain_core.documents import Document2from utils.constants import CHUNK_SIZE3 4def create_timestamp_aware_chunks(video_id: str, transcript_segments: list) -> list[Document]:5    """6    Groups transcript segments into chunks, preserving start and end timestamps.7    Returns a list of LangChain Document objects.8    """9    if not transcript_segments:10        return []11 12    documents = []13    current_text, current_start = [], None14    char_count = 015    approx_limit = CHUNK_SIZE * 4 # rough estimate for characters per chunk16    17    for seg in transcript_segments:18        if current_start is None:19            current_start = seg["start"]20            21        current_text.append(seg["text"])22        char_count += len(seg["text"]) + 123        24        if char_count > approx_limit:25            documents.append(Document(26                page_content=" ".join(current_text),27                metadata={28                    "video_id": video_id, 29                    "start_time": current_start, 30                    "end_time": seg["start"] + seg["duration"]31                }32            ))33            # Start next chunk with overlap34            current_text = [seg["text"]]35            current_start = seg["start"]36            char_count = len(seg["text"]) + 137            38    if current_text:39        documents.append(Document(40            page_content=" ".join(current_text),41            metadata={42                "video_id": video_id, 43                "start_time": current_start, 44                "end_time": transcript_segments[-1]["start"] + transcript_segments[-1]["duration"]45            }46        ))47        48    return documents49