admin/SpotiFLAC
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v2.7

Browse Source
This commit is contained in:
afkarxyz
2025-05-10 20:13:01 +07:00
parent 21946321f5
commit 966536f127
3 changed files with 468 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
SpotiFLAC.py
+178 -28
View File
@@ -5,19 +5,21 @@ from datetime import datetime
import requests
import re
from packaging import version
import json
from PyQt6.QtWidgets import (
QApplication, QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLineEdit,
QLabel, QFileDialog, QListWidget, QTextEdit, QTabWidget, QButtonGroup, QRadioButton,
QAbstractItemView, QSpacerItem, QSizePolicy, QProgressBar, QCheckBox, QDialog,
QDialogButtonBox, QComboBox, QStyledItemDelegate, QStyle
QDialogButtonBox, QComboBox, QStyledItemDelegate
)
from PyQt6.QtCore import Qt, QThread, pyqtSignal, QUrl, QTimer, QTime, QSettings, QSize
from PyQt6.QtGui import QIcon, QTextCursor, QDesktopServices, QPixmap, QBrush, QPalette
from PyQt6.QtGui import QIcon, QTextCursor, QDesktopServices, QPixmap, QBrush
from PyQt6.QtNetwork import QNetworkAccessManager, QNetworkRequest, QNetworkReply
from getMetadata import get_filtered_data, parse_uri, SpotifyInvalidUrlException
from getTracks import TrackDownloader
import SquidWTF
@dataclass
class Track:
@@ -120,6 +122,110 @@ class DownloadWorker(QThread):
else:
track_outpath = self.outpath
if self.service == "qobuz":
self.progress.emit(f"Getting track metadata for: {track.title} - {track.artists}", 0)
isrc = None
try:
from getMetadata import get_raw_spotify_data, parse_uri
track_url = track.external_urls
self.progress.emit(f"Fetching Spotify metadata for ISRC...", 0)
raw_data = get_raw_spotify_data(track_url)
if raw_data and "external_ids" in raw_data and "isrc" in raw_data["external_ids"]:
isrc = raw_data["external_ids"]["isrc"]
self.progress.emit(f"Found ISRC from Spotify: {isrc}", 0)
except Exception as e:
self.progress.emit(f"Could not get ISRC from Spotify raw data: {str(e)}", 0)
if not isrc:
self.progress.emit(f"No ISRC found, searching by title and artist", 0)
search_query = f"{track.title} {track.artists}"
try:
self.progress.emit(f"Searching Qobuz for: {search_query}", 0)
qobuz_track_info = SquidWTF.search_track(track.title, track.artists, strict_match=True)
if qobuz_track_info:
qobuz_track_id = qobuz_track_info["id"]
self.progress.emit(f"Found track on Qobuz by title search: {qobuz_track_info['title']} - {qobuz_track_info['performer']['name']}", 0)
found_artist = qobuz_track_info['performer']['name'].lower()
expected_artist = track.artists.lower()
if expected_artist not in found_artist and found_artist not in expected_artist:
self.progress.emit(f"Warning: Artist mismatch! Expected: {track.artists}, Found: {qobuz_track_info['performer']['name']}", 0)
raise Exception(f"Artist mismatch: Expected '{track.artists}', found '{qobuz_track_info['performer']['name']}'")
else:
raise Exception(f"Could not find track on Qobuz: {track.title} - {track.artists}")
except Exception as e:
self.progress.emit(f"Search by title failed: {str(e)}", 0)
raise Exception(f"Could not find track on Qobuz: {track.title} - {track.artists}")
else:
self.progress.emit(f"Searching Qobuz with ISRC: {isrc}", 0)
qobuz_track_info = SquidWTF.get_track_info(isrc)
qobuz_track_id = qobuz_track_info["id"]
self.progress.emit(f"Found track on Qobuz: {qobuz_track_info['title']} - {qobuz_track_info['performer']['name']}", 0)
found_artist = qobuz_track_info['performer']['name'].lower()
expected_artist = track.artists.lower()
if expected_artist not in found_artist and found_artist not in expected_artist:
self.progress.emit(f"Warning: Artist mismatch! Expected: {track.artists}, Found: {qobuz_track_info['performer']['name']}", 0)
download_url = SquidWTF.get_download_url(qobuz_track_id)
os.makedirs(track_outpath, exist_ok=True)
temp_filename = os.path.join(track_outpath, f"temp_{qobuz_track_id}.flac")
self.progress.emit(f"Downloading from Qobuz...", 0)
def progress_callback(current, total):
if total > 0:
percent = (current / total) * 100
current_mb = current / (1024 * 1024)
total_mb = total / (1024 * 1024)
self.progress.emit(f"Download progress: {percent:.2f}% ({current_mb:.2f}MB/{total_mb:.2f}MB)",
int(percent))
try:
SquidWTF.download_file(download_url, temp_filename, progress_callback)
if not os.path.exists(temp_filename) or os.path.getsize(temp_filename) == 0:
raise Exception(f"Downloaded file is empty or does not exist: {temp_filename}")
self.progress.emit(f"Embedding metadata...", 0)
SquidWTF.embed_metadata(temp_filename, qobuz_track_info)
if (self.is_album or (self.is_playlist and self.use_album_subfolders)) and self.use_track_numbers:
new_filename = f"{track.track_number:02d} - {self.get_formatted_filename(track)}"
else:
new_filename = self.get_formatted_filename(track)
new_filename = re.sub(r'[<>:"/\\|?*]', '_', new_filename)
new_filepath = os.path.join(track_outpath, new_filename)
if os.path.exists(new_filepath):
os.remove(new_filepath)
os.rename(temp_filename, new_filepath)
downloaded_file = new_filepath
self.progress.emit(f"File renamed to: {new_filename}", 0)
except Exception as e:
self.progress.emit(f"Error during download or processing: {str(e)}", 0)
if os.path.exists(temp_filename):
try:
os.remove(temp_filename)
self.progress.emit(f"Removed incomplete download file", 0)
except:
pass
raise Exception(f"Failed to download or process file: {str(e)}")
else:
import asyncio
metadata = asyncio.run(downloader.get_track_info(track_id, self.service))
@@ -218,23 +324,48 @@ class ServiceStatusChecker(QThread):
error = pyqtSignal(str)
def run(self):
services_status = {
'amazon': False,
'tidal': False,
'deezer': False,
'qobuz': False
}
try:
response = requests.get("https://lucida.to/api/stats", timeout=5)
if response.status_code == 200:
try:
data = response.json()
services_status = {}
current_services = data.get('all', {}).get('downloads', {}).get('current', {}).get('services', {})
services_status['amazon'] = current_services.get('amazon', 0) > 0
services_status['tidal'] = current_services.get('tidal', 0) > 0
services_status['deezer'] = current_services.get('deezer', 0) > 0
self.status_updated.emit(services_status)
else:
self.error.emit(f"Server returned status code: {response.status_code}")
print("Lucida services status check successful")
except json.JSONDecodeError as e:
print(f"Lucida API returned invalid JSON: {e}")
except Exception as e:
self.error.emit(f"Error checking service status: {str(e)}")
print(f"Error processing Lucida API response: {str(e)}")
else:
print(f"Lucida API returned status code: {response.status_code}")
except requests.exceptions.RequestException as e:
print(f"Error connecting to Lucida API: {str(e)}")
except Exception as e:
print(f"Unexpected error checking Lucida services: {str(e)}")
try:
qobuz_response = requests.get("https://us.qobuz.squid.wtf", timeout=5)
services_status['qobuz'] = qobuz_response.status_code in [200, 304]
print(f"SquidWTF (Qobuz) status check: {qobuz_response.status_code} - {'Online' if services_status['qobuz'] else 'Offline'}")
except requests.exceptions.RequestException as e:
print(f"Error connecting to SquidWTF API (Qobuz): {str(e)}")
services_status['qobuz'] = False
except Exception as e:
print(f"Unexpected error checking SquidWTF (Qobuz): {str(e)}")
services_status['qobuz'] = False
self.status_updated.emit(services_status)
class StatusIndicatorDelegate(QStyledItemDelegate):
def paint(self, painter, option, index):
@@ -243,11 +374,6 @@ class StatusIndicatorDelegate(QStyledItemDelegate):
super().paint(painter, option, index)
if option.state & QStyle.StateFlag.State_Selected:
text_color = option.palette.color(QPalette.ColorGroup.Active, QPalette.ColorRole.HighlightedText)
else:
text_color = option.palette.color(QPalette.ColorGroup.Active, QPalette.ColorRole.Text)
indicator_color = Qt.GlobalColor.green if is_online else Qt.GlobalColor.red
circle_size = 6
@@ -285,7 +411,8 @@ class ServiceComboBox(QComboBox):
self.services = [
{'id': 'tidal', 'name': 'Tidal', 'icon': 'tidal.png', 'online': False},
{'id': 'amazon', 'name': 'Amazon Music', 'icon': 'amazon.png', 'online': False},
{'id': 'deezer', 'name': 'Deezer', 'icon': 'deezer.png', 'online': False}
{'id': 'deezer', 'name': 'Deezer', 'icon': 'deezer.png', 'online': False},
{'id': 'qobuz', 'name': 'Qobuz', 'icon': 'qobuz.png', 'online': False}
]
for service in self.services:
@@ -331,7 +458,7 @@ class ServiceComboBox(QComboBox):
class SpotiFLACGUI(QWidget):
def __init__(self):
super().__init__()
self.current_version = "2.6"
self.current_version = "2.7"
self.tracks = []
self.reset_state()
@@ -419,6 +546,7 @@ class SpotiFLACGUI(QWidget):
self.setup_tabs()
self.setLayout(self.main_layout)
QTimer.singleShot(0, self.update_service_ui_visibility)
def setup_spotify_section(self):
spotify_layout = QHBoxLayout()
@@ -663,40 +791,47 @@ class SpotiFLACGUI(QWidget):
auth_layout = QVBoxLayout(auth_group)
auth_layout.setSpacing(5)
auth_label = QLabel('Lucida Settings')
auth_label = QLabel('Service Setting')
auth_label.setStyleSheet("font-weight: bold;")
auth_layout.addWidget(auth_label)
service_fallback_layout = QHBoxLayout()
source_fallback_layout = QHBoxLayout()
service_label = QLabel('Service:')
service_label = QLabel('Source:')
self.service_dropdown = ServiceComboBox()
self.service_dropdown.currentIndexChanged.connect(self.save_service_setting)
service_fallback_layout.addWidget(service_label)
service_fallback_layout.addWidget(self.service_dropdown)
saved_service = self.service
for i in range(self.service_dropdown.count()):
if self.service_dropdown.itemData(i, Qt.ItemDataRole.UserRole + 1) == saved_service:
self.service_dropdown.setCurrentIndex(i)
break
service_fallback_layout.addSpacing(20)
source_fallback_layout.addWidget(service_label)
source_fallback_layout.addWidget(self.service_dropdown)
source_fallback_layout.addSpacing(20)
self.fallback_checkbox = QCheckBox('Fallback')
self.fallback_checkbox.setCursor(Qt.CursorShape.PointingHandCursor)
self.fallback_checkbox.setChecked(self.use_fallback)
self.fallback_checkbox.toggled.connect(self.save_fallback_setting)
service_fallback_layout.addWidget(self.fallback_checkbox)
source_fallback_layout.addWidget(self.fallback_checkbox)
service_fallback_layout.addSpacing(20)
source_fallback_layout.addSpacing(20)
timeout_label = QLabel('Timeout:')
self.timeout_label = timeout_label
self.timeout_input = QLineEdit()
self.timeout_input.setText(str(self.timeout_value))
self.timeout_input.setFixedWidth(60)
self.timeout_input.textChanged.connect(self.save_timeout_setting)
service_fallback_layout.addWidget(timeout_label)
service_fallback_layout.addWidget(self.timeout_input)
source_fallback_layout.addWidget(timeout_label)
source_fallback_layout.addWidget(self.timeout_input)
service_fallback_layout.addStretch()
auth_layout.addLayout(service_fallback_layout)
source_fallback_layout.addStretch()
auth_layout.addLayout(source_fallback_layout)
settings_layout.addWidget(auth_group)
@@ -704,6 +839,19 @@ class SpotiFLACGUI(QWidget):
settings_tab.setLayout(settings_layout)
self.tab_widget.addTab(settings_tab, "Settings")
def update_service_ui_visibility(self):
if hasattr(self, 'fallback_checkbox') and hasattr(self, 'timeout_input') and hasattr(self, 'timeout_label'):
service = self.service_dropdown.currentData() if hasattr(self, 'service_dropdown') else self.service
if service == "qobuz":
self.fallback_checkbox.setVisible(False)
self.timeout_input.setVisible(False)
self.timeout_label.setVisible(False)
else:
self.fallback_checkbox.setVisible(True)
self.timeout_input.setVisible(True)
self.timeout_label.setVisible(True)
def setup_about_tab(self):
about_tab = QWidget()
about_layout = QVBoxLayout()
@@ -754,7 +902,7 @@ class SpotiFLACGUI(QWidget):
spacer = QSpacerItem(20, 6, QSizePolicy.Policy.Minimum, QSizePolicy.Policy.Fixed)
about_layout.addItem(spacer)
footer_label = QLabel("v2.6 | May 2025")
footer_label = QLabel("v2.7 | May 2025")
footer_label.setStyleSheet("font-size: 12px; margin-top: 10px;")
about_layout.addWidget(footer_label, alignment=Qt.AlignmentFlag.AlignCenter)
@@ -808,6 +956,8 @@ class SpotiFLACGUI(QWidget):
self.settings.sync()
self.log_output.append(f"Service setting saved: {self.service_dropdown.currentText()}")
self.update_service_ui_visibility()
def save_settings(self):
self.settings.setValue('output_path', self.output_dir.text().strip())
self.settings.sync()
SquidWTF.py
+228
View File
@@ -0,0 +1,228 @@
import requests
from mutagen.flac import FLAC, Picture
from datetime import datetime
import sys
import os
def get_track_info(isrc):
print(f"Search: {isrc}")
url = f"https://us.qobuz.squid.wtf/api/get-music?q={isrc}&offset=0"
response = requests.get(url)
data = response.json()
if not data.get("success"):
raise Exception("Failed to get track info")
tracks = data["data"]["tracks"]["items"]
if not tracks:
print(f"Not Found: {isrc}")
raise Exception(f"No tracks found for ISRC: {isrc}")
track = None
for item in tracks:
if item["isrc"] == isrc:
track = item
break
if not track:
print(f"Not Found: {isrc}")
raise Exception(f"No track with matching ISRC: {isrc}")
print(f"Found: {track['title']} - {track['performer']['name']}")
return track
def search_track(title, artist, strict_match=False):
print(f"Search by title/artist: {title} - {artist}")
search_query = f"{title} {artist}".replace("feat.", "").replace("ft.", "")
url = f"https://us.qobuz.squid.wtf/api/get-music?q={search_query}&offset=0"
response = requests.get(url)
data = response.json()
if not data.get("success"):
raise Exception("Failed to search for track")
tracks = data["data"]["tracks"]["items"]
if not tracks:
print(f"Not Found: {title} - {artist}")
raise Exception(f"No tracks found for: {title} - {artist}")
best_match = None
title_lower = title.lower()
artist_lower = artist.lower()
for item in tracks:
item_title = item["title"].lower()
item_artist = item["performer"]["name"].lower()
if title_lower == item_title and (artist_lower in item_artist or item_artist in artist_lower):
best_match = item
print(f"Found exact title match with artist: {item['title']} - {item['performer']['name']}")
break
if not best_match and not strict_match:
for item in tracks:
item_title = item["title"].lower()
item_artist = item["performer"]["name"].lower()
if title_lower in item_title and (artist_lower in item_artist or item_artist in artist_lower):
best_match = item
print(f"Found partial match: {item['title']} - {item['performer']['name']}")
break
if strict_match and best_match:
item_artist = best_match["performer"]["name"].lower()
if artist_lower not in item_artist and item_artist not in artist_lower:
print(f"Artist mismatch in strict mode: Expected '{artist}', found '{best_match['performer']['name']}'")
best_match = None
if not best_match and not strict_match and tracks:
best_match = tracks[0]
print(f"No good match, using first result: {best_match['title']} - {best_match['performer']['name']}")
if not best_match:
print(f"Not Found: {title} - {artist}")
raise Exception(f"No suitable track found for: {title} - {artist}")
print(f"Found by title search: {best_match['title']} - {best_match['performer']['name']}")
return best_match
def get_download_url(track_id):
url = f"https://us.qobuz.squid.wtf/api/download-music?track_id={track_id}&quality=27"
response = requests.get(url)
data = response.json()
if not data.get("success"):
raise Exception("Failed to get download URL")
return data["data"]["url"]
def download_file(url, filename, progress_callback=None):
directory = os.path.dirname(filename)
if directory and not os.path.exists(directory):
try:
os.makedirs(directory, exist_ok=True)
print(f"Created directory: {directory}")
except Exception as e:
raise Exception(f"Failed to create directory {directory}: {str(e)}")
try:
with open(filename, 'wb') as test_file:
pass
except Exception as e:
raise Exception(f"Cannot write to file {filename}: {str(e)}")
try:
response = requests.get(url, stream=True)
if response.status_code != 200:
raise Exception(f"Failed to download file: {response.status_code}")
total_size = int(response.headers.get('content-length', 0))
downloaded = 0
with open(filename, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
downloaded += len(chunk)
if total_size > 0 and progress_callback:
progress_callback(downloaded, total_size)
elif total_size > 0:
progress = (downloaded / total_size) * 100
sys.stdout.write(f"\rProgress Download: {progress:.1f}%")
sys.stdout.flush()
if total_size > 0:
sys.stdout.write("\n")
if not os.path.exists(filename) or os.path.getsize(filename) == 0:
raise Exception(f"Download failed: File {filename} is empty or does not exist")
return filename
except Exception as e:
if os.path.exists(filename):
try:
os.remove(filename)
print(f"Removed incomplete file: {filename}")
except:
pass
raise Exception(f"Download failed: {str(e)}")
def embed_metadata(filename, track_info):
if not os.path.exists(filename):
raise Exception(f"Cannot embed metadata: File {filename} does not exist")
try:
print("Embedding Tags...")
audio = FLAC(filename)
audio.clear()
audio["TITLE"] = track_info["title"]
audio["ARTIST"] = track_info["performer"]["name"]
audio["ALBUM"] = track_info["album"]["title"]
audio["ALBUMARTIST"] = track_info["album"]["artist"]["name"]
audio["TRACKNUMBER"] = str(track_info["track_number"])
audio["LABEL"] = track_info["album"]["label"]["name"]
audio["GENRE"] = track_info["album"]["genre"]["name"]
release_date = datetime.fromtimestamp(track_info["album"]["released_at"]).strftime("%Y-%m-%d")
release_year = release_date.split("-")[0]
audio["DATE"] = release_date
audio["YEAR"] = release_year
audio["ISRC"] = track_info["isrc"]
audio["COPYRIGHT"] = track_info["copyright"]
if track_info["album"]["image"]["large"]:
try:
cover_data = download_cover_image(track_info["album"]["image"]["large"])
picture = Picture()
picture.type = 3
picture.mime = "image/jpeg"
picture.desc = ""
picture.data = cover_data
audio.add_picture(picture)
except Exception as e:
print(f"Warning: Could not add cover image: {str(e)}")
audio.save()
except Exception as e:
raise Exception(f"Failed to embed metadata: {str(e)}")
def download_cover_image(url):
response = requests.get(url)
if response.status_code != 200:
raise Exception(f"Failed to download cover image: {response.status_code}")
return response.content
def main():
try:
isrc = "USUM72409273"
track_info = get_track_info(isrc)
track_id = track_info["id"]
if track_info["isrc"] != isrc:
raise Exception(f"ISRC mismatch: {track_info['isrc']} != {isrc}")
download_url = get_download_url(track_id)
filename = f"{track_info['title']} - {track_info['performer']['name']}.flac"
filename = filename.replace('/', '_').replace('\\', '_')
download_file(download_url, filename)
embed_metadata(filename, track_info)
print("Downloaded Successfully!")
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
main()
getMetadata.py
+27 -6
View File
@@ -319,6 +319,8 @@ def format_track_data(track_data):
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),
@@ -328,7 +330,8 @@ def format_track_data(track_data):
"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', '')
"external_urls": track_data.get('external_urls', {}).get('spotify', ''),
"isrc": isrc
}
}
@@ -345,6 +348,20 @@ def format_album_data(album_data):
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', ''),
@@ -353,7 +370,8 @@ def format_album_data(album_data):
"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', '')
"external_urls": track.get('external_urls', {}).get('spotify', ''),
"isrc": track_isrc
})
album_info = {
@@ -389,6 +407,8 @@ def format_playlist_data(playlist_data):
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', ''),
@@ -397,7 +417,8 @@ def format_playlist_data(playlist_data):
"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', '')
"external_urls": track.get('external_urls', {}).get('spotify', ''),
"isrc": track_isrc
})
playlist_info = {
@@ -443,9 +464,9 @@ def get_filtered_data(spotify_url, batch=False, delay=1.0):
return {"error": "Failed to get raw data"}
if __name__ == '__main__':
playlist = "https://open.spotify.com/playlist/5Qvz8wZIRYbEUUFoPueKI5"
album = "https://open.spotify.com/album/7kFyd5oyJdVX2pIi6P4iHE"
song = "https://open.spotify.com/track/4wJ5Qq0jBN4ajy7ouZIV1c"
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))