bel
/
JellyPlex-Watched
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Use list instead of tuple for servers 

Signed-off-by: Luis Garcia <git@luigi311.com>
pull/210/head
Luis Garcia 2024-11-12 22:49:20 -07:00
parent 170757aca1
commit 46fa5e7c9a
3 changed files with 67 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
src/connection.py
View File

@ -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")
)

66
src/main.py
View File

@ -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,

33
src/users.py
View File

@ -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 (