Spaces:
Sleeping
Sleeping
| """ | |
| 🎵 Lyric Sync — Gradio Space | |
| Automatic perfect song lyric acquisition and synchronization. | |
| Pipeline: | |
| 1. Vocal separation (Demucs htdemucs) | |
| 2. Word-level transcription (Whisper large-v3) | |
| 3. Song identification (AcoustID / transcript search) | |
| 4. Lyrics fetching (LRCLIB) | |
| 5. Sequence alignment (transfer timings to correct lyrics) | |
| 6. Timing refinement (onset/offset detection) | |
| """ | |
| import json | |
| import logging | |
| import os | |
| import re | |
| import subprocess | |
| import tempfile | |
| import time | |
| import unicodedata | |
| from dataclasses import dataclass, field | |
| from difflib import SequenceMatcher | |
| from pathlib import Path | |
| from typing import Optional | |
| import gradio as gr | |
| import librosa | |
| import numpy as np | |
| import requests | |
| import spaces | |
| import torch | |
| import torchaudio | |
| logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") | |
| logger = logging.getLogger(__name__) | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # AUDIO LOADING (ffmpeg-based, handles all formats) | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| def load_audio_as_tensor(audio_path: str, target_sr: int = 44100) -> tuple[torch.Tensor, int]: | |
| """ | |
| Load any audio file into a torch tensor using librosa (ffmpeg backend). | |
| Returns: (tensor [channels, samples], sample_rate) | |
| Always returns stereo at target_sr. | |
| """ | |
| # librosa.load handles MP3, FLAC, OGG, WAV, etc. via soundfile/ffmpeg | |
| y, sr = librosa.load(audio_path, sr=target_sr, mono=False) | |
| # y shape: (samples,) if mono or (channels, samples) if stereo | |
| if y.ndim == 1: | |
| # Mono → stereo by duplicating | |
| wav = torch.from_numpy(y).float().unsqueeze(0).repeat(2, 1) | |
| else: | |
| wav = torch.from_numpy(y).float() | |
| if wav.shape[0] > 2: | |
| wav = wav[:2] | |
| return wav, target_sr | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # DATA CLASSES | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| class TimedWord: | |
| word: str | |
| start: float | |
| end: float | |
| confidence: float = 1.0 | |
| def duration(self) -> float: | |
| return self.end - self.start | |
| class SongInfo: | |
| title: str | |
| artist: str | |
| album: Optional[str] = None | |
| method: str = "unknown" | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # STEP 1: VOCAL SEPARATION (Demucs) | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| _demucs_model = None | |
| def get_demucs_model(device="cpu"): | |
| global _demucs_model | |
| if _demucs_model is None: | |
| from demucs.pretrained import get_model | |
| _demucs_model = get_model("htdemucs") | |
| _demucs_model.eval() | |
| logger.info("Demucs htdemucs loaded") | |
| _demucs_model.to(device) | |
| return _demucs_model | |
| def separate_vocals(audio_path: str, device: str = "cpu") -> tuple[np.ndarray, int, np.ndarray, int]: | |
| """ | |
| Separate vocals from audio. | |
| Returns: (vocals_16k_mono, 16000, vocals_44k_mono, 44100) | |
| """ | |
| from demucs.apply import apply_model | |
| model = get_demucs_model(device) | |
| # Load audio using librosa (handles MP3, FLAC, OGG, etc.) | |
| wav, sr = load_audio_as_tensor(audio_path, target_sr=44100) | |
| # Run separation | |
| wav_batch = wav.unsqueeze(0).to(device) | |
| with torch.no_grad(): | |
| sources = apply_model( | |
| model, wav_batch, device=device, | |
| shifts=1, split=True, overlap=0.25, progress=False, | |
| ) | |
| # sources: [1, 4, 2, N] — drums, bass, other, vocals | |
| vocal_idx = model.sources.index("vocals") | |
| vocals_stereo = sources[0, vocal_idx].cpu() # [2, N] at 44100 | |
| vocals_44k = vocals_stereo.mean(dim=0) # mono | |
| # Resample to 16kHz for Whisper | |
| vocals_16k = torchaudio.functional.resample( | |
| vocals_44k.unsqueeze(0), 44100, 16000 | |
| ).squeeze(0) | |
| return vocals_16k.numpy(), 16000, vocals_44k.numpy(), 44100 | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # STEP 2: TRANSCRIPTION (Whisper) | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| _whisper_pipe = None | |
| def get_whisper_pipeline(device="cpu"): | |
| global _whisper_pipe | |
| if _whisper_pipe is None: | |
| from transformers import pipeline as hf_pipeline | |
| _whisper_pipe = hf_pipeline( | |
| task="automatic-speech-recognition", | |
| model="openai/whisper-large-v3", | |
| dtype=torch.float16 if device == "cuda" else torch.float32, | |
| device=device, | |
| model_kwargs={"attn_implementation": "sdpa"}, | |
| ) | |
| logger.info(f"Whisper large-v3 pipeline loaded on {device}") | |
| return _whisper_pipe | |
| def transcribe_vocals(audio: np.ndarray, sr: int = 16000) -> list[TimedWord]: | |
| pipe = get_whisper_pipeline("cuda" if torch.cuda.is_available() else "cpu") | |
| result = pipe( | |
| {"array": audio.astype(np.float32), "sampling_rate": sr}, | |
| return_timestamps="word", | |
| generate_kwargs={ | |
| "language": "english", | |
| "task": "transcribe", | |
| }, | |
| chunk_length_s=30, | |
| stride_length_s=5, | |
| ) | |
| words = [] | |
| for chunk in result.get("chunks", []): | |
| text = chunk["text"].strip() | |
| ts = chunk.get("timestamp", (None, None)) | |
| if text and ts[0] is not None and ts[1] is not None: | |
| words.append(TimedWord(word=text, start=ts[0], end=ts[1])) | |
| return words | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # STEP 3: SONG IDENTIFICATION | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| def identify_song(audio_path, user_artist="", user_title="", transcript_text=""): | |
| if user_artist.strip() and user_title.strip(): | |
| return SongInfo(title=user_title.strip(), artist=user_artist.strip(), method="user_provided") | |
| acoustid_key = os.environ.get("ACOUSTID_API_KEY") | |
| if acoustid_key: | |
| result = _acoustid_identify(audio_path, acoustid_key) | |
| if result: | |
| return result | |
| if transcript_text: | |
| words = transcript_text.split() | |
| if len(words) >= 5: | |
| mid = len(words) // 2 | |
| fragment = " ".join(words[max(0, mid-5):mid+5]) | |
| result = _search_lrclib(fragment) | |
| if result: | |
| return result | |
| return None | |
| def _acoustid_identify(audio_path, api_key): | |
| try: | |
| result = subprocess.run( | |
| ["fpcalc", "-json", "-length", "120", audio_path], | |
| capture_output=True, text=True, timeout=30, | |
| ) | |
| if result.returncode != 0: | |
| return None | |
| fp_data = json.loads(result.stdout) | |
| resp = requests.post("https://api.acoustid.org/v2/lookup", data={ | |
| "client": api_key, "duration": fp_data["duration"], | |
| "fingerprint": fp_data["fingerprint"], | |
| "meta": "recordings releasegroups", "format": "json", | |
| }, timeout=15) | |
| resp.raise_for_status() | |
| data = resp.json() | |
| if data.get("status") != "ok" or not data.get("results"): | |
| return None | |
| best = max(data["results"], key=lambda r: r.get("score", 0)) | |
| if best.get("score", 0) < 0.5 or not best.get("recordings"): | |
| return None | |
| rec = best["recordings"][0] | |
| artist = rec.get("artists", [{}])[0].get("name", "Unknown") | |
| album = rec.get("releasegroups", [{}])[0].get("title") if rec.get("releasegroups") else None | |
| return SongInfo(title=rec.get("title", "Unknown"), artist=artist, album=album, method="acoustid") | |
| except Exception as e: | |
| logger.warning(f"AcoustID failed: {e}") | |
| return None | |
| def _search_lrclib(query): | |
| try: | |
| resp = requests.get("https://lrclib.net/api/search", params={"q": query}, timeout=10) | |
| if resp.status_code == 200: | |
| results = resp.json() | |
| if results: | |
| best = results[0] | |
| return SongInfo(title=best.get("trackName", "Unknown"), artist=best.get("artistName", "Unknown"), album=best.get("albumName"), method="lrclib_search") | |
| except Exception as e: | |
| logger.debug(f"LRCLIB search failed: {e}") | |
| return None | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # STEP 4: LYRICS FETCHING | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| def fetch_lyrics(song): | |
| params = {"artist_name": song.artist, "track_name": song.title} | |
| if song.album: | |
| params["album_name"] = song.album | |
| try: | |
| resp = requests.get("https://lrclib.net/api/get", params=params, timeout=10) | |
| if resp.status_code == 200: | |
| data = resp.json() | |
| plain = data.get("plainLyrics") or data.get("syncedLyrics", "") | |
| if plain: | |
| plain = re.sub(r"\[\d{2}:\d{2}\.\d{2,3}\]\s*", "", plain) | |
| words = plain.split() | |
| if words: | |
| return plain, words | |
| except Exception as e: | |
| logger.warning(f"LRCLIB fetch failed: {e}") | |
| try: | |
| resp = requests.get("https://lrclib.net/api/search", params={"q": f"{song.artist} {song.title}"}, timeout=10) | |
| if resp.status_code == 200: | |
| results = resp.json() | |
| if results: | |
| plain = results[0].get("plainLyrics") or results[0].get("syncedLyrics", "") | |
| plain = re.sub(r"\[\d{2}:\d{2}\.\d{2,3}\]\s*", "", plain) | |
| words = plain.split() | |
| if words: | |
| return plain, words | |
| except Exception as e: | |
| logger.debug(f"LRCLIB search fallback failed: {e}") | |
| return None | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # STEP 5: SEQUENCE ALIGNMENT | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| def normalize_word(word): | |
| word = unicodedata.normalize("NFKD", word).lower() | |
| word = re.sub(r"[^\w']", "", word).strip("'") | |
| return word | |
| def align_words(asr_words, ref_words): | |
| if not asr_words or not ref_words: | |
| return asr_words or [] | |
| asr_norm = [normalize_word(w.word) for w in asr_words] | |
| ref_norm = [normalize_word(w) for w in ref_words] | |
| asr_valid = [(i, n) for i, n in enumerate(asr_norm) if n] | |
| ref_valid = [(i, n) for i, n in enumerate(ref_norm) if n] | |
| if not asr_valid or not ref_valid: | |
| return asr_words | |
| asr_indices, asr_normalized = zip(*asr_valid) | |
| ref_indices, ref_normalized = zip(*ref_valid) | |
| ref_set = set(ref_normalized) | |
| asr_for_matching = [] | |
| for aw in asr_normalized: | |
| if aw in ref_set: | |
| asr_for_matching.append(aw) | |
| else: | |
| best_match, best_ratio = aw, 0.0 | |
| for rw in ref_set: | |
| if abs(len(aw) - len(rw)) > max(len(aw), len(rw)) * 0.4: | |
| continue | |
| ratio = SequenceMatcher(None, aw, rw).ratio() | |
| if ratio > best_ratio: | |
| best_ratio, best_match = ratio, rw | |
| asr_for_matching.append(best_match if best_ratio >= 0.75 else aw) | |
| sm = SequenceMatcher(None, asr_for_matching, list(ref_normalized), autojunk=False) | |
| opcodes = sm.get_opcodes() | |
| result = [TimedWord(word=w, start=0.0, end=0.0, confidence=0.0) for w in ref_words] | |
| for tag, i1, i2, j1, j2 in opcodes: | |
| if tag == "equal": | |
| for asr_pos, ref_pos in zip(range(i1, i2), range(j1, j2)): | |
| orig_asr, orig_ref = asr_indices[asr_pos], ref_indices[ref_pos] | |
| result[orig_ref] = TimedWord(word=ref_words[orig_ref], start=asr_words[orig_asr].start, end=asr_words[orig_asr].end, confidence=asr_words[orig_asr].confidence) | |
| elif tag == "replace": | |
| t_start = asr_words[asr_indices[i1]].start | |
| t_end = asr_words[asr_indices[i2 - 1]].end | |
| n_ref = j2 - j1 | |
| duration = t_end - t_start | |
| for k, ref_pos in enumerate(range(j1, j2)): | |
| orig_ref = ref_indices[ref_pos] | |
| result[orig_ref] = TimedWord(word=ref_words[orig_ref], start=t_start + k * duration / n_ref, end=t_start + (k + 1) * duration / n_ref, confidence=0.5) | |
| _fill_gaps(result) | |
| return result | |
| def _fill_gaps(words): | |
| for i in range(len(words)): | |
| if words[i].start > 0 or words[i].end > 0: | |
| continue | |
| prev_end, next_start = 0.0, None | |
| for j in range(i - 1, -1, -1): | |
| if words[j].end > 0: | |
| prev_end = words[j].end | |
| break | |
| for j in range(i + 1, len(words)): | |
| if words[j].start > 0: | |
| next_start = words[j].start | |
| break | |
| if next_start is None: | |
| next_start = prev_end + 0.3 | |
| gap_count = 0 | |
| for j in range(i, len(words)): | |
| if words[j].start > 0 or words[j].end > 0: | |
| break | |
| gap_count += 1 | |
| position = 0 | |
| for j in range(i - 1, -1, -1): | |
| if words[j].start > 0 or words[j].end > 0: | |
| break | |
| position += 1 | |
| t_per_word = (next_start - prev_end) / max(gap_count, 1) | |
| words[i].start = prev_end + position * t_per_word | |
| words[i].end = prev_end + (position + 1) * t_per_word | |
| words[i].confidence = 0.3 | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # STEP 6: TIMING REFINEMENT | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| def refine_timings(vocals_44k, words, sr=44100, hop_length=256): | |
| if len(vocals_44k) == 0 or not words: | |
| return words | |
| odf = librosa.onset.onset_strength(y=vocals_44k, sr=sr, hop_length=hop_length, n_fft=1024, fmin=80.0, fmax=4000.0, aggregate=np.median, detrend=True) | |
| onset_frames = librosa.onset.onset_detect(onset_envelope=odf, sr=sr, hop_length=hop_length, backtrack=True, units='frames', pre_max=2, post_max=2, pre_avg=2, post_avg=4, delta=0.05, wait=8) | |
| rms = librosa.feature.rms(y=vocals_44k, frame_length=1024, hop_length=hop_length)[0] | |
| rms_smooth = np.convolve(rms, np.ones(7) / 7, mode='same') | |
| search_frames = int(0.08 * sr / hop_length) | |
| refined = [] | |
| for word in words: | |
| w = TimedWord(word=word.word, start=word.start, end=word.end, confidence=word.confidence) | |
| approx_frame = librosa.time_to_frames(w.start, sr=sr, hop_length=hop_length) | |
| lo, hi = max(0, approx_frame - search_frames), min(len(odf) - 1, approx_frame + search_frames) | |
| candidates = onset_frames[(onset_frames >= lo) & (onset_frames <= hi)] | |
| if len(candidates) > 0: | |
| w.start = librosa.frames_to_time(candidates[np.argmin(np.abs(candidates - approx_frame))], sr=sr, hop_length=hop_length) | |
| end_frame = librosa.time_to_frames(w.end, sr=sr, hop_length=hop_length) | |
| end_search = int(0.05 * sr / hop_length) | |
| elo, ehi = max(0, end_frame - end_search), min(len(rms_smooth) - 1, end_frame + end_search) | |
| if elo < ehi: | |
| rms_db = librosa.amplitude_to_db(rms_smooth[elo:ehi + 1] + 1e-10, ref=rms_smooth.max() + 1e-10) | |
| silent = np.where(rms_db < -40.0)[0] | |
| if len(silent) > 0: | |
| w.end = librosa.frames_to_time(elo + silent[0], sr=sr, hop_length=hop_length) | |
| if w.end <= w.start + 0.03: | |
| w.end = w.start + 0.03 | |
| refined.append(w) | |
| for i in range(len(refined) - 1): | |
| if refined[i].end > refined[i + 1].start: | |
| mid = (refined[i].end + refined[i + 1].start) / 2 | |
| refined[i] = TimedWord(word=refined[i].word, start=refined[i].start, end=mid, confidence=refined[i].confidence) | |
| refined[i + 1] = TimedWord(word=refined[i + 1].word, start=mid, end=refined[i + 1].end, confidence=refined[i + 1].confidence) | |
| return refined | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # OUTPUT FORMATTERS | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| def _lrc_ts(seconds): | |
| return f"{int(seconds // 60):02d}:{seconds % 60:05.2f}" | |
| def _srt_ts(seconds): | |
| h, m, s = int(seconds // 3600), int((seconds % 3600) // 60), seconds % 60 | |
| return f"{h:02d}:{m:02d}:{int(s):02d},{int((s % 1) * 1000):03d}" | |
| def format_enhanced_lrc(words, line_gap=1.0): | |
| if not words: return "" | |
| lines, current_line = [], [] | |
| for word in words: | |
| if current_line and word.start - current_line[-1].end > line_gap: | |
| lines.append(f"[{_lrc_ts(current_line[0].start)}] " + " ".join(f"<{_lrc_ts(w.start)}> {w.word}" for w in current_line) + f" <{_lrc_ts(current_line[-1].end)}>") | |
| current_line = [] | |
| current_line.append(word) | |
| if current_line: | |
| lines.append(f"[{_lrc_ts(current_line[0].start)}] " + " ".join(f"<{_lrc_ts(w.start)}> {w.word}" for w in current_line) + f" <{_lrc_ts(current_line[-1].end)}>") | |
| return "\n".join(lines) | |
| def format_standard_lrc(words, line_gap=1.0): | |
| if not words: return "" | |
| lines, current_line = [], [] | |
| for word in words: | |
| if current_line and word.start - current_line[-1].end > line_gap: | |
| lines.append(f"[{_lrc_ts(current_line[0].start)}] " + " ".join(w.word for w in current_line)) | |
| current_line = [] | |
| current_line.append(word) | |
| if current_line: | |
| lines.append(f"[{_lrc_ts(current_line[0].start)}] " + " ".join(w.word for w in current_line)) | |
| return "\n".join(lines) | |
| def format_srt(words, line_gap=1.0, max_words=10): | |
| if not words: return "" | |
| entries, current_line = [], [] | |
| for word in words: | |
| if current_line and (word.start - current_line[-1].end > line_gap or len(current_line) >= max_words): | |
| entries.append(current_line) | |
| current_line = [] | |
| current_line.append(word) | |
| if current_line: entries.append(current_line) | |
| return "\n".join(f"{idx}\n{_srt_ts(lw[0].start)} --> {_srt_ts(lw[-1].end)}\n{' '.join(w.word for w in lw)}\n" for idx, lw in enumerate(entries, 1)) | |
| def format_json(words): | |
| return [{"word": w.word, "start": round(w.start, 3), "end": round(w.end, 3), "confidence": round(w.confidence, 3)} for w in words] | |
| def format_ass(words, line_gap=1.0): | |
| header = "[Script Info]\nTitle: Lyric Sync\nScriptType: v4.00+\nPlayResX: 1920\nPlayResY: 1080\n\n[V4+ Styles]\nFormat: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding\nStyle: Default,Arial,48,&H00FFFFFF,&H000000FF,&H00000000,&H64000000,-1,0,0,0,100,100,0,0,1,2,1,2,10,10,40,1\n\n[Events]\nFormat: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text\n" | |
| line_groups, current = [], [] | |
| for word in words: | |
| if current and word.start - current[-1].end > line_gap: | |
| line_groups.append(current) | |
| current = [] | |
| current.append(word) | |
| if current: line_groups.append(current) | |
| def ass_ts(s): return f"{int(s//3600)}:{int((s%3600)//60):02d}:{s%60:05.2f}" | |
| events = [] | |
| for lw in line_groups: | |
| karaoke = " ".join(f"{{\\kf{int(w.duration * 100)}}}{w.word}" for w in lw) | |
| events.append(f"Dialogue: 0,{ass_ts(lw[0].start)},{ass_ts(lw[-1].end)},Default,,0,0,0,,{karaoke}") | |
| return header + "\n".join(events) | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # MAIN PIPELINE | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| def run_pipeline(audio_path: str, artist: str, title: str, progress=gr.Progress()): | |
| if audio_path is None: | |
| raise gr.Error("Please upload an audio file.") | |
| start_time = time.time() | |
| status_log = [] | |
| def log(msg): | |
| status_log.append(msg) | |
| logger.info(msg) | |
| device = "cuda" if torch.cuda.is_available() else "cpu" | |
| log(f"🖥️ Using device: {device}") | |
| progress(0.05, desc="🎵 Separating vocals (Demucs htdemucs)...") | |
| log("🎵 Separating vocals with Demucs htdemucs...") | |
| vocals_16k, sr_16k, vocals_44k, sr_44k = separate_vocals(audio_path, device=device) | |
| log(f" ✓ Vocals: {len(vocals_16k)/sr_16k:.1f}s extracted") | |
| progress(0.30, desc="📝 Transcribing vocals (Whisper large-v3)...") | |
| log("📝 Transcribing with Whisper large-v3 (word-level timestamps)...") | |
| asr_words = transcribe_vocals(vocals_16k, sr_16k) | |
| transcript_text = " ".join(w.word for w in asr_words) | |
| log(f" ✓ Transcribed: {len(asr_words)} words") | |
| progress(0.55, desc="🔍 Identifying song...") | |
| log("🔍 Identifying song...") | |
| song = identify_song(audio_path, artist, title, transcript_text) | |
| if song: | |
| log(f" ✓ Identified: {song.artist} — {song.title} (via {song.method})") | |
| else: | |
| log(" ⚠ Could not identify song — using raw transcription") | |
| ref_words = None | |
| if song: | |
| progress(0.65, desc="📖 Fetching reference lyrics (LRCLIB)...") | |
| log("📖 Fetching lyrics from LRCLIB...") | |
| lyrics_result = fetch_lyrics(song) | |
| if lyrics_result: | |
| _, ref_words = lyrics_result | |
| log(f" ✓ Lyrics: {len(ref_words)} words") | |
| else: | |
| log(" ⚠ No lyrics found — using raw transcription") | |
| if ref_words: | |
| progress(0.75, desc="🔗 Aligning transcript to reference lyrics...") | |
| log("🔗 Aligning ASR transcript to reference lyrics...") | |
| synced_words = align_words(asr_words, ref_words) | |
| matched = sum(1 for w in synced_words if w.confidence >= 0.8) | |
| total = len(synced_words) | |
| log(f" ✓ Aligned: {matched}/{total} direct matches ({matched/total*100:.0f}%)" if total else " ✓ Aligned") | |
| else: | |
| synced_words = asr_words | |
| log(" Using raw ASR words (no reference lyrics)") | |
| progress(0.88, desc="⚙️ Refining timings (onset/offset detection)...") | |
| log("⚙️ Refining timings via audio analysis (onset detection, librosa)...") | |
| synced_words = refine_timings(vocals_44k, synced_words, sr=sr_44k) | |
| log(" ✓ Timing refinement complete") | |
| progress(0.96, desc="📄 Formatting outputs...") | |
| lrc_enhanced = format_enhanced_lrc(synced_words) | |
| lrc_standard = format_standard_lrc(synced_words) | |
| srt_text = format_srt(synced_words) | |
| json_data = format_json(synced_words) | |
| ass_text = format_ass(synced_words) | |
| def write_tmp(content, suffix): | |
| f = tempfile.NamedTemporaryFile(suffix=suffix, delete=False, mode="w", encoding="utf-8") | |
| f.write(content); f.close(); return f.name | |
| lrc_file = write_tmp(lrc_enhanced, ".lrc") | |
| srt_file = write_tmp(srt_text, ".srt") | |
| json_file = write_tmp(json.dumps(json_data, indent=2, ensure_ascii=False), ".json") | |
| ass_file = write_tmp(ass_text, ".ass") | |
| elapsed = time.time() - start_time | |
| log(f"\n✅ Done in {elapsed:.1f}s — {len(synced_words)} words synchronized") | |
| progress(1.0, desc="✅ Done!") | |
| return ("\n".join(status_log), lrc_enhanced, lrc_file, lrc_standard, srt_text, srt_file, json_data, json_file, ass_text, ass_file) | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| # GRADIO UI | |
| # ═══════════════════════════════════════════════════════════════════════════════ | |
| with gr.Blocks(title="🎵 Lyric Sync", css=".main-title{text-align:center;margin-bottom:.5em}.subtitle{text-align:center;color:#666;margin-bottom:1.5em}") as demo: | |
| gr.HTML("<h1 class='main-title'>🎵 Lyric Sync</h1>") | |
| gr.HTML("<p class='subtitle'>Automatic perfect song lyric acquisition and synchronization</p>") | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| audio_input = gr.Audio(sources=["upload"], type="filepath", label="Upload Audio") | |
| gr.Markdown("### Song Metadata (optional)\n*Provide artist & title to skip identification, or leave blank for auto-detection.*") | |
| artist_input = gr.Textbox(label="Artist", placeholder="e.g. Radiohead") | |
| title_input = gr.Textbox(label="Song Title", placeholder="e.g. Creep") | |
| run_btn = gr.Button("▶ Sync Lyrics", variant="primary", size="lg") | |
| status_output = gr.Textbox(label="Pipeline Status", lines=12, interactive=False, show_copy_button=True) | |
| with gr.Column(scale=2): | |
| with gr.Tabs(): | |
| with gr.Tab("📝 Enhanced LRC (word-level)"): | |
| lrc_enhanced_output = gr.Textbox(label="Enhanced LRC", lines=20, show_copy_button=True, interactive=False) | |
| lrc_download = gr.File(label="⬇ Download .lrc") | |
| with gr.Tab("📄 Standard LRC (line-level)"): | |
| lrc_standard_output = gr.Textbox(label="Standard LRC", lines=20, show_copy_button=True, interactive=False) | |
| with gr.Tab("🎬 SRT Subtitles"): | |
| srt_output = gr.Textbox(label="SRT", lines=20, show_copy_button=True, interactive=False) | |
| srt_download = gr.File(label="⬇ Download .srt") | |
| with gr.Tab("📊 JSON"): | |
| json_output = gr.JSON(label="Word-level JSON") | |
| json_download = gr.File(label="⬇ Download .json") | |
| with gr.Tab("🎤 ASS Karaoke"): | |
| ass_output = gr.Textbox(label="ASS (karaoke \\k tags)", lines=20, show_copy_button=True, interactive=False) | |
| ass_download = gr.File(label="⬇ Download .ass") | |
| run_btn.click(fn=run_pipeline, inputs=[audio_input, artist_input, title_input], outputs=[status_output, lrc_enhanced_output, lrc_download, lrc_standard_output, srt_output, srt_download, json_output, json_download, ass_output, ass_download]) | |
| gr.Markdown("---\n### How it works\n| Step | What | Technology |\n|------|------|---|\n| 1 | Vocal separation | Demucs htdemucs (~9.2 dB SDR) |\n| 2 | Word-level transcription | Whisper large-v3 |\n| 3 | Song identification | AcoustID / LRCLIB text search |\n| 4 | Lyrics acquisition | LRCLIB (free, no auth) |\n| 5 | Alignment | Sequence alignment (LCS + fuzzy matching) |\n| 6 | Timing refinement | Onset/offset detection (librosa, 5.8ms resolution) |\n\n**Tips:** Provide artist & title for best results. Supported: MP3, WAV, FLAC, OGG, M4A. Processing: ~30-90s.") | |
| if __name__ == "__main__": | |
| demo.launch() | |