import asyncio import os import html import logging import aiosqlite import httpx import xml.etree.ElementTree as ET from googleapiclient.discovery import build from googleapiclient.errors import HttpError from auth import get_credentials from db import DB_PATH log = logging.getLogger(__name__) ATOM_NS = "http://www.w3.org/2005/Atom" async def _fetch_rss(client, channel_id): """Fetch latest videos from a channel's RSS feed (free). Returns list of (video_id, title).""" url = f"https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}" try: resp = await client.get(url, timeout=15) if resp.status_code != 200: return [] root = ET.fromstring(resp.text) entries = [] for entry in root.findall(f"{{{ATOM_NS}}}entry"): vid = entry.find(f"{{{ATOM_NS}}}link").get("href").split("=")[-1] title_el = entry.find(f"{{{ATOM_NS}}}title") title = html.unescape(title_el.text) if title_el is not None and title_el.text else "Untitled" entries.append((vid, title)) return entries except Exception as e: log.warning(f"RSS error for {channel_id}: {e}") return [] async def fetch_subscriptions(): """Fetch latest videos from subscribed channels using RSS + YouTube Data API v3.""" try: creds = get_credentials() except Exception as e: log.error(f"OAuth credential error: {e}") return 0 youtube = build("youtube", "v3", credentials=creds) # Step 1: Get subscribed channel IDs (1 unit) channel_ids = [] try: request = youtube.subscriptions().list(part="snippet", mine=True, maxResults=50, order="alphabetical") while request: response = request.execute() for item in response.get("items", []): cid = item["snippet"]["resourceId"]["channelId"] if cid not in channel_ids: channel_ids.append(cid) request = youtube.subscriptions().list_next(request, response) except HttpError as e: log.error(f"YouTube API error: {e}") return 0 log.info(f"Found {len(channel_ids)} subscribed channels") # Step 2: Fetch RSS from all channels (free, parallel) rss_videos = {} async with httpx.AsyncClient(timeout=20) as client: tasks = [_fetch_rss(client, cid) for cid in channel_ids] results = await asyncio.gather(*tasks) for cid, entries in zip(channel_ids, results): if entries: rss_videos[cid] = entries all_video_ids = [] rss_map = {} for cid, entries in rss_videos.items(): for vid, title in entries: if vid not in all_video_ids: all_video_ids.append(vid) rss_map[vid] = (cid, title) log.info(f"RSS discovered {len(all_video_ids)} videos from {len(rss_videos)} channels") if not all_video_ids: return 0 # Step 3: Check which videos are new (in SQLite) async with aiosqlite.connect(DB_PATH) as db: new_ids = [] for vid in all_video_ids: cursor = await db.execute("SELECT 1 FROM videos WHERE video_id = ?", (vid,)) if not await cursor.fetchone(): new_ids.append(vid) log.info(f"{len(new_ids)} new videos to fetch metadata for") if not new_ids: return 0 # Step 4: Fetch metadata via videos.list (1 unit per 50 videos) new_count = 0 async with aiosqlite.connect(DB_PATH) as db: for i in range(0, len(new_ids), 50): batch = new_ids[i:i+50] try: resp = youtube.videos().list(part="snippet", id=",".join(batch)).execute() for item in resp.get("items", []): vid = item["id"] snippet = item["snippet"] cid, rss_title = rss_map.get(vid, (snippet.get("channelId", ""), snippet.get("title", ""))) await db.execute(""" INSERT INTO videos (video_id, title, channel_name, channel_id, url, thumbnail_url, publish_date) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( vid, html.unescape(snippet.get("title", rss_title)), html.unescape(snippet.get("channelTitle", "Unknown")), snippet.get("channelId", cid), f"https://www.youtube.com/watch?v={vid}", snippet.get("thumbnails", {}).get("high", {}).get("url", ""), (snippet.get("publishTime") or "")[:10] )) new_count += 1 except HttpError as e: log.warning(f"videos.list error: {e}") continue await db.commit() log.info(f"Ingestion completed. Found {new_count} new videos.") return new_count if __name__ == "__main__": import sys logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') async def main(): from db import init_db await init_db() count = await fetch_subscriptions() print(f"\nDone. {count} new videos ingested.") asyncio.run(main())