import re def timestamp_to_seconds(ts_str: str) -> int | None: """ Converts an HH:MM:SS or MM:SS timestamp string to total seconds. Args: ts_str (str): Timestamp string in HH:MM:SS or MM:SS format. Returns: int or None: Total seconds from midnight, or None if parsing fails. """ try: # Remove milliseconds if present ts_str = ts_str.split(sep=".")[0] # Split timestamp into parts parts = list(map(int, ts_str.split(":"))) if len(parts) == 3: # HH:MM:SS format h, m, s = parts return h * 3600 + m * 60 + s elif len(parts) == 2: # MM:SS format m, s = parts return m * 60 + s else: # Invalid number of parts return None except (ValueError, AttributeError, IndexError): # Return None if parsing fails for any reason return None def seconds_to_timestamp(total_seconds: int) -> str: """ Converts total seconds to an HH:MM:SS timestamp string. (No changes needed here, always outputs full format) Args: total_seconds (int): Total seconds from midnight. Returns: str: Timestamp string in HH:MM:SS format. """ if total_seconds is None or total_seconds < 0: total_seconds = 0 # Default to 0 if input is invalid or negative # Calculate hours, minutes, and seconds hours, remainder = divmod(total_seconds, 3600) minutes, seconds = divmod(remainder, 60) # Format as HH:MM:SS with leading zeros return f"{int(hours):02}:{int(minutes):02}:{int(seconds):02}" def process_transcript(input_text: str, max_seg_duration: int = 30) -> str: """ Processes transcript text to join lines based on speaker and time. Joins consecutive lines if the speaker is the same AND the time elapsed since the start of the current segment is within max_seg_duration. Starts a new segment if the speaker changes OR the time limit is exceeded. Includes non-dialogue lines (like [MUSIC]) as separate lines. Args: input_text (str): The raw transcript text (multiline string). max_seg_duration (int): Maximum duration in seconds for a single speaker's segment before forcing a new timestamp. Defaults to 30. Returns: str: The processed transcript text as a multiline string. """ # Use splitlines() for more robust handling of different newline characters lines: list[str] = input_text.strip().splitlines() output_lines: list[str] = [] # List to store the processed output lines # Variables to keep track of the current segment being built current_segment_start_ts_str: str | None = None # Timestamp string of the segment start current_segment_start_seconds: int | None = None # Timestamp in seconds of the segment start current_speaker: str | None = None # Speaker of the current segment current_text_parts: list[str] = [] # List of text pieces in the current segment # --- UPDATED Regex --- # Capture timestamp [HH:MM:SS] or [MM:SS], speaker, and text content. # Made the HH: part optional using (?:...)? line_regex = re.compile(r"^\[((?:\d{2}:)?\d{2}:\d{2}(?:\.\d+)?)\]\s*([^:]+?):\s*(.*)$") # Breakdown of timestamp part: ((?:\d{2}:)?\d{2}:\d{2}(?:\.\d+)?) # ( - Start capture group 1 (full timestamp) # (?: - Start non-capturing group for HH: # \d{2}: - Match two digits and a colon (HH:) # )? - End non-capturing group, make it optional # \d{2}:\d{2} - Match MM:SS (required) # (?:\.\d+)? - Optionally match milliseconds (non-capturing) # ) - End capture group 1 for i, line in enumerate(lines): line = line.strip() # Remove leading/trailing whitespace if not line: continue # Skip empty lines match = line_regex.match(line) # --- Debugging Prints (Keep commented unless needed) --- # print(f"\nProcessing Line {i+1}: {line}") # if match: # _ts, _spk, _txt = match.groups() # _sec = timestamp_to_seconds(_ts) # print(f" Parsed: ts={_ts}, speaker='{_spk.strip()}', text='{_txt.strip()}', seconds={_sec}") # else: # print(f" No Match: Treating as non-dialogue.") # print(f" Current Segment State: speaker='{current_speaker}', start_ts='{current_segment_start_ts_str}', start_sec={current_segment_start_seconds}") # --- End Debugging Prints --- if not match: # --- Handle non-standard lines --- if current_speaker is not None: segment_text = " ".join(filter(None, current_text_parts)) # print(f" Finalizing previous (due to non-match): [{current_segment_start_ts_str}] {current_speaker}: {segment_text}") # DEBUG output_lines.append(f"[{current_segment_start_ts_str}] {current_speaker}: {segment_text}") current_speaker = None current_text_parts = [] current_segment_start_ts_str = None current_segment_start_seconds = None # print(f" Adding non-dialogue line: {line}") # DEBUG output_lines.append(line) continue # --- Process standard dialogue lines --- ts_str, speaker, text = match.groups() speaker = speaker.strip() text = text.strip() # Use the updated function to parse timestamp current_seconds = timestamp_to_seconds(ts_str) if current_seconds is None: print(f"Warning: Skipping line {i + 1} due to invalid timestamp format: {line}") continue # --- Logic to decide whether to start a new segment --- start_new_segment = False reason: str = "" # DEBUG if current_speaker is None: start_new_segment = True reason = "No active segment" # DEBUG elif speaker != current_speaker: start_new_segment = True reason = f"Speaker changed ('{speaker}' != '{current_speaker}')" # DEBUG # Ensure start_seconds is not None before comparison elif ( current_segment_start_seconds is not None and current_seconds - current_segment_start_seconds > max_seg_duration ): start_new_segment = True reason = f"Time limit exceeded ({current_seconds - current_segment_start_seconds}s > {max_seg_duration}s)" # DEBUG else: reason = "Continuing segment" # DEBUG # print(f" Decision: {'Start new segment' if start_new_segment else 'Continue segment'}. Reason: {reason}") # DEBUG # --- Process based on the decision --- if start_new_segment: if current_speaker is not None: segment_text = " ".join(filter(None, current_text_parts)) # print(f" Finalizing previous (due to new segment): [{current_segment_start_ts_str}] {current_speaker}: {segment_text}") # DEBUG output_lines.append(f"[{current_segment_start_ts_str}] {current_speaker}: {segment_text}") # Use seconds_to_timestamp to ensure consistent HH:MM:SS output format current_segment_start_ts_str = seconds_to_timestamp(current_seconds) current_segment_start_seconds = current_seconds current_speaker = speaker current_text_parts = [text] # print(f" Starting new segment: speaker='{speaker}', start_ts='{current_segment_start_ts_str}', start_sec={current_segment_start_seconds}, text='{text}'") # DEBUG else: # Continue the current segment if text: # print(f" Appending text: '{text}'") # DEBUG current_text_parts.append(text) # else: # DEBUG # print(f" Skipping empty text append.") # DEBUG # --- Finalize the very last segment --- if current_speaker is not None: segment_text = " ".join(filter(None, current_text_parts)) # print(f"Finalizing last segment: [{current_segment_start_ts_str}] {current_speaker}: {segment_text}") # DEBUG output_lines.append(f"[{current_segment_start_ts_str}] {current_speaker}: {segment_text}") return "\n".join(output_lines) if __name__ == "__main__": # Files path = "demos/audio/2025-4_transcript_gemini_2.5/" transcript_raw: str = f"{path}20250402_LetsVibe_Project_Update_raw.txt" transcript_out: str = transcript_raw[:-7] + "compact.txt" max_seg_duration: int = 300 # seconds # Read the original transcript with open(file=transcript_raw, mode="r", encoding="utf-8") as file: transcript_text: str = file.read() # Process the transcript processed_transcript: str = process_transcript(input_text=transcript_text, max_seg_duration=max_seg_duration) # Write the processed transcript to a new file with open(file=transcript_out, mode="w", encoding="utf-8") as file: file.write(processed_transcript)