root
117 lines
3.4 KiB
Python
Executable File
117 lines
3.4 KiB
Python
Executable File
import requests
|
|
import json
|
|
|
|
# ==========================================
|
|
# CONFIGURATION
|
|
# ==========================================
|
|
ND_PASSWORD = "G53^cZI3QVOMge*1x3$b%D2bJ$v*D&!"
|
|
ND_DIRECT_URL = "https://navidrome.wouterverduin.nl"
|
|
ND_USERNAME = "wouser"
|
|
|
|
# The webhook where data will be sent
|
|
#WEBHOOK_URL = "https://n8n.wouterverduin.nl/webhook-test/c3129d3a-dca8-43ae-8003-8d94b7f9097d"
|
|
WEBHOOK_URL = "https://n8n.wouterverduin.nl/webhook/c3129d3a-dca8-43ae-8003-8d94b7f9097d"
|
|
|
|
# ==========================================
|
|
# INTERNAL API AUTHENTICATION
|
|
# ==========================================
|
|
def get_auth_token():
|
|
auth_url = f"{ND_DIRECT_URL}/auth/login"
|
|
credentials = {
|
|
"username": ND_USERNAME,
|
|
"password": ND_PASSWORD
|
|
}
|
|
|
|
try:
|
|
response = requests.post(auth_url, json=credentials)
|
|
response.raise_for_status()
|
|
return response.json().get('token')
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"❌ Login Failed: {e}")
|
|
return None
|
|
|
|
# ==========================================
|
|
# DATA RETRIEVAL
|
|
# ==========================================
|
|
def get_song_stats(token):
|
|
headers = {
|
|
'x-nd-authorization': f'Bearer {token}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
api_url = f"{ND_DIRECT_URL}/api/song"
|
|
|
|
payload = {
|
|
"recently_played": [],
|
|
"most_played": []
|
|
}
|
|
|
|
# 1. Get Recently Played Songs
|
|
recent_params = {
|
|
'_end': 10,
|
|
'_order': 'DESC',
|
|
'_sort': 'play_date',
|
|
'_start': 0
|
|
}
|
|
|
|
try:
|
|
res = requests.get(api_url, headers=headers, params=recent_params)
|
|
res.raise_for_status()
|
|
songs = res.json()
|
|
for song in songs:
|
|
payload["recently_played"].append({
|
|
"artist": song.get('artist', 'Unknown'),
|
|
"title": song.get('title', 'Unknown')
|
|
})
|
|
except Exception as e:
|
|
print(f"Error fetching recent songs: {e}")
|
|
|
|
# 2. Get Most Played Songs (All Time)
|
|
top_params = {
|
|
'_end': 10,
|
|
'_order': 'DESC',
|
|
'_sort': 'play_count',
|
|
'_start': 0
|
|
}
|
|
|
|
try:
|
|
res = requests.get(api_url, headers=headers, params=top_params)
|
|
res.raise_for_status()
|
|
songs = res.json()
|
|
for song in songs:
|
|
payload["most_played"].append({
|
|
"artist": song.get('artist', 'Unknown'),
|
|
"title": song.get('title', 'Unknown')
|
|
})
|
|
except Exception as e:
|
|
print(f"Error fetching top songs: {e}")
|
|
|
|
return payload
|
|
|
|
# ==========================================
|
|
# WEBHOOK DELIVERY
|
|
# ==========================================
|
|
def send_to_webhook(data):
|
|
try:
|
|
print(f"Sending data to webhook...")
|
|
response = requests.post(WEBHOOK_URL, json=data)
|
|
response.raise_for_status()
|
|
print(f"✅ Successfully sent to webhook (Status: {response.status_code})")
|
|
except Exception as e:
|
|
print(f"❌ Failed to send to webhook: {e}")
|
|
|
|
# ==========================================
|
|
# MAIN
|
|
# ==========================================
|
|
if __name__ == "__main__":
|
|
if ND_PASSWORD == "YOUR_NEW_PASSWORD_HERE":
|
|
print("⚠️ Please update the ND_PASSWORD variable in the script first.")
|
|
else:
|
|
token = get_auth_token()
|
|
|
|
if token:
|
|
# Gather the data
|
|
stats_data = get_song_stats(token)
|
|
|
|
# Send the data to n8n
|
|
send_to_webhook(stats_data)
|