Compare commits

..

37 Commits

Author SHA1 Message Date
afkarxyz 3eda3245ca v4.9 2025-10-12 00:23:26 +07:00
afkarxyz f6f238361c v4.9 2025-10-12 00:22:30 +07:00
afkarxyz 998730bbb3 v4.8 2025-10-11 18:18:27 +07:00
afkarxyz 56a1d29d78 v4.8 2025-10-11 18:17:07 +07:00
afkarxyz 4b7316636e v4.7 2025-10-05 04:03:54 +07:00
afkarxyz 55669ec45f v4.7 2025-10-05 03:59:50 +07:00
afkarxyz 3304b13828 Update README.md 2025-10-05 03:42:30 +07:00
afkarxyz 579bb1415a Merge pull request #61 from value1338/main
Add button to auto-delete already downloaded files and retry failed ones
2025-10-05 03:40:21 +07:00
afkarxyz f0e71261a5 Update version.json 2025-10-05 03:39:40 +07:00
value1338 e2ad51da34 included Skipped Songs for Removal from Playlist 2025-10-04 13:06:13 +02:00
value1338 9e403ab1ba deleted comments in the code 2025-10-03 20:21:12 +02:00
value1338 7058559ddc Add button to auto-delete already downloaded files and retry failed ones 2025-10-03 19:47:51 +02:00
afkarxyz 861f303a4f v4.6 2025-09-20 08:34:27 +07:00
afkarxyz 9a28e8bd94 v4.6 2025-09-20 08:32:12 +07:00
afkarxyz f75385c4e8 open.spotify.com/intl-pt 2025-09-16 15:37:54 +07:00
afkarxyz 2eac274ee0 v4.5 2025-09-11 13:28:49 +07:00
afkarxyz 49a8de1b35 v4.5 2025-09-11 13:26:52 +07:00
afkarxyz cd2500d1df Update README.md 2025-09-11 13:24:17 +07:00
afkarxyz ea1372f1fe v4.4 2025-08-08 00:20:44 +07:00
afkarxyz 65fbb9a8e9 v4.4 2025-08-08 00:20:57 +07:00
afkarxyz de16d9e25d v4.4 2025-08-08 00:17:03 +07:00
afkarxyz 6dd19b563b v4.3 2025-08-06 20:24:58 +07:00
afkarxyz 303b76d1ec v4.3 2025-08-06 20:21:52 +07:00
afkarxyz dbcd49225d v4.2 2025-07-26 09:40:32 +07:00
afkarxyz bdc7717ef3 v4.2 2025-07-26 09:37:45 +07:00
afkarxyz 9a7c539418 Update README.md 2025-07-26 08:39:19 +07:00
afkarxyz 888ce2b61c v4.1 2025-07-24 06:35:18 +07:00
afkarxyz e2e1ab1cfa v4.1 2025-07-24 06:34:25 +07:00
afkarxyz 9bddeab0d1 v4.1 2025-07-24 06:14:42 +07:00
afkarxyz 03a30ee09a v4.0 2025-07-22 14:03:41 +07:00
afkarxyz 2d908e2f75 v4.0 2025-07-22 08:02:54 +07:00
afkarxyz e8f7bf7313 v4.0 2025-07-22 07:59:28 +07:00
afkarxyz 1f0922f358 v4.0 2025-07-22 07:54:03 +07:00
afkarxyz 3f267a3fa1 v3.9.5 2025-07-22 07:42:53 +07:00
afkarxyz 22da74a027 v3.9 2025-07-21 17:33:39 +07:00
afkarxyz 783350fe88 v3.9 2025-07-21 17:28:44 +07:00
afkarxyz 0057d43f46 v3.8 2025-07-14 13:20:14 +07:00
17 changed files with 1796 additions and 864 deletions
+6 -6
View File
@@ -3,20 +3,20 @@
![SpotiFLAC](https://github.com/user-attachments/assets/b4c4f403-edbd-4a71-b74b-c7d433d47d06) ![SpotiFLAC](https://github.com/user-attachments/assets/b4c4f403-edbd-4a71-b74b-c7d433d47d06)
<div align="center"> <div align="center">
<b>SpotiFLAC</b> allows you to download Spotify tracks in true FLAC format through services like Qobuz & Tidal. <b>SpotiFLAC</b> allows you to download Spotify tracks in true FLAC format through services like Tidal & Deezer.
</div> </div>
### [Download](https://github.com/afkarxyz/SpotiFLAC/releases/download/v3.7/SpotiFLAC.exe) ### [Download](https://github.com/afkarxyz/SpotiFLAC/releases/download/v4.9/SpotiFLAC.exe)
## Screenshots ## Screenshots
![image](https://github.com/user-attachments/assets/70a5dceb-3374-4255-8f6a-4afb5ee534b0) ![image](https://github.com/user-attachments/assets/416fca55-b885-45c0-af2a-b8b7ce5d5ef3)
![image](https://github.com/user-attachments/assets/9f0d6aa5-456b-4a90-b48a-7e0c22819ebd) ![image](https://github.com/user-attachments/assets/f9b11da4-dbc3-435e-8954-5627ebe2ccdc)
![image](https://github.com/user-attachments/assets/be9a79b5-260e-4948-9f72-10c735210ab7) ![image](https://github.com/user-attachments/assets/7507e58d-e228-4edf-adf7-675731731019)
![image](https://github.com/user-attachments/assets/1feec621-f8bf-4b2a-ae73-afcb1fb1deba) ![image](https://github.com/user-attachments/assets/1c3beda2-236b-4452-8afd-a2dfedf389e5)
## Lossless Audio Check ## Lossless Audio Check
+990 -397
View File
File diff suppressed because it is too large Load Diff
+241
View File
@@ -0,0 +1,241 @@
import requests
import asyncio
import os
import sys
from mutagen.flac import FLAC
from random import randrange
def get_random_user_agent():
return f"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_{randrange(11, 15)}_{randrange(4, 9)}) AppleWebKit/{randrange(530, 537)}.{randrange(30, 37)} (KHTML, like Gecko) Chrome/{randrange(80, 105)}.0.{randrange(3000, 4500)}.{randrange(60, 125)} Safari/{randrange(530, 537)}.{randrange(30, 36)}"
class DeezerDownloader:
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
'User-Agent': get_random_user_agent()
})
self.progress_callback = None
def set_progress_callback(self, callback):
self.progress_callback = callback
def get_track_by_isrc(self, isrc):
try:
url = f"https://api.deezer.com/2.0/track/isrc:{isrc}"
response = self.session.get(url)
response.raise_for_status()
data = response.json()
if 'error' in data:
print(f"Error from Deezer API: {data['error']['message']}")
return None
return data
except requests.exceptions.RequestException as e:
print(f"Error fetching track data: {e}")
return None
def extract_metadata(self, track_data):
metadata = {}
metadata['title'] = track_data.get('title', '')
metadata['title_short'] = track_data.get('title_short', '')
metadata['duration'] = track_data.get('duration', 0)
metadata['track_position'] = track_data.get('track_position', 1)
metadata['disk_number'] = track_data.get('disk_number', 1)
metadata['isrc'] = track_data.get('isrc', '')
metadata['release_date'] = track_data.get('release_date', '')
metadata['explicit_lyrics'] = track_data.get('explicit_lyrics', False)
if 'artist' in track_data:
metadata['artist'] = track_data['artist'].get('name', '')
metadata['artist_id'] = track_data['artist'].get('id', '')
if 'contributors' in track_data:
artists = []
for contributor in track_data['contributors']:
if contributor.get('role') == 'Main':
artists.append(contributor.get('name', ''))
metadata['artists'] = ', '.join(artists) if artists else metadata.get('artist', '')
if 'album' in track_data:
album = track_data['album']
metadata['album'] = album.get('title', '')
metadata['album_id'] = album.get('id', '')
metadata['cover_url'] = album.get('cover_xl', album.get('cover_big', ''))
metadata['cover_md5'] = album.get('md5_image', '')
metadata['deezer_link'] = track_data.get('link', '')
metadata['preview_url'] = track_data.get('preview', '')
return metadata
def download_cover_art(self, cover_url, filename):
if not cover_url:
return None
try:
response = self.session.get(cover_url)
response.raise_for_status()
cover_path = f"{filename}_cover.jpg"
with open(cover_path, 'wb') as f:
f.write(response.content)
return cover_path
except Exception as e:
print(f"Error downloading cover art: {e}")
return None
def embed_metadata(self, file_path, metadata, cover_path=None):
try:
audio = FLAC(file_path)
audio.clear()
if metadata.get('title'):
audio['TITLE'] = metadata['title']
if metadata.get('artists'):
audio['ARTIST'] = metadata['artists']
elif metadata.get('artist'):
audio['ARTIST'] = metadata['artist']
if metadata.get('album'):
audio['ALBUM'] = metadata['album']
if metadata.get('release_date'):
audio['DATE'] = metadata['release_date']
if metadata.get('track_position'):
audio['TRACKNUMBER'] = str(metadata['track_position'])
if metadata.get('disk_number'):
audio['DISCNUMBER'] = str(metadata['disk_number'])
if metadata.get('isrc'):
audio['ISRC'] = metadata['isrc']
if cover_path and os.path.exists(cover_path):
with open(cover_path, 'rb') as f:
cover_data = f.read()
from mutagen.flac import Picture
picture = Picture()
picture.type = 3
picture.mime = 'image/jpeg'
picture.desc = 'Cover'
picture.data = cover_data
audio.add_picture(picture)
audio.save()
print(f"Metadata embedded successfully in {file_path}")
except Exception as e:
print(f"Error embedding metadata: {e}")
async def download_by_isrc(self, isrc, output_dir="."):
print(f"Fetching track info for ISRC: {isrc}")
track_data = self.get_track_by_isrc(isrc)
if not track_data:
print("Failed to get track data from Deezer API")
return False
metadata = self.extract_metadata(track_data)
print(f"Found track: {metadata.get('artists', 'Unknown')} - {metadata.get('title', 'Unknown')}")
track_id = track_data.get('id')
if not track_id:
print("No track ID found in Deezer API response")
return False
print(f"Using track ID: {track_id}")
api_url = f"https://api.deezmate.com/dl/{track_id}"
print(f"Requesting download links from: {api_url}")
try:
response = self.session.get(api_url)
response.raise_for_status()
api_data = response.json()
if not api_data.get('success'):
print("API request failed")
return False
links = api_data.get('links', {})
flac_url = links.get('flac')
if not flac_url:
print("No FLAC download link found in API response")
return False
print(f"Successfully obtained FLAC download URL")
except Exception as e:
print(f"Error getting download URL from API: {e}")
return False
print("Downloading FLAC file...")
try:
response = self.session.get(flac_url)
response.raise_for_status()
safe_title = "".join(c for c in metadata.get('title', 'Unknown') if c.isalnum() or c in (' ', '-', '_')).rstrip()
safe_artist = "".join(c for c in metadata.get('artists', 'Unknown') if c.isalnum() or c in (' ', '-', '_')).rstrip()
filename = f"{safe_artist} - {safe_title}.flac"
file_path = os.path.join(output_dir, filename)
with open(file_path, 'wb') as f:
f.write(response.content)
downloaded = len(response.content)
print(f"File size: {downloaded} bytes ({downloaded / (1024*1024):.2f} MB)")
if self.progress_callback:
self.progress_callback(downloaded, downloaded)
print(f"Downloaded: {file_path}")
cover_path = None
if metadata.get('cover_url'):
print("Downloading cover art...")
cover_path = self.download_cover_art(metadata['cover_url'],
os.path.join(output_dir, f"{safe_artist} - {safe_title}"))
print("Embedding metadata...")
self.embed_metadata(file_path, metadata, cover_path)
if cover_path and os.path.exists(cover_path):
os.remove(cover_path)
print(f"Successfully downloaded and tagged: {filename}")
return True
except Exception as e:
print(f"Error downloading file: {e}")
return False
async def main():
print("=== DeezerDL - Deezer Downloader ===")
downloader = DeezerDownloader()
isrc = "USAT22409172"
output_dir = "."
success = await downloader.download_by_isrc(isrc, output_dir)
if success:
print("Download completed successfully!")
else:
print("Download failed!")
if __name__ == "__main__":
try:
import sys
if sys.platform == "win32":
import os
os.system("chcp 65001 > nul")
try:
sys.stdout.reconfigure(encoding='utf-8')
except:
pass
except:
pass
asyncio.run(main())
+240 -5
View File
@@ -1,5 +1,6 @@
from time import sleep from time import sleep
from urllib.parse import urlparse, parse_qs from urllib.parse import urlparse, parse_qs
from pathlib import Path
import requests import requests
import json import json
import time import time
@@ -12,20 +13,34 @@ from typing import Dict, Any, List, Tuple
def get_random_user_agent(): def get_random_user_agent():
return f"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_{randrange(11, 15)}_{randrange(4, 9)}) AppleWebKit/{randrange(530, 537)}.{randrange(30, 37)} (KHTML, like Gecko) Chrome/{randrange(80, 105)}.0.{randrange(3000, 4500)}.{randrange(60, 125)} Safari/{randrange(530, 537)}.{randrange(30, 36)}" return f"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_{randrange(11, 15)}_{randrange(4, 9)}) AppleWebKit/{randrange(530, 537)}.{randrange(30, 37)} (KHTML, like Gecko) Chrome/{randrange(80, 105)}.0.{randrange(3000, 4500)}.{randrange(60, 125)} Safari/{randrange(530, 537)}.{randrange(30, 36)}"
# https://github.com/xyloflake/spot-secrets-go
def generate_totp(): def generate_totp():
url = "https://raw.githubusercontent.com/Thereallo1026/spotify-secrets/refs/heads/main/secrets/secretBytes.json" local_path = Path.home() / ".spotify-secret" / "secretBytes.json"
used_local = False
try: try:
url = "https://raw.githubusercontent.com/afkarxyz/secretBytes/refs/heads/main/secrets/secretBytes.json"
resp = requests.get(url, timeout=10) resp = requests.get(url, timeout=10)
if resp.status_code != 200: if resp.status_code != 200:
raise Exception(f"Failed to fetch TOTP secrets from GitHub. Status: {resp.status_code}") raise Exception(f"GitHub fetch failed with status: {resp.status_code}")
secrets_list = resp.json() secrets_list = resp.json()
except Exception as github_error:
try:
if local_path.exists():
with open(local_path, 'r') as f:
secrets_list = json.load(f)
used_local = True
else:
raise Exception(f"GitHub failed ({github_error}) and no local file found at {local_path}")
except Exception as local_error:
raise Exception(f"Failed to fetch secrets from both GitHub and local: {local_error}")
try:
latest_entry = max(secrets_list, key=lambda x: x["version"]) latest_entry = max(secrets_list, key=lambda x: x["version"])
version = latest_entry["version"] version = latest_entry["version"]
secret_cipher = latest_entry["secret"] secret_cipher = latest_entry["secret"]
except Exception as e: except Exception as e:
raise Exception(f"Failed to fetch secrets from GitHub: {str(e)}") raise Exception(f"Failed to process secrets: {str(e)}")
processed = [byte ^ ((i % 33) + 9) for i, byte in enumerate(secret_cipher)] processed = [byte ^ ((i % 33) + 9) for i, byte in enumerate(secret_cipher)]
processed_str = "".join(map(str, processed)) processed_str = "".join(map(str, processed))
@@ -57,8 +72,10 @@ token_url = 'https://open.spotify.com/api/token'
playlist_base_url = 'https://api.spotify.com/v1/playlists/{}' playlist_base_url = 'https://api.spotify.com/v1/playlists/{}'
album_base_url = 'https://api.spotify.com/v1/albums/{}' album_base_url = 'https://api.spotify.com/v1/albums/{}'
track_base_url = 'https://api.spotify.com/v1/tracks/{}' track_base_url = 'https://api.spotify.com/v1/tracks/{}'
artist_base_url = 'https://api.spotify.com/v1/artists/{}'
artist_albums_url = 'https://api.spotify.com/v1/artists/{}/albums'
headers = { 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', 'User-Agent': get_random_user_agent(),
'Accept': 'application/json', 'Accept': 'application/json',
'Accept-Language': 'en-US,en;q=0.9', 'Accept-Language': 'en-US,en;q=0.9',
'Accept-Encoding': 'gzip, deflate, br', 'Accept-Encoding': 'gzip, deflate, br',
@@ -97,11 +114,22 @@ def parse_uri(uri):
if parts[1] == "embed": if parts[1] == "embed":
parts = parts[1:] parts = parts[1:]
if len(parts) > 1 and parts[1].startswith("intl-"):
parts = parts[1:]
l = len(parts) l = len(parts)
if l == 3 and parts[1] in ["album", "track", "playlist"]: if l == 3 and parts[1] in ["album", "track", "playlist", "artist"]:
return {"type": parts[1], "id": parts[2]} return {"type": parts[1], "id": parts[2]}
if l == 5 and parts[3] == "playlist": if l == 5 and parts[3] == "playlist":
return {"type": parts[3], "id": parts[4]} return {"type": parts[3], "id": parts[4]}
if l >= 4 and parts[1] == "artist" and len(parts) >= 4:
if parts[3] == "discography":
discography_type = "all"
if len(parts) >= 5 and parts[4] in ["all", "album", "single", "compilation"]:
discography_type = parts[4]
return {"type": "artist_discography", "id": parts[2], "discography_type": discography_type}
else:
return {"type": "artist", "id": parts[2]}
raise SpotifyInvalidUrlException("ERROR: unable to determine Spotify URL type or type is unsupported.") raise SpotifyInvalidUrlException("ERROR: unable to determine Spotify URL type or type is unsupported.")
@@ -337,6 +365,69 @@ def get_raw_spotify_data(spotify_url, batch: bool = False, delay: float = 1.0):
except Exception as e: except Exception as e:
return {"error": f"Failed to get track data: {str(e)}"} return {"error": f"Failed to get track data: {str(e)}"}
elif url_info["type"] == "artist_discography":
try:
artist_data = get_json_from_api(
artist_base_url.format(url_info["id"]),
access_token
)
if not artist_data:
return {"error": "Failed to get artist data"}
discography_type = url_info.get("discography_type", "all")
if discography_type == "all":
include_groups = "album,single,compilation"
else:
include_groups = discography_type
albums = []
albums_url = f'{artist_albums_url.format(url_info["id"])}?include_groups={include_groups}&limit=50'
if batch:
albums, num_batches = fetch_tracks_in_batches(albums_url, access_token, 50, delay)
raw_data = {
"artist_info": artist_data,
"albums": albums,
"discography_type": discography_type,
"_batch_count": num_batches,
"_batch_enabled": True
}
else:
while albums_url:
album_data = get_json_from_api(albums_url, access_token)
if not album_data:
break
albums.extend(album_data['items'])
albums_url = album_data.get('next')
if albums_url and "&locale=" in albums_url:
albums_url = albums_url.split("&locale=")[0]
raw_data = {
"artist_info": artist_data,
"albums": albums,
"discography_type": discography_type,
"_batch_enabled": False
}
raw_data['_token'] = access_token
except Exception as e:
return {"error": f"Failed to get artist discography data: {str(e)}"}
elif url_info["type"] == "artist":
try:
artist_data = get_json_from_api(
artist_base_url.format(url_info["id"]),
access_token
)
if not artist_data:
return {"error": "Failed to get artist data"}
raw_data = artist_data
except Exception as e:
return {"error": f"Failed to get artist data: {str(e)}"}
return raw_data return raw_data
def format_track_data(track_data): def format_track_data(track_data):
@@ -466,6 +557,134 @@ def format_playlist_data(playlist_data):
"track_list": track_list "track_list": track_list
} }
def format_artist_discography_data(discography_data):
artist_info = discography_data.get('artist_info', {})
albums = discography_data.get('albums', [])
access_token = discography_data.get('_token', '')
artist_image = ''
if artist_info.get('images'):
artist_image = artist_info.get('images', [{}])[0].get('url', '')
formatted_artist_info = {
"name": artist_info.get('name', ''),
"followers": artist_info.get('followers', {}).get('total', 0),
"genres": artist_info.get('genres', []),
"images": artist_image,
"external_urls": artist_info.get('external_urls', {}).get('spotify', ''),
"discography_type": discography_data.get('discography_type', 'all'),
"total_albums": len(albums)
}
if discography_data.get('_batch_enabled', False):
formatted_artist_info["batch"] = f"{discography_data.get('_batch_count', 1)}"
album_list = []
all_tracks = []
for album in albums:
album_image = ''
if album.get('images'):
album_image = album.get('images', [{}])[0].get('url', '')
album_artists = []
for artist in album.get('artists', []):
album_artists.append(artist['name'])
album_info = {
"id": album.get('id', ''),
"name": album.get('name', ''),
"album_type": album.get('album_type', ''),
"release_date": album.get('release_date', ''),
"total_tracks": album.get('total_tracks', 0),
"artists": ", ".join(album_artists),
"images": album_image,
"external_urls": album.get('external_urls', {}).get('spotify', '')
}
album_list.append(album_info)
if access_token and album.get('id'):
try:
album_tracks_data = get_json_from_api(
f'{album_base_url.format(album.get("id"))}/tracks?limit=50',
access_token
)
if album_tracks_data:
tracks = []
tracks_url = f'{album_base_url.format(album.get("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]
for track in tracks:
track_artists = []
for artist in track.get('artists', []):
track_artists.append(artist['name'])
track_id = track.get('id', '')
track_isrc = ''
if track_id:
try:
full_track_data = get_json_from_api(
track_base_url.format(track_id),
access_token
)
if full_track_data:
track_isrc = full_track_data.get('external_ids', {}).get('isrc', '')
except:
pass
formatted_track = {
"artists": ", ".join(track_artists),
"name": track.get('name', ''),
"album_name": album.get('name', ''),
"album_type": album.get('album_type', ''),
"duration_ms": track.get('duration_ms', 0),
"images": album_image,
"release_date": album.get('release_date', ''),
"track_number": track.get('track_number', 0),
"external_urls": track.get('external_urls', {}).get('spotify', ''),
"isrc": track_isrc
}
all_tracks.append(formatted_track)
except Exception as e:
print(f"Error getting tracks for album {album.get('name', '')}: {str(e)}")
continue
return {
"artist_info": formatted_artist_info,
"album_list": album_list,
"track_list": all_tracks
}
def format_artist_data(artist_data):
artist_image = ''
if artist_data.get('images'):
artist_image = artist_data.get('images', [{}])[0].get('url', '')
return {
"artist": {
"name": artist_data.get('name', ''),
"followers": artist_data.get('followers', {}).get('total', 0),
"genres": artist_data.get('genres', []),
"images": artist_image,
"external_urls": artist_data.get('external_urls', {}).get('spotify', ''),
"popularity": artist_data.get('popularity', 0)
}
}
def process_spotify_data(raw_data, data_type): def process_spotify_data(raw_data, data_type):
if not raw_data or "error" in raw_data: if not raw_data or "error" in raw_data:
return {"error": "Invalid data provided"} return {"error": "Invalid data provided"}
@@ -477,6 +696,10 @@ def process_spotify_data(raw_data, data_type):
return format_album_data(raw_data) return format_album_data(raw_data)
elif data_type == "playlist": elif data_type == "playlist":
return format_playlist_data(raw_data) return format_playlist_data(raw_data)
elif data_type == "artist_discography":
return format_artist_discography_data(raw_data)
elif data_type == "artist":
return format_artist_data(raw_data)
else: else:
return {"error": "Invalid data type"} return {"error": "Invalid data type"}
except Exception as e: except Exception as e:
@@ -495,11 +718,23 @@ if __name__ == '__main__':
album = "https://open.spotify.com/album/6J84szYCnMfzEcvIcfWMFL" album = "https://open.spotify.com/album/6J84szYCnMfzEcvIcfWMFL"
song = "https://open.spotify.com/track/7so0lgd0zP2Sbgs2d7a1SZ" song = "https://open.spotify.com/track/7so0lgd0zP2Sbgs2d7a1SZ"
artist_discography_all = "https://open.spotify.com/artist/0du5cEVh5yTK9QJze8zA0C/discography/all"
artist_discography_albums = "https://open.spotify.com/artist/0du5cEVh5yTK9QJze8zA0C/discography/album"
artist_discography_singles = "https://open.spotify.com/artist/0du5cEVh5yTK9QJze8zA0C/discography/single"
artist_discography_compilations = "https://open.spotify.com/artist/0du5cEVh5yTK9QJze8zA0C/discography/compilation"
print("=== Testing Artist Discography (All) ===")
filtered_discography = get_filtered_data(artist_discography_all, batch=True, delay=0.1)
print(json.dumps(filtered_discography, indent=2))
print("\n=== Testing Playlist ===")
filtered_playlist = get_filtered_data(playlist, batch=True, delay=0.1) filtered_playlist = get_filtered_data(playlist, batch=True, delay=0.1)
print(json.dumps(filtered_playlist, indent=2)) print(json.dumps(filtered_playlist, indent=2))
print("\n=== Testing Album ===")
filtered_album = get_filtered_data(album) filtered_album = get_filtered_data(album)
print(json.dumps(filtered_album, indent=2)) print(json.dumps(filtered_album, indent=2))
print("\n=== Testing Track ===")
filtered_track = get_filtered_data(song) filtered_track = get_filtered_data(song)
print(json.dumps(filtered_track, indent=2)) print(json.dumps(filtered_track, indent=2))
+111
View File
@@ -0,0 +1,111 @@
import json
import time
from pathlib import Path
from DrissionPage import ChromiumPage, ChromiumOptions
def summarise(caps):
real = {}
for cap in caps:
sec = cap.get("secret")
if not sec or not isinstance(sec, str):
continue
ver = cap.get("version") or cap.get("obj", {}).get("version")
if ver and ver != 0:
real[str(int(ver))] = sec
if not real:
return False, "No secrets found."
versions = sorted(int(k) for k in real.keys())
secret_bytes = [
{"version": v, "secret": [ord(c) for c in real[str(v)]]}
for v in versions
]
secrets_dir = Path.home() / ".spotify-secret"
secrets_dir.mkdir(exist_ok=True)
output_file = secrets_dir / "secretBytes.json"
with open(output_file, "w") as f:
json.dump(secret_bytes, f, indent=2)
return True, f"Saved to: {output_file}"
def grab_live(progress_callback=None):
"""
Grab secrets from Spotify web player
Args:
progress_callback: Optional callback function to report progress
Returns:
list: Captured secrets
"""
def emit_progress(msg):
if progress_callback:
progress_callback(msg)
else:
print(msg)
stealth = """(()=>{
Object.defineProperty(navigator,'webdriver',{get:()=>false});
Object.defineProperty(navigator,'languages',{get:()=>['en-US','en']});
Object.defineProperty(navigator,'plugins',{get:()=>[1,2,3,4,5]});
window.chrome={runtime:{}};
const q=navigator.permissions.query;
navigator.permissions.query=p=>p.name==='notifications'?Promise.resolve({state:Notification.permission}):q(p);
const g=WebGLRenderingContext.prototype.getParameter;
WebGLRenderingContext.prototype.getParameter=function(p){
if(p===37445)return'Intel Inc.';if(p===37446)return'Intel Iris OpenGL Engine';return g.call(this,p);
};
})();"""
hook = """(()=>{if(globalThis.__secretHookInstalled)return;
globalThis.__secretHookInstalled=true;globalThis.__captures=[];
Object.defineProperty(Object.prototype,'secret',{configurable:true,set:function(v){
try{__captures.push({secret:v,version:this.version,obj:this});}catch(e){}
Object.defineProperty(this,'secret',{value:v,writable:true,configurable:true,enumerable:true});}});
})();"""
co = ChromiumOptions()
co.headless(True)
co.set_argument('--disable-blink-features=AutomationControlled')
co.set_argument('--no-sandbox')
page = ChromiumPage(addr_or_opts=co)
try:
page.run_cdp('Page.addScriptToEvaluateOnNewDocument', source=stealth)
page.run_cdp('Page.addScriptToEvaluateOnNewDocument', source=hook)
emit_progress("Opening Spotify...")
page.get("https://open.spotify.com")
time.sleep(3)
caps = page.run_js("return globalThis.__captures || []")
for c in caps:
if isinstance(c, dict) and c.get("secret") and c.get("version"):
emit_progress(f"Secret({int(c['version'])}): {c['secret']}")
return caps or []
finally:
page.quit()
def scrape_and_save(progress_callback=None):
"""
Main function to scrape secrets and save to file
Args:
progress_callback: Optional callback function to report progress
Returns:
tuple: (success: bool, message: str)
"""
try:
caps = grab_live(progress_callback)
return summarise(caps)
except Exception as e:
return False, f"Error: {str(e)}"
def main():
success, message = scrape_and_save()
print(message)
return 0 if success else 1
if __name__ == "__main__":
import sys
sys.exit(main())
+5
View File
@@ -0,0 +1,5 @@
<svg xmlns="http://www.w3.org/2000/svg" width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round">
<circle cx="12" cy="12" r="10"/>
<line x1="15" y1="9" x2="9" y2="15"/>
<line x1="9" y1="9" x2="15" y2="15"/>
</svg>

After

Width:  |  Height:  |  Size: 304 B

BIN
View File
Binary file not shown.

After

Width:  |  Height:  |  Size: 6.4 KiB

+5
View File
@@ -0,0 +1,5 @@
<svg xmlns="http://www.w3.org/2000/svg" width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round">
<path d="M21 15v4a2 2 0 0 1-2 2H5a2 2 0 0 1-2-2v-4"/>
<polyline points="7 10 12 15 17 10"/>
<line x1="12" y1="15" x2="12" y2="3"/>
</svg>

After

Width:  |  Height:  |  Size: 326 B

BIN
View File
Binary file not shown.

After

Width:  |  Height:  |  Size: 169 KiB

+48
View File
@@ -0,0 +1,48 @@
<?xml version="1.0" encoding="utf-8"?>
<!-- Generator: Adobe Illustrator 28.0.0, SVG Export Plug-In . SVG Version: 6.00 Build 0) -->
<svg version="1.1" id="Layer_1" xmlns:sketch="http://www.bohemiancoding.com/sketch/ns"
xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" x="0px" y="0px" viewBox="0 0 512 512"
style="enable-background:new 0 0 512 512;" xml:space="preserve">
<style type="text/css">
.st0{fill-rule:evenodd;clip-rule:evenodd;fill:#FF0000;stroke:#373435;stroke-width:0.5669;stroke-miterlimit:10;}
.st1{fill-rule:evenodd;clip-rule:evenodd;fill:#FFFF00;stroke:#373435;stroke-width:0.5669;stroke-miterlimit:10;}
.st2{fill-rule:evenodd;clip-rule:evenodd;fill:#2AA125;stroke:#373435;stroke-width:0.5669;stroke-miterlimit:10;}
</style>
<g id="SVGRepo_bgCarrier">
</g>
<g id="SVGRepo_tracerCarrier">
</g>
<g id="Layer_x0020_1">
<g id="_1818452274576">
<g id="SVGRepo_bgCarrier_00000044893734704698182460000014884511992085247122_">
</g>
<g id="SVGRepo_tracerCarrier_00000067939915892718314930000001743019108017086612_">
</g>
<g id="SVGRepo_iconCarrier_00000176000695080737548300000008661679408724292005_">
<path d="M407.1,227.2c-54.7-27.6-119.3-43.8-187.6-43.8c-38.9,0-76.5,5.2-112.3,15l3-0.7c-2.1,0.7-4.5,1.1-7,1.1
c-13,0-23.5-10.5-23.5-23.5c0-10.5,6.8-19.3,16.3-22.4l0.2-0.1c90.9-26.9,240.6-21.8,335.4,34.6c7.3,4.4,12.1,12.3,12.1,21.3
c0,4.4-1.2,8.6-3.2,12.2l0.1-0.1c-5,5.9-12.3,9.6-20.6,9.6c-4.7,0-9-1.2-12.8-3.3L407.1,227.2L407.1,227.2z M404.5,298.9
c-3.4,5.7-9.6,9.4-16.6,9.4c-3.8,0-7.4-1.1-10.4-3l0.1,0.1c-46.8-26.8-102.8-42.5-162.5-42.5c-32.9,0-64.6,4.8-94.6,13.7l2.3-0.6
c-1.7,0.5-3.7,0.9-5.8,0.9c-10.7,0-19.4-8.7-19.4-19.4c0-8.7,5.7-16,13.5-18.5l0.1,0c30.8-9.1,66.2-14.4,102.8-14.4
c68.1,0,132,18.2,187,49.9l-1.8-1c5.1,3.3,8.4,8.9,8.4,15.3C407.7,292.5,406.5,296,404.5,298.9L404.5,298.9L404.5,298.9
L404.5,298.9z M373.8,369.3c-2.7,4.6-7.6,7.7-13.3,7.7c-3.2,0-6.1-1-8.6-2.6l0.1,0c-40.9-23-89.7-36.5-141.7-36.5
c-29.8,0-58.6,4.5-85.7,12.7l2.1-0.5c-1.1,0.3-2.5,0.5-3.8,0.5c-8.7,0-15.8-7.1-15.8-15.8c0-7.4,5.1-13.6,11.9-15.3l0.1,0
c27-8,58-12.6,90.1-12.6c58.1,0,112.6,15.1,159.9,41.7l-1.7-0.9c5.2,2.6,8.6,7.8,8.6,13.8C376,364.3,375.2,367.1,373.8,369.3
L373.8,369.3L373.8,369.3L373.8,369.3z M256-0.6L256-0.6C114.6-0.6,0,114,0,255.4s114.6,256,256,256s256-114.6,256-256l0,0
C511.6,114.2,397.2-0.2,256-0.6L256-0.6L256-0.6L256-0.6z"/>
</g>
</g>
<path class="st0" d="M406.9,227.2c0,0,0.1,0,0.1,0.1L406.9,227.2z M107.1,198.5c35.8-9.8,73.4-15,112.3-15
c68.3,0,132.8,16.1,187.5,43.7c3.8,2.1,8.2,3.3,12.8,3.3c8.2,0,15.6-3.7,20.5-9.6c0,0,0,0,0,0c2-3.6,3.1-7.7,3.1-12
c0-9-4.8-16.8-12.1-21.3c-94.7-56.4-244.5-61.5-335.4-34.6l-0.2,0.1c-9.4,3.1-16.3,11.9-16.3,22.4c0,13,10.5,23.5,23.5,23.5
c2.5,0,4.9-0.4,7-1.1L107.1,198.5L107.1,198.5z"/>
<path class="st1" d="M401.2,274.3c-55-31.8-118.9-49.9-187-49.9c-36.6,0-72,5.3-102.8,14.4l-0.1,0c-7.9,2.5-13.5,9.9-13.5,18.5
c0,10.7,8.7,19.4,19.4,19.4c1.3,0,2.5-0.1,3.6-0.3c29.9-8.8,61.5-13.6,94.3-13.6c59.7,0,115.7,15.7,162.4,42.5l0.1,0.1
c3,1.9,6.5,3,10.3,3c7,0,13.1-3.7,16.6-9.4l0,0c2-2.9,3.2-6.5,3.2-10.3c0-6.4-3.3-12-8.4-15.3L401.2,274.3L401.2,274.3z"/>
<path class="st2" d="M352,374.4C352,374.4,352,374.4,352,374.4L352,374.4z M373.8,369.3C373.8,369.3,373.7,369.4,373.8,369.3
L373.8,369.3L373.8,369.3z M367.8,347.8l-0.4-0.2C367.5,347.6,367.6,347.7,367.8,347.8z M369,348.4
c-47.3-26.5-101.8-41.7-159.9-41.7c-32.1,0-63.1,4.6-90.1,12.6l-0.1,0c-6.8,1.8-11.9,8-11.9,15.3c0,8.7,7.1,15.8,15.8,15.8
c1,0,1.9-0.1,2.8-0.3c26.8-8.1,55.2-12.4,84.6-12.4c52,0,100.8,13.5,141.7,36.5l0,0c2.4,1.6,5.4,2.6,8.5,2.6
c5.7,0,10.6-3.1,13.3-7.7h0c1.4-2.3,2.2-5,2.2-7.9c0-5.9-3.3-11-8.2-13.6L369,348.4L369,348.4z"/>
</g>
</svg>

After

Width:  |  Height:  |  Size: 3.7 KiB

BIN
View File
Binary file not shown.

After

Width:  |  Height:  |  Size: 6.1 KiB

+3
View File
@@ -0,0 +1,3 @@
<svg xmlns="http://www.w3.org/2000/svg" width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round">
<path d="M14.7 6.3a1 1 0 0 0 0 1.4l1.6 1.6a1 1 0 0 0 1.4 0l3.77-3.77a6 6 0 0 1-7.94 7.94l-6.91 6.91a2.12 2.12 0 0 1-3-3l6.91-6.91a6 6 0 0 1 7.94-7.94l-3.76 3.76z"/>
</svg>

After

Width:  |  Height:  |  Size: 356 B

+6
View File
@@ -0,0 +1,6 @@
<svg xmlns="http://www.w3.org/2000/svg" width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round">
<polyline points="3 6 5 6 21 6"/>
<path d="M19 6v14a2 2 0 0 1-2 2H7a2 2 0 0 1-2-2V6m3 0V4a2 2 0 0 1 2-2h4a2 2 0 0 1 2 2v2"/>
<line x1="10" y1="11" x2="10" y2="17"/>
<line x1="14" y1="11" x2="14" y2="17"/>
</svg>

After

Width:  |  Height:  |  Size: 402 B

-271
View File
@@ -1,271 +0,0 @@
import requests
import time
import os
import re
from datetime import datetime
from mutagen.flac import FLAC, Picture
from mutagen.id3 import PictureType
class ProgressCallback:
def __call__(self, current, total):
if total > 0:
percent = (current / total) * 100
print(f"\r{percent:.2f}% ({current}/{total})", end="")
else:
print(f"\r{current / (1024 * 1024):.2f} MB", end="")
class QobuzDownloader:
def __init__(self, region="us", timeout=30):
if region not in ["eu", "us"]:
raise ValueError("Region must be either 'us' or 'eu'")
self.region = region
self.timeout = timeout
self.session = 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.base_api_url = f"https://{region}.qobuz.squid.wtf/api"
self.download_chunk_size = 256 * 1024
self.progress_callback = ProgressCallback()
def set_progress_callback(self, callback):
self.progress_callback = callback
def sanitize_filename(self, filename):
if not filename:
return "Unknown Track"
sanitized = re.sub(r'[\\/*?:"<>|]', "", str(filename))
return re.sub(r'\s+', ' ', sanitized).strip() or "Unnamed Track"
def get_track_info(self, isrc):
print(f"Fetching: {isrc}")
search_url = f"{self.base_api_url}/get-music"
params = {'q': isrc, 'offset': 0, 'limit': 10}
try:
response = self.session.get(search_url, params=params, timeout=self.timeout)
response.raise_for_status()
data = response.json()
selected_track = None
if data and data.get("success"):
items = data.get("data", {}).get("tracks", {}).get("items", [])
priority = {24: 1, 16: 2}
for track in items:
if track.get("isrc") == isrc:
current_prio = priority.get(track.get("maximum_bit_depth"), 3)
if selected_track is None or current_prio < priority.get(selected_track.get("maximum_bit_depth"), 3):
selected_track = track
if current_prio == 1:
break
if not selected_track:
raise Exception(f"Track not found: {isrc}")
title = selected_track.get('title', 'Unknown')
bit_depth = selected_track.get('maximum_bit_depth', 'Unknown')
print(f"Found: {title} ({bit_depth}b)")
return selected_track
except requests.exceptions.RequestException as e:
raise Exception(f"Request error: {e}")
except Exception as e:
raise Exception(f"Error: {e}")
def get_download_url(self, track_id):
print("Fetching URL...")
download_api_url = f"{self.base_api_url}/download-music"
params = {'track_id': track_id, 'quality': 27}
try:
response = self.session.get(download_api_url, params=params, timeout=self.timeout)
response.raise_for_status()
data = response.json()
if data and data.get("success") and data.get("data", {}).get("url"):
download_url = data["data"]["url"]
print("URL found")
return download_url
else:
error_msg = data.get('error', {}).get('message', 'Unknown API error')
raise Exception(f"API error: {error_msg}")
except requests.exceptions.RequestException as e:
raise Exception(f"Request error: {e}")
except Exception as e:
raise Exception(f"Error: {e}")
def download(self, isrc, output_dir=".", is_paused_callback=None, is_stopped_callback=None):
if output_dir != ".":
try:
os.makedirs(output_dir, exist_ok=True)
except OSError as e:
raise Exception(f"Directory error: {e}")
track_info = self.get_track_info(isrc)
track_id = track_info.get("id")
if not track_id:
raise Exception("No track ID found")
artist_name = self.sanitize_filename(track_info.get('performer', {}).get('name'))
track_title = self.sanitize_filename(track_info.get('title'))
output_filename = os.path.join(output_dir, f"{artist_name} - {track_title}.flac")
if os.path.exists(output_filename):
file_size = os.path.getsize(output_filename)
if file_size > 0:
print(f"File already exists: {output_filename} ({file_size / (1024 * 1024):.2f} MB)")
return output_filename
download_url = self.get_download_url(track_id)
temp_filename = output_filename + ".part"
print(f"Downloading...")
try:
with self.session.get(download_url, stream=True, timeout=900) as response, \
open(temp_filename, 'wb') as f:
response.raise_for_status()
total_size = int(response.headers.get('content-length', 0))
downloaded_size = 0
start_time = time.time()
last_update_time = start_time
for chunk in response.iter_content(chunk_size=self.download_chunk_size):
if is_stopped_callback and is_stopped_callback():
f.close()
if os.path.exists(temp_filename):
os.remove(temp_filename)
raise Exception("Download stopped")
while is_paused_callback and is_paused_callback():
time.sleep(0.1)
if is_stopped_callback and is_stopped_callback():
f.close()
if os.path.exists(temp_filename):
os.remove(temp_filename)
raise Exception("Download stopped")
f.write(chunk)
downloaded_size += len(chunk)
current_time = time.time()
if current_time - last_update_time >= 1:
if total_size > 0:
progress_percent = (downloaded_size / total_size) * 100
elapsed_time = current_time - start_time
speed = downloaded_size / (1024 * 1024 * elapsed_time) if elapsed_time > 0 else 0
print(f"{progress_percent:.2f}% - {speed:.2f} MB/s")
else:
print(f"{downloaded_size / (1024 * 1024):.2f} MB")
last_update_time = current_time
if self.progress_callback:
self.progress_callback(downloaded_size, total_size)
os.rename(temp_filename, output_filename)
print("Download complete")
except requests.exceptions.RequestException as e:
if os.path.exists(temp_filename):
os.remove(temp_filename)
raise Exception(f"Download failed: {e}")
except Exception as e:
if os.path.exists(temp_filename):
os.remove(temp_filename)
raise Exception(f"File error: {e}")
print("Adding metadata...")
try:
self._embed_metadata(output_filename, track_info)
print("Metadata saved")
except Exception as e:
print(f"Tagging failed: {e}")
print(f"Done")
return output_filename
def _embed_metadata(self, filename, track_info):
try:
audio = FLAC(filename)
audio.delete()
audio.clear_pictures()
album_info = track_info.get('album', {})
artist = track_info.get('performer', {}).get('name')
if track_info.get('title'):
audio['TITLE'] = track_info['title']
if artist:
audio['ARTIST'] = artist
if album_info.get('title'):
audio['ALBUM'] = album_info['title']
if album_info.get('artist', {}).get('name', artist):
audio['ALBUMARTIST'] = album_info.get('artist', {}).get('name', artist)
if track_info.get('track_number'):
audio['TRACKNUMBER'] = str(track_info['track_number'])
if track_info.get('release_date_original'):
audio['DATE'] = track_info['release_date_original']
try:
audio['YEAR'] = str(datetime.strptime(track_info['release_date_original'], '%Y-%m-%d').year)
except ValueError:
pass
if album_info.get('genre', {}).get('name'):
audio['GENRE'] = album_info['genre']['name']
if track_info.get('copyright'):
audio['COPYRIGHT'] = track_info['copyright']
if track_info.get('isrc'):
audio['ISRC'] = track_info['isrc']
if album_info.get('label', {}).get('name'):
audio['ORGANIZATION'] = album_info['label']['name']
img_info = album_info.get('image', {})
cover_url = img_info.get('large') or img_info.get('small') or img_info.get('thumbnail')
if cover_url:
try:
img_response = self.session.get(cover_url, timeout=30)
img_response.raise_for_status()
mime_type = img_response.headers.get('Content-Type', 'image/jpeg').lower()
if mime_type in ['image/jpeg', 'image/png']:
picture = Picture()
picture.data = img_response.content
picture.type = PictureType.COVER_FRONT
picture.mime = mime_type
audio.add_picture(picture)
print("Cover added")
except Exception as e:
print(f"Cover error: {str(e)}")
audio.save()
except Exception as e:
raise Exception(f"Metadata error: {e}")
def main():
print("=== QobuzDL - Qobuz Downloader ===")
downloader = QobuzDownloader(region="us")
isrc = "USAT22409172"
output_dir = "."
try:
downloaded_file = downloader.download(isrc, output_dir)
print(f"Success: File saved as {downloaded_file}")
except Exception as e:
print(f"Error: {str(e)}")
if __name__ == "__main__":
try:
import sys
if sys.platform == "win32":
import os
os.system("chcp 65001 > nul")
try:
sys.stdout.reconfigure(encoding='utf-8')
except:
pass
except:
pass
main()
+9
View File
@@ -0,0 +1,9 @@
PyQt6
pyqt6-tools
pyqtdarktheme
requests
mutagen
pyotp
packaging
pyinstaller
DrissionPage
+96 -149
View File
@@ -2,9 +2,8 @@ import asyncio
import json import json
import os import os
import re import re
import tempfile
import time import time
import httpx import requests
from mutagen.flac import FLAC, Picture from mutagen.flac import FLAC, Picture
from mutagen.id3 import PictureType from mutagen.id3 import PictureType
@@ -24,22 +23,11 @@ class TidalDownloader:
self.progress_callback = ProgressCallback() self.progress_callback = ProgressCallback()
self.client_id = "zU4XHVVkc2tDPo4t" self.client_id = "zU4XHVVkc2tDPo4t"
self.client_secret = "VJKhDFqJPqvsPVNBV6ukXTJmwlvbttP7wlMlrc72se4=" self.client_secret = "VJKhDFqJPqvsPVNBV6ukXTJmwlvbttP7wlMlrc72se4="
self.temp_dir = tempfile.gettempdir()
self.token_path = os.path.join(self.temp_dir, "tidal_token.json")
self.access_token = None
self._load_token()
def set_progress_callback(self, callback): def set_progress_callback(self, callback):
self.progress_callback = callback self.progress_callback = callback
def _load_token(self):
if os.path.exists(self.token_path):
try:
with open(self.token_path, "r") as tok:
token = json.loads(tok.read())
self.access_token = token.get("access_token")
except:
pass
def sanitize_filename(self, filename): def sanitize_filename(self, filename):
if not filename: if not filename:
@@ -47,10 +35,7 @@ class TidalDownloader:
sanitized = re.sub(r'[\\/*?:"<>|]', "", str(filename)) sanitized = re.sub(r'[\\/*?:"<>|]', "", str(filename))
return re.sub(r'\s+', ' ', sanitized).strip() or "Unnamed Track" return re.sub(r'\s+', ' ', sanitized).strip() or "Unnamed Track"
async def get_access_token(self): def get_access_token(self):
if self.access_token:
return self.access_token
refresh_url = "https://auth.tidal.com/v1/oauth2/token" refresh_url = "https://auth.tidal.com/v1/oauth2/token"
payload = { payload = {
@@ -58,79 +43,67 @@ class TidalDownloader:
"grant_type": "client_credentials", "grant_type": "client_credentials",
} }
async with httpx.AsyncClient(http2=True) as client: try:
try: response = requests.post(
response = await client.post( url=refresh_url,
url=refresh_url, data=payload,
data=payload, auth=(self.client_id, self.client_secret),
auth=(self.client_id, self.client_secret), timeout=self.timeout
) )
if response.status_code == 200: if response.status_code == 200:
token_data = response.json() token_data = response.json()
new_token = token_data.get("access_token") return token_data.get("access_token")
else:
try:
with open(self.token_path, "w") as f:
json.dump({
"access_token": new_token
}, f)
except:
pass
self.access_token = new_token
return new_token
else:
return None
except:
return None return None
async def search_tracks(self, query): except:
return None
def search_tracks(self, query):
try: try:
tidal_token = await self.get_access_token() tidal_token = self.get_access_token()
if not tidal_token: if not tidal_token:
raise Exception("Failed to get access token") raise Exception("Failed to get access token")
search_url = f"https://api.tidal.com/v1/search/tracks?query={query}&limit=25&offset=0&countryCode=US" search_url = f"https://api.tidal.com/v1/search/tracks?query={query}&limit=25&offset=0&countryCode=US"
header = {"authorization": f"Bearer {tidal_token}"} header = {"authorization": f"Bearer {tidal_token}"}
async with httpx.AsyncClient(http2=True) as client: search_data = requests.get(url=search_url, headers=header, timeout=self.timeout)
search_data = await client.get(url=search_url, headers=header) response_data = search_data.json()
response_data = search_data.json()
filtered_items = [{ filtered_items = [{
"id": item.get("id"), "id": item.get("id"),
"title": item.get("title"), "title": item.get("title"),
"url": item.get("url"), "url": item.get("url"),
"isrc": item.get("isrc"), "isrc": item.get("isrc"),
"audioQuality": item.get("audioQuality"), "audioQuality": item.get("audioQuality"),
"mediaMetadata": item.get("mediaMetadata"), "mediaMetadata": item.get("mediaMetadata"),
"album": item.get("album", {}), "album": item.get("album", {}),
"artists": item.get("artists", []), "artists": item.get("artists", []),
"artist": item.get("artist", {}), "artist": item.get("artist", {}),
"trackNumber": item.get("trackNumber"), "trackNumber": item.get("trackNumber"),
"volumeNumber": item.get("volumeNumber"), "volumeNumber": item.get("volumeNumber"),
"duration": item.get("duration"), "duration": item.get("duration"),
"copyright": item.get("copyright"), "copyright": item.get("copyright"),
"explicit": item.get("explicit") "explicit": item.get("explicit")
} for item in response_data.get("items", [])] } for item in response_data.get("items", [])]
return { return {
"limit": response_data.get("limit"), "limit": response_data.get("limit"),
"offset": response_data.get("offset"), "offset": response_data.get("offset"),
"totalNumberOfItems": response_data.get("totalNumberOfItems"), "totalNumberOfItems": response_data.get("totalNumberOfItems"),
"items": filtered_items "items": filtered_items
} }
except Exception as e: except Exception as e:
raise Exception(f"Search error: {str(e)}") raise Exception(f"Search error: {str(e)}")
async def get_track_info(self, query, isrc=None): def get_track_info(self, query, isrc=None):
print(f"Fetching: {query}" + (f" (ISRC: {isrc})" if isrc else "")) print(f"Fetching: {query}" + (f" (ISRC: {isrc})" if isrc else ""))
try: try:
result = await self.search_tracks(query) result = self.search_tracks(query)
if not result or not result.get("items"): if not result or not result.get("items"):
raise Exception(f"No tracks found for query: {query}") raise Exception(f"No tracks found for query: {query}")
@@ -169,98 +142,72 @@ class TidalDownloader:
except Exception as e: except Exception as e:
raise Exception(f"Error getting track info: {str(e)}") raise Exception(f"Error getting track info: {str(e)}")
async def get_download_url(self, track_id, quality="LOSSLESS"): def get_download_url(self, track_id, quality="LOSSLESS"):
print("Fetching URL...") print("Fetching URL...")
download_api_url = f"https://hifi.401658.xyz/track/?id={track_id}&quality={quality}" download_api_url = f"https://tidal.401658.xyz/track/?id={track_id}&quality={quality}"
async with httpx.AsyncClient(http2=True, timeout=self.timeout) as client: try:
try: response = requests.get(download_api_url, timeout=self.timeout)
response = await client.get(download_api_url)
if response.status_code == 200: if response.status_code == 200:
data = response.json() data = response.json()
for item in data: for item in data:
if "OriginalTrackUrl" in item: if "OriginalTrackUrl" in item:
print("URL found") print("URL found")
return { return {
"download_url": item["OriginalTrackUrl"], "download_url": item["OriginalTrackUrl"],
"track_info": data[0] if data else {} "track_info": data[0] if data else {}
} }
raise Exception("Download URL not found in response") raise Exception("Download URL not found in response")
else: else:
raise Exception(f"API returned status code: {response.status_code}") raise Exception(f"API returned status code: {response.status_code}")
except Exception as e: except Exception as e:
raise Exception(f"Error getting download URL: {str(e)}") raise Exception(f"Error getting download URL: {str(e)}")
async def download_album_art(self, album_id, size="1280x1280"): def download_album_art(self, album_id, size="1280x1280"):
try: try:
art_url = f"https://resources.tidal.com/images/{album_id.replace('-', '/')}/{size}.jpg" art_url = f"https://resources.tidal.com/images/{album_id.replace('-', '/')}/{size}.jpg"
async with httpx.AsyncClient(http2=True, timeout=self.timeout) as client: response = requests.get(art_url, timeout=self.timeout)
response = await client.get(art_url)
if response.status_code == 200: if response.status_code == 200:
return response.content return response.content
else: else:
print(f"Failed to download album art: HTTP {response.status_code}") print(f"Failed to download album art: HTTP {response.status_code}")
return None return None
except Exception as e: except Exception as e:
print(f"Error downloading album art: {str(e)}") print(f"Error downloading album art: {str(e)}")
return None return None
async def download_file(self, url, filepath, is_paused_callback=None, is_stopped_callback=None): def download_file(self, url, filepath, is_paused_callback=None, is_stopped_callback=None):
temp_filepath = filepath + ".part" temp_filepath = filepath + ".part"
retry_count = 0 retry_count = 0
while retry_count <= self.max_retries: while retry_count <= self.max_retries:
try: try:
async with httpx.AsyncClient(http2=True, timeout=60.0) as client: response = requests.get(url, timeout=60.0)
async with client.stream('GET', url) as response: if response.status_code != 200:
if response.status_code != 200: raise Exception(f"HTTP {response.status_code}")
raise Exception(f"HTTP {response.status_code}")
total_size = int(response.headers.get('content-length', 0)) if is_stopped_callback and is_stopped_callback():
downloaded_size = 0 raise Exception("Download stopped")
start_time = time.time()
last_update_time = start_time
with open(temp_filepath, 'wb') as f: while is_paused_callback and is_paused_callback():
async for chunk in response.aiter_bytes(chunk_size=self.download_chunk_size): time.sleep(0.1)
if is_stopped_callback and is_stopped_callback(): if is_stopped_callback and is_stopped_callback():
f.close() raise Exception("Download stopped")
if os.path.exists(temp_filepath):
os.remove(temp_filepath)
raise Exception("Download stopped")
while is_paused_callback and is_paused_callback(): with open(temp_filepath, 'wb') as f:
await asyncio.sleep(0.1) f.write(response.content)
if is_stopped_callback and is_stopped_callback():
f.close()
if os.path.exists(temp_filepath):
os.remove(temp_filepath)
raise Exception("Download stopped")
f.write(chunk) downloaded_size = len(response.content)
downloaded_size += len(chunk)
current_time = time.time() if self.progress_callback:
if current_time - last_update_time >= 1: self.progress_callback(downloaded_size, downloaded_size)
if total_size > 0:
progress_percent = (downloaded_size / total_size) * 100
elapsed_time = current_time - start_time
speed = downloaded_size / (1024 * 1024 * elapsed_time) if elapsed_time > 0 else 0
print(f"{progress_percent:.2f}% - {speed:.2f} MB/s")
else:
print(f"{downloaded_size / (1024 * 1024):.2f} MB")
last_update_time = current_time
if self.progress_callback:
self.progress_callback(downloaded_size, total_size)
os.rename(temp_filepath, filepath) os.rename(temp_filepath, filepath)
print("Download complete") print("Download complete")
@@ -278,9 +225,9 @@ class TidalDownloader:
print(f"Download error (attempt {retry_count}/{self.max_retries}): {str(e)}") print(f"Download error (attempt {retry_count}/{self.max_retries}): {str(e)}")
print(f"Retrying in {retry_count * 2} seconds...") print(f"Retrying in {retry_count * 2} seconds...")
await asyncio.sleep(retry_count * 2) time.sleep(retry_count * 2)
async def embed_metadata(self, filepath, track_info, search_info=None): def embed_metadata(self, filepath, track_info, search_info=None):
try: try:
print("Embedding metadata...") print("Embedding metadata...")
audio = FLAC(filepath) audio = FLAC(filepath)
@@ -351,7 +298,7 @@ class TidalDownloader:
audio["COMMENT"] = f"Tidal {track_info['audioQuality']}" audio["COMMENT"] = f"Tidal {track_info['audioQuality']}"
if album_info.get("cover"): if album_info.get("cover"):
album_art = await self.download_album_art(album_info["cover"]) album_art = self.download_album_art(album_info["cover"])
if album_art: if album_art:
picture = Picture() picture = Picture()
picture.data = album_art picture.data = album_art
@@ -369,14 +316,14 @@ class TidalDownloader:
print(f"Error embedding metadata: {str(e)}") print(f"Error embedding metadata: {str(e)}")
return False return False
async def download(self, query, isrc=None, output_dir=".", quality="LOSSLESS", is_paused_callback=None, is_stopped_callback=None): def download(self, query, isrc=None, output_dir=".", quality="LOSSLESS", is_paused_callback=None, is_stopped_callback=None):
if output_dir != ".": if output_dir != ".":
try: try:
os.makedirs(output_dir, exist_ok=True) os.makedirs(output_dir, exist_ok=True)
except OSError as e: except OSError as e:
raise Exception(f"Directory error: {e}") raise Exception(f"Directory error: {e}")
track_info = await self.get_track_info(query, isrc) track_info = self.get_track_info(query, isrc)
track_id = track_info.get("id") track_id = track_info.get("id")
if not track_id: if not track_id:
@@ -402,12 +349,12 @@ class TidalDownloader:
print(f"File already exists: {output_filename} ({file_size / (1024 * 1024):.2f} MB)") print(f"File already exists: {output_filename} ({file_size / (1024 * 1024):.2f} MB)")
return output_filename return output_filename
download_info = await self.get_download_url(track_id, quality) download_info = self.get_download_url(track_id, quality)
download_url = download_info["download_url"] download_url = download_info["download_url"]
download_track_info = download_info["track_info"] download_track_info = download_info["track_info"]
print(f"Downloading to: {output_filename}") print(f"Downloading to: {output_filename}")
await self.download_file( self.download_file(
download_url, download_url,
output_filename, output_filename,
is_paused_callback=is_paused_callback, is_paused_callback=is_paused_callback,
@@ -416,7 +363,7 @@ class TidalDownloader:
print("Adding metadata...") print("Adding metadata...")
try: try:
await self.embed_metadata(output_filename, download_track_info, track_info) self.embed_metadata(output_filename, download_track_info, track_info)
print("Metadata saved") print("Metadata saved")
except Exception as e: except Exception as e:
print(f"Tagging failed: {e}") print(f"Tagging failed: {e}")
@@ -424,7 +371,7 @@ class TidalDownloader:
print("Done") print("Done")
return output_filename return output_filename
async def main(): def main():
print("=== TidalDL - Tidal Downloader ===") print("=== TidalDL - Tidal Downloader ===")
downloader = TidalDownloader(timeout=30, max_retries=3) downloader = TidalDownloader(timeout=30, max_retries=3)
@@ -433,7 +380,7 @@ async def main():
output_dir = "." output_dir = "."
try: try:
downloaded_file = await downloader.download(query, isrc, output_dir) downloaded_file = downloader.download(query, isrc, output_dir)
print(f"Success: File saved as {downloaded_file}") print(f"Success: File saved as {downloaded_file}")
except Exception as e: except Exception as e:
print(f"Error: {str(e)}") print(f"Error: {str(e)}")
@@ -451,4 +398,4 @@ if __name__ == "__main__":
except: except:
pass pass
asyncio.run(main()) main()
+1 -1
View File
@@ -1,3 +1,3 @@
{ {
"version": "3.7" "version": "4.8"
} }