#!/usr/bin/env python3 import json import sys import re from typing import List, Dict, Any, Optional from pathlib import Path # Pre-compile regex for efficiency SECTION_MARKER_PATTERN = re.compile(r'\[\w+(?:\s*\d*)?\]\s*') def format_vtt_time(seconds_float: float) -> str: """Format a time in seconds as a WebVTT timestamp (HH:MM:SS.mmm)""" if not isinstance(seconds_float, (int, float)) or seconds_float < 0: seconds_float = 0.0 # Default to 0 if invalid hours, remainder = divmod(seconds_float, 3600) minutes, seconds = divmod(remainder, 60) milliseconds = int((seconds_float * 1000) % 1000) return f"{int(hours):02d}:{int(minutes):02d}:{int(seconds):02d}.{milliseconds:03d}" def process_lyrics_simple(hoot_data: List[Dict[str, Any]]) -> tuple[str, str, str]: """ Process lyrics data using a single pass, relying on ASR word order and newlines. Returns: tuple[str, str, str]: (plain_text, line_timestamps_vtt, word_timestamps_vtt) """ lines_data = [] current_line_words = [] last_word_end_time = 0.0 # --- Pass 1: Aggregate words into lines based on '\n' --- for i, word_info in enumerate(hoot_data): if not isinstance(word_info, dict): continue # Skip non-dict items # Get word text and clean section markers raw_word_text = word_info.get("word", "") cleaned_word_text = SECTION_MARKER_PATTERN.sub('', raw_word_text) # Get timestamps, ensuring they are valid floats start_s = word_info.get("start_s") end_s = word_info.get("end_s") # Basic validation and fallback for timestamps try: start_s = float(start_s) if start_s is not None else None end_s = float(end_s) if end_s is not None else None except (ValueError, TypeError): start_s = None end_s = None print(f"Warning: Invalid timestamp format for word '{raw_word_text}'. Skipping timing.", file=sys.stderr) # If timestamps are missing or invalid, try to infer minimally if start_s is None: start_s = last_word_end_time # Start immediately after last word if end_s is None: end_s = start_s + 0.2 # Assign a small default duration if end is missing # Ensure times are monotonic relative to the *previous word* if start_s < last_word_end_time: start_s = last_word_end_time # Force monotonicity # Ensure end time is after start time if end_s <= start_s: end_s = start_s + 0.05 # Ensure a minimal duration # Update last known end time last_word_end_time = end_s # Split word by newlines, processing parts parts = cleaned_word_text.split('\n') num_parts = len(parts) for j, part in enumerate(parts): part_text = part # Preserve original text exactly is_last_part = (j == num_parts - 1) # If part has content, add it to the current line if part_text: word_part_info = { "text": part_text, "original_word": raw_word_text, "start_s": start_s, "end_s": end_s, } current_line_words.append(word_part_info) # If a newline was encountered or it's the last word, finalize the line if not is_last_part or i == len(hoot_data) - 1: if current_line_words: # Determine line start/end from its constituent words line_start_s = current_line_words[0]['start_s'] line_end_s = current_line_words[-1]['end_s'] # Ensure line end is strictly after line start if line_end_s <= line_start_s: line_end_s = line_start_s + 0.1 # SIMPLE APPROACH: Just concatenate the exact text values directly # This preserves any spaces in the original tokens and doesn't add any new ones line_text = "".join([w['text'] for w in current_line_words]) # Only trim whitespace at the very beginning and end line_text = line_text.strip() lines_data.append({ "text": line_text, "start_s": line_start_s, "end_s": line_end_s, "words": list(current_line_words) }) current_line_words = [] # Reset for the next line # --- Pass 2: Format Outputs --- plain_text_lines = [] line_timestamps_vtt = "WEBVTT\n\n" word_timestamps_vtt = "WEBVTT\n\n" last_line_end_time_vtt = -1.0 for idx, line_info in enumerate(lines_data): line_text = line_info['text'] start_s = line_info['start_s'] end_s = line_info['end_s'] # Ensure start time for VTT is monotonically increasing if start_s < last_line_end_time_vtt: start_s = last_line_end_time_vtt # Ensure end time is after start time for VTT if end_s <= start_s: end_s = start_s + 0.05 # Format VTT times start_time_str = format_vtt_time(start_s) end_time_str = format_vtt_time(end_s) # 1. Plain Text (add stanza breaks every 4 lines for readability) plain_text_lines.append(line_text) if (idx + 1) % 4 == 0 and idx < len(lines_data) - 1: plain_text_lines.append("") # Add blank line for stanza break # 2. Line-Level Timestamps VTT line_timestamps_vtt += f"{start_time_str} --> {end_time_str}\n" line_timestamps_vtt += f"{line_text}\n\n" # 3. Word-Level Timestamps VTT word_timestamps_vtt += f"{idx + 1}\n" # VTT Cue ID word_timestamps_vtt += f"{start_time_str} --> {end_time_str}\n" # Simplify word-level timing approach line_content_with_word_times = "" last_word_time_in_line = start_s for word_detail in line_info['words']: word_start_s = word_detail['start_s'] word_text = word_detail['text'] if not word_text: continue # Ensure monotonic timing within line if word_start_s < last_word_time_in_line: word_start_s = last_word_time_in_line word_time_str = format_vtt_time(word_start_s) # First word in line doesn't need a leading space if not line_content_with_word_times: line_content_with_word_times = f"<{word_time_str}>{word_text}" else: # Don't add space here - just concatenate with timing tag line_content_with_word_times += f"<{word_time_str}>{word_text}" last_word_time_in_line = word_start_s # Fallback if no words processed if not line_content_with_word_times and line_text: line_content_with_word_times = f"<{start_time_str}>{line_text}" word_timestamps_vtt += f"{line_content_with_word_times.strip()}\n\n" # Update last line end time for next iteration last_line_end_time_vtt = end_s # Join plain text lines plain_text = "\n".join(plain_text_lines) # Clean trailing '%' artifacts if present plain_text = plain_text.rstrip('%').strip() line_timestamps_vtt = line_timestamps_vtt.rstrip('%').strip() word_timestamps_vtt = word_timestamps_vtt.rstrip('%').strip() return plain_text, line_timestamps_vtt, word_timestamps_vtt # Pre-compile regex for efficiency SECTION_MARKER_PATTERN = re.compile(r"\[(\w+(?:\s*\d*)?)\]\s*") # TODO: we probably want first names or something ANONYMIZED_SPEAKER_NAMES = [str(n) for n in range(100)] def hoot_lyrics_to_timed_lyrics_json(hoot_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ Process lyrics data into a timed lyrics JSON format with proper tokenization. The output JSON structure is a list of lines, where each line has: - text: the full text of the line - start_time: when the line starts (in seconds) - end_time: when the line ends (in seconds) - words: array of word objects with text, start_time, and end_time - section: the most recent section marker (e.g., "Verse", "Chorus") Returns: List[Dict[str, Any]]: Timed lyrics-friendly JSON structure """ # Step 1: Extract clean tokens with timing and handle section markers tokens = [] current_section = "" for word_info in hoot_data: if not isinstance(word_info, dict): continue # Get word text and timing raw_word = word_info.get("word", "") start_time = float(word_info.get("start_s", word_info.get("start", 0))) end_time = float(word_info.get("end_s", word_info.get("end", start_time + 0.1))) # Ensure end time is after start time if end_time <= start_time: end_time = start_time + 0.05 # Check for section markers section_match = SECTION_MARKER_PATTERN.search(raw_word) if section_match: current_section = section_match.group(1).strip() # Remove section marker from display text raw_word = SECTION_MARKER_PATTERN.sub("", raw_word) # Skip empty tokens if not raw_word: continue # Handle newlines - split into multiple tokens with same timing parts = raw_word.split("\n") for i, part in enumerate(parts): if i > 0: # This is a new line tokens.append( { "type": "newline", "start_s": start_time, "end_s": start_time + 0.001, # Minimal duration for newline } ) if part: # Skip empty parts tokens.append( { "type": "word", "text": part, "start_s": start_time, "end_s": end_time, "section": current_section, } ) # Step 2: Group tokens into lines lines = [] current_line = [] for token in tokens: if token["type"] == "word": current_line.append(token) elif token["type"] == "newline" and current_line: # Finalize current line if current_line: line_text = "".join(t["text"] for t in current_line).strip() if line_text: # Only add non-empty lines lines.append( { "text": line_text, "tokens": current_line.copy(), "section": current_line[0]["section"], } ) # Start a new line current_line = [] # Add the last line if it exists if current_line: line_text = "".join(t["text"] for t in current_line).strip() if line_text: lines.append( {"text": line_text, "tokens": current_line.copy(), "section": current_line[0]["section"]} ) # Step 3: Process each line to create proper word tokens with accurate timing result = [] for line in lines: line_text = line["text"] section = line["section"] raw_tokens = line["tokens"] # Now we need to convert the raw tokens to proper word tokens # Split the line by spaces to get actual words words = line_text.split() if not words: continue # Map each word to its timing by finding its position in the raw tokens word_tokens = [] current_pos = 0 for word in words: # Find which raw tokens this word spans word_start = None word_end = None word_raw_tokens = [] # Find where this word appears in the line word_pos = line_text.find(word, current_pos) if word_pos >= 0: current_pos = word_pos + len(word) # Find which tokens contain this word char_pos = 0 for token in raw_tokens: token_len = len(token["text"]) token_end_pos = char_pos + token_len # Check if this token overlaps with the word if ( char_pos <= word_pos < token_end_pos or char_pos < word_pos + len(word) <= token_end_pos or word_pos <= char_pos < word_pos + len(word) ): word_raw_tokens.append(token) # Update timing if word_start is None or token["start_s"] < word_start: word_start = token["start_s"] if word_end is None or token["end_s"] > word_end: word_end = token["end_s"] char_pos += token_len # Fallback if we couldn't find tokens if word_start is None: # Use the timing from the first available token if raw_tokens: word_start = raw_tokens[0]["start_s"] word_end = raw_tokens[-1]["end_s"] else: # Last resort word_start = 0 word_end = 0.1 word_tokens.append({"text": word, "start_s": word_start, "end_s": word_end}) # Ensure timing monotonicity for i in range(1, len(word_tokens)): if word_tokens[i]["start_s"] < word_tokens[i - 1]["end_s"]: word_tokens[i]["start_s"] = word_tokens[i - 1]["end_s"] if word_tokens[i]["end_s"] <= word_tokens[i]["start_s"]: word_tokens[i]["end_s"] = word_tokens[i]["start_s"] + 0.05 # Create the final line object line_start = word_tokens[0]["start_s"] line_end = word_tokens[-1]["end_s"] result.append( { "text": line_text, "start_s": line_start, "end_s": line_end, "section": section, "words": word_tokens, } ) return result def main(): if len(sys.argv) != 2: print(f"Usage: {sys.argv[0]} ") sys.exit(1) hoot_path = Path(sys.argv[1]) if not hoot_path.exists(): print(f"Error: File {hoot_path} does not exist", file=sys.stderr) sys.exit(1) try: with open(hoot_path, 'r', encoding='utf-8') as f: hoot_data = json.load(f) # Ensure hoot_data is a list if not isinstance(hoot_data, list): print(f"Error: Expected a JSON list in {hoot_path}, found {type(hoot_data)}", file=sys.stderr) sys.exit(1) except json.JSONDecodeError as e: print(f"Error: File {hoot_path} is not valid JSON: {e}", file=sys.stderr) sys.exit(1) except Exception as e: print(f"Error reading file {hoot_path}: {e}", file=sys.stderr) sys.exit(1) # Process the lyrics using the simplified approach plain_text, line_timestamps_vtt, word_timestamps_vtt = process_lyrics_simple(hoot_data) # Determine output directory and base name output_dir = hoot_path.parent base_name = hoot_path.stem # Define output paths plain_path = output_dir / f"{base_name}_plain.txt" # Changed extension for clarity line_vtt_path = output_dir / f"{base_name}_line_timestamps.vtt" word_vtt_path = output_dir / f"{base_name}_word_timestamps.vtt" try: # Write plain text with open(plain_path, 'w', encoding='utf-8') as f: f.write(plain_text) # Write line-level timestamps VTT with open(line_vtt_path, 'w', encoding='utf-8') as f: f.write(line_timestamps_vtt) # Write word-level timestamps VTT with open(word_vtt_path, 'w', encoding='utf-8') as f: f.write(word_timestamps_vtt) print(f"Successfully processed {hoot_path}") print(f"Output files created in {output_dir}:") print(f"- {plain_path.name}") print(f"- {line_vtt_path.name}") print(f"- {word_vtt_path.name}") except IOError as e: print(f"Error writing output file: {e}", file=sys.stderr) sys.exit(1) if __name__ == "__main__": main()