Spaces:
Running
Running
| """ | |
| 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 | |