Upload 3 files

Dockerfile

@@ -1,11 +1,6 @@
 # Use an official Python runtime as a parent image
 FROM python:3.10-slim
 
-# Install system dependencies including ffmpeg
-RUN apt-get update && apt-get install -y \
-    ffmpeg \
-    && rm -rf /var/lib/apt/lists/*
-
 # Set up a new user 'user' with UID 1000
 RUN useradd -m -u 1000 user
 USER user
@@ -21,11 +16,8 @@ COPY --chown=user . .
 # Install dependencies
 RUN pip install --no-cache-dir --upgrade -r requirements.txt
 
-# Create a directory for yt-dlp cache
-RUN mkdir -p /tmp/.yt-dlp-cache && chmod 777 /tmp/.yt-dlp-cache
-
 # Expose port 7860
 EXPOSE 7860
 
-# Start the application using gunicorn with detailed logging
-CMD ["gunicorn", "--bind", "0.0.0.0:7860", "--workers", "2", "--timeout", "120", "--access-logfile", "-", "--error-logfile", "-", "app:app"]
+# Start the application using uvicorn
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

main.py

@@ -0,0 +1,223 @@
+import socket
+import sys
+import os
+import logging
+import requests
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
+from ytmusicapi import YTMusic
+from urllib.parse import urlparse, parse_qs
+from typing import Optional, Dict, Any
+
+# --- DNS BYPASS PATCH START ---
+BYPASS_DOMAINS = [
+    "youtube.com", "music.youtube.com", "googlevideo.com", "youtu.be"
+]
+
+try:
+    import dns.resolver
+    _original_getaddrinfo = socket.getaddrinfo
+
+    def patched_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
+        if host and any(domain in host for domain in BYPASS_DOMAINS):
+            try:
+                res = dns.resolver.Resolver()
+                res.nameservers = ['8.8.8.8', '1.1.1.1']
+                answers = res.resolve(host, 'A')
+                ip_address = answers[0].to_text()
+                return [(socket.AF_INET, type, proto, '', (ip_address, port))]
+            except Exception:
+                return _original_getaddrinfo(host, port, family, type, proto, flags)
+        return _original_getaddrinfo(host, port, family, type, proto, flags)
+
+    socket.getaddrinfo = patched_getaddrinfo
+    print(f"[INIT] DNS Bypass installed for {len(BYPASS_DOMAINS)} domains.", file=sys.stderr)
+except ImportError:
+    print("❌ CRITICAL: dnspython not installed. DNS Bypass failed.", file=sys.stderr)
+# --- DNS BYPASS PATCH END ---
+
+# 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.")

requirements.txt

@@ -1,5 +1,6 @@
-flask
-yt-dlp
+fastapi
+uvicorn
+pydantic
+ytmusicapi
+requests
 dnspython
-gunicorn
-cachetools