Spaces:

alexkolo
/

audio-transcript-gemini-2.5

Sleeping

audio-transcript-gemini-2.5 / transcript_compact.py

Upload 3 files

7b04714 verified over 1 year ago

9.17 kB

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)

	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)