JDM170/spotizerr-dev
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

man, markets are hard

Browse Source
This commit is contained in:
Xoconoch
2025-06-04 22:58:42 -06:00
parent 32040f77c6
commit 0277980a5e
15 changed files with 1496 additions and 1166 deletions
Download Patch File Download Diff File Expand all files Collapse all files

194
routes/utils/album.py
View File

@@ -4,6 +4,8 @@ import traceback
from deezspot.spotloader import SpoLogin
from deezspot.deezloader import DeeLogin
from pathlib import Path
from routes.utils.credentials import get_credential, _get_global_spotify_api_creds
from routes.utils.celery_config import get_config_params
def download_album(
url,
@@ -28,88 +30,49 @@ def download_album(
is_spotify_url = 'open.spotify.com' in url.lower()
is_deezer_url = 'deezer.com' in url.lower()
# Determine service exclusively from URL
service = ''
if is_spotify_url:
service = 'spotify'
elif is_deezer_url:
service = 'deezer'
else:
# If URL can't be detected, raise an error
error_msg = "Invalid URL: Must be from open.spotify.com or deezer.com"
print(f"ERROR: {error_msg}")
raise ValueError(error_msg)
print(f"DEBUG: album.py - URL detection: is_spotify_url={is_spotify_url}, is_deezer_url={is_deezer_url}")
print(f"DEBUG: album.py - Service determined from URL: {service}")
print(f"DEBUG: album.py - Credentials: main={main}, fallback={fallback}")
# Load Spotify client credentials if available
spotify_client_id = None
spotify_client_secret = None
# Smartly determine where to look for Spotify search credentials
if service == 'spotify' and fallback:
# If fallback is enabled, use the fallback account for Spotify search credentials
search_creds_path = Path(f'./data/creds/spotify/{fallback}/search.json')
print(f"DEBUG: Using Spotify search credentials from fallback: {search_creds_path}")
else:
# Otherwise use the main account for Spotify search credentials
search_creds_path = Path(f'./data/creds/spotify/{main}/search.json')
print(f"DEBUG: Using Spotify search credentials from main: {search_creds_path}")
if search_creds_path.exists():
try:
with open(search_creds_path, 'r') as f:
search_creds = json.load(f)
spotify_client_id = search_creds.get('client_id')
spotify_client_secret = search_creds.get('client_secret')
print(f"DEBUG: Loaded Spotify client credentials successfully")
except Exception as e:
print(f"Error loading Spotify search credentials: {e}")
# For Spotify URLs: check if fallback is enabled, if so use the fallback logic,
# otherwise download directly from Spotify
print(f"DEBUG: album.py - Credentials provided: main_account_name='{main}', fallback_account_name='{fallback}'")
# Get global Spotify API credentials
global_spotify_client_id, global_spotify_client_secret = _get_global_spotify_api_creds()
if not global_spotify_client_id or not global_spotify_client_secret:
warning_msg = "WARN: album.py - Global Spotify client_id/secret not found in search.json. Spotify operations will likely fail."
print(warning_msg)
if service == 'spotify':
if fallback:
if quality is None:
quality = 'FLAC'
if fall_quality is None:
fall_quality = 'HIGH'
if fallback: # Fallback is a Deezer account name for a Spotify URL
if quality is None: quality = 'FLAC' # Deezer quality for first attempt
if fall_quality is None: fall_quality = 'HIGH' # Spotify quality for fallback (if Deezer fails)
# First attempt: use DeeLogin's download_albumspo with the 'main' (Deezer credentials)
deezer_error = None
try:
# Load Deezer credentials from 'main' under deezer directory
deezer_creds_dir = os.path.join('./data/creds/deezer', main)
deezer_creds_path = os.path.abspath(os.path.join(deezer_creds_dir, 'credentials.json'))
# Attempt 1: Deezer via download_albumspo (using 'fallback' as Deezer account name)
print(f"DEBUG: album.py - Spotify URL. Attempt 1: Deezer (account: {fallback})")
deezer_fallback_creds = get_credential('deezer', fallback)
arl = deezer_fallback_creds.get('arl')
if not arl:
raise ValueError(f"ARL not found for Deezer account '{fallback}'.")
# DEBUG: Print Deezer credential paths being used
print(f"DEBUG: Looking for Deezer credentials at:")
print(f"DEBUG: deezer_creds_dir = {deezer_creds_dir}")
print(f"DEBUG: deezer_creds_path = {deezer_creds_path}")
print(f"DEBUG: Directory exists = {os.path.exists(deezer_creds_dir)}")
print(f"DEBUG: Credentials file exists = {os.path.exists(deezer_creds_path)}")
# List available directories to compare
print(f"DEBUG: Available Deezer credential directories:")
for dir_name in os.listdir('./data/creds/deezer'):
print(f"DEBUG: ./data/creds/deezer/{dir_name}")
with open(deezer_creds_path, 'r') as f:
deezer_creds = json.load(f)
# Initialize DeeLogin with Deezer credentials and Spotify client credentials
dl = DeeLogin(
arl=deezer_creds.get('arl', ''),
spotify_client_id=spotify_client_id,
spotify_client_secret=spotify_client_secret,
arl=arl,
spotify_client_id=global_spotify_client_id,
spotify_client_secret=global_spotify_client_secret,
progress_callback=progress_callback
)
print(f"DEBUG: Starting album download using Deezer credentials (download_albumspo)")
# Download using download_albumspo; pass real_time_dl accordingly and the custom formatting
dl.download_albumspo(
link_album=url,
link_album=url, # Spotify URL
output_dir="./downloads",
quality_download=quality,
quality_download=quality, # Deezer quality
recursive_quality=True,
recursive_download=False,
not_interface=False,
@@ -124,35 +87,33 @@ def download_album(
convert_to=convert_to,
bitrate=bitrate
)
print(f"DEBUG: Album download completed successfully using Deezer credentials")
print(f"DEBUG: album.py - Album download via Deezer (account: {fallback}) successful for Spotify URL.")
except Exception as e:
deezer_error = e
# Immediately report the Deezer error
print(f"ERROR: Deezer album download attempt failed: {e}")
print(f"ERROR: album.py - Deezer attempt (account: {fallback}) for Spotify URL failed: {e}")
traceback.print_exc()
print("Attempting Spotify fallback...")
print(f"DEBUG: album.py - Attempting Spotify direct download (account: {main} for blob)...")
# Load fallback Spotify credentials and attempt download
# Attempt 2: Spotify direct via download_album (using 'main' as Spotify account for blob)
try:
spo_creds_dir = os.path.join('./data/creds/spotify', fallback)
spo_creds_path = os.path.abspath(os.path.join(spo_creds_dir, 'credentials.json'))
print(f"DEBUG: Using Spotify fallback credentials from: {spo_creds_path}")
print(f"DEBUG: Fallback credentials exist: {os.path.exists(spo_creds_path)}")
# We've already loaded the Spotify client credentials above based on fallback
if not global_spotify_client_id or not global_spotify_client_secret:
raise ValueError("Global Spotify API credentials (client_id/secret) not configured for Spotify download.")
spotify_main_creds = get_credential('spotify', main) # For blob path
blob_file_path = spotify_main_creds.get('blob_file_path')
if not Path(blob_file_path).exists():
raise FileNotFoundError(f"Spotify credentials blob file not found at {blob_file_path} for account '{main}'")
spo = SpoLogin(
credentials_path=spo_creds_path,
spotify_client_id=spotify_client_id,
spotify_client_secret=spotify_client_secret,
credentials_path=blob_file_path,
spotify_client_id=global_spotify_client_id,
spotify_client_secret=global_spotify_client_secret,
progress_callback=progress_callback
)
print(f"DEBUG: Starting album download using Spotify fallback credentials")
spo.download_album(
link_album=url,
link_album=url, # Spotify URL
output_dir="./downloads",
quality_download=fall_quality,
quality_download=fall_quality, # Spotify quality
recursive_quality=True,
recursive_download=False,
not_interface=False,
@@ -168,34 +129,35 @@ def download_album(
convert_to=convert_to,
bitrate=bitrate
)
print(f"DEBUG: Album download completed successfully using Spotify fallback")
print(f"DEBUG: album.py - Spotify direct download (account: {main} for blob) successful.")
except Exception as e2:
# If fallback also fails, raise an error indicating both attempts failed
print(f"ERROR: Spotify fallback also failed: {e2}")
print(f"ERROR: album.py - Spotify direct download (account: {main} for blob) also failed: {e2}")
raise RuntimeError(
f"Both main (Deezer) and fallback (Spotify) attempts failed. "
f"Both Deezer attempt (account: {fallback}) and Spotify direct (account: {main} for blob) failed. "
f"Deezer error: {deezer_error}, Spotify error: {e2}"
) from e2
else:
# Original behavior: use Spotify main
if quality is None:
quality = 'HIGH'
creds_dir = os.path.join('./data/creds/spotify', main)
credentials_path = os.path.abspath(os.path.join(creds_dir, 'credentials.json'))
print(f"DEBUG: Using Spotify main credentials from: {credentials_path}")
print(f"DEBUG: Credentials exist: {os.path.exists(credentials_path)}")
# Spotify URL, no fallback. Direct Spotify download using 'main' (Spotify account for blob)
if quality is None: quality = 'HIGH' # Default Spotify quality
print(f"DEBUG: album.py - Spotify URL, no fallback. Direct download with Spotify account (for blob): {main}")
if not global_spotify_client_id or not global_spotify_client_secret:
raise ValueError("Global Spotify API credentials (client_id/secret) not configured for Spotify download.")
spotify_main_creds = get_credential('spotify', main) # For blob path
blob_file_path = spotify_main_creds.get('blob_file_path')
if not Path(blob_file_path).exists():
raise FileNotFoundError(f"Spotify credentials blob file not found at {blob_file_path} for account '{main}'")
spo = SpoLogin(
credentials_path=credentials_path,
spotify_client_id=spotify_client_id,
spotify_client_secret=spotify_client_secret,
credentials_path=blob_file_path,
spotify_client_id=global_spotify_client_id,
spotify_client_secret=global_spotify_client_secret,
progress_callback=progress_callback
)
print(f"DEBUG: Starting album download using Spotify main credentials")
spo.download_album(
link_album=url,
output_dir="./downloads",
quality_download=quality,
quality_download=quality,
recursive_quality=True,
recursive_download=False,
not_interface=False,
@@ -211,27 +173,24 @@ def download_album(
convert_to=convert_to,
bitrate=bitrate
)
print(f"DEBUG: Album download completed successfully using Spotify main")
# For Deezer URLs: download directly from Deezer
print(f"DEBUG: album.py - Direct Spotify download (account: {main} for blob) successful.")
elif service == 'deezer':
if quality is None:
quality = 'FLAC'
# Existing code remains the same, ignoring fallback
creds_dir = os.path.join('./data/creds/deezer', main)
creds_path = os.path.abspath(os.path.join(creds_dir, 'credentials.json'))
print(f"DEBUG: Using Deezer credentials from: {creds_path}")
print(f"DEBUG: Credentials exist: {os.path.exists(creds_path)}")
with open(creds_path, 'r') as f:
creds = json.load(f)
# Deezer URL. Direct Deezer download using 'main' (Deezer account name for ARL)
if quality is None: quality = 'FLAC' # Default Deezer quality
print(f"DEBUG: album.py - Deezer URL. Direct download with Deezer account: {main}")
deezer_main_creds = get_credential('deezer', main) # For ARL
arl = deezer_main_creds.get('arl')
if not arl:
raise ValueError(f"ARL not found for Deezer account '{main}'.")
dl = DeeLogin(
arl=creds.get('arl', ''),
spotify_client_id=spotify_client_id,
spotify_client_secret=spotify_client_secret,
arl=arl, # Account specific ARL
spotify_client_id=global_spotify_client_id, # Global Spotify keys
spotify_client_secret=global_spotify_client_secret, # Global Spotify keys
progress_callback=progress_callback
)
print(f"DEBUG: Starting album download using Deezer credentials (download_albumdee)")
dl.download_albumdee(
dl.download_albumdee( # Deezer URL, download via Deezer
link_album=url,
output_dir="./downloads",
quality_download=quality,
@@ -248,9 +207,10 @@ def download_album(
convert_to=convert_to,
bitrate=bitrate
)
print(f"DEBUG: Album download completed successfully using Deezer direct")
print(f"DEBUG: album.py - Direct Deezer download (account: {main}) successful.")
else:
raise ValueError(f"Unsupported service: {service}")
# Should be caught by initial service check, but as a safeguard
raise ValueError(f"Unsupported service determined: {service}")
except Exception as e:
print(f"ERROR: Album download failed with exception: {e}")
traceback.print_exc()

52
routes/utils/artist.py
View File

@@ -6,6 +6,7 @@ import logging
from flask import Blueprint, Response, request, url_for
from routes.utils.celery_queue_manager import download_queue_manager, get_config_params
from routes.utils.get_info import get_spotify_info
from routes.utils.credentials import get_credential, _get_global_spotify_api_creds
from routes.utils.celery_tasks import get_last_task_status, ProgressState
from deezspot.easy_spoty import Spo
@@ -19,36 +20,42 @@ def log_json(message_dict):
print(json.dumps(message_dict))
def get_artist_discography(url, main, album_type='album,single,compilation,appears_on', progress_callback=None):
def get_artist_discography(url, main_spotify_account_name, album_type='album,single,compilation,appears_on', progress_callback=None):
"""
Validate the URL, extract the artist ID, and retrieve the discography.
Uses global Spotify API client_id/secret for Spo initialization.
Args:
url (str): Spotify artist URL.
main_spotify_account_name (str): Name of the Spotify account (for context/logging, not API keys for Spo.__init__).
album_type (str): Types of albums to fetch.
progress_callback: Optional callback for progress.
"""
if not url:
log_json({"status": "error", "message": "No artist URL provided."})
raise ValueError("No artist URL provided.")
# This will raise an exception if the link is invalid.
link_is_valid(link=url)
link_is_valid(link=url) # This will raise an exception if the link is invalid.
# Initialize Spotify API with credentials
spotify_client_id = None
spotify_client_secret = None
search_creds_path = Path(f'./data/creds/spotify/{main}/search.json')
if search_creds_path.exists():
try:
with open(search_creds_path, 'r') as f:
search_creds = json.load(f)
spotify_client_id = search_creds.get('client_id')
spotify_client_secret = search_creds.get('client_secret')
except Exception as e:
log_json({"status": "error", "message": f"Error loading Spotify search credentials: {e}"})
raise
client_id, client_secret = _get_global_spotify_api_creds()
if not client_id or not client_secret:
log_json({"status": "error", "message": "Global Spotify API client_id or client_secret not configured."})
raise ValueError("Global Spotify API credentials are not configured.")
# Initialize the Spotify client with credentials
if spotify_client_id and spotify_client_secret:
Spo.__init__(spotify_client_id, spotify_client_secret)
if not main_spotify_account_name:
# This is a warning now, as API keys are global.
logger.warning("main_spotify_account_name not provided for get_artist_discography context. Using global API keys.")
else:
raise ValueError("No Spotify credentials found")
# Check if account exists for context, good for consistency
try:
get_credential('spotify', main_spotify_account_name)
logger.debug(f"Spotify account context '{main_spotify_account_name}' exists for get_artist_discography.")
except FileNotFoundError:
logger.warning(f"Spotify account '{main_spotify_account_name}' provided for discography context not found.")
except Exception as e:
logger.warning(f"Error checking Spotify account '{main_spotify_account_name}' for discography context: {e}")
Spo.__init__(client_id, client_secret) # Initialize with global API keys
try:
artist_id = get_ids(url)
@@ -58,6 +65,11 @@ def get_artist_discography(url, main, album_type='album,single,compilation,appea
raise ValueError(msg)
try:
# The progress_callback is not a standard param for Spo.get_artist
# If Spo.get_artist is meant to be Spo.get_artist_discography, that would take limit/offset
# Assuming it's Spo.get_artist which takes artist_id and album_type.
# If progress_callback was for a different Spo method, this needs review.
# For now, removing progress_callback from this specific call as Spo.get_artist doesn't use it.
discography = Spo.get_artist(artist_id, album_type=album_type)
return discography
except Exception as fetch_error:

4
routes/utils/celery_manager.py
View File

@@ -24,6 +24,8 @@ from .celery_tasks import (
from .celery_config import get_config_params
# Import history manager
from .history_manager import init_history_db
# Import credentials manager for DB init
from .credentials import init_credentials_db
# Configure logging
logger = logging.getLogger(__name__)
@@ -174,6 +176,8 @@ class CeleryManager:
# Initialize history database
init_history_db()
# Initialize credentials database
init_credentials_db()
# Clean up stale tasks BEFORE starting/restarting workers
self._cleanup_stale_tasks()

805
routes/utils/credentials.py
View File

@@ -1,447 +1,470 @@
import json
from pathlib import Path
import shutil
from deezspot.spotloader import SpoLogin
from deezspot.deezloader import DeeLogin
import sqlite3
import traceback # For logging detailed error messages
import time # For retry delays
import logging
def _get_spotify_search_creds(creds_dir: Path):
"""Helper to load client_id and client_secret from search.json for a Spotify account."""
search_file = creds_dir / 'search.json'
if search_file.exists():
# Assuming deezspot is in a location findable by Python's import system
# from deezspot.spotloader import SpoLogin # Used in validation
# from deezspot.deezloader import DeeLogin # Used in validation
# For now, as per original, validation calls these directly.
logger = logging.getLogger(__name__) # Assuming logger is configured elsewhere
# --- New Database and Path Definitions ---
CREDS_BASE_DIR = Path('./data/creds')
ACCOUNTS_DB_PATH = CREDS_BASE_DIR / 'accounts.db'
BLOBS_DIR = CREDS_BASE_DIR / 'blobs'
GLOBAL_SEARCH_JSON_PATH = CREDS_BASE_DIR / 'search.json' # Global Spotify API creds
EXPECTED_SPOTIFY_TABLE_COLUMNS = {
"name": "TEXT PRIMARY KEY",
# client_id and client_secret are now global
"region": "TEXT", # ISO 3166-1 alpha-2
"created_at": "REAL",
"updated_at": "REAL"
}
EXPECTED_DEEZER_TABLE_COLUMNS = {
"name": "TEXT PRIMARY KEY",
"arl": "TEXT",
"region": "TEXT", # ISO 3166-1 alpha-2
"created_at": "REAL",
"updated_at": "REAL"
}
def _get_db_connection():
ACCOUNTS_DB_PATH.parent.mkdir(parents=True, exist_ok=True)
BLOBS_DIR.mkdir(parents=True, exist_ok=True) # Ensure blobs directory also exists
conn = sqlite3.connect(ACCOUNTS_DB_PATH, timeout=10)
conn.row_factory = sqlite3.Row
return conn
def _ensure_table_schema(cursor: sqlite3.Cursor, table_name: str, expected_columns: dict):
"""Ensures the given table has all expected columns, adding them if necessary."""
try:
cursor.execute(f"PRAGMA table_info({table_name})")
existing_columns_info = cursor.fetchall()
existing_column_names = {col[1] for col in existing_columns_info}
added_columns = False
for col_name, col_type in expected_columns.items():
if col_name not in existing_column_names:
# Basic protection against altering PK after creation if table is not empty
if 'PRIMARY KEY' in col_type.upper() and existing_columns_info:
logger.warning(
f"Column '{col_name}' is part of PRIMARY KEY for table '{table_name}' "
f"and was expected to be created by CREATE TABLE. Skipping explicit ADD COLUMN."
)
continue
col_type_for_add = col_type.replace(' PRIMARY KEY', '').strip()
try:
cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {col_name} {col_type_for_add}")
logger.info(f"Added missing column '{col_name} {col_type_for_add}' to table '{table_name}'.")
added_columns = True
except sqlite3.OperationalError as alter_e:
logger.warning(
f"Could not add column '{col_name}' to table '{table_name}': {alter_e}. "
f"It might already exist with a different definition or there's another schema mismatch."
)
return added_columns
except sqlite3.Error as e:
logger.error(f"Error ensuring schema for table '{table_name}': {e}", exc_info=True)
return False
def init_credentials_db():
"""Initializes the accounts.db and its tables if they don't exist."""
try:
with _get_db_connection() as conn:
cursor = conn.cursor()
cursor.row_factory = sqlite3.Row # Apply row_factory here as well for consistency
# Spotify Table
cursor.execute("""
CREATE TABLE IF NOT EXISTS spotify (
name TEXT PRIMARY KEY,
region TEXT,
created_at REAL,
updated_at REAL
)
""")
if _ensure_table_schema(cursor, "spotify", EXPECTED_SPOTIFY_TABLE_COLUMNS):
conn.commit()
# Deezer Table
cursor.execute("""
CREATE TABLE IF NOT EXISTS deezer (
name TEXT PRIMARY KEY,
arl TEXT,
region TEXT,
created_at REAL,
updated_at REAL
)
""")
if _ensure_table_schema(cursor, "deezer", EXPECTED_DEEZER_TABLE_COLUMNS):
conn.commit()
# Ensure global search.json exists, create if not
if not GLOBAL_SEARCH_JSON_PATH.exists():
logger.info(f"Global Spotify search credential file not found at {GLOBAL_SEARCH_JSON_PATH}. Creating empty file.")
with open(GLOBAL_SEARCH_JSON_PATH, 'w') as f_search:
json.dump({"client_id": "", "client_secret": ""}, f_search, indent=4)
conn.commit()
logger.info(f"Credentials database initialized/schema checked at {ACCOUNTS_DB_PATH}")
except sqlite3.Error as e:
logger.error(f"Error initializing credentials database: {e}", exc_info=True)
raise
def _get_global_spotify_api_creds():
"""Loads client_id and client_secret from the global search.json."""
if GLOBAL_SEARCH_JSON_PATH.exists():
try:
with open(search_file, 'r') as f:
with open(GLOBAL_SEARCH_JSON_PATH, 'r') as f:
search_data = json.load(f)
return search_data.get('client_id'), search_data.get('client_secret')
except Exception:
# Log error if search.json is malformed or unreadable
print(f"Warning: Could not read Spotify search credentials from {search_file}")
traceback.print_exc()
return None, None
client_id = search_data.get('client_id')
client_secret = search_data.get('client_secret')
if client_id and client_secret:
return client_id, client_secret
else:
logger.warning(f"Global Spotify API credentials in {GLOBAL_SEARCH_JSON_PATH} are incomplete.")
except Exception as e:
logger.error(f"Error reading global Spotify API credentials from {GLOBAL_SEARCH_JSON_PATH}: {e}", exc_info=True)
else:
logger.warning(f"Global Spotify API credential file {GLOBAL_SEARCH_JSON_PATH} not found.")
return None, None # Return None if file doesn't exist or creds are incomplete/invalid
def _validate_with_retry(service_name, account_name, creds_dir_path, cred_file_path, data_for_validation, is_spotify):
def save_global_spotify_api_creds(client_id: str, client_secret: str):
"""Saves client_id and client_secret to the global search.json."""
try:
GLOBAL_SEARCH_JSON_PATH.parent.mkdir(parents=True, exist_ok=True)
with open(GLOBAL_SEARCH_JSON_PATH, 'w') as f:
json.dump({"client_id": client_id, "client_secret": client_secret}, f, indent=4)
logger.info(f"Global Spotify API credentials saved to {GLOBAL_SEARCH_JSON_PATH}")
return True
except Exception as e:
logger.error(f"Error saving global Spotify API credentials to {GLOBAL_SEARCH_JSON_PATH}: {e}", exc_info=True)
return False
def _validate_with_retry(service_name, account_name, validation_data):
"""
Attempts to validate credentials with retries for connection errors.
- For Spotify, cred_file_path is used.
- For Deezer, data_for_validation (which contains the 'arl' key) is used.
validation_data (dict): For Spotify, expects {'client_id': ..., 'client_secret': ..., 'blob_file_path': ...}
For Deezer, expects {'arl': ...}
Returns True if validated, raises ValueError if not.
"""
max_retries = 5
# Deezspot imports need to be available. Assuming they are.
from deezspot.spotloader import SpoLogin
from deezspot.deezloader import DeeLogin
max_retries = 3 # Reduced for brevity, was 5
last_exception = None
for attempt in range(max_retries):
try:
if is_spotify:
client_id, client_secret = _get_spotify_search_creds(creds_dir_path)
SpoLogin(credentials_path=str(cred_file_path), spotify_client_id=client_id, spotify_client_secret=client_secret)
if service_name == 'spotify':
# For Spotify, validation uses the account's blob and GLOBAL API creds
global_client_id, global_client_secret = _get_global_spotify_api_creds()
if not global_client_id or not global_client_secret:
raise ValueError("Global Spotify API client_id or client_secret not configured for validation.")
blob_file_path = validation_data.get('blob_file_path')
if not blob_file_path or not Path(blob_file_path).exists():
raise ValueError(f"Spotify blob file missing for validation of account {account_name}")
SpoLogin(credentials_path=str(blob_file_path), spotify_client_id=global_client_id, spotify_client_secret=global_client_secret)
else: # Deezer
arl = data_for_validation.get('arl')
arl = validation_data.get('arl')
if not arl:
# This should be caught by prior checks, but as a safeguard:
raise ValueError("Missing 'arl' for Deezer validation.")
DeeLogin(arl=arl)
print(f"{service_name.capitalize()} credentials for {account_name} validated successfully (attempt {attempt + 1}).")
return True # Validation successful
logger.info(f"{service_name.capitalize()} credentials for {account_name} validated successfully (attempt {attempt + 1}).")
return True
except Exception as e:
last_exception = e
error_str = str(e).lower()
# More comprehensive check for connection-related errors
is_connection_error = (
"connection refused" in error_str or
"connection error" in error_str or
"timeout" in error_str or
"temporary failure in name resolution" in error_str or
"dns lookup failed" in error_str or
"network is unreachable" in error_str or
"ssl handshake failed" in error_str or # Can be network-related
"connection reset by peer" in error_str
"connection refused" in error_str or "connection error" in error_str or
"timeout" in error_str or "temporary failure in name resolution" in error_str or
"dns lookup failed" in error_str or "network is unreachable" in error_str or
"ssl handshake failed" in error_str or "connection reset by peer" in error_str
)
if is_connection_error and attempt < max_retries - 1:
retry_delay = 2 + attempt # Increasing delay (2s, 3s, 4s, 5s)
print(f"Validation for {account_name} ({service_name}) failed on attempt {attempt + 1}/{max_retries} due to connection issue: {e}. Retrying in {retry_delay}s...")
retry_delay = 2 + attempt
logger.warning(f"Validation for {account_name} ({service_name}) failed (attempt {attempt + 1}) due to connection issue: {e}. Retrying in {retry_delay}s...")
time.sleep(retry_delay)
continue # Go to next retry attempt
continue
else:
# Not a connection error, or it's the last retry for a connection error
print(f"Validation for {account_name} ({service_name}) failed on attempt {attempt + 1} with non-retryable error or max retries reached for connection error.")
break # Exit retry loop
logger.error(f"Validation for {account_name} ({service_name}) failed on attempt {attempt + 1} (non-retryable or max retries).")
break
# If loop finished without returning True, validation failed
print(f"ERROR: Credential validation definitively failed for {service_name} account {account_name} after {attempt + 1} attempt(s).")
if last_exception:
base_error_message = str(last_exception).splitlines()[-1]
detailed_error_message = f"Invalid {service_name} credentials. Verification failed: {base_error_message}"
if is_spotify and "incorrect padding" in base_error_message.lower():
detailed_error_message += ". Hint: Do not throw your password here, read the docs"
# traceback.print_exc() # Already printed in create/edit, avoid duplicate full trace
detailed_error_message = f"Invalid {service_name} credentials for {account_name}. Verification failed: {base_error_message}"
if service_name == 'spotify' and "incorrect padding" in base_error_message.lower():
detailed_error_message += ". Hint: For Spotify, ensure the credentials blob content is correct."
raise ValueError(detailed_error_message)
else: # Should not happen if loop runs at least once
raise ValueError(f"Invalid {service_name} credentials. Verification failed (unknown reason after retries).")
def get_credential(service, name, cred_type='credentials'):
"""
Retrieves existing credential contents by name.
Args:
service (str): 'spotify' or 'deezer'
name (str): Custom name of the credential to retrieve
cred_type (str): 'credentials' or 'search' - type of credential file to read
Returns:
dict: Credential data as dictionary
Raises:
FileNotFoundError: If the credential doesn't exist
ValueError: For invalid service name or cred_type
"""
if service not in ['spotify', 'deezer']:
raise ValueError("Service must be 'spotify' or 'deezer'")
if cred_type not in ['credentials', 'search']:
raise ValueError("Credential type must be 'credentials' or 'search'")
# For Deezer, only credentials.json is supported
if service == 'deezer' and cred_type == 'search':
raise ValueError("Search credentials are only supported for Spotify")
creds_dir = Path('./data/creds') / service / name
file_path = creds_dir / f'{cred_type}.json'
if not file_path.exists():
if cred_type == 'search':
# Return empty dict if search.json doesn't exist
return {}
raise FileNotFoundError(f"Credential '{name}' not found for {service}")
with open(file_path, 'r') as f:
return json.load(f)
def list_credentials(service):
"""
Lists all available credential names for a service
Args:
service (str): 'spotify' or 'deezer'
Returns:
list: Array of credential names
Raises:
ValueError: For invalid service name
"""
if service not in ['spotify', 'deezer']:
raise ValueError("Service must be 'spotify' or 'deezer'")
service_dir = Path('./data/creds') / service
if not service_dir.exists():
return []
return [d.name for d in service_dir.iterdir() if d.is_dir()]
else:
raise ValueError(f"Invalid {service_name} credentials for {account_name}. Verification failed (unknown reason after retries).")
def create_credential(service, name, data, cred_type='credentials'):
def create_credential(service, name, data):
"""
Creates a new credential file for the specified service.
Creates a new credential.
Args:
service (str): 'spotify' or 'deezer'
name (str): Custom name for the credential
data (dict): Dictionary containing the credential data
cred_type (str): 'credentials' or 'search' - type of credential file to create
data (dict): For Spotify: {'client_id', 'client_secret', 'region', 'blob_content'}
For Deezer: {'arl', 'region'}
Raises:
ValueError: If service is invalid, data has invalid fields, or missing required fields
FileExistsError: If the credential directory already exists (for credentials.json)
ValueError, FileExistsError
"""
if service not in ['spotify', 'deezer']:
raise ValueError("Service must be 'spotify' or 'deezer'")
if cred_type not in ['credentials', 'search']:
raise ValueError("Credential type must be 'credentials' or 'search'")
# For Deezer, only credentials.json is supported
if service == 'deezer' and cred_type == 'search':
raise ValueError("Search credentials are only supported for Spotify")
# Validate data structure
required_fields = []
allowed_fields = []
if cred_type == 'credentials':
if service == 'spotify':
required_fields = ['username', 'credentials']
allowed_fields = required_fields + ['type']
data['type'] = 'AUTHENTICATION_STORED_SPOTIFY_CREDENTIALS'
else:
required_fields = ['arl']
allowed_fields = required_fields.copy()
# Check for extra fields
extra_fields = set(data.keys()) - set(allowed_fields)
if extra_fields:
raise ValueError(f"Deezer credentials can only contain 'arl'. Extra fields found: {', '.join(extra_fields)}")
elif cred_type == 'search':
required_fields = ['client_id', 'client_secret']
allowed_fields = required_fields.copy()
# Check for extra fields
extra_fields = set(data.keys()) - set(allowed_fields)
if extra_fields:
raise ValueError(f"Search credentials can only contain 'client_id' and 'client_secret'. Extra fields found: {', '.join(extra_fields)}")
for field in required_fields:
if field not in data:
raise ValueError(f"Missing required field for {cred_type}: {field}")
# Create directory
creds_dir = Path('./data/creds') / service / name
file_created_now = False
dir_created_now = False
if not name or not isinstance(name, str):
raise ValueError("Credential name must be a non-empty string.")
if cred_type == 'credentials':
current_time = time.time()
with _get_db_connection() as conn:
cursor = conn.cursor()
conn.row_factory = sqlite3.Row
try:
creds_dir.mkdir(parents=True, exist_ok=False)
dir_created_now = True
except FileExistsError:
# Directory already exists, which is fine for creating credentials.json
# if it doesn't exist yet, or if we are overwriting (though POST usually means new)
pass
except Exception as e:
raise ValueError(f"Could not create directory {creds_dir}: {e}")
if service == 'spotify':
required_fields = {'region', 'blob_content'} # client_id/secret are global
if not required_fields.issubset(data.keys()):
raise ValueError(f"Missing fields for Spotify. Required: {required_fields}")
file_path = creds_dir / 'credentials.json'
if file_path.exists() and request.method == 'POST': # type: ignore
# Safety check for POST to not overwrite if file exists unless it's an edit (PUT)
raise FileExistsError(f"Credential file {file_path} already exists. Use PUT to modify.")
# Write the credential file first
try:
with open(file_path, 'w') as f:
json.dump(data, f, indent=4)
file_created_now = True # Mark as created for potential cleanup
except Exception as e:
if dir_created_now: # Cleanup directory if file write failed
blob_path = BLOBS_DIR / name / 'credentials.json'
validation_data = {'blob_file_path': str(blob_path)} # Validation uses global API creds
blob_path.parent.mkdir(parents=True, exist_ok=True)
with open(blob_path, 'w') as f_blob:
if isinstance(data['blob_content'], dict):
json.dump(data['blob_content'], f_blob, indent=4)
else: # assume string
f_blob.write(data['blob_content'])
try:
creds_dir.rmdir()
except OSError: # rmdir fails if not empty, though it should be
pass
raise ValueError(f"Could not write credential file {file_path}: {e}")
# --- Validation Step ---
try:
_validate_with_retry(
service_name=service,
account_name=name,
creds_dir_path=creds_dir,
cred_file_path=file_path,
data_for_validation=data, # 'data' contains the arl for Deezer
is_spotify=(service == 'spotify')
)
except ValueError as val_err: # Catch the specific error from our helper
print(f"ERROR: Credential validation failed during creation for {service} account {name}: {val_err}")
traceback.print_exc() # Print full traceback here for creation failure context
# Clean up the created file and directory if validation fails
if file_created_now:
try:
file_path.unlink(missing_ok=True)
except OSError:
pass # Ignore if somehow already gone
if dir_created_now and not any(creds_dir.iterdir()): # Only remove if empty
try:
creds_dir.rmdir()
except OSError:
pass
raise # Re-raise the ValueError from validation
elif cred_type == 'search': # Spotify only
# For search.json, ensure the directory exists (it should if credentials.json exists)
if not creds_dir.exists():
# This implies credentials.json was not created first, which is an issue.
# However, the form logic might allow adding API creds to an existing empty dir.
# For now, let's create it if it's missing, assuming API creds can be standalone.
try:
creds_dir.mkdir(parents=True, exist_ok=True)
except Exception as e:
raise ValueError(f"Could not create directory for search credentials {creds_dir}: {e}")
file_path = creds_dir / 'search.json'
# No specific validation for client_id/secret themselves, they are validated in use.
with open(file_path, 'w') as f:
json.dump(data, f, indent=4)
def delete_credential(service, name, cred_type=None):
"""
Deletes an existing credential directory or specific credential file.
Args:
service (str): 'spotify' or 'deezer'
name (str): Name of the credential to delete
cred_type (str, optional): If specified ('credentials' or 'search'), only deletes
that specific file. If None, deletes the whole directory.
Raises:
FileNotFoundError: If the credential directory or specified file does not exist
"""
creds_dir = Path('./data/creds') / service / name
if cred_type:
if cred_type not in ['credentials', 'search']:
raise ValueError("Credential type must be 'credentials' or 'search'")
file_path = creds_dir / f'{cred_type}.json'
if not file_path.exists():
raise FileNotFoundError(f"{cred_type.capitalize()} credential '{name}' not found for {service}")
# Delete just the specific file
file_path.unlink()
# If it was credentials.json and no other credential files remain, also delete the directory
if cred_type == 'credentials' and not any(creds_dir.iterdir()):
creds_dir.rmdir()
else:
# Delete the entire directory
if not creds_dir.exists():
raise FileNotFoundError(f"Credential '{name}' not found for {service}")
shutil.rmtree(creds_dir)
def edit_credential(service, name, new_data, cred_type='credentials'):
"""
Edits an existing credential file.
Args:
service (str): 'spotify' or 'deezer'
name (str): Name of the credential to edit
new_data (dict): Dictionary containing fields to update
cred_type (str): 'credentials' or 'search' - type of credential file to edit
Raises:
FileNotFoundError: If the credential does not exist
ValueError: If new_data contains invalid fields or missing required fields after update
"""
if service not in ['spotify', 'deezer']:
raise ValueError("Service must be 'spotify' or 'deezer'")
if cred_type not in ['credentials', 'search']:
raise ValueError("Credential type must be 'credentials' or 'search'")
# For Deezer, only credentials.json is supported
if service == 'deezer' and cred_type == 'search':
raise ValueError("Search credentials are only supported for Spotify")
# Get file path
creds_dir = Path('./data/creds') / service / name
file_path = creds_dir / f'{cred_type}.json'
original_data_str = None # Store original data as string to revert
file_existed_before_edit = file_path.exists()
if file_existed_before_edit:
with open(file_path, 'r') as f:
original_data_str = f.read()
try:
data = json.loads(original_data_str)
except json.JSONDecodeError:
# If existing file is corrupt, treat as if we are creating it anew for edit
data = {}
original_data_str = None # Can't revert to corrupt data
else:
# If file doesn't exist, and we're editing (PUT), it's usually an error
# unless it's for search.json which can be created during an edit flow.
if cred_type == 'credentials':
raise FileNotFoundError(f"Cannot edit non-existent credentials file: {file_path}")
data = {} # Start with empty data for search.json creation
# Validate new_data fields (data to be merged)
allowed_fields = []
if cred_type == 'credentials':
if service == 'spotify':
allowed_fields = ['username', 'credentials']
else:
allowed_fields = ['arl']
else: # search.json
allowed_fields = ['client_id', 'client_secret']
for key in new_data.keys():
if key not in allowed_fields:
raise ValueError(f"Invalid field '{key}' for {cred_type} credentials")
# Update data (merging new_data into existing or empty data)
data.update(new_data)
# --- Write and Validate Step for 'credentials' type ---
if cred_type == 'credentials':
try:
# Temporarily write new data for validation
with open(file_path, 'w') as f:
json.dump(data, f, indent=4)
_validate_with_retry('spotify', name, validation_data)
cursor.execute(
"INSERT INTO spotify (name, region, created_at, updated_at) VALUES (?, ?, ?, ?)",
(name, data['region'], current_time, current_time)
)
except Exception as e:
if blob_path.exists(): blob_path.unlink() # Cleanup blob
if blob_path.parent.exists() and not any(blob_path.parent.iterdir()): blob_path.parent.rmdir()
raise # Re-raise validation or DB error
_validate_with_retry(
service_name=service,
account_name=name,
creds_dir_path=creds_dir,
cred_file_path=file_path,
data_for_validation=data, # 'data' is the merged data with 'arl' for Deezer
is_spotify=(service == 'spotify')
)
except ValueError as val_err: # Catch the specific error from our helper
print(f"ERROR: Edited credential validation failed for {service} account {name}: {val_err}")
traceback.print_exc() # Print full traceback here for edit failure context
# Revert or delete the file
if original_data_str is not None:
with open(file_path, 'w') as f:
f.write(original_data_str) # Restore original content
elif file_existed_before_edit: # file existed but original_data_str is None (corrupt)
pass
else: # File didn't exist before this edit attempt, so remove it
try:
file_path.unlink(missing_ok=True)
except OSError:
pass # Ignore if somehow already gone
raise # Re-raise the ValueError from validation
except Exception as e: # Catch other potential errors like file IO during temp write
print(f"ERROR: Unexpected error during edit/validation for {service} account {name}: {e}")
traceback.print_exc()
# Attempt revert/delete
if original_data_str is not None:
with open(file_path, 'w') as f: f.write(original_data_str)
elif file_existed_before_edit:
pass
else:
try:
file_path.unlink(missing_ok=True)
except OSError: pass
raise ValueError(f"Failed to save edited {service} credentials due to: {str(e).splitlines()[-1]}")
elif service == 'deezer':
required_fields = {'arl', 'region'}
if not required_fields.issubset(data.keys()):
raise ValueError(f"Missing fields for Deezer. Required: {required_fields}")
validation_data = {'arl': data['arl']}
_validate_with_retry('deezer', name, validation_data)
cursor.execute(
"INSERT INTO deezer (name, arl, region, created_at, updated_at) VALUES (?, ?, ?, ?, ?)",
(name, data['arl'], data['region'], current_time, current_time)
)
conn.commit()
logger.info(f"Credential '{name}' for {service} created successfully.")
return {"status": "created", "service": service, "name": name}
except sqlite3.IntegrityError:
raise FileExistsError(f"Credential '{name}' already exists for {service}.")
except Exception as e:
logger.error(f"Error creating credential {name} for {service}: {e}", exc_info=True)
raise ValueError(f"Could not create credential: {e}")
def get_credential(service, name):
"""
Retrieves a specific credential by name.
For Spotify, returns dict with name, region, and blob_content (from file).
For Deezer, returns dict with name, arl, and region.
Raises FileNotFoundError if the credential does not exist.
"""
if service not in ['spotify', 'deezer']:
raise ValueError("Service must be 'spotify' or 'deezer'")
# For 'search' type, just write, no specific validation here for client_id/secret
elif cred_type == 'search':
if not creds_dir.exists(): # Should not happen if we're editing
raise FileNotFoundError(f"Credential directory {creds_dir} not found for editing search credentials.")
with open(file_path, 'w') as f:
json.dump(data, f, indent=4) # `data` here is the merged data for search
with _get_db_connection() as conn:
cursor = conn.cursor()
conn.row_factory = sqlite3.Row # Ensure row_factory is set for this cursor
cursor.execute(f"SELECT * FROM {service} WHERE name = ?", (name,))
row = cursor.fetchone()
# For Deezer: Strip all fields except 'arl' - This should use `data` which is `updated_data`
if service == 'deezer' and cred_type == 'credentials':
if 'arl' not in data:
raise ValueError("Missing 'arl' field for Deezer credential after edit.")
data = {'arl': data['arl']}
if not row:
raise FileNotFoundError(f"No {service} credential found with name '{name}'")
data = dict(row)
# Ensure required fields are present
required_fields = []
if cred_type == 'credentials':
if service == 'spotify':
required_fields = ['username', 'credentials', 'type']
data['type'] = 'AUTHENTICATION_STORED_SPOTIFY_CREDENTIALS'
else:
required_fields = ['arl']
else: # search.json
required_fields = ['client_id', 'client_secret']
blob_file_path = BLOBS_DIR / name / 'credentials.json'
data['blob_file_path'] = str(blob_file_path) # Keep for internal use
try:
with open(blob_file_path, 'r') as f_blob:
blob_data = json.load(f_blob)
data['blob_content'] = blob_data
except FileNotFoundError:
logger.warning(f"Spotify blob file not found for {name} at {blob_file_path} during get_credential.")
data['blob_content'] = None
except json.JSONDecodeError:
logger.warning(f"Error decoding JSON from Spotify blob file for {name} at {blob_file_path}.")
data['blob_content'] = None
except Exception as e:
logger.error(f"Unexpected error reading Spotify blob for {name}: {e}", exc_info=True)
data['blob_content'] = None
cleaned_data = {
'name': data.get('name'),
'region': data.get('region'),
'blob_content': data.get('blob_content')
}
return cleaned_data
elif service == 'deezer':
cleaned_data = {
'name': data.get('name'),
'region': data.get('region'),
'arl': data.get('arl')
}
return cleaned_data
# Fallback, should not be reached if service is spotify or deezer
return None
def list_credentials(service):
if service not in ['spotify', 'deezer']:
raise ValueError("Service must be 'spotify' or 'deezer'")
for field in required_fields:
if field not in data:
raise ValueError(f"Missing required field '{field}' after update for {cred_type}")
with _get_db_connection() as conn:
cursor = conn.cursor()
conn.row_factory = sqlite3.Row
cursor.execute(f"SELECT name FROM {service}")
return [row['name'] for row in cursor.fetchall()]
def delete_credential(service, name):
if service not in ['spotify', 'deezer']:
raise ValueError("Service must be 'spotify' or 'deezer'")
# Save updated data
with open(file_path, 'w') as f:
json.dump(data, f, indent=4)
with _get_db_connection() as conn:
cursor = conn.cursor()
conn.row_factory = sqlite3.Row
cursor.execute(f"DELETE FROM {service} WHERE name = ?", (name,))
if cursor.rowcount == 0:
raise FileNotFoundError(f"Credential '{name}' not found for {service}.")
if service == 'spotify':
blob_dir = BLOBS_DIR / name
if blob_dir.exists():
shutil.rmtree(blob_dir)
conn.commit()
logger.info(f"Credential '{name}' for {service} deleted.")
return {"status": "deleted", "service": service, "name": name}
def edit_credential(service, name, new_data):
"""
Edits an existing credential.
new_data for Spotify can include: client_id, client_secret, region, blob_content.
new_data for Deezer can include: arl, region.
Fields not in new_data remain unchanged.
"""
if service not in ['spotify', 'deezer']:
raise ValueError("Service must be 'spotify' or 'deezer'")
current_time = time.time()
# Fetch existing data first to preserve unchanged fields and for validation backup
try:
existing_cred = get_credential(service, name) # This will raise FileNotFoundError if not found
except FileNotFoundError:
raise
except Exception as e: # Catch other errors from get_credential
raise ValueError(f"Could not retrieve existing credential {name} for edit: {e}")
updated_fields = new_data.copy()
with _get_db_connection() as conn:
cursor = conn.cursor()
conn.row_factory = sqlite3.Row
if service == 'spotify':
# Prepare data for DB update
db_update_data = {
'region': updated_fields.get('region', existing_cred['region']),
'updated_at': current_time,
'name': name # for WHERE clause
}
blob_path = Path(existing_cred['blob_file_path']) # Use path from existing
original_blob_content = None
if blob_path.exists():
with open(blob_path, 'r') as f_orig_blob:
original_blob_content = f_orig_blob.read()
# If blob_content is being updated, write it temporarily for validation
if 'blob_content' in updated_fields:
blob_path.parent.mkdir(parents=True, exist_ok=True)
with open(blob_path, 'w') as f_new_blob:
if isinstance(updated_fields['blob_content'], dict):
json.dump(updated_fields['blob_content'], f_new_blob, indent=4)
else:
f_new_blob.write(updated_fields['blob_content'])
validation_data = {'blob_file_path': str(blob_path)}
try:
_validate_with_retry('spotify', name, validation_data)
set_clause = ", ".join([f"{key} = ?" for key in db_update_data if key != 'name'])
values = [db_update_data[key] for key in db_update_data if key != 'name'] + [name]
cursor.execute(f"UPDATE spotify SET {set_clause} WHERE name = ?", tuple(values))
# If validation passed and blob was in new_data, it's already written.
# If blob_content was NOT in new_data, the existing blob (if any) remains.
except Exception as e:
# Revert blob if it was changed and validation failed
if 'blob_content' in updated_fields and original_blob_content is not None:
with open(blob_path, 'w') as f_revert_blob:
f_revert_blob.write(original_blob_content)
elif 'blob_content' in updated_fields and original_blob_content is None and blob_path.exists():
# If new blob was written but there was no original to revert to, delete the new one.
blob_path.unlink()
raise # Re-raise validation or DB error
elif service == 'deezer':
db_update_data = {
'arl': updated_fields.get('arl', existing_cred['arl']),
'region': updated_fields.get('region', existing_cred['region']),
'updated_at': current_time,
'name': name # for WHERE clause
}
validation_data = {'arl': db_update_data['arl']}
_validate_with_retry('deezer', name, validation_data) # Validation happens before DB write for Deezer
set_clause = ", ".join([f"{key} = ?" for key in db_update_data if key != 'name'])
values = [db_update_data[key] for key in db_update_data if key != 'name'] + [name]
cursor.execute(f"UPDATE deezer SET {set_clause} WHERE name = ?", tuple(values))
if cursor.rowcount == 0: # Should not happen if get_credential succeeded
raise FileNotFoundError(f"Credential '{name}' for {service} disappeared during edit.")
conn.commit()
logger.info(f"Credential '{name}' for {service} updated successfully.")
return {"status": "updated", "service": service, "name": name}
# --- Helper for credential file path (mainly for Spotify blob) ---
def get_spotify_blob_path(account_name: str) -> Path:
return BLOBS_DIR / account_name / 'credentials.json'
# It's good practice to call init_credentials_db() when the app starts.
# This can be done in the main application setup. For now, defining it here.
# If this script is run directly for setup, you could add:
# if __name__ == '__main__':
# init_credentials_db()
# print("Credentials database initialized.")

122
routes/utils/get_info.py
View File

@@ -4,48 +4,63 @@ from deezspot.easy_spoty import Spo
import json
from pathlib import Path
from routes.utils.celery_queue_manager import get_config_params
from routes.utils.credentials import get_credential, _get_global_spotify_api_creds
# Import Deezer API and logging
from deezspot.deezloader.dee_api import API as DeezerAPI
import logging
# Initialize logger
logger = logging.getLogger(__name__)
# We'll rely on get_config_params() instead of directly loading the config file
def get_spotify_info(spotify_id, spotify_type, limit=None, offset=None):
"""
Get info from Spotify API using the default Spotify account configured in main.json
Get info from Spotify API. Uses global client_id/secret from search.json.
The default Spotify account from main.json might still be relevant for other Spo settings or if Spo uses it.
Args:
spotify_id: The Spotify ID of the entity
spotify_type: The type of entity (track, album, playlist, artist)
limit (int, optional): The maximum number of items to return. Only used if spotify_type is "artist".
offset (int, optional): The index of the first item to return. Only used if spotify_type is "artist".
spotify_type: The type of entity (track, album, playlist, artist, artist_discography, episode)
limit (int, optional): The maximum number of items to return. Only used if spotify_type is "artist_discography".
offset (int, optional): The index of the first item to return. Only used if spotify_type is "artist_discography".
Returns:
Dictionary with the entity information
"""
client_id = None
client_secret = None
client_id, client_secret = _get_global_spotify_api_creds()
# Get config parameters including Spotify account
if not client_id or not client_secret:
raise ValueError("Global Spotify API client_id or client_secret not configured in ./data/creds/search.json.")
# Get config parameters including default Spotify account name
# This might still be useful if Spo uses the account name for other things (e.g. market/region if not passed explicitly)
# For now, we are just ensuring the API keys are set.
config_params = get_config_params()
main = config_params.get('spotify', '')
main_spotify_account_name = config_params.get('spotify', '') # Still good to know which account is 'default' contextually
if not main:
raise ValueError("No Spotify account configured in settings")
if spotify_id:
search_creds_path = Path(f'./data/creds/spotify/{main}/search.json')
if search_creds_path.exists():
try:
with open(search_creds_path, 'r') as f:
search_creds = json.load(f)
client_id = search_creds.get('client_id')
client_secret = search_creds.get('client_secret')
except Exception as e:
print(f"Error loading search credentials: {e}")
# Initialize the Spotify client with credentials (if available)
if client_id and client_secret:
Spo.__init__(client_id, client_secret)
if not main_spotify_account_name:
# This is less critical now that API keys are global, but could indicate a misconfiguration
# if other parts of Spo expect an account context.
print(f"WARN: No default Spotify account name configured in settings (main.json). API calls will use global keys.")
else:
raise ValueError("No Spotify credentials found")
# Optionally, one could load the specific account's region here if Spo.init or methods need it,
# but easy_spoty's Spo doesn't seem to take region directly in __init__.
# It might use it internally based on account details if credentials.json (blob) contains it.
try:
# We call get_credential just to check if the account exists,
# not for client_id/secret anymore for Spo.__init__
get_credential('spotify', main_spotify_account_name)
except FileNotFoundError:
# This is a more serious warning if an account is expected to exist.
print(f"WARN: Default Spotify account '{main_spotify_account_name}' configured in main.json was not found in credentials database.")
except Exception as e:
print(f"WARN: Error accessing default Spotify account '{main_spotify_account_name}': {e}")
# Initialize the Spotify client with GLOBAL credentials
Spo.__init__(client_id, client_secret)
if spotify_type == "track":
return Spo.get_track(spotify_id)
elif spotify_type == "album":
@@ -67,3 +82,58 @@ def get_spotify_info(spotify_id, spotify_type, limit=None, offset=None):
return Spo.get_episode(spotify_id)
else:
raise ValueError(f"Unsupported Spotify type: {spotify_type}")
def get_deezer_info(deezer_id, deezer_type, limit=None):
"""
Get info from Deezer API.
Args:
deezer_id: The Deezer ID of the entity.
deezer_type: The type of entity (track, album, playlist, artist, episode,
artist_top_tracks, artist_albums, artist_related,
artist_radio, artist_playlists).
limit (int, optional): The maximum number of items to return. Used for
artist_top_tracks, artist_albums, artist_playlists.
Deezer API methods usually have their own defaults (e.g., 25)
if limit is not provided or None is passed to them.
Returns:
Dictionary with the entity information.
Raises:
ValueError: If deezer_type is unsupported.
Various exceptions from DeezerAPI (NoDataApi, QuotaExceeded, requests.exceptions.RequestException, etc.)
"""
logger.debug(f"Fetching Deezer info for ID {deezer_id}, type {deezer_type}, limit {limit}")
# DeezerAPI uses class methods; its @classmethod __init__ handles setup.
# No specific ARL or account handling here as DeezerAPI seems to use general endpoints.
if deezer_type == "track":
return DeezerAPI.get_track(deezer_id)
elif deezer_type == "album":
return DeezerAPI.get_album(deezer_id)
elif deezer_type == "playlist":
return DeezerAPI.get_playlist(deezer_id)
elif deezer_type == "artist":
return DeezerAPI.get_artist(deezer_id)
elif deezer_type == "episode":
return DeezerAPI.get_episode(deezer_id)
elif deezer_type == "artist_top_tracks":
if limit is not None:
return DeezerAPI.get_artist_top_tracks(deezer_id, limit=limit)
return DeezerAPI.get_artist_top_tracks(deezer_id) # Use API default limit
elif deezer_type == "artist_albums": # Maps to get_artist_top_albums
if limit is not None:
return DeezerAPI.get_artist_top_albums(deezer_id, limit=limit)
return DeezerAPI.get_artist_top_albums(deezer_id) # Use API default limit
elif deezer_type == "artist_related":
return DeezerAPI.get_artist_related(deezer_id)
elif deezer_type == "artist_radio":
return DeezerAPI.get_artist_radio(deezer_id)
elif deezer_type == "artist_playlists":
if limit is not None:
return DeezerAPI.get_artist_top_playlists(deezer_id, limit=limit)
return DeezerAPI.get_artist_top_playlists(deezer_id) # Use API default limit
else:
logger.error(f"Unsupported Deezer type: {deezer_type}")
raise ValueError(f"Unsupported Deezer type: {deezer_type}")

76
routes/utils/history_manager.py
View File

@@ -9,13 +9,40 @@ logger = logging.getLogger(__name__)
HISTORY_DIR = Path('./data/history')
HISTORY_DB_FILE = HISTORY_DIR / 'download_history.db'
EXPECTED_COLUMNS = {
'task_id': 'TEXT PRIMARY KEY',
'download_type': 'TEXT',
'item_name': 'TEXT',
'item_artist': 'TEXT',
'item_album': 'TEXT',
'item_url': 'TEXT',
'spotify_id': 'TEXT',
'status_final': 'TEXT', # 'COMPLETED', 'ERROR', 'CANCELLED'
'error_message': 'TEXT',
'timestamp_added': 'REAL',
'timestamp_completed': 'REAL',
'original_request_json': 'TEXT',
'last_status_obj_json': 'TEXT',
'service_used': 'TEXT',
'quality_profile': 'TEXT',
'convert_to': 'TEXT',
'bitrate': 'TEXT'
}
def init_history_db():
"""Initializes the download history database and creates the table if it doesn't exist."""
"""Initializes the download history database, creates the table if it doesn't exist,
and adds any missing columns to an existing table."""
conn = None
try:
HISTORY_DIR.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(HISTORY_DB_FILE)
cursor = conn.cursor()
cursor.execute("""
# Create table if it doesn't exist (idempotent)
# The primary key constraint is handled by the initial CREATE TABLE.
# If 'task_id' is missing, it cannot be added as PRIMARY KEY to an existing table
# without complex migrations. We assume 'task_id' will exist if the table exists.
create_table_sql = f"""
CREATE TABLE IF NOT EXISTS download_history (
task_id TEXT PRIMARY KEY,
download_type TEXT,
@@ -24,7 +51,7 @@ def init_history_db():
item_album TEXT,
item_url TEXT,
spotify_id TEXT,
status_final TEXT, -- 'COMPLETED', 'ERROR', 'CANCELLED'
status_final TEXT,
error_message TEXT,
timestamp_added REAL,
timestamp_completed REAL,
@@ -35,9 +62,48 @@ def init_history_db():
convert_to TEXT,
bitrate TEXT
)
""")
"""
cursor.execute(create_table_sql)
conn.commit()
logger.info(f"Download history database initialized at {HISTORY_DB_FILE}")
# Check for missing columns and add them
cursor.execute("PRAGMA table_info(download_history)")
existing_columns_info = cursor.fetchall()
existing_column_names = {col[1] for col in existing_columns_info}
added_columns = False
for col_name, col_type in EXPECTED_COLUMNS.items():
if col_name not in existing_column_names:
if 'PRIMARY KEY' in col_type.upper() and col_name == 'task_id':
# This case should be handled by CREATE TABLE, but as a safeguard:
# If task_id is somehow missing and table exists, this is a problem.
# Adding it as PK here is complex and might fail if data exists.
# For now, we assume CREATE TABLE handles the PK.
# If we were to add it, it would be 'ALTER TABLE download_history ADD COLUMN task_id TEXT;'
# and then potentially a separate step to make it PK if table is empty, which is non-trivial.
logger.warning(f"Column '{col_name}' is part of PRIMARY KEY and was expected to be created by CREATE TABLE. Skipping explicit ADD COLUMN.")
continue
# For other columns, just add them.
# Remove PRIMARY KEY from type definition if present, as it's only for table creation.
col_type_for_add = col_type.replace(' PRIMARY KEY', '').strip()
try:
cursor.execute(f"ALTER TABLE download_history ADD COLUMN {col_name} {col_type_for_add}")
logger.info(f"Added missing column '{col_name} {col_type_for_add}' to download_history table.")
added_columns = True
except sqlite3.OperationalError as alter_e:
# This might happen if a column (e.g. task_id) without "PRIMARY KEY" is added by this loop
# but the initial create table already made it a primary key.
# Or other more complex scenarios.
logger.warning(f"Could not add column '{col_name}': {alter_e}. It might already exist or there's a schema mismatch.")
if added_columns:
conn.commit()
logger.info(f"Download history table schema updated at {HISTORY_DB_FILE}")
else:
logger.info(f"Download history database schema is up-to-date at {HISTORY_DB_FILE}")
except sqlite3.Error as e:
logger.error(f"Error initializing download history database: {e}", exc_info=True)
finally:

188
routes/utils/playlist.py
View File

@@ -4,6 +4,8 @@ import traceback
from deezspot.spotloader import SpoLogin
from deezspot.deezloader import DeeLogin
from pathlib import Path
from routes.utils.credentials import get_credential, _get_global_spotify_api_creds
from routes.utils.celery_config import get_config_params
def download_playlist(
url,
@@ -28,83 +30,49 @@ def download_playlist(
is_spotify_url = 'open.spotify.com' in url.lower()
is_deezer_url = 'deezer.com' in url.lower()
# Determine service exclusively from URL
service = ''
if is_spotify_url:
service = 'spotify'
elif is_deezer_url:
service = 'deezer'
else:
# If URL can't be detected, raise an error
error_msg = "Invalid URL: Must be from open.spotify.com or deezer.com"
print(f"ERROR: {error_msg}")
raise ValueError(error_msg)
print(f"DEBUG: playlist.py - URL detection: is_spotify_url={is_spotify_url}, is_deezer_url={is_deezer_url}")
print(f"DEBUG: playlist.py - Service determined from URL: {service}")
print(f"DEBUG: playlist.py - Credentials: main={main}, fallback={fallback}")
# Load Spotify client credentials if available
spotify_client_id = None
spotify_client_secret = None
# Smartly determine where to look for Spotify search credentials
if service == 'spotify' and fallback:
# If fallback is enabled, use the fallback account for Spotify search credentials
search_creds_path = Path(f'./data/creds/spotify/{fallback}/search.json')
print(f"DEBUG: Using Spotify search credentials from fallback: {search_creds_path}")
else:
# Otherwise use the main account for Spotify search credentials
search_creds_path = Path(f'./data/creds/spotify/{main}/search.json')
print(f"DEBUG: Using Spotify search credentials from main: {search_creds_path}")
if search_creds_path.exists():
try:
with open(search_creds_path, 'r') as f:
search_creds = json.load(f)
spotify_client_id = search_creds.get('client_id')
spotify_client_secret = search_creds.get('client_secret')
print(f"DEBUG: Loaded Spotify client credentials successfully")
except Exception as e:
print(f"Error loading Spotify search credentials: {e}")
# For Spotify URLs: check if fallback is enabled, if so use the fallback logic,
# otherwise download directly from Spotify
print(f"DEBUG: playlist.py - Credentials provided: main_account_name='{main}', fallback_account_name='{fallback}'")
# Get global Spotify API credentials
global_spotify_client_id, global_spotify_client_secret = _get_global_spotify_api_creds()
if not global_spotify_client_id or not global_spotify_client_secret:
warning_msg = "WARN: playlist.py - Global Spotify client_id/secret not found in search.json. Spotify operations will likely fail."
print(warning_msg)
if service == 'spotify':
if fallback:
if quality is None:
quality = 'FLAC'
if fall_quality is None:
fall_quality = 'HIGH'
if fallback: # Fallback is a Deezer account name for a Spotify URL
if quality is None: quality = 'FLAC' # Deezer quality for first attempt
if fall_quality is None: fall_quality = 'HIGH' # Spotify quality for fallback (if Deezer fails)
# First attempt: use DeeLogin's download_playlistspo with the 'main' (Deezer credentials)
deezer_error = None
try:
# Load Deezer credentials from 'main' under deezer directory
deezer_creds_dir = os.path.join('./data/creds/deezer', main)
deezer_creds_path = os.path.abspath(os.path.join(deezer_creds_dir, 'credentials.json'))
# Attempt 1: Deezer via download_playlistspo (using 'fallback' as Deezer account name)
print(f"DEBUG: playlist.py - Spotify URL. Attempt 1: Deezer (account: {fallback})")
deezer_fallback_creds = get_credential('deezer', fallback)
arl = deezer_fallback_creds.get('arl')
if not arl:
raise ValueError(f"ARL not found for Deezer account '{fallback}'.")
# DEBUG: Print Deezer credential paths being used
print(f"DEBUG: Looking for Deezer credentials at:")
print(f"DEBUG: deezer_creds_dir = {deezer_creds_dir}")
print(f"DEBUG: deezer_creds_path = {deezer_creds_path}")
print(f"DEBUG: Directory exists = {os.path.exists(deezer_creds_dir)}")
print(f"DEBUG: Credentials file exists = {os.path.exists(deezer_creds_path)}")
with open(deezer_creds_path, 'r') as f:
deezer_creds = json.load(f)
# Initialize DeeLogin with Deezer credentials
dl = DeeLogin(
arl=deezer_creds.get('arl', ''),
spotify_client_id=spotify_client_id,
spotify_client_secret=spotify_client_secret,
arl=arl,
spotify_client_id=global_spotify_client_id,
spotify_client_secret=global_spotify_client_secret,
progress_callback=progress_callback
)
print(f"DEBUG: Starting playlist download using Deezer credentials (download_playlistspo)")
# Download using download_playlistspo; pass the custom formatting parameters.
dl.download_playlistspo(
link_playlist=url,
link_playlist=url, # Spotify URL
output_dir="./downloads",
quality_download=quality,
quality_download=quality, # Deezer quality
recursive_quality=True,
recursive_download=False,
not_interface=False,
@@ -119,35 +87,33 @@ def download_playlist(
convert_to=convert_to,
bitrate=bitrate
)
print(f"DEBUG: Playlist download completed successfully using Deezer credentials")
print(f"DEBUG: playlist.py - Playlist download via Deezer (account: {fallback}) successful for Spotify URL.")
except Exception as e:
deezer_error = e
# Immediately report the Deezer error
print(f"ERROR: Deezer playlist download attempt failed: {e}")
print(f"ERROR: playlist.py - Deezer attempt (account: {fallback}) for Spotify URL failed: {e}")
traceback.print_exc()
print("Attempting Spotify fallback...")
print(f"DEBUG: playlist.py - Attempting Spotify direct download (account: {main} for blob)...")
# Load fallback Spotify credentials and attempt download
# Attempt 2: Spotify direct via download_playlist (using 'main' as Spotify account for blob)
try:
spo_creds_dir = os.path.join('./data/creds/spotify', fallback)
spo_creds_path = os.path.abspath(os.path.join(spo_creds_dir, 'credentials.json'))
print(f"DEBUG: Using Spotify fallback credentials from: {spo_creds_path}")
print(f"DEBUG: Fallback credentials exist: {os.path.exists(spo_creds_path)}")
# We've already loaded the Spotify client credentials above based on fallback
if not global_spotify_client_id or not global_spotify_client_secret:
raise ValueError("Global Spotify API credentials (client_id/secret) not configured for Spotify download.")
spotify_main_creds = get_credential('spotify', main) # For blob path
blob_file_path = spotify_main_creds.get('blob_file_path')
if not Path(blob_file_path).exists():
raise FileNotFoundError(f"Spotify credentials blob file not found at {blob_file_path} for account '{main}'")
spo = SpoLogin(
credentials_path=spo_creds_path,
spotify_client_id=spotify_client_id,
spotify_client_secret=spotify_client_secret,
credentials_path=blob_file_path,
spotify_client_id=global_spotify_client_id,
spotify_client_secret=global_spotify_client_secret,
progress_callback=progress_callback
)
print(f"DEBUG: Starting playlist download using Spotify fallback credentials")
spo.download_playlist(
link_playlist=url,
link_playlist=url, # Spotify URL
output_dir="./downloads",
quality_download=fall_quality,
quality_download=fall_quality, # Spotify quality
recursive_quality=True,
recursive_download=False,
not_interface=False,
@@ -163,34 +129,36 @@ def download_playlist(
convert_to=convert_to,
bitrate=bitrate
)
print(f"DEBUG: Playlist download completed successfully using Spotify fallback")
print(f"DEBUG: playlist.py - Spotify direct download (account: {main} for blob) successful.")
except Exception as e2:
# If fallback also fails, raise an error indicating both attempts failed
print(f"ERROR: Spotify fallback also failed: {e2}")
print(f"ERROR: playlist.py - Spotify direct download (account: {main} for blob) also failed: {e2}")
raise RuntimeError(
f"Both main (Deezer) and fallback (Spotify) attempts failed. "
f"Both Deezer attempt (account: {fallback}) and Spotify direct (account: {main} for blob) failed. "
f"Deezer error: {deezer_error}, Spotify error: {e2}"
) from e2
else:
# Original behavior: use Spotify main
if quality is None:
quality = 'HIGH'
creds_dir = os.path.join('./data/creds/spotify', main)
credentials_path = os.path.abspath(os.path.join(creds_dir, 'credentials.json'))
print(f"DEBUG: Using Spotify main credentials from: {credentials_path}")
print(f"DEBUG: Credentials exist: {os.path.exists(credentials_path)}")
# Spotify URL, no fallback. Direct Spotify download using 'main' (Spotify account for blob)
if quality is None: quality = 'HIGH' # Default Spotify quality
print(f"DEBUG: playlist.py - Spotify URL, no fallback. Direct download with Spotify account (for blob): {main}")
if not global_spotify_client_id or not global_spotify_client_secret:
raise ValueError("Global Spotify API credentials (client_id/secret) not configured for Spotify download.")
spotify_main_creds = get_credential('spotify', main) # For blob path
blob_file_path = spotify_main_creds.get('blob_file_path')
if not Path(blob_file_path).exists():
raise FileNotFoundError(f"Spotify credentials blob file not found at {blob_file_path} for account '{main}'")
spo = SpoLogin(
credentials_path=credentials_path,
spotify_client_id=spotify_client_id,
spotify_client_secret=spotify_client_secret,
credentials_path=blob_file_path,
spotify_client_id=global_spotify_client_id,
spotify_client_secret=global_spotify_client_secret,
progress_callback=progress_callback
)
print(f"DEBUG: Starting playlist download using Spotify main credentials")
spo.download_playlist(
link_playlist=url,
output_dir="./downloads",
quality_download=quality,
quality_download=quality,
recursive_quality=True,
recursive_download=False,
not_interface=False,
@@ -206,31 +174,28 @@ def download_playlist(
convert_to=convert_to,
bitrate=bitrate
)
print(f"DEBUG: Playlist download completed successfully using Spotify main")
# For Deezer URLs: download directly from Deezer
print(f"DEBUG: playlist.py - Direct Spotify download (account: {main} for blob) successful.")
elif service == 'deezer':
if quality is None:
quality = 'FLAC'
# Existing code for Deezer, using main as Deezer account.
creds_dir = os.path.join('./data/creds/deezer', main)
creds_path = os.path.abspath(os.path.join(creds_dir, 'credentials.json'))
print(f"DEBUG: Using Deezer credentials from: {creds_path}")
print(f"DEBUG: Credentials exist: {os.path.exists(creds_path)}")
with open(creds_path, 'r') as f:
creds = json.load(f)
# Deezer URL. Direct Deezer download using 'main' (Deezer account name for ARL)
if quality is None: quality = 'FLAC' # Default Deezer quality
print(f"DEBUG: playlist.py - Deezer URL. Direct download with Deezer account: {main}")
deezer_main_creds = get_credential('deezer', main) # For ARL
arl = deezer_main_creds.get('arl')
if not arl:
raise ValueError(f"ARL not found for Deezer account '{main}'.")
dl = DeeLogin(
arl=creds.get('arl', ''),
spotify_client_id=spotify_client_id,
spotify_client_secret=spotify_client_secret,
arl=arl, # Account specific ARL
spotify_client_id=global_spotify_client_id, # Global Spotify keys
spotify_client_secret=global_spotify_client_secret, # Global Spotify keys
progress_callback=progress_callback
)
print(f"DEBUG: Starting playlist download using Deezer direct")
dl.download_playlistdee(
dl.download_playlistdee( # Deezer URL, download via Deezer
link_playlist=url,
output_dir="./downloads",
quality_download=quality,
recursive_quality=False,
recursive_quality=False, # Usually False for playlists to get individual track qualities
recursive_download=False,
make_zip=False,
custom_dir_format=custom_dir_format,
@@ -243,9 +208,10 @@ def download_playlist(
convert_to=convert_to,
bitrate=bitrate
)
print(f"DEBUG: Playlist download completed successfully using Deezer direct")
print(f"DEBUG: playlist.py - Direct Deezer download (account: {main}) successful.")
else:
raise ValueError(f"Unsupported service: {service}")
# Should be caught by initial service check, but as a safeguard
raise ValueError(f"Unsupported service determined: {service}")
except Exception as e:
print(f"ERROR: Playlist download failed with exception: {e}")
traceback.print_exc()

55
routes/utils/search.py
View File

@@ -2,6 +2,7 @@ from deezspot.easy_spoty import Spo
import json
from pathlib import Path
import logging
from routes.utils.credentials import get_credential, _get_global_spotify_api_creds
# Configure logger
logger = logging.getLogger(__name__)
@@ -12,48 +13,38 @@ def search(
limit: int = 3,
main: str = None
) -> dict:
logger.info(f"Search requested: query='{query}', type={search_type}, limit={limit}, main={main}")
logger.info(f"Search requested: query='{query}', type={search_type}, limit={limit}, main_account_name={main}")
# If main account is specified, load client ID and secret from the account's search.json
client_id = None
client_secret = None
client_id, client_secret = _get_global_spotify_api_creds()
if not client_id or not client_secret:
logger.error("Global Spotify API client_id or client_secret not configured in ./data/creds/search.json.")
raise ValueError("Spotify API credentials are not configured globally for search.")
if main:
search_creds_path = Path(f'./data/creds/spotify/{main}/search.json')
logger.debug(f"Looking for credentials at: {search_creds_path}")
if search_creds_path.exists():
try:
with open(search_creds_path, 'r') as f:
search_creds = json.load(f)
client_id = search_creds.get('client_id')
client_secret = search_creds.get('client_secret')
logger.debug(f"Credentials loaded successfully for account: {main}")
except Exception as e:
logger.error(f"Error loading search credentials: {e}")
print(f"Error loading search credentials: {e}")
else:
logger.warning(f"Credentials file not found at: {search_creds_path}")
# Initialize the Spotify client with credentials (if available)
if client_id and client_secret:
logger.debug("Initializing Spotify client with account credentials")
Spo.__init__(client_id, client_secret)
logger.debug(f"Spotify account context '{main}' was provided for search. API keys are global, but this account might be used for other context by Spo if relevant.")
try:
get_credential('spotify', main)
logger.debug(f"Spotify account '{main}' exists.")
except FileNotFoundError:
logger.warning(f"Spotify account '{main}' provided for search context not found in credentials. Search will proceed with global API keys.")
except Exception as e:
logger.warning(f"Error checking existence of Spotify account '{main}': {e}. Search will proceed with global API keys.")
else:
logger.debug("Using default Spotify client credentials")
logger.debug("No specific 'main' account context provided for search. Using global API keys.")
# Perform the Spotify search
logger.debug(f"Executing Spotify search with query='{query}', type={search_type}")
logger.debug(f"Initializing Spotify client with global API credentials for search.")
Spo.__init__(client_id, client_secret)
logger.debug(f"Executing Spotify search with query='{query}', type={search_type}, limit={limit}")
try:
spotify_response = Spo.search(
query=query,
search_type=search_type,
limit=limit,
client_id=client_id,
client_secret=client_secret
limit=limit
)
logger.info(f"Search completed successfully")
logger.info(f"Search completed successfully for query: '{query}'")
return spotify_response
except Exception as e:
logger.error(f"Error during Spotify search: {e}")
logger.error(f"Error during Spotify search for query '{query}': {e}", exc_info=True)
raise

176
routes/utils/track.py
View File

@@ -4,6 +4,8 @@ import traceback
from deezspot.spotloader import SpoLogin
from deezspot.deezloader import DeeLogin
from pathlib import Path
from routes.utils.credentials import get_credential, _get_global_spotify_api_creds, get_spotify_blob_path
from routes.utils.celery_config import get_config_params
def download_track(
url,
@@ -28,84 +30,53 @@ def download_track(
is_spotify_url = 'open.spotify.com' in url.lower()
is_deezer_url = 'deezer.com' in url.lower()
# Determine service exclusively from URL
service = ''
if is_spotify_url:
service = 'spotify'
elif is_deezer_url:
service = 'deezer'
else:
# If URL can't be detected, raise an error
error_msg = "Invalid URL: Must be from open.spotify.com or deezer.com"
print(f"ERROR: {error_msg}")
raise ValueError(error_msg)
print(f"DEBUG: track.py - URL detection: is_spotify_url={is_spotify_url}, is_deezer_url={is_deezer_url}")
print(f"DEBUG: track.py - Service determined from URL: {service}")
print(f"DEBUG: track.py - Credentials: main={main}, fallback={fallback}")
# Load Spotify client credentials if available
spotify_client_id = None
spotify_client_secret = None
# Smartly determine where to look for Spotify search credentials
if service == 'spotify' and fallback:
# If fallback is enabled, use the fallback account for Spotify search credentials
search_creds_path = Path(f'./data/creds/spotify/{fallback}/search.json')
print(f"DEBUG: Using Spotify search credentials from fallback: {search_creds_path}")
else:
# Otherwise use the main account for Spotify search credentials
search_creds_path = Path(f'./data/creds/spotify/{main}/search.json')
print(f"DEBUG: Using Spotify search credentials from main: {search_creds_path}")
if search_creds_path.exists():
try:
with open(search_creds_path, 'r') as f:
search_creds = json.load(f)
spotify_client_id = search_creds.get('client_id')
spotify_client_secret = search_creds.get('client_secret')
print(f"DEBUG: Loaded Spotify client credentials successfully")
except Exception as e:
print(f"Error loading Spotify search credentials: {e}")
# For Spotify URLs: check if fallback is enabled, if so use the fallback logic,
# otherwise download directly from Spotify
print(f"DEBUG: track.py - Credentials provided: main_account_name='{main}', fallback_account_name='{fallback}'")
# Get global Spotify API credentials for SpoLogin and DeeLogin (if it uses Spotify search)
global_spotify_client_id, global_spotify_client_secret = _get_global_spotify_api_creds()
if not global_spotify_client_id or not global_spotify_client_secret:
# This is a critical failure if Spotify operations are involved
warning_msg = "WARN: track.py - Global Spotify client_id/secret not found in search.json. Spotify operations will likely fail."
print(warning_msg)
# Depending on flow, might want to raise error here if service is 'spotify'
# For now, let it proceed and fail at SpoLogin/DeeLogin init if keys are truly needed and missing.
if service == 'spotify':
if fallback:
if quality is None:
quality = 'FLAC'
if fall_quality is None:
fall_quality = 'HIGH'
if fallback: # Fallback is a Deezer account name for a Spotify URL
if quality is None: quality = 'FLAC' # Deezer quality for first attempt
if fall_quality is None: fall_quality = 'HIGH' # Spotify quality for fallback (if Deezer fails)
# First attempt: use Deezer's download_trackspo with 'main' (Deezer credentials)
deezer_error = None
try:
deezer_creds_dir = os.path.join('./data/creds/deezer', main)
deezer_creds_path = os.path.abspath(os.path.join(deezer_creds_dir, 'credentials.json'))
# Attempt 1: Deezer via download_trackspo (using 'fallback' as Deezer account name)
print(f"DEBUG: track.py - Spotify URL. Attempt 1: Deezer (account: {fallback})")
deezer_fallback_creds = get_credential('deezer', fallback)
arl = deezer_fallback_creds.get('arl')
if not arl:
raise ValueError(f"ARL not found for Deezer account '{fallback}'.")
# DEBUG: Print Deezer credential paths being used
print(f"DEBUG: Looking for Deezer credentials at:")
print(f"DEBUG: deezer_creds_dir = {deezer_creds_dir}")
print(f"DEBUG: deezer_creds_path = {deezer_creds_path}")
print(f"DEBUG: Directory exists = {os.path.exists(deezer_creds_dir)}")
print(f"DEBUG: Credentials file exists = {os.path.exists(deezer_creds_path)}")
# List available directories to compare
print(f"DEBUG: Available Deezer credential directories:")
for dir_name in os.listdir('./data/creds/deezer'):
print(f"DEBUG: ./data/creds/deezer/{dir_name}")
with open(deezer_creds_path, 'r') as f:
deezer_creds = json.load(f)
dl = DeeLogin(
arl=deezer_creds.get('arl', ''),
spotify_client_id=spotify_client_id,
spotify_client_secret=spotify_client_secret,
arl=arl,
spotify_client_id=global_spotify_client_id, # Global creds
spotify_client_secret=global_spotify_client_secret, # Global creds
progress_callback=progress_callback
)
# download_trackspo means: Spotify URL, download via Deezer
dl.download_trackspo(
link_track=url,
link_track=url, # Spotify URL
output_dir="./downloads",
quality_download=quality,
quality_download=quality, # Deezer quality
recursive_quality=False,
recursive_download=False,
not_interface=False,
@@ -118,30 +89,33 @@ def download_track(
convert_to=convert_to,
bitrate=bitrate
)
print(f"DEBUG: track.py - Track download via Deezer (account: {fallback}) successful for Spotify URL.")
except Exception as e:
deezer_error = e
# Immediately report the Deezer error
print(f"ERROR: Deezer download attempt failed: {e}")
print(f"ERROR: track.py - Deezer attempt (account: {fallback}) for Spotify URL failed: {e}")
traceback.print_exc()
print("Attempting Spotify fallback...")
print(f"DEBUG: track.py - Attempting Spotify direct download (account: {main})...")
# If the first attempt fails, use the fallback Spotify credentials
# Attempt 2: Spotify direct via download_track (using 'main' as Spotify account for blob)
try:
spo_creds_dir = os.path.join('./data/creds/spotify', fallback)
spo_creds_path = os.path.abspath(os.path.join(spo_creds_dir, 'credentials.json'))
# We've already loaded the Spotify client credentials above based on fallback
if not global_spotify_client_id or not global_spotify_client_secret:
raise ValueError("Global Spotify API credentials (client_id/secret) not configured for Spotify download.")
# Use get_spotify_blob_path directly
blob_file_path = get_spotify_blob_path(main)
if not blob_file_path.exists(): # Check existence on the Path object
raise FileNotFoundError(f"Spotify credentials blob file not found at {str(blob_file_path)} for account '{main}'")
spo = SpoLogin(
credentials_path=spo_creds_path,
spotify_client_id=spotify_client_id,
spotify_client_secret=spotify_client_secret,
credentials_path=str(blob_file_path), # Account specific blob
spotify_client_id=global_spotify_client_id, # Global API keys
spotify_client_secret=global_spotify_client_secret, # Global API keys
progress_callback=progress_callback
)
spo.download_track(
link_track=url,
link_track=url, # Spotify URL
output_dir="./downloads",
quality_download=fall_quality,
quality_download=fall_quality, # Spotify quality
recursive_quality=False,
recursive_download=False,
not_interface=False,
@@ -156,28 +130,36 @@ def download_track(
convert_to=convert_to,
bitrate=bitrate
)
print(f"DEBUG: track.py - Spotify direct download (account: {main} for blob) successful.")
except Exception as e2:
# If fallback also fails, raise an error indicating both attempts failed
print(f"ERROR: track.py - Spotify direct download (account: {main} for blob) also failed: {e2}")
raise RuntimeError(
f"Both main (Deezer) and fallback (Spotify) attempts failed. "
f"Both Deezer attempt (account: {fallback}) and Spotify direct (account: {main} for blob) failed. "
f"Deezer error: {deezer_error}, Spotify error: {e2}"
) from e2
else:
# Directly use Spotify main account
if quality is None:
quality = 'HIGH'
creds_dir = os.path.join('./data/creds/spotify', main)
credentials_path = os.path.abspath(os.path.join(creds_dir, 'credentials.json'))
# Spotify URL, no fallback. Direct Spotify download using 'main' (Spotify account for blob)
if quality is None: quality = 'HIGH' # Default Spotify quality
print(f"DEBUG: track.py - Spotify URL, no fallback. Direct download with Spotify account (for blob): {main}")
if not global_spotify_client_id or not global_spotify_client_secret:
raise ValueError("Global Spotify API credentials (client_id/secret) not configured for Spotify download.")
# Use get_spotify_blob_path directly
blob_file_path = get_spotify_blob_path(main)
if not blob_file_path.exists(): # Check existence on the Path object
raise FileNotFoundError(f"Spotify credentials blob file not found at {str(blob_file_path)} for account '{main}'")
spo = SpoLogin(
credentials_path=credentials_path,
spotify_client_id=spotify_client_id,
spotify_client_secret=spotify_client_secret,
credentials_path=str(blob_file_path), # Account specific blob
spotify_client_id=global_spotify_client_id, # Global API keys
spotify_client_secret=global_spotify_client_secret, # Global API keys
progress_callback=progress_callback
)
spo.download_track(
link_track=url,
output_dir="./downloads",
quality_download=quality,
quality_download=quality,
recursive_quality=False,
recursive_download=False,
not_interface=False,
@@ -192,22 +174,24 @@ def download_track(
convert_to=convert_to,
bitrate=bitrate
)
# For Deezer URLs: download directly from Deezer
print(f"DEBUG: track.py - Direct Spotify download (account: {main} for blob) successful.")
elif service == 'deezer':
if quality is None:
quality = 'FLAC'
# Deezer download logic remains unchanged, with the custom formatting parameters passed along.
creds_dir = os.path.join('./data/creds/deezer', main)
creds_path = os.path.abspath(os.path.join(creds_dir, 'credentials.json'))
with open(creds_path, 'r') as f:
creds = json.load(f)
# Deezer URL. Direct Deezer download using 'main' (Deezer account name for ARL)
if quality is None: quality = 'FLAC' # Default Deezer quality
print(f"DEBUG: track.py - Deezer URL. Direct download with Deezer account: {main}")
deezer_main_creds = get_credential('deezer', main) # For ARL
arl = deezer_main_creds.get('arl')
if not arl:
raise ValueError(f"ARL not found for Deezer account '{main}'.")
dl = DeeLogin(
arl=creds.get('arl', ''),
spotify_client_id=spotify_client_id,
spotify_client_secret=spotify_client_secret,
arl=arl, # Account specific ARL
spotify_client_id=global_spotify_client_id, # Global Spotify keys for internal Spo use by DeeLogin
spotify_client_secret=global_spotify_client_secret, # Global Spotify keys
progress_callback=progress_callback
)
dl.download_trackdee(
dl.download_trackdee( # Deezer URL, download via Deezer
link_track=url,
output_dir="./downloads",
quality_download=quality,
@@ -223,8 +207,10 @@ def download_track(
convert_to=convert_to,
bitrate=bitrate
)
print(f"DEBUG: track.py - Direct Deezer download (account: {main}) successful.")
else:
raise ValueError(f"Unsupported service: {service}")
# Should be caught by initial service check, but as a safeguard
raise ValueError(f"Unsupported service determined: {service}")
except Exception as e:
traceback.print_exc()
raise

159
routes/utils/watch/db.py
View File

@@ -14,6 +14,101 @@ ARTISTS_DB_PATH = DB_DIR / 'artists.db'
# Config path for watch.json is managed in routes.utils.watch.manager now
# CONFIG_PATH = Path('./data/config/watch.json') # Removed
# Expected column definitions
EXPECTED_WATCHED_PLAYLISTS_COLUMNS = {
'spotify_id': 'TEXT PRIMARY KEY',
'name': 'TEXT',
'owner_id': 'TEXT',
'owner_name': 'TEXT',
'total_tracks': 'INTEGER',
'link': 'TEXT',
'snapshot_id': 'TEXT',
'last_checked': 'INTEGER',
'added_at': 'INTEGER',
'is_active': 'INTEGER DEFAULT 1'
}
EXPECTED_PLAYLIST_TRACKS_COLUMNS = {
'spotify_track_id': 'TEXT PRIMARY KEY',
'title': 'TEXT',
'artist_names': 'TEXT',
'album_name': 'TEXT',
'album_artist_names': 'TEXT',
'track_number': 'INTEGER',
'album_spotify_id': 'TEXT',
'duration_ms': 'INTEGER',
'added_at_playlist': 'TEXT',
'added_to_db': 'INTEGER',
'is_present_in_spotify': 'INTEGER DEFAULT 1',
'last_seen_in_spotify': 'INTEGER'
}
EXPECTED_WATCHED_ARTISTS_COLUMNS = {
'spotify_id': 'TEXT PRIMARY KEY',
'name': 'TEXT',
'link': 'TEXT',
'total_albums_on_spotify': 'INTEGER', # Number of albums found via API
'last_checked': 'INTEGER',
'added_at': 'INTEGER',
'is_active': 'INTEGER DEFAULT 1',
'genres': 'TEXT', # Comma-separated
'popularity': 'INTEGER',
'image_url': 'TEXT'
}
EXPECTED_ARTIST_ALBUMS_COLUMNS = {
'album_spotify_id': 'TEXT PRIMARY KEY',
'artist_spotify_id': 'TEXT', # Foreign key to watched_artists
'name': 'TEXT',
'album_group': 'TEXT', # album, single, compilation, appears_on
'album_type': 'TEXT', # album, single, compilation
'release_date': 'TEXT',
'release_date_precision': 'TEXT', # year, month, day
'total_tracks': 'INTEGER',
'link': 'TEXT',
'image_url': 'TEXT',
'added_to_db': 'INTEGER',
'last_seen_on_spotify': 'INTEGER', # Timestamp when last confirmed via API
'download_task_id': 'TEXT',
'download_status': 'INTEGER DEFAULT 0', # 0: Not Queued, 1: Queued/In Progress, 2: Downloaded, 3: Error
'is_fully_downloaded_managed_by_app': 'INTEGER DEFAULT 0' # 0: No, 1: Yes (app has marked all its tracks as downloaded)
}
def _ensure_table_schema(cursor: sqlite3.Cursor, table_name: str, expected_columns: dict, table_description: str):
"""
Ensures the given table has all expected columns, adding them if necessary.
"""
try:
cursor.execute(f"PRAGMA table_info({table_name})")
existing_columns_info = cursor.fetchall()
existing_column_names = {col[1] for col in existing_columns_info}
added_columns_to_this_table = False
for col_name, col_type in expected_columns.items():
if col_name not in existing_column_names:
if 'PRIMARY KEY' in col_type.upper() and existing_columns_info: # Only warn if table already exists
logger.warning(
f"Column '{col_name}' is part of PRIMARY KEY for {table_description} '{table_name}' "
f"and was expected to be created by CREATE TABLE. Skipping explicit ADD COLUMN. "
f"Manual schema review might be needed if this table was not empty."
)
continue
col_type_for_add = col_type.replace(' PRIMARY KEY', '').strip()
try:
cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {col_name} {col_type_for_add}")
logger.info(f"Added missing column '{col_name} {col_type_for_add}' to {table_description} table '{table_name}'.")
added_columns_to_this_table = True
except sqlite3.OperationalError as alter_e:
logger.warning(
f"Could not add column '{col_name}' to {table_description} table '{table_name}': {alter_e}. "
f"It might already exist with a different definition or there's another schema mismatch."
)
return added_columns_to_this_table
except sqlite3.Error as e:
logger.error(f"Error ensuring schema for {table_description} table '{table_name}': {e}", exc_info=True)
return False
def _get_playlists_db_connection():
DB_DIR.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(PLAYLISTS_DB_PATH, timeout=10)
@@ -27,7 +122,7 @@ def _get_artists_db_connection():
return conn
def init_playlists_db():
"""Initializes the playlists database and creates the main watched_playlists table if it doesn't exist."""
"""Initializes the playlists database and creates/updates the main watched_playlists table."""
try:
with _get_playlists_db_connection() as conn:
cursor = conn.cursor()
@@ -45,15 +140,17 @@ def init_playlists_db():
is_active INTEGER DEFAULT 1
)
""")
conn.commit()
logger.info(f"Playlists database initialized successfully at {PLAYLISTS_DB_PATH}")
# Ensure schema
if _ensure_table_schema(cursor, 'watched_playlists', EXPECTED_WATCHED_PLAYLISTS_COLUMNS, "watched playlists"):
conn.commit()
logger.info(f"Playlists database initialized/updated successfully at {PLAYLISTS_DB_PATH}")
except sqlite3.Error as e:
logger.error(f"Error initializing watched_playlists table: {e}", exc_info=True)
raise
def _create_playlist_tracks_table(playlist_spotify_id: str):
"""Creates a table for a specific playlist to store its tracks if it doesn't exist in playlists.db."""
table_name = f"playlist_{playlist_spotify_id.replace('-', '_')}" # Sanitize table name
"""Creates or updates a table for a specific playlist to store its tracks in playlists.db."""
table_name = f"playlist_{playlist_spotify_id.replace('-', '_').replace(' ', '_')}" # Sanitize table name
try:
with _get_playlists_db_connection() as conn: # Use playlists connection
cursor = conn.cursor()
@@ -73,8 +170,10 @@ def _create_playlist_tracks_table(playlist_spotify_id: str):
last_seen_in_spotify INTEGER -- Timestamp when last confirmed in Spotify playlist
)
""")
conn.commit()
logger.info(f"Tracks table '{table_name}' created or already exists in {PLAYLISTS_DB_PATH}.")
# Ensure schema
if _ensure_table_schema(cursor, table_name, EXPECTED_PLAYLIST_TRACKS_COLUMNS, f"playlist tracks ({playlist_spotify_id})"):
conn.commit()
logger.info(f"Tracks table '{table_name}' created/updated or already exists in {PLAYLISTS_DB_PATH}.")
except sqlite3.Error as e:
logger.error(f"Error creating playlist tracks table {table_name} in {PLAYLISTS_DB_PATH}: {e}", exc_info=True)
raise
@@ -388,50 +487,66 @@ def add_single_track_to_playlist_db(playlist_spotify_id: str, track_item_for_db:
# --- Artist Watch Database Functions ---
def init_artists_db():
"""Initializes the artists database and creates the watched_artists table if it doesn't exist."""
"""Initializes the artists database and creates/updates the main watched_artists table."""
try:
with _get_artists_db_connection() as conn:
cursor = conn.cursor()
# Note: total_albums_on_spotify, genres, popularity, image_url added to EXPECTED_WATCHED_ARTISTS_COLUMNS
# and will be added by _ensure_table_schema if missing.
cursor.execute("""
CREATE TABLE IF NOT EXISTS watched_artists (
spotify_id TEXT PRIMARY KEY,
name TEXT,
total_albums_on_spotify INTEGER,
link TEXT,
total_albums_on_spotify INTEGER, -- Number of albums found via API on last full check
last_checked INTEGER,
added_at INTEGER,
is_active INTEGER DEFAULT 1,
last_known_status TEXT,
last_task_id TEXT
genres TEXT, -- Comma-separated list of genres
popularity INTEGER, -- Artist popularity (0-100)
image_url TEXT -- URL of the artist's image
)
""")
conn.commit()
logger.info(f"Artists database initialized successfully at {ARTISTS_DB_PATH}")
# Ensure schema
if _ensure_table_schema(cursor, 'watched_artists', EXPECTED_WATCHED_ARTISTS_COLUMNS, "watched artists"):
conn.commit()
logger.info(f"Artists database initialized/updated successfully at {ARTISTS_DB_PATH}")
except sqlite3.Error as e:
logger.error(f"Error initializing watched_artists table in {ARTISTS_DB_PATH}: {e}", exc_info=True)
raise
def _create_artist_albums_table(artist_spotify_id: str):
"""Creates a table for a specific artist to store its albums if it doesn't exist in artists.db."""
table_name = f"artist_{artist_spotify_id.replace('-', '_')}_albums"
"""Creates or updates a table for a specific artist to store their albums in artists.db."""
table_name = f"artist_{artist_spotify_id.replace('-', '_').replace(' ', '_')}" # Sanitize table name
try:
with _get_artists_db_connection() as conn:
with _get_artists_db_connection() as conn: # Use artists connection
cursor = conn.cursor()
# Note: Several columns including artist_spotify_id, release_date_precision, image_url,
# last_seen_on_spotify, download_task_id, download_status, is_fully_downloaded_managed_by_app
# are part of EXPECTED_ARTIST_ALBUMS_COLUMNS and will be added by _ensure_table_schema.
cursor.execute(f"""
CREATE TABLE IF NOT EXISTS {table_name} (
album_spotify_id TEXT PRIMARY KEY,
artist_spotify_id TEXT,
name TEXT,
album_group TEXT,
album_type TEXT,
release_date TEXT,
release_date_precision TEXT,
total_tracks INTEGER,
added_to_db_at INTEGER,
is_download_initiated INTEGER DEFAULT 0,
task_id TEXT,
last_checked_for_tracks INTEGER
link TEXT,
image_url TEXT,
added_to_db INTEGER,
last_seen_on_spotify INTEGER,
download_task_id TEXT,
download_status INTEGER DEFAULT 0,
is_fully_downloaded_managed_by_app INTEGER DEFAULT 0
)
""")
conn.commit()
logger.info(f"Albums table '{table_name}' for artist {artist_spotify_id} created or exists in {ARTISTS_DB_PATH}.")
# Ensure schema for the specific artist's album table
if _ensure_table_schema(cursor, table_name, EXPECTED_ARTIST_ALBUMS_COLUMNS, f"artist albums ({artist_spotify_id})"):
conn.commit()
logger.info(f"Albums table '{table_name}' created/updated or already exists in {ARTISTS_DB_PATH}.")
except sqlite3.Error as e:
logger.error(f"Error creating artist albums table {table_name} in {ARTISTS_DB_PATH}: {e}", exc_info=True)
raise