From d87c2aba44d06a191313fd10a9284052cf73bf26 Mon Sep 17 00:00:00 2001 From: "Gan, Jimmy" Date: Wed, 24 Jun 2026 02:47:34 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20RSS=20hybrid=20fetcher=20=E2=80=94=2022?= =?UTF-8?q?1=20channels,=202,414=20videos,=20~50=20units=20(was=202,000)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/fetcher.py | 128 ++++++++++++++++++++++++--------------------- 1 file changed, 67 insertions(+), 61 deletions(-) diff --git a/backend/fetcher.py b/backend/fetcher.py index d3518bf..26030c8 100644 --- a/backend/fetcher.py +++ b/backend/fetcher.py @@ -3,6 +3,8 @@ 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 @@ -10,55 +12,42 @@ from db import DB_PATH log = logging.getLogger(__name__) -ROTATION_FILE = os.environ.get("ROTATION_FILE", os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config", "rotation.txt")) +ATOM_NS = "http://www.w3.org/2005/Atom" -def get_rotation_batch(channel_ids, batch_size=50): - """Read last position, return next batch of channels, save new position.""" - offset = 0 +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: - if os.path.exists(ROTATION_FILE): - with open(ROTATION_FILE) as f: - offset = int(f.read().strip()) - except: pass - - total = len(channel_ids) - if offset >= total: - offset = 0 - - end = min(offset + batch_size, total) - batch = channel_ids[offset:end] - - # Save next offset - new_offset = end if end < total else 0 - os.makedirs(os.path.dirname(ROTATION_FILE), exist_ok=True) - with open(ROTATION_FILE, 'w') as f: - f.write(str(new_offset)) - - log.info(f"Rotation: channels {offset+1}-{end} of {total} (next starts at {new_offset})") - return batch + 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 YouTube Data API v3.""" + """Fetch latest videos from subscribed channels using RSS + YouTube Data API v3.""" try: creds = get_credentials() - except FileNotFoundError as e: - log.error(f"OAuth setup incomplete: {e}") - return 0 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 + # Step 1: Get subscribed channel IDs (1 unit) channel_ids = [] try: - request = youtube.subscriptions().list( - part="snippet", - mine=True, - maxResults=50, - order="alphabetical" - ) + request = youtube.subscriptions().list(part="snippet", mine=True, maxResults=50, order="alphabetical") while request: response = request.execute() for item in response.get("items", []): @@ -67,57 +56,74 @@ async def fetch_subscriptions(): channel_ids.append(cid) request = youtube.subscriptions().list_next(request, response) except HttpError as e: - log.error(f"YouTube API error fetching subscriptions: {e}") + log.error(f"YouTube API error: {e}") return 0 log.info(f"Found {len(channel_ids)} subscribed channels") - if not channel_ids: - log.warning("No subscribed channels found. Make sure the authenticated account has subscriptions.") + # 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 2: For each channel in today's rotation batch, get latest videos - batch = get_rotation_batch(channel_ids) + # 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 cid in batch: + for i in range(0, len(new_ids), 50): + batch = new_ids[i:i+50] try: - req = youtube.search().list( - part="snippet", - channelId=cid, - maxResults=10, - order="date", - type="video" - ) - resp = req.execute() + resp = youtube.videos().list(part="snippet", id=",".join(batch)).execute() for item in resp.get("items", []): - vid = item["id"]["videoId"] + vid = item["id"] snippet = item["snippet"] + cid, rss_title = rss_map.get(vid, (snippet.get("channelId", ""), snippet.get("title", ""))) - cursor = await db.execute( - "SELECT 1 FROM videos WHERE video_id = ?", (vid,) - ) - if await cursor.fetchone(): - continue - - pub_time = snippet.get("publishTime", "") 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", "Untitled")), + 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", ""), - pub_time[:10] if pub_time else None + (snippet.get("publishTime") or "")[:10] )) new_count += 1 - log.debug(f"New video: {snippet.get('title', 'Untitled')[:50]}") except HttpError as e: - log.warning(f"Channel {cid} fetch error: {e}") + log.warning(f"videos.list error: {e}") continue await db.commit()