Compare commits

...

65 Commits

Author SHA1 Message Date
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
afkarxyz 2acd6fcba1 Update v1.6 2025-02-18 09:46:44 +07:00
afkarxyz 85a5bb2321 Update README.md 2025-02-18 09:45:29 +07:00
afkarxyz 70a955f531 Update README.md 2025-02-18 09:10:34 +07:00
afkarxyz 71c8070ec0 Create version.json 2025-02-18 07:58:39 +07:00
afkarxyz 177bc06b79 Update README.md 2025-02-04 14:24:02 +07:00
afkarxyz 2aec9c0185 Update v1.5 2025-02-04 14:22:51 +07:00
afkarxyz a6a84cf869 Update v1.5 2025-02-04 14:22:32 +07:00
afkarxyz 3577574ad8 Update v1.5 2025-02-04 14:22:01 +07:00
afkarxyz 3696fc95a7 Update README.md 2025-02-04 14:21:20 +07:00
afkarxyz effa462810 Update README.md 2025-01-22 12:50:04 +07:00
afkarxyz 921faefecf Update README.md 2025-01-22 05:39:13 +07:00
afkarxyz a4168450d1 Update README.md 2025-01-22 05:35:54 +07:00
afkarxyz 7ba3efb75b Update README.md 2025-01-22 05:34:59 +07:00
10 changed files with 3172 additions and 1086 deletions
-99
View File
@@ -1,99 +0,0 @@
import asyncio
import zendriver as zd
async def get_metadata(page):
max_attempts = 40
attempts = 0
await asyncio.sleep(2)
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;
const artists = Array.from(document.querySelectorAll('h2.svelte-6pt9ji a.normal'))
.map(a => a.textContent)
.join(', ');
const cover = document.querySelector('.svelte-6pt9ji .meta.svelte-6pt9ji a').href;
window.downloadInfo = {
url: payload.url,
cover: cover,
title: title,
artists: artists,
token: payload.token.primary,
expiry: payload.token.expiry
};
}
return originalFetch.apply(this, args);
};
""")
await page.evaluate("""
function waitForElement(selector) {
return new Promise(resolve => {
if (document.querySelector(selector)) {
return resolve(document.querySelector(selector));
}
const observer = new MutationObserver(mutations => {
if (document.querySelector(selector)) {
observer.disconnect();
resolve(document.querySelector(selector));
}
});
observer.observe(document.documentElement, {
childList: true,
subtree: true
});
});
}
(async () => {
if (!window.location.hostname.includes('lucida.')) return;
await Promise.race([
waitForElement('.d1-track button'),
waitForElement('button[class*="download-button"]')
]);
const clickDownloadButton = () => {
const button = document.querySelector('.d1-track button') ||
document.querySelector('button[class*="download-button"]');
if (button) button.click();
};
clickDownloadButton();
})();
""")
while attempts < max_attempts:
download_info = await page.evaluate("window.downloadInfo")
if download_info:
return download_info
await asyncio.sleep(0.5)
attempts += 1
raise TimeoutError("Timeout")
async def main():
browser = await zd.start(headless=False)
try:
track_id = "2plbrEY59IikOBgBGLjaoe"
url = f"https://lucida.to/?url=https%3A%2F%2Fopen.spotify.com%2Ftrack%2F{track_id}&country=auto&to=tidal"
page = await browser.get(url)
download_info = await get_metadata(page)
print(download_info)
return download_info
finally:
await browser.stop()
if __name__ == "__main__":
asyncio.run(main())
-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())
+99 -99
View File
@@ -1,99 +1,99 @@
import asyncio
import zendriver as zd
async def get_metadata(page, headless=True):
max_attempts = 40
attempts = 0
await asyncio.sleep(2)
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;
window.downloadInfo = {
url: payload.url,
cover: cover,
title: title,
artists: artists,
token: payload.token.primary,
expiry: payload.token.expiry
};
}
return originalFetch.apply(this, args);
};
""")
await page.evaluate("""
function waitForElement(selector) {
return new Promise(resolve => {
if (document.querySelector(selector)) {
return resolve(document.querySelector(selector));
}
const observer = new MutationObserver(mutations => {
if (document.querySelector(selector)) {
observer.disconnect();
resolve(document.querySelector(selector));
}
});
observer.observe(document.documentElement, {
childList: true,
subtree: true
});
});
}
(async () => {
if (!window.location.hostname.includes('lucida.')) return;
await Promise.race([
waitForElement('.d1-track button'),
waitForElement('button[class*="download-button"]')
]);
const clickDownloadButton = () => {
const button = document.querySelector('.d1-track button') ||
document.querySelector('button[class*="download-button"]');
if (button) button.click();
};
clickDownloadButton();
})();
""")
while attempts < max_attempts:
download_info = await page.evaluate("window.downloadInfo")
if download_info:
return download_info
await asyncio.sleep(0.5)
attempts += 1
raise TimeoutError("Timeout")
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"
page = await browser.get(url)
download_info = await get_metadata(page)
print(download_info)
return download_info
finally:
await browser.stop()
if __name__ == "__main__":
asyncio.run(main())
import asyncio
import zendriver as zd
async def get_metadata(page, headless=True):
max_attempts = 40
attempts = 0
await asyncio.sleep(2)
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;
window.downloadInfo = {
url: payload.url,
cover: cover,
title: title,
artists: artists,
token: payload.token.primary,
expiry: payload.token.expiry
};
}
return originalFetch.apply(this, args);
};
""")
await page.evaluate("""
function waitForElement(selector) {
return new Promise(resolve => {
if (document.querySelector(selector)) {
return resolve(document.querySelector(selector));
}
const observer = new MutationObserver(mutations => {
if (document.querySelector(selector)) {
observer.disconnect();
resolve(document.querySelector(selector));
}
});
observer.observe(document.documentElement, {
childList: true,
subtree: true
});
});
}
(async () => {
if (!window.location.hostname.includes('lucida.')) return;
await Promise.race([
waitForElement('.d1-track button'),
waitForElement('button[class*="download-button"]')
]);
const clickDownloadButton = () => {
const button = document.querySelector('.d1-track button') ||
document.querySelector('button[class*="download-button"]');
if (button) button.click();
};
clickDownloadButton();
})();
""")
while attempts < max_attempts:
download_info = await page.evaluate("window.downloadInfo")
if download_info:
return download_info
await asyncio.sleep(0.5)
attempts += 1
raise TimeoutError("Timeout")
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"
page = await browser.get(url)
download_info = await get_metadata(page)
print(download_info)
return download_info
finally:
await browser.stop()
if __name__ == "__main__":
asyncio.run(main())
-143
View File
@@ -1,143 +0,0 @@
import requests
import time
import os
import asyncio
from GetMetadata import main as get_metadata
class TrackDownloader:
def __init__(self, use_fallback=False):
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'
}
self.progress_callback = None
self.filename_format = 'title_artist'
self.use_fallback = use_fallback
self.base_domain = "lucida.su" if use_fallback else "lucida.to"
def set_progress_callback(self, callback):
self.progress_callback = callback
def set_filename_format(self, format_type):
self.filename_format = format_type
def generate_filename(self, metadata):
if self.filename_format == 'artist_title':
filename = f"{metadata['artists']} - {metadata['title']}.flac"
else:
filename = f"{metadata['title']} - {metadata['artists']}.flac"
return self.sanitize_filename(filename)
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(f"https://{self.base_domain}/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 = self.generate_filename(metadata)
completion_url = f"https://{server}.{self.base_domain}/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}.{self.base_domain}/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))
downloaded_size = 0
file_path = os.path.join(output_dir, file_name)
try:
with open(file_path, 'wb') as file:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
file.write(chunk)
downloaded_size += len(chunk)
if self.progress_callback:
self.progress_callback(downloaded_size, total_size)
if downloaded_size == 0:
raise Exception("No data received from server")
return file_path
except Exception as e:
if os.path.exists(file_path) and os.path.getsize(file_path) == 0:
try:
os.remove(file_path)
except:
pass
raise e
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())
+20 -16
View File
@@ -1,29 +1,33 @@
[![GitHub All Releases](https://img.shields.io/github/downloads/afkarxyz/SpotifyFLAC/total?style=for-the-badge)](https://github.com/afkarxyz/SpotifyFLAC/releases)
[![GitHub All Releases](https://img.shields.io/github/downloads/afkarxyz/SpotiFLAC/total?style=for-the-badge)](https://github.com/afkarxyz/SpotiFLAC/releases)
![spotifyflac](https://github.com/user-attachments/assets/a11fde95-e756-4592-982f-b567d4a85f3c)
![SpotiFLAC](https://github.com/user-attachments/assets/b4c4f403-edbd-4a71-b74b-c7d433d47d06)
**Spotify FLAC** allows you to download Spotify tracks in true FLAC format through services like Tidal, Amazon Music, Qobuz, and Deezer with the help of Lucida.
<div align="center">
<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>
> [!NOTE]
> Requires **Google Chrome**
### [Download](https://github.com/afkarxyz/SpotiFLAC/releases/download/v2.7/SpotiFLAC.exe)
> [!WARNING]
Sometimes, the **download speed** from Lucida can be fast or slow; it varies unpredictably.
#
#### [Download](https://github.com/afkarxyz/SpotifyFLAC/releases/download/v1.4/SpotifyFLAC.exe) Spotify FLAC
> [!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/c2057543-7f15-470e-beeb-2451a3764d15)
![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` and `Deezer` occasionally experience issues.
![image](https://github.com/user-attachments/assets/9f0d6aa5-456b-4a90-b48a-7e0c22819ebd)
![image](https://github.com/user-attachments/assets/75a61cef-05a8-4f2c-b40b-ba5d49885ffe)
![image](https://github.com/user-attachments/assets/be9a79b5-260e-4948-9f72-10c735210ab7)
![image](https://github.com/user-attachments/assets/84dfcfec-7c9d-4b5b-8624-3558cd3155be)
![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
@@ -31,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/SpotifyFLAC/releases/download/v0/FLAC-Checker.zip) FLAC Checker
#### [Download](https://github.com/afkarxyz/SpotiFLAC/releases/download/v0/FLAC-Checker.zip) FLAC Checker
+1520
View File
File diff suppressed because it is too large Load Diff
-609
View File
@@ -1,609 +0,0 @@
import sys
import asyncio
import os
import time
from pathlib import Path
from PyQt6.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout,
QHBoxLayout, QLabel, QLineEdit, QPushButton,
QProgressBar, QFileDialog, QCheckBox, QRadioButton,
QGroupBox, QComboBox)
from PyQt6.QtCore import QThread, pyqtSignal, Qt, QSettings, QSize
from PyQt6.QtGui import QIcon, QPixmap, QCursor
from GetMetadata import get_metadata
from LucidaDownloader import TrackDownloader
class ImageDownloader(QThread):
finished = pyqtSignal(bytes)
def __init__(self, url):
super().__init__()
self.url = url
def run(self):
import requests
response = requests.get(self.url)
if response.status_code == 200:
self.finished.emit(response.content)
class MetadataFetcher(QThread):
finished = pyqtSignal(dict)
error = pyqtSignal(str)
def __init__(self, url, headless=True, service="tidal", use_fallback=False):
super().__init__()
self.url = url
self.headless_mode = headless
self.service = service
self.use_fallback = use_fallback
self.max_retries = 3
def extract_track_id(self, url):
if "track/" in url:
return url.split("track/")[1].split("?")[0]
return None
async def fetch_metadata(self, track_id):
import zendriver as zd
from asyncio import sleep
domain = "lucida.su" if self.use_fallback else "lucida.to"
for attempt in range(self.max_retries):
try:
lucida_url = f"https://{domain}/?url=https%3A%2F%2Fopen.spotify.com%2Ftrack%2F{track_id}&country=auto&to={self.service}"
browser = await zd.start(headless=self.headless_mode)
try:
page = await browser.get(lucida_url)
return await get_metadata(page)
finally:
await browser.stop()
except Exception as e:
if "refused" in str(e).lower() and attempt < self.max_retries - 1:
await sleep(2 * (attempt + 1))
continue
raise e
def run(self):
try:
track_id = self.extract_track_id(self.url)
if not track_id:
self.error.emit("Invalid Spotify URL")
return
metadata = asyncio.run(self.fetch_metadata(track_id))
if metadata:
self.finished.emit(metadata)
else:
self.error.emit("Failed to fetch track metadata")
except Exception as e:
error_msg = str(e)
if "refused" in error_msg.lower():
self.error.emit("Connection refused. Please check your internet connection and try again.")
elif "timeout" in error_msg.lower():
self.error.emit("Connection timed out. Please check your internet connection and try again.")
else:
self.error.emit(f"Error: {error_msg}")
class DownloaderWorker(QThread):
progress = pyqtSignal(int)
status = pyqtSignal(str)
finished = pyqtSignal(str)
error = pyqtSignal(str)
def __init__(self, metadata, output_dir, filename_format='title_artist', use_fallback=False):
super().__init__()
self.metadata = metadata
self.output_dir = output_dir
self.filename_format = filename_format
self.use_fallback = use_fallback
self.downloader = TrackDownloader(use_fallback=use_fallback)
self.last_update_time = 0
self.last_downloaded_size = 0
def format_size(self, size_bytes):
units = ['B', 'KB', 'MB', 'GB']
index = 0
while size_bytes >= 1024 and index < len(units) - 1:
size_bytes /= 1024
index += 1
return f"{size_bytes:.2f}{units[index]}"
def format_speed(self, speed_bytes):
speed_bits = speed_bytes * 8
if speed_bits >= 1024 * 1024:
speed_mbps = speed_bits / (1024 * 1024)
return f"{speed_mbps:.2f}Mbps"
else:
speed_kbps = speed_bits / 1024
return f"{speed_kbps:.2f}Kbps"
def progress_callback(self, downloaded_size, total_size):
current_time = time.time()
if current_time - self.last_update_time >= 0.5:
progress = int((downloaded_size / total_size) * 100) if total_size > 0 else 0
self.progress.emit(progress)
time_diff = current_time - self.last_update_time
if time_diff > 0:
speed = (downloaded_size - self.last_downloaded_size) / time_diff
status = f"Downloading... {self.format_size(downloaded_size)}/{self.format_size(total_size)} | {self.format_speed(speed)}"
self.status.emit(status)
self.last_update_time = current_time
self.last_downloaded_size = downloaded_size
def run(self):
try:
self.status.emit("Preparing...")
self.downloader.set_progress_callback(self.progress_callback)
self.downloader.set_filename_format(self.filename_format)
self.progress.emit(0)
downloaded_file = self.downloader.download(self.metadata, self.output_dir)
self.progress.emit(100)
self.finished.emit("Download complete!")
except Exception as e:
self.error.emit(f"Error: {str(e)}")
class ServiceComboBox(QComboBox):
def __init__(self, parent=None):
super().__init__(parent)
self.setIconSize(QSize(16, 16))
self.setup_items()
def setup_items(self):
current_dir = os.path.dirname(os.path.abspath(__file__))
icons_dir = os.path.join(current_dir, 'icons')
if not os.path.exists(icons_dir):
os.makedirs(icons_dir)
services = [
{'id': 'tidal', 'name': 'Tidal', 'icon': 'tidal.png'},
{'id': 'amazon', 'name': 'Amazon Music', 'icon': 'amazon.png'},
{'id': 'qobuz', 'name': 'Qobuz', 'icon': 'qobuz.png'},
{'id': 'deezer', 'name': 'Deezer', 'icon': 'deezer.png'}
]
for service in services:
icon_path = os.path.join(icons_dir, service['icon'])
if not os.path.exists(icon_path):
self.create_placeholder_icon(icon_path)
icon = QIcon(icon_path)
self.addItem(icon, service['name'], service['id'])
def create_placeholder_icon(self, path):
pixmap = QPixmap(16, 16)
pixmap.fill(Qt.GlobalColor.transparent)
pixmap.save(path)
class SpotifyFlacGUI(QMainWindow):
def __init__(self):
super().__init__()
self.settings = QSettings('SpotifyFlac', 'Settings')
self.setWindowTitle("Spotify FLAC")
icon_path = os.path.join(os.path.dirname(__file__), "icon.svg")
if os.path.exists(icon_path):
self.setWindowIcon(QIcon(icon_path))
self.setFixedWidth(600)
self.setFixedHeight(180)
self.default_music_dir = str(Path.home() / "Music")
if not os.path.exists(self.default_music_dir):
os.makedirs(self.default_music_dir)
self.metadata = None
self.init_ui()
self.url_input.textChanged.connect(self.validate_url)
self.load_settings()
self.setup_settings_persistence()
def load_settings(self):
headless = self.settings.value('headless', True, type=bool)
fallback = self.settings.value('fallback', False, type=bool)
service = self.settings.value('service', 'tidal')
format_type = self.settings.value('format', 'title_artist')
output_dir = self.settings.value('output_dir', self.default_music_dir)
self.headless_checkbox.setChecked(headless)
self.fallback_checkbox.setChecked(fallback)
for i in range(self.service_combo.count()):
if self.service_combo.itemData(i) == service:
self.service_combo.setCurrentIndex(i)
break
self.format_title_artist.setChecked(format_type == 'title_artist')
self.format_artist_title.setChecked(format_type == 'artist_title')
self.dir_input.setText(output_dir)
def setup_settings_persistence(self):
self.headless_checkbox.stateChanged.connect(
lambda x: self.settings.setValue('headless', bool(x)))
self.fallback_checkbox.stateChanged.connect(
lambda x: self.settings.setValue('fallback', bool(x)))
self.service_combo.currentIndexChanged.connect(
lambda i: self.settings.setValue('service', self.service_combo.itemData(i)))
self.format_title_artist.toggled.connect(
lambda x: self.settings.setValue('format', 'title_artist' if x else 'artist_title'))
self.dir_input.textChanged.connect(
lambda x: self.settings.setValue('output_dir', x))
def init_ui(self):
central_widget = QWidget()
self.setCentralWidget(central_widget)
self.main_layout = QVBoxLayout(central_widget)
self.main_layout.setContentsMargins(10, 10, 10, 10)
self.input_widget = QWidget()
input_layout = QVBoxLayout(self.input_widget)
input_layout.setSpacing(10)
url_layout = QHBoxLayout()
url_label = QLabel("Track URL:")
url_label.setFixedWidth(100)
self.url_input = QLineEdit()
self.url_input.setPlaceholderText("Please enter track URL")
self.url_input.setClearButtonEnabled(True)
self.fetch_button = QPushButton("Fetch")
self.fetch_button.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.fetch_button.setFixedWidth(100)
self.fetch_button.setEnabled(False)
self.fetch_button.clicked.connect(self.fetch_track_info)
url_layout.addWidget(url_label)
url_layout.addWidget(self.url_input)
url_layout.addWidget(self.fetch_button)
input_layout.addLayout(url_layout)
dir_layout = QHBoxLayout()
dir_label = QLabel("Output Directory:")
dir_label.setFixedWidth(100)
self.dir_input = QLineEdit(self.default_music_dir)
self.dir_button = QPushButton("Browse")
self.dir_button.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.dir_button.setFixedWidth(100)
dir_layout.addWidget(dir_label)
dir_layout.addWidget(self.dir_input)
dir_layout.addWidget(self.dir_button)
self.dir_button.clicked.connect(self.select_directory)
input_layout.addLayout(dir_layout)
settings_group = QGroupBox("Settings")
settings_layout = QHBoxLayout(settings_group)
settings_layout.setContentsMargins(10, 0, 10, 10)
settings_layout.setSpacing(10)
settings_container = QWidget()
settings_container_layout = QHBoxLayout(settings_container)
settings_container_layout.setContentsMargins(0, 0, 0, 0)
settings_container_layout.setSpacing(10)
self.headless_checkbox = QCheckBox("Headless")
self.headless_checkbox.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.headless_checkbox.setChecked(True)
settings_container_layout.addWidget(self.headless_checkbox)
self.fallback_checkbox = QCheckBox("Fallback")
self.fallback_checkbox.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.fallback_checkbox.setChecked(False)
settings_container_layout.addWidget(self.fallback_checkbox)
service_widget = QWidget()
service_layout = QHBoxLayout(service_widget)
service_layout.setContentsMargins(0, 0, 0, 0)
service_layout.setSpacing(10)
service_label = QLabel("Service:")
self.service_combo = ServiceComboBox()
self.service_combo.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
service_layout.addWidget(service_label)
service_layout.addWidget(self.service_combo)
settings_container_layout.addWidget(service_widget)
format_widget = QWidget()
format_layout = QHBoxLayout(format_widget)
format_layout.setContentsMargins(0, 0, 0, 0)
format_layout.setSpacing(10)
format_label = QLabel("Filename:")
self.format_title_artist = QRadioButton("Title")
self.format_artist_title = QRadioButton("Artist")
self.format_title_artist.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.format_artist_title.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.format_title_artist.setChecked(True)
format_layout.addWidget(format_label)
format_layout.addWidget(self.format_title_artist)
format_layout.addWidget(self.format_artist_title)
settings_container_layout.addWidget(format_widget)
settings_layout.addStretch()
settings_layout.addWidget(settings_container)
settings_layout.addStretch()
input_layout.addWidget(settings_group)
self.main_layout.addWidget(self.input_widget)
self.track_widget = QWidget()
self.track_widget.hide()
track_layout = QHBoxLayout(self.track_widget)
track_layout.setContentsMargins(0, 0, 0, 0)
track_layout.setSpacing(10)
cover_container = QWidget()
cover_layout = QVBoxLayout(cover_container)
cover_layout.setContentsMargins(0, 0, 0, 0)
cover_layout.setAlignment(Qt.AlignmentFlag.AlignTop)
self.cover_label = QLabel()
self.cover_label.setFixedSize(100, 100)
self.cover_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
cover_layout.addWidget(self.cover_label)
track_layout.addWidget(cover_container)
track_details_container = QWidget()
track_details_layout = QVBoxLayout(track_details_container)
track_details_layout.setContentsMargins(0, 0, 0, 0)
track_details_layout.setSpacing(2)
track_details_layout.setAlignment(Qt.AlignmentFlag.AlignTop)
self.title_label = QLabel()
self.title_label.setStyleSheet("font-size: 14px; font-weight: bold;")
self.title_label.setWordWrap(True)
self.title_label.setMinimumWidth(400)
self.artist_label = QLabel()
self.artist_label.setStyleSheet("font-size: 12px;")
self.artist_label.setWordWrap(True)
self.artist_label.setMinimumWidth(400)
track_details_layout.addWidget(self.title_label)
track_details_layout.addWidget(self.artist_label)
track_layout.addWidget(track_details_container, stretch=1)
track_layout.addStretch()
self.main_layout.addWidget(self.track_widget)
self.download_button = QPushButton("Download")
self.download_button.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.download_button.setFixedWidth(100)
self.download_button.clicked.connect(self.button_clicked)
self.download_button.hide()
self.cancel_button = QPushButton("Cancel")
self.cancel_button.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.cancel_button.setFixedWidth(100)
self.cancel_button.clicked.connect(self.cancel_clicked)
self.cancel_button.hide()
self.open_button = QPushButton("Open")
self.open_button.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.open_button.setFixedWidth(100)
self.open_button.clicked.connect(self.open_output_directory)
self.open_button.hide()
download_layout = QHBoxLayout()
download_layout.addStretch()
download_layout.addWidget(self.open_button)
download_layout.addWidget(self.download_button)
download_layout.addWidget(self.cancel_button)
download_layout.addStretch()
self.main_layout.addLayout(download_layout)
self.progress_bar = QProgressBar()
self.progress_bar.hide()
self.main_layout.addWidget(self.progress_bar)
bottom_layout = QHBoxLayout()
self.status_label = QLabel("")
bottom_layout.addWidget(self.status_label, stretch=1)
self.update_button = QPushButton()
icon_path = os.path.join(os.path.dirname(__file__), "update.svg")
if os.path.exists(icon_path):
self.update_button.setIcon(QIcon(icon_path))
self.update_button.setFixedSize(16, 16)
self.update_button.setStyleSheet("""
QPushButton {
border: none;
background: transparent;
}
QPushButton:hover {
background: transparent;
}
""")
self.update_button.setCursor(QCursor(Qt.CursorShape.PointingHandCursor))
self.update_button.setToolTip("Check for Updates")
self.update_button.clicked.connect(self.open_update_page)
bottom_layout.addWidget(self.update_button)
self.main_layout.addLayout(bottom_layout)
def open_update_page(self):
import webbrowser
webbrowser.open('https://github.com/afkarxyz/SpotifyFLAC/releases')
def validate_url(self, url):
url = url.strip()
self.fetch_button.setEnabled(False)
if not url:
self.status_label.clear()
return
if "open.spotify.com/" not in url:
self.status_label.setText("Please enter a valid Spotify URL")
return
if "/album/" in url:
self.status_label.setText("Album URLs are not supported. Please enter a track URL.")
return
if "/playlist/" in url:
self.status_label.setText("Playlist URLs are not supported. Please enter a track URL.")
return
if "/track/" not in url:
self.status_label.setText("Please enter a valid Spotify track URL")
return
self.fetch_button.setEnabled(True)
self.status_label.clear()
def fetch_track_info(self):
url = self.url_input.text().strip()
if not url:
self.status_label.setText("Please enter a Track URL")
return
self.fetch_button.setEnabled(False)
self.status_label.setText("Fetching track information...")
headless = self.headless_checkbox.isChecked()
fallback = self.fallback_checkbox.isChecked()
service = self.service_combo.currentData()
self.fetcher = MetadataFetcher(url, headless=headless, service=service, use_fallback=fallback)
self.fetcher.finished.connect(self.handle_track_info)
self.fetcher.error.connect(self.handle_fetch_error)
self.fetcher.start()
def handle_track_info(self, metadata):
self.metadata = metadata
self.fetch_button.setEnabled(True)
self.title_label.setText(metadata['title'].strip())
self.artist_label.setText(metadata['artists'].strip())
self.image_downloader = ImageDownloader(metadata['cover'])
self.image_downloader.finished.connect(self.update_cover_art)
self.image_downloader.start()
self.input_widget.hide()
self.track_widget.show()
self.download_button.show()
self.cancel_button.show()
self.update_button.hide()
self.status_label.clear()
self.adjustWindowHeight()
def adjustWindowHeight(self):
title_height = self.title_label.sizeHint().height()
artist_height = self.artist_label.sizeHint().height()
base_height = 180
additional_height = max(0, (title_height + artist_height) - 40)
new_height = min(300, base_height + additional_height)
self.setFixedHeight(int(new_height))
def update_cover_art(self, image_data):
pixmap = QPixmap()
pixmap.loadFromData(image_data)
scaled_pixmap = pixmap.scaled(100, 100, Qt.AspectRatioMode.KeepAspectRatio, Qt.TransformationMode.SmoothTransformation)
self.cover_label.setPixmap(scaled_pixmap)
def handle_fetch_error(self, error):
self.fetch_button.setEnabled(True)
self.status_label.setText(f"Error fetching track info: {error}")
def select_directory(self):
directory = QFileDialog.getExistingDirectory(self, "Select Output Directory")
if directory:
self.dir_input.setText(directory)
def open_output_directory(self):
output_dir = self.dir_input.text().strip() or self.default_music_dir
os.startfile(output_dir)
def cancel_clicked(self):
self.track_widget.hide()
self.input_widget.show()
self.download_button.hide()
self.cancel_button.hide()
self.progress_bar.hide()
self.progress_bar.setValue(0)
self.status_label.clear()
self.metadata = None
self.fetch_button.setEnabled(True)
self.update_button.show()
self.setFixedHeight(180)
def button_clicked(self):
if self.download_button.text() == "Clear":
self.clear_form()
else:
self.start_download()
def clear_form(self):
self.url_input.clear()
self.progress_bar.hide()
self.progress_bar.setValue(0)
self.status_label.clear()
self.download_button.setText("Download")
self.download_button.hide()
self.cancel_button.hide()
self.open_button.hide()
self.track_widget.hide()
self.input_widget.show()
self.metadata = None
self.update_button.show()
self.setFixedHeight(180)
def start_download(self):
output_dir = self.dir_input.text().strip()
if not self.metadata:
self.status_label.setText("Please fetch track information first")
return
if not output_dir:
output_dir = self.default_music_dir
self.dir_input.setText(output_dir)
self.download_button.hide()
self.cancel_button.hide()
self.progress_bar.show()
self.progress_bar.setValue(0)
self.status_label.setText("Preparing...")
format_type = 'artist_title' if self.format_artist_title.isChecked() else 'title_artist'
fallback = self.fallback_checkbox.isChecked()
self.worker = DownloaderWorker(
metadata=self.metadata,
output_dir=output_dir,
filename_format=format_type,
use_fallback=fallback
)
self.worker.progress.connect(self.update_progress)
self.worker.status.connect(self.update_status)
self.worker.finished.connect(self.download_finished)
self.worker.error.connect(self.download_error)
self.worker.start()
def update_status(self, status):
self.status_label.setText(status)
def update_progress(self, value):
self.progress_bar.setValue(value)
def download_finished(self, message):
self.progress_bar.hide()
self.status_label.setText(message)
self.open_button.show()
self.download_button.setText("Clear")
self.download_button.show()
self.cancel_button.hide()
self.download_button.setEnabled(True)
def download_error(self, error_message):
self.progress_bar.hide()
self.status_label.setText(error_message)
self.download_button.setText("Retry")
self.download_button.show()
self.cancel_button.show()
self.download_button.setEnabled(True)
self.cancel_button.setEnabled(True)
def main():
app = QApplication(sys.argv)
window = SpotifyFlacGUI()
window.show()
sys.exit(app.exec())
if __name__ == "__main__":
main()
+478
View File
@@ -0,0 +1,478 @@
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
_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])
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()
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,
)
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'
}
class SpotifyInvalidUrlException(Exception):
pass
class SpotifyWebsiteParserException(Exception):
pass
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,
}
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))
+1052
View File
File diff suppressed because it is too large Load Diff
+3
View File
@@ -0,0 +1,3 @@
{
"version": "2.7"
}