Add code

.gitignore

@@ -158,3 +158,5 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+
+config.yaml

jellyfinstats/__main__.py

@@ -0,0 +1,16 @@
+from argparse import ArgumentParser
+from .audio import generate_audio_report
+from .config import Config
+from .db import PlaybackReportingDb, LibraryDb
+
+
+p = ArgumentParser()
+p.add_argument("-c", "--config", help="The path to config file.", default="config.yaml")  # noqa: E501
+p.add_argument("--playback-reporting-db", help="The path to playback_reporting.db")  # noqa: E501
+p.add_argument("--library-db", help="The path to library.db")
+p.add_argument("--jellyfin-data-dir", help="The path to data directory.")
+arg = p.parse_intermixed_args()
+cfg = Config(arg.config, arg)
+with PlaybackReportingDb(cfg.playback_reporting_db) as pdb:
+    with LibraryDb(cfg.library_db) as ldb:
+        generate_audio_report(pdb, ldb)

jellyfinstats/audio.py

@@ -0,0 +1,39 @@
+from .db import PlaybackReportingDb, LibraryDb
+from re import compile
+
+
+ITEMNAME_PATTERN = compile(r'(?P<album_artist>.*) - (?P<track>.*) \((?P<album>.*)\)')  # noqa: E501
+
+
+def generate_audio_report(pdb: PlaybackReportingDb, ldb: LibraryDb):
+    offset = 0
+    data = pdb.get_activitys(offset, itemType='Audio')
+    count = 0
+    while len(data) > 0:
+        for d in data:
+            itemId = d['ItemId']
+            item = ldb.get_item(itemId)
+            if item:
+                pass
+            else:
+                itemName = d['ItemName']
+                re = ITEMNAME_PATTERN.match(itemName)
+                if re is None:
+                    raise ValueError(f"Failed to parse ItemName: {itemName}")
+                re = re.groupdict()
+                if re['album_artist'] == 'Not Known':
+                    re['album_artist'] = None
+                if re['album'] == 'Not Known':
+                    re['album'] = None
+                items = ldb.get_audios(re['track'], re['album'])
+                if len(items) == 1:
+                    pass
+                else:
+                    if len(items):
+                        print(items)
+                    else:
+                        print(re)
+                    count += 1
+        offset += len(data)
+        data = pdb.get_activitys(offset, itemType='Audio')
+    print('Count', count)

jellyfinstats/config.py

@@ -0,0 +1,47 @@
+from argparse import Namespace
+try:
+    from functools import cached_property
+except ImportError:
+    cached_property = property
+from os.path import join
+from yaml import load as loadyaml
+try:
+    from yaml import CSafeLoader as SafeLoader
+except ImportError:
+    from yaml import SafeLoader
+
+
+class Config:
+    def __init__(self, path: str, args: Namespace = None):
+        with open(path, encoding="UTF-8") as f:
+            self._data = loadyaml(f, Loader=SafeLoader)
+        self._args = args
+
+    @cached_property
+    def playback_reporting_db(self) -> str:
+        if self._args and self._args.playback_reporting_db:
+            return self._args.playback_reporting_db
+        if 'playback_reporting_db' in self._data and self._data['playback_reporting_db']:  # noqa: E501
+            return self._data['playback_reporting_db']
+        d = self.jellyfin_data_dir
+        if d:
+            return join(d, "playback_reporting.db")
+        raise ValueError('playback_reporting.db not set.')
+
+    @cached_property
+    def library_db(self) -> str:
+        if self._args and self._args.library_db:
+            return self._args.library_db
+        if 'library_db' in self._data and self._data['library_db']:
+            return self._data['library_db']
+        d = self.jellyfin_data_dir
+        if d:
+            return join(d, "library.db")
+        raise ValueError('library.db not set.')
+
+    @cached_property
+    def jellyfin_data_dir(self) -> str | None:
+        if self._args and self._args.jellyfin_data_dir:
+            return self._args.jellyfin_data_dir
+        if 'jellyfin_data_dir' in self._data and self._data['jellyfin_data_dir']:  # noqa: E501
+            return self._data['jellyfin_data_dir']

jellyfinstats/db.py

@@ -0,0 +1,71 @@
+import sqlite3
+
+
+class PlaybackReportingDb:
+    def __init__(self, fn: str):
+        self._db = sqlite3.connect(fn)
+        self._closed = False
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, tp, val, trace):
+        self.close()
+
+    def close(self):
+        if self._closed:
+            return
+        self._db.close()
+        self._closed = True
+
+    def get_activitys(self, offset: int = 0, limit: int = 100,
+                      itemType: str = None):
+        where_sql = ''
+        args = []
+        if itemType is not None:
+            where_sql = ' WHERE ItemType = ?'
+            args.append(itemType)
+        args.append(limit)
+        args.append(offset)
+        cur = self._db.execute(f"SELECT * FROM PlaybackActivity{where_sql} LIMIT ? OFFSET ?;", args)  # noqa: E501
+        cur.row_factory = sqlite3.Row
+        return [dict(i) for i in cur.fetchall()]
+
+
+class LibraryDb:
+    def __init__(self, fn: str):
+        self._db = sqlite3.connect(fn)
+        self._closed = False
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, tp, val, trace):
+        self.close()
+
+    def close(self):
+        if self._closed:
+            return
+        self._db.close()
+        self._closed = True
+
+    def get_items(self, offset: int = 0, limit: int = 100):
+        cur = self._db.execute("SELECT * FROM TypedBaseItems LIMIT ? OFFSET ?;", [limit, offset])  # noqa: E501
+        cur.row_factory = sqlite3.Row
+        return [dict(i) for i in cur.fetchall()]
+
+    def get_item(self, itemId: str):
+        cur = self._db.execute("SELECT * FROM TypedBaseItems WHERE PresentationUniqueKey = ?;", [itemId])  # noqa: E501
+        cur.row_factory = sqlite3.Row
+        re = cur.fetchone()
+        return dict(re) if re is not None else None
+
+    def get_audios(self, track: str, album: str = None):
+        args = ['Audio', track]
+        where_sql = ''
+        if album is not None:
+            where_sql = ' AND Album = ?'
+            args.append(album)
+        cur = self._db.execute(f"SELECT * FROM TypedBaseItems WHERE MediaType = ? AND Name = ?{where_sql};", args)  # noqa: E501
+        cur.row_factory = sqlite3.Row
+        return [dict(i) for i in cur.fetchall()]

requirements.txt

@@ -0,0 +1 @@
+pyyaml