nmv/change_path.py

#!/usr/bin/env python3
# This script changes the path for a folder or file in Navidrome's database, allowing music files to be
# moved or renamed on the file system without losing associated metadata in Navidrome. Since the original
# version, it has been updated to account for the media_file IDs, which are calculated from the path value
# and referenced in several tables.
#
# This script is based on Navidrome version 0.49.2. If you are running an older version of Navidrome, it
# will likely fail. If you are running a newer version of Navidrome, your mileage may vary.
#
# It does NOT make any modifications to the file system - only to the Navidrome database. 
#
# It does not rescan the file; it assumes nothing has changed but the path. If you're moving files
# and also updating their contents (e.g. tags or bitrate), run this to change the path(s) in the
# database, and then run a full scan to update the metadata.
#
# Place this file in the same directory as navidrome.db, which is /var/lib/navidrome on Linux, and be sure  
# to use fully qualified paths for arguments. It must be run as a user that has write access to the
# navidrome.db file.
#
# Generic use - note that you may need to use python3 instead of python, depending on your system:
#   python change_path.py FROM_PATH TO_PATH
#
# Example: Rename/move a folder (note trailing slashes):
#   python change_path.py /mnt/music/artists/Bjork/ /mnt/music/artists/Björk/
# 
# Example: Rename a song file (use quotes for paths with spaces):
#   python change_path.py "/mnt/music/artists/Test 1/song.mp3" "/mnt/music/artists/Test 2/01 - Song.mp3"
#
# The script's output lists each path updated along with the MD5 ID calculated from it and the row
# count updates for each table referencing the old ID.
#
# Note that Navidrome's scanner will automatically remove files from its database if they're found to 
# be missing, so it's important to stop the Navidrome server process before moving files, and to update
# the database with this script prior to restarting it.
#
# Steps to use:
# 1. Stop the Navidrome server.
# 2. Make a backup of your navidrome.db file, so you can roll back any unwanted changes.
# 3. Move or rename folders and files as needed on the file system.
# 4. Run this script to update the Navidrome database for any files/folders moved or renamed.
# 5. Start Navidrome service.
# 6. Optionally run a full scan if file contents have changed.
#
# Source: https://gist.github.com/bagaag/3d64e3349b6ed3bfd6e01813222db055
#

import hashlib
import os
import sqlite3
import sys

from shared import Song, get_media_by_path


def exec_update(sql: str, params: tuple, conn: sqlite3.Connection):
    cur = conn.cursor()
    cur.execute(sql, params)
    # conn.commit()
    rc = cur.rowcount
    cur.close()
    return rc


def replace_values(res: tuple[Song, ...], to_path: str, from_path: str, conn: sqlite3.Connection):
    replaced = []
    for row in res:
        old_id = row.id
        old_path = str(row.path)

        if from_path.endswith('/'):
            new_path = old_path.replace(from_path, to_path)
        else:
            new_path = to_path

        new_id = md5(new_path)
        sql = f"""
            UPDATE media_file 
            SET path = ?, id = ?
            WHERE path = ? and id = ?;
            """
        media_file = exec_update(sql, (new_path, new_id, old_path, old_id), conn)
        sql = f"""
            UPDATE annotation
            SET item_id = ? 
            WHERE item_id = ? 
            AND item_type='media_file';
            """
        annotation = exec_update(sql, (new_id, old_id), conn)
        sql = f"""
            UPDATE media_file_genres
            SET media_file_id = ?
            WHERE media_file_id = ?;
            """
        media_file_genres = exec_update(sql, (new_id, old_id), conn)
        sql = f"""
            UPDATE playlist_tracks
            SET media_file_id = ?
            WHERE media_file_id = ?;
            """
        playlist_tracks = exec_update(sql, (new_id, old_id), conn)
        sql = f"""
            UPDATE bookmark
            SET item_id = ?
            WHERE item_id = ?
            AND item_type = 'media_file';
            """
        bookmark = exec_update(sql, (new_id, old_id), conn)
        sql = f"""        
            UPDATE album
            SET embed_art_path = ?
            WHERE embed_art_path = ?;
            """
        album = exec_update(sql, (new_path, old_path), conn)
        replaced.append({
            'old_id': old_id,
            'old_path': old_path,
            'new_id': new_id,
            'new_path': new_path,
            'media_file': media_file,
            'annotation': annotation,
            'media_file_genres': media_file_genres,
            'playlist_tracks': playlist_tracks,
            'bookmark': bookmark,
            'album': album
        })
    return replaced


def md5(s: str):
    return hashlib.md5(s.encode('utf-8')).hexdigest()


def main(conn: sqlite3.Connection, from_path: str, to_path: str):
    res = get_media_by_path(conn, from_path)
    print(f'Found {len(res)} path matches.')
    updated = replace_values(res, to_path, from_path, conn)

    for update in updated:
        print('FROM: ' + update['old_path'])
        print('      ' + update['old_id'])
        print('TO:   ' + update['new_path'])
        print('      ' + update['new_id'])
        print('      media_file:        ' + str(update['media_file']))
        print('      annotation:        ' + str(update['annotation']))
        print('      media_file_genres: ' + str(update['media_file_genres']))
        print('      playlist_tracks:   ' + str(update['playlist_tracks']))
        print('      bookmark:          ' + str(update['bookmark']))
        print('      album:             ' + str(update['album']))


def cli_entrypoint():
    if len(sys.argv) < 3:
        print("Usage: python change_path.py FROM_PATH TO_PATH")
        exit()

    from_path = sys.argv[1]
    to_path = sys.argv[2]

    if (from_path.endswith(os.sep) and not to_path.endswith(os.sep)) or (
            not from_path.endswith(os.sep) and to_path.endswith(os.sep)):
        print("One path has a trailing slash and the other doesn't. That's probably not right. Check your inputs.")
        exit()

    conn = sqlite3.connect('/data/navidrome.db')
    main(conn, from_path, to_path)
    conn.commit()
    conn.close()


if __name__ == '__main__':
    cli_entrypoint()