musicfetch/musicfetch

#!/usr/bin/env python3
import json
import sys
import os
import re
import subprocess
import requests
from requests.exceptions import Timeout

# === CONFIGURATION ===
LIDARR_URL = "http://localhost:8686"  # Your Lidarr base URL
API_KEY = os.environ.get("LIDARR_API_KEY", "")      # Your Lidarr API key
ROOT_FOLDER = "/media/music"

headers = {
    "X-Api-Key": API_KEY,
    "Content-Type": "application/json"
}

def is_url(string):
    return re.match(r'https?://', string)

def run_yt_dlp_get_metadata(url):
    try:
        result = subprocess.run(
            ["yt-dlp", "-j", "--no-playlist", url],
            capture_output=True,
            text=True,
            check=True
        )
        metadata = json.loads(result.stdout)
        return metadata  # Return full metadata dict here
    except subprocess.CalledProcessError as e:
        print(f"yt-dlp subprocess error: {e}")
        print(f"stderr: {e.stderr}")
    except json.JSONDecodeError as e:
        print(f"JSON decoding error: {e}")
        print(f"Output was: {result.stdout}")
    except Exception as e:
        print(f"Failed to extract metadata from URL: {e}")
    return None

def get_artist_from_metadata(metadata):
    # Try different possible keys for artist/creator in metadata
    artist = None
    for key in ["artist", "creator", "uploader", "channel"]:
        artist = metadata.get(key)
        if artist:
            break
    # fallback to title parsing if nothing found
    if not artist and "title" in metadata:
        # Try to parse artist from title "Artist - Track"
        parts = metadata["title"].split(" - ", 1)
        if len(parts) == 2:
            artist = parts[0].strip()
    return artist or "Unknown Artist"

def yt_dlp_download(url_or_query, target_folder):
    print(f"Downloading via yt-dlp to: {target_folder}")
    os.makedirs(target_folder, exist_ok=True)
    subprocess.run([
        "yt-dlp",
        "--embed-metadata",
        "--parse-metadata", "%(title)s:%(album)s",
        "--embed-thumbnail",
        "-f", "bestaudio",
        "-x",
        "-P", target_folder,
        url_or_query
    ])

def search_artist(name, timeout_seconds=15):
    try:
        resp = requests.get(
            f"{LIDARR_URL}/api/v1/artist/lookup",
            headers=headers,
            params={"term": name},
            timeout=timeout_seconds
        )
        resp.raise_for_status()
        results = resp.json()
        return results[0] if results else None
    except Timeout:
        print(f"Lidarr artist search timed out after {timeout_seconds} seconds.")
        return None
    except requests.RequestException as e:
        print(f"Lidarr artist search failed: {e}")
        return None

def get_existing_artist(name):
    print(f"--> Checking if artist '{name}' already exists in Lidarr...")
    try:
        print("...Sending request to /api/v1/artist")
        resp = requests.get(f"{LIDARR_URL}/api/v1/artist", headers=headers, timeout=10)
        print("...Got response from Lidarr")
        resp.raise_for_status()
        for artist in resp.json():
            if artist["artistName"].lower() == name.lower():
                print("...Artist match found!")
                return artist
        print("...Artist not found in existing list.")
    except requests.exceptions.Timeout:
        print("!!! Timeout during existing artist check.")
    except requests.exceptions.RequestException as e:
        print(f"!!! Exception during existing artist check: {e}")
    return None

def add_artist(metadata_artist):
    foreign_id = metadata_artist.get("foreignArtistId") or metadata_artist.get("id")
    artist_name = metadata_artist.get("artistName") or metadata_artist.get("title")

    if not foreign_id or not artist_name:
        raise ValueError("Could not find foreignArtistId or artistName in metadata.")

    data = {
        "foreignArtistId": foreign_id,
        "artistName": artist_name,
        "qualityProfileId": 1,
        "rootFolderPath": ROOT_FOLDER,
        "monitored": True,
        "addOptions": {
            "searchForMissingAlbums": True,
            "monitor": "all"
        }
    }

    resp = requests.post(
        f"{LIDARR_URL}/api/v1/artist",
        headers=headers,
        json=data
    )
    resp.raise_for_status()
    return resp.json()

def search_local_album(track_name, artist_id=None, artist_name=None):
    try:
        print(f"--> Searching local albums for track: '{track_name}'")
        resp = requests.get(
            f"{LIDARR_URL}/api/v1/album",
            headers=headers,
            timeout=10
        )
        resp.raise_for_status()
        albums = resp.json()
        print(f"...Loaded {len(albums)} local albums")

        # Search by artist_id and track_name
        for album in albums:
            # Check artist ID match if provided
            if artist_id and album.get("artistId") != artist_id:
                continue
            # Check artist name match if provided
            if artist_name and album.get("artist", {}).get("artistName", "").lower() != artist_name.lower():
                continue
            # Check if track_name matches album title (case insensitive substring)
            if track_name.lower() in album.get("title", "").lower():
                print(f"!!! Found local album: {album.get('title')} by {album.get('artist', {}).get('artistName')}")
                return album
        print("!!! No matching local album found.")
    except requests.exceptions.Timeout:
        print("!!! Timeout during local album search.")
    except requests.exceptions.RequestException as e:
        print(f"!!! Exception during local album search: {e}")
    return None

def search_album(track_name, artist_id):
    # Check local albums first
    album = search_local_album(track_name, artist_id=artist_id)
    if album:
        return album

    # Fallback to external lookup
    try:
        print(f"--> Searching album externally with track: '{track_name}' and artist ID: {artist_id}")
        resp = requests.get(
            f"{LIDARR_URL}/api/v1/album/lookup",
            headers=headers,
            params={"term": track_name, "artistId": artist_id},
            timeout=10
        )
        resp.raise_for_status()
        results = resp.json()
        print(f"...Found {len(results)} results from album lookup with artist ID.")
        return results[0] if results else None
    except requests.exceptions.Timeout:
        print("!!! Timeout during album search (with artist ID).")
    except requests.exceptions.RequestException as e:
        print(f"!!! Exception during album search (with artist ID): {e}")
    return None

def search_album_by_artist(track_name, artist_name):
    # Check local albums first
    album = search_local_album(track_name, artist_name=artist_name)
    if album:
        return album

    # Fallback to external lookup
    try:
        print(f"--> Fallback external search: '{artist_name} {track_name}'")
        resp = requests.get(
            f"{LIDARR_URL}/api/v1/album/lookup",
            headers=headers,
            params={"term": f"{artist_name} {track_name}"},
            timeout=10
        )
        resp.raise_for_status()
        results = resp.json()
        print(f"...Found {len(results)} results from fallback search.")
        return results[0] if results else None
    except requests.exceptions.Timeout:
        print("!!! Timeout during fallback album search.")
    except requests.exceptions.RequestException as e:
        print(f"!!! Exception during fallback album search: {e}")
    return None


def trigger_album_search(album_id):
    data = {
        "name": "AlbumSearch",
        "albumIds": [album_id]
    }
    resp = requests.post(
        f"{LIDARR_URL}/api/v1/command",
        headers=headers,
        json=data
    )
    resp.raise_for_status()
    print(f"Triggered Lidarr search for album ID {album_id}")

def main():
    if len(sys.argv) < 2:
        print("Usage: musicfetch.py 'Artist - Track' OR 'https://youtube.com/watch?...'")
        sys.exit(1)

    input_str = sys.argv[1]

    if is_url(input_str):
        print("Input is a URL. Extracting metadata to find artist...")
        metadata = run_yt_dlp_get_metadata(input_str)
        if metadata:
            artist_name = get_artist_from_metadata(metadata)
            print(f"Extracted artist name: {artist_name}")
        else:
            print("Failed to get metadata from URL.")
            artist_name = "Unknown Artist"

        target_folder = os.path.join(ROOT_FOLDER, artist_name, "youtube")
        yt_dlp_download(input_str, target_folder)
        return

    if " - " not in input_str:
        print("Input must be in 'Artist - Track' format.")
        sys.exit(1)

    artist_name, track_name = input_str.split(" - ", 1)
    artist_name = artist_name.strip()
    track_name = track_name.strip()

    print(f"Looking up artist: {artist_name}")
    artist = get_existing_artist(artist_name)

    if not artist:
        print("Artist not found. Searching Lidarr metadata (with timeout)...")
        metadata = search_artist(artist_name)
        if not metadata:
            print("No match found or search timed out. Falling back to yt-dlp.")
            yt_dlp_download(f"ytsearch:{input_str}", os.path.join(ROOT_FOLDER, artist_name, "youtube"))
            return

        print(f"Adding artist: {metadata.get('artistName') or metadata.get('title')}")
        artist = add_artist(metadata)

    album = search_album(track_name, artist["id"]) or search_album_by_artist(track_name, artist_name)

    if album:
        print(f"Found album '{album['title']}' for track '{track_name}', triggering search...")
        trigger_album_search(album["id"])
    else:
        print("No album found in Lidarr. Falling back to yt-dlp.")
        yt_dlp_download(input_str, os.path.join(ROOT_FOLDER, artist_name, "youtube"))

if __name__ == "__main__":
    main()