Compare commits

...

56 Commits

Author SHA1 Message Date
afkarxyz 3ac4c34d73 v3.0 2025-05-31 19:33:16 +07:00
afkarxyz 88e303cbe4 v2.9 2025-05-30 22:28:24 +07:00
afkarxyz c13855fadd v2.9 2025-05-30 22:22:14 +07:00
afkarxyz 2b12684960 v2.8 2025-05-23 16:47:38 +07:00
afkarxyz 4bc164cc56 v2.8 2025-05-23 16:43:45 +07:00
afkarxyz 46cb65665e Update README.md 2025-05-19 10:33:04 +07:00
afkarxyz 276b3b4951 Update README.md 2025-05-13 20:13:37 +07:00
afkarxyz e15aadbd61 Update README.md 2025-05-13 20:11:35 +07:00
afkarxyz d7639bae8f v2.7 2025-05-13 20:11:32 +07:00
afkarxyz 1af7ab65c9 v2.7 2025-05-13 20:07:19 +07:00
afkarxyz c5240596cb Revert 2025-05-13 12:06:55 +07:00
afkarxyz c4a9042adc v2.9 2025-05-12 00:08:55 +07:00
afkarxyz 45ac08ecbd v2.9 2025-05-12 00:05:34 +07:00
afkarxyz 0add305d9c v2.8 2025-05-11 18:34:58 +07:00
afkarxyz 9b6b43c0a4 v2.8 2025-05-11 18:31:28 +07:00
afkarxyz 60d20cbebe Revert 2025-05-11 17:03:23 +07:00
afkarxyz 626d58667e v2.8 2025-05-11 16:32:26 +07:00
afkarxyz 4dd1a7ea12 v2.8 2025-05-11 15:58:46 +07:00
afkarxyz 67964e4acb Update README.md 2025-05-11 04:40:22 +07:00
afkarxyz 1486fb13df v2.7 2025-05-10 20:33:17 +07:00
afkarxyz 966536f127 v2.7 2025-05-10 20:13:01 +07:00
afkarxyz 21946321f5 Update README.md 2025-05-06 13:25:27 +07:00
afkarxyz 3e3cb0610d Update README.md 2025-05-06 12:45:25 +07:00
afkarxyz 160eba0987 Update README.md 2025-05-06 12:43:29 +07:00
afkarxyz 71a60ded47 Update README.md 2025-05-06 10:42:11 +07:00
afkarxyz e0a0514df9 v2.6 2025-05-06 10:41:11 +07:00
afkarxyz 1e7a48d263 v2.6 2025-05-06 10:38:01 +07:00
afkarxyz 0a83a0dd6e Update README.md 2025-05-06 10:35:17 +07:00
afkarxyz da429d9410 v2.5 2025-04-25 06:33:32 +07:00
afkarxyz 63211c726b v2.5 2025-04-25 06:30:14 +07:00
afkarxyz 055cb6991a Update v2.4 2025-04-08 13:10:12 +07:00
afkarxyz 222d681551 Update v2.4 2025-04-08 13:07:26 +07:00
afkarxyz 479c6ede2b Update v2.3 2025-03-20 05:47:20 +07:00
afkarxyz ceb727adb9 Update v2.3 2025-03-20 05:41:28 +07:00
afkarxyz bbea8ca493 Update README.md 2025-03-20 05:36:38 +07:00
afkarxyz f567dd19bf Update v2.2 2025-03-18 08:01:11 +07:00
afkarxyz 3651833e2a Update v2.2 2025-03-18 07:56:51 +07:00
afkarxyz 8403b96306 Update README.md 2025-03-18 07:55:16 +07:00
afkarxyz d977829e36 Update v2.1 2025-03-02 15:21:25 +07:00
afkarxyz 2aaf123c98 Update v2.1 2025-03-02 15:21:14 +07:00
afkarxyz 13567802a0 Update v2.1 2025-03-02 15:18:04 +07:00
afkarxyz d704782519 Update v2.0 2025-03-02 14:52:04 +07:00
afkarxyz 884c02278f Update v2.0 2025-03-02 14:51:46 +07:00
afkarxyz d6abe2bae3 Update v2.0 2025-03-02 14:48:15 +07:00
afkarxyz 7b858dd0ce Update v2.0 2025-03-02 14:47:55 +07:00
afkarxyz cfeb9a2ef2 Update v1.9 2025-03-02 14:05:17 +07:00
afkarxyz 81a78832ff Update v1.9 2025-03-02 14:02:16 +07:00
afkarxyz 071f20deff Update README.md 2025-03-02 14:01:46 +07:00
afkarxyz 03de68ac7b Rename to getMetadata.py 2025-03-02 06:45:20 +07:00
afkarxyz 77363f9e61 Update README.md 2025-03-02 06:33:53 +07:00
afkarxyz 2df77120cf Update v1.8 2025-03-02 06:30:12 +07:00
afkarxyz e6e953b2ed Update README.md 2025-03-02 06:18:19 +07:00
afkarxyz 72f17479e8 Update getTracks.py 2025-02-19 15:48:43 +07:00
afkarxyz 9286fba63c Update README.md 2025-02-19 09:42:23 +07:00
afkarxyz 6bf7084959 Update version.json 2025-02-19 09:42:11 +07:00
afkarxyz 03cc3d82a7 Update v1.7 2025-02-19 09:41:58 +07:00
7 changed files with 2862 additions and 841 deletions
-120
View File
@@ -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"
+16 -16
View File
@@ -1,33 +1,33 @@
[![GitHub All Releases](https://img.shields.io/github/downloads/afkarxyz/SpotiFLAC/total?style=for-the-badge)](https://github.com/afkarxyz/SpotiFLAC/releases)
![spotiflac](https://github.com/user-attachments/assets/a233a276-14a4-4f4c-b267-f182dd3912a0)
![SpotiFLAC](https://github.com/user-attachments/assets/b4c4f403-edbd-4a71-b74b-c7d433d47d06)
<div align="center">
<b>SpotiFLAC</b> allows you to download Spotify tracks in true FLAC format through services like Tidal, Amazon Music and 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
![image](https://github.com/user-attachments/assets/5f36d815-da1d-4fc1-b85e-e7c1fe5c9842)
![image](https://github.com/user-attachments/assets/70a5dceb-3374-4255-8f6a-4afb5ee534b0)
> - 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.
![image](https://github.com/user-attachments/assets/9f0d6aa5-456b-4a90-b48a-7e0c22819ebd)
![image](https://github.com/user-attachments/assets/cf45ba1b-f048-4284-8d90-dc90d2c37745)
![image](https://github.com/user-attachments/assets/be9a79b5-260e-4948-9f72-10c735210ab7)
![image](https://github.com/user-attachments/assets/adc7c2fe-9758-4371-b186-d690bd06e3b0)
![image](https://github.com/user-attachments/assets/c4403934-9003-447e-a27b-fc74cab23454)
![image](https://github.com/user-attachments/assets/1feec621-f8bf-4b2a-ae73-afcb1fb1deba)
![image](https://github.com/user-attachments/assets/c64b9a08-c99a-4d3a-ae8b-5f834623915b)
> 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
![image](https://github.com/user-attachments/assets/7649e6e1-d5d1-49b3-b83f-965d44651d05)
### [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
+1363 -520
View File
File diff suppressed because it is too large Load Diff
+460 -81
View File
@@ -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])
(async () => {
if (!window.location.hostname.includes('lucida.')) return;
if not u.scheme and not u.netloc:
return {"type": "playlist", "id": u.path}
await Promise.race([
waitForElement('.d1-track button'),
waitForElement('button[class*="download-button"]')
]);
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("/")
const clickDownloadButton = () => {
const button = document.querySelector('.d1-track button') ||
document.querySelector('button[class*="download-button"]');
if (button) button.click();
};
if parts[1] == "embed":
parts = parts[1:]
clickDownloadButton();
})();
""")
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]}
while attempts < max_attempts:
download_info = await page.evaluate("window.downloadInfo")
if download_info:
return download_info
raise SpotifyInvalidUrlException("ERROR: unable to determine Spotify URL type or type is unsupported.")
await asyncio.sleep(0.5)
attempts += 1
def get_json_from_api(api_url, access_token):
headers.update({'Authorization': 'Bearer {}'.format(access_token)})
raise TimeoutError("Timeout")
req = requests.get(api_url, headers=headers, timeout=10)
async def main(headless=True):
browser = await zd.start(headless=headless)
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:
track_id = "2plbrEY59IikOBgBGLjaoe"
url = f"https://lucida.to/?url=https%3A%2F%2Fopen.spotify.com%2Ftrack%2F{track_id}&country=auto&to=tidal"
totp, timestamp = generate_totp()
page = await browser.get(url)
download_info = await get_metadata(page)
print(download_info)
return download_info
finally:
await browser.stop()
params = {
"reason": "init",
"productType": "web-player",
"totp": totp,
"totpVer": 5,
"ts": timestamp,
}
if __name__ == "__main__":
asyncio.run(main())
req = requests.get(token_url, headers=headers, params=params, timeout=10)
if req.status_code != 200:
return {"error": f"Failed to get access token. Status code: {req.status_code}"}
return req.json()
except Exception as e:
return {"error": f"Failed to get access token: {str(e)}"}
def fetch_tracks_in_batches(url: str, access_token: str, batch_size: int = 100, delay: float = 1.0) -> Tuple[List[Dict[str, Any]], int]:
all_tracks = []
current_batch = 0
while url:
print(f"Batch : {current_batch}")
url_parts = url.split("offset=")
if len(url_parts) > 1:
offset_part = url_parts[1].split("&")[0]
print(f"Offset : {offset_part}")
print("-------------")
track_data = get_json_from_api(url, access_token)
if not track_data:
break
items = track_data.get('items', [])
all_tracks.extend(items)
url = track_data.get('next')
if url and "&locale=" in url:
url = url.split("&locale=")[0]
if url and delay > 0:
sleep(delay)
current_batch += 1
return all_tracks, current_batch
def get_raw_spotify_data(spotify_url, batch: bool = False, delay: float = 1.0):
url_info = parse_uri(spotify_url)
token = get_access_token()
if "error" in token:
return token
access_token = token["accessToken"]
raw_data = {}
if url_info['type'] == "playlist":
try:
playlist_data = get_json_from_api(
playlist_base_url.format(url_info["id"]),
access_token
)
if not playlist_data:
return {"error": "Failed to get playlist data"}
raw_data = playlist_data
total_tracks = playlist_data.get('tracks', {}).get('total', 0)
if batch:
tracks_url = f'https://api.spotify.com/v1/playlists/{url_info["id"]}/tracks?limit=100'
tracks, num_batches = fetch_tracks_in_batches(tracks_url, access_token, 100, delay)
raw_data['tracks']['items'] = tracks
raw_data['_batch_count'] = num_batches
raw_data['_batch_enabled'] = True
if len(tracks) < total_tracks:
last_offset = len(tracks)
remaining_tracks = []
while last_offset < total_tracks:
print(f"Batch : {num_batches}")
print(f"Offset : {last_offset}")
print("-------------")
remainder_url = f'https://api.spotify.com/v1/playlists/{url_info["id"]}/tracks?offset={last_offset}&limit=100'
track_data = get_json_from_api(remainder_url, access_token)
if not track_data or not track_data.get('items'):
break
items = track_data.get('items', [])
remaining_tracks.extend(items)
if len(items) < 100:
break
last_offset += len(items)
num_batches += 1
if delay > 0:
sleep(delay)
tracks.extend(remaining_tracks)
raw_data['tracks']['items'] = tracks
raw_data['_batch_count'] = num_batches
else:
tracks = []
tracks_url = f'https://api.spotify.com/v1/playlists/{url_info["id"]}/tracks?limit=100'
while tracks_url:
track_data = get_json_from_api(tracks_url, access_token)
if not track_data:
break
tracks.extend(track_data['items'])
tracks_url = track_data.get('next')
if tracks_url and "&locale=" in tracks_url:
tracks_url = tracks_url.split("&locale=")[0]
raw_data['tracks']['items'] = tracks
raw_data['_batch_enabled'] = False
except Exception as e:
return {"error": f"Failed to get playlist data: {str(e)}"}
elif url_info["type"] == "album":
try:
album_data = get_json_from_api(
album_base_url.format(url_info["id"]),
access_token
)
if not album_data:
return {"error": "Failed to get album data"}
album_data['_token'] = access_token
raw_data = album_data
total_tracks = album_data.get('total_tracks', 0)
if batch:
tracks_url = f'{album_base_url.format(url_info["id"])}/tracks?limit=50'
tracks, num_batches = fetch_tracks_in_batches(tracks_url, access_token, 50, delay)
raw_data['tracks']['items'] = tracks
raw_data['_batch_count'] = num_batches
raw_data['_batch_enabled'] = True
if len(tracks) < total_tracks:
last_offset = len(tracks)
remaining_tracks = []
while last_offset < total_tracks:
print(f"Batch : {num_batches}")
print(f"Offset : {last_offset}")
print("-------------")
remainder_url = f'{album_base_url.format(url_info["id"])}/tracks?offset={last_offset}&limit=50'
track_data = get_json_from_api(remainder_url, access_token)
if not track_data or not track_data.get('items'):
break
items = track_data.get('items', [])
remaining_tracks.extend(items)
if len(items) < 50:
break
last_offset += len(items)
num_batches += 1
if delay > 0:
sleep(delay)
tracks.extend(remaining_tracks)
raw_data['tracks']['items'] = tracks
raw_data['_batch_count'] = num_batches
else:
tracks = []
tracks_url = f'{album_base_url.format(url_info["id"])}/tracks?limit=50'
while tracks_url:
track_data = get_json_from_api(tracks_url, access_token)
if not track_data:
break
tracks.extend(track_data['items'])
tracks_url = track_data.get('next')
if tracks_url and "&locale=" in tracks_url:
tracks_url = tracks_url.split("&locale=")[0]
raw_data['tracks']['items'] = tracks
raw_data['_batch_enabled'] = False
except Exception as e:
return {"error": f"Failed to get album data: {str(e)}"}
elif url_info["type"] == "track":
try:
track_data = get_json_from_api(
track_base_url.format(url_info["id"]),
access_token
)
if not track_data:
return {"error": "Failed to get track data"}
raw_data = track_data
except Exception as e:
return {"error": f"Failed to get track data: {str(e)}"}
return raw_data
def format_track_data(track_data):
artists = []
for artist in track_data.get('artists', []):
artists.append(artist['name'])
image_url = track_data.get('album', {}).get('images', [{}])[0].get('url', '') if track_data.get('album', {}).get('images') else ''
isrc = track_data.get('external_ids', {}).get('isrc', '')
return {
"track": {
"artists": ", ".join(artists),
"name": track_data.get('name', ''),
"album_name": track_data.get('album', {}).get('name', ''),
"duration_ms": track_data.get('duration_ms', 0),
"images": image_url,
"release_date": track_data.get('album', {}).get('release_date', ''),
"track_number": track_data.get('track_number', 0),
"external_urls": track_data.get('external_urls', {}).get('spotify', ''),
"isrc": isrc
}
}
def format_album_data(album_data):
artists = []
for artist in album_data.get('artists', []):
artists.append(artist['name'])
image_url = album_data.get('images', [{}])[0].get('url', '') if album_data.get('images') else ''
track_list = []
for track in album_data.get('tracks', {}).get('items', []):
track_artists = []
for artist in track.get('artists', []):
track_artists.append(artist['name'])
track_id = track.get('id', '')
track_isrc = ''
if track_id and album_data.get('_token'):
try:
full_track_data = get_json_from_api(
track_base_url.format(track_id),
album_data.get('_token')
)
if full_track_data:
track_isrc = full_track_data.get('external_ids', {}).get('isrc', '')
except:
pass
track_list.append({
"artists": ", ".join(track_artists),
"name": track.get('name', ''),
"album_name": album_data.get('name', ''),
"duration_ms": track.get('duration_ms', 0),
"images": image_url,
"release_date": album_data.get('release_date', ''),
"track_number": track.get('track_number', 0),
"external_urls": track.get('external_urls', {}).get('spotify', ''),
"isrc": track_isrc
})
album_info = {
"total_tracks": album_data.get('total_tracks', 0),
"name": album_data.get('name', ''),
"release_date": album_data.get('release_date', ''),
"artists": ", ".join(artists),
"images": image_url
}
if album_data.get('_batch_enabled', False):
album_info["batch"] = f"{album_data.get('_batch_count', 1)}"
return {
"album_info": album_info,
"track_list": track_list
}
def format_playlist_data(playlist_data):
image_url = playlist_data.get('images', [{}])[0].get('url', '') if playlist_data.get('images') else ''
track_list = []
for item in playlist_data.get('tracks', {}).get('items', []):
track = item.get('track', {})
if not track:
continue
artists = []
for artist in track.get('artists', []):
artists.append(artist['name'])
track_image = ''
if track.get('album', {}).get('images'):
track_image = track.get('album', {}).get('images', [{}])[0].get('url', '')
track_isrc = track.get('external_ids', {}).get('isrc', '')
track_list.append({
"artists": ", ".join(artists),
"name": track.get('name', ''),
"album_name": track.get('album', {}).get('name', ''),
"duration_ms": track.get('duration_ms', 0),
"images": track_image,
"release_date": track.get('album', {}).get('release_date', ''),
"track_number": track.get('track_number', 0),
"external_urls": track.get('external_urls', {}).get('spotify', ''),
"isrc": track_isrc
})
playlist_info = {
"tracks": {"total": playlist_data.get('tracks', {}).get('total', 0)},
"followers": {"total": playlist_data.get('followers', {}).get('total', 0)},
"owner": {
"display_name": playlist_data.get('owner', {}).get('display_name', ''),
"name": playlist_data.get('name', ''),
"images": image_url
}
}
if playlist_data.get('_batch_enabled', False):
playlist_info["batch"] = f"{playlist_data.get('_batch_count', 1)}"
return {
"playlist_info": playlist_info,
"track_list": track_list
}
def process_spotify_data(raw_data, data_type):
if not raw_data or "error" in raw_data:
return {"error": "Invalid data provided"}
try:
if data_type == "track":
return format_track_data(raw_data)
elif data_type == "album":
return format_album_data(raw_data)
elif data_type == "playlist":
return format_playlist_data(raw_data)
else:
return {"error": "Invalid data type"}
except Exception as e:
return {"error": f"Error processing data: {str(e)}"}
def get_filtered_data(spotify_url, batch=False, delay=1.0):
raw_data = get_raw_spotify_data(spotify_url, batch=batch, delay=delay)
if raw_data and "error" not in raw_data:
url_info = parse_uri(spotify_url)
filtered_data = process_spotify_data(raw_data, url_info['type'])
return filtered_data
return {"error": "Failed to get raw data"}
if __name__ == '__main__':
playlist = "https://open.spotify.com/playlist/37i9dQZEVXbNG2KDcFcKOF"
album = "https://open.spotify.com/album/6J84szYCnMfzEcvIcfWMFL"
song = "https://open.spotify.com/track/7so0lgd0zP2Sbgs2d7a1SZ"
filtered_playlist = get_filtered_data(playlist, batch=True, delay=0.1)
print(json.dumps(filtered_playlist, indent=2))
filtered_album = get_filtered_data(album)
print(json.dumps(filtered_album, indent=2))
filtered_track = get_filtered_data(song)
print(json.dumps(filtered_track, indent=2))
+966 -47
View File
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -1,3 +1,3 @@
{
"version": "1.6"
"version": "2.9"
}