Script per l'importazione delle stazioni radio

This commit is contained in:
tommal
2025-11-30 15:21:34 +01:00
commit fe23acafdf
2 changed files with 137 additions and 0 deletions

3
readme Normal file
View File

@@ -0,0 +1,3 @@
sqlite3 -header -csv ~/snap/shortwave/114/.local/share/Shortwave/Shortwave.db "SELECT * FROM library;" > ~/my_shortwave_stations.csv

134
shortwave_to_navidrome.py Normal file
View File

@@ -0,0 +1,134 @@
#!/usr/bin/env python3
"""
Import Shortwave radio stations into Navidrome via Subsonic API
"""
import sqlite3
import json
import requests
import hashlib
import random
import string
from urllib.parse import urljoin
# ============= CONFIGURATION =============
# Edit these values for your setup
NAVIDROME_URL = "http://192.168.1.105:4533" # Your Navidrome URL
USERNAME = "tommal" # Your Navidrome username
PASSWORD = "tommal" # Your Navidrome password
SHORTWAVE_DB = "/home/tommal/snap/shortwave/114/.local/share/Shortwave/Shortwave.db"
# =========================================
def generate_token(password, salt):
"""Generate authentication token for Subsonic API"""
token_str = password + salt
return hashlib.md5(token_str.encode()).hexdigest()
def create_radio_station(base_url, username, token, salt, name, stream_url, homepage_url=""):
"""Create internet radio station in Navidrome"""
url = urljoin(base_url, "/rest/createInternetRadioStation")
params = {
'u': username,
't': token,
's': salt,
'v': '1.16.1',
'c': 'ShortwaveImport',
'f': 'json',
'name': name,
'streamUrl': stream_url,
'homepageUrl': homepage_url
}
try:
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
result = response.json()
if result.get('subsonic-response', {}).get('status') == 'ok':
return True, "Success"
else:
error = result.get('subsonic-response', {}).get('error', {})
return False, f"Error: {error.get('message', 'Unknown error')}"
except Exception as e:
return False, str(e)
def main():
# Generate salt and token
salt = ''.join(random.choices(string.ascii_letters + string.digits, k=16))
token = generate_token(PASSWORD, salt)
print(f"Connecting to Shortwave database: {SHORTWAVE_DB}")
# Connect to Shortwave database
conn = sqlite3.connect(SHORTWAVE_DB)
cursor = conn.cursor()
# Get all stations from library
cursor.execute("SELECT uuid, data FROM library")
stations = cursor.fetchall()
print(f"\nFound {len(stations)} stations in Shortwave library")
print(f"Importing to Navidrome at {NAVIDROME_URL}...\n")
success_count = 0
failed_count = 0
for uuid, data_json in stations:
try:
# Parse JSON data
data = json.loads(data_json)
name = data.get('name', '').strip()
# Prefer url_resolved over url
stream_url = data.get('url_resolved') or data.get('url', '')
homepage = data.get('homepage', '')
if not name or not stream_url:
print(f"⚠️ Skipping station with missing data (UUID: {uuid})")
failed_count += 1
continue
print(f"Importing: {name}")
print(f" URL: {stream_url}")
success, message = create_radio_station(
NAVIDROME_URL,
USERNAME,
token,
salt,
name,
stream_url,
homepage
)
if success:
print(f" ✓ Success\n")
success_count += 1
else:
print(f" ✗ Failed: {message}\n")
failed_count += 1
except json.JSONDecodeError:
print(f"⚠️ Skipping station with invalid JSON (UUID: {uuid})")
failed_count += 1
except Exception as e:
print(f"⚠️ Error processing station {uuid}: {e}")
failed_count += 1
conn.close()
# Summary
print("=" * 50)
print(f"Import complete!")
print(f" ✓ Successfully imported: {success_count}")
print(f" ✗ Failed: {failed_count}")
print(f" Total: {len(stations)}")
print("=" * 50)
if __name__ == "__main__":
main()