128 lines
4.9 KiB
Python
128 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
|
import requests
|
|
import json
|
|
import time
|
|
import logging
|
|
import sys
|
|
import os
|
|
import musicbrainzngs
|
|
from typing import List, Dict, Optional
|
|
from flask import Flask, jsonify
|
|
import threading
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Logging
|
|
# ----------------------------------------------------------------------
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s',
|
|
handlers=[
|
|
logging.FileHandler('/app/logs/lidarr_api.log'),
|
|
logging.StreamHandler(sys.stdout)
|
|
]
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Global state and Flask app
|
|
# ----------------------------------------------------------------------
|
|
app = Flask(__name__)
|
|
artist_ids = []
|
|
|
|
# Configure MusicBrainzngs
|
|
musicbrainzngs.set_useragent(
|
|
"Maloja-Lidarr-Sync-API",
|
|
"1.0",
|
|
"your-email@example.com"
|
|
)
|
|
|
|
# ----------------------------------------------------------------------
|
|
# MusicBrainz API helpers
|
|
# ----------------------------------------------------------------------
|
|
def get_artist_id_from_recording(recording_mbid: str) -> Optional[str]:
|
|
"""Finds the artist MBID for a given recording MBID."""
|
|
try:
|
|
result = musicbrainzngs.get_recording_by_id(recording_mbid, includes=["artist-credits"])
|
|
if result.get('recording', {}).get('artist-credit', []):
|
|
artist = result['recording']['artist-credit'][0]
|
|
if artist.get('artist', {}).get('id'):
|
|
return artist['artist']['id']
|
|
except musicbrainzngs.MusicBrainzError as e:
|
|
logger.error(f"MusicBrainz API error fetching recording {recording_mbid}: {e}")
|
|
except Exception as e:
|
|
logger.error(f"An unexpected error occurred: {e}")
|
|
return None
|
|
|
|
def get_artist_id_by_name(artist_name: str) -> Optional[str]:
|
|
"""Finds the artist MBID by searching for their name."""
|
|
try:
|
|
result = musicbrainzngs.search_artists(artist=artist_name, limit=1)
|
|
if result.get('artist-list', []):
|
|
return result['artist-list'][0].get('id')
|
|
except musicbrainzngs.MusicBrainzError as e:
|
|
logger.error(f"MusicBrainz API error searching for artist '{artist_name}': {e}")
|
|
except Exception as e:
|
|
logger.error(f"An unexpected error occurred: {e}")
|
|
return None
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Main logic to fetch and process artists
|
|
# ----------------------------------------------------------------------
|
|
def fetch_and_process_artists():
|
|
"""Fetches tracks from Maloja and populates the global artist_ids list."""
|
|
global artist_ids
|
|
while True:
|
|
try:
|
|
api_url = os.getenv("SOURCE_API_URL")
|
|
api_key = os.getenv("SOURCE_API_KEY")
|
|
|
|
if not api_url or not api_key:
|
|
logger.error("API URL or key not set. Exiting thread.")
|
|
break
|
|
|
|
resp = requests.get(api_url, headers={'x-api-key': api_key}, params={'key': api_key}, timeout=15)
|
|
resp.raise_for_status()
|
|
tracks_data = resp.json().get('list', [])
|
|
|
|
unique_artist_ids = set()
|
|
for track in tracks_data:
|
|
if not isinstance(track, dict):
|
|
continue
|
|
|
|
recording_mbid = track.get("track", {}).get("recording_mbid")
|
|
artist_name = track.get("track", {}).get("artists", [""])[0]
|
|
|
|
if recording_mbid:
|
|
artist_id = get_artist_id_from_recording(recording_mbid)
|
|
if artist_id:
|
|
unique_artist_ids.add(artist_id)
|
|
elif artist_name:
|
|
artist_id = get_artist_id_by_name(artist_name)
|
|
if artist_id:
|
|
unique_artist_ids.add(artist_id)
|
|
|
|
artist_ids = sorted(list(unique_artist_ids))
|
|
logger.info(f"Artist list updated. Found {len(artist_ids)} unique artist IDs.")
|
|
except Exception as e:
|
|
logger.error(f"An error occurred in the fetch thread: {e}")
|
|
|
|
time.sleep(3600)
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Flask routes
|
|
# ----------------------------------------------------------------------
|
|
@app.route('/artists')
|
|
def get_artists():
|
|
if not artist_ids:
|
|
return jsonify([])
|
|
|
|
lidarr_list = [{"foreignId": artist_id} for artist_id in artist_ids]
|
|
return jsonify(lidarr_list)
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Main execution
|
|
# ----------------------------------------------------------------------
|
|
if __name__ == '__main__':
|
|
thread = threading.Thread(target=fetch_and_process_artists, daemon=True)
|
|
thread.start()
|
|
app.run(host='0.0.0.0', port=5000) |