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())