from sqlalchemy import select from sqlalchemy.orm import Session from app.models.category import Category from app.models.channel import Channel from app.models.channel_category import channel_categories from app.models.download_job import DownloadJob from app.models.video import Video def channel_categories_map(db: Session, channel_ids: list[int]) -> dict[int, list[dict]]: if not channel_ids: return {} rows = db.execute( select(channel_categories.c.channel_id, Category.id, Category.name) .join(Category, Category.id == channel_categories.c.category_id) .where(channel_categories.c.channel_id.in_(channel_ids)) ).all() result: dict[int, list[dict]] = {} for channel_id, category_id, category_name in rows: result.setdefault(channel_id, []).append({"id": category_id, "name": category_name}) return result def _serialize_local(job: DownloadJob | None) -> dict: if job is None: return {"available": False, "status": "not_downloaded", "progress_percent": None, "media_url": None} if job.status == "completed": return {"available": True, "status": "completed", "progress_percent": 100, "media_url": job.media_url} return { "available": False, "status": job.status, "progress_percent": job.progress_percent, "media_url": None, } def serialize_video( video: Video, channel: Channel, categories: list[dict], download_job: DownloadJob | None = None ) -> dict: return { "youtube_video_id": video.youtube_video_id, "title": video.title, "description": video.description, "channel": { "id": channel.id, "youtube_channel_id": channel.youtube_channel_id, "title": channel.title, "thumbnail_url": channel.thumbnail_url, }, "thumbnail_url": video.thumbnail_url, "published_at": video.published_at, "duration_seconds": video.duration_seconds, "youtube_url": video.youtube_url, "categories": categories, "local": _serialize_local(download_job), }