"""Video downloader for public Google Drive (gdown).""" import re import sys from pathlib import Path from typing import Optional class GoogleDriveDownloader: """Handles downloading public videos from Google Drive URLs.""" GDRIVE_PATTERNS = [ r'https?://drive\.google\.com/file/d/[\w-]+', r'https?://drive\.google\.com/open\?.*id=[\w-]+', r'https?://drive\.google\.com/uc\?.*id=[\w-]+', ] def __init__(self, cache_dir: Optional[Path] = None) -> None: self.cache_dir = cache_dir or Path("google-drive") self.cache_dir.mkdir(parents=True, exist_ok=True) def is_google_drive_url(self, url: str) -> bool: """Check if string is a public Google Drive URL.""" return any(re.search(pattern, url) for pattern in self.GDRIVE_PATTERNS) def download(self, url: str) -> Path: """ Download video from a public Google Drive URL. Returns: Path to downloaded video file Raises: ValueError for invalid URL, RuntimeError for download failures (e.g. private file) """ if not self.is_google_drive_url(url): raise ValueError(f"Invalid Google Drive URL: {url}") file_id = self._extract_file_id(url) cached = self._get_cached_file(file_id) if cached: print(f"Using cached video: {cached}", file=sys.stderr) return cached output_path = self.cache_dir / f"{file_id}.mp4" try: import gdown result = gdown.download(id=file_id, output=str(output_path), quiet=False) if result is None: raise RuntimeError( "Download failed — file may be private, restricted, or the URL is invalid." ) downloaded = Path(result) if not downloaded.exists(): raise RuntimeError(f"Download completed but file not found: {downloaded}") return downloaded except RuntimeError: raise except Exception as e: if output_path.exists(): output_path.unlink() raise RuntimeError(f"Failed to download from Google Drive: {e}") def _extract_file_id(self, url: str) -> str: """Extract file ID from a Google Drive URL.""" patterns = [ r'drive\.google\.com/file/d/([\w-]+)', r'[?&]id=([\w-]+)', ] for pattern in patterns: match = re.search(pattern, url) if match: return match.group(1) raise ValueError(f"Could not extract file ID from URL: {url}") def _get_cached_file(self, file_id: str) -> Optional[Path]: """Check if file already downloaded.""" for ext in ['.mp4', '.webm', '.mkv', '.mov']: cached = self.cache_dir / f"{file_id}{ext}" if cached.exists(): return cached return None