Restore your download history with original filenames and hardlinks from your radarr database

arr-restore.py

###########################
###  *ARR RESTORE v1.2  ###
###      by xzin        ###
###########################

# About #
# This script automatically restores the original downloaded media files from your Radarr and Sonarr databases
# These files will be named as they were when originally downloaded, before Radarr/Sonarr performed any renaming
# These files will also be hardlinked to your existing media, so you won't incur additional storage space usage

# Protip #
# I recommend that you output to a new directory, not to where your download client(s) are currently pointed.
# This will let you test the script and ensure everything worked as intended.
# (You can safely run the script a few times without worrying about storage space or IO)
# If everything worked, you can move the files over to your download directory and start seeding again. hooray!


### CHANGE THE VARIABLES BELOW BASED ON YOUR SYSTEM ###

# Database Paths #
#----------------#
# These should point to the radarr.db & sonarr.db files within your radarr and sonarr config directories
# For docker users, it's in the location that your docker's /config volume points to
PATH_TO_RADARR_DB = "/mnt/user/appdata/radarr/radarr.db"
PATH_TO_SONARR_DB = "/mnt/user/appdata/sonarr/sonarr.db"


# Path To Output Directory #
#--------------------------#
# Where do you want the files saved to
OUTPUT_DIRECTORY = "/mnt/user/data/arr-restore/files"

# Media Directories #
#-------------------#
# Where do your movies and shows live
RADARR_DIRECTORY = "/mnt/user/data/media/movies"
SONARR_DIRECTORY = "/mnt/user/data/media/tv"

# Are you on Windows or Linux/Mac? #
#----------------------------------#
# Valid values are "windows" or "unix".
OPERATING_SYSTEM = "unix"


### STOP CHANGING VARIABLES ###
###############################


import os
from pathlib import Path
import subprocess
import shlex
import sqlite3
from contextlib import closing


RADARR_QUERY = '''SELECT Movies.Path || "''' + os.path.sep + '''" || MovieFiles.RelativePath AS RelativePath, MovieFiles.OriginalFilePath
                    FROM Movies INNER JOIN MovieFiles ON Movies.Id = MovieFiles.MovieId
                    WHERE OriginalFilePath IS NOT NULL'''

SONARR_QUERY = '''SELECT Series.Path || "''' + os.path.sep + '''" || EpisodeFiles.RelativePath AS RelativePath, EpisodeFiles.OriginalFilePath
                    FROM Series INNER JOIN EpisodeFiles ON Series.Id = EpisodeFiles.SeriesId
                    WHERE OriginalFilePath IS NOT NULL'''

def create_hard_links(output_pathlib: Path, arr_directory: Path, radarr_name: Path, original_name):
    '''
    Invokes either ln or mklink to establish hardlink between data in media directory and output directory

    :param output_pathlib: The destination directory where the user wants to new file saved
    :param arr_directory: The sonarr/radarr media directory where the file currently lives
    :param radarr_name: The abbreviated arr path (eg: 'John Wick (2014)/JohnWick.mkv' or 'Fringe/Season 01/Fringe S01E01.mkv')
    :param original_name: The filename (including parent directory, if applicable) as downloaded, before any arr renaming
    '''
   # Build the full file paths
    download_path = output_pathlib / Path(original_name)
    arr_path = arr_directory / radarr_name

    # Normalize and resolve the absolute path to fix any relative path references
    absolute_arr_path = arr_path.resolve()

    # In case the media was in its own folder, we'll re-create the folder first
    if not download_path.parent.exists():
        os.makedirs(download_path.parent)

    # Create the hard link
    try:
        if OPERATING_SYSTEM == "unix":
            resp = subprocess.run(["ln", absolute_arr_path, download_path], capture_output=True, text=True)
        else:
            resp = subprocess.run(["cmd", "/c", f"mklink /h {shlex.quote(str(download_path))} {shlex.quote(str(absolute_arr_path))}"], shell=True, capture_output=True, text=True)

        if resp.returncode == 0:
            print(f"Hardlink between {absolute_arr_path} and {download_path} successfully created")
        else:
            print(f"ERROR: Could not create hardlink between {absolute_arr_path} and {download_path}: {resp.stderr}")
    
    except OSError as e:
        print(f"ERROR: Could not create hardlink between {absolute_arr_path} and {download_path}: {e}")


def main():
    output_pathlib = Path(OUTPUT_DIRECTORY)
    # Ensure that the target directory exists
    if not output_pathlib.exists():
        print("ERROR: The output directory does not exist. Please create it prior to running this script.")
        return

    # Ensure radarr database file is found
    radarr_db_pathlib = Path(PATH_TO_RADARR_DB)
    if not radarr_db_pathlib.exists():
        print(f"ERROR: Cannot find radarr database file at specified location: {radarr_db_pathlib}")
        return
    
    # Ensure radarr database file is found
    sonarr_db_pathlib = Path(PATH_TO_SONARR_DB)
    if not sonarr_db_pathlib.exists():
        print(f"ERROR: Cannot find sonarr database file at specified location: {sonarr_db_pathlib}")
        return 
    
    # Ensure radarr media directory is found
    radarr_media_pathlib = Path(RADARR_DIRECTORY)
    if not radarr_media_pathlib.exists():
        print("ERROR: Cannot find radarr media directory")
        return
    
    # Ensure sonarr media directory is found
    sonarr_media_pathlib = Path(SONARR_DIRECTORY)
    if not sonarr_media_pathlib.exists():
        print("ERROR: Cannot find sonarr media directory")
        return
    
    # Loop through the Radarr library
    with closing(sqlite3.connect(radarr_db_pathlib)) as connection:
        with closing(connection.cursor()) as cursor:
            db = cursor.execute(RADARR_QUERY)
            history = db.fetchall()
            for item in history:
                # Split the relative path so that it just contains /Movie Name/FileName.ext
                relative_path = Path(item[0])
                sanitized_path = Path(relative_path.parts[-2:][0], relative_path.name)

                # Create the hardlink
                create_hard_links(output_pathlib, radarr_media_pathlib, sanitized_path, item[1])

    # Loop through the Sonarr library
    with closing(sqlite3.connect(sonarr_db_pathlib)) as connection:
        with closing(connection.cursor()) as cursor:
            db = cursor.execute(SONARR_QUERY)
            history = db.fetchall()
            for item in history:
                # Split the relative path so that it just contains Series Name/Season/FileName.ext
                relative_path = Path(item[0])
                # Retain 2 subfolders: one for the series, one for the season
                sanitized_path = Path(relative_path.parts[-3:][0], relative_path.parts[-3:][1], relative_path.name)

                # Create the hardlink
                create_hard_links(output_pathlib, sonarr_media_pathlib, sanitized_path, item[1])

if __name__ == "__main__":
    main()