root
381 lines
12 KiB
Python
381 lines
12 KiB
Python
import requests
|
|
import hashlib
|
|
import secrets
|
|
import time
|
|
import sys
|
|
import re
|
|
import os
|
|
import json
|
|
import argparse
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
# ==========================================
|
|
# CONFIGURATION
|
|
# ==========================================
|
|
|
|
# Logging Configuration
|
|
LOG_FILE = "octotrigger.log"
|
|
|
|
# Setup Logger
|
|
logger = logging.getLogger("OctoTrigger")
|
|
logger.setLevel(logging.INFO)
|
|
|
|
# Formatter
|
|
formatter = logging.Formatter('%(asctime)s [%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
|
|
|
|
# File Handler
|
|
file_handler = logging.FileHandler(LOG_FILE)
|
|
file_handler.setFormatter(formatter)
|
|
logger.addHandler(file_handler)
|
|
|
|
# Console Handler
|
|
stream_handler = logging.StreamHandler(sys.stdout)
|
|
stream_handler.setFormatter(formatter)
|
|
logger.addHandler(stream_handler)
|
|
|
|
# URLs
|
|
ND_PROXY_URL = os.getenv("ND_PROXY_URL", "https://ndproxy.wouterverduin.nl")
|
|
ND_DIRECT_URL = os.getenv("ND_DIRECT_URL", "https://navidrome.wouterverduin.nl")
|
|
|
|
# Credentials (Environment variables preferred, fallback to hardcoded)
|
|
ND_USERNAME = os.getenv("ND_USERNAME", "wouser")
|
|
ND_PASSWORD = os.getenv("ND_PASSWORD", "G53^cZI3QVOMge*1x3$b%D2bJ$v*D&!")
|
|
|
|
# Client identification
|
|
CLIENT_NAME = "OctoTriggerScript"
|
|
CLIENT_VERSION = "2.1.0"
|
|
|
|
# Delays
|
|
SCAN_DELAY_SECONDS = 60
|
|
POST_SCAN_WAIT_SECONDS = 30
|
|
|
|
# ==========================================
|
|
# AUTH & HELPERS
|
|
# ==========================================
|
|
|
|
def get_auth_params():
|
|
salt = secrets.token_hex(6)
|
|
token = hashlib.md5((ND_PASSWORD + salt).encode('utf-8')).hexdigest()
|
|
return {
|
|
'u': ND_USERNAME,
|
|
't': token,
|
|
's': salt,
|
|
'v': '1.16.1',
|
|
'c': CLIENT_NAME,
|
|
'f': 'json'
|
|
}
|
|
|
|
def validate_input(json_data):
|
|
"""Validate format of incoming JSON"""
|
|
if 'playlist_name' not in json_data:
|
|
raise ValueError("Missing 'playlist_name' in JSON")
|
|
if 'tracks' not in json_data or not isinstance(json_data['tracks'], list):
|
|
raise ValueError("Missing 'tracks' list in JSON")
|
|
return True
|
|
|
|
def is_local_file(song):
|
|
"""
|
|
Determines if a song object from the API is a local file.
|
|
"""
|
|
song_id = song.get('id', '')
|
|
path = song.get('path', '').lower()
|
|
suffix = song.get('suffix', '').lower()
|
|
|
|
local_exts = ('.mp3', '.flac', '.m4a', '.ogg', '.opus', '.wav', '.alac', '.ape')
|
|
|
|
# CHECK 1: Valid File Extension in Path or Suffix
|
|
if path.endswith(local_exts) or suffix in [e.replace('.', '') for e in local_exts]:
|
|
return True
|
|
|
|
# CHECK 2: Standard Navidrome MD5 ID (32 chars, hex only)
|
|
if len(song_id) == 32 and re.match(r'^[0-9a-fA-F]+$', song_id):
|
|
return True
|
|
|
|
return False
|
|
|
|
def find_best_match(songs, artist, title):
|
|
"""
|
|
Prioritizes LOCAL matches over remote ones.
|
|
"""
|
|
artist_clean = artist.lower().strip()
|
|
title_clean = title.lower().strip()
|
|
|
|
local_matches = []
|
|
remote_matches = []
|
|
|
|
for song in songs:
|
|
res_artist = song.get('artist', '').lower()
|
|
res_title = song.get('title', '').lower()
|
|
|
|
if artist_clean in res_artist and title_clean in res_title:
|
|
if is_local_file(song):
|
|
local_matches.append(song)
|
|
else:
|
|
remote_matches.append(song)
|
|
|
|
if local_matches:
|
|
logger.info(f" [+] Found {len(local_matches)} local candidate(s). Selecting local.")
|
|
return local_matches[0]
|
|
|
|
if remote_matches:
|
|
logger.info(f" [-] No local copy found. Selected remote candidate.")
|
|
return remote_matches[0]
|
|
|
|
return None
|
|
|
|
def search_song(session, artist, title, base_url=ND_PROXY_URL):
|
|
query = f"{artist} {title}"
|
|
params = get_auth_params()
|
|
params['query'] = query
|
|
params['songCount'] = 20
|
|
|
|
target_name = "Proxy" if base_url == ND_PROXY_URL else "Direct"
|
|
logger.info(f"[*] Searching ({target_name}): {artist} - {title}...")
|
|
|
|
try:
|
|
url = f"{base_url}/rest/search3"
|
|
r = session.get(url, params=params, timeout=15)
|
|
r.raise_for_status()
|
|
|
|
data = r.json()
|
|
subsonic_resp = data.get('subsonic-response', {})
|
|
songs = subsonic_resp.get('searchResult3', {}).get('song', [])
|
|
|
|
if not songs:
|
|
logger.warning(f" ! No results found for: {artist} - {title}")
|
|
return None
|
|
|
|
selected_song = find_best_match(songs, artist, title)
|
|
|
|
if selected_song:
|
|
origin = "LOCAL" if is_local_file(selected_song) else "REMOTE"
|
|
logger.info(f" -> Match ({origin}): {selected_song['artist']} - {selected_song['title']} (ID: {selected_song['id']})")
|
|
|
|
return selected_song
|
|
|
|
except Exception as e:
|
|
logger.error(f" ! Error searching: {e}")
|
|
return None
|
|
|
|
def trigger_stream_download(session, song_id):
|
|
params = get_auth_params()
|
|
params['id'] = song_id
|
|
params['maxBitRate'] = 320
|
|
|
|
logger.info(f" -> Triggering proxy download (Streaming)...")
|
|
|
|
try:
|
|
url = f"{ND_PROXY_URL}/rest/stream"
|
|
|
|
with session.get(url, params=params, stream=True, timeout=20) as r:
|
|
r.raise_for_status()
|
|
|
|
total_bytes = 0
|
|
for chunk in r.iter_content(chunk_size=8192):
|
|
if chunk:
|
|
total_bytes += len(chunk)
|
|
if total_bytes > 2 * 1024 * 1024:
|
|
break
|
|
|
|
logger.info(f" [OK] Trigger sent. (Read {total_bytes/1024:.2f} KB)")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f" [X] Stream failed: {e}")
|
|
return False
|
|
|
|
def trigger_navidrome_scan(session):
|
|
"""
|
|
Triggers a library scan in Navidrome after a delay.
|
|
"""
|
|
logger.info("=" * 40)
|
|
logger.info(f"Waiting {SCAN_DELAY_SECONDS} seconds for downloads to settle before scanning...")
|
|
time.sleep(SCAN_DELAY_SECONDS)
|
|
|
|
logger.info(f"[*] Triggering Navidrome Library Scan...")
|
|
params = get_auth_params()
|
|
|
|
try:
|
|
url = f"{ND_PROXY_URL}/rest/startScan"
|
|
r = session.get(url, params=params, timeout=15)
|
|
r.raise_for_status()
|
|
|
|
data = r.json()
|
|
resp = data.get('subsonic-response', {})
|
|
|
|
if resp.get('status') == 'ok':
|
|
scan_status = resp.get('scanStatus', {})
|
|
count = scan_status.get('count', 'Unknown')
|
|
logger.info(f" [OK] Scan started successfully. (Current Count: {count})")
|
|
|
|
# Post-Scan Wait
|
|
logger.info(f"[*] Waiting {POST_SCAN_WAIT_SECONDS} seconds for scan initialization...")
|
|
time.sleep(POST_SCAN_WAIT_SECONDS)
|
|
logger.info(" [OK] Wait complete.")
|
|
|
|
else:
|
|
logger.warning(f" [!] Scan trigger request completed but status was not 'ok'.")
|
|
|
|
except Exception as e:
|
|
logger.error(f" [X] Failed to trigger scan: {e}")
|
|
|
|
def update_navidrome_playlist(session, playlist_name, song_ids):
|
|
"""
|
|
Creates or updates a playlist on the direct Navidrome instance.
|
|
"""
|
|
logger.info("=" * 40)
|
|
logger.info(f"[*] Configuring Playlist: '{playlist_name}'")
|
|
logger.info(f" Tracks to add: {len(song_ids)}")
|
|
|
|
if not song_ids:
|
|
logger.warning(" [!] No tracks to add. Aborting playlist update.")
|
|
return
|
|
|
|
params = get_auth_params()
|
|
|
|
try:
|
|
# 1. Check for existing playlist to avoid duplicates
|
|
url_get = f"{ND_DIRECT_URL}/rest/getPlaylists"
|
|
r = session.get(url_get, params=params, timeout=15)
|
|
r.raise_for_status()
|
|
|
|
data = r.json()
|
|
playlists = data.get('subsonic-response', {}).get('playlists', {}).get('playlist', [])
|
|
|
|
target_id = None
|
|
for pl in playlists:
|
|
if pl.get('name') == playlist_name:
|
|
target_id = pl.get('id')
|
|
logger.info(f" [i] Found existing playlist ID: {target_id} (Updating)")
|
|
break
|
|
|
|
if not target_id:
|
|
logger.info(f" [i] Playlist does not exist. Creating new.")
|
|
|
|
# 2. Create or Update Playlist
|
|
params = get_auth_params()
|
|
params['name'] = playlist_name
|
|
params['songId'] = song_ids # requests handles list -> songId=1&songId=2
|
|
|
|
if target_id:
|
|
params['playlistId'] = target_id
|
|
|
|
url_create = f"{ND_DIRECT_URL}/rest/createPlaylist"
|
|
r = session.get(url_create, params=params, timeout=15)
|
|
r.raise_for_status()
|
|
|
|
logger.info(f" [OK] Playlist '{playlist_name}' successfully saved.")
|
|
|
|
except Exception as e:
|
|
logger.error(f" [X] Failed to update playlist: {e}")
|
|
|
|
# ==========================================
|
|
# MAIN PROCESS EXECUTION
|
|
# ==========================================
|
|
|
|
def run_process(playlist_data):
|
|
"""
|
|
Main Logic Function.
|
|
Can be called by CLI main() or external Import.
|
|
"""
|
|
|
|
# 1. Prepare Data
|
|
base_playlist_name = playlist_data.get("playlist_name", "Unknown Playlist")
|
|
track_list = playlist_data.get("tracks", [])
|
|
|
|
# 2. Append Date (DD-MM-YY)
|
|
current_date = datetime.now().strftime("%d-%m-%y")
|
|
#final_playlist_name = f"{base_playlist_name} ({current_date})"
|
|
final_playlist_name = f"{current_date} {base_playlist_name}" #deze zelf gemaakt ipv bovenste
|
|
|
|
logger.info(f"--- {CLIENT_NAME} v{CLIENT_VERSION} ---")
|
|
logger.info(f"Playlist Target: {final_playlist_name}")
|
|
logger.info(f"Tracks Loaded: {len(track_list)}")
|
|
|
|
session = requests.Session()
|
|
stats = {'cached': 0, 'downloaded': 0, 'failed': 0}
|
|
|
|
# ---------------------------------------------------------
|
|
# PHASE 1: Download Missing Tracks (Using Proxy)
|
|
# ---------------------------------------------------------
|
|
logger.info(">>> PHASE 1: Checking and Downloading Tracks")
|
|
for artist, title in track_list:
|
|
song = search_song(session, artist, title, base_url=ND_PROXY_URL)
|
|
|
|
if song:
|
|
if is_local_file(song):
|
|
logger.info(f" [i] File exists locally.")
|
|
stats['cached'] += 1
|
|
else:
|
|
logger.info(f" [!] NEW SONG DETECTED")
|
|
if trigger_stream_download(session, song['id']):
|
|
stats['downloaded'] += 1
|
|
else:
|
|
stats['failed'] += 1
|
|
else:
|
|
logger.warning(" [X] Not found in Search")
|
|
stats['failed'] += 1
|
|
|
|
logger.info("-" * 40)
|
|
time.sleep(1)
|
|
|
|
logger.info(f"Phase 1 Done: {stats['cached']} Existing, {stats['downloaded']} New, {stats['failed']} Failed.")
|
|
|
|
# ---------------------------------------------------------
|
|
# PHASE 2: Trigger Scan (If needed)
|
|
# ---------------------------------------------------------
|
|
if stats['downloaded'] > 0:
|
|
trigger_navidrome_scan(session)
|
|
else:
|
|
logger.info("No new downloads. Skipping library scan.")
|
|
|
|
# ---------------------------------------------------------
|
|
# PHASE 3: Build Playlist (Using Direct URL)
|
|
# ---------------------------------------------------------
|
|
logger.info(">>> PHASE 3: Resolving Local IDs for Playlist")
|
|
|
|
final_song_ids = []
|
|
|
|
for artist, title in track_list:
|
|
song = search_song(session, artist, title, base_url=ND_DIRECT_URL)
|
|
|
|
if song:
|
|
final_song_ids.append(song['id'])
|
|
else:
|
|
logger.warning(f" [X] Could not resolve ID for: {artist} - {title}")
|
|
|
|
time.sleep(0.5)
|
|
|
|
update_navidrome_playlist(session, final_playlist_name, final_song_ids)
|
|
|
|
logger.info("--- End of Script ---")
|
|
|
|
|
|
def main():
|
|
# Argument Parsing
|
|
parser = argparse.ArgumentParser(description="Octo-Fiesta Smart Trigger")
|
|
parser.add_argument('json_input', help='JSON string containing playlist name and tracks')
|
|
args = parser.parse_args()
|
|
|
|
# Load JSON
|
|
try:
|
|
playlist_data = json.loads(args.json_input)
|
|
validate_input(playlist_data)
|
|
|
|
# Call the logic
|
|
run_process(playlist_data)
|
|
|
|
except json.JSONDecodeError as e:
|
|
logger.critical(f"Invalid JSON provided: {e}")
|
|
sys.exit(1)
|
|
except ValueError as e:
|
|
logger.critical(f"Invalid Data Structure: {e}")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
logger.critical(f"Unexpected Error: {e}")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|