diff --git a/README.md b/README.md index 85e288e..7a7ef25 100644 --- a/README.md +++ b/README.md @@ -9,8 +9,12 @@ Each user signs in with their own Google account (invite-only) and sees only the subscriptions. All the expensive data (channels, videos, metadata) is fetched from YouTube once and stored locally, so filtering/searching/sorting are instant and don't burn API quota. -> Status: early development. Milestone **M1** (foundation) is in place: docker-compose stack, -> FastAPI backend, Google OAuth login with an email invite-list, and encrypted token storage. +> Status: early development. +> - **M1** (foundation): docker-compose stack, FastAPI backend, Google OAuth login with an +> email invite-list, encrypted token storage. +> - **M2** (ingest core): subscription import, free RSS detection, recent-first + deep backfill +> from the uploads playlist, enrichment (duration/stats/category, Shorts & livestream +> classification), a shared daily quota guard, and a background scheduler. ## Requirements diff --git a/backend/app/config.py b/backend/app/config.py index 43ac6d3..5142571 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -31,9 +31,23 @@ class Settings(BaseSettings): # Daily quota budget (units). Default leaves headroom under the 10,000/day free limit. quota_daily_budget: int = 9000 # Recent-first backfill: how far back to fetch on the first pass per channel. - backfill_recent_max_videos: int = 200 + backfill_recent_max_videos: int = 100 backfill_recent_max_days: int = 365 + # Videos at or below this duration (seconds) are treated as Shorts. + shorts_max_seconds: int = 60 + # videos.list accepts up to 50 ids per call. + enrich_batch_size: int = 50 + + # --- Background scheduler --- + scheduler_enabled: bool = True + rss_poll_minutes: int = 20 + enrich_interval_minutes: int = 3 + backfill_interval_minutes: int = 10 + # Keep this many quota units in reserve so scheduled backfill never starves + # interactive syncs / enrichment of fresh videos. + backfill_quota_reserve: int = 2000 + @property def allowed_email_set(self) -> set[str]: return {e.strip().lower() for e in self.allowed_emails.split(",") if e.strip()} diff --git a/backend/app/main.py b/backend/app/main.py index f884cfb..0f40203 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -1,3 +1,4 @@ +from contextlib import asynccontextmanager from pathlib import Path from fastapi import FastAPI @@ -9,8 +10,19 @@ from starlette.middleware.sessions import SessionMiddleware from app import auth from app.config import settings from app.routes import health, sync +from app.scheduler import shutdown_scheduler, start_scheduler -app = FastAPI(title=settings.app_name) + +@asynccontextmanager +async def lifespan(app: FastAPI): + start_scheduler() + try: + yield + finally: + shutdown_scheduler() + + +app = FastAPI(title=settings.app_name, lifespan=lifespan) app.add_middleware( SessionMiddleware, diff --git a/backend/app/routes/sync.py b/backend/app/routes/sync.py index 9f4fb8a..e76493a 100644 --- a/backend/app/routes/sync.py +++ b/backend/app/routes/sync.py @@ -6,11 +6,24 @@ from app import quota from app.auth import current_user from app.db import get_db from app.models import Channel, Subscription, User, Video +from app.sync.runner import run_enrich, run_recent_backfill, run_rss_poll from app.sync.subscriptions import import_subscriptions router = APIRouter(prefix="/api/sync", tags=["sync"]) +def _user_channels(db: Session, user: User) -> list[Channel]: + return ( + db.execute( + select(Channel) + .join(Subscription, Subscription.channel_id == Channel.id) + .where(Subscription.user_id == user.id) + ) + .scalars() + .all() + ) + + @router.post("/subscriptions") def sync_subscriptions( user: User = Depends(current_user), db: Session = Depends(get_db) @@ -22,6 +35,38 @@ def sync_subscriptions( return result +@router.post("/rss") +def sync_rss( + user: User = Depends(current_user), db: Session = Depends(get_db) +) -> dict: + new = run_rss_poll(db, _user_channels(db, user)) + return {"new_videos": new} + + +@router.post("/backfill") +def sync_backfill( + user: User = Depends(current_user), + db: Session = Depends(get_db), + max_channels: int = 25, +) -> dict: + before = quota.units_used_today(db) + result = run_recent_backfill(db, _user_channels(db, user), max_channels=max_channels) + result["quota_used_estimate"] = quota.units_used_today(db) - before + return result + + +@router.post("/enrich") +def sync_enrich( + user: User = Depends(current_user), db: Session = Depends(get_db) +) -> dict: + before = quota.units_used_today(db) + enriched = run_enrich(db) + return { + "enriched": enriched, + "quota_used_estimate": quota.units_used_today(db) - before, + } + + @router.get("/status") def sync_status( user: User = Depends(current_user), db: Session = Depends(get_db) diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py new file mode 100644 index 0000000..3b3b3ad --- /dev/null +++ b/backend/app/scheduler.py @@ -0,0 +1,77 @@ +"""Background scheduler: free RSS detection, enrichment of new videos, and quota-aware +recent-first / deep backfill. Runs inside the API process (single worker).""" +import logging + +from apscheduler.schedulers.background import BackgroundScheduler + +from app.config import settings +from app.db import SessionLocal +from app.sync.runner import ( + run_deep_backfill, + run_enrich, + run_recent_backfill, + run_rss_poll, +) + +logger = logging.getLogger("subfeed.scheduler") + +_scheduler: BackgroundScheduler | None = None + + +def _job(name: str, fn) -> None: + db = SessionLocal() + try: + result = fn(db) + logger.info("job %s -> %s", name, result) + except Exception: + db.rollback() + logger.exception("job %s failed", name) + finally: + db.close() + + +def _rss_job() -> None: + _job("rss_poll", lambda db: run_rss_poll(db)) + + +def _enrich_job() -> None: + _job("enrich", lambda db: run_enrich(db)) + + +def _backfill_job() -> None: + # Recent-first for not-yet-synced channels, then deep backfill for the rest. + def work(db): + recent = run_recent_backfill(db, max_channels=25) + deep = run_deep_backfill(db, max_channels=10) + return {"recent": recent, "deep": deep} + + _job("backfill", work) + + +def start_scheduler() -> None: + global _scheduler + if not settings.scheduler_enabled or _scheduler is not None: + return + scheduler = BackgroundScheduler(timezone="UTC") + scheduler.add_job( + _rss_job, "interval", minutes=settings.rss_poll_minutes, id="rss_poll" + ) + scheduler.add_job( + _enrich_job, "interval", minutes=settings.enrich_interval_minutes, id="enrich" + ) + scheduler.add_job( + _backfill_job, + "interval", + minutes=settings.backfill_interval_minutes, + id="backfill", + ) + scheduler.start() + _scheduler = scheduler + logger.info("scheduler started") + + +def shutdown_scheduler() -> None: + global _scheduler + if _scheduler is not None: + _scheduler.shutdown(wait=False) + _scheduler = None diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py new file mode 100644 index 0000000..9f14932 --- /dev/null +++ b/backend/app/sync/runner.py @@ -0,0 +1,116 @@ +"""Reusable sync routines shared by the scheduler and the manual /api/sync endpoints. + +Channel/video data is shared across users, so background work acts through a "service +user" (any user with a stored refresh token) unless a YOUTUBE_API_KEY is configured for +public reads. +""" +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app import quota +from app.config import settings +from app.models import Channel, OAuthToken, User +from app.sync.videos import ( + backfill_channel_deep, + backfill_channel_recent, + enrich_pending, + poll_rss_channel, +) +from app.youtube.client import YouTubeClient + + +def get_service_user(db: Session) -> User | None: + return ( + db.execute( + select(User) + .join(OAuthToken) + .where(OAuthToken.refresh_token_enc.is_not(None)) + .order_by(User.id) + ) + .scalars() + .first() + ) + + +def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> int: + if channels is None: + channels = db.execute(select(Channel)).scalars().all() + new = 0 + for channel in channels: + try: + new += poll_rss_channel(db, channel) + except Exception: + db.rollback() + return new + + +def run_enrich(db: Session, max_batches: int = 200, floor: int = 200) -> int: + user = get_service_user(db) + if user is None: + return 0 + total = 0 + with YouTubeClient(db, user) as yt: + for _ in range(max_batches): + if quota.remaining_today(db) <= floor: + break + n = enrich_pending(db, yt) + total += n + if n == 0: + break + return total + + +def run_recent_backfill( + db: Session, channels: list[Channel] | None = None, max_channels: int | None = None +) -> dict: + user = get_service_user(db) + if user is None: + return {"channels_processed": 0, "videos_added": 0, "reason": "no service user"} + if channels is None: + channels = ( + db.execute(select(Channel).where(Channel.recent_synced_at.is_(None))) + .scalars() + .all() + ) + if max_channels: + channels = channels[:max_channels] + processed = 0 + videos_added = 0 + with YouTubeClient(db, user) as yt: + for channel in channels: + if quota.remaining_today(db) <= settings.backfill_quota_reserve: + break + try: + videos_added += backfill_channel_recent(db, yt, channel) + processed += 1 + except Exception: + db.rollback() + return {"channels_processed": processed, "videos_added": videos_added} + + +def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) -> dict: + user = get_service_user(db) + if user is None: + return {"channels_processed": 0, "videos_added": 0, "reason": "no service user"} + channels = ( + db.execute( + select(Channel).where( + Channel.backfill_done.is_(False), + Channel.recent_synced_at.is_not(None), + ) + ) + .scalars() + .all() + )[:max_channels] + processed = 0 + videos_added = 0 + with YouTubeClient(db, user) as yt: + for channel in channels: + if quota.remaining_today(db) <= settings.backfill_quota_reserve: + break + try: + videos_added += backfill_channel_deep(db, yt, channel, max_pages) + processed += 1 + except Exception: + db.rollback() + return {"channels_processed": processed, "videos_added": videos_added} diff --git a/backend/app/sync/videos.py b/backend/app/sync/videos.py new file mode 100644 index 0000000..5818e5c --- /dev/null +++ b/backend/app/sync/videos.py @@ -0,0 +1,227 @@ +"""Video ingestion: free RSS detection, recent-first (and deep) backfill from the +uploads playlist, and enrichment via videos.list (duration, stats, category, Shorts +and livestream classification).""" +import re +from datetime import datetime, timedelta, timezone + +from sqlalchemy import select +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.orm import Session + +from app.config import settings +from app.models import Channel, Video +from app.youtube.client import YouTubeClient, best_thumbnail +from app.youtube.rss import fetch_channel_feed + +_DURATION_RE = re.compile( + r"P(?:(\d+)D)?T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?", re.IGNORECASE +) + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def parse_iso8601_duration(value: str | None) -> int | None: + if not value: + return None + match = _DURATION_RE.fullmatch(value) + if not match: + return None + days, hours, minutes, seconds = (int(g) if g else 0 for g in match.groups()) + return days * 86400 + hours * 3600 + minutes * 60 + seconds + + +def parse_dt(value: str | None) -> datetime | None: + if not value: + return None + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + + +def _insert_stubs(db: Session, rows: list[dict]) -> int: + """Idempotently insert video stubs; returns the number of newly inserted rows. + + Uses RETURNING so the count is exact (ON CONFLICT DO NOTHING leaves rowcount + unreliable across drivers). Duplicates within the batch are dropped first so a + single INSERT can't raise CardinalityViolation on the conflict target.""" + if not rows: + return 0 + seen: dict[str, dict] = {} + for row in rows: + seen[row["id"]] = row + stmt = ( + pg_insert(Video) + .values(list(seen.values())) + .on_conflict_do_nothing(index_elements=[Video.id]) + .returning(Video.id) + ) + inserted = len(db.execute(stmt).fetchall()) + db.commit() + return inserted + + +# --- RSS (free) --- +def poll_rss_channel(db: Session, channel: Channel) -> int: + entries = fetch_channel_feed(channel.id) + inserted = _insert_stubs(db, entries) + channel.last_rss_at = _now() + db.add(channel) + db.commit() + return inserted + + +# --- Backfill (uploads playlist; costs quota) --- +def _stub_from_playlist_item(item: dict, channel_id: str) -> dict | None: + content = item.get("contentDetails", {}) + snippet = item.get("snippet", {}) + video_id = content.get("videoId") + if not video_id: + return None + published = parse_dt(content.get("videoPublishedAt") or snippet.get("publishedAt")) + return { + "id": video_id, + "channel_id": channel_id, + "title": snippet.get("title"), + "published_at": published, + "thumbnail_url": best_thumbnail(snippet.get("thumbnails")), + } + + +def backfill_channel_recent(db: Session, yt: YouTubeClient, channel: Channel) -> int: + """First pass: newest videos up to the configured count or age cutoff. Records a + resume cursor for the later deep backfill.""" + if not channel.uploads_playlist_id: + channel.recent_synced_at = _now() + db.add(channel) + db.commit() + return 0 + + cutoff = _now() - timedelta(days=settings.backfill_recent_max_days) + collected: list[dict] = [] + page_token: str | None = None + next_cursor: str | None = None + done = False + + while len(collected) < settings.backfill_recent_max_videos: + data = yt.get_playlist_items(channel.uploads_playlist_id, page_token) + reached_cutoff = False + for item in data.get("items", []): + stub = _stub_from_playlist_item(item, channel.id) + if stub is None: + continue + if stub["published_at"] and stub["published_at"] < cutoff: + reached_cutoff = True + break + collected.append(stub) + if len(collected) >= settings.backfill_recent_max_videos: + break + page_token = data.get("nextPageToken") + if reached_cutoff or not page_token: + next_cursor = page_token + done = not page_token + break + next_cursor = page_token + + channel.recent_synced_at = _now() + channel.backfill_cursor = next_cursor + channel.backfill_done = done + inserted = _insert_stubs(db, collected) + db.add(channel) + db.commit() + return inserted + + +def backfill_channel_deep( + db: Session, yt: YouTubeClient, channel: Channel, max_pages: int = 5 +) -> int: + """Continue paging older uploads from the stored cursor (bounded per run).""" + if channel.backfill_done or not channel.uploads_playlist_id: + return 0 + page_token = channel.backfill_cursor + total = 0 + for _ in range(max_pages): + data = yt.get_playlist_items(channel.uploads_playlist_id, page_token) + rows = [ + s + for s in ( + _stub_from_playlist_item(it, channel.id) for it in data.get("items", []) + ) + if s is not None + ] + total += _insert_stubs(db, rows) + page_token = data.get("nextPageToken") + channel.backfill_cursor = page_token + if not page_token: + channel.backfill_done = True + break + db.add(channel) + db.commit() + return total + + +# --- Enrichment (videos.list) --- +def _live_status(snippet: dict, live_details: dict | None) -> str: + broadcast = snippet.get("liveBroadcastContent") + if broadcast == "live": + return "live" + if broadcast == "upcoming": + return "upcoming" + if live_details and live_details.get("actualEndTime"): + return "was_live" + return "none" + + +def apply_video_details(video: Video, item: dict) -> None: + snippet = item.get("snippet", {}) + content = item.get("contentDetails", {}) + stats = item.get("statistics", {}) + topics = item.get("topicDetails", {}) + live = item.get("liveStreamingDetails") + + video.title = snippet.get("title") or video.title + video.description = snippet.get("description") + if not video.published_at: + video.published_at = parse_dt(snippet.get("publishedAt")) + video.thumbnail_url = best_thumbnail(snippet.get("thumbnails")) or video.thumbnail_url + video.duration_seconds = parse_iso8601_duration(content.get("duration")) + video.view_count = int(stats["viewCount"]) if stats.get("viewCount") else None + video.like_count = int(stats["likeCount"]) if stats.get("likeCount") else None + video.category_id = int(snippet["categoryId"]) if snippet.get("categoryId") else None + video.topic_categories = topics.get("topicCategories") + video.default_language = snippet.get("defaultLanguage") or snippet.get( + "defaultAudioLanguage" + ) + video.live_status = _live_status(snippet, live) + video.is_short = bool( + video.live_status == "none" + and video.duration_seconds + and 0 < video.duration_seconds <= settings.shorts_max_seconds + ) + + +def enrich_pending(db: Session, yt: YouTubeClient, limit: int | None = None) -> int: + limit = limit or settings.enrich_batch_size + videos = ( + db.execute(select(Video).where(Video.enriched_at.is_(None)).limit(limit)) + .scalars() + .all() + ) + if not videos: + return 0 + items = {it["id"]: it for it in yt.get_videos([v.id for v in videos])} + now = _now() + enriched = 0 + for video in videos: + item = items.get(video.id) + if item is None: + # Unavailable (deleted/private): stamp so we don't keep retrying. + video.enriched_at = now + continue + apply_video_details(video, item) + video.enriched_at = now + enriched += 1 + db.commit() + return enriched diff --git a/backend/app/youtube/client.py b/backend/app/youtube/client.py index b1902a1..87305cd 100644 --- a/backend/app/youtube/client.py +++ b/backend/app/youtube/client.py @@ -133,3 +133,29 @@ class YouTubeClient: ) items.extend(data.get("items", [])) return items + + def get_playlist_items(self, playlist_id: str, page_token: str | None = None) -> dict: + """One page (up to 50) of a playlist's items, most-recent-first for uploads + playlists. Returns the raw response (items + nextPageToken).""" + params = { + "part": "contentDetails,snippet", + "playlistId": playlist_id, + "maxResults": 50, + } + if page_token: + params["pageToken"] = page_token + return self._get("playlistItems", params) + + def get_videos(self, video_ids: list[str]) -> list[dict]: + items: list[dict] = [] + for batch in _chunks(video_ids, 50): + data = self._get( + "videos", + { + "part": "snippet,contentDetails,statistics,topicDetails,liveStreamingDetails", + "id": ",".join(batch), + "maxResults": 50, + }, + ) + items.extend(data.get("items", [])) + return items diff --git a/backend/app/youtube/rss.py b/backend/app/youtube/rss.py new file mode 100644 index 0000000..4653d32 --- /dev/null +++ b/backend/app/youtube/rss.py @@ -0,0 +1,42 @@ +"""Free per-channel RSS feed reader (no API quota). Returns the channel's most recent +~15 uploads as lightweight stubs for fast fresh-video detection.""" +from datetime import datetime, timezone + +import feedparser +import httpx + +RSS_URL = "https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}" + + +def fetch_channel_feed(channel_id: str) -> list[dict]: + url = RSS_URL.format(channel_id=channel_id) + try: + resp = httpx.get(url, timeout=20.0, headers={"User-Agent": "Subfeed/1.0"}) + except httpx.HTTPError: + return [] + if resp.status_code != 200: + return [] + + parsed = feedparser.parse(resp.content) + out: list[dict] = [] + for entry in parsed.entries: + video_id = entry.get("yt_videoid") + if not video_id: + continue + published = None + if entry.get("published_parsed"): + published = datetime(*entry.published_parsed[:6], tzinfo=timezone.utc) + thumbnail = None + media = entry.get("media_thumbnail") or [] + if media: + thumbnail = media[0].get("url") + out.append( + { + "id": video_id, + "channel_id": channel_id, + "title": entry.get("title"), + "published_at": published, + "thumbnail_url": thumbnail, + } + ) + return out diff --git a/backend/requirements.txt b/backend/requirements.txt index 3d74bde..44e98f0 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -7,5 +7,7 @@ pydantic>=2.7,<3.0 pydantic-settings>=2.3,<3.0 authlib>=1.3,<2.0 httpx>=0.27,<1.0 +feedparser>=6.0,<7.0 +apscheduler>=3.10,<4.0 itsdangerous>=2.1,<3.0 cryptography>=42,<46