import base64 import logging from datetime import datetime, timedelta, timezone from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel from sqlalchemy import func, select from sqlalchemy.orm import Session from app.config import settings from app.core.auth_dependency import require_session from app.db import get_db from app.models.category import Category from app.models.channel import Channel from app.models.channel_category import channel_categories from app.models.video import Video from app.services import sync from app.services.download_jobs import latest_jobs_map from app.services.google_oauth import OAuthNotConnected from app.services.video_presentation import channel_categories_map, serialize_video from app.services.youtube_client import YouTubeAPIError, YouTubeInsufficientScope, YouTubeQuotaExceeded logger = logging.getLogger(__name__) router = APIRouter(dependencies=[Depends(require_session)]) ACTIVITY_DEFAULT_LIMIT = 20 ACTIVITY_MAX_LIMIT = 100 ACTIVITY_VIDEOS_PER_CHANNEL = 3 class ChannelCategoriesUpdate(BaseModel): category_ids: list[int] def _category_ids_by_channel(db: Session, channel_ids: list[int]) -> dict[int, list[int]]: if not channel_ids: return {} rows = db.execute( select(channel_categories.c.channel_id, channel_categories.c.category_id).where( channel_categories.c.channel_id.in_(channel_ids) ) ).all() result: dict[int, list[int]] = {} for channel_id, category_id in rows: result.setdefault(channel_id, []).append(category_id) return result def _new_videos_counts(db: Session, channel_ids: list[int]) -> dict[int, int]: """Videos published within the new-videos window, per channel, in one aggregate query.""" if not channel_ids: return {} since = datetime.now(timezone.utc) - timedelta(days=settings.new_videos_window_days) rows = ( db.query(Video.channel_id, func.count(Video.id)) .filter(Video.channel_id.in_(channel_ids), Video.published_at >= since) .group_by(Video.channel_id) .all() ) return {channel_id: count for channel_id, count in rows} def _serialize(channel: Channel, category_ids: list[int], new_videos_count: int = 0) -> dict: return { "id": channel.id, "youtube_channel_id": channel.youtube_channel_id, "title": channel.title, "description": channel.description, "thumbnail_url": channel.thumbnail_url, "uploads_playlist_id": channel.uploads_playlist_id, "subscriber_count": channel.subscriber_count, "subscribed": channel.subscribed, "last_synced_at": channel.last_synced_at, "category_ids": category_ids, "new_videos_count": new_videos_count, } @router.get("/channels") def list_channels( subscribed: bool | None = None, search: str | None = None, category_id: int | None = None, uncategorized: bool = False, db: Session = Depends(get_db), ) -> list[dict]: query = db.query(Channel) if subscribed is not None: query = query.filter(Channel.subscribed == subscribed) if search: query = query.filter(Channel.title.ilike(f"%{search}%")) if uncategorized: categorized_ids = select(channel_categories.c.channel_id) query = query.filter(~Channel.id.in_(categorized_ids)) elif category_id is not None: channel_ids_in_category = select(channel_categories.c.channel_id).where( channel_categories.c.category_id == category_id ) query = query.filter(Channel.id.in_(channel_ids_in_category)) channels = query.order_by(Channel.title.asc()).all() channel_ids = [c.id for c in channels] category_map = _category_ids_by_channel(db, channel_ids) new_videos_map = _new_videos_counts(db, channel_ids) return [_serialize(c, category_map.get(c.id, []), new_videos_map.get(c.id, 0)) for c in channels] def _encode_activity_cursor(last_published_at: datetime | None, channel_id: int) -> str: raw = f"{last_published_at.isoformat() if last_published_at else ''}|{channel_id}" return base64.urlsafe_b64encode(raw.encode()).decode() def _decode_activity_cursor(cursor: str) -> tuple[datetime | None, int]: try: raw = base64.urlsafe_b64decode(cursor.encode()).decode() iso_raw, channel_id_raw = raw.rsplit("|", 1) published_at = datetime.fromisoformat(iso_raw) if iso_raw else None if published_at is not None and published_at.tzinfo is None: published_at = published_at.replace(tzinfo=timezone.utc) return published_at, int(channel_id_raw) except Exception: raise HTTPException(status_code=400, detail="Invalid cursor") # NOTE: this route must stay declared BEFORE /channels/{channel_id}, otherwise # "activity" is captured by the path parameter and validation fails with 422. @router.get("/channels/activity") def channel_activity( category_id: int | None = None, uncategorized: bool = False, limit: int = Query(ACTIVITY_DEFAULT_LIMIT, ge=1, le=ACTIVITY_MAX_LIMIT), cursor: str | None = None, db: Session = Depends(get_db), ) -> dict: """Subscribed channels ordered by the freshness of their latest video (channels without videos go last), with up to 3 recent videos each. Cursor is (last_published_at, channel_id); an empty iso marks the no-videos tail so pagination stays lossless across that boundary.""" latest_video = ( select(Video.channel_id, func.max(Video.published_at).label("last_published_at")) .group_by(Video.channel_id) .subquery() ) query = db.query(Channel, latest_video.c.last_published_at).outerjoin( latest_video, latest_video.c.channel_id == Channel.id ) query = query.filter(Channel.subscribed.is_(True)) if uncategorized: categorized_ids = select(channel_categories.c.channel_id) query = query.filter(~Channel.id.in_(categorized_ids)) elif category_id is not None: channel_ids_in_category = select(channel_categories.c.channel_id).where( channel_categories.c.category_id == category_id ) query = query.filter(Channel.id.in_(channel_ids_in_category)) if cursor: cursor_published_at, cursor_channel_id = _decode_activity_cursor(cursor) if cursor_published_at is not None: # Channels without videos sort last (NULLS LAST), so they are # also "after" this cursor and must pass the filter; the ORDER BY # keeps them behind every channel that still has videos. query = query.filter( (latest_video.c.last_published_at < cursor_published_at) | ( (latest_video.c.last_published_at == cursor_published_at) & (Channel.id < cursor_channel_id) ) | latest_video.c.last_published_at.is_(None) ) else: query = query.filter( latest_video.c.last_published_at.is_(None) & (Channel.id < cursor_channel_id) ) rows = ( query.order_by(latest_video.c.last_published_at.desc().nullslast(), Channel.id.desc()) .limit(limit + 1) .all() ) next_cursor = None if len(rows) > limit: last_channel, last_published_at = rows[limit - 1] next_cursor = _encode_activity_cursor(last_published_at, last_channel.id) rows = rows[:limit] channels_by_id = {channel.id: channel for channel, _ in rows} channel_ids = list(channels_by_id) videos_by_channel: dict[int, list[Video]] = {channel_id: [] for channel_id in channel_ids} if channel_ids: # Top-N videos per channel in a single windowed query (no N+1). row_number = func.row_number().over( partition_by=Video.channel_id, order_by=(Video.published_at.desc(), Video.id.desc()), ).label("rn") windowed = ( db.query(Video.id, row_number) .filter(Video.channel_id.in_(channel_ids)) .subquery() ) recent_ids = select(windowed.c.id).where(windowed.c.rn <= ACTIVITY_VIDEOS_PER_CHANNEL) videos = db.query(Video).filter(Video.id.in_(recent_ids)).all() for video in videos: videos_by_channel[video.channel_id].append(video) for channel_videos in videos_by_channel.values(): channel_videos.sort(key=lambda video: (video.published_at, video.id), reverse=True) categories_map = channel_categories_map(db, channel_ids) jobs_map = latest_jobs_map(db, [video.id for videos in videos_by_channel.values() for video in videos]) new_videos_map = _new_videos_counts(db, channel_ids) items = [] for channel, _ in rows: categories = categories_map.get(channel.id, []) items.append( { "channel": _serialize( channel, [category["id"] for category in categories], new_videos_map.get(channel.id, 0) ), "videos": [ serialize_video(video, channel, categories, jobs_map.get(video.id)) for video in videos_by_channel.get(channel.id, []) ], } ) return {"items": items, "next_cursor": next_cursor} @router.get("/channels/{channel_id}") def get_channel(channel_id: int, db: Session = Depends(get_db)) -> dict: channel = db.get(Channel, channel_id) if channel is None: raise HTTPException(status_code=404, detail="Channel not found") category_map = _category_ids_by_channel(db, [channel_id]) new_videos_map = _new_videos_counts(db, [channel_id]) return _serialize(channel, category_map.get(channel_id, []), new_videos_map.get(channel_id, 0)) @router.put("/channels/{channel_id}/categories") def set_channel_categories(channel_id: int, payload: ChannelCategoriesUpdate, db: Session = Depends(get_db)) -> dict: channel = db.get(Channel, channel_id) if channel is None: raise HTTPException(status_code=404, detail="Channel not found") unique_ids = set(payload.category_ids) if unique_ids: found = db.query(Category.id).filter(Category.id.in_(unique_ids)).all() found_ids = {row[0] for row in found} missing = unique_ids - found_ids if missing: raise HTTPException(status_code=400, detail=f"Unknown category ids: {sorted(missing)}") db.execute(channel_categories.delete().where(channel_categories.c.channel_id == channel_id)) if unique_ids: db.execute( channel_categories.insert(), [{"channel_id": channel_id, "category_id": cid} for cid in unique_ids], ) db.commit() new_videos_map = _new_videos_counts(db, [channel_id]) return _serialize(channel, sorted(unique_ids), new_videos_map.get(channel_id, 0)) @router.post("/channels/{channel_id}/unsubscribe") def unsubscribe_channel(channel_id: int, db: Session = Depends(get_db)) -> dict: channel = db.get(Channel, channel_id) if channel is None: raise HTTPException(status_code=404, detail="Channel not found") try: sync.unsubscribe_channel(db, channel) except sync.ChannelHasNoSubscriptionId as exc: raise HTTPException(status_code=400, detail=str(exc)) except OAuthNotConnected: raise HTTPException(status_code=400, detail="Google account is not connected") except YouTubeInsufficientScope: raise HTTPException( status_code=403, detail="Недостаточно прав для отписки — переподключи YouTube, чтобы выдать разрешение на управление подписками", ) except YouTubeQuotaExceeded: raise HTTPException(status_code=503, detail="YouTube API quota exhausted") except YouTubeAPIError as exc: raise HTTPException(status_code=502, detail=f"YouTube API error: {exc}") except Exception: logger.exception("Failed to unsubscribe from channel %s", channel_id) raise HTTPException(status_code=502, detail="YouTube is unavailable") category_map = _category_ids_by_channel(db, [channel_id]) new_videos_map = _new_videos_counts(db, [channel_id]) return _serialize(channel, category_map.get(channel_id, []), new_videos_map.get(channel_id, 0))