quicktools.videotools

  1"""Video utilities: transcription, audio extraction, metadata, frame capture, 
  2and speaker diarization — powered by PyAV, faster-whisper, and Pyannote.
  3
  4Requires the optional 'av' package: pip install av
  5(This is already installed automatically as a dependency of faster-whisper.)
  6
  7Supports common video containers: MP4, MOV, MKV, AVI, WEBM, and more — since PyAV
  8uses FFmpeg's decoding engine internally, the same one that powers audiotools.
  9"""
 10import os
 11import shutil
 12import tempfile
 13import random
 14import subprocess
 15import requests
 16
 17
 18def get_video_info(path: str) -> dict:
 19    """Return basic metadata about a video file: duration (seconds), width, height, and frame rate."""
 20    import av
 21    container = av.open(path)
 22    stream = container.streams.video[0]
 23    duration = float(container.duration / av.time_base) if container.duration else None
 24    info = {
 25        "duration_seconds": duration,
 26        "width": stream.width,
 27        "height": stream.height,
 28        "fps": float(stream.average_rate) if stream.average_rate else None,
 29    }
 30    container.close()
 31    return info
 32
 33
 34def get_video_duration(path: str) -> float:
 35    """Return the duration of a video file in seconds."""
 36    return get_video_info(path)["duration_seconds"]
 37
 38
 39def extract_audio_from_video(video_path: str, output_audio_path: str) -> None:
 40    """Extract the audio track from a video file and save it as a standalone audio file
 41    (format is inferred from output_audio_path's extension, e.g. .mp3, .wav, .m4a)."""
 42    import av
 43
 44    input_container = av.open(video_path)
 45    audio_stream = input_container.streams.audio[0]
 46
 47    output_container = av.open(output_audio_path, mode="w")
 48    output_stream = output_container.add_stream("aac" if output_audio_path.endswith((".m4a", ".mp4")) else "mp3")
 49
 50    for frame in input_container.decode(audio_stream):
 51        for packet in output_stream.encode(frame):
 52            output_container.mux(packet)
 53
 54    for packet in output_stream.encode(None):
 55        output_container.mux(packet)
 56
 57    output_container.close()
 58    input_container.close()
 59
 60
 61def extract_video_frame(video_path: str, timestamp_seconds: float, output_image_path: str) -> None:
 62    """Extract a single frame from a video at the given timestamp (seconds) and save it as an image."""
 63    import av
 64
 65    container = av.open(video_path)
 66    stream = container.streams.video[0]
 67
 68    target_pts = int(timestamp_seconds / stream.time_base)
 69    container.seek(target_pts, stream=stream)
 70
 71    for frame in container.decode(stream):
 72        if frame.time >= timestamp_seconds:
 73            frame.to_image().save(output_image_path)
 74            break
 75
 76    container.close()
 77
 78
 79def transcribe_video(path: str, model_size: str = "base", language: str | None = None) -> str:
 80    """Transcribe the spoken audio in a video file to plain text. Works directly on
 81    video containers (MP4, MOV, MKV, etc.) — the audio track is extracted automatically."""
 82    from quicktools.audiotools import transcribe_audio
 83    return transcribe_audio(path, model_size=model_size, language=language)
 84
 85
 86def transcribe_video_with_timestamps(path: str, model_size: str = "base") -> list[dict]:
 87    """Transcribe a video's audio into timestamped segments, each with 'start', 'end', and 'text'."""
 88    from quicktools.audiotools import transcribe_audio_with_timestamps
 89    return transcribe_audio_with_timestamps(path, model_size=model_size)
 90
 91
 92def transcribe_video_word_level(path: str, model_size: str = "base") -> list[dict]:
 93    """Transcribe a video's audio into word-by-word timestamps, each with 'word', 'start', and 'end'.
 94    Useful for generating captions synced precisely to speech."""
 95    from quicktools.audiotools import transcribe_audio_word_level
 96    return transcribe_audio_word_level(path, model_size=model_size)
 97
 98
 99# --- NEW DIARIZATION AND STYLING FEATURES ---
100
101from quicktools import audiotools
102
103def transcribe_video_with_speakers(path_or_url: str, hf_token: str, model_size: str = "base", device: str = "auto") -> list[dict]:
104    """
105    Extracts audio from a local video file or web URL (YouTube, TikTok, IG, X), 
106    transcribes it, and maps the text to specific speakers.
107    """
108    print(f"Processing video source: {path_or_url}")
109    return audiotools.transcribe_with_speakers(path_or_url, hf_token, model_size, device)
110
111
112def save_video_script_to_docx(transcript_data: list[dict], output_path: str, title_text: str = "Video Script & Transcript") -> None:
113    """
114    Formats speaker-mapped video transcript data as a professional script in Word (.docx).
115    """
116    try:
117        from docx import Document
118        from docx.shared import Pt, RGBColor
119    except ImportError:
120        raise ImportError("Saving to Word requires python-docx. Run: pip install python-docx")
121
122    doc = Document()
123    title = doc.add_heading(title_text, level=1)
124    title.alignment = 1  # Center align
125
126    last_speaker = None
127
128    for entry in transcript_data:
129        p = doc.add_paragraph()
130
131        # Format timestamp as [MM:SS]
132        start_m, start_s = divmod(int(entry["start"]), 60)
133        time_str = f"[{start_m:02d}:{start_s:02d}]"
134
135        # Speaker header line if speaker changed
136        if entry["speaker"] != last_speaker:
137            speaker_run = p.add_run(f"{time_str} {entry['speaker']}:\n")
138            speaker_run.bold = True
139            speaker_run.font.color.rgb = RGBColor(112, 48, 160)  # Distinct Purple highlight for video speakers
140            last_speaker = entry["speaker"]
141
142        p.add_run(entry["text"])
143        p.paragraph_format.space_after = Pt(8)
144
145    doc.save(output_path)
146
147
148
149def download_video(url: str, output_dir: str = ".", filename: str | None = None, resolution: str = "best", cookiefile: str | None = None) -> str:
150    try:
151        import yt_dlp
152    except ImportError:
153        raise ImportError("Downloading videos requires 'yt-dlp'. Install it with: pip install yt-dlp")
154
155    os.makedirs(output_dir, exist_ok=True)
156    out_template = f"{filename}.%(ext)s" if filename else "%(title)s.%(ext)s"
157    target_path = os.path.join(output_dir, out_template)
158
159    # 1. RESOLVE SHORTLINKS & NORMALIZE X.COM
160    if "x.com" in url:
161        url = url.replace("x.com", "twitter.com")
162
163    final_url = url
164    if "vt.tiktok.com" in url or "vm.tiktok.com" in url:
165        try:
166            resp = requests.head(url, allow_redirects=True, timeout=8, headers={'User-Agent': 'Mozilla/5.0'})
167            final_url = resp.url
168            print(f"  [quicktools] Resolved shortlink: {url} -> {final_url}")
169        except Exception as e_redirect:
170            print(f"  [quicktools] Shortlink resolution warning: {str(e_redirect)}")
171
172    if not cookiefile and os.path.exists('/app/cookies.txt'):
173        cookiefile = '/app/cookies.txt'
174
175    has_ffmpeg = shutil.which("ffmpeg") is not None
176    if resolution == "best":
177        fmt = 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best' if has_ffmpeg else 'best[ext=mp4]/best'
178    else:
179        fmt = 'worst'
180
181    ydl_opts = {
182        'format': fmt,
183        'outtmpl': target_path,
184        'quiet': False,
185        'no_warnings': True,
186        'socket_timeout': 15,
187        'merge_output_format': 'mp4' if has_ffmpeg else None,
188        'http_headers': {
189            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
190        },
191        # === THE CORRECTED API BACKDOORS (Wrapped in Lists!) ===
192        'extractor_args': {
193            'youtube': {'player_client': ['android']},
194            'twitter': {'api': ['syndication']},   
195            'tiktok': {'app_version': ['123456']}  
196        }
197    }
198
199    if has_ffmpeg:
200        ydl_opts['postprocessor_args'] = ['-movflags', '+faststart']
201
202    if cookiefile and os.path.exists(cookiefile):
203        ydl_opts['cookiefile'] = cookiefile
204
205    def attempt_download(opts: dict, target_url: str) -> str:
206        with yt_dlp.YoutubeDL(opts) as ydl:
207            info = ydl.extract_info(target_url, download=True)
208            filepath = ydl.prepare_filename(info)
209            base, _ = os.path.splitext(filepath)
210            if os.path.exists(f"{base}.mp4"):
211                filepath = f"{base}.mp4"
212            if not os.path.exists(filepath) or os.path.getsize(filepath) == 0:
213                raise RuntimeError("Download succeeded but file is missing or empty.")
214            return os.path.abspath(filepath)
215
216    # === ATTEMPT 1: Direct IPv6 ===
217    print(f"  [Attempt 1] Direct IPv6 download from {final_url}...")
218    ydl_opts_ipv6 = ydl_opts.copy()
219    ydl_opts_ipv6['source_address'] = '0::0'
220    try:
221        return attempt_download(ydl_opts_ipv6, final_url)
222    except Exception as e_ipv6:
223        print(f"  IPv6 Attempt Failed: {str(e_ipv6)}")
224
225    # === ATTEMPT 2: Direct IPv4 ===
226    print(f"  [Attempt 2] Direct IPv4 download...")
227    ydl_opts_ipv4 = ydl_opts.copy()
228    try:
229        return attempt_download(ydl_opts_ipv4, final_url)
230    except Exception as e_ipv4:
231        print(f"  IPv4 Attempt Failed: {str(e_ipv4)}")
232
233    # === ATTEMPT 3: Rotating Proxy Endpoint ===
234    proxy_url = os.getenv("RESIDENTIAL_PROXY")
235    if proxy_url:
236        print("  [Attempt 3] Routing through Rotating Proxy Endpoint...")
237        ydl_opts_proxy = ydl_opts.copy()
238        ydl_opts_proxy['proxy'] = proxy_url.strip()
239        try:
240            return attempt_download(ydl_opts_proxy, final_url)
241        except Exception as e_proxy:
242            raise RuntimeError(f"Rotating Proxy failed: {str(e_proxy)}")
243    else:
244        raise RuntimeError(f"Direct downloads failed (IPv6/IPv4), and no rotating proxy configured. Last error: {str(e_ipv4)}")
245
246
247import subprocess
248def convert_video_to_animated(input_path: str, output_path: str, target_format: str = "gif", fps: int = 15, width: int = 512) -> None:
249    """Converts a video to an animated GIF or WebP sticker using FFmpeg."""
250    if not shutil.which("ffmpeg"):
251        raise RuntimeError("FFmpeg is required for video-to-animation conversion. Please install it.")
252    
253    # WhatsApp/Telegram stickers prefer a 512px boundary. We scale proportionally.
254    scale_filter = f"fps={fps},scale={width}:-1:flags=lanczos"
255    
256    if target_format.lower() == "gif":
257        # High-quality GIF generation using a 2-pass color palette
258        vf_cmd = f"{scale_filter},split[s0][s1];[s0]palettegen[p];[s1][p]paletteuse"
259        cmd = ["ffmpeg", "-y", "-i", input_path, "-vf", vf_cmd, "-loop", "0", output_path]
260    elif target_format.lower() == "webp":
261        # Force exact 512x512 transparent canvas, 15fps, and loop infinitely for WhatsApp
262        wa_filter = "fps=15,scale=512:512:force_original_aspect_ratio=decrease,format=rgba,pad=512:512:(ow-iw)/2:(oh-ih)/2:color=#00000000"
263        cmd = [
264            "ffmpeg", "-y", "-i", input_path, 
265            "-vcodec", "libwebp", 
266            "-vf", wa_filter,
267            "-lossless", "0", 
268            "-compression_level", "6", 
269            "-q:v", "40", 
270            "-loop", "0", 
271            "-preset", "picture", 
272            "-an", "-vsync", "0", 
273            output_path
274        ]
275    else:
276        raise ValueError("Target format must be 'gif' or 'webp'")
277
278    # Execute FFmpeg silently
279    result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
280    if result.returncode != 0:
281        raise RuntimeError(f"FFmpeg conversion failed: {result.stderr}")