"""YouTube Data API v3 + ytmusicapi client for fetching artist songs and view counts.""" import os import json import logging import datetime import time from traceback import format_exc from typing import Optional, List, Dict, Any, Union, cast from dotenv import load_dotenv import googleapiclient.discovery # type: ignore load_dotenv() # Try YouTube Data API v3 first try: from googleapiclient.discovery import build as build_yt_service HAS_YT_DATA_API = True except ImportError: HAS_YT_DATA_API = False # Try ytmusicapi try: from ytmusicapi import YTMusic # type: ignore HAS_YTMUSIC = True except ImportError: HAS_YTMUSIC = False class YouTubeClient: """Fetches song data from YouTube using Data API v3 and/or ytmusicapi.""" def __init__(self): self.api_key = os.getenv("YOUTUBE_API_KEY") self.youtube: Optional[googleapiclient.discovery.Resource] = None self.ytmusic: Optional[YTMusic] = None if HAS_YT_DATA_API and self.api_key and self.api_key != "your_youtube_api_key_here": self.youtube = build_yt_service("youtube", "v3", developerKey=self.api_key) print("[YouTube] Data API v3 initialized") else: print("[YouTube] Data API v3 not available (no API key or missing library)") if HAS_YTMUSIC: self.ytmusic = YTMusic() print("[YouTube] ytmusicapi initialized (unauthenticated)") else: print("[YouTube] ytmusicapi not installed") def resolve_channel_by_handle(self, handle: str) -> Optional[Dict[str, Any]]: """Resolve a YouTube @handle to channel info using Data API v3. Args: handle: YouTube handle without '@' prefix (e.g. 'agsyworld') Returns: dict with channel_id, name, thumbnail, subscriber_count, description or None if not found / API unavailable. """ if not self.youtube: print("[YouTube] Data API v3 not available for handle resolution") return None handle = handle.lstrip("@") try: yt_api = cast(Any, self.youtube) response = yt_api.channels().list( forHandle=handle, part="snippet,statistics", maxResults=1, ).execute() items = response.get("items", []) if not items: return None ch = items[0] snippet = ch.get("snippet", {}) stats = ch.get("statistics", {}) thumbnails = snippet.get("thumbnails", {}) thumb = ( thumbnails.get("high", {}).get("url") or thumbnails.get("medium", {}).get("url") or thumbnails.get("default", {}).get("url") ) return { "channel_id": ch.get("id"), "name": snippet.get("title", ""), "thumbnail": thumb, "subscriber_count": int(stats.get("subscriberCount", 0)), "description": snippet.get("description", ""), } except Exception as e: print(f"[YouTube] Error resolving handle @{handle}: {e}") return None def get_artist_songs_ytmusic(self, channel_id: str, artist_name: str, fast_mode: bool = False) -> List[Dict[str, Any]]: """Get artist's songs via ytmusicapi. Returns list of song dicts. If fast_mode=True, skips recursive album fetching for speed.""" if not self.ytmusic: return [] try: # Search for the artist on YT Music ytm = cast(Any, self.ytmusic) search_results = ytm.search(artist_name, filter="artists", limit=5) if not search_results: print(f" [ytmusic] No results for '{artist_name}'") return [] # Find the best match artist_browse_id = None for result in search_results: if result.get("browseId"): artist_browse_id = result["browseId"] break if not artist_browse_id: print(f" [ytmusic] No browseId for '{artist_name}'") return [] # Get artist details artist_data = ytm.get_artist(artist_browse_id) songs = [] # Get songs from the artist page if "songs" in artist_data and "results" in artist_data["songs"]: for track in artist_data["songs"]["results"]: song = { "title": track.get("title", "Unknown"), "video_id": track.get("videoId"), "album": track.get("album", {}).get("name") if track.get("album") else None, "views": self._parse_view_count(track.get("views", "")), "thumbnail": self._get_best_thumbnail(track.get("thumbnails", [])), } if song["video_id"]: songs.append(song) # Also try to get albums/singles for release dates (skip in fast_mode) if not fast_mode and "albums" in dict(artist_data) and "results" in dict(artist_data).get("albums", {}): for album in artist_data["albums"]["results"][:10]: album_id = str(album.get("browseId", "")) if album_id and self.ytmusic is not None: try: ytm: Any = self.ytmusic album_data = ytm.get_album(album_id) release_date = album_data.get("year") for track in album_data.get("tracks", []): # Check if we already have this song vid = track.get("videoId") if vid and not any(s["video_id"] == vid for s in songs): songs.append({ "title": track.get("title", "Unknown"), "video_id": vid, "album": album_data.get("title"), "views": self._parse_view_count(track.get("views", "")), "thumbnail": self._get_best_thumbnail(track.get("thumbnails", [])), "release_year": release_date, }) time.sleep(0.3) # Rate limit except Exception as e: print(f" [ytmusic] Error fetching album {album_id}: {e}") # Try singles too if "singles" in dict(artist_data) and "results" in dict(artist_data).get("singles", {}): for single in artist_data["singles"]["results"][:10]: single_id = str(single.get("browseId", "")) if single_id and self.ytmusic is not None: try: ytm: Any = self.ytmusic single_data = ytm.get_album(single_id) release_date = single_data.get("year") for track in single_data.get("tracks", []): vid = track.get("videoId") if vid and not any(s["video_id"] == vid for s in songs): songs.append({ "title": track.get("title", "Unknown"), "video_id": vid, "album": single_data.get("title"), "views": self._parse_view_count(track.get("views", "")), "thumbnail": self._get_best_thumbnail(track.get("thumbnails", [])), "release_year": release_date, }) time.sleep(0.3) except Exception as e: print(f" [ytmusic] Error fetching single {single_id}: {e}") print(f" [ytmusic] Found {len(songs)} songs for '{artist_name}'") return songs except Exception as e: print(f" [ytmusic] Error for '{artist_name}': {e}") return [] def get_video_stats(self, video_ids: List[str]) -> Dict[str, Dict[str, Union[int, str]]]: """Get view/like/comment counts for a list of video IDs using YouTube Data API v3. Returns dict of {video_id: {views, likes, comments, published_at}}""" if not self.youtube: return {} stats: Dict[str, Dict[str, Union[int, str]]] = {} # Process in batches of 50 (API limit) for i in range(0, len(video_ids), 50): # Pyre gets confused with slicing lists sometimes, so cast explicitly chunk_end = min(i + 50, len(video_ids)) batch: List[str] = [video_ids[j] for j in range(i, chunk_end)] try: yt_api = cast(Any, self.youtube) response = yt_api.videos().list( part="statistics,snippet", id=",".join(batch) ).execute() for item in response.get("items", []): vid = item["id"] statistics = item.get("statistics", {}) snippet = item.get("snippet", {}) stats[vid] = { "views": int(statistics.get("viewCount", 0)), "likes": int(statistics.get("likeCount", 0)), "comments": int(statistics.get("commentCount", 0)), "published_at": snippet.get("publishedAt", ""), "title": snippet.get("title", ""), "channel_title": snippet.get("channelTitle", ""), "thumbnail": snippet.get("thumbnails", {}).get("high", {}).get("url", ""), } time.sleep(0.1) # Small delay between batches except Exception as e: print(f" [yt-data-api] Error fetching stats for batch: {e}") return stats def search_videos_by_query(self, query: str, max_results: int = 5) -> List[Dict[str, Any]]: """Search YouTube for a specific string (e.g., 'Artist Title')""" if self.youtube is None: return [] try: yt_api = cast(Any, self.youtube) response = yt_api.search().list( q=query, part="id,snippet", type="video", maxResults=max_results, ).execute() videos = [] for item in response.get("items", []): videos.append({ "video_id": item["id"]["videoId"], "title": item["snippet"]["title"], "published_at": item["snippet"]["publishedAt"], "channel_title": item["snippet"]["channelTitle"], "thumbnail": item["snippet"]["thumbnails"].get("high", {}).get("url", ""), }) return videos except Exception as e: print(f" [yt-data-api] Error searching videos for '{query}': {e}") return [] def get_channel_videos(self, channel_id: str, max_results: int = 50) -> list: """Get recent videos from a YouTube channel using Data API v3.""" if not self.youtube: return [] try: # Search for videos by the channel yt_api = cast(Any, self.youtube) response = yt_api.search().list( part="id,snippet", channelId=channel_id, type="video", order="date", maxResults=min(max_results, 50), videoCategoryId="10" # Music category ).execute() videos = [] for item in response.get("items", []): videos.append({ "video_id": item["id"]["videoId"], "title": item["snippet"]["title"], "published_at": item["snippet"]["publishedAt"], "thumbnail": item["snippet"]["thumbnails"].get("high", {}).get("url", ""), }) return videos except Exception as e: print(f" [yt-data-api] Error fetching channel videos: {e}") return [] def _parse_view_count(self, view_str: str) -> Optional[int]: """Parse view count strings like '1.5M views' or '500K views'.""" if not view_str: return None view_str = view_str.lower().replace(",", "").replace(" views", "").replace(" plays", "").strip() try: if "b" in view_str: return int(float(view_str.replace("b", "")) * 1_000_000_000) elif "m" in view_str: return int(float(view_str.replace("m", "")) * 1_000_000) elif "k" in view_str: return int(float(view_str.replace("k", "")) * 1_000) else: return int(float(view_str)) except (ValueError, TypeError): return None def _get_best_thumbnail(self, thumbnails: list) -> str: """Get the highest resolution thumbnail URL.""" if not thumbnails: return "" return thumbnails[-1].get("url", "")