""" YouTube Shorts Scraper using Piped API Piped is a privacy-friendly YouTube proxy that works without JS """ import requests import json import time import threading _cache = {} _lock = threading.Lock() CACHE_TTL = 900 # 15 min # Piped API instances (public) PIPED_INSTANCES = [ "https://pipedapi.kavin.rocks", "https://pipedapi.adminforge.de", "https://api.piped.projectsegfau.lt", ] UA = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} 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 _piped_request(path, params=None): """Try multiple Piped instances""" last_err = None for base in PIPED_INSTANCES: try: url = f"{base}{path}" r = requests.get(url, params=params, headers=UA, timeout=15) if r.status_code == 200: return r.json() except Exception as e: last_err = e continue raise Exception(f"All Piped instances failed: {last_err}") def get_channel_videos(channel_id, max_videos=200): """Get all videos from a channel using Piped API with pagination""" cached = _cached(f'ch_vids_{channel_id}') if cached is not None: return cached all_videos = [] page = None while len(all_videos) < max_videos: try: if page: data = _piped_request(f"/channels/{channel_id}/videos", {"nextpage": page}) else: data = _piped_request(f"/channels/{channel_id}/videos") videos = data.get('relatedStreams', []) if not videos: break all_videos.extend(videos) # Check for next page next_page = data.get('nextpage') if not next_page or next_page == page: break page = next_page # Small delay to be polite time.sleep(0.3) if len(all_videos) >= max_videos: break except Exception as e: print(f"Piped pagination error: {e}") break result = all_videos[:max_videos] _set_cache(f'ch_vids_{channel_id}', result) return result def get_vtvnambo_shorts_piped(max_count=50): """Get shorts from VTV Nam Bộ using Piped API""" # VTV Nam Bộ channel ID channel_id = "UCJ0btJV8qh7J7R2aXb9GmGA" try: videos = get_channel_videos(channel_id, 200) shorts = [] for v in videos: title = v.get('title', '') vid = v.get('url', '').replace('/watch?v=', '') if not vid: continue # Filter for shorts: title has #shorts, or duration <= 60s duration = v.get('duration', 0) is_short = ( '#shorts' in title.lower() or '#short' in title.lower() or (duration > 0 and duration <= 60) ) if is_short: shorts.append({ 'id': vid, 'title': title, 'img': f"https://i.ytimg.com/vi/{vid}/hqdefault.jpg", 'channel': 'vtvnambo', }) if shorts: return shorts[:max_count] except Exception as e: print(f"Piped shorts error: {e}") return [] def get_vtvnambo_shorts_rss(max_count=50): """Get shorts from YouTube RSS feed""" cached = _cached('vtvnambo_rss') if cached is not None: return cached from xml.etree import ElementTree as ET # First get channel ID from page channel_id = None try: r = requests.get("https://www.youtube.com/@vtvnambo", headers=UA, timeout=15) if r.status_code == 200: m = re.search(r'"channelId":"(UC[^"]+)"', r.text) if m: channel_id = m.group(1) except: pass if not channel_id: channel_id = "UCJ0btJV8qh7J7R2aXb9GmGA" # fallback try: url = f"https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}" r = requests.get(url, headers=UA, timeout=15) if r.status_code != 200: return [] root = ET.fromstring(r.text) ns = {'atom': 'http://www.w3.org/2005/Atom', 'yt': 'http://www.youtube.com/xml/schemas/2015'} shorts = [] for entry in root.findall('atom:entry', ns)[:max_count * 2]: title_el = entry.find('atom:title', ns) title = title_el.text if title_el is not None and title_el.text else '' vid_el = entry.find('yt:videoId', ns) vid = vid_el.text if vid_el is not None else '' if not vid: continue is_short = '#shorts' in title.lower() or '#short' in title.lower() link_el = entry.find('atom:link', ns) link = link_el.get('href', '') if link_el is not None else '' if '/shorts/' in link: is_short = True if is_short: shorts.append({ 'id': vid, 'title': title, 'img': f"https://i.ytimg.com/vi/{vid}/hqdefault.jpg", 'channel': 'vtvnambo', }) _set_cache('vtvnambo_rss', shorts[:max_count]) return shorts[:max_count] except Exception as e: print(f"RSS error: {e}") return [] def get_vtvnambo_shorts(max_count=50): """Get all shorts from VTV Nam Bộ. Tries Piped API first, then RSS.""" cached = _cached('vtvnambo_shorts_v3') if cached is not None: return cached all_shorts = [] seen_ids = set() # Method 1: Piped API (most reliable) try: piped_shorts = get_vtvnambo_shorts_piped(max_count) for s in piped_shorts: if s['id'] not in seen_ids: seen_ids.add(s['id']) all_shorts.append(s) print(f"Piped API found {len(piped_shorts)} shorts") except Exception as e: print(f"Piped method failed: {e}") # Method 2: RSS feed if len(all_shorts) < 3: try: rss_shorts = get_vtvnambo_shorts_rss(max_count) for s in rss_shorts: if s['id'] not in seen_ids: seen_ids.add(s['id']) all_shorts.append(s) print(f"RSS found {len(rss_shorts)} shorts") except Exception as e: print(f"RSS method failed: {e}") result = all_shorts[:max_count] _set_cache('vtvnambo_shorts_v3', result) 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] import re # Alias for backward compatibility get_vtvnamo_shorts = get_vtvnambo_shorts