af5a2bf825
- Integrated `plexapi` and `python-dotenv` as dependencies in `flake.nix` and `pyproject.toml` for enhanced functionality. - Implemented new modules for audio verification and duplicate tracking, including `audio_verification.py`, `duplicate_finder.py`, and `track_verification.py`. - Updated `main.py` to utilize the new modules for identifying and managing duplicate single tracks in Lidarr, with detailed logging and confidence scoring. - Enhanced the `find_duplicate_singles` function to support audio verification results and metadata migration to Plex. - Refactored existing code for improved structure and maintainability, ensuring better integration of new features.
90 lines
2.5 KiB
Python
90 lines
2.5 KiB
Python
"""Lidarr API client functions"""
|
|
|
|
import logging
|
|
from typing import Dict, List, Optional
|
|
|
|
import requests
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_json(
|
|
url: str,
|
|
headers: Dict[str, str],
|
|
params: Optional[Dict[str, object]] = None,
|
|
raise_on_error: bool = True,
|
|
) -> List[Dict]:
|
|
"""Fetch JSON from URL with error handling"""
|
|
try:
|
|
resp = requests.get(url, headers=headers, params=params, timeout=60)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except requests.exceptions.RequestException as e:
|
|
logger.warning(f"Error fetching {url}: {e}")
|
|
if raise_on_error:
|
|
raise
|
|
return []
|
|
|
|
|
|
def get_trackfile_info(
|
|
base_url: str, track_file_id: int, headers: Dict[str, str]
|
|
) -> Optional[Dict]:
|
|
"""Get trackfile information including file path and quality"""
|
|
try:
|
|
resp = requests.get(
|
|
f"{base_url.rstrip('/')}/api/v1/trackfile/{track_file_id}",
|
|
headers=headers,
|
|
timeout=30,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except requests.exceptions.RequestException as e:
|
|
logger.warning(f"Could not fetch trackfile {track_file_id}: {e}")
|
|
return None
|
|
|
|
|
|
def get_track_info(
|
|
base_url: str, track_id: int, headers: Dict[str, str]
|
|
) -> Optional[Dict]:
|
|
"""Get track information including MusicBrainz recording ID"""
|
|
try:
|
|
resp = requests.get(
|
|
f"{base_url.rstrip('/')}/api/v1/track/{track_id}",
|
|
headers=headers,
|
|
timeout=30,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except requests.exceptions.RequestException as e:
|
|
logger.warning(f"Could not fetch track {track_id}: {e}")
|
|
return None
|
|
|
|
|
|
def fetch_all_artists(base_url: str, headers: Dict[str, str]) -> List[Dict]:
|
|
"""Fetch all artists from Lidarr"""
|
|
return get_json(f"{base_url}/api/v1/artist", headers)
|
|
|
|
|
|
def fetch_albums_for_artist(
|
|
base_url: str, headers: Dict[str, str], artist_id: int
|
|
) -> List[Dict]:
|
|
"""Fetch all albums for an artist"""
|
|
return get_json(
|
|
f"{base_url}/api/v1/album",
|
|
headers,
|
|
params={"artistId": artist_id},
|
|
raise_on_error=False,
|
|
)
|
|
|
|
|
|
def fetch_tracks_for_album(
|
|
base_url: str, headers: Dict[str, str], album_id: int
|
|
) -> List[Dict]:
|
|
"""Fetch all tracks for an album"""
|
|
return get_json(
|
|
f"{base_url.rstrip('/')}/api/v1/track",
|
|
headers,
|
|
params={"albumId": album_id},
|
|
raise_on_error=False,
|
|
)
|