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

added support for path substiution

Browse Source
This commit is contained in:
Djairo Hougee
2025-02-06 21:15:56 +01:00
parent f87c9a34c5
commit 57fa326da6
2 changed files with 90 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
backerupper.py
View File

@@ -1,5 +1,6 @@
#!/usr/bin/python3
import re
import sys
from time import sleep
import argparse
@@ -33,19 +34,28 @@ def get_user_id(db: str, name: str) -> str:
class ExportService():
@staticmethod
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)
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)
ExportService.write_annotations(a)
@staticmethod
def get_annotations(db: str, user_id: str) -> dict:
def get_annotations(db: str, user_id: str, re_nav: str = "", re_re:str = "") -> dict:
"""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)}.
"""
@@ -55,7 +65,7 @@ class ExportService():
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[p] = (pc,r,s)
annos[re.sub(re_nav, re_re, p)] = (pc,r,s)
conn.close()
return annos
except sqlite3.OperationalError as e:
@@ -64,18 +74,20 @@ class ExportService():
@staticmethod
def write_annotations(annos: dict):
def write_annotations(annos: dict, re_nav: str = "", re_re: str = ""):
"""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.
"""
count = 0
for k,v in annos.items():
try:
file = id3.ID3(k)
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])))
@@ -89,18 +101,28 @@ class ExportService():
class ImportService():
@staticmethod
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)
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)
ImportService.write_annotations(a, db, user_id)
@staticmethod
def get_annotations(path: str) -> dict:
def get_annotations(path: str, re_nav: str = "", re_re: 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)}.
"""
@@ -126,26 +148,28 @@ class ImportService():
starred = int(f.text[0])
if play_count or rating or starred:
annos[dir+'/'+mp3_file] = (play_count, rating, starred)
annos[re.sub(re_re, re_nav, 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):
def write_annotations(annos: dict, db: str, user_id: str, re_nav: str = "", re_re: 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(annos.keys())}')""")
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()])}')""")
# step 2: create list with item_ids
items = []
@@ -170,6 +194,8 @@ 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":
@@ -180,15 +206,15 @@ def main():
print(f"Importing annotations from '{args.path[0]}'...")
elif args.ie[0] == "export":
print(f"Exporting annotations...")
sleep(2)
sleep(1)
# run code
user_id = get_user_id(args.db, args.user)
user_id = get_user_id(args.db[0], args.user[0])
if args.ie[0] == "import":
ImportService.imp(args.path[0], args.db, user_id)
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])
elif args.ie[0] == "export":
ExportService.exp(args.db, user_id)
ExportService.exp(db=args.db[0], user_id=user_id, re_nav=args.replace_navidrome[0], re_re=args.replace_real[0])
if __name__ == '__main__':