Update file system paths in Navidrome's database - bagaag / change_path.py

change_path.py

# 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 updatd 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 Navidrom, 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 sqlite3
import hashlib
import os
import sys

def path_clause(path):
    if path.endswith('/'):
        return ('LIKE', '%')
    else:
        return ('=', '')

def get_matching_library(path):
    """Gets library path from the database."""
    path = path.rstrip(os.sep)
    sql = """
        SELECT path FROM library;"""
    cur = con.cursor()
    res = cur.execute(sql)
    row = res.fetchone()
    cur.close()
    return row[0]

def get_matching_album(path):
    """Gets album image_files and paths from the database."""
    clause_and_suffix = path_clause(path)
    clause = clause_and_suffix[0]
    if path.endswith('/'):
        path = path + clause_and_suffix[1]
    else:
        path = os.path.dirname(path) + clause_and_suffix[1]
    # paths doesn't have a empty values unlike the rest of the columns might.
    sql = f"""
        SELECT embed_art_path, image_files, paths FROM album
        WHERE paths {clause} ?;"""
    cur = con.cursor()
    res = cur.execute(sql, [path])
    ret = res.fetchall()
    cur.close()
    return ret

def get_matching_media(path):
    clause_and_suffix = path_clause(path)
    clause = clause_and_suffix[0]
    path = path + clause_and_suffix[1]
    sql = f"""
        SELECT id, path FROM media_file
        WHERE path {clause} ?
        ORDER BY path;"""
    cur = con.cursor()
    res = cur.execute(sql, [path])
    ret = res.fetchall()
    cur.close()
    return ret

def exec_update(sql, params):
    cur = con.cursor()
    cur.execute(sql, params)
    con.commit()
    rc = cur.rowcount
    cur.close()
    return rc

def replace_library_values():
    """Replaces library path in the database."""
    replaced = []
    old_path = from_path.rstrip(os.sep)
    new_path = to_path.rstrip(os.sep)
    sql = """
        UPDATE library
        SET path = ?
        WHERE path = ?;
        """
    exec_update(sql, (new_path, old_path))
    replaced.append({
        'old_path': old_path,
        'new_path': new_path
    })
    return replaced

def replace_album_values(res):
    """Replaces album table's paths and image_files columns."""
    replaced = []
    old_path = from_path
    new_path = to_path
    if from_path.endswith('/'):
        new_path = old_path.replace(from_path, to_path)
    else:
        new_path = to_path
    for row in res:
        res_embed_art = row[0]
        res_image_files = row[1]
        #res_paths = row[2] # Unused, but kept for reference.

        album_embed_art = 0
        if len(res_embed_art) > 0:
            if from_path.endswith('/'):
                new_embed_art_path = res_embed_art.replace(from_path, to_path)
            else:
                new_embed_art_path = new_path
            sql = """
                UPDATE album
                SET embed_art_path = ?
                WHERE embed_art_path = ?;
                """
            album_embed_art = exec_update(sql, (new_embed_art_path, res_embed_art))

        album_images = 0
        if len(res_image_files) > 0:
            if from_path.endswith('/'):
                new_image_path = res_image_files.replace(from_path, to_path)
            else:
                new_image_path = new_path
            sql = """
                UPDATE album
                SET image_files = ?
                WHERE image_files = ?;
                """
            album_images = exec_update(sql, (new_image_path, res_image_files))

        # paths might include multiple paths joined to a single string.
        sql = """
        UPDATE album
        SET paths = REPLACE(paths, ?, ?)
        WHERE paths LIKE ?;
        """
        album_paths = exec_update(sql, (old_path, new_path, '%' + old_path + '%'))
        replaced.append({
            'old_path': old_path,
            'new_path': new_path,
            'album_embed_art': album_embed_art,
            'album_image_files': album_images,
            'album_paths': album_paths
        })
    return replaced

def replace_values(res):
    replaced = []
    for row in res:
        old_id = row[0]
        old_path = row[1]
        new_path = to_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 = """
            UPDATE media_file
            SET path = ?, id = ?
            WHERE path = ? and id = ?;
            """
        media_file = exec_update(sql, (new_path, new_id, old_path, old_id))
        sql = """
            UPDATE annotation
            SET item_id = ?
            WHERE item_id = ?
            AND item_type='media_file';
            """
        annotation = exec_update(sql, (new_id, old_id))
        sql = """
            UPDATE media_file_genres
            SET media_file_id = ?
            WHERE media_file_id = ?;
            """
        media_file_genres = exec_update(sql, (new_id, old_id))
        sql = """
            UPDATE playlist_tracks
            SET media_file_id = ?
            WHERE media_file_id = ?;
            """
        playlist_tracks = exec_update(sql, (new_id, old_id))
        sql = """
            UPDATE bookmark
            SET item_id = ?
            WHERE item_id = ?
            AND item_type = 'media_file';
            """
        bookmark = exec_update(sql, (new_id, old_id))
        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
        })
    return replaced

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

#
# main
#
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()


con = sqlite3.connect('navidrome.db')

# Check the library table and update accordingly.
res_library = get_matching_library(from_path)
updated_library = []
if not from_path.rstrip(os.sep).startswith(res_library):
    print("FROM_PATH doesn't exist in the database. Is there a typo in the path?")
    print(f'\n{from_path}')
    exit()
elif not (to_path.rstrip(os.sep).startswith(res_library)):
    # If to_path changes the wholy library path, update all media references.
    #
    # NOTE: Could be better to ask a confirmation from user, so that the whole music
    # library isn't accidentally moved to an album's location for eg.
    from_path = res_library + os.sep # Replace from_path with the value from library path. Might not be the best idea to change global variable here.
    updated_library = replace_library_values() 
    # TODO: Parse user provided to_path parameter? Too tired atm. :)

# Update the media_file table.
res = get_matching_media(from_path)
print(f'Found {len(res)} path matches.')
updated = replace_values(res)

# Update the album table.
res_album = get_matching_album(from_path)
print(f'Found {len(res_album)} album matches.')
updated_album = replace_album_values(res_album)

con.close()

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']))

# Creates log spam: there are only few changes per album.
#for update in updated_album:
#    print("\nAlbum:\n")
#    print('FROM: ' + update['old_path'])
#    print('TO:   ' + update['new_path'])
#    print('      album_embed_art:   ' + str(update['album_embed_art']))
#    print('      album_paths:       ' + str(update['album_paths']))
#    print('      album_image_files: ' + str(update['album_image_files']))

if len(updated_library) > 0:
    for update in updated_library:
        print("\nLibrary:\n")
        print("NOTE: Library location has changed!")
        print('FROM: ' + update['old_path'])
        print('TO:   ' + update['new_path'])

Needs fixing when renaming a single song.

Song renaming fixed.

This needs to be updated for Navidrome v0.55.0

Obsolete.