#!/usr/bin/env python3 """ TTS Dataset Builder — Professional Gradio Space YouTube playlist/video -> VAD segmentation -> Qwen3-ASR -> HuggingFace dataset With checkpoint/resume support for crash recovery. """ import os import json import glob import shutil import hashlib import logging import subprocess from pathlib import Path from concurrent.futures import ThreadPoolExecutor, as_completed import gradio as gr import numpy as np import soundfile as sf import torch try: import spaces IS_HF_SPACE = True except ImportError: IS_HF_SPACE = False logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") logger = logging.getLogger(__name__) SAMPLE_RATE = 16000 MIN_SEGMENT_SEC = 8.0 MAX_SEGMENT_SEC = 12.0 TARGET_SEGMENT_SEC = 10.0 HARD_MAX_SEC = 15.0 # /data kalici (persistent storage), yoksa home dizini if os.path.isdir("/data"): WORK_ROOT = "/data/tts_sessions" else: WORK_ROOT = os.path.join(os.path.expanduser("~"), "tts_sessions") os.makedirs(WORK_ROOT, exist_ok=True) # =========================================================================== # SESSION / CHECKPOINT # =========================================================================== def _session_dir(hf_repo): """HF repo adina gore sabit calisma dizini. Sayfa yenilense bile ayni.""" slug = hashlib.md5(hf_repo.strip().encode()).hexdigest()[:12] d = os.path.join(WORK_ROOT, slug) os.makedirs(d, exist_ok=True) return d def _load_checkpoint(session_dir): cp_path = os.path.join(session_dir, "checkpoint.json") if os.path.exists(cp_path): with open(cp_path) as f: return json.load(f) return {"phase": "idle", "downloaded_videos": [], "processed_files": [], "segments": [], "results": [], "config": {}} def _save_checkpoint(session_dir, checkpoint): cp_path = os.path.join(session_dir, "checkpoint.json") with open(cp_path, "w") as f: json.dump(checkpoint, f, ensure_ascii=False) def _cleanup_session(session_dir): try: shutil.rmtree(session_dir, ignore_errors=True) except Exception: pass # =========================================================================== # 1) VIDEO LISTELEME # =========================================================================== def list_videos(youtube_url, cookies_text): if not youtube_url.strip(): return gr.CheckboxGroup(choices=[], value=[]), "Please enter a URL." cookies_path = _save_cookies(cookies_text) cmd = [ "yt-dlp", "--flat-playlist", "--print", "%(id)s\t%(title)s\t%(duration_string)s", "--ignore-errors", "--no-check-formats", "--skip-download", "--remote-components", "ejs:github", ] if cookies_path: cmd += ["--cookies", cookies_path] cmd.append(youtube_url.strip()) try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=180) lines = [l.strip() for l in result.stdout.strip().split("\n") if l.strip()] except Exception as e: return gr.CheckboxGroup(choices=[], value=[]), f"Error: {e}" if not lines: stderr_msg = result.stderr[:1000] if result.stderr else "" return gr.CheckboxGroup(choices=[], value=[]), f"No videos found.\n{stderr_msg}" choices = [] for line in lines: if "\t" in line: parts = line.split("\t") else: parts = line.split("\\t") if len(parts) >= 3: vid, title, dur = parts[0].strip(), parts[1].strip(), parts[2].strip() choices.append((f"[{dur}] {title}", vid)) elif len(parts) >= 2: choices.append((parts[1].strip(), parts[0].strip())) return ( gr.CheckboxGroup(choices=choices, value=[c[1] for c in choices]), f"{len(choices)} videos found.", ) # =========================================================================== # 2) PIPELINE — checkpoint destekli # =========================================================================== def run_pipeline( youtube_url, cookies_text, selected_videos, hf_repo, hf_token, speaker_id, trim_start, trim_end, use_deepfilter, sleep_interval, language, progress=gr.Progress(track_tqdm=False), ): if not youtube_url.strip(): yield "Please enter a YouTube URL." return if not hf_repo.strip(): yield "Please enter a HuggingFace repository name." return if not selected_videos: yield "Please select at least one video." return token = hf_token.strip() or os.environ.get("HF_TOKEN", "") if not token: yield "Please provide an HF Token or add HF_TOKEN to Space Secrets." return session_dir = _session_dir(hf_repo) audio_dir = os.path.join(session_dir, "audio") segments_dir = os.path.join(session_dir, "segments") os.makedirs(audio_dir, exist_ok=True) os.makedirs(segments_dir, exist_ok=True) cp = _load_checkpoint(session_dir) cp["config"] = { "hf_repo": hf_repo.strip(), "speaker_id": speaker_id, "trim_start": trim_start, "trim_end": trim_end, "language": language, "use_deepfilter": use_deepfilter, } _save_checkpoint(session_dir, cp) cookies_path = _save_cookies(cookies_text) # ---- PHASE 1: INDIRME (checkpoint destekli) ---- # Hem checkpoint'taki kaydi hem diskteki dosyalari kontrol et already_downloaded = set(cp.get("downloaded_videos", [])) existing_on_disk = set() for ext in ("*.opus", "*.m4a", "*.mp3", "*.wav", "*.ogg", "*.flac", "*.webm"): for f in glob.glob(os.path.join(audio_dir, ext)): vid_id = Path(f).stem existing_on_disk.add(vid_id) already_downloaded = already_downloaded | existing_on_disk to_download = [v for v in selected_videos if v not in already_downloaded] if to_download: yield f"[1/5] Downloading ({len(already_downloaded)} cached, {len(to_download)} remaining)..." for i, vid in enumerate(to_download): yield f"[1/5] Downloading {i+1}/{len(to_download)} — {vid}" url = f"https://www.youtube.com/watch?v={vid}" cmd = [ "yt-dlp", url, "-f", "bestaudio/best", "--extract-audio", "--remote-components", "ejs:github", "--concurrent-fragments", "4", "-o", os.path.join(audio_dir, "%(id)s.%(ext)s"), "--no-warnings", ] if cookies_path: cmd += ["--cookies", cookies_path] if sleep_interval > 0 and i < len(to_download) - 1: cmd += ["--sleep-interval", str(int(sleep_interval))] try: subprocess.run(cmd, capture_output=True, text=True, timeout=600) cp["downloaded_videos"].append(vid) cp["phase"] = "downloading" _save_checkpoint(session_dir, cp) except Exception as e: yield f"[1/5] Download error ({vid}): {e}" else: yield f"[1/5] All videos already downloaded ({len(already_downloaded)}). Skipping..." audio_files = [] for ext in ("*.opus", "*.m4a", "*.mp3", "*.wav", "*.ogg", "*.flac", "*.webm"): audio_files.extend(glob.glob(os.path.join(audio_dir, ext))) audio_files = sorted(set(audio_files)) if not audio_files: yield "No files downloaded. Please check your cookies." return yield f"[1/5] Done: {len(audio_files)} files ready." # ---- PHASE 2: FFMPEG TRIM + WAV ---- already_processed = set(cp.get("processed_files", [])) to_process = [f for f in audio_files if f not in already_processed] if to_process: yield f"[2/5] Trimming & converting ({len(to_process)} files)..." wav_dir = os.path.join(session_dir, "wavs") os.makedirs(wav_dir, exist_ok=True) def _convert_one(args): input_path, idx = args wav_path = os.path.join(wav_dir, f"{idx:04d}.wav") try: probe = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", input_path], capture_output=True, text=True, timeout=30, ) duration = float(probe.stdout.strip()) total_trim = trim_start + trim_end if duration > total_trim + 10: cmd = ["ffmpeg", "-y", "-i", input_path, "-ss", str(trim_start)] if trim_end > 0: cmd += ["-to", str(duration - trim_end)] cmd += ["-ar", str(SAMPLE_RATE), "-ac", "1", "-f", "wav", "-loglevel", "error", wav_path] subprocess.run(cmd, capture_output=True, check=True, timeout=300) else: subprocess.run( ["ffmpeg", "-y", "-i", input_path, "-ar", str(SAMPLE_RATE), "-ac", "1", "-f", "wav", "-loglevel", "error", wav_path], capture_output=True, check=True, timeout=300, ) return (input_path, idx, wav_path, True) except Exception as e: logger.error(f"[CONV] FAIL: {e}") return (input_path, idx, wav_path, False) converted = [] with ThreadPoolExecutor(max_workers=8) as ex: futs = [ex.submit(_convert_one, (f, i)) for i, f in enumerate(to_process)] for fut in as_completed(futs): inp, idx, wpath, ok = fut.result() if ok: converted.append((inp, idx, wpath)) cp["processed_files"].append(inp) converted.sort(key=lambda x: x[1]) cp["phase"] = "trimmed" _save_checkpoint(session_dir, cp) yield f"[2/5] Done: {len(converted)} files converted." else: yield "[2/5] All files already converted. Skipping..." wav_dir = os.path.join(session_dir, "wavs") converted = [] if os.path.exists(wav_dir): for idx, wf in enumerate(sorted(glob.glob(os.path.join(wav_dir, "*.wav")))): converted.append(("", idx, wf)) # ---- PHASE 3: VAD ---- existing_segments = glob.glob(os.path.join(segments_dir, "*.wav")) if not existing_segments and converted: yield "[3/5] VAD segmentation..." all_seg_infos = _run_vad_on_files(converted, segments_dir) # WAV dosyalarini temizle — segmentler olusturuldu wav_dir = os.path.join(session_dir, "wavs") if os.path.exists(wav_dir): shutil.rmtree(wav_dir, ignore_errors=True) if not all_seg_infos: yield "[3/5] No segments created. Check trim settings." return cp["segments"] = all_seg_infos cp["phase"] = "segmented" _save_checkpoint(session_dir, cp) yield f"[3/5] Done: {len(all_seg_infos)} segments." elif existing_segments: yield f"[3/5] {len(existing_segments)} segments cached. Skipping..." all_seg_infos = cp.get("segments", []) if not all_seg_infos: all_seg_infos = [{"seg_path": p, "duration": 0, "source_file": ""} for p in sorted(existing_segments)] else: yield "No segments could be created." return if not all_seg_infos: yield "No segments found." return # ---- PHASE 3.5: DEEPFILTERNET (opsiyonel) ---- if use_deepfilter and cp.get("phase") != "enhanced": yield f"[3.5/5] DeepFilterNet enhancement ({len(all_seg_infos)} segments)..." try: from df.enhance import enhance, init_df, load_audio, save_audio df_model, df_state, _ = init_df() count = 0 for seg_info in all_seg_infos: try: audio_df, _ = load_audio(seg_info["seg_path"], sr=df_state.sr()) enhanced_audio = enhance(df_model, df_state, audio_df) save_audio(seg_info["seg_path"], enhanced_audio, sr=df_state.sr()) count += 1 if count % 50 == 0: yield f"Phase 3.5: DeepFilter {count}/{len(all_seg_infos)}" except Exception as e: logger.warning(f"[DF] Skip: {e}") del df_model, df_state torch.cuda.empty_cache() cp["phase"] = "enhanced" _save_checkpoint(session_dir, cp) yield f"[3.5/5] Done: {count} segments enhanced." except ImportError: yield "[3.5/5] DeepFilterNet not installed. Skipping." # ---- PHASE 4: QWEN3-ASR ---- existing_results = cp.get("results", []) if not existing_results: yield f"[4/5] Transcribing ({len(all_seg_infos)} segments)..." try: all_results = _transcribe_segments(all_seg_infos, speaker_id, language=language) except Exception as e: yield f"[4/5] Transcription error: {e}" return cp["results"] = all_results cp["phase"] = "transcribed" _save_checkpoint(session_dir, cp) yield f"[4/5] Done: {len(all_results)} transcriptions." else: all_results = existing_results yield f"[4/5] {len(all_results)} transcriptions cached. Skipping..." if not all_results: yield "No transcriptions produced." return # ---- PHASE 5: HF PUSH ---- yield f"[5/5] Pushing to HuggingFace ({hf_repo})..." try: from datasets import Dataset, Audio as HfAudio ds = Dataset.from_dict({ "audio": [r["audio_path"] for r in all_results], "text": [r["text"] for r in all_results], "speaker_id": [r["speaker_id"] for r in all_results], }) ds = ds.cast_column("audio", HfAudio(sampling_rate=SAMPLE_RATE)) ds.push_to_hub(hf_repo.strip(), token=token, private=False) yield f"Done! {len(all_results)} segments pushed to {hf_repo}." _cleanup_session(session_dir) except Exception as e: yield f"Push failed: {e}\nProgress is saved — retry to continue." # =========================================================================== # ASR (Zero GPU uyumlu) # =========================================================================== def _run_vad_on_files(converted, segments_dir): """Enerji tabanli VAD — hicbir GPU/CUDA bagimliligina ihtiyac duymaz.""" all_seg_infos = [] for ci, (orig, fidx, wav_path) in enumerate(converted): try: audio = _read_audio_sf(wav_path, sampling_rate=SAMPLE_RATE) audio_np = audio.numpy() timestamps = _energy_vad(audio_np, sr=SAMPLE_RATE) segments = _vad_segment(timestamps, audio_np, len(audio)) for si, (s, e) in enumerate(segments): seg = audio_np[s:e] dur = len(seg) / SAMPLE_RATE if dur < 1.0: continue seg_path = os.path.join(segments_dir, f"f{fidx:04d}_s{si:04d}.wav") sf.write(seg_path, seg, SAMPLE_RATE) all_seg_infos.append({ "seg_path": seg_path, "duration": round(dur, 2), "source_file": Path(orig).name if orig else "", }) except Exception as e: logger.error(f"[VAD] ERR: {e}") return all_seg_infos def _energy_vad(audio_np, sr=16000, frame_ms=30, energy_threshold=0.01, min_speech_ms=250, min_silence_ms=300): """Enerji tabanli basit VAD. Silero benzeri cikti uretir: [{'start': sample, 'end': sample}, ...]""" frame_size = int(sr * frame_ms / 1000) hop = frame_size // 2 n_frames = (len(audio_np) - frame_size) // hop + 1 # Her frame icin RMS enerji hesapla energies = np.zeros(n_frames) for i in range(n_frames): start = i * hop frame = audio_np[start:start + frame_size] energies[i] = np.sqrt(np.mean(frame ** 2)) # Adaptif esik: medyan enerjinin 2 kati veya sabit esik adaptive_threshold = max(np.median(energies) * 2, energy_threshold) # Konusma/sessizlik etiketleme is_speech = energies > adaptive_threshold # Kisa bosluk/konusma temizleme min_speech_frames = int(min_speech_ms / frame_ms) min_silence_frames = int(min_silence_ms / frame_ms) # Kisa sessizlikleri doldur i = 0 while i < len(is_speech): if not is_speech[i]: j = i while j < len(is_speech) and not is_speech[j]: j += 1 if j - i < min_silence_frames and i > 0 and j < len(is_speech): is_speech[i:j] = True i = j else: i += 1 # Konusma segmentlerini bul timestamps = [] in_speech = False speech_start = 0 for i in range(len(is_speech)): if is_speech[i] and not in_speech: speech_start = i * hop in_speech = True elif not is_speech[i] and in_speech: speech_end = i * hop + frame_size dur_frames = i - (speech_start // hop) if dur_frames >= min_speech_frames: timestamps.append({"start": speech_start, "end": min(speech_end, len(audio_np))}) in_speech = False if in_speech: speech_end = len(audio_np) timestamps.append({"start": speech_start, "end": speech_end}) return timestamps _qwen_model = None def _transcribe_batch_gpu(batch_paths, language="Japanese"): """GPU'da batch transkripsiyon.""" global _qwen_model if _qwen_model is None: from qwen_asr import Qwen3ASRModel logger.info("Loading Qwen3-ASR-1.7B...") _qwen_model = Qwen3ASRModel.from_pretrained( "Qwen/Qwen3-ASR-1.7B", dtype=torch.bfloat16, device_map="cuda:0", max_inference_batch_size=64, max_new_tokens=512, ) logger.info("Qwen3-ASR loaded!") return _qwen_model.transcribe(audio=batch_paths, language=language) if IS_HF_SPACE: _transcribe_batch_gpu = spaces.GPU(duration=120)(_transcribe_batch_gpu) def _transcribe_segments(all_seg_infos, speaker_id, batch_size=64, language="Japanese"): all_results = [] for batch_start in range(0, len(all_seg_infos), batch_size): batch = all_seg_infos[batch_start:batch_start + batch_size] batch_paths = [s["seg_path"] for s in batch] try: results = _transcribe_batch_gpu(batch_paths, language) for seg_info, res in zip(batch, results): text = res.text.strip() if res and res.text else "" if text: all_results.append({ "audio_path": seg_info["seg_path"], "text": text, "speaker_id": speaker_id, "duration": seg_info["duration"], "source_file": seg_info["source_file"], }) except Exception as e: logger.warning(f"[TR] Batch failed, single fallback: {e}") for seg_info in batch: try: results = _transcribe_batch_gpu([seg_info["seg_path"]], language) text = results[0].text.strip() if results and results[0].text else "" except Exception: text = "" if text: all_results.append({ "audio_path": seg_info["seg_path"], "text": text, "speaker_id": speaker_id, "duration": seg_info["duration"], "source_file": seg_info["source_file"], }) logger.info(f"[TR] {min(batch_start+batch_size, len(all_seg_infos))}/{len(all_seg_infos)} done, {len(all_results)} with text") return all_results # =========================================================================== # YARDIMCI FONKSIYONLAR # =========================================================================== def _save_cookies(cookies_text): if not cookies_text or not cookies_text.strip(): return None path = os.path.join(WORK_ROOT, "cookies.txt") with open(path, "w") as f: f.write(cookies_text) return path def _read_audio_sf(path, sampling_rate=16000): data, sr = sf.read(path, dtype="float32") if len(data.shape) > 1: data = data.mean(axis=1) if sr != sampling_rate: ratio = sampling_rate / sr new_len = int(len(data) * ratio) indices = np.arange(new_len) / ratio idx_floor = np.floor(indices).astype(int) idx_ceil = np.minimum(idx_floor + 1, len(data) - 1) frac = indices - idx_floor data = data[idx_floor] * (1 - frac) + data[idx_ceil] * frac return torch.FloatTensor(data) def _find_best_split(audio_np, start, end, sr=SAMPLE_RATE): seg = audio_np[start:end] win_samples = int(0.03 * sr) hop = win_samples // 3 s_start = int(len(seg) * 0.2) s_end = int(len(seg) * 0.8) if s_end - s_start < win_samples * 2: return start + len(seg) // 2 best_energy = float('inf') best_pos = start + len(seg) // 2 for pos in range(s_start, s_end - win_samples, hop): win = seg[pos:pos + win_samples] rms = np.sqrt(np.mean(win ** 2)) if rms < best_energy: best_energy = rms best_pos = start + pos + win_samples // 2 return best_pos def _split_long_segment(audio_np, start, end, sr=SAMPLE_RATE): dur = (end - start) / sr if dur <= HARD_MAX_SEC: return [(start, end)] split_pt = _find_best_split(audio_np, start, end, sr) left = _split_long_segment(audio_np, start, split_pt, sr) right = _split_long_segment(audio_np, split_pt, end, sr) return left + right def _vad_segment(timestamps, audio_np, audio_len, sr=SAMPLE_RATE): if not timestamps: total_sec = audio_len / sr if total_sec < MIN_SEGMENT_SEC: return [(0, audio_len)] segs = [] pos = 0 while pos < audio_len: end = min(pos + int(TARGET_SEGMENT_SEC * sr), audio_len) segs.append((pos, end)) pos = end return segs pad = int(0.15 * sr) segments = [] group_start = timestamps[0]["start"] group_end = timestamps[0]["end"] for ts in timestamps[1:]: cs, ce = ts["start"], ts["end"] new_dur = (ce - group_start) / sr if new_dur <= MAX_SEGMENT_SEC: group_end = ce elif new_dur <= HARD_MAX_SEC: cur_dur = (group_end - group_start) / sr if cur_dur >= MIN_SEGMENT_SEC: segments.append((group_start, group_end)) group_start = cs group_end = ce else: group_end = ce else: if (group_end - group_start) / sr >= 1.0: segments.append((group_start, group_end)) group_start = cs group_end = ce last_dur = (group_end - group_start) / sr if last_dur < MIN_SEGMENT_SEC and segments: prev_s, prev_e = segments[-1] merged_dur = (group_end - prev_s) / sr if merged_dur <= HARD_MAX_SEC: segments[-1] = (prev_s, group_end) elif last_dur >= 1.0: segments.append((group_start, group_end)) elif last_dur >= 1.0: segments.append((group_start, group_end)) final = [] for s, e in segments: dur = (e - s) / sr if dur > HARD_MAX_SEC: sub = _split_long_segment(audio_np, s, e, sr) for ss, se in sub: final.append((max(0, ss - pad), min(audio_len, se + pad))) else: final.append((max(0, s - pad), min(audio_len, e + pad))) return final # =========================================================================== # GRADIO UI # =========================================================================== def build_ui(): with gr.Blocks(title="TTS Dataset Builder") as demo: gr.Markdown("# TTS Dataset Builder") with gr.Row(): with gr.Column(scale=2): youtube_url = gr.Textbox( label="YouTube URL", placeholder="Playlist or video URL", ) cookies_text = gr.Textbox( label="Cookies", placeholder="Paste cookies.txt content here", lines=4, ) with gr.Column(scale=1): hf_repo = gr.Textbox(label="Dataset Repository", placeholder="org/dataset-name") hf_token = gr.Textbox( label="HF Token", type="password", value=os.environ.get("HF_TOKEN", ""), ) speaker_id = gr.Textbox(label="Speaker ID", value="speaker_0001") with gr.Row(): trim_start = gr.Number(label="Trim Start (s)", value=10, minimum=0) trim_end = gr.Number(label="Trim End (s)", value=10, minimum=0) sleep_interval = gr.Number(label="Download Delay (s)", value=30, minimum=0) use_deepfilter = gr.Checkbox(label="DeepFilterNet", value=False) language = gr.Dropdown( label="Language", choices=[ "Japanese", "Turkish", "English", "Chinese", "Korean", "German", "French", "Spanish", "Arabic", "Russian", ], value="Japanese", ) list_btn = gr.Button("Fetch Videos", variant="secondary") list_status = gr.Textbox(label="Status", interactive=False) video_selector = gr.CheckboxGroup( label="Select videos to include", choices=[], ) run_btn = gr.Button("Run Pipeline", variant="primary") output_log = gr.Textbox(label="Log", lines=12, interactive=False) list_btn.click( fn=list_videos, inputs=[youtube_url, cookies_text], outputs=[video_selector, list_status], ) run_btn.click( fn=run_pipeline, inputs=[ youtube_url, cookies_text, video_selector, hf_repo, hf_token, speaker_id, trim_start, trim_end, use_deepfilter, sleep_interval, language, ], outputs=output_log, ) return demo def _setup_runtime(): deno_bin = os.path.expanduser("~/.deno/bin/deno") if not os.path.exists(deno_bin): logger.info("Installing deno runtime...") subprocess.run( "curl -fsSL https://deno.land/install.sh | sh", shell=True, capture_output=True, ) if os.path.exists(deno_bin): deno_dir = os.path.dirname(deno_bin) if deno_dir not in os.environ.get("PATH", ""): os.environ["PATH"] = deno_dir + ":" + os.environ.get("PATH", "") if __name__ == "__main__": _setup_runtime() demo = build_ui() demo.launch()