Migrate library data (playcounts, favorites, import ages etc.) from Rhythmbox to Navidrome

transfer_rhytmbox_to_navidrome.py

from datetime import datetime, UTC
import logging
from typing import NamedTuple
from pathlib import Path
import sqlite3
from urllib.parse import unquote, urlparse

import untangle

# Path of the music files in rhythmdb (used for conversion of absolute to library-relative)
OLD_PATH = Path("/data/Musik")
# In-library subfolder of the music files in navidrome library path
NEW_PATH = ""
NAVIDROME_DB_PATH = Path("/data/Musik/navidrome.db")
RHYTHMBOX_PATH = Path("/data/Musik/rhythmbox")
# ID of the target user in Navidrome
NAVIDROME_USER_ID = "..."
# All albums contained in this playlist will be favorited
FAVORITE_ALBUMS_IN_PLAYLIST = "My favorite albums"
# All songs contained in this playlist will be favorited
FAVORITE_SONGS_IN_PLAYLIST = "My favorite tracks"

logging.basicConfig(level=logging.INFO)


def adapt_datetime_iso(val: datetime) -> str:
    """Adapt datetime to offset-aware ISO 8601 date."""
    # Use same format as navidrome
    return val.isoformat(sep=" ", timespec="milliseconds")


def convert_datetime(val: bytes) -> datetime:
    """Convert ISO 8601 datetime to datetime object."""
    return datetime.fromisoformat(val.decode())


sqlite3.register_adapter(datetime, adapt_datetime_iso)
sqlite3.register_converter("datetime", convert_datetime)

navidrome_db = sqlite3.connect(NAVIDROME_DB_PATH, detect_types=sqlite3.PARSE_DECLTYPES)
cur = navidrome_db.cursor()


class NavidromeSong(NamedTuple):
    id: str
    album_id: str
    artist_id: str
    created_at: datetime


def _navidrome_get_all_songs() -> dict[str, NavidromeSong]:
    # Get all songs
    logging.info("Get all songs from navidrome")
    songs = cur.execute(
        "SELECT path, id, album_id, artist_id, created_at FROM media_file"
    )

    return {song_data[0]: NavidromeSong(*song_data[1:]) for song_data in songs}


songs_by_path = _navidrome_get_all_songs()

# Load rhythmbox DB
logging.info("Parse rhythmdb")
rhythmdb = untangle.parse(str(RHYTHMBOX_PATH / "rhythmdb.xml")).rhythmdb
rhythmdb_playlists = untangle.parse(
    str(RHYTHMBOX_PATH / "playlists.xml")
).rhythmdb_playlists


def _find_navidrome_song(loc: str) -> NavidromeSong | None:
    loc_parts = urlparse(loc)

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

    return songs_by_path.get(navidrome_path)


logging.info("Process songs")
touched_songs = set[str]()

for entry in rhythmdb.entry:
    if entry["type"] != "song":
        continue

    loc = entry.location.cdata

    rating = float(entry.rating.cdata) if "rating" in entry else 0
    if int(rating) != rating:
        logging.warning(f"Invalid non-integer rating {rating} for {loc}")
    rating = int(rating)

    first_seen = datetime.fromtimestamp(int(entry.first_seen.cdata), tz=UTC)
    play_count = int(entry.play_count.cdata) if "play_count" in entry else 0
    last_played = (
        datetime.fromtimestamp(int(entry.last_played.cdata), tz=UTC)
        if "last_played" in entry
        else None
    )

    navidrome_song = _find_navidrome_song(loc)
    if navidrome_song is None:
        logging.warning(f"Did not find Navidrome entry for {loc}")
        continue
    navidrome_id = navidrome_song.id
    touched_songs.add(navidrome_id)

    logging.debug(f"{navidrome_id} set created_at={first_seen}")
    upd_result = cur.execute(
        "UPDATE media_file SET created_at=? WHERE id=?", (first_seen, navidrome_id)
    )
    assert upd_result.rowcount

    # Find existing annotation, if possible
    where_clause = "user_id=? AND item_type=? AND item_id=?"
    ident_values = (NAVIDROME_USER_ID, "media_file", navidrome_id)

    annotation = cur.execute(
        f"SELECT play_count, play_date, rating FROM annotation WHERE {where_clause}",
        ident_values,
    ).fetchone()
    if annotation is None:
        logging.debug(f"insert empty for {ident_values}")
        _ = cur.execute(
            "INSERT INTO annotation(user_id, item_type, item_id) VALUES(?, ?, ?)",
            ident_values,
        )
        navidrome_play_count, navidrome_play_date, navidrome_rating = 0, None, 0
    else:
        navidrome_play_count, navidrome_play_date, navidrome_rating = annotation

    update_clause = "play_count=?, play_date=?, rating=?"
    update_values = (
        max(play_count, navidrome_play_count),
        max(list(filter(None, (last_played, navidrome_play_date))), default=None),
        max(rating, navidrome_rating),
    )

    logging.debug(f"[{ident_values}] set {update_values}")

    _ = cur.execute(
        f"UPDATE annotation SET {update_clause} WHERE {where_clause}",
        (*update_values, *ident_values),
    )


logging.info("Process favorite albums")
fav_song_playlist = next(
    p for p in rhythmdb_playlists.playlist if p["name"] == FAVORITE_ALBUMS_IN_PLAYLIST
)
fav_time = datetime.now(UTC)
favd_navidrome_albums = set[str]()
for loc_node in fav_song_playlist.location:
    loc = loc_node.cdata
    navidrome_song = _find_navidrome_song(loc)
    if navidrome_song is None:
        logging.warning(f"Did not find Navidrome entry for {loc}")
        continue
    navidrome_album_id = navidrome_song.album_id

    if navidrome_album_id in favd_navidrome_albums:
        continue

    logging.debug(f"Favoriting {navidrome_album_id} due to {loc}")
    _ = cur.execute(
        "INSERT INTO annotation (user_id, item_id, item_type, starred, starred_at) VALUES(?, ?, 'album', TRUE, ?) ON CONFLICT DO UPDATE SET starred = excluded.starred, starred_at = excluded.starred_at",
        (NAVIDROME_USER_ID, navidrome_album_id, fav_time),
    )

    favd_navidrome_albums.add(navidrome_album_id)


logging.info("Process favorite songs")
fav_song_playlist = next(
    p for p in rhythmdb_playlists.playlist if p["name"] == FAVORITE_SONGS_IN_PLAYLIST
)
for loc_node in fav_song_playlist.location:
    loc = loc_node.cdata
    navidrome_song = _find_navidrome_song(loc)
    if navidrome_song is None:
        logging.warning(f"Did not find Navidrome entry for {loc}")
        continue

    logging.debug(f"Favoriting {navidrome_song.id}")
    _ = cur.execute(
        "INSERT INTO annotation (user_id, item_id, item_type, starred, starred_at) VALUES(?, ?, 'media_file', TRUE, ?) ON CONFLICT DO UPDATE SET starred = excluded.starred, starred_at = excluded.starred_at",
        (NAVIDROME_USER_ID, navidrome_song.id, fav_time),
    )

# Reload data
songs_by_path = _navidrome_get_all_songs()

logging.info("Determine ages")
earliest_date_by_album = dict[str, datetime]()
earliest_date_by_artist = dict[str, datetime]()


def _update_age_dict(
    target: dict[str, datetime],
    key: str,
    song_created_at: datetime,
) -> None:
    if key not in target or target[key] > song_created_at:
        target[key] = song_created_at


for song in songs_by_path.values():
    if song.id not in touched_songs:
        # Only touch songs that were also in Rhythmbox
        continue

    _update_age_dict(earliest_date_by_album, song.album_id, song.created_at)
    _update_age_dict(earliest_date_by_artist, song.artist_id, song.created_at)


def _fixup_age(table: str, values: dict[str, datetime]) -> None:
    for key, value in values.items():
        logging.debug(f"Set {table}/{key} created at {value}")
        res = cur.execute(
            f"UPDATE {table} SET created_at = ? WHERE id = ?", (value, key)
        )
        # assert res.rowcount


logging.info("Fixup album age")
_fixup_age("album", earliest_date_by_album)
logging.info("Fixup artist age")
_fixup_age("artist", earliest_date_by_artist)

logging.info("Committing DB changes")
navidrome_db.commit()
navidrome_db.close()

logging.info("Done!")