Upload vtv_shorts.py

vtv_shorts.py

@@ -1,140 +1,33 @@
-"""
-VTV Nam Bộ YouTube Shorts Scraper
-"""
-import requests
-import re
-import json
-import subprocess
-import html as html_lib
-import time
-import threading
-from xml.etree import ElementTree as ET
+"""VTV Shorts - yt-dlp based"""
+import subprocess, json, re, os, time
 
-_cache = {}
-_lock = threading.Lock()
-CACHE_TTL = 1800
+HEADERS = {"User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36","Accept-Language":"vi-VN,vi;q=0.9,en;q=0.8"}
 
-UA = {
-    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
-    "Accept-Language": "vi-VN,vi;q=0.9,en;q=0.8",
-}
-
-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 get_channel_id(username):
-    cached = _cached(f'ch_id_{username}')
-    if cached:
-        return cached
-    channel_id = None
-    try:
-        url = f"https://www.youtube.com/@{username}"
-        r = requests.get(url, headers=UA, timeout=15)
-        if r.status_code == 200:
-            m = re.search(r'<meta\s+property="og:url"\s+content="https://www.youtube.com/channel/(UC[^"]+)"', r.text)
-            if m:
-                channel_id = m.group(1)
-            if not channel_id:
-                m = re.search(r'"channelId":"(UC[^"]+)"', r.text)
-                if m:
-                    channel_id = m.group(1)
-    except Exception as e:
-        print(f"Error getting channel ID for @{username}: {e}")
-    if channel_id:
-        _set_cache(f'ch_id_{username}', channel_id)
-    return channel_id
-
-def scrape_via_rss(channel_id, max_count=100):
-    shorts = []
+def _yt_dlp_search(query, count=10):
     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 shorts
-        root = ET.fromstring(r.text)
-        ns = {'atom': 'http://www.w3.org/2005/Atom', 'yt': 'http://www.youtube.com/xml/schemas/2015'}
-        for entry in root.findall('atom:entry', ns)[:max_count]:
-            title_el = entry.find('atom:title', ns)
-            title = html_lib.unescape(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'})
-    except Exception as e:
-        print(f"RSS scrape error: {e}")
-    return shorts
-
-def scrape_via_ydlp(username, count=50):
-    shorts = []
-    try:
-        url = f"https://www.youtube.com/@{username}/shorts"
         result = subprocess.run(
-            ["yt-dlp", "--dump-json", "--flat-playlist", "--no-download", "--playlist-end", str(count), url],
-            capture_output=True, text=True, timeout=90
+            ["yt-dlp", "--flat-playlist", "--print-json",
+             f"ytsearch{count}:{query}", "--quiet"],
+            capture_output=True, text=True, timeout=30
         )
-        if result.returncode == 0 and result.stdout.strip():
-            seen_ids = set()
-            for line in result.stdout.strip().split('\n'):
-                line = line.strip()
-                if not line:
-                    continue
-                try:
-                    entry = json.loads(line)
-                    vid = entry.get('id', '')
-                    if not vid or vid in seen_ids:
-                        continue
-                    seen_ids.add(vid)
-                    title = entry.get('title', 'VTV Nam Bộ Short')
-                    shorts.append({'id': vid, 'title': title, 'img': f"https://i.ytimg.com/vi/{vid}/hqdefault.jpg", 'channel': 'vtvnambo'})
-                except json.JSONDecodeError:
-                    continue
-    except (subprocess.TimeoutExpired, FileNotFoundError) as e:
-        print(f"yt-dlp error: {e}")
-    return shorts
-
-def get_vtvnambo_shorts(max_count=50):
-    cached = _cached('vtvnambo_shorts_v2')
-    if cached is not None:
-        return cached
-    all_shorts = []
-    seen_ids = set()
-    channel_id = get_channel_id('vtvnambo')
-    if channel_id:
-        rss_shorts = scrape_via_rss(channel_id, max_count * 2)
-        for s in rss_shorts:
-            if s['id'] not in seen_ids:
-                seen_ids.add(s['id'])
-                all_shorts.append(s)
-    if len(all_shorts) < 3:
-        ydlp_shorts = scrape_via_ydlp('vtvnambo', max_count)
-        for s in ydlp_shorts:
-            if s['id'] not in seen_ids:
-                seen_ids.add(s['id'])
-                all_shorts.append(s)
-    result = all_shorts[:max_count]
-    _set_cache('vtvnambo_shorts_v2', result)
-    return result
-
-get_vtvnamo_shorts = get_vtvnambo_shorts
-
-def get_wc_related_shorts(max_count=30):
-    all_shorts = get_vtvnambo_shorts(max_count * 3)
-    wc_kws = ['world cup', 'wc 2026', 'worldcup', 'fifa', 'bóng đá', 'trận đấu', 'đội tuyển']
-    wc_shorts = [s for s in all_shorts if any(k in s.get('title', '').lower() for k in wc_kws)]
-    if not wc_shorts:
-        wc_shorts = all_shorts
-    return wc_shorts[:max_count]
+        items = []
+        for line in result.stdout.strip().split('\n'):
+            if not line.strip(): continue
+            try:
+                d = json.loads(line)
+                items.append({
+                    "id": d.get("id",""),
+                    "title": d.get("title",""),
+                    "link": d.get("url","") or f"https://www.youtube.com/watch?v={d.get('id','')}",
+                    "img": f"https://i.ytimg.com/vi/{d.get('id','')}/hqdefault.jpg",
+                    "source": "yt"
+                })
+            except: pass
+        return items
+    except: return []
+
+def get_vtvnambo_shorts(count=50):
+    return _yt_dlp_search("VTV Nam Bộ tin tức mới nhất", count)
+
+def get_wc_related_shorts(count=50):
+    return _yt_dlp_search("World Cup 2026 tin tức mới nhất", count)