Compare commits
56 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 3ac4c34d73 | |||
| 88e303cbe4 | |||
| c13855fadd | |||
| 2b12684960 | |||
| 4bc164cc56 | |||
| 46cb65665e | |||
| 276b3b4951 | |||
| e15aadbd61 | |||
| d7639bae8f | |||
| 1af7ab65c9 | |||
| c5240596cb | |||
| c4a9042adc | |||
| 45ac08ecbd | |||
| 0add305d9c | |||
| 9b6b43c0a4 | |||
| 60d20cbebe | |||
| 626d58667e | |||
| 4dd1a7ea12 | |||
| 67964e4acb | |||
| 1486fb13df | |||
| 966536f127 | |||
| 21946321f5 | |||
| 3e3cb0610d | |||
| 160eba0987 | |||
| 71a60ded47 | |||
| e0a0514df9 | |||
| 1e7a48d263 | |||
| 0a83a0dd6e | |||
| da429d9410 | |||
| 63211c726b | |||
| 055cb6991a | |||
| 222d681551 | |||
| 479c6ede2b | |||
| ceb727adb9 | |||
| bbea8ca493 | |||
| f567dd19bf | |||
| 3651833e2a | |||
| 8403b96306 | |||
| d977829e36 | |||
| 2aaf123c98 | |||
| 13567802a0 | |||
| d704782519 | |||
| 884c02278f | |||
| d6abe2bae3 | |||
| 7b858dd0ce | |||
| cfeb9a2ef2 | |||
| 81a78832ff | |||
| 071f20deff | |||
| 03de68ac7b | |||
| 77363f9e61 | |||
| 2df77120cf | |||
| e6e953b2ed | |||
| 72f17479e8 | |||
| 9286fba63c | |||
| 6bf7084959 | |||
| 03cc3d82a7 |
@@ -1,120 +0,0 @@
|
||||
import requests
|
||||
from tqdm import tqdm
|
||||
import time
|
||||
import os
|
||||
import asyncio
|
||||
from GetMetadata import main as get_metadata
|
||||
|
||||
class TrackDownloader:
|
||||
def __init__(self):
|
||||
self.client = requests.Session()
|
||||
self.headers = {
|
||||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
|
||||
}
|
||||
|
||||
async def get_track_info(self):
|
||||
metadata = await get_metadata()
|
||||
return metadata
|
||||
|
||||
def sanitize_filename(self, filename):
|
||||
invalid_chars = '<>:"/\\|?*'
|
||||
for char in invalid_chars:
|
||||
filename = filename.replace(char, '')
|
||||
|
||||
filename = ' '.join(filename.split())
|
||||
filename = filename.replace(' ,', ',')
|
||||
filename = filename.replace(',', ', ')
|
||||
while ' ' in filename:
|
||||
filename = filename.replace(' ', ' ')
|
||||
filename = filename.rsplit('.', 1)
|
||||
filename[0] = filename[0].strip()
|
||||
return '.'.join(filename)
|
||||
|
||||
def download(self, metadata, output_dir):
|
||||
track_url = metadata['url']
|
||||
primary_token = metadata['token']
|
||||
expiry = metadata['expiry']
|
||||
|
||||
print(f"Starting download for: {track_url}")
|
||||
|
||||
initial_request = {
|
||||
"account": {"id": "auto", "type": "country"},
|
||||
"compat": "false",
|
||||
"downscale": "original",
|
||||
"handoff": True,
|
||||
"metadata": True,
|
||||
"private": True,
|
||||
"token": {
|
||||
"expiry": expiry,
|
||||
"primary": primary_token
|
||||
},
|
||||
"upload": {"enabled": False, "service": "pixeldrain"},
|
||||
"url": track_url
|
||||
}
|
||||
|
||||
response = self.client.post("https://lucida.to/api/load?url=/api/fetch/stream/v2",
|
||||
json=initial_request,
|
||||
headers=self.headers)
|
||||
|
||||
csrf_token = response.cookies.get('csrf_token')
|
||||
if csrf_token:
|
||||
self.headers['X-CSRF-Token'] = csrf_token
|
||||
|
||||
initial_response = response.json()
|
||||
|
||||
if not initial_response.get("success", False):
|
||||
raise Exception(f"Initial request failed: {initial_response.get('error', 'Unknown error')}")
|
||||
|
||||
handoff = initial_response["handoff"]
|
||||
server = initial_response["server"]
|
||||
|
||||
file_name = f"{metadata['title']} - {metadata['artists']}.flac"
|
||||
file_name = self.sanitize_filename(file_name)
|
||||
|
||||
completion_url = f"https://{server}.lucida.to/api/fetch/request/{handoff}"
|
||||
|
||||
print("Waiting for track processing to complete")
|
||||
while True:
|
||||
completion_response = self.client.get(completion_url, headers=self.headers).json()
|
||||
if completion_response["status"] == "completed":
|
||||
break
|
||||
elif completion_response["status"] == "error":
|
||||
raise Exception(f"API request failed: {completion_response.get('message', 'Unknown error')}")
|
||||
time.sleep(1)
|
||||
|
||||
download_url = f"https://{server}.lucida.to/api/fetch/request/{handoff}/download"
|
||||
print(f"Starting download of: {file_name}")
|
||||
|
||||
response = self.client.get(download_url, stream=True, headers=self.headers)
|
||||
total_size = int(response.headers.get('content-length', 0))
|
||||
|
||||
file_path = os.path.join(output_dir, file_name)
|
||||
|
||||
with open(file_path, 'wb') as file, tqdm(
|
||||
desc=file_name,
|
||||
total=total_size,
|
||||
unit='iB',
|
||||
unit_scale=True,
|
||||
unit_divisor=1024,
|
||||
) as progress_bar:
|
||||
for data in response.iter_content(chunk_size=1024):
|
||||
size = file.write(data)
|
||||
progress_bar.update(size)
|
||||
|
||||
print(f"Download completed: {file_path}")
|
||||
return file_path
|
||||
|
||||
async def main():
|
||||
downloader = TrackDownloader()
|
||||
output_dir = "."
|
||||
|
||||
try:
|
||||
metadata = await downloader.get_track_info()
|
||||
|
||||
downloaded_file = downloader.download(metadata, output_dir)
|
||||
print(f"File downloaded successfully: {downloaded_file}")
|
||||
except Exception as e:
|
||||
print(f"An error occurred: {str(e)}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,7 +1,7 @@
|
||||
import asyncio
|
||||
import zendriver as zd
|
||||
|
||||
async def get_metadata(page):
|
||||
async def get_metadata(page, headless=True):
|
||||
max_attempts = 40
|
||||
attempts = 0
|
||||
|
||||
@@ -14,9 +14,9 @@ async def get_metadata(page):
|
||||
const [url, config] = args;
|
||||
if (url.includes('/api/load?url=%2Fapi%2Ffetch%2Fstream%2Fv2')) {
|
||||
const payload = JSON.parse(config.body);
|
||||
const title = document.querySelector('h1.svelte-6pt9ji').textContent;
|
||||
const title = document.querySelector('h1.svelte-6pt9ji').textContent.trim();
|
||||
const artists = Array.from(document.querySelectorAll('h2.svelte-6pt9ji a.normal'))
|
||||
.map(a => a.textContent)
|
||||
.map(a => a.textContent.trim())
|
||||
.join(', ');
|
||||
const cover = document.querySelector('.svelte-6pt9ji .meta.svelte-6pt9ji a').href;
|
||||
|
||||
@@ -82,8 +82,8 @@ async def get_metadata(page):
|
||||
|
||||
raise TimeoutError("Timeout")
|
||||
|
||||
async def main():
|
||||
browser = await zd.start(headless=False)
|
||||
async def main(headless=True):
|
||||
browser = await zd.start(headless=headless)
|
||||
try:
|
||||
track_id = "2plbrEY59IikOBgBGLjaoe"
|
||||
url = f"https://lucida.to/?url=https%3A%2F%2Fopen.spotify.com%2Ftrack%2F{track_id}&country=auto&to=tidal"
|
||||
@@ -1,33 +1,33 @@
|
||||
[](https://github.com/afkarxyz/SpotiFLAC/releases)
|
||||
|
||||

|
||||

|
||||
|
||||
<div align="center">
|
||||
<b>SpotiFLAC</b> allows you to download Spotify tracks in true FLAC format through services like Tidal, Amazon Music and Qobuz with the help of Lucida.
|
||||
<b>SpotiFLAC</b> allows you to download Spotify tracks in true FLAC format through services like Tidal, Amazon Music, and Deezer <code>(via Lucida)</code>, as well as Qobuz <code>(via SquidWTF)</code>.
|
||||
</div>
|
||||
|
||||
### [Download](https://github.com/afkarxyz/SpotiFLAC/releases/download/v1.6/SpotiFLAC.exe)
|
||||
### [Download](https://github.com/afkarxyz/SpotiFLAC/releases/download/v2.9/SpotiFLAC.exe)
|
||||
|
||||
#
|
||||
|
||||
> [!NOTE]
|
||||
> Requires **Google Chrome**
|
||||
|
||||
> [!WARNING]
|
||||
Sometimes, the **download speed** from Lucida can be fast or slow; it varies unpredictably.
|
||||
> [!Note]
|
||||
**Download speed** from Lucida is unpredictable—sometimes fast, sometimes slow. Join their [Discord](https://discord.com/invite/dXEGRWqEbS) for updates.
|
||||
|
||||
## Screenshots
|
||||
|
||||

|
||||

|
||||
|
||||
> - When **Headless** is enabled, the browser runs in the background without a graphical interface, improving performance and allowing seamless automation.
|
||||
> - When **Fallback** is enabled, it will use the backup server Lucida.su
|
||||
> - **Filename: Title** means the filename format is `Title - Artist`, and vice versa.
|
||||
> - I highly recommend **Tidal** or **Amazon Music** because `Qobuz` occasionally experience issues.
|
||||

|
||||
|
||||

|
||||

|
||||
|
||||

|
||||

|
||||
|
||||

|
||||
|
||||

|
||||
|
||||
> When **Fallback** is enabled, it will use the backup server `Lucida.su`
|
||||
|
||||
## Lossless Audio Check
|
||||
|
||||
@@ -35,4 +35,4 @@ Sometimes, the **download speed** from Lucida can be fast or slow; it varies unp
|
||||
|
||||

|
||||
|
||||
### [Download](https://github.com/afkarxyz/SpotiFLAC/releases/download/v0/FLAC-Checker.zip) FLAC Checker
|
||||
#### [Download](https://github.com/afkarxyz/SpotiFLAC/releases/download/v0/FLAC-Checker.zip) FLAC Checker
|
||||
|
||||
+1361
-518
File diff suppressed because it is too large
Load Diff
+459
-80
@@ -1,99 +1,478 @@
|
||||
import asyncio
|
||||
import zendriver as zd
|
||||
from time import sleep
|
||||
from urllib.parse import urlparse, parse_qs
|
||||
import requests
|
||||
import json
|
||||
import hmac
|
||||
import time
|
||||
import hashlib
|
||||
from typing import Tuple, Callable, Dict, Any, List
|
||||
|
||||
async def get_metadata(page, headless=True):
|
||||
max_attempts = 40
|
||||
attempts = 0
|
||||
_TOTP_SECRET = bytearray([53,53,48,55,49,52,53,56,53,51,52,56,55,52,57,57,53,57,50,50,52,56,54,51,48,51,50,57,51,52,55])
|
||||
|
||||
await asyncio.sleep(2)
|
||||
def generate_totp(
|
||||
secret: bytes = _TOTP_SECRET,
|
||||
algorithm: Callable[[], object] = hashlib.sha1,
|
||||
digits: int = 6,
|
||||
counter_factory: Callable[[], int] = lambda: int(time.time()) // 30,
|
||||
) -> Tuple[str, int]:
|
||||
counter = counter_factory()
|
||||
hmac_result = hmac.new(
|
||||
secret, counter.to_bytes(8, byteorder="big"), algorithm
|
||||
).digest()
|
||||
|
||||
await page.evaluate("""
|
||||
window.downloadInfo = null;
|
||||
const originalFetch = window.fetch;
|
||||
window.fetch = async function(...args) {
|
||||
const [url, config] = args;
|
||||
if (url.includes('/api/load?url=%2Fapi%2Ffetch%2Fstream%2Fv2')) {
|
||||
const payload = JSON.parse(config.body);
|
||||
const title = document.querySelector('h1.svelte-6pt9ji').textContent.trim();
|
||||
const artists = Array.from(document.querySelectorAll('h2.svelte-6pt9ji a.normal'))
|
||||
.map(a => a.textContent.trim())
|
||||
.join(', ');
|
||||
const cover = document.querySelector('.svelte-6pt9ji .meta.svelte-6pt9ji a').href;
|
||||
offset = hmac_result[-1] & 15
|
||||
truncated_value = (
|
||||
(hmac_result[offset] & 127) << 24
|
||||
| (hmac_result[offset + 1] & 255) << 16
|
||||
| (hmac_result[offset + 2] & 255) << 8
|
||||
| (hmac_result[offset + 3] & 255)
|
||||
)
|
||||
return (
|
||||
str(truncated_value % (10**digits)).zfill(digits),
|
||||
counter * 30_000,
|
||||
)
|
||||
|
||||
window.downloadInfo = {
|
||||
url: payload.url,
|
||||
cover: cover,
|
||||
title: title,
|
||||
artists: artists,
|
||||
token: payload.token.primary,
|
||||
expiry: payload.token.expiry
|
||||
};
|
||||
}
|
||||
return originalFetch.apply(this, args);
|
||||
};
|
||||
""")
|
||||
token_url = 'https://open.spotify.com/get_access_token'
|
||||
playlist_base_url = 'https://api.spotify.com/v1/playlists/{}'
|
||||
album_base_url = 'https://api.spotify.com/v1/albums/{}'
|
||||
track_base_url = 'https://api.spotify.com/v1/tracks/{}'
|
||||
headers = {
|
||||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36',
|
||||
'Accept': 'application/json',
|
||||
'Accept-Language': 'en-US,en;q=0.9',
|
||||
'Accept-Encoding': 'gzip, deflate, br',
|
||||
'sec-ch-ua-platform': '"Windows"',
|
||||
'sec-fetch-dest': 'empty',
|
||||
'sec-fetch-mode': 'cors',
|
||||
'sec-fetch-site': 'same-origin',
|
||||
'Referer': 'https://open.spotify.com/',
|
||||
'Origin': 'https://open.spotify.com'
|
||||
}
|
||||
|
||||
await page.evaluate("""
|
||||
function waitForElement(selector) {
|
||||
return new Promise(resolve => {
|
||||
if (document.querySelector(selector)) {
|
||||
return resolve(document.querySelector(selector));
|
||||
}
|
||||
class SpotifyInvalidUrlException(Exception):
|
||||
pass
|
||||
|
||||
const observer = new MutationObserver(mutations => {
|
||||
if (document.querySelector(selector)) {
|
||||
observer.disconnect();
|
||||
resolve(document.querySelector(selector));
|
||||
}
|
||||
});
|
||||
class SpotifyWebsiteParserException(Exception):
|
||||
pass
|
||||
|
||||
observer.observe(document.documentElement, {
|
||||
childList: true,
|
||||
subtree: true
|
||||
});
|
||||
});
|
||||
def parse_uri(uri):
|
||||
u = urlparse(uri)
|
||||
if u.netloc == "embed.spotify.com":
|
||||
if not u.query:
|
||||
raise SpotifyInvalidUrlException("ERROR: url {} is not supported".format(uri))
|
||||
qs = parse_qs(u.query)
|
||||
return parse_uri(qs['uri'][0])
|
||||
|
||||
if not u.scheme and not u.netloc:
|
||||
return {"type": "playlist", "id": u.path}
|
||||
|
||||
if u.scheme == "spotify":
|
||||
parts = uri.split(":")
|
||||
else:
|
||||
if u.netloc != "open.spotify.com" and u.netloc != "play.spotify.com":
|
||||
raise SpotifyInvalidUrlException("ERROR: url {} is not supported".format(uri))
|
||||
parts = u.path.split("/")
|
||||
|
||||
if parts[1] == "embed":
|
||||
parts = parts[1:]
|
||||
|
||||
l = len(parts)
|
||||
if l == 3 and parts[1] in ["album", "track", "playlist"]:
|
||||
return {"type": parts[1], "id": parts[2]}
|
||||
if l == 5 and parts[3] == "playlist":
|
||||
return {"type": parts[3], "id": parts[4]}
|
||||
|
||||
raise SpotifyInvalidUrlException("ERROR: unable to determine Spotify URL type or type is unsupported.")
|
||||
|
||||
def get_json_from_api(api_url, access_token):
|
||||
headers.update({'Authorization': 'Bearer {}'.format(access_token)})
|
||||
|
||||
req = requests.get(api_url, headers=headers, timeout=10)
|
||||
|
||||
if req.status_code == 429:
|
||||
seconds = int(req.headers.get("Retry-After", "5")) + 1
|
||||
print(f"INFO: rate limited! Sleeping for {seconds} seconds")
|
||||
sleep(seconds)
|
||||
return None
|
||||
|
||||
if req.status_code != 200:
|
||||
raise SpotifyWebsiteParserException(f"ERROR: {api_url} gave us not a 200. Instead: {req.status_code}")
|
||||
|
||||
return req.json()
|
||||
|
||||
def get_access_token():
|
||||
try:
|
||||
totp, timestamp = generate_totp()
|
||||
|
||||
params = {
|
||||
"reason": "init",
|
||||
"productType": "web-player",
|
||||
"totp": totp,
|
||||
"totpVer": 5,
|
||||
"ts": timestamp,
|
||||
}
|
||||
|
||||
(async () => {
|
||||
if (!window.location.hostname.includes('lucida.')) return;
|
||||
req = requests.get(token_url, headers=headers, params=params, timeout=10)
|
||||
if req.status_code != 200:
|
||||
return {"error": f"Failed to get access token. Status code: {req.status_code}"}
|
||||
return req.json()
|
||||
except Exception as e:
|
||||
return {"error": f"Failed to get access token: {str(e)}"}
|
||||
|
||||
await Promise.race([
|
||||
waitForElement('.d1-track button'),
|
||||
waitForElement('button[class*="download-button"]')
|
||||
]);
|
||||
def fetch_tracks_in_batches(url: str, access_token: str, batch_size: int = 100, delay: float = 1.0) -> Tuple[List[Dict[str, Any]], int]:
|
||||
all_tracks = []
|
||||
current_batch = 0
|
||||
|
||||
const clickDownloadButton = () => {
|
||||
const button = document.querySelector('.d1-track button') ||
|
||||
document.querySelector('button[class*="download-button"]');
|
||||
if (button) button.click();
|
||||
};
|
||||
while url:
|
||||
print(f"Batch : {current_batch}")
|
||||
|
||||
clickDownloadButton();
|
||||
})();
|
||||
""")
|
||||
url_parts = url.split("offset=")
|
||||
if len(url_parts) > 1:
|
||||
offset_part = url_parts[1].split("&")[0]
|
||||
print(f"Offset : {offset_part}")
|
||||
print("-------------")
|
||||
|
||||
while attempts < max_attempts:
|
||||
download_info = await page.evaluate("window.downloadInfo")
|
||||
if download_info:
|
||||
return download_info
|
||||
track_data = get_json_from_api(url, access_token)
|
||||
if not track_data:
|
||||
break
|
||||
|
||||
await asyncio.sleep(0.5)
|
||||
attempts += 1
|
||||
items = track_data.get('items', [])
|
||||
all_tracks.extend(items)
|
||||
|
||||
raise TimeoutError("Timeout")
|
||||
url = track_data.get('next')
|
||||
if url and "&locale=" in url:
|
||||
url = url.split("&locale=")[0]
|
||||
|
||||
if url and delay > 0:
|
||||
sleep(delay)
|
||||
|
||||
current_batch += 1
|
||||
|
||||
return all_tracks, current_batch
|
||||
|
||||
def get_raw_spotify_data(spotify_url, batch: bool = False, delay: float = 1.0):
|
||||
url_info = parse_uri(spotify_url)
|
||||
token = get_access_token()
|
||||
|
||||
if "error" in token:
|
||||
return token
|
||||
|
||||
access_token = token["accessToken"]
|
||||
raw_data = {}
|
||||
|
||||
if url_info['type'] == "playlist":
|
||||
try:
|
||||
playlist_data = get_json_from_api(
|
||||
playlist_base_url.format(url_info["id"]),
|
||||
access_token
|
||||
)
|
||||
if not playlist_data:
|
||||
return {"error": "Failed to get playlist data"}
|
||||
|
||||
raw_data = playlist_data
|
||||
total_tracks = playlist_data.get('tracks', {}).get('total', 0)
|
||||
|
||||
if batch:
|
||||
tracks_url = f'https://api.spotify.com/v1/playlists/{url_info["id"]}/tracks?limit=100'
|
||||
tracks, num_batches = fetch_tracks_in_batches(tracks_url, access_token, 100, delay)
|
||||
raw_data['tracks']['items'] = tracks
|
||||
raw_data['_batch_count'] = num_batches
|
||||
raw_data['_batch_enabled'] = True
|
||||
|
||||
if len(tracks) < total_tracks:
|
||||
last_offset = len(tracks)
|
||||
remaining_tracks = []
|
||||
|
||||
while last_offset < total_tracks:
|
||||
print(f"Batch : {num_batches}")
|
||||
print(f"Offset : {last_offset}")
|
||||
print("-------------")
|
||||
|
||||
remainder_url = f'https://api.spotify.com/v1/playlists/{url_info["id"]}/tracks?offset={last_offset}&limit=100'
|
||||
track_data = get_json_from_api(remainder_url, access_token)
|
||||
|
||||
if not track_data or not track_data.get('items'):
|
||||
break
|
||||
|
||||
items = track_data.get('items', [])
|
||||
remaining_tracks.extend(items)
|
||||
|
||||
if len(items) < 100:
|
||||
break
|
||||
|
||||
last_offset += len(items)
|
||||
num_batches += 1
|
||||
|
||||
if delay > 0:
|
||||
sleep(delay)
|
||||
|
||||
tracks.extend(remaining_tracks)
|
||||
raw_data['tracks']['items'] = tracks
|
||||
raw_data['_batch_count'] = num_batches
|
||||
else:
|
||||
tracks = []
|
||||
tracks_url = f'https://api.spotify.com/v1/playlists/{url_info["id"]}/tracks?limit=100'
|
||||
while tracks_url:
|
||||
track_data = get_json_from_api(tracks_url, access_token)
|
||||
if not track_data:
|
||||
break
|
||||
|
||||
tracks.extend(track_data['items'])
|
||||
tracks_url = track_data.get('next')
|
||||
if tracks_url and "&locale=" in tracks_url:
|
||||
tracks_url = tracks_url.split("&locale=")[0]
|
||||
|
||||
raw_data['tracks']['items'] = tracks
|
||||
raw_data['_batch_enabled'] = False
|
||||
|
||||
except Exception as e:
|
||||
return {"error": f"Failed to get playlist data: {str(e)}"}
|
||||
|
||||
elif url_info["type"] == "album":
|
||||
try:
|
||||
album_data = get_json_from_api(
|
||||
album_base_url.format(url_info["id"]),
|
||||
access_token
|
||||
)
|
||||
if not album_data:
|
||||
return {"error": "Failed to get album data"}
|
||||
|
||||
album_data['_token'] = access_token
|
||||
raw_data = album_data
|
||||
total_tracks = album_data.get('total_tracks', 0)
|
||||
|
||||
if batch:
|
||||
tracks_url = f'{album_base_url.format(url_info["id"])}/tracks?limit=50'
|
||||
tracks, num_batches = fetch_tracks_in_batches(tracks_url, access_token, 50, delay)
|
||||
raw_data['tracks']['items'] = tracks
|
||||
raw_data['_batch_count'] = num_batches
|
||||
raw_data['_batch_enabled'] = True
|
||||
|
||||
if len(tracks) < total_tracks:
|
||||
last_offset = len(tracks)
|
||||
remaining_tracks = []
|
||||
|
||||
while last_offset < total_tracks:
|
||||
print(f"Batch : {num_batches}")
|
||||
print(f"Offset : {last_offset}")
|
||||
print("-------------")
|
||||
|
||||
remainder_url = f'{album_base_url.format(url_info["id"])}/tracks?offset={last_offset}&limit=50'
|
||||
track_data = get_json_from_api(remainder_url, access_token)
|
||||
|
||||
if not track_data or not track_data.get('items'):
|
||||
break
|
||||
|
||||
items = track_data.get('items', [])
|
||||
remaining_tracks.extend(items)
|
||||
|
||||
if len(items) < 50:
|
||||
break
|
||||
|
||||
last_offset += len(items)
|
||||
num_batches += 1
|
||||
|
||||
if delay > 0:
|
||||
sleep(delay)
|
||||
|
||||
tracks.extend(remaining_tracks)
|
||||
raw_data['tracks']['items'] = tracks
|
||||
raw_data['_batch_count'] = num_batches
|
||||
else:
|
||||
tracks = []
|
||||
tracks_url = f'{album_base_url.format(url_info["id"])}/tracks?limit=50'
|
||||
while tracks_url:
|
||||
track_data = get_json_from_api(tracks_url, access_token)
|
||||
if not track_data:
|
||||
break
|
||||
|
||||
tracks.extend(track_data['items'])
|
||||
tracks_url = track_data.get('next')
|
||||
if tracks_url and "&locale=" in tracks_url:
|
||||
tracks_url = tracks_url.split("&locale=")[0]
|
||||
|
||||
raw_data['tracks']['items'] = tracks
|
||||
raw_data['_batch_enabled'] = False
|
||||
|
||||
except Exception as e:
|
||||
return {"error": f"Failed to get album data: {str(e)}"}
|
||||
|
||||
elif url_info["type"] == "track":
|
||||
try:
|
||||
track_data = get_json_from_api(
|
||||
track_base_url.format(url_info["id"]),
|
||||
access_token
|
||||
)
|
||||
if not track_data:
|
||||
return {"error": "Failed to get track data"}
|
||||
|
||||
raw_data = track_data
|
||||
except Exception as e:
|
||||
return {"error": f"Failed to get track data: {str(e)}"}
|
||||
|
||||
return raw_data
|
||||
|
||||
def format_track_data(track_data):
|
||||
artists = []
|
||||
for artist in track_data.get('artists', []):
|
||||
artists.append(artist['name'])
|
||||
|
||||
image_url = track_data.get('album', {}).get('images', [{}])[0].get('url', '') if track_data.get('album', {}).get('images') else ''
|
||||
|
||||
isrc = track_data.get('external_ids', {}).get('isrc', '')
|
||||
|
||||
return {
|
||||
"track": {
|
||||
"artists": ", ".join(artists),
|
||||
"name": track_data.get('name', ''),
|
||||
"album_name": track_data.get('album', {}).get('name', ''),
|
||||
"duration_ms": track_data.get('duration_ms', 0),
|
||||
"images": image_url,
|
||||
"release_date": track_data.get('album', {}).get('release_date', ''),
|
||||
"track_number": track_data.get('track_number', 0),
|
||||
"external_urls": track_data.get('external_urls', {}).get('spotify', ''),
|
||||
"isrc": isrc
|
||||
}
|
||||
}
|
||||
|
||||
def format_album_data(album_data):
|
||||
artists = []
|
||||
for artist in album_data.get('artists', []):
|
||||
artists.append(artist['name'])
|
||||
|
||||
image_url = album_data.get('images', [{}])[0].get('url', '') if album_data.get('images') else ''
|
||||
|
||||
track_list = []
|
||||
for track in album_data.get('tracks', {}).get('items', []):
|
||||
track_artists = []
|
||||
for artist in track.get('artists', []):
|
||||
track_artists.append(artist['name'])
|
||||
|
||||
track_id = track.get('id', '')
|
||||
track_isrc = ''
|
||||
|
||||
if track_id and album_data.get('_token'):
|
||||
try:
|
||||
full_track_data = get_json_from_api(
|
||||
track_base_url.format(track_id),
|
||||
album_data.get('_token')
|
||||
)
|
||||
if full_track_data:
|
||||
track_isrc = full_track_data.get('external_ids', {}).get('isrc', '')
|
||||
except:
|
||||
pass
|
||||
|
||||
track_list.append({
|
||||
"artists": ", ".join(track_artists),
|
||||
"name": track.get('name', ''),
|
||||
"album_name": album_data.get('name', ''),
|
||||
"duration_ms": track.get('duration_ms', 0),
|
||||
"images": image_url,
|
||||
"release_date": album_data.get('release_date', ''),
|
||||
"track_number": track.get('track_number', 0),
|
||||
"external_urls": track.get('external_urls', {}).get('spotify', ''),
|
||||
"isrc": track_isrc
|
||||
})
|
||||
|
||||
album_info = {
|
||||
"total_tracks": album_data.get('total_tracks', 0),
|
||||
"name": album_data.get('name', ''),
|
||||
"release_date": album_data.get('release_date', ''),
|
||||
"artists": ", ".join(artists),
|
||||
"images": image_url
|
||||
}
|
||||
|
||||
if album_data.get('_batch_enabled', False):
|
||||
album_info["batch"] = f"{album_data.get('_batch_count', 1)}"
|
||||
|
||||
return {
|
||||
"album_info": album_info,
|
||||
"track_list": track_list
|
||||
}
|
||||
|
||||
def format_playlist_data(playlist_data):
|
||||
image_url = playlist_data.get('images', [{}])[0].get('url', '') if playlist_data.get('images') else ''
|
||||
|
||||
track_list = []
|
||||
for item in playlist_data.get('tracks', {}).get('items', []):
|
||||
track = item.get('track', {})
|
||||
if not track:
|
||||
continue
|
||||
|
||||
artists = []
|
||||
for artist in track.get('artists', []):
|
||||
artists.append(artist['name'])
|
||||
|
||||
track_image = ''
|
||||
if track.get('album', {}).get('images'):
|
||||
track_image = track.get('album', {}).get('images', [{}])[0].get('url', '')
|
||||
|
||||
track_isrc = track.get('external_ids', {}).get('isrc', '')
|
||||
|
||||
track_list.append({
|
||||
"artists": ", ".join(artists),
|
||||
"name": track.get('name', ''),
|
||||
"album_name": track.get('album', {}).get('name', ''),
|
||||
"duration_ms": track.get('duration_ms', 0),
|
||||
"images": track_image,
|
||||
"release_date": track.get('album', {}).get('release_date', ''),
|
||||
"track_number": track.get('track_number', 0),
|
||||
"external_urls": track.get('external_urls', {}).get('spotify', ''),
|
||||
"isrc": track_isrc
|
||||
})
|
||||
|
||||
playlist_info = {
|
||||
"tracks": {"total": playlist_data.get('tracks', {}).get('total', 0)},
|
||||
"followers": {"total": playlist_data.get('followers', {}).get('total', 0)},
|
||||
"owner": {
|
||||
"display_name": playlist_data.get('owner', {}).get('display_name', ''),
|
||||
"name": playlist_data.get('name', ''),
|
||||
"images": image_url
|
||||
}
|
||||
}
|
||||
|
||||
if playlist_data.get('_batch_enabled', False):
|
||||
playlist_info["batch"] = f"{playlist_data.get('_batch_count', 1)}"
|
||||
|
||||
return {
|
||||
"playlist_info": playlist_info,
|
||||
"track_list": track_list
|
||||
}
|
||||
|
||||
def process_spotify_data(raw_data, data_type):
|
||||
if not raw_data or "error" in raw_data:
|
||||
return {"error": "Invalid data provided"}
|
||||
|
||||
async def main(headless=True):
|
||||
browser = await zd.start(headless=headless)
|
||||
try:
|
||||
track_id = "2plbrEY59IikOBgBGLjaoe"
|
||||
url = f"https://lucida.to/?url=https%3A%2F%2Fopen.spotify.com%2Ftrack%2F{track_id}&country=auto&to=tidal"
|
||||
if data_type == "track":
|
||||
return format_track_data(raw_data)
|
||||
elif data_type == "album":
|
||||
return format_album_data(raw_data)
|
||||
elif data_type == "playlist":
|
||||
return format_playlist_data(raw_data)
|
||||
else:
|
||||
return {"error": "Invalid data type"}
|
||||
except Exception as e:
|
||||
return {"error": f"Error processing data: {str(e)}"}
|
||||
|
||||
page = await browser.get(url)
|
||||
download_info = await get_metadata(page)
|
||||
print(download_info)
|
||||
return download_info
|
||||
finally:
|
||||
await browser.stop()
|
||||
def get_filtered_data(spotify_url, batch=False, delay=1.0):
|
||||
raw_data = get_raw_spotify_data(spotify_url, batch=batch, delay=delay)
|
||||
if raw_data and "error" not in raw_data:
|
||||
url_info = parse_uri(spotify_url)
|
||||
filtered_data = process_spotify_data(raw_data, url_info['type'])
|
||||
return filtered_data
|
||||
return {"error": "Failed to get raw data"}
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
if __name__ == '__main__':
|
||||
playlist = "https://open.spotify.com/playlist/37i9dQZEVXbNG2KDcFcKOF"
|
||||
album = "https://open.spotify.com/album/6J84szYCnMfzEcvIcfWMFL"
|
||||
song = "https://open.spotify.com/track/7so0lgd0zP2Sbgs2d7a1SZ"
|
||||
|
||||
filtered_playlist = get_filtered_data(playlist, batch=True, delay=0.1)
|
||||
print(json.dumps(filtered_playlist, indent=2))
|
||||
|
||||
filtered_album = get_filtered_data(album)
|
||||
print(json.dumps(filtered_album, indent=2))
|
||||
|
||||
filtered_track = get_filtered_data(song)
|
||||
print(json.dumps(filtered_track, indent=2))
|
||||
+966
-47
File diff suppressed because it is too large
Load Diff
+1
-1
@@ -1,3 +1,3 @@
|
||||
{
|
||||
"version": "1.6"
|
||||
"version": "2.9"
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user