v2.7

README.md

@@ -1,28 +1,32 @@
 [![GitHub All Releases](https://img.shields.io/github/downloads/afkarxyz/SpotiFLAC/total?style=for-the-badge)](https://github.com/afkarxyz/SpotiFLAC/releases)
 
-![spotiflac](https://github.com/user-attachments/assets/a233a276-14a4-4f4c-b267-f182dd3912a0)
+![SpotiFLAC](https://github.com/user-attachments/assets/b4c4f403-edbd-4a71-b74b-c7d433d47d06)
 
 <div align="center">
 <b>SpotiFLAC</b> allows you to download Spotify tracks in true FLAC format through services like Tidal, Amazon Music and Deezer with the help of Lucida.
 </div>
 
-### [Download](https://github.com/afkarxyz/SpotiFLAC/releases/download/v2.0/SpotiFLAC.exe)
+### [Download](https://github.com/afkarxyz/SpotiFLAC/releases/download/v2.6/SpotiFLAC.exe)
 
 #
 
-> [!WARNING] 
-Sometimes, the **download speed** from Lucida can be fast or slow; it varies unpredictably.
+> [!Note] 
+**Download speed** from Lucida is unpredictable—sometimes fast, sometimes slow. Join their [Discord](https://discord.com/invite/dXEGRWqEbS) for updates.
 
 ## Screenshots
 
+![image](https://github.com/user-attachments/assets/70a5dceb-3374-4255-8f6a-4afb5ee534b0)
+
+![image](https://github.com/user-attachments/assets/9f0d6aa5-456b-4a90-b48a-7e0c22819ebd)
+
+![image](https://github.com/user-attachments/assets/be9a79b5-260e-4948-9f72-10c735210ab7)
+
+![image](https://github.com/user-attachments/assets/1feec621-f8bf-4b2a-ae73-afcb1fb1deba)
+
+![image](https://github.com/user-attachments/assets/66cc3398-547d-4568-8d49-a05ad4997370)
+
 > When **Fallback** is enabled, it will use the backup server `Lucida.su`
 
-![image](https://github.com/user-attachments/assets/3db51367-45dc-470f-8d6e-8f783ebd6340)
-
-![image](https://github.com/user-attachments/assets/a9020973-f79c-40ba-ab76-e4a3955a1ba4)
-
-![image](https://github.com/user-attachments/assets/cf4d09dd-144f-4e7f-a204-78aad353cdbf)
-
 ## Lossless Audio Check
 
 ![image](https://github.com/user-attachments/assets/d63b422d-0ea3-4307-850f-96c99d7eaa9a)

SquidWTF.py

@@ -0,0 +1,228 @@
+import requests
+from mutagen.flac import FLAC, Picture
+from datetime import datetime
+import sys
+import os
+
+def get_track_info(isrc):
+    print(f"Search: {isrc}")
+    url = f"https://us.qobuz.squid.wtf/api/get-music?q={isrc}&offset=0"
+    response = requests.get(url)
+    data = response.json()
+    
+    if not data.get("success"):
+        raise Exception("Failed to get track info")
+    
+    tracks = data["data"]["tracks"]["items"]
+    if not tracks:
+        print(f"Not Found: {isrc}")
+        raise Exception(f"No tracks found for ISRC: {isrc}")
+    
+    track = None
+    for item in tracks:
+        if item["isrc"] == isrc:
+            track = item
+            break
+    
+    if not track:
+        print(f"Not Found: {isrc}")
+        raise Exception(f"No track with matching ISRC: {isrc}")
+    
+    print(f"Found: {track['title']} - {track['performer']['name']}")
+    return track
+
+def search_track(title, artist, strict_match=False):
+    print(f"Search by title/artist: {title} - {artist}")
+    
+    search_query = f"{title} {artist}".replace("feat.", "").replace("ft.", "")
+    
+    url = f"https://us.qobuz.squid.wtf/api/get-music?q={search_query}&offset=0"
+    response = requests.get(url)
+    data = response.json()
+    
+    if not data.get("success"):
+        raise Exception("Failed to search for track")
+    
+    tracks = data["data"]["tracks"]["items"]
+    if not tracks:
+        print(f"Not Found: {title} - {artist}")
+        raise Exception(f"No tracks found for: {title} - {artist}")
+    
+    best_match = None
+    title_lower = title.lower()
+    artist_lower = artist.lower()
+    
+    for item in tracks:
+        item_title = item["title"].lower()
+        item_artist = item["performer"]["name"].lower()
+        
+        if title_lower == item_title and (artist_lower in item_artist or item_artist in artist_lower):
+            best_match = item
+            print(f"Found exact title match with artist: {item['title']} - {item['performer']['name']}")
+            break
+    
+    if not best_match and not strict_match:
+        for item in tracks:
+            item_title = item["title"].lower()
+            item_artist = item["performer"]["name"].lower()
+            
+            if title_lower in item_title and (artist_lower in item_artist or item_artist in artist_lower):
+                best_match = item
+                print(f"Found partial match: {item['title']} - {item['performer']['name']}")
+                break
+    
+    if strict_match and best_match:
+        item_artist = best_match["performer"]["name"].lower()
+        if artist_lower not in item_artist and item_artist not in artist_lower:
+            print(f"Artist mismatch in strict mode: Expected '{artist}', found '{best_match['performer']['name']}'")
+            best_match = None
+    
+    if not best_match and not strict_match and tracks:
+        best_match = tracks[0]
+        print(f"No good match, using first result: {best_match['title']} - {best_match['performer']['name']}")
+    
+    if not best_match:
+        print(f"Not Found: {title} - {artist}")
+        raise Exception(f"No suitable track found for: {title} - {artist}")
+    
+    print(f"Found by title search: {best_match['title']} - {best_match['performer']['name']}")
+    return best_match
+
+def get_download_url(track_id):
+    url = f"https://us.qobuz.squid.wtf/api/download-music?track_id={track_id}&quality=27"
+    response = requests.get(url)
+    data = response.json()
+    
+    if not data.get("success"):
+        raise Exception("Failed to get download URL")
+    
+    return data["data"]["url"]
+
+def download_file(url, filename, progress_callback=None):
+    directory = os.path.dirname(filename)
+    if directory and not os.path.exists(directory):
+        try:
+            os.makedirs(directory, exist_ok=True)
+            print(f"Created directory: {directory}")
+        except Exception as e:
+            raise Exception(f"Failed to create directory {directory}: {str(e)}")
+    
+    try:
+        with open(filename, 'wb') as test_file:
+            pass
+    except Exception as e:
+        raise Exception(f"Cannot write to file {filename}: {str(e)}")
+    
+    try:
+        response = requests.get(url, stream=True)
+        
+        if response.status_code != 200:
+            raise Exception(f"Failed to download file: {response.status_code}")
+        
+        total_size = int(response.headers.get('content-length', 0))
+        downloaded = 0
+        
+        with open(filename, 'wb') as f:
+            for chunk in response.iter_content(chunk_size=8192):
+                if chunk:
+                    f.write(chunk)
+                    downloaded += len(chunk)
+                    
+                    if total_size > 0 and progress_callback:
+                        progress_callback(downloaded, total_size)
+                    elif total_size > 0:
+                        progress = (downloaded / total_size) * 100
+                        sys.stdout.write(f"\rProgress Download: {progress:.1f}%")
+                        sys.stdout.flush()
+        
+        if total_size > 0:
+            sys.stdout.write("\n")
+        
+        if not os.path.exists(filename) or os.path.getsize(filename) == 0:
+            raise Exception(f"Download failed: File {filename} is empty or does not exist")
+        
+        return filename
+    except Exception as e:
+        if os.path.exists(filename):
+            try:
+                os.remove(filename)
+                print(f"Removed incomplete file: {filename}")
+            except:
+                pass
+        raise Exception(f"Download failed: {str(e)}")
+
+def embed_metadata(filename, track_info):
+    if not os.path.exists(filename):
+        raise Exception(f"Cannot embed metadata: File {filename} does not exist")
+    
+    try:
+        print("Embedding Tags...")
+        audio = FLAC(filename)
+        audio.clear()
+        
+        audio["TITLE"] = track_info["title"]
+        audio["ARTIST"] = track_info["performer"]["name"]
+        audio["ALBUM"] = track_info["album"]["title"]
+        audio["ALBUMARTIST"] = track_info["album"]["artist"]["name"]
+        audio["TRACKNUMBER"] = str(track_info["track_number"])
+        audio["LABEL"] = track_info["album"]["label"]["name"]
+        audio["GENRE"] = track_info["album"]["genre"]["name"]
+        
+        release_date = datetime.fromtimestamp(track_info["album"]["released_at"]).strftime("%Y-%m-%d")
+        release_year = release_date.split("-")[0]
+        
+        audio["DATE"] = release_date
+        audio["YEAR"] = release_year
+        audio["ISRC"] = track_info["isrc"]
+        audio["COPYRIGHT"] = track_info["copyright"]
+        
+        if track_info["album"]["image"]["large"]:
+            try:
+                cover_data = download_cover_image(track_info["album"]["image"]["large"])
+                picture = Picture()
+                picture.type = 3
+                picture.mime = "image/jpeg"
+                picture.desc = ""
+                picture.data = cover_data
+                
+                audio.add_picture(picture)
+            except Exception as e:
+                print(f"Warning: Could not add cover image: {str(e)}")
+        
+        audio.save()
+    except Exception as e:
+        raise Exception(f"Failed to embed metadata: {str(e)}")
+
+def download_cover_image(url):
+    response = requests.get(url)
+    
+    if response.status_code != 200:
+        raise Exception(f"Failed to download cover image: {response.status_code}")
+    
+    return response.content
+
+def main():
+    try:
+        isrc = "USUM72409273"
+        
+        track_info = get_track_info(isrc)
+        track_id = track_info["id"]
+        
+        if track_info["isrc"] != isrc:
+            raise Exception(f"ISRC mismatch: {track_info['isrc']} != {isrc}")
+        
+        download_url = get_download_url(track_id)
+        
+        filename = f"{track_info['title']} - {track_info['performer']['name']}.flac"
+        filename = filename.replace('/', '_').replace('\\', '_')
+        
+        download_file(download_url, filename)
+        embed_metadata(filename, track_info)
+        
+        print("Downloaded Successfully!")
+        
+    except Exception as e:
+        print(f"Error: {e}")
+
+if __name__ == "__main__":
+    main()

getMetadata.py

@@ -0,0 +1,478 @@
+from time import sleep
+from urllib.parse import urlparse, parse_qs
+import requests
+import json
+import hmac
+import time
+import hashlib
+from typing import Tuple, Callable, Dict, Any, List
+
+_TOTP_SECRET = bytearray([53,53,48,55,49,52,53,56,53,51,52,56,55,52,57,57,53,57,50,50,52,56,54,51,48,51,50,57,51,52,55])
+
+def generate_totp(
+    secret: bytes = _TOTP_SECRET,
+    algorithm: Callable[[], object] = hashlib.sha1,
+    digits: int = 6,
+    counter_factory: Callable[[], int] = lambda: int(time.time()) // 30,
+) -> Tuple[str, int]:
+    counter = counter_factory()
+    hmac_result = hmac.new(
+        secret, counter.to_bytes(8, byteorder="big"), algorithm
+    ).digest()
+
+    offset = hmac_result[-1] & 15
+    truncated_value = (
+        (hmac_result[offset] & 127) << 24
+        | (hmac_result[offset + 1] & 255) << 16
+        | (hmac_result[offset + 2] & 255) << 8
+        | (hmac_result[offset + 3] & 255)
+    )
+    return (
+        str(truncated_value % (10**digits)).zfill(digits),
+        counter * 30_000,
+    )
+
+token_url = 'https://open.spotify.com/get_access_token'
+playlist_base_url = 'https://api.spotify.com/v1/playlists/{}'
+album_base_url = 'https://api.spotify.com/v1/albums/{}'
+track_base_url = 'https://api.spotify.com/v1/tracks/{}'
+headers = {
+    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36',
+    'Accept': 'application/json',
+    'Accept-Language': 'en-US,en;q=0.9',
+    'Accept-Encoding': 'gzip, deflate, br',
+    'sec-ch-ua-platform': '"Windows"',
+    'sec-fetch-dest': 'empty',
+    'sec-fetch-mode': 'cors',
+    'sec-fetch-site': 'same-origin',
+    'Referer': 'https://open.spotify.com/',
+    'Origin': 'https://open.spotify.com'
+}
+
+class SpotifyInvalidUrlException(Exception):
+    pass
+
+class SpotifyWebsiteParserException(Exception):
+    pass
+
+def parse_uri(uri):
+    u = urlparse(uri)
+    if u.netloc == "embed.spotify.com":
+        if not u.query:
+            raise SpotifyInvalidUrlException("ERROR: url {} is not supported".format(uri))
+        qs = parse_qs(u.query)
+        return parse_uri(qs['uri'][0])
+
+    if not u.scheme and not u.netloc:
+        return {"type": "playlist", "id": u.path}
+
+    if u.scheme == "spotify":
+        parts = uri.split(":")
+    else:
+        if u.netloc != "open.spotify.com" and u.netloc != "play.spotify.com":
+            raise SpotifyInvalidUrlException("ERROR: url {} is not supported".format(uri))
+        parts = u.path.split("/")
+
+    if parts[1] == "embed":
+        parts = parts[1:]
+
+    l = len(parts)
+    if l == 3 and parts[1] in ["album", "track", "playlist"]:
+        return {"type": parts[1], "id": parts[2]}
+    if l == 5 and parts[3] == "playlist":
+        return {"type": parts[3], "id": parts[4]}
+
+    raise SpotifyInvalidUrlException("ERROR: unable to determine Spotify URL type or type is unsupported.")
+
+def get_json_from_api(api_url, access_token):
+    headers.update({'Authorization': 'Bearer {}'.format(access_token)})
+    
+    req = requests.get(api_url, headers=headers, timeout=10)
+
+    if req.status_code == 429:
+        seconds = int(req.headers.get("Retry-After", "5")) + 1
+        print(f"INFO: rate limited! Sleeping for {seconds} seconds")
+        sleep(seconds)
+        return None
+
+    if req.status_code != 200:
+        raise SpotifyWebsiteParserException(f"ERROR: {api_url} gave us not a 200. Instead: {req.status_code}")
+        
+    return req.json()
+
+def get_access_token():
+    try:
+        totp, timestamp = generate_totp()
+        
+        params = {
+            "reason": "init",
+            "productType": "web-player",
+            "totp": totp,
+            "totpVer": 5,
+            "ts": timestamp,
+        }
+        
+        req = requests.get(token_url, headers=headers, params=params, timeout=10)
+        if req.status_code != 200:
+            return {"error": f"Failed to get access token. Status code: {req.status_code}"}
+        return req.json()
+    except Exception as e:
+        return {"error": f"Failed to get access token: {str(e)}"}
+
+def fetch_tracks_in_batches(url: str, access_token: str, batch_size: int = 100, delay: float = 1.0) -> Tuple[List[Dict[str, Any]], int]:
+    all_tracks = []
+    current_batch = 0
+    
+    while url:
+        print(f"Batch : {current_batch}")
+        
+        url_parts = url.split("offset=")
+        if len(url_parts) > 1:
+            offset_part = url_parts[1].split("&")[0]
+            print(f"Offset : {offset_part}")
+        print("-------------")
+        
+        track_data = get_json_from_api(url, access_token)
+        if not track_data:
+            break
+        
+        items = track_data.get('items', [])
+        all_tracks.extend(items)
+        
+        url = track_data.get('next')
+        if url and "&locale=" in url:
+            url = url.split("&locale=")[0]
+            
+        if url and delay > 0:
+            sleep(delay)
+        
+        current_batch += 1
+        
+    return all_tracks, current_batch
+
+def get_raw_spotify_data(spotify_url, batch: bool = False, delay: float = 1.0):
+    url_info = parse_uri(spotify_url)
+    token = get_access_token()
+    
+    if "error" in token:
+        return token
+    
+    access_token = token["accessToken"]
+    raw_data = {}
+    
+    if url_info['type'] == "playlist":
+        try:
+            playlist_data = get_json_from_api(
+                playlist_base_url.format(url_info["id"]), 
+                access_token
+            )
+            if not playlist_data:
+                return {"error": "Failed to get playlist data"}
+                
+            raw_data = playlist_data
+            total_tracks = playlist_data.get('tracks', {}).get('total', 0)
+            
+            if batch:
+                tracks_url = f'https://api.spotify.com/v1/playlists/{url_info["id"]}/tracks?limit=100'
+                tracks, num_batches = fetch_tracks_in_batches(tracks_url, access_token, 100, delay)
+                raw_data['tracks']['items'] = tracks
+                raw_data['_batch_count'] = num_batches
+                raw_data['_batch_enabled'] = True
+                
+                if len(tracks) < total_tracks:
+                    last_offset = len(tracks)
+                    remaining_tracks = []
+                    
+                    while last_offset < total_tracks:
+                        print(f"Batch : {num_batches}")
+                        print(f"Offset : {last_offset}")
+                        print("-------------")
+                        
+                        remainder_url = f'https://api.spotify.com/v1/playlists/{url_info["id"]}/tracks?offset={last_offset}&limit=100'
+                        track_data = get_json_from_api(remainder_url, access_token)
+                        
+                        if not track_data or not track_data.get('items'):
+                            break
+                            
+                        items = track_data.get('items', [])
+                        remaining_tracks.extend(items)
+                        
+                        if len(items) < 100:
+                            break
+                            
+                        last_offset += len(items)
+                        num_batches += 1
+                        
+                        if delay > 0:
+                            sleep(delay)
+                    
+                    tracks.extend(remaining_tracks)
+                    raw_data['tracks']['items'] = tracks
+                    raw_data['_batch_count'] = num_batches
+            else:
+                tracks = []
+                tracks_url = f'https://api.spotify.com/v1/playlists/{url_info["id"]}/tracks?limit=100'
+                while tracks_url:
+                    track_data = get_json_from_api(tracks_url, access_token)
+                    if not track_data:
+                        break
+                        
+                    tracks.extend(track_data['items'])
+                    tracks_url = track_data.get('next')
+                    if tracks_url and "&locale=" in tracks_url:
+                        tracks_url = tracks_url.split("&locale=")[0]
+                    
+                raw_data['tracks']['items'] = tracks
+                raw_data['_batch_enabled'] = False
+                
+        except Exception as e:
+            return {"error": f"Failed to get playlist data: {str(e)}"}
+            
+    elif url_info["type"] == "album":
+        try:
+            album_data = get_json_from_api(
+                album_base_url.format(url_info["id"]),
+                access_token
+            )
+            if not album_data:
+                return {"error": "Failed to get album data"}
+                
+            album_data['_token'] = access_token
+            raw_data = album_data
+            total_tracks = album_data.get('total_tracks', 0)
+            
+            if batch:
+                tracks_url = f'{album_base_url.format(url_info["id"])}/tracks?limit=50'
+                tracks, num_batches = fetch_tracks_in_batches(tracks_url, access_token, 50, delay)
+                raw_data['tracks']['items'] = tracks
+                raw_data['_batch_count'] = num_batches
+                raw_data['_batch_enabled'] = True
+                
+                if len(tracks) < total_tracks:
+                    last_offset = len(tracks)
+                    remaining_tracks = []
+                    
+                    while last_offset < total_tracks:
+                        print(f"Batch : {num_batches}")
+                        print(f"Offset : {last_offset}")
+                        print("-------------")
+                        
+                        remainder_url = f'{album_base_url.format(url_info["id"])}/tracks?offset={last_offset}&limit=50'
+                        track_data = get_json_from_api(remainder_url, access_token)
+                        
+                        if not track_data or not track_data.get('items'):
+                            break
+                            
+                        items = track_data.get('items', [])
+                        remaining_tracks.extend(items)
+                        
+                        if len(items) < 50:
+                            break
+                            
+                        last_offset += len(items)
+                        num_batches += 1
+                        
+                        if delay > 0:
+                            sleep(delay)
+                    
+                    tracks.extend(remaining_tracks)
+                    raw_data['tracks']['items'] = tracks
+                    raw_data['_batch_count'] = num_batches
+            else:
+                tracks = []
+                tracks_url = f'{album_base_url.format(url_info["id"])}/tracks?limit=50'
+                while tracks_url:
+                    track_data = get_json_from_api(tracks_url, access_token)
+                    if not track_data:
+                        break
+                        
+                    tracks.extend(track_data['items'])
+                    tracks_url = track_data.get('next')
+                    if tracks_url and "&locale=" in tracks_url:
+                        tracks_url = tracks_url.split("&locale=")[0]
+                    
+                raw_data['tracks']['items'] = tracks
+                raw_data['_batch_enabled'] = False
+                
+        except Exception as e:
+            return {"error": f"Failed to get album data: {str(e)}"}
+                
+    elif url_info["type"] == "track":
+        try:
+            track_data = get_json_from_api(
+                track_base_url.format(url_info["id"]),
+                access_token
+            )
+            if not track_data:
+                return {"error": "Failed to get track data"}
+                
+            raw_data = track_data
+        except Exception as e:
+            return {"error": f"Failed to get track data: {str(e)}"}
+
+    return raw_data
+
+def format_track_data(track_data):
+    artists = []
+    for artist in track_data.get('artists', []):
+        artists.append(artist['name'])
+    
+    image_url = track_data.get('album', {}).get('images', [{}])[0].get('url', '') if track_data.get('album', {}).get('images') else ''
+    
+    isrc = track_data.get('external_ids', {}).get('isrc', '')
+    
+    return {
+        "track": {
+            "artists": ", ".join(artists),
+            "name": track_data.get('name', ''),
+            "album_name": track_data.get('album', {}).get('name', ''),
+            "duration_ms": track_data.get('duration_ms', 0),
+            "images": image_url,
+            "release_date": track_data.get('album', {}).get('release_date', ''),
+            "track_number": track_data.get('track_number', 0),
+            "external_urls": track_data.get('external_urls', {}).get('spotify', ''),
+            "isrc": isrc  
+        }
+    }
+
+def format_album_data(album_data):
+    artists = []
+    for artist in album_data.get('artists', []):
+        artists.append(artist['name'])
+    
+    image_url = album_data.get('images', [{}])[0].get('url', '') if album_data.get('images') else ''
+    
+    track_list = []
+    for track in album_data.get('tracks', {}).get('items', []):
+        track_artists = []
+        for artist in track.get('artists', []):
+            track_artists.append(artist['name'])
+        
+        track_id = track.get('id', '')
+        track_isrc = ''
+        
+        if track_id and album_data.get('_token'):
+            try:
+                full_track_data = get_json_from_api(
+                    track_base_url.format(track_id),
+                    album_data.get('_token')
+                )
+                if full_track_data:
+                    track_isrc = full_track_data.get('external_ids', {}).get('isrc', '')
+            except:
+                pass
+            
+        track_list.append({
+            "artists": ", ".join(track_artists),
+            "name": track.get('name', ''),
+            "album_name": album_data.get('name', ''),
+            "duration_ms": track.get('duration_ms', 0),
+            "images": image_url,
+            "release_date": album_data.get('release_date', ''),
+            "track_number": track.get('track_number', 0),
+            "external_urls": track.get('external_urls', {}).get('spotify', ''),
+            "isrc": track_isrc  
+        })
+    
+    album_info = {
+        "total_tracks": album_data.get('total_tracks', 0),
+        "name": album_data.get('name', ''),
+        "release_date": album_data.get('release_date', ''),
+        "artists": ", ".join(artists),
+        "images": image_url
+    }
+    
+    if album_data.get('_batch_enabled', False):
+        album_info["batch"] = f"{album_data.get('_batch_count', 1)}"
+    
+    return {
+        "album_info": album_info,
+        "track_list": track_list
+    }
+
+def format_playlist_data(playlist_data):
+    image_url = playlist_data.get('images', [{}])[0].get('url', '') if playlist_data.get('images') else ''
+    
+    track_list = []
+    for item in playlist_data.get('tracks', {}).get('items', []):
+        track = item.get('track', {})
+        if not track:
+            continue
+            
+        artists = []
+        for artist in track.get('artists', []):
+            artists.append(artist['name'])
+            
+        track_image = ''
+        if track.get('album', {}).get('images'):
+            track_image = track.get('album', {}).get('images', [{}])[0].get('url', '')
+        
+        track_isrc = track.get('external_ids', {}).get('isrc', '')
+        
+        track_list.append({
+            "artists": ", ".join(artists),
+            "name": track.get('name', ''),
+            "album_name": track.get('album', {}).get('name', ''),
+            "duration_ms": track.get('duration_ms', 0),
+            "images": track_image,
+            "release_date": track.get('album', {}).get('release_date', ''),
+            "track_number": track.get('track_number', 0),
+            "external_urls": track.get('external_urls', {}).get('spotify', ''),
+            "isrc": track_isrc  
+        })
+    
+    playlist_info = {
+        "tracks": {"total": playlist_data.get('tracks', {}).get('total', 0)},
+        "followers": {"total": playlist_data.get('followers', {}).get('total', 0)},
+        "owner": {
+            "display_name": playlist_data.get('owner', {}).get('display_name', ''),
+            "name": playlist_data.get('name', ''),
+            "images": image_url
+        }
+    }
+    
+    if playlist_data.get('_batch_enabled', False):
+        playlist_info["batch"] = f"{playlist_data.get('_batch_count', 1)}"
+    
+    return {
+        "playlist_info": playlist_info,
+        "track_list": track_list
+    }
+
+def process_spotify_data(raw_data, data_type):
+    if not raw_data or "error" in raw_data:
+        return {"error": "Invalid data provided"}
+        
+    try:
+        if data_type == "track":
+            return format_track_data(raw_data)
+        elif data_type == "album":
+            return format_album_data(raw_data)
+        elif data_type == "playlist":
+            return format_playlist_data(raw_data)
+        else:
+            return {"error": "Invalid data type"}
+    except Exception as e:
+        return {"error": f"Error processing data: {str(e)}"}
+
+def get_filtered_data(spotify_url, batch=False, delay=1.0):
+    raw_data = get_raw_spotify_data(spotify_url, batch=batch, delay=delay)
+    if raw_data and "error" not in raw_data:
+        url_info = parse_uri(spotify_url)
+        filtered_data = process_spotify_data(raw_data, url_info['type'])
+        return filtered_data
+    return {"error": "Failed to get raw data"}
+
+if __name__ == '__main__':
+    playlist = "https://open.spotify.com/playlist/37i9dQZEVXbNG2KDcFcKOF"
+    album = "https://open.spotify.com/album/6J84szYCnMfzEcvIcfWMFL"
+    song = "https://open.spotify.com/track/7so0lgd0zP2Sbgs2d7a1SZ"
+    
+    filtered_playlist = get_filtered_data(playlist, batch=True, delay=0.1)
+    print(json.dumps(filtered_playlist, indent=2))
+    
+    filtered_album = get_filtered_data(album)
+    print(json.dumps(filtered_album, indent=2))
+    
+    filtered_track = get_filtered_data(song)
+    print(json.dumps(filtered_track, indent=2))

getTracks.py

@@ -2,67 +2,158 @@ import requests
 import time
 import os
 import asyncio
+import re
+import base64
 
 class TrackDownloader:
-    def __init__(self, use_fallback=False):
+    def __init__(self, use_fallback=False, timeout=30):
         self.client = requests.Session()
         self.headers = {
             'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
         }
         self.progress_callback = None
-        self.filename_format = 'title_artist'
         self.use_fallback = use_fallback
+        self.timeout = timeout
         self.base_domain = "lucida.su" if use_fallback else "lucida.to"
-        self.api_base = "https://apislucida.vercel.app"
 
     def set_progress_callback(self, callback):
         self.progress_callback = callback
-        
-    def set_filename_format(self, format_type):
-        self.filename_format = format_type
 
-    def generate_filename(self, metadata):
-        if self.filename_format == 'artist_title':
-            filename = f"{metadata['artists']} - {metadata['title']}.flac"
-        else:
-            filename = f"{metadata['title']} - {metadata['artists']}.flac"
-        return self.sanitize_filename(filename)
+    def generate_filename(self, track_id, service):
+        return f"{track_id}_{service}.flac"
 
     async def get_track_info(self, track_id, service="amazon", use_fallback=None):
         if use_fallback is None:
             use_fallback = self.use_fallback
             
-        fallback = "su" if use_fallback else "to"
-        api_url = f"{self.api_base}/{fallback}/{track_id}/{service}"
+        domain_type = "su" if use_fallback else "to"
+        
+        spotify_url = f"https://open.spotify.com/track/{track_id}"
+        
+        result = self.convert_spotify_link(spotify_url, service, domain_type)
+        
+        if "error" in result:
+            raise Exception(f"Failed to get track info: {result['error']}")
+        
+        result["track_id"] = track_id
+        
+        return result
+
+    def convert_spotify_link(self, spotify_url, target_service="amazon", domain_type="to"):
+        track_id_match = re.search(r'track/([a-zA-Z0-9]+)', spotify_url)
+        if not track_id_match:
+            return {"error": "Invalid Spotify URL"}
+        
+        domain = "lucida.to" if domain_type == "to" else "lucida.su"
+        base_url = f"https://{domain}"
+        
+        headers = {
+            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
+            "Accept-Language": "id-ID,id;q=0.9",
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "Host": domain,
+            "Pragma": "no-cache",
+            "Upgrade-Insecure-Requests": "1",
+            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
+        }
         
         try:
-            response = requests.get(api_url)
-            response.raise_for_status()
-            return response.json()
-        except requests.exceptions.RequestException as e:
-            raise Exception(f"Failed to get track info: {str(e)}")
-
-    def sanitize_filename(self, filename):
-        invalid_chars = '<>:"/\\|?*'
-        for char in invalid_chars:
-            filename = filename.replace(char, '')
+            headers["Referer"] = f"{base_url}/?url={spotify_url}&country=auto"
             
-        filename = ' '.join(filename.split())
-        filename = filename.replace(' ,', ',')
-        filename = filename.replace(',', ', ')
-        while '  ' in filename:
-            filename = filename.replace('  ', ' ')
-        filename = filename.rsplit('.', 1)
-        filename[0] = filename[0].strip()
-        return '.'.join(filename)
+            request_params = {
+                "url": spotify_url,
+                "country": "auto",
+                "to": target_service
+            }
+            
+            session = requests.Session()
+            session.verify = True
+            
+            response = session.get(
+                base_url,
+                params=request_params,
+                headers=headers,
+                timeout=self.timeout
+            )
+            
+            html_content = response.text
+            
+            token_match = re.search(r'token:"([^"]+)"', html_content)
+            token_expiry_match = re.search(r'tokenExpiry:(\d+)', html_content)
+            
+            token = token_match.group(1) if token_match else None
+            token_expiry = int(token_expiry_match.group(1)) if token_expiry_match else None
+            
+            url = None
+            url_patterns = [
+                r'"url":"([^"]+)"',
+                r'href="(https?://[^"]*' + re.escape(target_service) + r'[^"]*track[^"]*)"',
+            ]
+            
+            for pattern in url_patterns:
+                url_match = re.search(pattern, html_content)
+                if url_match:
+                    url = url_match.group(1).replace('\\/', '/')
+                    break
+            
+            if not url:
+                redirect_patterns = [
+                    r'url=([^&"]+)',
+                    r'href="([^"]+)"',
+                    r'window\.location\.href\s*=\s*[\'"]([^\'"]+)[\'"]',
+                ]
+                
+                for pattern in redirect_patterns:
+                    matches = re.finditer(pattern, html_content)
+                    for match in matches:
+                        potential_url = match.group(1)
+                        if potential_url.startswith('http') and target_service.lower() in potential_url.lower():
+                            url = potential_url.replace('\\/', '/')
+                            break
+                
+                if not url:
+                    service_urls = re.finditer(r'(https?://[^"\s]+' + re.escape(target_service) + r'[^"\s]+)', html_content)
+                    for match in service_urls:
+                        url = match.group(1).replace('\\/', '/')
+                        break
+            
+            result = {
+                "service": target_service,
+                "url": url,
+                "token": {
+                    "primary": None,
+                    "expiry": None
+                }
+            }
+            
+            if token:
+                try:
+                    decoded_once = base64.b64decode(token).decode('latin1')
+                    decoded_token = base64.b64decode(decoded_once).decode('latin1')
+                    result["token"]["primary"] = decoded_token
+                except Exception:
+                    result["token"]["primary"] = token
+            
+            result["token"]["expiry"] = token_expiry
+            
+            return result
+                
+        except Exception as error:
+            return {"error": str(error)}
 
-    def download(self, metadata, output_dir):
+    def download(self, metadata, output_dir, is_paused_callback=None, is_stopped_callback=None):
         track_url = metadata['url']
         primary_token = metadata['token']['primary']
         expiry = metadata['token']['expiry']
+        track_id = metadata['track_id']
+        service = metadata['service']
         
         print(f"Starting download for: {track_url}")
         
+        if is_stopped_callback and is_stopped_callback():
+            raise Exception("Download stopped by user")
+        
         initial_request = {
             "account": {"id": "auto", "type": "country"},
             "compat": "false",
@@ -94,12 +185,20 @@ class TrackDownloader:
         handoff = initial_response["handoff"]
         server = initial_response["server"]
 
-        file_name = self.generate_filename(metadata)
+        file_name = self.generate_filename(track_id, service)
 
         completion_url = f"https://{server}.{self.base_domain}/api/fetch/request/{handoff}"
 
         print("Waiting for track processing to complete")
         while True:
+            if is_stopped_callback and is_stopped_callback():
+                raise Exception("Download stopped by user")
+                
+            while is_paused_callback and is_paused_callback():
+                time.sleep(0.1)
+                if is_stopped_callback and is_stopped_callback():
+                    raise Exception("Download stopped by user")
+            
             completion_response = self.client.get(completion_url, headers=self.headers).json()
             
             status = completion_response["status"]
@@ -143,6 +242,20 @@ class TrackDownloader:
                 last_update_time = start_time
                 
                 for chunk in response.iter_content(chunk_size=8192):
+                    if is_stopped_callback and is_stopped_callback():
+                        file.close()
+                        if os.path.exists(file_path):
+                            os.remove(file_path)
+                        raise Exception("Download stopped by user")
+                        
+                    while is_paused_callback and is_paused_callback():
+                        time.sleep(0.1)
+                        if is_stopped_callback and is_stopped_callback():
+                            file.close()
+                            if os.path.exists(file_path):
+                                os.remove(file_path)
+                            raise Exception("Download stopped by user")
+                    
                     if chunk:
                         file.write(chunk)
                         downloaded_size += len(chunk)
@@ -177,10 +290,12 @@ class TrackDownloader:
             raise e
 
 async def main():
-    downloader = TrackDownloader()
+    use_fallback = False  
+    downloader = TrackDownloader(use_fallback)
+    
     output_dir = "."
     track_id = "2plbrEY59IikOBgBGLjaoe"
-    service = "amazon"
+    service = "tidal"
     
     def progress_update(current, total):
         if total > 0:
@@ -192,7 +307,7 @@ async def main():
     try:
         print(f"Getting track info for ID: {track_id} from {service}")
         metadata = await downloader.get_track_info(track_id, service)
-        print(f"Track info received: {metadata['title']} by {metadata['artists']}")
+        print(f"Track info received, starting download process")
         
         downloaded_file = downloader.download(metadata, output_dir)
         print(f"\nFile downloaded successfully: {downloaded_file}")
@@ -200,4 +315,4 @@ async def main():
         print(f"An error occurred: {str(e)}")
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    asyncio.run(main())

version.json

@@ -1,3 +1,3 @@
 {
-  "version": "2.0"
+  "version": "2.6"
 }