Update app.py

app.py

@@ -1,4 +1,8 @@
-from flask import Flask, render_template, jsonify, request
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.responses import HTMLResponse
+from fastapi.staticfiles import StaticFiles
+from fastapi.templating import Jinja2Templates
+from pydantic import BaseModel
 from ytmusicapi import YTMusic
 import os
 import logging
@@ -7,43 +11,64 @@ from datetime import datetime, timedelta
 import time
 import asyncio
 import cloudscraper
-from pydantic import BaseModel
 from urllib.parse import urlparse, parse_qs
 from collections import defaultdict
 import threading
+from typing import Optional, Dict, Any
 
-app = Flask(__name__)
-ytmusic = YTMusic()
+app = FastAPI()
+
+# Mount static files and templates
+app.mount("/static", StaticFiles(directory="static"), name="static")
+templates = Jinja2Templates(directory="templates")
 
+ytmusic = YTMusic()
 
-# Configure logging 
+# Configure logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
-@app.route('/')
-def index():
-    return render_template('index.html')
-
-@app.route('/search', methods=['POST'])
-def search():
-    query = request.json.get('query', '')
-    search_results = ytmusic.search(query, filter="songs")
-    return jsonify(search_results)
-
-@app.route('/searcht', methods=['POST'])
-def searcht(): 
-    query = request.json.get('query', '')
-    logger.info(f"serch query: {query}")
-    search_results = ytmusic.search(query, filter="songs")
-    first_song = next((song for song in search_results if 'videoId' in song and song['videoId']), {}) if search_results else {}
-    return jsonify(first_song)
+# Pydantic models for request/response validation
+class SearchRequest(BaseModel):
+    query: str
+
+class MatchRequest(BaseModel):
+    url: str
+
+class TrackDownloadRequest(BaseModel):
+    track_id: str
+    quality: str = "128"
 
+class MatchResponse(BaseModel):
+    url: str
+    filename: str
+    track_id: str
 
-def extract_amazon_track_id(url: str):
+class ErrorResponse(BaseModel):
+    detail: str
+    premium: Optional[str] = None
+
+@app.get("/", response_class=HTMLResponse)
+async def index(request: Request):
+    return templates.TemplateResponse("index.html", {"request": request})
+
+@app.post("/search")
+async def search(request: SearchRequest):
+    search_results = ytmusic.search(request.query, filter="songs")
+    return search_results
+
+@app.post("/searcht")
+async def searcht(request: SearchRequest):
+    logger.info(f"search query: {request.query}")
+    search_results = ytmusic.search(request.query, filter="songs")
+    first_song = next((song for song in search_results if 'videoId' in song and song['videoId']), {}) if search_results else {}
+    return first_song
+
+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: # MODIFIED: Slight logic inversion for early exit compared to original, same effective outcome.
+    if "music.amazon.com" not in url:
         return None
 
     parsed_url = urlparse(url)
@@ -52,153 +77,135 @@ def extract_amazon_track_id(url: str):
     if "trackAsin" in query_params:
         return query_params["trackAsin"][0]
 
-    path_parts = parsed_url.path.split('/') # MODIFIED: Changed from simple `url.split` to more robust path parsing for Case 2.
+    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] # MODIFIED: Accessing specific part after "tracks".
+                return path_parts[track_id_index]
         except (ValueError, IndexError):
             pass
 
-    logger.warning(f"Could not extract Amazon track ID from URL: {url}") # ADDED: Logging for when no ID is found.
+    logger.warning(f"Could not extract Amazon track ID from URL: {url}")
     return None
 
-
-def get_song_link_info(url: str):
+def get_song_link_info(url: str) -> Optional[Dict[str, Any]]:
     """
     Fetches track information from the Song.link API.
     Uses requests.get() which is a blocking call.
     """
-    api_base_url = "https://api.song.link/v1-alpha.1/links" # ADDED: Defined base URL for clarity.
-    params = {"userCountry": "US"} # MODIFIED: Using a params dictionary for requests.get().
+    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" # MODIFIED: Populating params dict.
+            params["platform"] = "amazonMusic"
             params["id"] = track_id
             params["type"] = "song"
         else:
-            params["url"] = url # MODIFIED: Populating params dict.
+            params["url"] = url
     else:
-        params["url"] = url # MODIFIED: Populating params dict.
+        params["url"] = url
 
-    try: # ADDED: try-except block for robust error handling during API call.
-        logger.info(f"Querying Song.link API with params: {params}") # ADDED: Logging the API query.
-        response = requests.get(api_base_url, params=params, timeout=10) # MODIFIED: Call uses base_url and params. ADDED: timeout.
-        response.raise_for_status() # ADDED: Checks for HTTP errors (4xx or 5xx responses).
+    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: # ADDED: Catching network/request related exceptions.
-        logger.error(f"Error fetching from Song.link API: {e}") # ADDED: Logging the specific error.
+    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):
+def extract_url(links_by_platform: dict, platform: str) -> Optional[str]:
     """
     Extracts a specific platform URL from Song.link API response.
     """
-    # MODIFIED: Added .get("url") for safer access to prevent KeyError if "url" key is missing.
     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()}") # ADDED: Logging if platform URL not found.
+    
+    logger.warning(f"No URL found for platform '{platform}' in links: {links_by_platform.keys()}")
     return None
 
-@app.route('/match', methods=['POST'])
-def match(): # MODIFIED: Changed from `async def` to `def` for synchronous Flask.
+@app.post("/match", response_model=MatchResponse)
+async def match(request: MatchRequest):
     """
     Matches a given music track URL to a YouTube Music URL.
-    Expects a JSON body with "url".
     """
-    data = request.get_json()
-    if not data: # ADDED: Check for empty JSON payload.
-        logger.error("Match endpoint: No JSON payload received.") # ADDED: Logging.
-        return jsonify({"detail": "No JSON payload received."}), 400 # MODIFIED: Flask-style JSON error response.
-
-    track_url = data.get('url')
-    # MODIFIED: Added more specific validation for track_url presence and type.
-    if not track_url or not isinstance(track_url, str):
-        logger.error(f"Match endpoint: Invalid or missing URL: {track_url}") # ADDED: Logging.
-        return jsonify({"detail": "Valid 'url' string is required in request body."}), 400 # MODIFIED: Flask-style JSON error response.
-
-    logger.info(f"Match endpoint: Processing URL: {track_url}") # ADDED: Logging.
+    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}") # ADDED: Logging.
-        # MODIFIED: Flask-style JSON error response instead of HTTPException.
-        return jsonify({"detail": "Could not fetch track info from Song.link API."}), 404
+        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") # MODIFIED: Used .get() for safer access.
+    entity_unique_id = track_info.get("entityUniqueId")
     title = None
     artist = None
 
-    # MODIFIED: More robust extraction of title and artist with checks and logging.
     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}'") # ADDED: Logging.
+        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}") # ADDED: Logging.
-        # ADDED: Fallback logic to find title/artist from other entities if main one fails.
+        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}") # ADDED: Logging.
+                logger.info(f"Match endpoint: Using fallback entity - Title: '{title}', Artist: '{artist}' from entity ID {entity_id}")
                 break
-        if not title or not artist: # ADDED: Check if title/artist still not found after fallback.
-             logger.error(f"Match endpoint: Could not determine title and artist for URL: {track_url}") # ADDED: Logging.
-             return jsonify({"detail": "Could not determine title and artist from Song.link info."}), 404 # MODIFIED: Flask-style JSON error.
-
 
-    youtube_url = extract_url(track_info.get("linksByPlatform", {}), "youtube") # MODIFIED: Used .get() for safer access.
+    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.")
 
+    youtube_url = extract_url(track_info.get("linksByPlatform", {}), "youtube")
+    
     if youtube_url:
         video_id = None
-        # MODIFIED: Improved video_id extraction from youtube_url, handles direct watch links and youtu.be, and strips extra params.
+        
         if "v=" in youtube_url:
             video_id = youtube_url.split("v=")[1].split("&")[0]
-        elif "youtu.be/" in youtube_url: # MODIFIED: Handling for youtu.be links if present in song.link
+        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}") # ADDED: Logging.
-        # MODIFIED: Flask-style JSON response instead of returning dict directly.
-        return jsonify({"url": youtube_url, "filename": filename, "track_id": video_id}), 200
+        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:
-        logger.info(f"Match endpoint: No direct YouTube URL. Searching YTMusic with: '{title} - {artist}'") # ADDED: Logging.
-        # ADDED: Explicit check if title or artist is missing before searching.
-        if not title or not artist:
-            logger.error("Match endpoint: Cannot search YTMusic without title and artist.") # ADDED: Logging.
-            return jsonify({"detail": "Cannot search on YouTube Music without title and artist information."}), 400 # MODIFIED: Flask-style JSON error.
-
-        search_query = f'{title} {artist}' # MODIFIED: Changed from '+' to space for a more natural search query.
+        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:
-            # MODIFIED: Improved logic to pick the first song with a videoId using next() and .get().
             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"]
-                # MODIFIED: Changed ym_url to a standard YouTube watch URL format.
                 ym_url = f'https://music.youtube.com/watch?v={video_id}'
-                # MODIFIED: More robust filename generation using .get() and providing fallbacks.
-                filename = f"{first_song.get('title', title)} - {first_song.get('artists', [{'name': artist}])[0]['name']}"
-                logger.info(f"Match endpoint: Found YTMusic search result - URL: {ym_url}, Video ID: {video_id}") # ADDED: Logging.
-                # MODIFIED: Flask-style JSON response.
-                return jsonify({"filename": filename, "url": ym_url, "track_id": video_id}), 200
+                
+                # 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.") # ADDED: Logging.
-                # MODIFIED: Flask-style JSON error response.
-                return jsonify({"detail": "No matching video ID found on YouTube Music after search."}), 404
+                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.") # ADDED: Logging.
-            # MODIFIED: Flask-style JSON error response.
-            return jsonify({"detail": "No results found on YouTube Music for the track."}), 404
-    # REMOVED: The final `raise HTTPException` was determined to be unreachable and removed.
-
-    
+            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.")
 
 class ApiRotator:
     def __init__(self, apis):
@@ -207,10 +214,9 @@ class ApiRotator:
 
     def get_prioritized_apis(self):
         if self.last_successful_index is not None:
-            # Move the last successful API to the front
             rotated_apis = (
-                [self.apis[self.last_successful_index]] + 
-                self.apis[:self.last_successful_index] + 
+                [self.apis[self.last_successful_index]] +
+                self.apis[:self.last_successful_index] +
                 self.apis[self.last_successful_index+1:]
             )
             return rotated_apis
@@ -219,21 +225,18 @@ class ApiRotator:
     def update_last_successful(self, index):
         self.last_successful_index = index
 
-# In your function:
 api_rotator = ApiRotator([
     "https://cobalt-api.ayo.tf",
     "https://cobalt-api.kwiatekmiki.com",
     "http://34.107.254.11",
     "https://dwnld.nichind.dev",
     "https://yt.edd1e.xyz/"
-    
 ])
 
-
-
 async def get_track_download_url(track_id: str, quality: str) -> str:
     apis = api_rotator.get_prioritized_apis()
-    session = cloudscraper.create_scraper()  # Requires cloudscraper package
+    session = cloudscraper.create_scraper()
+    
     headers = {
         "Accept": "application/json",
         "Content-Type": "application/json",
@@ -244,12 +247,19 @@ async def get_track_download_url(track_id: str, quality: str) -> str:
         try:
             logger.info(f"Attempting to get download URL from: {api_url}")
             y_url = f"https://youtu.be/{track_id}"
-            response = session.post(
-                api_url,
-                timeout=20,
-                json={"url": y_url, "audioFormat": "mp3", "downloadMode": "audio", "audioBitrate": quality},
-                headers=headers
+            
+            # Use asyncio to run the blocking request in a thread pool
+            loop = asyncio.get_event_loop()
+            response = await loop.run_in_executor(
+                None,
+                lambda: session.post(
+                    api_url,
+                    timeout=20,
+                    json={"url": y_url, "audioFormat": "mp3", "downloadMode": "audio", "audioBitrate": quality},
+                    headers=headers
+                )
             )
+
             logger.info(f"Response status: {response.status_code}")
             logger.info(f"Response content: {response.content}")
 
@@ -259,81 +269,72 @@ async def get_track_download_url(track_id: str, quality: str) -> str:
                 
                 if error_code == "error.api.content.video.unavailable":
                     logger.warning(f"Video unavailable error from {api_url}")
-                    break  # Only break for specific error
+                    break
                 
                 if "url" in json_response:
                     api_rotator.update_last_successful(i)
                     return json_response["url"]
-                
+
         except Exception as e:
             logger.error(f"Failed with {api_url}: {str(e)}")
             continue
-    
-    logger.error(f"No download URL found")
-    return {"error": "Download URL not found"}
-
-
-
-    
-
 
+    logger.error(f"No download URL found")
+    return ""
 
-@app.route('/track_dl', methods=['POST'])
-async def track_dl():
-   
-    data = request.get_json()
-    track_id = data.get('track_id')
-    quality = data.get('quality', '128')
-    
+@app.post("/track_dl")
+async def track_dl(request: TrackDownloadRequest):
     try:
-        quality_num = int(quality)
-        if quality_num > 128 or quality.upper() == 'FLAC':
-            return jsonify({
-                "error": "Quality above 128 or FLAC is for Premium users Only.",
-                "premium": "https://chrunos.com/premium-shortcuts/"
-            }), 400
-            
-        dl_url = await get_track_download_url(track_id, quality)
+        quality_num = int(request.quality)
+        if quality_num > 128 or request.quality.upper() == 'FLAC':
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error": "Quality above 128 or FLAC is for Premium users Only.",
+                    "premium": "https://chrunos.com/premium-shortcuts/"
+                }
+            )
+
+        dl_url = await get_track_download_url(request.track_id, request.quality)
         
         if dl_url and "http" in dl_url:
-                       
-            result = {
+            return {
                 "url": dl_url,
                 "premium": "https://chrunos.com/premium-shortcuts/"
             }
-            return jsonify(result)
         else:
-            return jsonify({
-                "error": "Failed to Fetch the Track.",
-                "premium": "https://chrunos.com/premium-shortcuts/"
-            }), 400
-            
-    except ValueError:
-        return jsonify({
-            "error": "Invalid quality value provided. It should be a valid integer or FLAC.",
-            "premium": "https://chrunos.com/premium-shortcuts/"
-        }), 400
-    
-
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error": "Failed to Fetch the Track.",
+                    "premium": "https://chrunos.com/premium-shortcuts/"
+                }
+            )
 
+    except ValueError:
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error": "Invalid quality value provided. It should be a valid integer or FLAC.",
+                "premium": "https://chrunos.com/premium-shortcuts/"
+            }
+        )
 
-@app.route('/get_artist', methods=['GET'])
-def get_artist():
-    artist_id = request.args.get('id')
-    artist_info = ytmusic.get_artist(artist_id)
-    return jsonify(artist_info)
+@app.get("/get_artist")
+async def get_artist(id: str):
+    artist_info = ytmusic.get_artist(id)
+    return artist_info
 
-@app.route('/get_album', methods=['GET'])
-def get_album():
-    album_id = request.args.get('id')
-    album_info = ytmusic.get_album(album_id)
-    return jsonify(album_info)
+@app.get("/get_album")
+async def get_album(id: str):
+    album_info = ytmusic.get_album(id)
+    return album_info
 
-@app.route('/get_song', methods=['GET'])
-def get_song():
-    song_id = request.args.get('id')
-    song_info = ytmusic.get_song(song_id)
-    return jsonify(song_info)
+@app.get("/get_song")
+async def get_song(id: str):
+    song_info = ytmusic.get_song(id)
+    return song_info
 
 if __name__ == '__main__':
-    app.run(host='0.0.0.0', port=7860)
+    import uvicorn
+    uvicorn.run(app, host='0.0.0.0', port=7860)