Compare commits
51 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 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 |
@@ -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 asyncio
|
||||||
import zendriver as zd
|
import zendriver as zd
|
||||||
|
|
||||||
async def get_metadata(page):
|
async def get_metadata(page, headless=True):
|
||||||
max_attempts = 40
|
max_attempts = 40
|
||||||
attempts = 0
|
attempts = 0
|
||||||
|
|
||||||
@@ -14,9 +14,9 @@ async def get_metadata(page):
|
|||||||
const [url, config] = args;
|
const [url, config] = args;
|
||||||
if (url.includes('/api/load?url=%2Fapi%2Ffetch%2Fstream%2Fv2')) {
|
if (url.includes('/api/load?url=%2Fapi%2Ffetch%2Fstream%2Fv2')) {
|
||||||
const payload = JSON.parse(config.body);
|
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'))
|
const artists = Array.from(document.querySelectorAll('h2.svelte-6pt9ji a.normal'))
|
||||||
.map(a => a.textContent)
|
.map(a => a.textContent.trim())
|
||||||
.join(', ');
|
.join(', ');
|
||||||
const cover = document.querySelector('.svelte-6pt9ji .meta.svelte-6pt9ji a').href;
|
const cover = document.querySelector('.svelte-6pt9ji .meta.svelte-6pt9ji a').href;
|
||||||
|
|
||||||
@@ -82,8 +82,8 @@ async def get_metadata(page):
|
|||||||
|
|
||||||
raise TimeoutError("Timeout")
|
raise TimeoutError("Timeout")
|
||||||
|
|
||||||
async def main():
|
async def main(headless=True):
|
||||||
browser = await zd.start(headless=False)
|
browser = await zd.start(headless=headless)
|
||||||
try:
|
try:
|
||||||
track_id = "2plbrEY59IikOBgBGLjaoe"
|
track_id = "2plbrEY59IikOBgBGLjaoe"
|
||||||
url = f"https://lucida.to/?url=https%3A%2F%2Fopen.spotify.com%2Ftrack%2F{track_id}&country=auto&to=tidal"
|
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)
|
[](https://github.com/afkarxyz/SpotiFLAC/releases)
|
||||||
|
|
||||||

|

|
||||||
|
|
||||||
<div align="center">
|
<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>
|
</div>
|
||||||
|
|
||||||
### [Download](https://github.com/afkarxyz/SpotiFLAC/releases/download/v1.7/SpotiFLAC.exe)
|
### [Download](https://github.com/afkarxyz/SpotiFLAC/releases/download/v2.8/SpotiFLAC.exe)
|
||||||
|
|
||||||
#
|
#
|
||||||
|
|
||||||
> [!NOTE]
|
> [!Note]
|
||||||
> Requires **Google Chrome**
|
**Download speed** from Lucida is unpredictable—sometimes fast, sometimes slow. Join their [Discord](https://discord.com/invite/dXEGRWqEbS) for updates.
|
||||||
|
|
||||||
> [!WARNING]
|
|
||||||
Sometimes, the **download speed** from Lucida can be fast or slow; it varies unpredictably.
|
|
||||||
|
|
||||||
## Screenshots
|
## 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
|
## 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
|
||||||
|
|||||||
+1357
-517
File diff suppressed because it is too large
Load Diff
+459
-80
@@ -1,99 +1,478 @@
|
|||||||
import asyncio
|
from time import sleep
|
||||||
import zendriver as zd
|
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):
|
_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])
|
||||||
max_attempts = 40
|
|
||||||
attempts = 0
|
|
||||||
|
|
||||||
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("""
|
offset = hmac_result[-1] & 15
|
||||||
window.downloadInfo = null;
|
truncated_value = (
|
||||||
const originalFetch = window.fetch;
|
(hmac_result[offset] & 127) << 24
|
||||||
window.fetch = async function(...args) {
|
| (hmac_result[offset + 1] & 255) << 16
|
||||||
const [url, config] = args;
|
| (hmac_result[offset + 2] & 255) << 8
|
||||||
if (url.includes('/api/load?url=%2Fapi%2Ffetch%2Fstream%2Fv2')) {
|
| (hmac_result[offset + 3] & 255)
|
||||||
const payload = JSON.parse(config.body);
|
)
|
||||||
const title = document.querySelector('h1.svelte-6pt9ji').textContent.trim();
|
return (
|
||||||
const artists = Array.from(document.querySelectorAll('h2.svelte-6pt9ji a.normal'))
|
str(truncated_value % (10**digits)).zfill(digits),
|
||||||
.map(a => a.textContent.trim())
|
counter * 30_000,
|
||||||
.join(', ');
|
)
|
||||||
const cover = document.querySelector('.svelte-6pt9ji .meta.svelte-6pt9ji a').href;
|
|
||||||
|
|
||||||
window.downloadInfo = {
|
token_url = 'https://open.spotify.com/get_access_token'
|
||||||
url: payload.url,
|
playlist_base_url = 'https://api.spotify.com/v1/playlists/{}'
|
||||||
cover: cover,
|
album_base_url = 'https://api.spotify.com/v1/albums/{}'
|
||||||
title: title,
|
track_base_url = 'https://api.spotify.com/v1/tracks/{}'
|
||||||
artists: artists,
|
headers = {
|
||||||
token: payload.token.primary,
|
'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',
|
||||||
expiry: payload.token.expiry
|
'Accept': 'application/json',
|
||||||
};
|
'Accept-Language': 'en-US,en;q=0.9',
|
||||||
}
|
'Accept-Encoding': 'gzip, deflate, br',
|
||||||
return originalFetch.apply(this, args);
|
'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("""
|
class SpotifyInvalidUrlException(Exception):
|
||||||
function waitForElement(selector) {
|
pass
|
||||||
return new Promise(resolve => {
|
|
||||||
if (document.querySelector(selector)) {
|
|
||||||
return resolve(document.querySelector(selector));
|
|
||||||
}
|
|
||||||
|
|
||||||
const observer = new MutationObserver(mutations => {
|
class SpotifyWebsiteParserException(Exception):
|
||||||
if (document.querySelector(selector)) {
|
pass
|
||||||
observer.disconnect();
|
|
||||||
resolve(document.querySelector(selector));
|
|
||||||
}
|
|
||||||
});
|
|
||||||
|
|
||||||
observer.observe(document.documentElement, {
|
def parse_uri(uri):
|
||||||
childList: true,
|
u = urlparse(uri)
|
||||||
subtree: true
|
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 () => {
|
req = requests.get(token_url, headers=headers, params=params, timeout=10)
|
||||||
if (!window.location.hostname.includes('lucida.')) return;
|
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([
|
def fetch_tracks_in_batches(url: str, access_token: str, batch_size: int = 100, delay: float = 1.0) -> Tuple[List[Dict[str, Any]], int]:
|
||||||
waitForElement('.d1-track button'),
|
all_tracks = []
|
||||||
waitForElement('button[class*="download-button"]')
|
current_batch = 0
|
||||||
]);
|
|
||||||
|
|
||||||
const clickDownloadButton = () => {
|
while url:
|
||||||
const button = document.querySelector('.d1-track button') ||
|
print(f"Batch : {current_batch}")
|
||||||
document.querySelector('button[class*="download-button"]');
|
|
||||||
if (button) button.click();
|
|
||||||
};
|
|
||||||
|
|
||||||
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:
|
track_data = get_json_from_api(url, access_token)
|
||||||
download_info = await page.evaluate("window.downloadInfo")
|
if not track_data:
|
||||||
if download_info:
|
break
|
||||||
return download_info
|
|
||||||
|
|
||||||
await asyncio.sleep(0.5)
|
items = track_data.get('items', [])
|
||||||
attempts += 1
|
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:
|
try:
|
||||||
track_id = "2plbrEY59IikOBgBGLjaoe"
|
if data_type == "track":
|
||||||
url = f"https://lucida.to/?url=https%3A%2F%2Fopen.spotify.com%2Ftrack%2F{track_id}&country=auto&to=tidal"
|
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)
|
def get_filtered_data(spotify_url, batch=False, delay=1.0):
|
||||||
download_info = await get_metadata(page)
|
raw_data = get_raw_spotify_data(spotify_url, batch=batch, delay=delay)
|
||||||
print(download_info)
|
if raw_data and "error" not in raw_data:
|
||||||
return download_info
|
url_info = parse_uri(spotify_url)
|
||||||
finally:
|
filtered_data = process_spotify_data(raw_data, url_info['type'])
|
||||||
await browser.stop()
|
return filtered_data
|
||||||
|
return {"error": "Failed to get raw data"}
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == '__main__':
|
||||||
asyncio.run(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.7"
|
"version": "2.8"
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user