t-youtube/backend/fetcher.py

import asyncio
import os
import json
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"

CHANNELS_PER_SYNC = 50  # Rotate through all channels — N per sync
_ROTATION_FILE = os.path.join(os.environ.get("DB_DIR", os.path.dirname(os.path.abspath(__file__))), "sync_rotation.json")


def _get_rotation_slice(total: int) -> tuple[int, int]:
    """Return (start, end) indices for this sync's channel slice."""
    state = {"offset": 0, "total": total}
    if os.path.exists(_ROTATION_FILE):
        try:
            with open(_ROTATION_FILE) as f:
                state = json.load(f)
        except (json.JSONDecodeError, OSError):
            pass
    # If total changed (subscribed/unsubscribed), reset
    if state.get("total") != total:
        state = {"offset": 0, "total": total}
    start = state["offset"]
    end = min(start + CHANNELS_PER_SYNC, total)
    return start, end


def _save_rotation_slice(total: int, start: int, end: int):
    """Persist next sync offset."""
    next_offset = end  # Continue from where we left off
    if next_offset >= total:
        next_offset = 0  # Wrap around — full cycle complete
    os.makedirs(os.path.dirname(_ROTATION_FILE), exist_ok=True)
    with open(_ROTATION_FILE, "w") as f:
        json.dump({"offset": next_offset, "total": total}, f)

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: Rotate — only process CHANNELS_PER_SYNC channels per sync
    start, end = _get_rotation_slice(len(channel_ids))
    slice_ids = channel_ids[start:end]
    log.info(f"Processing channels {start+1}–{end} of {len(channel_ids)} ({len(slice_ids)} this sync)")

    # Step 3: Fetch RSS from slice channels (free, parallel)
    rss_videos = {}
    async with httpx.AsyncClient(timeout=20) as client:
        tasks = [_fetch_rss(client, cid) for cid in slice_ids]
        results = await asyncio.gather(*tasks)
        for cid, entries in zip(slice_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:
        _save_rotation_slice(len(channel_ids), start, end)
        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:
        _save_rotation_slice(len(channel_ids), start, end)
        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.")

    # Save rotation state for next sync
    _save_rotation_slice(len(channel_ids), start, end)

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