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
 13
 14
 15def get_video_info(path: str) -> dict:
 16    """Return basic metadata about a video file: duration (seconds), width, height, and frame rate."""
 17    import av
 18    container = av.open(path)
 19    stream = container.streams.video[0]
 20    duration = float(container.duration / av.time_base) if container.duration else None
 21    info = {
 22        "duration_seconds": duration,
 23        "width": stream.width,
 24        "height": stream.height,
 25        "fps": float(stream.average_rate) if stream.average_rate else None,
 26    }
 27    container.close()
 28    return info
 29
 30
 31def get_video_duration(path: str) -> float:
 32    """Return the duration of a video file in seconds."""
 33    return get_video_info(path)["duration_seconds"]
 34
 35
 36def extract_audio_from_video(video_path: str, output_audio_path: str) -> None:
 37    """Extract the audio track from a video file and save it as a standalone audio file
 38    (format is inferred from output_audio_path's extension, e.g. .mp3, .wav, .m4a)."""
 39    import av
 40
 41    input_container = av.open(video_path)
 42    audio_stream = input_container.streams.audio[0]
 43
 44    output_container = av.open(output_audio_path, mode="w")
 45    output_stream = output_container.add_stream("aac" if output_audio_path.endswith((".m4a", ".mp4")) else "mp3")
 46
 47    for frame in input_container.decode(audio_stream):
 48        for packet in output_stream.encode(frame):
 49            output_container.mux(packet)
 50
 51    for packet in output_stream.encode(None):
 52        output_container.mux(packet)
 53
 54    output_container.close()
 55    input_container.close()
 56
 57
 58def extract_video_frame(video_path: str, timestamp_seconds: float, output_image_path: str) -> None:
 59    """Extract a single frame from a video at the given timestamp (seconds) and save it as an image."""
 60    import av
 61
 62    container = av.open(video_path)
 63    stream = container.streams.video[0]
 64
 65    target_pts = int(timestamp_seconds / stream.time_base)
 66    container.seek(target_pts, stream=stream)
 67
 68    for frame in container.decode(stream):
 69        if frame.time >= timestamp_seconds:
 70            frame.to_image().save(output_image_path)
 71            break
 72
 73    container.close()
 74
 75
 76def transcribe_video(path: str, model_size: str = "base", language: str | None = None) -> str:
 77    """Transcribe the spoken audio in a video file to plain text. Works directly on
 78    video containers (MP4, MOV, MKV, etc.) — the audio track is extracted automatically."""
 79    from quicktools.audiotools import transcribe_audio
 80    return transcribe_audio(path, model_size=model_size, language=language)
 81
 82
 83def transcribe_video_with_timestamps(path: str, model_size: str = "base") -> list[dict]:
 84    """Transcribe a video's audio into timestamped segments, each with 'start', 'end', and 'text'."""
 85    from quicktools.audiotools import transcribe_audio_with_timestamps
 86    return transcribe_audio_with_timestamps(path, model_size=model_size)
 87
 88
 89def transcribe_video_word_level(path: str, model_size: str = "base") -> list[dict]:
 90    """Transcribe a video's audio into word-by-word timestamps, each with 'word', 'start', and 'end'.
 91    Useful for generating captions synced precisely to speech."""
 92    from quicktools.audiotools import transcribe_audio_word_level
 93    return transcribe_audio_word_level(path, model_size=model_size)
 94
 95
 96# --- NEW DIARIZATION AND STYLING FEATURES ---
 97
 98from quicktools import audiotools
 99
100def transcribe_video_with_speakers(path_or_url: str, hf_token: str, model_size: str = "base", device: str = "auto") -> list[dict]:
101    """
102    Extracts audio from a local video file or web URL (YouTube, TikTok, IG, X), 
103    transcribes it, and maps the text to specific speakers.
104    """
105    print(f"Processing video source: {path_or_url}")
106    return audiotools.transcribe_with_speakers(path_or_url, hf_token, model_size, device)
107
108
109def save_video_script_to_docx(transcript_data: list[dict], output_path: str, title_text: str = "Video Script & Transcript") -> None:
110    """
111    Formats speaker-mapped video transcript data as a professional script in Word (.docx).
112    """
113    try:
114        from docx import Document
115        from docx.shared import Pt, RGBColor
116    except ImportError:
117        raise ImportError("Saving to Word requires python-docx. Run: pip install python-docx")
118
119    doc = Document()
120    title = doc.add_heading(title_text, level=1)
121    title.alignment = 1  # Center align
122
123    last_speaker = None
124
125    for entry in transcript_data:
126        p = doc.add_paragraph()
127
128        # Format timestamp as [MM:SS]
129        start_m, start_s = divmod(int(entry["start"]), 60)
130        time_str = f"[{start_m:02d}:{start_s:02d}]"
131
132        # Speaker header line if speaker changed
133        if entry["speaker"] != last_speaker:
134            speaker_run = p.add_run(f"{time_str} {entry['speaker']}:\n")
135            speaker_run.bold = True
136            speaker_run.font.color.rgb = RGBColor(112, 48, 160)  # Distinct Purple highlight for video speakers
137            last_speaker = entry["speaker"]
138
139        p.add_run(entry["text"])
140        p.paragraph_format.space_after = Pt(8)
141
142    doc.save(output_path)
143
144
145import os
146import shutil
147import random
148
149def download_video(url: str, output_dir: str = ".", filename: str | None = None, resolution: str = "best", cookiefile: str | None = None) -> str:
150    """
151    Downloads a video file from supported web URLs (YouTube, TikTok, Instagram, X/Twitter, etc.).
152    
153    :param url: The web video URL to download.
154    :param output_dir: Destination folder (defaults to current directory).
155    :param filename: Optional custom filename (without extension). Defaults to video title.
156    :param resolution: Quality target ('best' or 'worst').
157    :param cookiefile: Path to a cookies.txt file to bypass login restrictions.
158    :return: Absolute file path of the downloaded video.
159    """
160    try:
161        import yt_dlp
162    except ImportError:
163        raise ImportError(
164            "Downloading videos requires 'yt-dlp'. Install it with: pip install yt-dlp"
165        )
166
167    os.makedirs(output_dir, exist_ok=True)
168    out_template = f"{filename}.%(ext)s" if filename else "%(title)s.%(ext)s"
169    target_path = os.path.join(output_dir, out_template)
170
171    # --- AUTO-DETECT SERVER COOKIES ---
172    if not cookiefile and os.path.exists('/app/cookies.txt'):
173        cookiefile = '/app/cookies.txt'
174    # ----------------------------------
175
176    # --- THE SMART FFMPEG FALLBACK ---
177    has_ffmpeg = shutil.which("ffmpeg") is not None
178    
179    if resolution == "best":
180        if has_ffmpeg:
181            # Grab max quality separate streams and merge them
182            fmt = 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best'
183        else:
184            # Fall back to the best PRE-MERGED format (usually 720p max) to prevent a crash
185            print("⚠️ [quicktools] FFmpeg not found. Falling back to pre-merged 720p format.")
186            print("💡 Tip: Install FFmpeg ('winget install ffmpeg') to enable 1080p+ downloads.")
187            fmt = 'best[ext=mp4]/best'
188    else:
189        fmt = 'worst'
190    # ---------------------------------
191
192    ydl_opts = {
193        'format': fmt,
194        'outtmpl': target_path,
195        'quiet': False,
196        'no_warnings': True,
197        'merge_output_format': 'mp4' if has_ffmpeg else None,
198        'extractor_args': {'youtube': ['player_client=android']}, 
199    }
200
201    # --- CRITICAL APPLE iOS FIXES ---
202    if has_ffmpeg:
203        # Force FFmpeg to move the moov atom to the top of the file (Fast Start)
204        ydl_opts['postprocessor_args'] = ['-movflags', '+faststart']
205    # --------------------------------
206
207    # --- ATTACH COOKIE IF AVAILABLE ---
208    if cookiefile and os.path.exists(cookiefile):
209        print(f"🔑 Using cookie authentication file: {cookiefile}")
210        ydl_opts['cookiefile'] = cookiefile
211    # ----------------------------------
212
213    # --- HELPER FUNCTION FOR EXECUTION AND VALIDATION ---
214    def attempt_download(opts: dict) -> str:
215        with yt_dlp.YoutubeDL(opts) as ydl:
216            info = ydl.extract_info(url, download=True)
217            filepath = ydl.prepare_filename(info)
218
219            # Check if output was merged into an mp4 container
220            base, _ = os.path.splitext(filepath)
221            if os.path.exists(f"{base}.mp4"):
222                filepath = f"{base}.mp4"
223
224            # CRITICAL: verify the file actually exists AND has real content.
225            # A silent merge failure can leave a 0-byte or missing file that would
226            # otherwise look "successful" to the caller.
227            if not os.path.exists(filepath):
228                raise RuntimeError(f"Download appeared to succeed but no file was found at {filepath}")
229            file_size = os.path.getsize(filepath)
230            if file_size == 0:
231                raise RuntimeError(f"Downloaded file is empty (0 bytes): {filepath}")
232
233            print(f"✅ Video saved to: {filepath} ({file_size / 1024 / 1024:.1f} MB)")
234            return os.path.abspath(filepath)
235
236    # === ATTEMPT 1: IPv6 Workaround ===
237    print(f"📥 [Attempt 1] Downloading video from {url} via IPv6...")
238    ydl_opts_ipv6 = ydl_opts.copy()
239    ydl_opts_ipv6['source_address'] = '0::0'
240
241    try:
242        return attempt_download(ydl_opts_ipv6)
243    except Exception as e_ipv6:
244        print(f"⚠️ IPv6 Attempt Failed/Blocked: {str(e_ipv6)}")
245
246        # === ATTEMPT 2: Rotating Webshare Proxy Fallback ===
247        proxy_env = os.getenv("RESIDENTIAL_PROXY")
248        if proxy_env:
249            # Parse the comma-separated string of 10 proxies you added to .env
250            proxy_list = [p.strip() for p in proxy_env.split(",") if p.strip()]
251            chosen_proxy = random.choice(proxy_list)
252            
253            # Print masked proxy IP for terminal debugging (hides your username/password)
254            masked_ip = chosen_proxy.split('@')[-1] if '@' in chosen_proxy else chosen_proxy
255            print(f"🛡️ [Attempt 2] Routing download through random Webshare proxy ({masked_ip})...")
256            
257            ydl_opts_proxy = ydl_opts.copy()
258            ydl_opts_proxy['proxy'] = chosen_proxy
259
260            try:
261                return attempt_download(ydl_opts_proxy)
262            except Exception as e_proxy:
263                print(f"❌ Proxy Attempt Failed: {str(e_proxy)}")
264                raise RuntimeError(f"Both IPv6 and Proxy downloads failed. Last error: {str(e_proxy)}")
265        else:
266            print("❌ No RESIDENTIAL_PROXY found in environment variables.")
267            raise RuntimeError(f"IPv6 download failed ({str(e_ipv6)}), and no Webshare Proxy was configured.")
268
269
270import subprocess
271def convert_video_to_animated(input_path: str, output_path: str, target_format: str = "gif", fps: int = 15, width: int = 512) -> None:
272    """Converts a video to an animated GIF or WebP sticker using FFmpeg."""
273    if not shutil.which("ffmpeg"):
274        raise RuntimeError("FFmpeg is required for video-to-animation conversion. Please install it.")
275    
276    # WhatsApp/Telegram stickers prefer a 512px boundary. We scale proportionally.
277    scale_filter = f"fps={fps},scale={width}:-1:flags=lanczos"
278    
279    if target_format.lower() == "gif":
280        # High-quality GIF generation using a 2-pass color palette
281        vf_cmd = f"{scale_filter},split[s0][s1];[s0]palettegen[p];[s1][p]paletteuse"
282        cmd = ["ffmpeg", "-y", "-i", input_path, "-vf", vf_cmd, "-loop", "0", output_path]
283    elif target_format.lower() == "webp":
284        # Animated WebP optimized for social media stickers
285        cmd = [
286            "ffmpeg", "-y", "-i", input_path, 
287            "-vcodec", "libwebp", 
288            "-vf", scale_filter,
289            "-lossless", "0", 
290            "-compression_level", "4", 
291            "-q:v", "50", 
292            "-loop", "0", 
293            "-preset", "picture", 
294            "-an", "-vsync", "0", 
295            output_path
296        ]
297    else:
298        raise ValueError("Target format must be 'gif' or 'webp'")
299
300    # Execute FFmpeg silently
301    result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
302    if result.returncode != 0:
303        raise RuntimeError(f"FFmpeg conversion failed: {result.stderr}")