Transfer song ratings from Rhythmbox to OpenSubsonic server

transfer.py

import libopensonic
import logging
import untangle
from urllib.parse import urlparse, unquote
from pathlib import Path

# Paths in rhythmdb
OLD_PATH = Path("/data/music")
# Paths in subsonic API
NEW_PATH = Path("/music")
RHYTHMDB_PATH = OLD_PATH / "rhythmbox" / "rhythmdb.xml"

logging.basicConfig(level=logging.DEBUG)

logging.info("Login")
server = libopensonic.Connection(
    base_url="...",
    username="...",
    password="...",
    port=...,
    server_path="...",
)

# Get all songs
logging.info("Get all songs from subsonic")
search_res = server.search2("", artist_count=0, album_count=0, song_count=999999)
songs = search_res.song
songs_by_path = {song.path: song for song in songs}

# Load rhythmbox DB
logging.info("Parse rhythmdb")
rhythmdb = untangle.parse(str(RHYTHMDB_PATH)).rhythmdb

logging.info("Process songs")
for entry in rhythmdb.entry:
    if entry["type"] != "song":
        continue
    if "rating" not in entry:
        continue

    loc = entry.location.cdata

    rating = float(entry.rating.cdata)
    if not rating:
        continue
    if int(rating) != rating:
        logging.warning(f"Invalid rating {rating} for {loc}")

    rating = int(rating)

    loc_parts = urlparse(loc)

    path = unquote(loc_parts.path).removeprefix(f"{OLD_PATH}/")
    subsonic_path = f"{NEW_PATH}/{path}"

    subsonic_entry = songs_by_path.get(subsonic_path)
    if subsonic_entry is None:
        logging.warning(f"Did not find Subsonic entry for {subsonic_path}")
        continue

    if subsonic_entry.user_rating == rating:
        continue

    logging.debug(f"Set rating for {subsonic_path} [{subsonic_entry.id}] to {rating}")
    server.set_rating(item_id=subsonic_entry.id, rating=rating)