Compare commits

...

43 Commits

Author SHA1 Message Date
afkarxyz a620c16b1c v5.3 2025-10-22 05:13:44 +07:00
afkarxyz cf27ae098d v5.2 2025-10-21 03:26:31 +07:00
afkarxyz a0c60a473a . 2025-10-21 03:26:01 +07:00
afkarxyz 25c5a4d175 v5.2 2025-10-21 03:23:28 +07:00
afkarxyz 33a6137f75 v5.1 2025-10-21 02:16:31 +07:00
afkarxyz b4fcb6bca6 v5.1 2025-10-21 02:11:41 +07:00
afkarxyz 5ab19a6d37 . 2025-10-21 02:10:50 +07:00
afkarxyz 8547e6d410 v5.0 2025-10-13 05:37:10 +07:00
afkarxyz 17666d8027 Update version.json 2025-10-13 05:16:51 +07:00
afkarxyz ab208482ca v5.0 2025-10-13 05:09:53 +07:00
afkarxyz 76e02d77e8 v5.0 2025-10-13 05:05:01 +07:00
afkarxyz 75cc4543ad Merge pull request #65 from petacz/patch-1
Use Embedded ISRC Tags to check for existing files
2025-10-13 04:55:34 +07:00
Petr V 0b468c4b60 Use Embedded ISRC Tags to check for existing files 2025-10-12 19:53:28 +02:00
afkarxyz 87a6a778f7 v4.9 2025-10-12 00:30:15 +07:00
afkarxyz ef893ab9f4 v4.9 2025-10-12 00:26:01 +07:00
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
18 changed files with 1713 additions and 1026 deletions
+6 -12
View File
@@ -3,26 +3,20 @@
![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 Qobuz, Tidal & Deezer.
<b>SpotiFLAC</b> allows you to download Spotify tracks in true FLAC format through services like Tidal & Deezer.
</div>
### [Download](https://github.com/afkarxyz/SpotiFLAC/releases/download/v4.0/SpotiFLAC.exe)
#
> [!Important]
> - Requires **Google Chrome, Chromium, Microsoft Edge,** or **Brave** to use `Deezer`
> - If after **Cloudflare** verification nothing happens, use a `VPN`, your country is likely blocked by `corsproxy.io`
### [Download](https://github.com/afkarxyz/SpotiFLAC/releases/download/v5.2/SpotiFLAC.exe)
## 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/e81f69f3-1552-4c12-a4e3-2f6d4b322d84)
## Lossless Audio Check
+1003 -405
View File
File diff suppressed because it is too large Load Diff
+30 -21
View File
@@ -3,12 +3,16 @@ 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': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
'User-Agent': get_random_user_agent()
})
self.progress_callback = None
@@ -170,27 +174,22 @@ class DeezerDownloader:
print("Downloading FLAC file...")
try:
response = self.session.get(flac_url, stream=True)
response = self.session.get(flac_url)
response.raise_for_status()
total_size = int(response.headers.get('content-length', 0))
print(f"File size: {total_size} bytes ({total_size / (1024*1024):.2f} MB)")
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)
downloaded = 0
with open(file_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
downloaded += len(chunk)
if self.progress_callback and total_size > 0:
current_mb = downloaded / (1024 * 1024)
total_mb = total_size / (1024 * 1024)
percent = (downloaded / total_size) * 100
self.progress_callback(downloaded, total_size)
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}")
@@ -214,19 +213,29 @@ class DeezerDownloader:
return False
async def main():
if len(sys.argv) != 2:
print("Usage: python deezerDL.py <ISRC>")
print("Example: python deezerDL.py USUM72409273")
return
isrc = sys.argv[1]
print("=== DeezerDL - Deezer Downloader ===")
downloader = DeezerDownloader()
success = await downloader.download_by_isrc(isrc)
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())
-130
View File
@@ -1,130 +0,0 @@
import nodriver as uc
import asyncio
async def download_deezer_track(deezer_link=None, initial_delay=7.5):
if deezer_link is None:
deezer_link = "https://www.deezer.com/us/track/2947516331"
browser = None
try:
browser = await uc.start(headless=False)
page = await browser.get("https://deezmate.com/en")
print("Loading...")
await asyncio.sleep(initial_delay)
input_selector = 'input[placeholder="Paste your Deezer link here..."]'
await page.wait_for(input_selector, timeout=15)
input_element = await page.select(input_selector)
await input_element.clear_input()
await input_element.send_keys(deezer_link)
print("Link entered")
await page.evaluate("""
window.apiResponse = null;
window.originalFetch = window.fetch;
window.fetch = function(...args) {
return window.originalFetch(...args).then(async response => {
if (response.url.includes('api.deezmate.com/dl/')) {
try {
const data = await response.clone().json();
window.apiResponse = data;
console.log('Captured API response:', data);
} catch (e) {
console.log('Error parsing API response:', e);
}
}
return response;
});
};
""")
max_retries = 3
download_button_clicked = False
for attempt in range(max_retries):
try:
download_button_selector = 'button.bg-purple.hover\\:bg-purple-dark.cursor-pointer.transition.text-white.rounded-xl.p-2.mt-2.w-full.mb-5'
await page.wait_for(download_button_selector, timeout=15)
download_button = await page.select(download_button_selector)
await download_button.click()
print("Processing...")
download_button_clicked = True
break
except Exception as e:
if attempt < max_retries - 1:
print(f"Turnstile verification failed, retrying... ({attempt + 1}/{max_retries})")
await asyncio.sleep(0.5)
await page.evaluate("window.apiResponse = null;")
else:
print("Failed to pass Turnstile verification after all retries")
raise e
if not download_button_clicked:
return None
try:
track_download_selector = 'button.bg-purple.text-white.flex.items-center.gap-2.px-3.py-1.rounded-full.hover\\:bg-purple-dark.transition'
await page.wait_for(track_download_selector, timeout=15)
track_download_button = await page.select(track_download_selector)
await track_download_button.click()
except Exception as e:
print(f"Failed to click track download button: {e}")
return None
print("Getting FLAC URL from API response...")
api_response = None
for i in range(30):
api_response = await page.evaluate("window.apiResponse")
if api_response:
break
await asyncio.sleep(0.2)
if not api_response:
return None
def parse_nodriver_response(data):
if isinstance(data, list):
result = {}
for item in data:
if isinstance(item, list) and len(item) == 2:
key = item[0]
value_obj = item[1]
if isinstance(value_obj, dict) and 'value' in value_obj:
if value_obj.get('type') == 'object':
result[key] = parse_nodriver_response(value_obj['value'])
else:
result[key] = value_obj['value']
return result
return data
parsed_response = parse_nodriver_response(api_response)
if parsed_response.get('success') and parsed_response.get('links'):
flac_url = parsed_response['links'].get('flac')
if flac_url:
print(f"Successfully obtained FLAC download URL: {flac_url}")
return flac_url
return None
except Exception as e:
print(f"Error: {e}")
return None
finally:
if browser:
try:
await browser.stop()
except:
pass
async def main(deezer_link=None, initial_delay=7.5):
flac_url = await download_deezer_track(deezer_link, initial_delay)
if not flac_url:
print("Failed to download track")
return flac_url
if __name__ == "__main__":
uc.loop().run_until_complete(main())
+240 -5
View File
@@ -1,5 +1,6 @@
from time import sleep
from urllib.parse import urlparse, parse_qs
from pathlib import Path
import requests
import json
import time
@@ -12,20 +13,34 @@ from typing import Dict, Any, List, Tuple
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)}"
# https://github.com/xyloflake/spot-secrets-go
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:
url = "https://raw.githubusercontent.com/afkarxyz/secretBytes/refs/heads/main/secrets/secretBytes.json"
resp = requests.get(url, timeout=10)
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()
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"])
version = latest_entry["version"]
secret_cipher = latest_entry["secret"]
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_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/{}'
album_base_url = 'https://api.spotify.com/v1/albums/{}'
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 = {
'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-Language': 'en-US,en;q=0.9',
'Accept-Encoding': 'gzip, deflate, br',
@@ -97,11 +114,22 @@ def parse_uri(uri):
if parts[1] == "embed":
parts = parts[1:]
if len(parts) > 1 and parts[1].startswith("intl-"):
parts = parts[1:]
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]}
if l == 5 and parts[3] == "playlist":
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.")
@@ -337,6 +365,69 @@ def get_raw_spotify_data(spotify_url, batch: bool = False, delay: float = 1.0):
except Exception as 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
def format_track_data(track_data):
@@ -466,6 +557,134 @@ def format_playlist_data(playlist_data):
"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):
if not raw_data or "error" in raw_data:
return {"error": "Invalid data provided"}
@@ -477,6 +696,10 @@ def process_spotify_data(raw_data, data_type):
return format_album_data(raw_data)
elif data_type == "playlist":
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:
return {"error": "Invalid data type"}
except Exception as e:
@@ -495,11 +718,23 @@ if __name__ == '__main__':
album = "https://open.spotify.com/album/6J84szYCnMfzEcvIcfWMFL"
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)
print(json.dumps(filtered_playlist, indent=2))
print("\n=== Testing Album ===")
filtered_album = get_filtered_data(album)
print(json.dumps(filtered_album, indent=2))
print("\n=== Testing Track ===")
filtered_track = get_filtered_data(song)
print(json.dumps(filtered_track, indent=2))
+95
View File
@@ -0,0 +1,95 @@
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):
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):
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
+211 -130
View File
@@ -1,9 +1,9 @@
import asyncio
import json
import os
import re
import time
import httpx
import base64
import requests
import json
from mutagen.flac import FLAC, Picture
from mutagen.id3 import PictureType
@@ -16,26 +16,95 @@ class ProgressCallback:
print(f"\r{current / (1024 * 1024):.2f} MB", end="")
class TidalDownloader:
def __init__(self, timeout=30, max_retries=3):
def __init__(self, timeout=30, max_retries=3, api_url=None):
self.timeout = timeout
self.max_retries = max_retries
self.download_chunk_size = 256 * 1024
self.progress_callback = ProgressCallback()
self.client_id = "zU4XHVVkc2tDPo4t"
self.client_secret = "VJKhDFqJPqvsPVNBV6ukXTJmwlvbttP7wlMlrc72se4="
self.client_id = base64.b64decode("NkJEU1JkcEs5aHFFQlRnVQ==").decode()
self.client_secret = base64.b64decode("eGV1UG1ZN25icFo5SUliTEFjUTkzc2hrYTFWTmhlVUFxTjZJY3N6alRHOD0=").decode()
self.api_url = api_url or "https://hifi.401658.xyz"
@staticmethod
def get_available_apis():
try:
response = requests.get("https://status.monochrome.tf/api/stream", timeout=10, stream=True)
for line in response.iter_lines():
if line:
line_str = line.decode('utf-8')
if line_str.startswith('data: '):
data = json.loads(line_str[6:])
api_instances = [
inst for inst in data.get('instances', [])
if inst.get('instance_type') == 'api' and inst.get('last_check', {}).get('success')
]
api_instances.sort(key=lambda x: x.get('avg_response_time', 9999))
return api_instances
except Exception as e:
print(f"Failed to fetch API list: {e}")
return []
@staticmethod
def select_api_interactive():
apis = TidalDownloader.get_available_apis()
if not apis:
print("No APIs available, using default: https://hifi.401658.xyz")
return "https://hifi.401658.xyz"
print("\n=== Available API Instances ===")
print(f"{'No':<4} {'URL':<40} {'Status':<8} {'Uptime':<8} {'Avg Response':<12}")
print("-" * 80)
for i, api in enumerate(apis, 1):
url = api.get('url', 'N/A')
status = "UP" if api.get('last_check', {}).get('success') else "DOWN"
uptime = f"{api.get('uptime', 0):.1f}%"
avg_time = f"{api.get('avg_response_time', 0)}ms"
print(f"{i:<4} {url:<40} {status:<8} {uptime:<8} {avg_time:<12}")
print("\n0 Use default (https://hifi.401658.xyz)")
print("-" * 80)
while True:
try:
choice = input(f"\nSelect API (0-{len(apis)}) [1 for fastest]: ").strip()
if not choice:
choice = "1"
choice_num = int(choice)
if choice_num == 0:
return "https://hifi.401658.xyz"
elif 1 <= choice_num <= len(apis):
selected_url = apis[choice_num - 1]['url']
print(f"\nSelected: {selected_url}")
return selected_url
else:
print(f"Invalid choice. Please enter 0-{len(apis)}")
except ValueError:
print("Invalid input. Please enter a number.")
except KeyboardInterrupt:
print("\nUsing default API")
return "https://hifi.401658.xyz"
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"
async def get_access_token(self):
def get_access_token(self):
refresh_url = "https://auth.tidal.com/v1/oauth2/token"
payload = {
@@ -43,68 +112,67 @@ class TidalDownloader:
"grant_type": "client_credentials",
}
async with httpx.AsyncClient(http2=True) as client:
try:
response = await client.post(
url=refresh_url,
data=payload,
auth=(self.client_id, self.client_secret),
)
try:
response = requests.post(
url=refresh_url,
data=payload,
auth=(self.client_id, self.client_secret),
timeout=self.timeout
)
if response.status_code == 200:
token_data = response.json()
return token_data.get("access_token")
else:
return None
except:
if response.status_code == 200:
token_data = response.json()
return token_data.get("access_token")
else:
return None
async def search_tracks(self, query):
except:
return None
def search_tracks(self, query):
try:
tidal_token = await self.get_access_token()
tidal_token = self.get_access_token()
if not tidal_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"
header = {"authorization": f"Bearer {tidal_token}"}
async with httpx.AsyncClient(http2=True) as client:
search_data = await client.get(url=search_url, headers=header)
response_data = search_data.json()
search_data = requests.get(url=search_url, headers=header, timeout=self.timeout)
response_data = search_data.json()
filtered_items = [{
"id": item.get("id"),
"title": item.get("title"),
"url": item.get("url"),
"isrc": item.get("isrc"),
"audioQuality": item.get("audioQuality"),
"mediaMetadata": item.get("mediaMetadata"),
"album": item.get("album", {}),
"artists": item.get("artists", []),
"artist": item.get("artist", {}),
"trackNumber": item.get("trackNumber"),
"volumeNumber": item.get("volumeNumber"),
"duration": item.get("duration"),
"copyright": item.get("copyright"),
"explicit": item.get("explicit")
} for item in response_data.get("items", [])]
filtered_items = [{
"id": item.get("id"),
"title": item.get("title"),
"url": item.get("url"),
"isrc": item.get("isrc"),
"audioQuality": item.get("audioQuality"),
"mediaMetadata": item.get("mediaMetadata"),
"album": item.get("album", {}),
"artists": item.get("artists", []),
"artist": item.get("artist", {}),
"trackNumber": item.get("trackNumber"),
"volumeNumber": item.get("volumeNumber"),
"duration": item.get("duration"),
"copyright": item.get("copyright"),
"explicit": item.get("explicit")
} for item in response_data.get("items", [])]
return {
"limit": response_data.get("limit"),
"offset": response_data.get("offset"),
"totalNumberOfItems": response_data.get("totalNumberOfItems"),
"items": filtered_items
}
return {
"limit": response_data.get("limit"),
"offset": response_data.get("offset"),
"totalNumberOfItems": response_data.get("totalNumberOfItems"),
"items": filtered_items
}
except Exception as 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 ""))
try:
result = await self.search_tracks(query)
result = self.search_tracks(query)
if not result or not result.get("items"):
raise Exception(f"No tracks found for query: {query}")
@@ -143,98 +211,76 @@ class TidalDownloader:
except Exception as 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...")
download_api_url = f"https://hifi.401658.xyz/track/?id={track_id}&quality={quality}"
download_api_url = f"{self.api_url}/track/?id={track_id}&quality={quality}"
async with httpx.AsyncClient(http2=True, timeout=self.timeout) as client:
try:
response = await client.get(download_api_url)
try:
response = requests.get(download_api_url, timeout=self.timeout)
if response.status_code == 200:
data = response.json()
if response.status_code == 200:
data = response.json()
for item in data:
if "OriginalTrackUrl" in item:
print("URL found")
return {
"download_url": item["OriginalTrackUrl"],
"track_info": data[0] if data else {}
}
for item in data:
if "OriginalTrackUrl" in item:
print("URL found")
return {
"download_url": item["OriginalTrackUrl"],
"track_info": data[0] if data else {}
}
raise Exception("Download URL not found in response")
else:
raise Exception(f"API returned status code: {response.status_code}")
raise Exception("Download URL not found in response")
else:
raise Exception(f"API returned status code: {response.status_code}")
except Exception as e:
raise Exception(f"Error getting download URL: {str(e)}")
except Exception as 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:
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 = await client.get(art_url)
response = requests.get(art_url, timeout=self.timeout)
if response.status_code == 200:
return response.content
else:
print(f"Failed to download album art: HTTP {response.status_code}")
return None
if response.status_code == 200:
return response.content
else:
print(f"Failed to download album art: HTTP {response.status_code}")
return None
except Exception as e:
print(f"Error downloading album art: {str(e)}")
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):
file_dir = os.path.dirname(filepath)
if file_dir and not os.path.exists(file_dir):
os.makedirs(file_dir, exist_ok=True)
temp_filepath = filepath + ".part"
retry_count = 0
while retry_count <= self.max_retries:
try:
async with httpx.AsyncClient(http2=True, timeout=60.0) as client:
async with client.stream('GET', url) as response:
if response.status_code != 200:
raise Exception(f"HTTP {response.status_code}")
response = requests.get(url, timeout=60.0)
if response.status_code != 200:
raise Exception(f"HTTP {response.status_code}")
total_size = int(response.headers.get('content-length', 0))
downloaded_size = 0
start_time = time.time()
last_update_time = start_time
if is_stopped_callback and is_stopped_callback():
raise Exception("Download stopped")
with open(temp_filepath, 'wb') as f:
async for chunk in response.aiter_bytes(chunk_size=self.download_chunk_size):
if is_stopped_callback and is_stopped_callback():
f.close()
if os.path.exists(temp_filepath):
os.remove(temp_filepath)
raise Exception("Download stopped")
while is_paused_callback and is_paused_callback():
time.sleep(0.1)
if is_stopped_callback and is_stopped_callback():
raise Exception("Download stopped")
while is_paused_callback and is_paused_callback():
await asyncio.sleep(0.1)
if is_stopped_callback and is_stopped_callback():
f.close()
if os.path.exists(temp_filepath):
os.remove(temp_filepath)
raise Exception("Download stopped")
with open(temp_filepath, 'wb') as f:
f.write(response.content)
f.write(chunk)
downloaded_size += len(chunk)
downloaded_size = len(response.content)
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)
if self.progress_callback:
self.progress_callback(downloaded_size, downloaded_size)
os.rename(temp_filepath, filepath)
print("Download complete")
@@ -252,9 +298,9 @@ class TidalDownloader:
print(f"Download error (attempt {retry_count}/{self.max_retries}): {str(e)}")
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:
print("Embedding metadata...")
audio = FLAC(filepath)
@@ -325,7 +371,7 @@ class TidalDownloader:
audio["COMMENT"] = f"Tidal {track_info['audioQuality']}"
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:
picture = Picture()
picture.data = album_art
@@ -343,14 +389,47 @@ class TidalDownloader:
print(f"Error embedding metadata: {str(e)}")
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, auto_fallback=False):
if output_dir != ".":
try:
os.makedirs(output_dir, exist_ok=True)
except OSError as e:
raise Exception(f"Directory error: {e}")
track_info = await self.get_track_info(query, isrc)
if auto_fallback:
apis = self.get_available_apis()
if not apis:
print("No APIs available for fallback, using current API")
return self._download_single(query, isrc, output_dir, quality, is_paused_callback, is_stopped_callback)
last_error = None
for i, api in enumerate(apis, 1):
api_url = api.get('url')
try:
print(f"[Auto Fallback {i}/{len(apis)}] Trying: {api_url}")
fallback_downloader = TidalDownloader(api_url=api_url)
fallback_downloader.set_progress_callback(self.progress_callback)
result = fallback_downloader._download_single(
query, isrc, output_dir, quality,
is_paused_callback, is_stopped_callback
)
print(f"✓ Success with: {api_url}")
return result
except Exception as e:
last_error = str(e)
print(f"✗ Failed with {api_url}: {last_error[:80]}")
continue
raise Exception(f"All {len(apis)} APIs failed. Last error: {last_error}")
return self._download_single(query, isrc, output_dir, quality, is_paused_callback, is_stopped_callback)
def _download_single(self, query, isrc, output_dir, quality, is_paused_callback, is_stopped_callback):
track_info = self.get_track_info(query, isrc)
track_id = track_info.get("id")
if not track_id:
@@ -376,12 +455,12 @@ class TidalDownloader:
print(f"File already exists: {output_filename} ({file_size / (1024 * 1024):.2f} MB)")
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_track_info = download_info["track_info"]
print(f"Downloading to: {output_filename}")
await self.download_file(
self.download_file(
download_url,
output_filename,
is_paused_callback=is_paused_callback,
@@ -390,7 +469,7 @@ class TidalDownloader:
print("Adding metadata...")
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")
except Exception as e:
print(f"Tagging failed: {e}")
@@ -398,16 +477,18 @@ class TidalDownloader:
print("Done")
return output_filename
async def main():
def main():
print("=== TidalDL - Tidal Downloader ===")
downloader = TidalDownloader(timeout=30, max_retries=3)
selected_api = TidalDownloader.select_api_interactive()
downloader = TidalDownloader(timeout=30, max_retries=3, api_url=selected_api)
query = "APT."
isrc = "USAT22409172"
output_dir = "."
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}")
except Exception as e:
print(f"Error: {str(e)}")
@@ -425,4 +506,4 @@ if __name__ == "__main__":
except:
pass
asyncio.run(main())
main()
+1 -1
View File
@@ -1,3 +1,3 @@
{
"version": "4.0"
"version": "5.2"
}