File size: 9,166 Bytes
7b04714 | 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 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 | 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)
|