djairoh
/
nd-backerupper
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: djairoh:57fa326da6b3db2e430ac70ae2c649907142aa4e
Branches Tags
djairoh:main
..
compare: djairoh:1af229d59695fea66d25790336982aa85a97610b
Branches Tags
djairoh:main

No commits in common. "57fa326da6b3db2e430ac70ae2c649907142aa4e" and "1af229d59695fea66d25790336982aa85a97610b" have entirely different histories.
57fa326da6 ... 1af229d596

2 changed files with 25 additions and 100 deletions
Show Stats Expand all files Collapse all files

46
README.md
View File

@ -1,46 +0,0 @@
<p align="center">
A CLI tool to manage Navidrome annotations persistently.
</p>
Helps backing up (and restoring) annotation data from Navidrome databases. Aimed at mp3 files.
## Table of Contents
* [Introduction](#introduction)
* [Usage](#usage)
<!-- * [License](#license) -->
## Introduction
This script facilitates backups of the Navidrome annotation data (playcount, rating, starred) for mp3 files. This persistence only depends on the actual mp3 files in question (so no relative filepaths determining the entire database state). It's great for when you want to have that annotation data actually stored somewhere, as opposed to having it be entirely dependent on the whims of the Navidrome db.
The Export functionality works by writing to 3 custom 'TXXX' ID3 tags - which are stored as metadata on the music file itself. Similarly, Importing reads these data from the music files, and adds the appropriate entries in the Navidrome database.
## Usage
There's a few optional arguments, and one mandatory one:
* -u | --user : the username(or real name) of the Navidrome user for which annotations should be managed.
* -d | --db : path to the Navidrome database file.
* -p | --path (import only): path to the root of your music folder.
* -f | --replace-navidrome: substitution to make between Navidrome paths and actual paths; Navidrome component. See [substitution](###substitution).
* -t | --replace-real: substitution to make between Navidrome paths and actual paths; real component. See [substitution](###substitution).
* import|export : whether to import annotations (from file -> db) or export them (from db -> file). Note that importing is destructive, that is, it will overwrite existing annotation data.
### Substitution
As most Navidrome installations live within a Docker environment, generally with a mounted music folder, the filepaths Navidrome sees may not line up with the actual folder structure.
For these cases, the flag `-r | --replace` can be used to 'translate' between Docker environment and real folder structure.
An example: suppose your music folder lives in `/data/music` and your Navidrome configuration has this folder mounted under `/media/drive/music`, like so:
```yaml
...
volumes:
- /data/music:/media/drive/music:ro
```
Without the replace flag, backerupper would not be able to find your music files - as it would read the Navidrome database and try to look under `/media/drive/music` - which may or may not exist (but in any case does not contain your music files). By instead using ` backeruppry.py -f '/media/drive/' -t '/data/`, backerupper will translate these paths appropriately.
Keep in mind that this is a very limited feature, using essentially direct substitution. Results may vary - be sure make **Backups** of your Navidrome database.

79
backerupper.py
View File

@ -1,6 +1,5 @@
#!/usr/bin/python3
import re
import sys
from time import sleep
import argparse
@ -31,66 +30,52 @@ def get_user_id(db: str, name: str) -> str:
print(f"Failed to read id from {db}: {e}")
return ''
class ExportService():
@staticmethod
def exp(db: str, user_id: str, re_nav: str, re_re: str):
"""Exports media_file annotations from a given navidrome database <db> to id3v2 tags. Only exports annotations belonging to the navidrome user with id <user_id>.
input:
db: path to a navidrome sql3 database. This database must contain at least the annotation and media_file tables.
user_id: the user_id to fetch annotations from.
re_nav: path substitution (if necessary); Navidrome folder structure component.
re_re: path substitution (if necessary); real folder structure component.
"""
a = ExportService.get_annotations(db, user_id, re_nav, re_re)
def exp(db: str, user_id: str):
"""Exports media_file annotations from a given navidrome database <db> to id3v2 tags. Only exports annotations belonging to the navidrome user with id <user_id>"""
a = ExportService.get_annotations(db, user_id)
ExportService.write_annotations(a)
@staticmethod
def get_annotations(db: str, user_id: str, re_nav: str = "", re_re:str = "") -> dict:
def get_annotations(db: str, user_id: str) -> list:
"""Fetches all annotations belonging to the user with id <user_id> from the provided sql3 navidrome database <db>.
input:
db: path to a navidrome sql3 database. This database must contain at least the annotation and media_file tables.
user_id: the user_id to fetch annotations from.
re_nav: path substitution (if necessary); Navidrome folder structure component.
re_re: path substitution (if necessary); real folder structure component.
output:
dict of annotations in the format {path: (playCount, rating, starred)}.
list of annotations in the format (path, playCount, rating, starred).
"""
annos = {}
try:
conn = sqlite3.connect(db)
cursor = conn.cursor()
cursor.execute(f"""SELECT path,play_count,rating,starred FROM annotation INNER JOIN media_file ON item_id=id WHERE user_id='{user_id}' AND item_type='media_file'""")
for p,pc,r,s in cursor.fetchall():
annos[re.sub(re_nav, re_re, p)] = (pc,r,s)
annos = cursor.fetchall()
conn.close()
return annos
except sqlite3.OperationalError as e:
print(f"Failed to read data from {db}: {e}")
return annos
return []
@staticmethod
def write_annotations(annos: dict, re_nav: str = "", re_re: str = ""):
def write_annotations(annos: list):
"""Writes a list of annotations to id3v2 tags.
The tags used are TXXX=FNVD_<NAME>, where <NAME> is one of PlayCount, Rating, Starred.
input:
annos: dict of annotations. Format must be {path: (playCount, rating, starred)}.
re_nav: path substitution (if necessary); Navidrome folder structure component.
re_re: path substitution (if necessary); real folder structure component.
annos: list of annotations. Format must be (path, playCount, rating, starred).
"""
count = 0
for k,v in annos.items():
for a in annos:
try:
file = id3.ID3(re.sub(re_nav, re_re, k))
file.add(TXXX(desc="FNVD_PlayCount", text=str(v[0])))
file.add(TXXX(desc="FNVD_Rating", text=str(v[1])))
file.add(TXXX(desc="FNVD_Starred", text=str(v[2])))
file = id3.ID3(a[0])
file.add(TXXX(desc="FNVD_PlayCount", text=str(a[1])))
file.add(TXXX(desc="FNVD_Rating", text=str(a[2])))
file.add(TXXX(desc="FNVD_Starred", text=str(a[3])))
file.save()
count += 1
except MutagenError as e:
@ -101,28 +86,18 @@ class ExportService():
class ImportService():
@staticmethod
def imp(path: str, db: str, user_id: str, re_nav: str, re_re: str):
"""Imports media_file annotations from files located in the <path> directory into to he given navidrome database <db>, as belonging to user with id <user_id>.
input:
path: string containing the path for the root directory of the music folder.
db: path to a navidrome sql3 database. This database must contain at least the annotation and media_file tables.
user_id: the user_id to ascribe annotations to.
re_nav: path substitution (if necessary); Navidrome folder structure component.
re_re: path substitution (if necessary); real folder structure component.
"""
a = ImportService.get_annotations(path, re_nav, re_re)
def imp(path: str, db: str, user_id: str):
"""Imports media_file annotations from files located in the <path> directory into to he given navidrome database <db>, as belonging to user with id <user_id>."""
a = ImportService.get_annotations(path)
ImportService.write_annotations(a, db, user_id)
@staticmethod
def get_annotations(path: str, re_nav: str = "", re_re: str = "") -> dict:
def get_annotations(path: str) -> dict:
"""Finds all mp3 files in path <path> and extracts the FNVD_PlayCount, FNVD_Rating, and FNVD_Starred id3v2 fields into a dict
input:
path: string containing the path for the root directory of the music folder.
re_nav: path substitution (if necessary); Navidrome folder structure component.
re_re: path substitution (if necessary); real folder structure component.
output:
a dict containing annotations in the format {path: (playCount, rating, starred)}.
"""
@ -148,28 +123,26 @@ class ImportService():
starred = int(f.text[0])
if play_count or rating or starred:
annos[re.sub(re_re, re_nav, dir)+'/'+mp3_file] = (play_count, rating, starred)
annos[dir+'/'+mp3_file] = (play_count, rating, starred)
except MutagenError as e:
print(f"failed for file {dir+'/'+mp3_file}: {e}")
return annos
@staticmethod
def write_annotations(annos: dict, db: str, user_id: str, re_nav: str = "", re_re: str = ""):
def write_annotations(annos: dict, db: str, user_id: str):
"""Writes a dict of annotations to the navidrome database <db> as user with id <user_id>.
input:
annos: dict of annotations. Format must be {path: (playCount, rating, starred)}.
db: path to a navidrome sql3 database. This database must contain at least the annotation and media_file tables.
user_id: the user_id to ascribe annotations to.
re_nav: path substitution (if necessary); Navidrome folder structure component.
re_re: path substitution (if necessary); real folder structure component.
"""
try:
conn = sqlite3.connect(db)
cursor = conn.cursor()
# step 1: fetch path and item_id for existing annotations WHERE path IN annos.paths
cursor.execute(f"""SELECT path,id FROM media_file WHERE path IN ('{"','".join([ re.sub(re_re, re_nav, k) for k in annos.keys()])}')""")
cursor.execute(f"""SELECT path,id FROM media_file WHERE path IN ('{"','".join(annos.keys())}')""")
# step 2: create list with item_ids
items = []
@ -194,8 +167,6 @@ def main():
parser.add_argument("-d", "--db", type=str, nargs=1, metavar="database", default='./navidrome.db', help="Filepath to the navidrome database you want to backup (or restore)")
parser.add_argument("ie", type=str, nargs=1, metavar="import|export", default=None, choices=['import', 'export'], help="What you want to do with the annotations. choices: [import, export]")
parser.add_argument("-p", "--path", type=str, nargs=1, required='import' in sys.argv, metavar="mediapath", default=None, help="The folder to import metadata from (import only)")
parser.add_argument("-f", "--replace-navidrome", type=str, nargs=1, metavar="from", default="", help="Path substitution to make when translating between Navidrome db and real folder structure; Navidrome component")
parser.add_argument("-t", "--replace-real", type=str, nargs=1, metavar="to", default="", help="Path substitution to make when translating between Navidrome db and real folder structure; real folder component.")
args = parser.parse_args()
if args.user == "user":
@ -206,15 +177,15 @@ def main():
print(f"Importing annotations from '{args.path[0]}'...")
elif args.ie[0] == "export":
print(f"Exporting annotations...")
sleep(1)
sleep(2)
# run code
user_id = get_user_id(args.db[0], args.user[0])
user_id = get_user_id(args.db, args.user)
if args.ie[0] == "import":
ImportService.imp(path=args.path[0], db=args.db[0], user_id=user_id, re_nav=args.replace_navidrome[0], re_re=args.replace_real[0])
ImportService.imp(args.path[0], args.db, user_id)
elif args.ie[0] == "export":
ExportService.exp(db=args.db[0], user_id=user_id, re_nav=args.replace_navidrome[0], re_re=args.replace_real[0])
ExportService.exp(args.db, user_id)
if __name__ == '__main__':