from fastapi import FastAPI, HTTPException from pydantic import BaseModel from ytmusicapi import YTMusic import logging import requests from urllib.parse import urlparse, parse_qs from typing import Optional, Dict, Any # Initialize FastAPI and YTMusic app = FastAPI() ytmusic = YTMusic() # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # --- Pydantic Models --- class SearchRequest(BaseModel): query: str class MatchRequest(BaseModel): url: str class MatchResponse(BaseModel): url: str filename: str track_id: str # --- Helper Functions --- def extract_amazon_track_id(url: str) -> Optional[str]: """ Extracts track ID from various Amazon Music URL formats. """ if "music.amazon.com" not in url: return None parsed_url = urlparse(url) query_params = parse_qs(parsed_url.query) if "trackAsin" in query_params: return query_params["trackAsin"][0] path_parts = parsed_url.path.split('/') if "tracks" in path_parts: try: track_id_index = path_parts.index("tracks") + 1 if track_id_index < len(path_parts): return path_parts[track_id_index] except (ValueError, IndexError): pass logger.warning(f"Could not extract Amazon track ID from URL: {url}") return None def get_song_link_info(url: str) -> Optional[Dict[str, Any]]: """ Fetches track information from the Song.link API. """ api_base_url = "https://api.song.link/v1-alpha.1/links" params = {"userCountry": "US"} if "music.amazon.com" in url: track_id = extract_amazon_track_id(url) if track_id: params["platform"] = "amazonMusic" params["id"] = track_id params["type"] = "song" else: params["url"] = url else: params["url"] = url try: logger.info(f"Querying Song.link API with params: {params}") response = requests.get(api_base_url, params=params, timeout=10) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: logger.error(f"Error fetching from Song.link API: {e}") return None def extract_url(links_by_platform: dict, platform: str) -> Optional[str]: """ Extracts a specific platform URL from Song.link API response. """ if platform in links_by_platform and links_by_platform[platform].get("url"): return links_by_platform[platform]["url"] logger.warning(f"No URL found for platform '{platform}' in links: {links_by_platform.keys()}") return None # --- Endpoints --- @app.get("/") async def root(): return {"message": "Music Match API is running. Use /match or /searcht endpoints."} @app.post("/searcht") async def searcht(request: SearchRequest): """ Searches YouTube Music for a song and returns the first result with a videoId. """ logger.info(f"search query: {request.query}") search_results = ytmusic.search(request.query, filter="songs") # Return the first song that has a valid videoId, or an empty dict first_song = next((song for song in search_results if 'videoId' in song and song['videoId']), {}) if search_results else {} return first_song @app.post("/match", response_model=MatchResponse) async def match(request: MatchRequest): """ Matches a given music track URL (Spotify, Apple Music, Amazon, etc.) to a YouTube Music URL. """ track_url = request.url logger.info(f"Match endpoint: Processing URL: {track_url}") track_info = get_song_link_info(track_url) if not track_info: logger.error(f"Match endpoint: Could not fetch track info for URL: {track_url}") raise HTTPException(status_code=404, detail="Could not fetch track info from Song.link API.") entity_unique_id = track_info.get("entityUniqueId") title = None artist = None # Attempt to find main entity details if entity_unique_id and entity_unique_id in track_info.get("entitiesByUniqueId", {}): main_entity = track_info["entitiesByUniqueId"][entity_unique_id] title = main_entity.get("title") artist = main_entity.get("artistName") logger.info(f"Match endpoint: Found main entity - Title: '{title}', Artist: '{artist}'") else: logger.warning(f"Match endpoint: Could not find main entity details for {track_url} using entityUniqueId: {entity_unique_id}") # Fallback logic to find title/artist from other entities for entity_id, entity_data in track_info.get("entitiesByUniqueId", {}).items(): if entity_data.get("title") and entity_data.get("artistName"): title = entity_data.get("title") artist = entity_data.get("artistName") logger.info(f"Match endpoint: Using fallback entity - Title: '{title}', Artist: '{artist}' from entity ID {entity_id}") break if not title or not artist: logger.error(f"Match endpoint: Could not determine title and artist for URL: {track_url}") raise HTTPException(status_code=404, detail="Could not determine title and artist from Song.link info.") # Try to find a direct YouTube link from the API response youtube_url = extract_url(track_info.get("linksByPlatform", {}), "youtube") if youtube_url: video_id = None # Extract Video ID from URL if "v=" in youtube_url: video_id = youtube_url.split("v=")[1].split("&")[0] elif "youtu.be/" in youtube_url: video_id = youtube_url.split("youtu.be/")[1].split("?")[0] filename = f"{title} - {artist}" if title and artist else "Unknown Track - Unknown Artist" logger.info(f"Match endpoint: Found direct YouTube URL: {youtube_url}, Video ID: {video_id}") return MatchResponse(url=youtube_url, filename=filename, track_id=video_id) else: # Fallback: Search YouTube Music logger.info(f"Match endpoint: No direct YouTube URL. Searching YTMusic with: '{title} - {artist}'") search_query = f'{title} {artist}' search_results = ytmusic.search(search_query, filter="songs") if search_results: first_song = next((song for song in search_results if song.get('videoId')), None) if first_song and first_song.get('videoId'): video_id = first_song["videoId"] ym_url = f'https://music.youtube.com/watch?v={video_id}' # Get artist name safely artist_name = artist if first_song.get('artists') and len(first_song['artists']) > 0: artist_name = first_song['artists'][0]['name'] filename = f"{first_song.get('title', title)} - {artist_name}" logger.info(f"Match endpoint: Found YTMusic search result - URL: {ym_url}, Video ID: {video_id}") return MatchResponse(filename=filename, url=ym_url, track_id=video_id) else: logger.error(f"Match endpoint: YTMusic search for '{search_query}' yielded no results with a videoId.") raise HTTPException(status_code=404, detail="No matching video ID found on YouTube Music after search.") else: logger.error(f"Match endpoint: YTMusic search for '{search_query}' yielded no results.") raise HTTPException(status_code=404, detail="No results found on YouTube Music for the track.")