""" YouTube Shorts Scraper using yt-dlp (already installed on Space) Runs yt-dlp as subprocess to extract video info and direct URLs """ import subprocess import json import time import threading import os import re as re_mod _cache = {} _lock = threading.Lock() CACHE_TTL = 600 # 10 min cache def _cached(key): with _lock: if key in _cache and time.time() - _cache[key]['t'] < CACHE_TTL: return _cache[key]['d'] return None def _set_cache(key, data): with _lock: _cache[key] = {'t': time.time(), 'd': data} def run_yt_dlp(args, timeout=120): """Run yt-dlp and return parsed JSON lines""" try: result = subprocess.run( ["yt-dlp"] + args, capture_output=True, text=True, timeout=timeout ) if result.returncode != 0 and not result.stdout.strip(): print(f"yt-dlp error: {result.stderr[:200]}") return [] lines = result.stdout.strip().split('\n') items = [] for line in lines: line = line.strip() if not line: continue try: items.append(json.loads(line)) except json.JSONDecodeError: continue return items except subprocess.TimeoutExpired: print("yt-dlp timeout") return [] except FileNotFoundError: print("yt-dlp not found!") return [] except Exception as e: print(f"yt-dlp exception: {e}") return [] def get_channel_shorts_via_playlist(channel_username, max_count=100): """Get shorts from channel's shorts page using yt-dlp""" shorts = [] # Method 1: Fetch from /shorts page url = f"https://www.youtube.com/@{channel_username}/shorts" items = run_yt_dlp([ "--dump-json", "--flat-playlist", "--no-download", "--playlist-end", str(max_count), "--no-check-certificates", "--user-agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", url ], timeout=90) seen_ids = set() for item in items: vid = item.get('id', '') if not vid or vid in seen_ids: continue seen_ids.add(vid) title = item.get('title', 'VTV Nam Bộ Short') duration = item.get('duration', 0) or 0 # Only include actual shorts (<= 120s to be safe) if duration <= 120 or '#shorts' in title.lower() or '#short' in title.lower(): shorts.append({ 'id': vid, 'title': title, 'duration': duration, 'channel': channel_username, 'img': f"https://i.ytimg.com/vi/{vid}/hqdefault.jpg", }) return shorts def get_channel_videos_filter_shorts(channel_username, max_count=200): """Get all videos from /videos page and filter for shorts by duration""" shorts = [] url = f"https://www.youtube.com/@{channel_username}/videos" items = run_yt_dlp([ "--dump-json", "--flat-playlist", "--no-download", "--playlist-end", str(max_count), "--match-filter", "duration > 0 and duration <= 120", "--no-check-certificates", "--user-agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", url ], timeout=120) seen_ids = set() for item in items: vid = item.get('id', '') if not vid or vid in seen_ids: continue seen_ids.add(vid) title = item.get('title', 'VTV Nam Bộ Short') duration = item.get('duration', 0) or 0 shorts.append({ 'id': vid, 'title': title, 'duration': duration, 'channel': channel_username, 'img': f"https://i.ytimg.com/vi/{vid}/hqdefault.jpg", }) return shorts def get_shorts_with_direct_url(video_ids): """Get direct video download URLs for given video IDs""" results = [] for vid in video_ids[:20]: # Limit to avoid timeout try: url = f"https://www.youtube.com/shorts/{vid}" items = run_yt_dlp([ "--dump-json", "--no-download", "--no-check-certificates", "--format", "best[filesize<10M]/best", url ], timeout=30) if items: info = items[0] direct_url = info.get('url', '') if not direct_url: # Try to get from formats formats = info.get('formats', []) for f in formats: if f.get('vcodec') != 'none' and f.get('acodec') != 'none': direct_url = f.get('url', '') break if direct_url: results.append({ 'id': vid, 'title': info.get('title', ''), 'direct_url': direct_url, 'thumbnail': info.get('thumbnail', f"https://i.ytimg.com/vi/{vid}/hqdefault.jpg"), 'duration': info.get('duration', 0), }) except Exception as e: print(f"Error getting URL for {vid}: {e}") return results def get_vtvnambo_shorts(max_count=50): """Get all shorts from VTV Nam Bộ using yt-dlp""" cached = _cached('vtvnambo_shorts_yt') if cached is not None: return cached all_shorts = [] seen_ids = set() # Method 1: /shorts page print(f"[yt-dlp] Fetching /shorts page...") shorts_page = get_channel_shorts_via_playlist('vtvnambo', max_count) for s in shorts_page: if s['id'] not in seen_ids: seen_ids.add(s['id']) all_shorts.append(s) print(f"[yt-dlp] /shorts page: {len(shorts_page)} shorts") # Method 2: /videos page with duration filter if len(all_shorts) < 5: print(f"[yt-dlp] Fetching /videos page with filter...") videos_filtered = get_channel_videos_filter_shorts('vtvnambo', max_count * 2) for s in videos_filtered: if s['id'] not in seen_ids: seen_ids.add(s['id']) all_shorts.append(s) print(f"[yt-dlp] /videos filter: {len(videos_filtered)} shorts") result = all_shorts[:max_count] _set_cache('vtvnambo_shorts_yt', result) print(f"[yt-dlp] Total: {len(result)} shorts from VTV Nam Bộ") return result def get_wc_related_shorts(max_count=30): """Get World Cup / football related shorts""" all_shorts = get_vtvnambo_shorts(max_count * 3) wc_kws = [ 'world cup', 'wc 2026', 'worldcup', 'fifa', 'bóng đá', 'trận đấu', 'đội tuyển', 'tuyển', 'vòng loại', 'khoảnh khắc', 'highlights', 'bàn thắng', 'goal', 'kết quả', 'tỉ số', 'việt nam', 'vn', 'ngoại hạng', 'premier league', 'champions league', 'laliga', 'serie a', 'bundesliga', 'ligue 1', 'copa', 'europa', 'c1', 'c2', 'messi', 'ronaldo', 'neymar', 'mbappe', 'haaland', 'v-league', 'vleague', 'bóng đá việt', 'đội bóng', 'hlv', 'huấn luyện viên', 'chuyển nhượng', 'transfer', 'asian cup', 'aff cup', 'sea games', 'olympic', 'u23', 'u20', 'u17', ] wc_shorts = [] for s in all_shorts: tl = s.get('title', '').lower() if any(k in tl for k in wc_kws): wc_shorts.append(s) if not wc_shorts: wc_shorts = all_shorts return wc_shorts[:max_count] # Aliases get_vtvnamo_shorts = get_vtvnambo_shorts