Upload shorts_updater.py with huggingface_hub

shorts_updater.py

@@ -0,0 +1,300 @@
+"""
+Shorts Auto-Updater: Resilient YouTube shorts fetching with timeout handling
+- Background updates every 10 minutes
+- Hard timeout per channel (25s) to prevent hanging
+- Stale-while-revalidate: serve cached data immediately, update in background
+- Multiprocess isolation for yt-dlp to prevent crashes
+"""
+import json
+import os
+import time
+import threading
+import subprocess
+import signal
+from datetime import datetime, timezone, timedelta
+
+DATA_DIR = "/data" if os.path.isdir("/data") else "/app"
+SHORTS_CACHE_FILE = os.path.join(DATA_DIR, "shorts_cache.json")
+SHORTS_METADATA_FILE = os.path.join(DATA_DIR, "shorts_meta.json")
+
+YOUTUBE_HANDLES = ["baodantri7941", "baosuckhoedoisongboyte", "vtvnambo"]
+YOUTUBE_TIMEOUT = 25  # Hard timeout per channel in seconds
+UPDATE_INTERVAL = 600  # 10 minutes
+STALE_TTL = 3600  # 1 hour - serve stale if no fresh data
+
+_lock = threading.Lock()
+_cache = {"t": 0, "d": [], "error": False}
+_meta = {"last_update": None, "last_success": None, "errors": {}}
+
+
+def _load_cache():
+    """Load shorts cache from disk"""
+    try:
+        if os.path.exists(SHORTS_CACHE_FILE):
+            with open(SHORTS_CACHE_FILE, 'r', encoding='utf-8') as f:
+                return json.load(f)
+    except Exception as e:
+        print(f"Cache load error: {e}")
+    return {"t": 0, "d": [], "error": False}
+
+
+def _save_cache(data):
+    """Save shorts cache to disk atomically"""
+    try:
+        os.makedirs(DATA_DIR, exist_ok=True)
+        tmp = SHORTS_CACHE_FILE + ".tmp"
+        with open(tmp, 'w', encoding='utf-8') as f:
+            json.dump(data, f, ensure_ascii=False)
+        os.replace(tmp, SHORTS_CACHE_FILE)
+    except Exception as e:
+        print(f"Cache save error: {e}")
+
+
+def _load_meta():
+    """Load metadata for tracking update status"""
+    try:
+        if os.path.exists(SHORTS_METADATA_FILE):
+            with open(SHORTS_METADATA_FILE, 'r', encoding='utf-8') as f:
+                return json.load(f)
+    except Exception:
+        pass
+    return {"last_update": None, "last_success": None, "errors": {}}
+
+
+def _save_meta(data):
+    """Save metadata to disk"""
+    try:
+        tmp = SHORTS_METADATA_FILE + ".tmp"
+        with open(tmp, 'w', encoding='utf-8') as f:
+            json.dump(data, f, ensure_ascii=False)
+        os.replace(tmp, SHORTS_METADATA_FILE)
+    except Exception:
+        pass
+
+
+def _fetch_shorts_via_ydlp(username, count=50):
+    """Fetch shorts from a channel using yt-dlp with hard timeout"""
+    shorts = []
+    try:
+        url = f"https://www.youtube.com/@{username}/shorts"
+        
+        # Use subprocess with hard timeout
+        proc = subprocess.Popen(
+            ["yt-dlp", "--dump-json", "--flat-playlist", "--no-download", 
+             "--playlist-end", str(count), "--no-check-certificates",
+             "--user-agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
+             url],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True
+        )
+        
+        try:
+            stdout, stderr = proc.communicate(timeout=YOUTUBE_TIMEOUT)
+        except subprocess.TimeoutExpired:
+            proc.kill()
+            proc.communicate()
+            print(f"yt-dlp timeout for @{username}")
+            return []
+        
+        if proc.returncode != 0:
+            print(f"yt-dlp failed for @{username}: {stderr[:100]}")
+            return []
+        
+        seen_ids = set()
+        for line in 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', 'VNEWS Short')[:120]
+                
+                shorts.append({
+                    'id': vid,
+                    'title': title,
+                    'img': f"https://i.ytimg.com/vi/{vid}/hqdefault.jpg",
+                    'link': f"https://www.youtube.com/shorts/{vid}",
+                    'channel': username,
+                    'source': 'yt'
+                })
+            except json.JSONDecodeError:
+                continue
+            
+            if len(shorts) >= count:
+                break
+                
+    except Exception as e:
+        print(f"yt-dlp error for @{username}: {e}")
+    
+    return shorts
+
+
+def _fetch_all_shorts():
+    """Fetch shorts from all configured channels"""
+    all_shorts = []
+    errors = {}
+    
+    for handle in YOUTUBE_HANDLES:
+        shorts = _fetch_shorts_via_ydlp(handle, 30)
+        if shorts:
+            all_shorts.extend(shorts)
+        else:
+            errors[handle] = "fetch_failed"
+    
+    # Remove duplicates by ID
+    seen = set()
+    unique_shorts = []
+    for s in all_shorts:
+        if s['id'] not in seen:
+            seen.add(s['id'])
+            unique_shorts.append(s)
+    
+    return unique_shorts[:100], errors
+
+
+def update_shorts_background(force=False):
+    """Background update of shorts - call from scheduler"""
+    global _cache, _meta
+    
+    now = time.time()
+    
+    with _lock:
+        if not force and _cache.get("updating"):
+            return
+        _cache["updating"] = True
+    
+    try:
+        old_meta = _load_meta()
+        shorts, errors = _fetch_all_shorts()
+        
+        new_cache = {
+            "t": now,
+            "d": shorts,
+            "error": len(errors) > 0 and len(shorts) == 0,
+            "updating": False
+        }
+        
+        new_meta = {
+            "last_update": datetime.now(timezone.utc).isoformat(),
+            "last_success": datetime.now(timezone.utc).isoformat() if shorts else old_meta.get("last_success"),
+            "errors": errors,
+            "count": len(shorts)
+        }
+        
+        with _lock:
+            _cache = new_cache
+        
+        _save_cache(new_cache)
+        _save_meta(new_meta)
+        
+        print(f"Shorts updated: {len(shorts)} videos, errors: {errors if errors else 'none'}")
+        
+    except Exception as e:
+        print(f"Shorts update error: {e}")
+        with _lock:
+            if "error" not in _cache or not _cache["error"]:
+                _cache["error"] = True
+    finally:
+        with _lock:
+            _cache["updating"] = False
+
+
+def get_shorts(stale_ok=True):
+    """
+    Get shorts with stale-while-revalidate pattern.
+    Returns cached data immediately if available, triggers async update if stale.
+    """
+    global _cache
+    
+    now = time.time()
+    
+    with _lock:
+        cache_copy = dict(_cache)
+    
+    # Load from disk if memory cache is empty
+    if not cache_copy.get("d"):
+        disk_cache = _load_cache()
+        if disk_cache.get("d"):
+            cache_copy = disk_cache
+    
+    # Check if cache is stale
+    cache_age = now - cache_copy.get("t", 0)
+    is_stale = cache_age > 600  # 10 minutes
+    
+    # Trigger background update if stale or empty
+    if is_stale or cache_copy.get("error"):
+        if not cache_copy.get("updating"):
+            threading.Thread(target=update_shorts_background, daemon=True).start()
+    
+    # Return cache (even if stale, but warn timeout)
+    return cache_copy.get("d", [])
+
+
+def get_shorts_with_status():
+    """Get shorts with metadata about freshness"""
+    now = time.time()
+    
+    with _lock:
+        cache = dict(_cache)
+    
+    if not cache.get("d"):
+        cache = _load_cache()
+    
+    meta = _load_meta()
+    cache_age = now - cache.get("t", 0)
+    
+    stale = cache_age > 600
+    
+    return {
+        "shorts": cache.get("d", []),
+        "meta": {
+            "fresh": not stale,
+            "age_seconds": cache_age,
+            "last_update": meta.get("last_update"),
+            "last_success": meta.get("last_success"),
+            "count": len(cache.get("d", [])),
+            "errors": meta.get("errors", {})
+        }
+    }
+
+
+# Initialize - trigger first update on module load
+def init_updater():
+    """Initialize the shorts updater - run on app startup"""
+    # Load existing cache
+    global _cache, _meta
+    _cache = _load_cache()
+    _meta = _load_meta()
+    
+    # Start background thread for periodic updates
+    def scheduler():
+        # First update immediately
+        time.sleep(2)  # Wait for app to be ready
+        update_shorts_background(force=True)
+        
+        while True:
+            time.sleep(UPDATE_INTERVAL)
+            update_shorts_background(force=False)
+    
+    threading.Thread(target=scheduler, daemon=True).start()
+
+
+# Fallback data for when all sources fail
+FALLBACK_SHORTS = [
+    {"id": "nqlLH6chLRo", "title": "Tin nóng VTV Nam Bộ | #shorts", "img": "https://i.ytimg.com/vi/nqlLH6chLRo/hqdefault.jpg", "channel": "vtvnambo", "link": "https://www.youtube.com/shorts/nqlLH6chLRo", "source": "yt"},
+    {"id": "E7Kq0v3hG6w", "title": "VTV Nam Bộ - Tin tức miền Nam | #shorts", "img": "https://i.ytimg.com/vi/E7Kq0v3hG6w/hqdefault.jpg", "channel": "vtvnambo", "link": "https://www.youtube.com/shorts/E7Kq0v3hG6w", "source": "yt"},
+    {"id": "Lu_iCQ5YwNM", "title": "Công an lập hồ sơ xử lý người phụ nữ chửi bới tát nam tài xế ô tô ở Hà Nội", "img": "https://i.ytimg.com/vi/Lu_iCQ5YwNM/hqdefault.jpg", "channel": "baodantri7941", "link": "https://www.youtube.com/shorts/Lu_iCQ5YwNM", "source": "yt"},
+]
+
+
+if __name__ == "__main__":
+    # Test run
+    init_updater()
+    time.sleep(5)
+    result = get_shorts_with_status()
+    print(json.dumps(result, ensure_ascii=False, indent=2))