diff --git a/skills/indic-voice-pipeline/SKILL.md b/skills/indic-voice-pipeline/SKILL.md new file mode 100644 index 00000000..053ec3bb --- /dev/null +++ b/skills/indic-voice-pipeline/SKILL.md @@ -0,0 +1,141 @@ +--- +name: indic-voice-pipeline +description: | + Download, transcribe, and translate audio/video in 12 Indian languages locally using fine-tuned Whisper models. + Use when the user mentions: transcribe, transcription, speech to text, audio to text, translate audio, + download video, subtitle, caption, detect language, whisper, dictation, Indian language, + Telugu, Hindi, Kannada, Tamil, Bengali, Malayalam, Marathi, Gujarati, Odia, Punjabi, Sanskrit, Urdu. + Works with any audio (mp3, wav, m4a, ogg, flac) or video (mp4, mkv, webm, mov) file. + Downloads from YouTube, Twitter/X, TikTok, Instagram, and 1000+ sites via yt-dlp. + Supports 12 Indian languages with fine-tuned models from vasista22 (IIT Madras) and AI4Bharat IndicWhisper. + Runs 100% locally — no API keys needed. +license: MIT +--- + +# Indic Voice Pipeline + +Download, transcribe, and translate audio/video in 12 Indian languages — entirely on your machine. No cloud APIs, no data leaving your machine, no API keys. + +**By [HumanCTO](https://github.com/humancto) | [GitHub](https://github.com/humancto/indic-voice-pipeline) | [Documentation](https://humancto.github.io/indic-voice-pipeline/)** + +## Prerequisites + +Run once to install dependencies: + +```bash +pip install yt-dlp openai-whisper "transformers==4.46.3" accelerate "numpy<2" --quiet +brew install ffmpeg # macOS +``` + +## Step-by-Step Workflow + +**For ANY download, transcription, or translation request, follow these steps:** + +### Step 1: Determine intent and run the appropriate command + +**User wants to download a video:** + +```bash +python3 {SKILL_PATH}/scripts/video_downloader.py download "" --output-dir ~/Downloads +``` + +**User wants audio only:** + +```bash +python3 {SKILL_PATH}/scripts/video_downloader.py audio "" --output-dir ~/Downloads --audio-format mp3 +``` + +**User wants to transcribe audio/video to text:** + +```bash +python3 {SKILL_PATH}/scripts/whisper_transcribe.py transcribe "" --output-dir ~/Downloads +``` + +**User wants to transcribe with a specific Indian language (uses fine-tuned model automatically):** + +```bash +python3 {SKILL_PATH}/scripts/whisper_transcribe.py transcribe "" --language te --output-dir ~/Downloads +``` + +**User wants to translate audio/video to English:** + +```bash +python3 {SKILL_PATH}/scripts/whisper_transcribe.py translate "" --output-dir ~/Downloads +``` + +**User wants to detect the language:** + +```bash +python3 {SKILL_PATH}/scripts/whisper_transcribe.py detect "" +``` + +**User wants video/audio info without processing:** + +```bash +python3 {SKILL_PATH}/scripts/video_downloader.py info "" +python3 {SKILL_PATH}/scripts/whisper_transcribe.py info "" +``` + +### Step 2: Report results + +Tell the user: + +- Where files were saved (full paths) +- Detected language and confidence +- The full transcription text +- Processing time +- Output files: .txt (plain text), .srt (subtitles), .json (structured) + +## Supported Indian Languages + +The skill auto-selects the best fine-tuned model when you pass `--language `: + +### vasista22 Models (HuggingFace — auto-downloaded) + +| Language | Code | Model | +| -------- | ---- | --------------------------------- | +| Telugu | `te` | vasista22/whisper-telugu-large-v2 | +| Hindi | `hi` | vasista22/whisper-hindi-large-v2 | +| Kannada | `kn` | vasista22/whisper-kannada-medium | +| Gujarati | `gu` | vasista22/whisper-gujarati-medium | +| Tamil | `ta` | vasista22/whisper-tamil-medium | + +### AI4Bharat IndicWhisper Models (ZIP download — cached at ~/.cache/indicwhisper/) + +| Language | Code | +| --------- | ---- | +| Bengali | `bn` | +| Malayalam | `ml` | +| Marathi | `mr` | +| Odia | `or` | +| Punjabi | `pa` | +| Sanskrit | `sa` | +| Urdu | `ur` | + +## Full Pipeline Example + +Download a Telugu video and transcribe it: + +```bash +# Download +python3 {SKILL_PATH}/scripts/video_downloader.py download "https://youtube.com/watch?v=..." --output-dir ~/Downloads + +# Transcribe with fine-tuned Telugu model +python3 {SKILL_PATH}/scripts/whisper_transcribe.py transcribe ~/Downloads/video.mp4 --language te --output-dir ~/Downloads + +# Translate to English +python3 {SKILL_PATH}/scripts/whisper_transcribe.py translate ~/Downloads/video.mp4 --language te --output-dir ~/Downloads +``` + +## Important Notes + +- Default output: `~/Downloads` +- Three output files per transcription: `.txt`, `.srt` (subtitles), `.json` (structured) +- Video files have audio extracted automatically (ffmpeg, 16kHz mono WAV) +- Smart chunking: 25s windows with 5s overlap — no words lost at boundaries +- Hardware: Apple Silicon (MPS), NVIDIA GPU (CUDA), or CPU — auto-detected +- Translation always outputs English (Whisper limitation) +- First run downloads the model — subsequent runs use cache +- IndicWhisper models (~600 MB each) are downloaded and cached on first use +- Use `--model medium` or `--model large` for better accuracy on noisy audio +- Use `--hf-model` to specify any custom HuggingFace Whisper model diff --git a/skills/indic-voice-pipeline/scripts/check_deps.py b/skills/indic-voice-pipeline/scripts/check_deps.py new file mode 100644 index 00000000..f179cc1d --- /dev/null +++ b/skills/indic-voice-pipeline/scripts/check_deps.py @@ -0,0 +1,106 @@ +#!/usr/bin/env python3 +"""Whisper Transcribe Dependency Checker — Verify all requirements are met.""" + +import shutil +import subprocess +import sys +from pathlib import Path + + +def check(name, found, version=None, install_hint=None): + if found: + ver = f" ({version})" if version else "" + print(f" [OK] {name}{ver}") + else: + hint = f" -> {install_hint}" if install_hint else "" + print(f" [--] {name} — NOT FOUND{hint}") + return found + + +def get_cmd_version(cmd, args=None): + try: + result = subprocess.run( + [cmd] + (args or ["--version"]), + capture_output=True, text=True, timeout=10, + ) + return result.stdout.strip().splitlines()[0] if result.returncode == 0 else None + except (FileNotFoundError, subprocess.TimeoutExpired): + return None + + +def check_python_package(pkg_name): + try: + result = subprocess.run( + [sys.executable, "-c", f"import {pkg_name}; print({pkg_name}.__version__ if hasattr({pkg_name}, '__version__') else 'installed')"], + capture_output=True, text=True, timeout=10, + ) + if result.returncode == 0: + return result.stdout.strip() + return None + except (FileNotFoundError, subprocess.TimeoutExpired): + return None + + +def main(): + print("=" * 50) + print(" Whisper Transcribe — Dependency Check") + print("=" * 50) + all_ok = True + + # --- Python --- + print(f"\nPython") + py_ver = sys.version.split()[0] + py_ok = sys.version_info >= (3, 10) + if not check("Python 3.10+", py_ok, py_ver, "Install Python 3.10+ from python.org"): + all_ok = False + + # --- CLI tools --- + print(f"\nCLI Tools") + + ffmpeg_ver = get_cmd_version("ffmpeg", ["-version"]) + if ffmpeg_ver: + ffmpeg_ver = ffmpeg_ver.split("version ")[-1].split(" ")[0] if "version" in ffmpeg_ver else ffmpeg_ver[:40] + if not check("ffmpeg", ffmpeg_ver, ffmpeg_ver, "brew install ffmpeg"): + all_ok = False + + ffprobe_ver = get_cmd_version("ffprobe", ["-version"]) + if not check("ffprobe", ffprobe_ver, install_hint="Included with ffmpeg"): + all_ok = False + + # --- Python packages --- + print(f"\nPython Packages") + + whisper_ver = check_python_package("whisper") + if not check("openai-whisper", whisper_ver, whisper_ver, "pip install openai-whisper"): + all_ok = False + + torch_ver = check_python_package("torch") + if not check("torch (PyTorch)", torch_ver, torch_ver, "pip install torch"): + all_ok = False + + transformers_ver = check_python_package("transformers") + if not check("transformers (HuggingFace)", transformers_ver, transformers_ver, "pip install transformers accelerate"): + all_ok = False + + # --- Skill installation --- + print(f"\nClaude Code Skill") + skill_path = Path.home() / ".claude" / "skills" / "whisper-transcribe" / "SKILL.md" + check("Skill installed", skill_path.exists(), str(skill_path) if skill_path.exists() else None, + "Copy files to ~/.claude/skills/whisper-transcribe/") + + # --- Summary --- + print("\n" + "=" * 50) + if all_ok: + print(" All good! Ready to use Whisper Transcribe.") + else: + print(" Some dependencies are missing.") + print(" Quick install:") + print(" pip install openai-whisper") + print(" brew install ffmpeg (macOS)") + print("=" * 50) + + sys.exit(0 if all_ok else 1) + + +if __name__ == "__main__": + main() diff --git a/skills/indic-voice-pipeline/scripts/check_deps_downloader.py b/skills/indic-voice-pipeline/scripts/check_deps_downloader.py new file mode 100644 index 00000000..a9a1c01f --- /dev/null +++ b/skills/indic-voice-pipeline/scripts/check_deps_downloader.py @@ -0,0 +1,82 @@ +#!/usr/bin/env python3 +"""Video Downloader Dependency Checker — Verify all requirements are met.""" + +import shutil +import subprocess +import sys +from pathlib import Path + + +def check(name, found, version=None, install_hint=None): + if found: + ver = f" ({version})" if version else "" + print(f" [OK] {name}{ver}") + else: + hint = f" -> {install_hint}" if install_hint else "" + print(f" [--] {name} — NOT FOUND{hint}") + return found + + +def get_cmd_version(cmd, args=None): + try: + result = subprocess.run( + [cmd] + (args or ["--version"]), + capture_output=True, text=True, timeout=5, + ) + return result.stdout.strip().splitlines()[0] if result.returncode == 0 else None + except (FileNotFoundError, subprocess.TimeoutExpired): + return None + + +def main(): + print("=" * 50) + print(" Video Downloader — Dependency Check") + print("=" * 50) + all_ok = True + + # --- Python --- + print(f"\nPython") + py_ver = sys.version.split()[0] + py_ok = sys.version_info >= (3, 10) + if not check("Python 3.10+", py_ok, py_ver, "Install Python 3.10+ from python.org"): + all_ok = False + + # --- CLI tools --- + print(f"\nCLI Tools") + + ytdlp_ver = get_cmd_version("yt-dlp") + if not check("yt-dlp", ytdlp_ver, ytdlp_ver, "pip install yt-dlp"): + all_ok = False + + ffmpeg_ver = get_cmd_version("ffmpeg", ["-version"]) + if ffmpeg_ver: + ffmpeg_ver = ffmpeg_ver.split("version ")[-1].split(" ")[0] if "version" in ffmpeg_ver else ffmpeg_ver[:40] + if not check("ffmpeg", ffmpeg_ver, ffmpeg_ver, "brew install ffmpeg (macOS) / sudo apt install ffmpeg (Linux)"): + all_ok = False + + ffprobe_ver = get_cmd_version("ffprobe", ["-version"]) + if not check("ffprobe", ffprobe_ver, install_hint="Included with ffmpeg"): + all_ok = False + + # --- Skill installation --- + print(f"\nClaude Code Skill") + skill_path = Path.home() / ".claude" / "skills" / "video-downloader" / "SKILL.md" + check("Skill installed", skill_path.exists(), str(skill_path) if skill_path.exists() else None, + "Copy files to ~/.claude/skills/video-downloader/") + + # --- Summary --- + print("\n" + "=" * 50) + if all_ok: + print(" All good! Ready to use Video Downloader.") + else: + print(" Some dependencies are missing.") + print(" Quick install:") + print(" pip install yt-dlp") + print(" brew install ffmpeg (macOS)") + print("=" * 50) + + sys.exit(0 if all_ok else 1) + + +if __name__ == "__main__": + main() diff --git a/skills/indic-voice-pipeline/scripts/video_downloader.py b/skills/indic-voice-pipeline/scripts/video_downloader.py new file mode 100644 index 00000000..246dd249 --- /dev/null +++ b/skills/indic-voice-pipeline/scripts/video_downloader.py @@ -0,0 +1,528 @@ +#!/usr/bin/env python3 +"""Video Downloader — Download videos, audio, and playlists from any URL. + +Commands: + download — Download single video (best quality MP4) + audio — Download audio only (MP3/M4A/Opus) + playlist — Download all videos in a playlist + formats — List available formats/qualities for a URL + info — Show video metadata without downloading + +Examples: + python video_downloader.py download "https://youtube.com/watch?v=..." + python video_downloader.py audio "https://youtube.com/watch?v=..." --audio-format m4a + python video_downloader.py playlist "https://youtube.com/playlist?list=..." + python video_downloader.py formats "https://youtube.com/watch?v=..." + python video_downloader.py info "https://youtube.com/watch?v=..." +""" + +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +from pathlib import Path + + +# ─── HELPERS ──────────────────────────────────────────────────────────────────── + + +def _timestamp_str(seconds: float | int | None) -> str: + """Convert seconds to HH:MM:SS or MM:SS.""" + if not seconds: + return "00:00" + seconds = float(seconds) + h = int(seconds // 3600) + m = int((seconds % 3600) // 60) + s = int(seconds % 60) + return f"{h:02d}:{m:02d}:{s:02d}" if h > 0 else f"{m:02d}:{s:02d}" + + +def _get_format_string(quality: str | None) -> str: + """Get yt-dlp format string for desired quality.""" + if not quality: + return "bestvideo[height<=1080][ext=mp4]+bestaudio[ext=m4a]/best[height<=1080][ext=mp4]/best" + + q = str(quality).lower().replace("p", "") + if q == "4k": + q = "2160" + try: + height = int(q) + return f"bestvideo[height<={height}][ext=mp4]+bestaudio[ext=m4a]/best[height<={height}][ext=mp4]/best" + except ValueError: + return "best[ext=mp4]/best" + + +def _find_downloaded_file(directory: Path, extensions: list[str]) -> Path | None: + """Find the most recently created file with given extensions.""" + candidates = [] + for ext in extensions: + candidates.extend(directory.glob(f"*{ext}")) + if not candidates: + return None + return max(candidates, key=lambda p: p.stat().st_mtime) + + +def _check_ytdlp() -> dict | None: + """Check yt-dlp is available. Returns error dict if not.""" + try: + subprocess.run(["yt-dlp", "--version"], capture_output=True, check=True) + return None + except (FileNotFoundError, subprocess.CalledProcessError): + return {"error": "yt-dlp not found. Install with: pip install yt-dlp"} + + +def _get_metadata(url: str, playlist: bool = False) -> tuple[dict | None, dict | None]: + """Fetch metadata from URL. Returns (metadata, error).""" + meta_cmd = ["yt-dlp", "--dump-json", "--no-download"] + if not playlist: + meta_cmd.append("--no-playlist") + meta_cmd.append(url) + + result = subprocess.run(meta_cmd, capture_output=True, text=True) + if result.returncode != 0: + return None, {"error": f"Failed to get video metadata: {result.stderr.strip()[:500]}"} + + try: + meta = json.loads(result.stdout.strip().splitlines()[0]) + return meta, None + except (json.JSONDecodeError, IndexError): + return None, {"error": "Failed to parse video metadata"} + + +def _format_filesize(size_bytes: int | float | None) -> str: + """Human-readable file size.""" + if not size_bytes: + return "unknown" + size = float(size_bytes) + for unit in ["B", "KB", "MB", "GB"]: + if size < 1024: + return f"{size:.1f} {unit}" + size /= 1024 + return f"{size:.1f} TB" + + +def _save_meta(data: dict, output_dir: Path) -> None: + """Save download_meta.json alongside the download.""" + meta_path = output_dir / "download_meta.json" + with open(meta_path, "w") as f: + json.dump(data, f, indent=2) + + +def _build_metadata_dict(meta: dict, url: str) -> dict: + """Build a standard metadata dict from yt-dlp JSON.""" + return { + "title": meta.get("title", "Unknown"), + "uploader": meta.get("uploader", "Unknown"), + "upload_date": meta.get("upload_date", ""), + "duration_seconds": meta.get("duration", 0), + "duration": _timestamp_str(meta.get("duration", 0)), + "view_count": meta.get("view_count"), + "like_count": meta.get("like_count"), + "description": (meta.get("description") or "")[:500], + "webpage_url": meta.get("webpage_url", url), + "resolution": f"{meta.get('width', '?')}x{meta.get('height', '?')}", + "thumbnail": meta.get("thumbnail", ""), + } + + +# ─── COMMANDS ─────────────────────────────────────────────────────────────────── + + +def cmd_download( + url: str, + output_dir: str = "~/Downloads", + quality: str | None = None, + format_id: str | None = None, + extract_subs: bool = True, +) -> dict: + """Download a single video from URL.""" + err = _check_ytdlp() + if err: + return err + + out_path = Path(output_dir).expanduser() + out_path.mkdir(parents=True, exist_ok=True) + + print("Fetching metadata...", file=sys.stderr) + meta, err = _get_metadata(url) + if err: + return err + + video_template = str(out_path / "%(title).80s.%(ext)s") + + if format_id: + fmt = format_id + else: + fmt = _get_format_string(quality) + + dl_cmd = [ + "yt-dlp", + "-f", fmt, + "--merge-output-format", "mp4", + "-o", video_template, + "--no-playlist", + "--no-overwrites", + ] + + if extract_subs: + dl_cmd.extend([ + "--write-subs", + "--write-auto-subs", + "--sub-langs", "en", + "--sub-format", "srt/vtt/best", + "--convert-subs", "srt", + ]) + + dl_cmd.append(url) + + print("Downloading...", file=sys.stderr) + dl_result = subprocess.run(dl_cmd, capture_output=True, text=True) + + if dl_result.returncode != 0: + # Fallback: simpler format + print("Retrying with fallback format...", file=sys.stderr) + dl_cmd_simple = [ + "yt-dlp", + "-f", "best[ext=mp4]/best", + "-o", video_template, + "--no-playlist", + "--no-overwrites", + url, + ] + dl_result = subprocess.run(dl_cmd_simple, capture_output=True, text=True) + if dl_result.returncode != 0: + return {"error": f"Download failed: {dl_result.stderr.strip()[:500]}"} + + video_file = _find_downloaded_file(out_path, [".mp4", ".mkv", ".webm", ".mov"]) + if not video_file: + return {"error": f"Download completed but video file not found in {out_path}"} + + sub_file = _find_downloaded_file(out_path, [".srt", ".vtt"]) + + file_size = video_file.stat().st_size + + result = { + "status": "downloaded", + "video_path": str(video_file), + "video_filename": video_file.name, + "file_size": _format_filesize(file_size), + "file_size_bytes": file_size, + "output_dir": str(out_path), + "metadata": _build_metadata_dict(meta, url), + } + + if sub_file: + result["subtitle_path"] = str(sub_file) + result["subtitle_filename"] = sub_file.name + + _save_meta(result, out_path) + return result + + +def cmd_audio( + url: str, + output_dir: str = "~/Downloads", + audio_format: str = "mp3", +) -> dict: + """Download audio only from URL.""" + err = _check_ytdlp() + if err: + return err + + out_path = Path(output_dir).expanduser() + out_path.mkdir(parents=True, exist_ok=True) + + print("Fetching metadata...", file=sys.stderr) + meta, err = _get_metadata(url) + if err: + return err + + audio_template = str(out_path / "%(title).80s.%(ext)s") + dl_cmd = [ + "yt-dlp", + "-x", + "--audio-format", audio_format, + "--audio-quality", "0", + "-o", audio_template, + "--no-playlist", + "--no-overwrites", + url, + ] + + print("Downloading audio...", file=sys.stderr) + dl_result = subprocess.run(dl_cmd, capture_output=True, text=True) + if dl_result.returncode != 0: + return {"error": f"Audio download failed: {dl_result.stderr.strip()[:500]}"} + + ext_map = {"mp3": [".mp3"], "m4a": [".m4a"], "opus": [".opus"]} + search_exts = ext_map.get(audio_format, [f".{audio_format}"]) + # Also search common fallbacks + search_exts.extend([".mp3", ".m4a", ".opus", ".ogg", ".webm"]) + + audio_file = _find_downloaded_file(out_path, search_exts) + if not audio_file: + return {"error": f"Download completed but audio file not found in {out_path}"} + + file_size = audio_file.stat().st_size + + result = { + "status": "downloaded", + "type": "audio", + "audio_path": str(audio_file), + "audio_filename": audio_file.name, + "audio_format": audio_format, + "file_size": _format_filesize(file_size), + "file_size_bytes": file_size, + "output_dir": str(out_path), + "metadata": _build_metadata_dict(meta, url), + } + + _save_meta(result, out_path) + return result + + +def cmd_playlist( + url: str, + output_dir: str = "~/Downloads", + quality: str | None = None, +) -> dict: + """Download all videos in a playlist.""" + err = _check_ytdlp() + if err: + return err + + out_path = Path(output_dir).expanduser() + out_path.mkdir(parents=True, exist_ok=True) + + video_template = str(out_path / "%(playlist_index)03d - %(title).70s.%(ext)s") + fmt = _get_format_string(quality) + + dl_cmd = [ + "yt-dlp", + "-f", fmt, + "--merge-output-format", "mp4", + "-o", video_template, + "--no-overwrites", + "--yes-playlist", + url, + ] + + print("Downloading playlist...", file=sys.stderr) + dl_result = subprocess.run(dl_cmd, capture_output=True, text=True, timeout=3600) + + if dl_result.returncode != 0 and not list(out_path.glob("*.mp4")): + return {"error": f"Playlist download failed: {dl_result.stderr.strip()[:500]}"} + + # Collect all downloaded files + video_files = sorted(out_path.glob("*.mp4"), key=lambda p: p.name) + if not video_files: + # Try other extensions + for ext in [".mkv", ".webm", ".mov"]: + video_files.extend(out_path.glob(f"*{ext}")) + video_files.sort(key=lambda p: p.name) + + if not video_files: + return {"error": f"Playlist download completed but no video files found in {out_path}"} + + items = [] + total_size = 0 + for vf in video_files: + sz = vf.stat().st_size + total_size += sz + items.append({ + "video_path": str(vf), + "video_filename": vf.name, + "file_size": _format_filesize(sz), + "file_size_bytes": sz, + }) + + result = { + "status": "downloaded", + "type": "playlist", + "total_videos": len(items), + "total_size": _format_filesize(total_size), + "total_size_bytes": total_size, + "output_dir": str(out_path), + "videos": items, + } + + _save_meta(result, out_path) + return result + + +def cmd_formats(url: str) -> dict: + """List all available formats for a URL.""" + err = _check_ytdlp() + if err: + return err + + result = subprocess.run( + ["yt-dlp", "-F", "--no-playlist", url], + capture_output=True, text=True, + ) + if result.returncode != 0: + return {"error": f"Failed to list formats: {result.stderr.strip()[:500]}"} + + lines = result.stdout.strip().splitlines() + + formats = [] + header_found = False + for line in lines: + if line.startswith("ID") or line.startswith("---"): + header_found = True + continue + if not header_found: + continue + if not line.strip(): + continue + + parts = line.split() + if len(parts) < 3: + continue + + fmt = { + "format_id": parts[0], + "ext": parts[1] if len(parts) > 1 else "", + "description": " ".join(parts[2:]) if len(parts) > 2 else "", + "raw": line.strip(), + } + formats.append(fmt) + + # Also get metadata for context + meta, _ = _get_metadata(url) + title = meta.get("title", "Unknown") if meta else "Unknown" + + return { + "title": title, + "url": url, + "total_formats": len(formats), + "formats": formats, + } + + +def cmd_info(url: str) -> dict: + """Show video metadata without downloading.""" + err = _check_ytdlp() + if err: + return err + + print("Fetching info...", file=sys.stderr) + meta, err = _get_metadata(url) + if err: + return err + + return { + "title": meta.get("title", "Unknown"), + "uploader": meta.get("uploader", "Unknown"), + "uploader_url": meta.get("uploader_url", ""), + "upload_date": meta.get("upload_date", ""), + "duration_seconds": meta.get("duration", 0), + "duration": _timestamp_str(meta.get("duration", 0)), + "view_count": meta.get("view_count"), + "like_count": meta.get("like_count"), + "comment_count": meta.get("comment_count"), + "description": (meta.get("description") or "")[:1000], + "webpage_url": meta.get("webpage_url", url), + "resolution": f"{meta.get('width', '?')}x{meta.get('height', '?')}", + "fps": meta.get("fps"), + "vcodec": meta.get("vcodec", ""), + "acodec": meta.get("acodec", ""), + "filesize_approx": _format_filesize(meta.get("filesize_approx")), + "thumbnail": meta.get("thumbnail", ""), + "categories": meta.get("categories", []), + "tags": (meta.get("tags") or [])[:20], + "subtitles_available": bool(meta.get("subtitles") or meta.get("automatic_captions")), + } + + +# ─── CLI ──────────────────────────────────────────────────────────────────────── + + +def main(): + parser = argparse.ArgumentParser( + description="Video Downloader — Download videos, audio, and playlists from any URL", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + %(prog)s download "https://youtube.com/watch?v=dQw4w9WgXcQ" + %(prog)s download "https://youtube.com/watch?v=..." --quality 720 --output-dir ~/Videos + %(prog)s audio "https://youtube.com/watch?v=..." --audio-format m4a + %(prog)s playlist "https://youtube.com/playlist?list=..." + %(prog)s formats "https://youtube.com/watch?v=..." + %(prog)s info "https://youtube.com/watch?v=..." + """, + ) + sub = parser.add_subparsers(dest="command", required=True) + + # download + dl_p = sub.add_parser("download", help="Download single video") + dl_p.add_argument("url", help="Video URL") + dl_p.add_argument("--output-dir", default="~/Downloads", help="Output directory (default: ~/Downloads)") + dl_p.add_argument("--quality", default=None, help="Max quality: 360, 480, 720, 1080, 4k (default: 1080)") + dl_p.add_argument("--format", dest="format_id", default=None, help="Specific format ID (from 'formats' output)") + dl_p.add_argument("--no-subs", action="store_true", help="Skip subtitle download") + + # audio + au_p = sub.add_parser("audio", help="Download audio only") + au_p.add_argument("url", help="Video URL") + au_p.add_argument("--output-dir", default="~/Downloads", help="Output directory (default: ~/Downloads)") + au_p.add_argument("--audio-format", default="mp3", choices=["mp3", "m4a", "opus"], help="Audio format (default: mp3)") + + # playlist + pl_p = sub.add_parser("playlist", help="Download all videos in a playlist") + pl_p.add_argument("url", help="Playlist URL") + pl_p.add_argument("--output-dir", default="~/Downloads", help="Output directory (default: ~/Downloads)") + pl_p.add_argument("--quality", default=None, help="Max quality: 360, 480, 720, 1080, 4k (default: 1080)") + + # formats + fmt_p = sub.add_parser("formats", help="List available formats/qualities") + fmt_p.add_argument("url", help="Video URL") + + # info + info_p = sub.add_parser("info", help="Show video metadata without downloading") + info_p.add_argument("url", help="Video URL") + + args = parser.parse_args() + + if args.command == "download": + result = cmd_download(args.url, args.output_dir, args.quality, args.format_id, not args.no_subs) + print(json.dumps(result, indent=2)) + if "error" not in result: + print(f"\nDownloaded to {result['video_path']}", file=sys.stderr) + print(f"Size: {result['file_size']}", file=sys.stderr) + + elif args.command == "audio": + result = cmd_audio(args.url, args.output_dir, args.audio_format) + print(json.dumps(result, indent=2)) + if "error" not in result: + print(f"\nDownloaded audio to {result['audio_path']}", file=sys.stderr) + print(f"Size: {result['file_size']}", file=sys.stderr) + + elif args.command == "playlist": + result = cmd_playlist(args.url, args.output_dir, args.quality) + print(json.dumps(result, indent=2)) + if "error" not in result: + print(f"\nDownloaded {result['total_videos']} videos to {result['output_dir']}", file=sys.stderr) + print(f"Total size: {result['total_size']}", file=sys.stderr) + + elif args.command == "formats": + result = cmd_formats(args.url) + print(json.dumps(result, indent=2)) + if "error" not in result: + print(f"\n{result['total_formats']} formats available for: {result['title']}", file=sys.stderr) + + elif args.command == "info": + result = cmd_info(args.url) + print(json.dumps(result, indent=2)) + if "error" not in result: + print(f"\n{result['title']} ({result['duration']})", file=sys.stderr) + + if "error" in result: + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/skills/indic-voice-pipeline/scripts/whisper_transcribe.py b/skills/indic-voice-pipeline/scripts/whisper_transcribe.py new file mode 100644 index 00000000..ea40a4d6 --- /dev/null +++ b/skills/indic-voice-pipeline/scripts/whisper_transcribe.py @@ -0,0 +1,772 @@ +#!/usr/bin/env python3 +""" +Whisper Transcribe — Local audio transcription, translation, and analysis. + +Supports both standard OpenAI Whisper models and HuggingFace fine-tuned models +(e.g., vasista22/whisper-telugu-large-v2 for Telugu). + +Commands: + transcribe — Transcribe audio/video to text (auto-detects language) + translate — Transcribe and translate to English + detect — Detect the language of an audio file + info — Show audio file metadata without transcribing + +Usage: + whisper_transcribe.py transcribe [options] + whisper_transcribe.py translate [options] + whisper_transcribe.py detect + whisper_transcribe.py info +""" + +import argparse +import json +import os +import subprocess +import sys +import time +import zipfile +from pathlib import Path +from urllib.request import urlretrieve + + +# Well-known HuggingFace fine-tuned Whisper models for specific languages. +# When --language is specified and no explicit --hf-model is given, these are +# used automatically for better accuracy. +HF_LANGUAGE_MODELS = { + "te": "vasista22/whisper-telugu-large-v2", + "hi": "vasista22/whisper-hindi-large-v2", + "kn": "vasista22/whisper-kannada-medium", + "gu": "vasista22/whisper-gujarati-medium", + "ta": "vasista22/whisper-tamil-medium", +} + +# AI4Bharat IndicWhisper models — distributed as ZIP archives (not on HuggingFace). +# Fine-tuned on Whisper-medium (769M params) using the Vistaar training dataset. +# Covers languages where vasista22 models may not be available. +# Models are downloaded once, extracted, and cached locally. +INDICWHISPER_BASE_URL = "https://indicwhisper.objectstore.e2enetworks.net" +INDICWHISPER_CACHE_DIR = Path.home() / ".cache" / "indicwhisper" + +INDICWHISPER_MODELS = { + "bn": {"name": "Bengali", "zip": "bengali_models.zip"}, + "ml": {"name": "Malayalam", "zip": "malayalam_models.zip"}, + "mr": {"name": "Marathi", "zip": "marathi_models.zip"}, + "or": {"name": "Odia", "zip": "odia_models.zip"}, + "pa": {"name": "Punjabi", "zip": "punjabi_models.zip"}, + "sa": {"name": "Sanskrit", "zip": "sanskrit_models.zip"}, + "ur": {"name": "Urdu", "zip": "urdu_models.zip"}, +} + +LANG_NAMES = { + "en": "English", "es": "Spanish", "fr": "French", "de": "German", + "it": "Italian", "pt": "Portuguese", "nl": "Dutch", "ru": "Russian", + "zh": "Chinese", "ja": "Japanese", "ko": "Korean", "ar": "Arabic", + "hi": "Hindi", "te": "Telugu", "ta": "Tamil", "bn": "Bengali", + "tr": "Turkish", "pl": "Polish", "uk": "Ukrainian", "vi": "Vietnamese", + "th": "Thai", "id": "Indonesian", "ms": "Malay", "sv": "Swedish", + "da": "Danish", "no": "Norwegian", "fi": "Finnish", "cs": "Czech", + "ro": "Romanian", "hu": "Hungarian", "el": "Greek", "he": "Hebrew", + "ur": "Urdu", "fa": "Persian", "mr": "Marathi", "gu": "Gujarati", + "kn": "Kannada", "ml": "Malayalam", "pa": "Punjabi", "or": "Odia", + "sa": "Sanskrit", +} + + +def eprint(*args, **kwargs): + print(*args, file=sys.stderr, **kwargs) + + +def get_audio_info(file_path): + """Get audio metadata using ffprobe.""" + try: + result = subprocess.run( + [ + "ffprobe", "-v", "quiet", "-print_format", "json", + "-show_format", "-show_streams", str(file_path) + ], + capture_output=True, text=True, timeout=30, + ) + if result.returncode == 0: + return json.loads(result.stdout) + return None + except Exception: + return None + + +def format_duration(seconds): + """Format seconds into HH:MM:SS.""" + seconds = int(float(seconds)) + h, remainder = divmod(seconds, 3600) + m, s = divmod(remainder, 60) + if h > 0: + return f"{h:02d}:{m:02d}:{s:02d}" + return f"{m:02d}:{s:02d}" + + +def format_timestamp(seconds): + """Format seconds into SRT-style timestamp.""" + h, remainder = divmod(seconds, 3600) + m, s = divmod(remainder, 60) + ms = int((s - int(s)) * 1000) + return f"{int(h):02d}:{int(m):02d}:{int(s):02d}.{ms:03d}" + + +def extract_audio_if_video(file_path, output_dir): + """If the input is a video file, extract audio to a temp WAV. Returns path to audio.""" + probe = get_audio_info(file_path) + if not probe: + return str(file_path) + + streams = probe.get("streams", []) + has_video = any(s.get("codec_type") == "video" for s in streams) + has_audio = any(s.get("codec_type") == "audio" for s in streams) + + if has_video and has_audio: + eprint("Extracting audio from video...") + audio_path = Path(output_dir) / f"_whisper_temp_{Path(file_path).stem}.wav" + subprocess.run( + [ + "ffmpeg", "-y", "-i", str(file_path), + "-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1", + str(audio_path) + ], + capture_output=True, timeout=300, + ) + if audio_path.exists(): + return str(audio_path) + + return str(file_path) + + +def _download_progress(block_num, block_size, total_size): + """Print download progress.""" + if total_size > 0: + downloaded = block_num * block_size + pct = min(100, downloaded * 100 / total_size) + mb_done = downloaded / (1024 * 1024) + mb_total = total_size / (1024 * 1024) + eprint(f"\r Downloaded {mb_done:.1f} / {mb_total:.1f} MB ({pct:.0f}%)", end="") + else: + downloaded = block_num * block_size + eprint(f"\r Downloaded {downloaded / (1024*1024):.1f} MB...", end="") + + +def download_and_cache_indicwhisper(lang_code): + """Download and extract an IndicWhisper model ZIP if not already cached. + Returns the local directory path to the extracted model.""" + info = INDICWHISPER_MODELS[lang_code] + zip_name = info["zip"] + lang_name = info["name"].lower() + url = f"{INDICWHISPER_BASE_URL}/{zip_name}" + + # Cache directory for this language + lang_cache = INDICWHISPER_CACHE_DIR / lang_name + marker = lang_cache / ".download_complete" + + if marker.exists(): + # Already downloaded and extracted — find the model directory + model_dir = _find_indicwhisper_model_dir(lang_cache) + if model_dir: + eprint(f"IndicWhisper {info['name']} model already cached at {model_dir}") + return str(model_dir) + + # Download + INDICWHISPER_CACHE_DIR.mkdir(parents=True, exist_ok=True) + zip_path = INDICWHISPER_CACHE_DIR / zip_name + + eprint(f"Downloading IndicWhisper {info['name']} model...") + eprint(f" URL: {url}") + eprint(f" (This is a one-time download — ~500-800 MB)") + + try: + urlretrieve(url, str(zip_path), reporthook=_download_progress) + eprint() # newline after progress + except Exception as e: + eprint(f"\nFailed to download: {e}") + if zip_path.exists(): + zip_path.unlink() + return None + + # Extract + eprint(f"Extracting to {lang_cache}...") + lang_cache.mkdir(parents=True, exist_ok=True) + try: + with zipfile.ZipFile(str(zip_path), "r") as zf: + zf.extractall(str(lang_cache)) + except Exception as e: + eprint(f"Failed to extract: {e}") + return None + finally: + # Clean up zip + if zip_path.exists(): + zip_path.unlink() + + # Mark as complete + marker.touch() + + model_dir = _find_indicwhisper_model_dir(lang_cache) + if model_dir: + eprint(f"IndicWhisper {info['name']} model ready at {model_dir}") + return str(model_dir) + + eprint(f"Warning: Could not locate model files in {lang_cache}") + return str(lang_cache) + + +def _find_indicwhisper_model_dir(cache_dir): + """Find the actual model directory inside extracted IndicWhisper files. + The ZIPs may contain a nested folder structure. We look for a directory + containing config.json (standard HF model marker).""" + cache_path = Path(cache_dir) + + # Check if config.json is directly in the cache dir + if (cache_path / "config.json").exists(): + return cache_path + + # Search one or two levels deep + for depth_pattern in ["*/config.json", "*/*/config.json"]: + matches = list(cache_path.glob(depth_pattern)) + if matches: + return matches[0].parent + + # Fallback: return the first subdirectory + subdirs = [d for d in cache_path.iterdir() if d.is_dir() and not d.name.startswith(".")] + if subdirs: + return subdirs[0] + + return None + + +def resolve_hf_model(language, hf_model_arg): + """Resolve which HuggingFace model to use, if any. + Returns model ID/path string or None (use standard whisper). + Checks vasista22 HF models first, then IndicWhisper (AI4Bharat).""" + if hf_model_arg: + return hf_model_arg + if language and language in HF_LANGUAGE_MODELS: + return HF_LANGUAGE_MODELS[language] + if language and language in INDICWHISPER_MODELS: + # Download/cache and return local path + local_path = download_and_cache_indicwhisper(language) + if local_path: + return local_path + return None + + +def load_whisper_model(model_name): + """Load a standard OpenAI Whisper model.""" + eprint(f"Loading Whisper model '{model_name}'...") + import whisper + model = whisper.load_model(model_name) + eprint(f"Model '{model_name}' loaded.") + return model + + +def load_hf_model_and_processor(hf_model_id): + """Load a HuggingFace Whisper model and processor directly (not via pipeline). + hf_model_id can be a HuggingFace repo ID (e.g. 'vasista22/whisper-telugu-large-v2') + or a local directory path (for IndicWhisper cached models).""" + is_local = os.path.isdir(hf_model_id) + label = hf_model_id if not is_local else f"local:{Path(hf_model_id).name}" + eprint(f"Loading {'local' if is_local else 'HuggingFace'} model '{label}'...") + if not is_local: + eprint("(First run will download the model — this may take a few minutes)") + import torch + from transformers import WhisperForConditionalGeneration, WhisperProcessor + + device = "mps" if torch.backends.mps.is_available() else ("cuda" if torch.cuda.is_available() else "cpu") + torch_dtype = torch.float16 if device != "cpu" else torch.float32 + + processor = WhisperProcessor.from_pretrained(hf_model_id) + model = WhisperForConditionalGeneration.from_pretrained( + hf_model_id, + torch_dtype=torch_dtype, + low_cpu_mem_usage=True, + ) + model.to(device) + + # Fix generation config for fine-tuned models with missing/empty suppress_tokens + if model.generation_config.suppress_tokens is not None and len(model.generation_config.suppress_tokens) == 0: + model.generation_config.suppress_tokens = None + if not hasattr(model.generation_config, "no_timestamps_token_id") or model.generation_config.no_timestamps_token_id is None: + model.generation_config.no_timestamps_token_id = processor.tokenizer.convert_tokens_to_ids("<|notimestamps|>") + + eprint(f"Model '{label}' loaded on {device}.") + return model, processor, device, torch_dtype + + +def transcribe_with_hf(hf_model_id, audio_path, task="transcribe"): + """Transcribe using a HuggingFace model directly. Returns dict with text and segments. + + Strategy: + - Short audio (<= 30s): single pass for best accuracy. + - Longer audio: sliding window with overlap + intelligent word-level merge + to ensure no words are lost at chunk boundaries. + """ + import torch + + model, processor, device, torch_dtype = load_hf_model_and_processor(hf_model_id) + + # Load audio + eprint("Loading audio...") + try: + import librosa + audio_array, _sr = librosa.load(audio_path, sr=16000) + except ImportError: + import whisper + audio_array = whisper.load_audio(audio_path) + + SAMPLE_RATE = 16000 + total_samples = len(audio_array) + total_duration = total_samples / SAMPLE_RATE + eprint(f"Audio duration: {total_duration:.1f}s") + + # Detect language for forced decoder IDs from model name/path + HF_MODEL_TO_LANG = { + "telugu": "te", "hindi": "hi", "kannada": "kn", + "gujarati": "gu", "tamil": "ta", + "bengali": "bn", "malayalam": "ml", "marathi": "mr", + "odia": "or", "punjabi": "pa", "sanskrit": "sa", "urdu": "ur", + } + lang_code = None + model_lower = hf_model_id.lower() + for lang_name, code in HF_MODEL_TO_LANG.items(): + if lang_name in model_lower: + lang_code = code + break + + def _generate_text(chunk): + """Run inference on a single audio chunk. Returns decoded text.""" + input_features = processor( + chunk, sampling_rate=SAMPLE_RATE, return_tensors="pt" + ).input_features.to(device, dtype=torch_dtype) + + # Odia ("or") is not in Whisper's built-in language list — pass None + effective_lang = None if lang_code == "or" else lang_code + forced_decoder_ids = processor.get_decoder_prompt_ids( + language=effective_lang, + task=task, + ) + + # Compute safe max_new_tokens: Whisper max is 448 total sequence length. + # forced_decoder_ids adds prompt tokens; we need to leave room. + prompt_len = len(forced_decoder_ids) + 1 if forced_decoder_ids else 1 + safe_max_tokens = 448 - prompt_len + + with torch.no_grad(): + predicted_ids = model.generate( + input_features, + forced_decoder_ids=forced_decoder_ids, + max_new_tokens=safe_max_tokens, + ) + + return processor.batch_decode(predicted_ids, skip_special_tokens=True)[0].strip() + + # ── Single pass for short audio (up to 30s) ── + if total_duration <= 30.0: + eprint("Short audio — single pass...") + text = _generate_text(audio_array) + segments = [{ + "id": 0, + "start": 0.0, + "end": round(total_duration, 2), + "text": text, + }] if text else [] + return {"text": text, "segments": segments} + + # ── Sliding window for longer audio ── + CHUNK_SEC = 25 # each chunk is 25s of audio + OVERLAP_SEC = 5 # 5s overlap between consecutive chunks + CHUNK_LEN = CHUNK_SEC * SAMPLE_RATE + OVERLAP_LEN = OVERLAP_SEC * SAMPLE_RATE + STRIDE = CHUNK_LEN - OVERLAP_LEN + + eprint(f"Chunked mode: {CHUNK_SEC}s windows, {OVERLAP_SEC}s overlap, {STRIDE/SAMPLE_RATE:.0f}s stride") + + raw_chunks = [] + start_sample = 0 + idx = 0 + + while start_sample < total_samples: + end_sample = min(start_sample + CHUNK_LEN, total_samples) + chunk = audio_array[start_sample:end_sample] + + # Skip very short trailing fragments + if len(chunk) / SAMPLE_RATE < 2.0: + break + + idx += 1 + t0 = start_sample / SAMPLE_RATE + t1 = end_sample / SAMPLE_RATE + eprint(f" Chunk {idx}: {t0:.1f}s – {t1:.1f}s ({len(chunk)/SAMPLE_RATE:.1f}s)") + + text = _generate_text(chunk) + if text: + raw_chunks.append({"start": t0, "end": t1, "text": text}) + + start_sample += STRIDE + + # ── Merge overlapping chunks ── + if not raw_chunks: + return {"text": "", "segments": []} + + if len(raw_chunks) == 1: + c = raw_chunks[0] + return { + "text": c["text"], + "segments": [{"id": 0, "start": round(c["start"], 2), + "end": round(c["end"], 2), "text": c["text"]}], + } + + def _find_best_overlap(prev_words, curr_words): + """Find longest suffix of prev_words matching a prefix of curr_words. + Also tries fuzzy matching: if exact match fails, look for the longest + matching subsequence at the boundary.""" + max_search = min(len(prev_words), len(curr_words), 20) + + # Exact suffix-prefix match (greedy: find longest) + best = 0 + for length in range(1, max_search + 1): + if prev_words[-length:] == curr_words[:length]: + best = length + + if best > 0: + return best + + # Fuzzy: find any single word from end of prev that appears near + # start of curr — use it as anchor to skip overlap region + search_end = min(len(curr_words), 8) + for pw_idx in range(min(len(prev_words), 10) - 1, -1, -1): + word = prev_words[pw_idx] + if len(word) < 3: + continue # skip short words, too ambiguous + for cw_idx in range(search_end): + if curr_words[cw_idx] == word: + return cw_idx + 1 + + return 0 + + merged_parts = [] + segments = [] + + for i, chunk in enumerate(raw_chunks): + if i == 0: + merged_parts.append(chunk["text"]) + segments.append({ + "id": 0, + "start": round(chunk["start"], 2), + "end": round(chunk["end"], 2), + "text": chunk["text"], + }) + continue + + prev_words = raw_chunks[i - 1]["text"].split() + curr_words = chunk["text"].split() + + overlap_words = _find_best_overlap(prev_words, curr_words) + + if overlap_words > 0: + unique = " ".join(curr_words[overlap_words:]) + else: + # Heuristic fallback: skip ~first OVERLAP/CHUNK fraction of words + skip = max(1, int(len(curr_words) * (OVERLAP_SEC / CHUNK_SEC) * 0.5)) + unique = " ".join(curr_words[skip:]) + + if unique: + merged_parts.append(unique) + seg_start = chunk["start"] + OVERLAP_SEC if overlap_words == 0 else chunk["start"] + (overlap_words / max(len(curr_words), 1)) * (chunk["end"] - chunk["start"]) + segments.append({ + "id": i, + "start": round(seg_start, 2), + "end": round(chunk["end"], 2), + "text": unique, + }) + + full_text = " ".join(merged_parts) + return {"text": full_text, "segments": segments} + + +def cmd_info(args): + """Show audio file metadata.""" + file_path = Path(args.file).expanduser().resolve() + if not file_path.exists(): + print(json.dumps({"status": "error", "error": f"File not found: {file_path}"})) + return 1 + + probe = get_audio_info(file_path) + if not probe: + print(json.dumps({"status": "error", "error": "Could not read file metadata. Is ffprobe installed?"})) + return 1 + + fmt = probe.get("format", {}) + streams = probe.get("streams", []) + + audio_streams = [s for s in streams if s.get("codec_type") == "audio"] + video_streams = [s for s in streams if s.get("codec_type") == "video"] + + duration = float(fmt.get("duration", 0)) + size_bytes = int(fmt.get("size", 0)) + + info = { + "status": "ok", + "file": str(file_path), + "filename": file_path.name, + "format": fmt.get("format_long_name", fmt.get("format_name", "unknown")), + "duration": format_duration(duration), + "duration_seconds": round(duration, 2), + "file_size": f"{size_bytes / (1024*1024):.1f} MB" if size_bytes > 1024*1024 else f"{size_bytes / 1024:.1f} KB", + "file_size_bytes": size_bytes, + "bitrate": f"{int(fmt.get('bit_rate', 0)) // 1000} kbps" if fmt.get("bit_rate") else "unknown", + "has_video": len(video_streams) > 0, + "has_audio": len(audio_streams) > 0, + } + + if audio_streams: + a = audio_streams[0] + info["audio"] = { + "codec": a.get("codec_long_name", a.get("codec_name", "unknown")), + "sample_rate": f"{a.get('sample_rate', 'unknown')} Hz", + "channels": a.get("channels", "unknown"), + "channel_layout": a.get("channel_layout", "unknown"), + } + + if video_streams: + v = video_streams[0] + info["video"] = { + "codec": v.get("codec_long_name", v.get("codec_name", "unknown")), + "resolution": f"{v.get('width', '?')}x{v.get('height', '?')}", + } + + print(json.dumps(info, indent=2, ensure_ascii=False)) + return 0 + + +def cmd_detect(args): + """Detect language of audio file.""" + file_path = Path(args.file).expanduser().resolve() + if not file_path.exists(): + print(json.dumps({"status": "error", "error": f"File not found: {file_path}"})) + return 1 + + import whisper + + model = load_whisper_model(args.model) + + temp_audio = extract_audio_if_video(file_path, file_path.parent) + + eprint("Detecting language...") + audio = whisper.load_audio(temp_audio) + audio = whisper.pad_or_trim(audio) + mel = whisper.log_mel_spectrogram(audio).to(model.device) + _, probs = model.detect_language(mel) + + # Clean up temp file + if temp_audio != str(file_path) and Path(temp_audio).exists(): + Path(temp_audio).unlink() + + top_langs = sorted(probs.items(), key=lambda x: x[1], reverse=True)[:5] + + result = { + "status": "ok", + "file": str(file_path), + "detected_language": top_langs[0][0], + "detected_language_name": LANG_NAMES.get(top_langs[0][0], top_langs[0][0]), + "confidence": round(top_langs[0][1], 4), + "top_languages": [ + { + "code": code, + "name": LANG_NAMES.get(code, code), + "probability": round(prob, 4), + } + for code, prob in top_langs + ], + } + + print(json.dumps(result, indent=2, ensure_ascii=False)) + return 0 + + +def cmd_transcribe(args): + """Transcribe audio to text using standard Whisper or a HuggingFace fine-tuned model.""" + file_path = Path(args.file).expanduser().resolve() + if not file_path.exists(): + print(json.dumps({"status": "error", "error": f"File not found: {file_path}"})) + return 1 + + output_dir = Path(args.output_dir).expanduser().resolve() + output_dir.mkdir(parents=True, exist_ok=True) + + temp_audio = extract_audio_if_video(file_path, output_dir) + + hf_model_id = resolve_hf_model(args.language, getattr(args, "hf_model", None)) + + eprint("Transcribing...") + start_time = time.time() + + if hf_model_id: + # --- HuggingFace fine-tuned model path --- + model_label = hf_model_id + hf_result = transcribe_with_hf(hf_model_id, temp_audio, task=args.task) + + full_text = hf_result["text"] + segments = hf_result["segments"] + language = args.language or "auto" + + # Build segment data with timestamps + segment_data = [] + for seg in segments: + segment_data.append({ + "id": seg["id"], + "start": seg["start"], + "end": seg["end"], + "start_ts": format_timestamp(seg["start"]), + "end_ts": format_timestamp(seg["end"]), + "text": seg["text"], + }) + else: + # --- Standard OpenAI Whisper path --- + model_label = args.model + model = load_whisper_model(args.model) + + transcribe_opts = {"verbose": False} + + if args.language: + transcribe_opts["language"] = args.language + + if args.task == "translate": + transcribe_opts["task"] = "translate" + + result = model.transcribe(temp_audio, **transcribe_opts) + + full_text = result.get("text", "").strip() + language = result.get("language", "unknown") + raw_segments = result.get("segments", []) + + segment_data = [] + for seg in raw_segments: + segment_data.append({ + "id": seg.get("id", 0), + "start": round(seg.get("start", 0), 2), + "end": round(seg.get("end", 0), 2), + "start_ts": format_timestamp(seg.get("start", 0)), + "end_ts": format_timestamp(seg.get("end", 0)), + "text": seg.get("text", "").strip(), + }) + + elapsed = time.time() - start_time + + # Clean up temp file + if temp_audio != str(file_path) and Path(temp_audio).exists(): + Path(temp_audio).unlink() + + # Save text file + stem = file_path.stem + task_suffix = "_translated" if args.task == "translate" else "" + txt_path = output_dir / f"{stem}{task_suffix}.txt" + with open(txt_path, "w", encoding="utf-8") as f: + f.write(full_text + "\n") + + # Save SRT file + srt_path = output_dir / f"{stem}{task_suffix}.srt" + with open(srt_path, "w", encoding="utf-8") as f: + for i, seg in enumerate(segment_data, 1): + f.write(f"{i}\n") + f.write(f"{seg['start_ts'].replace('.', ',')} --> {seg['end_ts'].replace('.', ',')}\n") + f.write(f"{seg['text']}\n\n") + + # Save JSON transcript + json_path = output_dir / f"{stem}{task_suffix}.json" + transcript_json = { + "file": str(file_path), + "language": language, + "task": args.task, + "model": model_label, + "text": full_text, + "segments": segment_data, + } + with open(json_path, "w", encoding="utf-8") as f: + json.dump(transcript_json, f, indent=2, ensure_ascii=False) + + # Build output + output = { + "status": "ok", + "task": args.task, + "file": str(file_path), + "model": model_label, + "language": language, + "language_name": LANG_NAMES.get(language, language), + "duration_seconds": round(segment_data[-1]["end"], 2) if segment_data else 0, + "duration": format_duration(segment_data[-1]["end"]) if segment_data else "00:00", + "processing_time": f"{elapsed:.1f}s", + "text": full_text, + "segment_count": len(segment_data), + "segments": segment_data, + "output_files": { + "text": str(txt_path), + "srt": str(srt_path), + "json": str(json_path), + }, + } + + if hf_model_id: + output["hf_model"] = hf_model_id + + print(json.dumps(output, indent=2, ensure_ascii=False)) + + eprint(f"\nTranscription complete in {elapsed:.1f}s") + eprint(f" Text: {txt_path}") + eprint(f" SRT: {srt_path}") + eprint(f" JSON: {json_path}") + + return 0 + + +def main(): + parser = argparse.ArgumentParser(description="Whisper Transcribe — Local audio transcription and translation") + subparsers = parser.add_subparsers(dest="command", required=True) + + # Common args for transcribe/translate + def add_common_args(p): + p.add_argument("file", help="Path to audio or video file") + p.add_argument("--model", default="base", choices=["tiny", "base", "small", "medium", "large"], + help="Whisper model size (default: base). Ignored when using a HuggingFace model.") + p.add_argument("--language", default=None, + help="Source language code (e.g. 'en', 'te', 'hi'). Auto-detected if not set. " + "For Telugu ('te'), automatically uses vasista22/whisper-telugu-large-v2.") + p.add_argument("--hf-model", default=None, + help="HuggingFace model ID to use instead of standard Whisper " + "(e.g. 'vasista22/whisper-telugu-large-v2').") + p.add_argument("--output-dir", default="~/Downloads", + help="Output directory (default: ~/Downloads)") + + # --- transcribe --- + p_transcribe = subparsers.add_parser("transcribe", help="Transcribe audio/video to text") + add_common_args(p_transcribe) + + # --- translate --- + p_translate = subparsers.add_parser("translate", help="Transcribe and translate to English") + add_common_args(p_translate) + + # --- detect --- + p_detect = subparsers.add_parser("detect", help="Detect language of audio") + p_detect.add_argument("file", help="Path to audio or video file") + p_detect.add_argument("--model", default="base", choices=["tiny", "base", "small", "medium", "large"], + help="Whisper model size (default: base)") + + # --- info --- + p_info = subparsers.add_parser("info", help="Show audio file metadata") + p_info.add_argument("file", help="Path to audio or video file") + + args = parser.parse_args() + + if args.command == "info": + return cmd_info(args) + elif args.command == "detect": + return cmd_detect(args) + elif args.command in ("transcribe", "translate"): + args.task = "translate" if args.command == "translate" else "transcribe" + return cmd_transcribe(args) + + +if __name__ == "__main__": + sys.exit(main() or 0)