moved from list to dict for export

This commit is contained in:
Djairo Hougee 2025-02-06 20:41:39 +01:00
parent 1af229d596
commit f87c9a34c5
1 changed files with 14 additions and 11 deletions

View File

@ -30,6 +30,7 @@ def get_user_id(db: str, name: str) -> str:
print(f"Failed to read id from {db}: {e}") print(f"Failed to read id from {db}: {e}")
return '' return ''
class ExportService(): class ExportService():
@staticmethod @staticmethod
def exp(db: str, user_id: str): def exp(db: str, user_id: str):
@ -39,43 +40,45 @@ class ExportService():
@staticmethod @staticmethod
def get_annotations(db: str, user_id: str) -> list: def get_annotations(db: str, user_id: str) -> dict:
"""Fetches all annotations belonging to the user with id <user_id> from the provided sql3 navidrome database <db>. """Fetches all annotations belonging to the user with id <user_id> from the provided sql3 navidrome database <db>.
input: input:
db: path to a navidrome sql3 database. This database must contain at least the annotation and media_file tables. 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. user_id: the user_id to fetch annotations from.
output: output:
list of annotations in the format (path, playCount, rating, starred). dict of annotations in the format {path: (playCount, rating, starred)}.
""" """
annos = {}
try: try:
conn = sqlite3.connect(db) conn = sqlite3.connect(db)
cursor = conn.cursor() 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'""") 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'""")
annos = cursor.fetchall() for p,pc,r,s in cursor.fetchall():
annos[p] = (pc,r,s)
conn.close() conn.close()
return annos return annos
except sqlite3.OperationalError as e: except sqlite3.OperationalError as e:
print(f"Failed to read data from {db}: {e}") print(f"Failed to read data from {db}: {e}")
return [] return annos
@staticmethod @staticmethod
def write_annotations(annos: list): def write_annotations(annos: dict):
"""Writes a list of annotations to id3v2 tags. """Writes a list of annotations to id3v2 tags.
The tags used are TXXX=FNVD_<NAME>, where <NAME> is one of PlayCount, Rating, Starred. The tags used are TXXX=FNVD_<NAME>, where <NAME> is one of PlayCount, Rating, Starred.
input: input:
annos: list of annotations. Format must be (path, playCount, rating, starred). annos: dict of annotations. Format must be {path: (playCount, rating, starred)}.
""" """
count = 0 count = 0
for a in annos: for k,v in annos.items():
try: try:
file = id3.ID3(a[0]) file = id3.ID3(k)
file.add(TXXX(desc="FNVD_PlayCount", text=str(a[1]))) file.add(TXXX(desc="FNVD_PlayCount", text=str(v[0])))
file.add(TXXX(desc="FNVD_Rating", text=str(a[2]))) file.add(TXXX(desc="FNVD_Rating", text=str(v[1])))
file.add(TXXX(desc="FNVD_Starred", text=str(a[3]))) file.add(TXXX(desc="FNVD_Starred", text=str(v[2])))
file.save() file.save()
count += 1 count += 1
except MutagenError as e: except MutagenError as e: