From 24b6e0026d97fc75109f59d96d867e36337cfa27 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Thu, 11 Jun 2026 01:22:07 +0200 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20M2=20(part=201)=20=E2=80=94=20subsc?= =?UTF-8?q?ription=20import,=20YouTube=20client,=20quota=20guard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Channel/Subscription/Video/ApiQuotaUsage models + migration 0002 - Synchronous YouTube Data API client with OAuth token refresh and per-call quota accounting (API key preferred for public reads when configured) - Subscription import: upsert channels + subscription links, prune unsubscribed, fetch channel details (uploads playlist, topics, stats, handle, country) - Central quota guard tracking units per US-Pacific day against a daily budget - POST /api/sync/subscriptions and GET /api/sync/status endpoints --- backend/alembic/versions/0002_ingest_core.py | 127 +++++++++++++++++ backend/app/config.py | 10 ++ backend/app/main.py | 3 +- backend/app/models.py | 126 ++++++++++++++++- backend/app/quota.py | 50 +++++++ backend/app/routes/sync.py | 40 ++++++ backend/app/sync/__init__.py | 0 backend/app/sync/subscriptions.py | 115 ++++++++++++++++ backend/app/youtube/__init__.py | 0 backend/app/youtube/client.py | 135 +++++++++++++++++++ 10 files changed, 603 insertions(+), 3 deletions(-) create mode 100644 backend/alembic/versions/0002_ingest_core.py create mode 100644 backend/app/quota.py create mode 100644 backend/app/routes/sync.py create mode 100644 backend/app/sync/__init__.py create mode 100644 backend/app/sync/subscriptions.py create mode 100644 backend/app/youtube/__init__.py create mode 100644 backend/app/youtube/client.py diff --git a/backend/alembic/versions/0002_ingest_core.py b/backend/alembic/versions/0002_ingest_core.py new file mode 100644 index 0000000..5c05bad --- /dev/null +++ b/backend/alembic/versions/0002_ingest_core.py @@ -0,0 +1,127 @@ +"""ingest core: channels, subscriptions, videos, api_quota_usage + +Revision ID: 0002_ingest_core +Revises: 0001_initial +Create Date: 2026-06-11 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + +revision: str = "0002_ingest_core" +down_revision: Union[str, None] = "0001_initial" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "channels", + sa.Column("id", sa.String(length=32), primary_key=True), + sa.Column("title", sa.String(length=255), nullable=True), + sa.Column("handle", sa.String(length=255), nullable=True), + sa.Column("description", sa.Text(), nullable=True), + sa.Column("thumbnail_url", sa.String(length=1024), nullable=True), + sa.Column("uploads_playlist_id", sa.String(length=64), nullable=True), + sa.Column("subscriber_count", sa.BigInteger(), nullable=True), + sa.Column("video_count", sa.BigInteger(), nullable=True), + sa.Column("topic_categories", sa.JSON(), nullable=True), + sa.Column("default_language", sa.String(length=16), nullable=True), + sa.Column("country", sa.String(length=8), nullable=True), + sa.Column("details_synced_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("recent_synced_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_rss_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("backfill_cursor", sa.String(length=128), nullable=True), + sa.Column("backfill_done", sa.Boolean(), nullable=False, server_default="false"), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.func.now(), + nullable=False, + ), + ) + + op.create_table( + "subscriptions", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column( + "user_id", + sa.Integer(), + sa.ForeignKey("users.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column( + "channel_id", + sa.String(length=32), + sa.ForeignKey("channels.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("yt_subscription_id", sa.String(length=128), nullable=True), + sa.Column("priority", sa.Integer(), nullable=False, server_default="0"), + sa.Column("hidden", sa.Boolean(), nullable=False, server_default="false"), + sa.Column("subscribed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.func.now(), + nullable=False, + ), + sa.UniqueConstraint("user_id", "channel_id", name="uq_user_channel"), + ) + op.create_index("ix_subscriptions_user_id", "subscriptions", ["user_id"]) + op.create_index("ix_subscriptions_channel_id", "subscriptions", ["channel_id"]) + + op.create_table( + "videos", + sa.Column("id", sa.String(length=16), primary_key=True), + sa.Column( + "channel_id", + sa.String(length=32), + sa.ForeignKey("channels.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("title", sa.Text(), nullable=True), + sa.Column("description", sa.Text(), nullable=True), + sa.Column("published_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("thumbnail_url", sa.String(length=1024), nullable=True), + sa.Column("duration_seconds", sa.Integer(), nullable=True), + sa.Column("view_count", sa.BigInteger(), nullable=True), + sa.Column("like_count", sa.BigInteger(), nullable=True), + sa.Column("category_id", sa.Integer(), nullable=True), + sa.Column("topic_categories", sa.JSON(), nullable=True), + sa.Column("default_language", sa.String(length=16), nullable=True), + sa.Column("detected_language", sa.String(length=16), nullable=True), + sa.Column("is_short", sa.Boolean(), nullable=False, server_default="false"), + sa.Column("live_status", sa.String(length=16), nullable=False, server_default="none"), + sa.Column("enriched_at", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.func.now(), + nullable=False, + ), + ) + op.create_index("ix_videos_channel_id", "videos", ["channel_id"]) + op.create_index("ix_videos_published_at", "videos", ["published_at"]) + op.create_index("ix_videos_is_short", "videos", ["is_short"]) + op.create_index("ix_videos_live_status", "videos", ["live_status"]) + + op.create_table( + "api_quota_usage", + sa.Column("day", sa.Date(), primary_key=True), + sa.Column("units_used", sa.Integer(), nullable=False, server_default="0"), + ) + + +def downgrade() -> None: + op.drop_table("api_quota_usage") + op.drop_index("ix_videos_live_status", table_name="videos") + op.drop_index("ix_videos_is_short", table_name="videos") + op.drop_index("ix_videos_published_at", table_name="videos") + op.drop_index("ix_videos_channel_id", table_name="videos") + op.drop_table("videos") + op.drop_index("ix_subscriptions_channel_id", table_name="subscriptions") + op.drop_index("ix_subscriptions_user_id", table_name="subscriptions") + op.drop_table("subscriptions") + op.drop_table("channels") diff --git a/backend/app/config.py b/backend/app/config.py index 25284b5..43ac6d3 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -24,6 +24,16 @@ class Settings(BaseSettings): # Origin of a separately served frontend dev server (enables CORS). Empty in production. frontend_origin: str = "" + # --- Sync / YouTube Data API --- + # Optional API key for public reads (channels/videos/playlistItems). When set it is + # preferred for shared backfill/enrichment so it doesn't depend on a specific user. + youtube_api_key: str = "" + # 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_days: int = 365 + @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 17122a2..f884cfb 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -8,7 +8,7 @@ from starlette.middleware.sessions import SessionMiddleware from app import auth from app.config import settings -from app.routes import health +from app.routes import health, sync app = FastAPI(title=settings.app_name) @@ -30,6 +30,7 @@ if settings.frontend_origin: app.include_router(health.router) app.include_router(auth.router) +app.include_router(sync.router) STATIC_DIR = Path(__file__).parent / "static" app.mount("/assets", StaticFiles(directory=STATIC_DIR / "assets"), name="assets") diff --git a/backend/app/models.py b/backend/app/models.py index 92e8b91..3087abd 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -1,6 +1,18 @@ -from datetime import datetime +from datetime import date, datetime -from sqlalchemy import DateTime, ForeignKey, String, func +from sqlalchemy import ( + BigInteger, + Boolean, + Date, + DateTime, + ForeignKey, + Integer, + JSON, + String, + Text, + UniqueConstraint, + func, +) from sqlalchemy.orm import Mapped, mapped_column, relationship from app.db import Base @@ -22,6 +34,9 @@ class User(Base): token: Mapped["OAuthToken | None"] = relationship( back_populates="user", uselist=False, cascade="all, delete-orphan" ) + subscriptions: Mapped[list["Subscription"]] = relationship( + back_populates="user", cascade="all, delete-orphan" + ) class OAuthToken(Base): @@ -41,3 +56,110 @@ class OAuthToken(Base): ) user: Mapped["User"] = relationship(back_populates="token") + + +class Channel(Base): + """A YouTube channel. Shared across all users (one channel's videos are the same + for everyone), so its expensive metadata is fetched and stored only once.""" + + __tablename__ = "channels" + + id: Mapped[str] = mapped_column(String(32), primary_key=True) # "UC..." channel id + title: Mapped[str | None] = mapped_column(String(255)) + handle: Mapped[str | None] = mapped_column(String(255)) + description: Mapped[str | None] = mapped_column(Text) + thumbnail_url: Mapped[str | None] = mapped_column(String(1024)) + uploads_playlist_id: Mapped[str | None] = mapped_column(String(64)) + subscriber_count: Mapped[int | None] = mapped_column(BigInteger) + video_count: Mapped[int | None] = mapped_column(BigInteger) + topic_categories: Mapped[list | None] = mapped_column(JSON) + default_language: Mapped[str | None] = mapped_column(String(16)) + country: Mapped[str | None] = mapped_column(String(8)) + + # Sync bookkeeping. + details_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + recent_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + last_rss_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + backfill_cursor: Mapped[str | None] = mapped_column(String(128)) + backfill_done: Mapped[bool] = mapped_column( + Boolean, default=False, server_default="false" + ) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + + videos: Mapped[list["Video"]] = relationship(back_populates="channel") + subscriptions: Mapped[list["Subscription"]] = relationship(back_populates="channel") + + +class Subscription(Base): + """Per-user link to a channel, with the user's own priority/visibility overrides.""" + + __tablename__ = "subscriptions" + __table_args__ = (UniqueConstraint("user_id", "channel_id", name="uq_user_channel"),) + + id: Mapped[int] = mapped_column(primary_key=True) + user_id: Mapped[int] = mapped_column( + ForeignKey("users.id", ondelete="CASCADE"), index=True + ) + channel_id: Mapped[str] = mapped_column( + ForeignKey("channels.id", ondelete="CASCADE"), index=True + ) + # YouTube's subscription resource id, needed to unsubscribe via the API. + yt_subscription_id: Mapped[str | None] = mapped_column(String(128)) + priority: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + hidden: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") + subscribed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + + user: Mapped["User"] = relationship(back_populates="subscriptions") + channel: Mapped["Channel"] = relationship(back_populates="subscriptions") + + +class Video(Base): + """A YouTube video. Shared across users; per-user state lives elsewhere.""" + + __tablename__ = "videos" + + id: Mapped[str] = mapped_column(String(16), primary_key=True) # video id + channel_id: Mapped[str] = mapped_column( + ForeignKey("channels.id", ondelete="CASCADE"), index=True + ) + title: Mapped[str | None] = mapped_column(Text) + description: Mapped[str | None] = mapped_column(Text) + published_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), index=True + ) + thumbnail_url: Mapped[str | None] = mapped_column(String(1024)) + duration_seconds: Mapped[int | None] = mapped_column(Integer) + view_count: Mapped[int | None] = mapped_column(BigInteger) + like_count: Mapped[int | None] = mapped_column(BigInteger) + category_id: Mapped[int | None] = mapped_column(Integer) + topic_categories: Mapped[list | None] = mapped_column(JSON) + default_language: Mapped[str | None] = mapped_column(String(16)) + detected_language: Mapped[str | None] = mapped_column(String(16)) + is_short: Mapped[bool] = mapped_column( + Boolean, default=False, server_default="false", index=True + ) + # none | live | upcoming | premiere | was_live + live_status: Mapped[str] = mapped_column( + String(16), default="none", server_default="none", index=True + ) + enriched_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + + channel: Mapped["Channel"] = relationship(back_populates="videos") + + +class ApiQuotaUsage(Base): + """Tracks YouTube Data API units spent per Pacific-time day (the quota resets at + midnight Pacific). The whole app shares one daily budget.""" + + __tablename__ = "api_quota_usage" + + day: Mapped[date] = mapped_column(Date, primary_key=True) + units_used: Mapped[int] = mapped_column(Integer, default=0, server_default="0") diff --git a/backend/app/quota.py b/backend/app/quota.py new file mode 100644 index 0000000..b2acbd8 --- /dev/null +++ b/backend/app/quota.py @@ -0,0 +1,50 @@ +"""Central YouTube Data API quota guard. + +The whole app shares one daily budget (the API resets at midnight US Pacific time). +Steady-state work (RSS detection is free; enrichment of new videos is cheap) should +always be allowed; expensive backfill must yield to the remaining budget. +""" +from datetime import datetime +from zoneinfo import ZoneInfo + +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.orm import Session + +from app.config import settings +from app.models import ApiQuotaUsage + +_PACIFIC = ZoneInfo("America/Los_Angeles") + + +def pacific_today(): + return datetime.now(_PACIFIC).date() + + +def units_used_today(db: Session) -> int: + row = db.get(ApiQuotaUsage, pacific_today()) + return row.units_used if row else 0 + + +def remaining_today(db: Session) -> int: + return max(0, settings.quota_daily_budget - units_used_today(db)) + + +def can_spend(db: Session, units: int) -> bool: + return remaining_today(db) >= units + + +def record_usage(db: Session, units: int) -> None: + """Atomically add `units` to today's counter (upsert).""" + if units <= 0: + return + day = pacific_today() + stmt = ( + pg_insert(ApiQuotaUsage) + .values(day=day, units_used=units) + .on_conflict_do_update( + index_elements=[ApiQuotaUsage.day], + set_={"units_used": ApiQuotaUsage.units_used + units}, + ) + ) + db.execute(stmt) + db.commit() diff --git a/backend/app/routes/sync.py b/backend/app/routes/sync.py new file mode 100644 index 0000000..9f4fb8a --- /dev/null +++ b/backend/app/routes/sync.py @@ -0,0 +1,40 @@ +from fastapi import APIRouter, Depends +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +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.subscriptions import import_subscriptions + +router = APIRouter(prefix="/api/sync", tags=["sync"]) + + +@router.post("/subscriptions") +def sync_subscriptions( + user: User = Depends(current_user), db: Session = Depends(get_db) +) -> dict: + before = quota.units_used_today(db) + result = import_subscriptions(db, user) + result["quota_used_estimate"] = quota.units_used_today(db) - before + result["quota_remaining_today"] = quota.remaining_today(db) + return result + + +@router.get("/status") +def sync_status( + user: User = Depends(current_user), db: Session = Depends(get_db) +) -> dict: + sub_count = db.scalar( + select(func.count()) + .select_from(Subscription) + .where(Subscription.user_id == user.id) + ) + return { + "subscriptions": sub_count, + "channels_total": db.scalar(select(func.count()).select_from(Channel)), + "videos_total": db.scalar(select(func.count()).select_from(Video)), + "quota_used_today": quota.units_used_today(db), + "quota_remaining_today": quota.remaining_today(db), + } diff --git a/backend/app/sync/__init__.py b/backend/app/sync/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/sync/subscriptions.py b/backend/app/sync/subscriptions.py new file mode 100644 index 0000000..653eba1 --- /dev/null +++ b/backend/app/sync/subscriptions.py @@ -0,0 +1,115 @@ +"""Import the authenticated user's YouTube subscriptions and channel metadata.""" +from datetime import datetime, timezone + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.models import Channel, Subscription, User +from app.youtube.client import YouTubeClient, best_thumbnail + + +def _to_int(value) -> int | None: + try: + return int(value) if value is not None else None + except (TypeError, ValueError): + return None + + +def apply_channel_details(db: Session, items: list[dict]) -> None: + now = datetime.now(timezone.utc) + for item in items: + channel = db.get(Channel, item["id"]) + if channel is None: + continue + snippet = item.get("snippet", {}) + content = item.get("contentDetails", {}) + stats = item.get("statistics", {}) + topics = item.get("topicDetails", {}) + + channel.title = snippet.get("title") or channel.title + channel.handle = snippet.get("customUrl") + channel.description = snippet.get("description") + channel.thumbnail_url = ( + best_thumbnail(snippet.get("thumbnails")) or channel.thumbnail_url + ) + channel.default_language = snippet.get("defaultLanguage") + channel.country = snippet.get("country") + channel.uploads_playlist_id = content.get("relatedPlaylists", {}).get("uploads") + channel.subscriber_count = _to_int(stats.get("subscriberCount")) + channel.video_count = _to_int(stats.get("videoCount")) + channel.topic_categories = topics.get("topicCategories") + channel.details_synced_at = now + + +def import_subscriptions(db: Session, user: User) -> dict: + """Pull all subscriptions for `user`, upsert channels + subscription links, prune + channels the user has unsubscribed from on YouTube, and fetch channel details for + any channel we don't have details for yet.""" + fetched: dict[str, dict] = {} + new_channels = 0 + + with YouTubeClient(db, user) as yt: + for sub in yt.iter_subscriptions(): + cid = sub.get("channel_id") + if cid: + fetched[cid] = sub + + for cid, sub in fetched.items(): + channel = db.get(Channel, cid) + if channel is None: + channel = Channel( + id=cid, + title=sub.get("title"), + description=sub.get("description"), + thumbnail_url=sub.get("thumbnail_url"), + ) + db.add(channel) + new_channels += 1 + sub_row = db.execute( + select(Subscription).where( + Subscription.user_id == user.id, Subscription.channel_id == cid + ) + ).scalar_one_or_none() + if sub_row is None: + sub_row = Subscription(user_id=user.id, channel_id=cid) + db.add(sub_row) + sub_row.yt_subscription_id = sub.get("yt_subscription_id") + db.flush() + + # Prune subscriptions removed on YouTube. + removed = 0 + existing = ( + db.execute(select(Subscription).where(Subscription.user_id == user.id)) + .scalars() + .all() + ) + for sub_row in existing: + if sub_row.channel_id not in fetched: + db.delete(sub_row) + removed += 1 + db.commit() + + # Fetch details for channels that don't have them yet. + need_details = ( + db.execute( + select(Channel).where( + Channel.id.in_(fetched.keys()), + Channel.details_synced_at.is_(None), + ) + ) + .scalars() + .all() + ) + detailed = 0 + if need_details: + items = yt.get_channels([c.id for c in need_details]) + apply_channel_details(db, items) + detailed = len(items) + db.commit() + + return { + "subscriptions": len(fetched), + "channels_new": new_channels, + "channels_detailed": detailed, + "removed_stale": removed, + } diff --git a/backend/app/youtube/__init__.py b/backend/app/youtube/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/youtube/client.py b/backend/app/youtube/client.py new file mode 100644 index 0000000..b1902a1 --- /dev/null +++ b/backend/app/youtube/client.py @@ -0,0 +1,135 @@ +"""Thin synchronous YouTube Data API v3 client bound to a user's OAuth token. + +Each list call costs 1 quota unit and is recorded via the central quota guard. +Public reads (channels/videos/playlistItems) use the configured API key when available +so they don't depend on a specific user's token; subscriptions.list?mine=true always +uses OAuth. +""" +from collections.abc import Iterator +from datetime import datetime, timedelta, timezone + +import httpx + +from app import quota +from app.config import settings +from app.models import User +from app.security import decrypt + +GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token" +API_BASE = "https://www.googleapis.com/youtube/v3" + + +class YouTubeError(Exception): + pass + + +def _chunks(items: list, size: int) -> Iterator[list]: + for i in range(0, len(items), size): + yield items[i : i + size] + + +def best_thumbnail(thumbnails: dict | None) -> str | None: + if not thumbnails: + return None + for key in ("maxres", "standard", "high", "medium", "default"): + if key in thumbnails and thumbnails[key].get("url"): + return thumbnails[key]["url"] + return None + + +class YouTubeClient: + def __init__(self, db, user: User): + self.db = db + self.user = user + self._http = httpx.Client(timeout=30.0) + + def __enter__(self) -> "YouTubeClient": + return self + + def __exit__(self, *exc) -> None: + self._http.close() + + # --- auth --- + def _access_token(self) -> str: + tok = self.user.token + if tok is None: + raise YouTubeError("User has no stored OAuth token") + now = datetime.now(timezone.utc) + if tok.access_token and tok.expiry and tok.expiry > now + timedelta(seconds=60): + return tok.access_token + refresh = decrypt(tok.refresh_token_enc) + if not refresh: + raise YouTubeError("No refresh token; user must re-authenticate") + resp = self._http.post( + GOOGLE_TOKEN_URL, + data={ + "client_id": settings.google_client_id, + "client_secret": settings.google_client_secret, + "refresh_token": refresh, + "grant_type": "refresh_token", + }, + ) + if resp.status_code != 200: + raise YouTubeError(f"Token refresh failed: {resp.status_code} {resp.text[:200]}") + data = resp.json() + tok.access_token = data["access_token"] + tok.expiry = now + timedelta(seconds=int(data.get("expires_in", 3600))) + self.db.add(tok) + self.db.commit() + return tok.access_token + + # --- core request --- + def _get(self, path: str, params: dict, cost: int = 1, allow_key: bool = True) -> dict: + p = dict(params) + headers = {} + if allow_key and settings.youtube_api_key: + p["key"] = settings.youtube_api_key + else: + headers["Authorization"] = f"Bearer {self._access_token()}" + resp = self._http.get(f"{API_BASE}/{path}", params=p, headers=headers) + quota.record_usage(self.db, cost) + if resp.status_code != 200: + raise YouTubeError(f"GET {path} -> {resp.status_code}: {resp.text[:300]}") + return resp.json() + + # --- endpoints --- + def iter_subscriptions(self) -> Iterator[dict]: + """Yield the authenticated user's subscriptions (requires OAuth).""" + page_token = None + while True: + params = { + "part": "snippet", + "mine": "true", + "maxResults": 50, + "order": "alphabetical", + } + if page_token: + params["pageToken"] = page_token + data = self._get("subscriptions", params, cost=1, allow_key=False) + for item in data.get("items", []): + snippet = item.get("snippet", {}) + resource = snippet.get("resourceId", {}) + yield { + "channel_id": resource.get("channelId"), + "title": snippet.get("title"), + "description": snippet.get("description"), + "thumbnail_url": best_thumbnail(snippet.get("thumbnails")), + "yt_subscription_id": item.get("id"), + } + page_token = data.get("nextPageToken") + if not page_token: + break + + def get_channels(self, channel_ids: list[str]) -> list[dict]: + items: list[dict] = [] + for batch in _chunks(channel_ids, 50): + data = self._get( + "channels", + { + "part": "snippet,contentDetails,statistics,topicDetails", + "id": ",".join(batch), + "maxResults": 50, + }, + ) + items.extend(data.get("items", [])) + return items From cff1d2307166f40a69f4731ecd5d9a50819efb62 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Thu, 11 Jun 2026 01:36:41 +0200 Subject: [PATCH 2/2] =?UTF-8?q?feat:=20M2=20(part=202)=20=E2=80=94=20RSS?= =?UTF-8?q?=20poller,=20backfill,=20enrichment,=20scheduler?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Free per-channel RSS reader for quota-less fresh-video detection - Recent-first backfill (configurable: 100 videos / 1 year) plus resumable deep backfill from the uploads playlist - Enrichment via videos.list: duration, view/like counts, category, topics, language, Shorts heuristic and livestream/premiere classification - Reusable sync runners + APScheduler jobs (rss / enrich / backfill), all quota-aware with a reserve so backfill never starves fresh enrichment - Manual triggers: POST /api/sync/{rss,backfill,enrich} - Exact insert counting via RETURNING with in-batch de-duplication --- README.md | 8 +- backend/app/config.py | 16 ++- backend/app/main.py | 14 ++- backend/app/routes/sync.py | 45 +++++++ backend/app/scheduler.py | 77 ++++++++++++ backend/app/sync/runner.py | 116 +++++++++++++++++ backend/app/sync/videos.py | 227 ++++++++++++++++++++++++++++++++++ backend/app/youtube/client.py | 26 ++++ backend/app/youtube/rss.py | 42 +++++++ backend/requirements.txt | 2 + 10 files changed, 569 insertions(+), 4 deletions(-) create mode 100644 backend/app/scheduler.py create mode 100644 backend/app/sync/runner.py create mode 100644 backend/app/sync/videos.py create mode 100644 backend/app/youtube/rss.py 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