More cleanup

src/black_white.py

@@ -2,14 +2,14 @@
 
 
 def setup_black_white_lists(
-    blacklist_library: str,
-    whitelist_library: str,
-    blacklist_library_type: str,
-    whitelist_library_type: str,
-    blacklist_users: str,
-    whitelist_users: str,
-    library_mapping=None,
-    user_mapping=None,
+    blacklist_library: list[str] | None,
+    whitelist_library: list[str] | None,
+    blacklist_library_type: list[str] | None,
+    whitelist_library_type: list[str] | None,
+    blacklist_users: list[str] | None,
+    whitelist_users: list[str] | None,
+    library_mapping: dict[str, str] | None = None,
+    user_mapping: dict[str, str] | None = None,
 ):
     blacklist_library, blacklist_library_type, blacklist_users = setup_x_lists(
         blacklist_library,
@@ -40,53 +40,44 @@ def setup_black_white_lists(
 
 
 def setup_x_lists(
-    xlist_library,
-    xlist_library_type,
-    xlist_users,
-    xlist_type,
-    library_mapping=None,
-    user_mapping=None,
-):
+    xlist_library: list[str] | None,
+    xlist_library_type: list[str] | None,
+    xlist_users: list[str] | None,
+    xlist_type: str | None,
+    library_mapping: dict[str, str] | None = None,
+    user_mapping: dict[str, str] | None = None,
+) -> tuple[list[str], list[str], list[str]]:
+    out_library: list[str] = []
     if xlist_library:
-        if len(xlist_library) > 0:
-            xlist_library = xlist_library.split(",")
-            xlist_library = [x.strip() for x in xlist_library]
-            if library_mapping:
-                temp_library = []
-                for library in xlist_library:
-                    library_other = search_mapping(library_mapping, library)
-                    if library_other:
-                        temp_library.append(library_other)
+        out_library = [x.strip() for x in xlist_library]
+        if library_mapping:
+            temp_library: list[str] = []
+            for library in xlist_library:
+                library_other = search_mapping(library_mapping, library)
+                if library_other:
+                    temp_library.append(library_other)
 
-                xlist_library = xlist_library + temp_library
-    else:
-        xlist_library = []
-    logger(f"{xlist_type}list Library: {xlist_library}", 1)
+            out_library = out_library + temp_library
+        logger(f"{xlist_type}list Library: {xlist_library}", 1)
 
+    out_library_type: list[str] = []
     if xlist_library_type:
-        if len(xlist_library_type) > 0:
-            xlist_library_type = xlist_library_type.split(",")
-            xlist_library_type = [x.lower().strip() for x in xlist_library_type]
-    else:
-        xlist_library_type = []
-    logger(f"{xlist_type}list Library Type: {xlist_library_type}", 1)
+        out_library_type = [x.lower().strip() for x in xlist_library_type]
+
+        logger(f"{xlist_type}list Library Type: {out_library_type}", 1)
 
+    out_users: list[str] = []
     if xlist_users:
-        if len(xlist_users) > 0:
-            xlist_users = xlist_users.split(",")
-            xlist_users = [x.lower().strip() for x in xlist_users]
-            if user_mapping:
-                temp_users = []
-                for user in xlist_users:
-                    user_other = search_mapping(user_mapping, user)
-                    if user_other:
-                        temp_users.append(user_other)
+        out_users = [x.lower().strip() for x in xlist_users]
+        if user_mapping:
+            temp_users: list[str] = []
+            for user in out_users:
+                user_other = search_mapping(user_mapping, user)
+                if user_other:
+                    temp_users.append(user_other)
+
+            out_users = out_users + temp_users
 
-                xlist_users = xlist_users + temp_users
-        else:
-            xlist_users = []
-    else:
-        xlist_users = []
-    logger(f"{xlist_type}list Users: {xlist_users}", 1)
+        logger(f"{xlist_type}list Users: {out_users}", 1)
 
-    return xlist_library, xlist_library_type, xlist_users
+    return out_library, out_library_type, out_users

src/connection.py

@@ -1,4 +1,5 @@
 import os
+from typing import Literal
 from dotenv import load_dotenv
 
 from src.functions import logger, str_to_bool
@@ -9,39 +10,32 @@
 load_dotenv(override=True)
 
 
-def jellyfin_emby_server_connection(server_baseurl, server_token, server_type):
-    servers = []
+def jellyfin_emby_server_connection(
+    server_baseurl: str, server_token: str, server_type: Literal["jellyfin", "emby"]
+) -> list[Jellyfin | Emby]:
+    servers: list[Jellyfin | Emby] = []
+    server: Jellyfin | Emby
 
-    server_baseurl = server_baseurl.split(",")
-    server_token = server_token.split(",")
+    server_baseurls = server_baseurl.split(",")
+    server_tokens = server_token.split(",")
 
-    if len(server_baseurl) != len(server_token):
+    if len(server_baseurls) != len(server_tokens):
         raise Exception(
             f"{server_type.upper()}_BASEURL and {server_type.upper()}_TOKEN must have the same number of entries"
         )
 
-    for i, baseurl in enumerate(server_baseurl):
+    for i, baseurl in enumerate(server_baseurls):
         baseurl = baseurl.strip()
         if baseurl[-1] == "/":
             baseurl = baseurl[:-1]
 
         if server_type == "jellyfin":
-            server = Jellyfin(baseurl=baseurl, token=server_token[i].strip())
-            servers.append(
-                (
-                    "jellyfin",
-                    server,
-                )
-            )
+            server = Jellyfin(baseurl=baseurl, token=server_tokens[i].strip())
+            servers.append(server)
 
         elif server_type == "emby":
-            server = Emby(baseurl=baseurl, token=server_token[i].strip())
-            servers.append(
-                (
-                    "emby",
-                    server,
-                )
-            )
+            server = Emby(baseurl=baseurl, token=server_tokens[i].strip())
+            servers.append(server)
         else:
             raise Exception("Unknown server type")
 
@@ -50,19 +44,19 @@ def jellyfin_emby_server_connection(server_baseurl, server_token, server_type):
     return servers
 
 
-def generate_server_connections():
-    servers = []
+def generate_server_connections() -> list[Plex | Jellyfin | Emby]:
+    servers: list[Plex | Jellyfin | Emby] = []
 
-    plex_baseurl = os.getenv("PLEX_BASEURL", None)
-    plex_token = os.getenv("PLEX_TOKEN", None)
-    plex_username = os.getenv("PLEX_USERNAME", None)
-    plex_password = os.getenv("PLEX_PASSWORD", None)
-    plex_servername = os.getenv("PLEX_SERVERNAME", None)
+    plex_baseurl_str: str | None = os.getenv("PLEX_BASEURL", None)
+    plex_token_str: str | None = os.getenv("PLEX_TOKEN", None)
+    plex_username_str: str | None = os.getenv("PLEX_USERNAME", None)
+    plex_password_str: str | None = os.getenv("PLEX_PASSWORD", None)
+    plex_servername_str: str | None = os.getenv("PLEX_SERVERNAME", None)
     ssl_bypass = str_to_bool(os.getenv("SSL_BYPASS", "False"))
 
-    if plex_baseurl and plex_token:
-        plex_baseurl = plex_baseurl.split(",")
-        plex_token = plex_token.split(",")
+    if plex_baseurl_str and plex_token_str:
+        plex_baseurl = plex_baseurl_str.split(",")
+        plex_token = plex_token_str.split(",")
 
         if len(plex_baseurl) != len(plex_token):
             raise Exception(
@@ -81,17 +75,12 @@ def generate_server_connections():
 
             logger(f"Plex Server {i} info: {server.info()}", 3)
 
-            servers.append(
-                (
-                    "plex",
-                    server,
-                )
-            )
+            servers.append(server)
 
-    if plex_username and plex_password and plex_servername:
-        plex_username = plex_username.split(",")
-        plex_password = plex_password.split(",")
-        plex_servername = plex_servername.split(",")
+    if plex_username_str and plex_password_str and plex_servername_str:
+        plex_username = plex_username_str.split(",")
+        plex_password = plex_password_str.split(",")
+        plex_servername = plex_servername_str.split(",")
 
         if len(plex_username) != len(plex_password) or len(plex_username) != len(
             plex_servername
@@ -111,16 +100,10 @@ def generate_server_connections():
             )
 
             logger(f"Plex Server {i} info: {server.info()}", 3)
-            servers.append(
-                (
-                    "plex",
-                    server,
-                )
-            )
+            servers.append(server)
 
     jellyfin_baseurl = os.getenv("JELLYFIN_BASEURL", None)
     jellyfin_token = os.getenv("JELLYFIN_TOKEN", None)
-
     if jellyfin_baseurl and jellyfin_token:
         servers.extend(
             jellyfin_emby_server_connection(
@@ -130,7 +113,6 @@ def generate_server_connections():
 
     emby_baseurl = os.getenv("EMBY_BASEURL", None)
     emby_token = os.getenv("EMBY_TOKEN", None)
-
     if emby_baseurl and emby_token:
         servers.extend(
             jellyfin_emby_server_connection(emby_baseurl, emby_token, "emby")

src/custom_types.py

@@ -0,0 +1,12 @@
+UsersWatchedStatus = dict[str, str | tuple[str] | dict[str, bool | int]]
+UsersWatched = (
+    dict[
+        frozenset[UsersWatchedStatus],
+        dict[
+            str,
+            list[UsersWatchedStatus]
+            | dict[frozenset[UsersWatchedStatus], list[UsersWatchedStatus]],
+        ],
+    ]
+    | list[UsersWatchedStatus]
+)

src/emby.py

@@ -1,5 +1,5 @@
 from src.jellyfin_emby import JellyfinEmby
-from packaging import version
+from packaging.version import parse, Version
 
 
 class Emby(JellyfinEmby):
@@ -21,5 +21,5 @@ def __init__(self, baseurl, token):
             server_type="Emby", baseurl=baseurl, token=token, headers=headers
         )
 
-    def is_partial_update_supported(self, server_version):
-        return server_version > version.parse("4.4")
+    def is_partial_update_supported(self, server_version: Version) -> bool:
+        return server_version > parse("4.4")