diff --git a/src/connection.py b/src/connection.py index ce4192b..39be764 100644 --- a/src/connection.py +++ b/src/connection.py @@ -12,8 +12,9 @@ load_dotenv(override=True) def jellyfin_emby_server_connection( server_baseurl: str, server_token: str, server_type: Literal["jellyfin", "emby"] -) -> list[tuple[Literal["jellyfin", "emby"], Jellyfin | Emby]]: - servers: list[tuple[Literal["jellyfin", "emby"], Jellyfin | Emby]] = [] +) -> list[Jellyfin | Emby]: + servers: list[Jellyfin | Emby] = [] + server: Jellyfin | Emby server_baseurls = server_baseurl.split(",") server_tokens = server_token.split(",") @@ -30,21 +31,11 @@ def jellyfin_emby_server_connection( if server_type == "jellyfin": server = Jellyfin(baseurl=baseurl, token=server_tokens[i].strip()) - servers.append( - ( - "jellyfin", - server, - ) - ) + servers.append(server) elif server_type == "emby": server = Emby(baseurl=baseurl, token=server_tokens[i].strip()) - servers.append( - ( - "emby", - server, - ) - ) + servers.append(server) else: raise Exception("Unknown server type") @@ -53,23 +44,19 @@ def jellyfin_emby_server_connection( return servers -def generate_server_connections() -> ( - list[tuple[Literal["plex", "jellyfin", "emby"], Plex | Jellyfin | Emby]] -): - servers: list[ - tuple[Literal["plex", "jellyfin", "emby"], Plex | Jellyfin | Emby] - ] = [] +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( @@ -88,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 @@ -118,12 +100,7 @@ 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) @@ -137,7 +114,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") ) diff --git a/src/main.py b/src/main.py index 116d75f..91765b1 100644 --- a/src/main.py +++ b/src/main.py @@ -1,8 +1,10 @@ import os, traceback, json -from typing import Literal from dotenv import load_dotenv from time import sleep, perf_counter +from src.emby import Emby +from src.jellyfin import Jellyfin +from src.plex import Plex from src.library import setup_libraries from src.functions import ( logger, @@ -20,8 +22,8 @@ load_dotenv(override=True) def should_sync_server( - server_1_type: Literal["plex", "jellyfin", "emby"], - server_2_type: Literal["plex", "jellyfin", "emby"], + server_1: Plex | Jellyfin | Emby, + server_2: Plex | Jellyfin | Emby, ) -> bool: sync_from_plex_to_jellyfin = str_to_bool( os.getenv("SYNC_FROM_PLEX_TO_JELLYFIN", "True") @@ -45,42 +47,42 @@ def should_sync_server( ) sync_from_emby_to_emby = str_to_bool(os.getenv("SYNC_FROM_EMBY_TO_EMBY", "True")) - if server_1_type == "plex": - if server_2_type == "jellyfin" and not sync_from_plex_to_jellyfin: + if isinstance(server_1, Plex): + if isinstance(server_2, Jellyfin) and not sync_from_plex_to_jellyfin: logger("Sync from plex -> jellyfin is disabled", 1) return False - if server_2_type == "emby" and not sync_from_plex_to_emby: + if isinstance(server_2, Emby) and not sync_from_plex_to_emby: logger("Sync from plex -> emby is disabled", 1) return False - if server_2_type == "plex" and not sync_from_plex_to_plex: + if isinstance(server_2, Plex) and not sync_from_plex_to_plex: logger("Sync from plex -> plex is disabled", 1) return False - if server_1_type == "jellyfin": - if server_2_type == "plex" and not sync_from_jelly_to_plex: + if isinstance(server_1, Jellyfin): + if isinstance(server_2, Plex) and not sync_from_jelly_to_plex: logger("Sync from jellyfin -> plex is disabled", 1) return False - if server_2_type == "jellyfin" and not sync_from_jelly_to_jellyfin: + if isinstance(server_2, Jellyfin) and not sync_from_jelly_to_jellyfin: logger("Sync from jellyfin -> jellyfin is disabled", 1) return False - if server_2_type == "emby" and not sync_from_jelly_to_emby: + if isinstance(server_2, Emby) and not sync_from_jelly_to_emby: logger("Sync from jellyfin -> emby is disabled", 1) return False - if server_1_type == "emby": - if server_2_type == "plex" and not sync_from_emby_to_plex: + if isinstance(server_1, Emby): + if isinstance(server_2, Plex) and not sync_from_emby_to_plex: logger("Sync from emby -> plex is disabled", 1) return False - if server_2_type == "jellyfin" and not sync_from_emby_to_jellyfin: + if isinstance(server_2, Jellyfin) and not sync_from_emby_to_jellyfin: logger("Sync from emby -> jellyfin is disabled", 1) return False - if server_2_type == "emby" and not sync_from_emby_to_emby: + if isinstance(server_2, Emby) and not sync_from_emby_to_emby: logger("Sync from emby -> emby is disabled", 1) return False @@ -147,13 +149,13 @@ def main_loop(): # Start server_2 at the next server in the list for server_2 in servers[servers.index(server_1) + 1 :]: # Check if server 1 and server 2 are going to be synced in either direction, skip if not - if not should_sync_server( - server_1[0], server_2[0] - ) and not should_sync_server(server_2[0], server_1[0]): + if not should_sync_server(server_1, server_2) and not should_sync_server( + server_2, server_1 + ): continue - logger(f"Server 1: {server_1[0].capitalize()}: {server_1[1].info()}", 0) - logger(f"Server 2: {server_2[0].capitalize()}: {server_2[1].info()}", 0) + logger(f"Server 1: {type(server_1)}: {server_1.info()}", 0) + logger(f"Server 2: {type(server_2)}: {server_2.info()}", 0) # Create users list logger("Creating users list", 1) @@ -162,8 +164,8 @@ def main_loop(): ) server_1_libraries, server_2_libraries = setup_libraries( - server_1[1], - server_2[1], + server_1, + server_2, blacklist_library, blacklist_library_type, whitelist_library, @@ -172,14 +174,10 @@ def main_loop(): ) logger("Creating watched lists", 1) - server_1_watched = server_1[1].get_watched( - server_1_users, server_1_libraries - ) + server_1_watched = server_1.get_watched(server_1_users, server_1_libraries) logger("Finished creating watched list server 1", 1) - server_2_watched = server_2[1].get_watched( - server_2_users, server_2_libraries - ) + server_2_watched = server_2.get_watched(server_2_users, server_2_libraries) logger("Finished creating watched list server 2", 1) logger(f"Server 1 watched: {server_1_watched}", 3) @@ -204,18 +202,18 @@ def main_loop(): 1, ) - if should_sync_server(server_2[0], server_1[0]): - logger(f"Syncing {server_2[1].info()} -> {server_1[1].info()}", 0) - server_1[1].update_watched( + if should_sync_server(server_2, server_1): + logger(f"Syncing {server_2.info()} -> {server_1.info()}", 0) + server_1.update_watched( server_2_watched_filtered, user_mapping, library_mapping, dryrun, ) - if should_sync_server(server_1[0], server_2[0]): - logger(f"Syncing {server_1[1].info()} -> {server_2[1].info()}", 0) - server_2[1].update_watched( + if should_sync_server(server_1, server_2): + logger(f"Syncing {server_1.info()} -> {server_2.info()}", 0) + server_2.update_watched( server_1_watched_filtered, user_mapping, library_mapping, diff --git a/src/users.py b/src/users.py index a659cdd..1163655 100644 --- a/src/users.py +++ b/src/users.py @@ -1,4 +1,3 @@ -from typing import Literal from plexapi.myplex import MyPlexAccount from src.emby import Emby from src.jellyfin import Jellyfin @@ -9,22 +8,18 @@ from src.functions import ( ) -def generate_user_list( - server: tuple[Literal["plex", "jellyfin", "emby"], Plex | Jellyfin | Emby] -) -> list[str]: +def generate_user_list(server: Plex | Jellyfin | Emby) -> list[str]: # generate list of users from server 1 and server 2 - server_type = server[0] - server_connection = server[1] server_users: list[str] = [] - if server_type == "plex": - for user in server_connection.users: + if isinstance(server, Plex): + for user in server.users: server_users.append( user.username.lower() if user.username else user.title.lower() ) - elif server_type in ["jellyfin", "emby"]: - server_users = [key.lower() for key in server_connection.users.keys()] + elif isinstance(server, (Jellyfin, Emby)): + server_users = [key.lower() for key in server.users.keys()] return server_users @@ -78,12 +73,12 @@ def filter_user_lists( def generate_server_users( - server: tuple[Literal["plex", "jellyfin", "emby"], Plex | Jellyfin | Emby], + server: Plex | Jellyfin | Emby, users: dict[str, str], ) -> list[MyPlexAccount] | dict[str, str] | None: - if server[0] == "plex": + if isinstance(server, Plex): plex_server_users: list[MyPlexAccount] = [] - for plex_user in server[1].users: + for plex_user in server.users: username_title = ( plex_user.username if plex_user.username else plex_user.title ) @@ -95,9 +90,9 @@ def generate_server_users( plex_server_users.append(plex_user) return plex_server_users - elif server[0] in ["jellyfin", "emby"]: + elif isinstance(server, (Jellyfin, Emby)): jelly_emby_server_users: dict[str, str] = {} - for jellyfin_user, jellyfin_id in server[1].users.items(): + for jellyfin_user, jellyfin_id in server.users.items(): if ( jellyfin_user.lower() in users.keys() or jellyfin_user.lower() in users.values() @@ -110,8 +105,8 @@ def generate_server_users( def setup_users( - server_1: tuple[Literal["plex", "jellyfin", "emby"], Plex | Jellyfin | Emby], - server_2: tuple[Literal["plex", "jellyfin", "emby"], Plex | Jellyfin | Emby], + server_1: Plex | Jellyfin | Emby, + server_2: Plex | Jellyfin | Emby, blacklist_users: list[str], whitelist_users: list[str], user_mapping: dict[str, str] | None = None, @@ -133,12 +128,12 @@ def setup_users( # Check if users is none or empty if output_server_1_users is None or len(output_server_1_users) == 0: logger( - f"No users found for server 1 {server_1[0]}, users: {server_1_users}, overlapping users {users}, filtered users {users_filtered}, server 1 users {server_1[1].users}" + f"No users found for server 1 {type(server_1)}, users: {server_1_users}, overlapping users {users}, filtered users {users_filtered}, server 1 users {server_1.users}" ) if output_server_2_users is None or len(output_server_2_users) == 0: logger( - f"No users found for server 2 {server_2[0]}, users: {server_2_users}, overlapping users {users} filtered users {users_filtered}, server 2 users {server_2[1].users}" + f"No users found for server 2 {type(server_2)}, users: {server_2_users}, overlapping users {users} filtered users {users_filtered}, server 2 users {server_2.users}" ) if (