""" YouTube Data API v3 adapter - collect video mentions from whitelisted channels. Gracefully skips if YOUTUBE_API_KEY is not configured. """ import logging from datetime import datetime, timedelta, timezone from typing import Dict, List, Optional import aiohttp from sqlalchemy.ext.asyncio import AsyncSession from app.core.http_client import get_http_session from sqlalchemy import select from app.core.config import settings from app.models.overlay_raw_event import OverlayVideoEvent from app.models.overlay_registry import YouTubeChannelRegistry from app.services.overlay.entity_resolver import EntityResolver logger = logging.getLogger(__name__) YOUTUBE_SEARCH_URL = "https://www.googleapis.com/youtube/v3/search" YOUTUBE_VIDEOS_URL = "https://www.googleapis.com/youtube/v3/videos" class YouTubeAdapter: """Collect YouTube video data from whitelisted channels.""" def __init__(self): self.api_key: str = getattr(settings, "YOUTUBE_API_KEY", "") self.resolver = EntityResolver() @property def enabled(self) -> bool: return bool(self.api_key) async def _get_channels(self, db: AsyncSession) -> List[YouTubeChannelRegistry]: result = await db.execute( select(YouTubeChannelRegistry).where(YouTubeChannelRegistry.active == True) ) return result.scalars().all() async def _search_channel_videos( self, channel_id: str, published_after: datetime ) -> List[dict]: """Search for recent videos from a channel via YouTube Data API.""" params = { "part": "snippet", "channelId": channel_id, "type": "video", "order": "date", "publishedAfter": published_after.strftime("%Y-%m-%dT%H:%M:%SZ"), "maxResults": 20, "key": self.api_key, } try: session = await get_http_session() async with session.get( YOUTUBE_SEARCH_URL, params=params, timeout=aiohttp.ClientTimeout(total=30) ) as resp: if resp.status == 403: logger.warning("YouTube API: quota exceeded or invalid key") return [] resp.raise_for_status() data = await resp.json() return data.get("items", []) except Exception as e: logger.error(f"YouTube search error for channel {channel_id}: {e}") return [] async def _get_video_stats(self, video_ids: List[str]) -> Dict[str, Dict]: """Fetch view/comment counts for a list of video IDs.""" if not video_ids: return {} params = { "part": "statistics", "id": ",".join(video_ids), "key": self.api_key, } try: session = await get_http_session() async with session.get( YOUTUBE_VIDEOS_URL, params=params, timeout=aiohttp.ClientTimeout(total=30) ) as resp: resp.raise_for_status() data = await resp.json() stats = {} for item in data.get("items", []): vid_id = item["id"] s = item.get("statistics", {}) stats[vid_id] = { "view_count": int(s.get("viewCount", 0)), "comment_count": int(s.get("commentCount", 0)), } return stats except Exception as e: logger.error(f"YouTube stats fetch error: {e}") return {} async def collect(self, db: AsyncSession, days_back: int = 2) -> int: """ Collect YouTube videos from whitelisted channels. Gracefully returns 0 if API key is not configured. Returns number of new records inserted. """ if not self.enabled: logger.info("YouTube adapter: API key not configured, skipping") return 0 await self.resolver.load_aliases(db) channels = await self._get_channels(db) if not channels: logger.info("YouTube: no active channels in registry") return 0 published_after = datetime.now(timezone.utc) - timedelta(days=days_back) inserted = 0 for channel in channels: items = await self._search_channel_videos(channel.channel_id, published_after) video_ids = [ item["id"]["videoId"] for item in items if isinstance(item.get("id"), dict) and "videoId" in item["id"] ] stats = await self._get_video_stats(video_ids) for item in items: vid_id = item.get("id", {}).get("videoId") if isinstance(item.get("id"), dict) else None if not vid_id: continue # Check duplicate existing = await db.execute( select(OverlayVideoEvent.id).where(OverlayVideoEvent.video_id == vid_id) ) if existing.first(): continue snippet = item.get("snippet", {}) title = snippet.get("title", "") pub_at_str = snippet.get("publishedAt", "") try: pub_at = datetime.fromisoformat(pub_at_str.replace("Z", "+00:00")) except Exception: pub_at = datetime.now(timezone.utc) stat = stats.get(vid_id, {}) matched = self.resolver.resolve_from_title(title) event = OverlayVideoEvent( video_id=vid_id, channel_id=channel.channel_id, title=title, view_count=stat.get("view_count", 0), comment_count=stat.get("comment_count", 0), published_at=pub_at, matched_symbols=matched, channel_weight=channel.channel_weight, ) db.add(event) inserted += 1 if inserted: await db.commit() logger.info(f"YouTube: inserted {inserted} video events") return inserted