|
|
@@ -1,11 +1,12 @@
|
|
|
# /// script
|
|
|
-# dependencies = []
|
|
|
+# dependencies = ["requests", "silero-vad", "torch"]
|
|
|
# ///
|
|
|
-"""Extract timed text from video subtitles; grab single frames.
|
|
|
+"""Extract timed text from video subtitles; transcribe via ASR; grab frames.
|
|
|
|
|
|
Commands:
|
|
|
- parse <input> [--outdir <dir>] extract timed text from subtitles
|
|
|
- frame <video> <timestamp> --outpath <path> extract single frame
|
|
|
+ parse <input> [--outdir <dir>] extract timed text from subtitles
|
|
|
+ transcribe <video> [--outpath <path>] transcribe speech via VAD+ASR
|
|
|
+ frame <video> <timestamp> --outpath <p> extract single frame
|
|
|
"""
|
|
|
|
|
|
import argparse
|
|
|
@@ -13,6 +14,7 @@ import os
|
|
|
import re
|
|
|
import subprocess
|
|
|
import sys
|
|
|
+import tempfile
|
|
|
from pathlib import Path
|
|
|
|
|
|
SRT_PATTERN = re.compile(
|
|
|
@@ -164,6 +166,101 @@ def cmd_frame(args: argparse.Namespace) -> None:
|
|
|
print(f"[OK] Frame saved -> {out}")
|
|
|
|
|
|
|
|
|
+def extract_audio(video_path: Path, out_wav: Path) -> None:
|
|
|
+ """Extract 16kHz mono WAV from video."""
|
|
|
+ cmd = [
|
|
|
+ "ffmpeg", "-y", "-i", str(video_path),
|
|
|
+ "-vn", "-ar", "16000", "-ac", "1",
|
|
|
+ str(out_wav),
|
|
|
+ ]
|
|
|
+ subprocess.run(cmd, capture_output=True, check=True)
|
|
|
+
|
|
|
+
|
|
|
+def read_audio_tensor(wav_path: Path) -> "torch.Tensor":
|
|
|
+ """Read WAV file as mono float32 tensor (compat with silero-vad)."""
|
|
|
+ import numpy as np
|
|
|
+ import torch
|
|
|
+ # Use ffmpeg to decode to raw PCM float32
|
|
|
+ result = subprocess.run(
|
|
|
+ ["ffmpeg", "-y", "-i", str(wav_path),
|
|
|
+ "-f", "f32le", "-ac", "1", "-ar", "16000", "-"],
|
|
|
+ capture_output=True, check=True,
|
|
|
+ )
|
|
|
+ arr = np.frombuffer(result.stdout, dtype=np.float32).copy()
|
|
|
+ return torch.from_numpy(arr).squeeze()
|
|
|
+
|
|
|
+
|
|
|
+def _find_model_dir(model_size: str) -> str | None:
|
|
|
+ """Find cached faster-whisper model directory."""
|
|
|
+ repo = f"Systran/faster-whisper-{model_size}"
|
|
|
+ cache = Path(os.path.expanduser("~/.cache/huggingface/hub"))
|
|
|
+ repo_dir = cache / f"models--{repo.replace('/', '--')}"
|
|
|
+ if not repo_dir.exists():
|
|
|
+ return None
|
|
|
+ snapshots = repo_dir / "snapshots"
|
|
|
+ if not snapshots.exists():
|
|
|
+ return None
|
|
|
+ snaps = list(snapshots.iterdir())
|
|
|
+ return str(snaps[0]) if snaps else None
|
|
|
+
|
|
|
+
|
|
|
+def cmd_transcribe(args: argparse.Namespace) -> None:
|
|
|
+ video = Path(args.video)
|
|
|
+ if not video.exists():
|
|
|
+ print(f"[ERR] Video not found: {video}", file=sys.stderr)
|
|
|
+ sys.exit(1)
|
|
|
+
|
|
|
+ outpath = Path(args.outpath) if args.outpath else None
|
|
|
+ model_size = args.model
|
|
|
+
|
|
|
+ model_dir = _find_model_dir(model_size)
|
|
|
+ if model_dir is None:
|
|
|
+ print(f"[ERR] Model '{model_size}' not cached locally", file=sys.stderr)
|
|
|
+ sys.exit(1)
|
|
|
+
|
|
|
+ print(f"[...] Extracting audio from {video.name} ...", file=sys.stderr)
|
|
|
+ wav = Path(tempfile.mktemp(suffix=".wav"))
|
|
|
+ try:
|
|
|
+ extract_audio(video, wav)
|
|
|
+
|
|
|
+ print(f"[...] Loading faster-whisper ({model_size}) ...", file=sys.stderr)
|
|
|
+ import os as _os
|
|
|
+ _os.environ["HF_HUB_DISABLE_SYMLINKS_WARNING"] = "1"
|
|
|
+ from faster_whisper import WhisperModel
|
|
|
+ model = WhisperModel(model_dir, device="cpu", compute_type="int8")
|
|
|
+
|
|
|
+ print(f"[...] Transcribing. This may take a few minutes ...", file=sys.stderr)
|
|
|
+ segments, _info = model.transcribe(
|
|
|
+ str(wav), beam_size=5, vad_filter=True,
|
|
|
+ vad_parameters=dict(min_silence_duration_ms=300),
|
|
|
+ )
|
|
|
+
|
|
|
+ segs: list[tuple[float, str]] = []
|
|
|
+ for seg in segments:
|
|
|
+ text = seg.text.strip()
|
|
|
+ if text:
|
|
|
+ segs.append((seg.start, text))
|
|
|
+
|
|
|
+ print(f"[...] {len(segs)} segments", file=sys.stderr)
|
|
|
+ for s, t in segs:
|
|
|
+ print(f" [{sec_to_ts(s)}] {t[:80]}", file=sys.stderr)
|
|
|
+
|
|
|
+ finally:
|
|
|
+ wav.unlink(missing_ok=True)
|
|
|
+
|
|
|
+ if not segs:
|
|
|
+ print(f"[WARN] No speech transcribed", file=sys.stderr)
|
|
|
+ return
|
|
|
+
|
|
|
+ lines = [f"[{sec_to_ts(s)}] {t}" for s, t in segs]
|
|
|
+ if outpath:
|
|
|
+ outpath.parent.mkdir(parents=True, exist_ok=True)
|
|
|
+ outpath.write_text("\n".join(lines), encoding="utf-8")
|
|
|
+ print(f"[OK] Transcribed -> {outpath} ({len(segs)} segments)")
|
|
|
+ else:
|
|
|
+ print("\n".join(lines))
|
|
|
+
|
|
|
+
|
|
|
def main() -> None:
|
|
|
parser = argparse.ArgumentParser(description="Video subtitle extractor & frame grabber")
|
|
|
sub = parser.add_subparsers(dest="command")
|
|
|
@@ -172,6 +269,11 @@ def main() -> None:
|
|
|
p.add_argument("input", help="Video file or directory")
|
|
|
p.add_argument("--outdir", default="./transcripts", help="Output directory")
|
|
|
|
|
|
+ p = sub.add_parser("transcribe", help="Transcribe speech via local faster-whisper")
|
|
|
+ p.add_argument("video", help="Path to video file")
|
|
|
+ p.add_argument("--outpath", help="Output transcript path (default: stdout)")
|
|
|
+ p.add_argument("--model", default="base", help="Model size: tiny/base/small/medium/large")
|
|
|
+
|
|
|
p = sub.add_parser("frame", help="Extract single frame as JPEG")
|
|
|
p.add_argument("video", help="Path to video file")
|
|
|
p.add_argument("timestamp", help="Timestamp (HH:MM:SS or seconds)")
|
|
|
@@ -180,6 +282,8 @@ def main() -> None:
|
|
|
args = parser.parse_args()
|
|
|
if args.command == "parse":
|
|
|
cmd_parse(args)
|
|
|
+ elif args.command == "transcribe":
|
|
|
+ cmd_transcribe(args)
|
|
|
elif args.command == "frame":
|
|
|
cmd_frame(args)
|
|
|
else:
|
OpenCode