Add parallel threading

.gitignore

@@ -1,4 +1,5 @@
 .env
+*.prof
 
 # Byte-compiled / optimized / DLL files
 __pycache__/

main.py

@@ -2,7 +2,8 @@ import copy, os, traceback, json
 from dotenv import load_dotenv
 from time import sleep
 
-from src.functions import logger, str_to_bool, search_mapping, generate_library_guids_dict
+
+from src.functions import logger, str_to_bool, search_mapping, generate_library_guids_dict, future_thread_executor
 from src.plex import Plex
 from src.jellyfin import Jellyfin
 
@@ -346,8 +347,12 @@ def main():
             # Create users list
             server_1_users, server_2_users = setup_users(server_1, server_2, blacklist_users, whitelist_users, user_mapping)
 
-            server_1_watched = server_1_connection.get_watched(server_1_users, blacklist_library, whitelist_library, blacklist_library_type, whitelist_library_type, library_mapping)
+            args = [[server_1_connection.get_watched, server_1_users, blacklist_library, whitelist_library, blacklist_library_type, whitelist_library_type, library_mapping]
-            server_2_watched = server_2_connection.get_watched(server_2_users, blacklist_library, whitelist_library, blacklist_library_type, whitelist_library_type, library_mapping)
+                    , [server_2_connection.get_watched, server_2_users, blacklist_library, whitelist_library, blacklist_library_type, whitelist_library_type, library_mapping]]
+
+            results = future_thread_executor(args)
+            server_1_watched = results[0]
+            server_2_watched = results[1]
 
             # clone watched so it isnt modified in the cleanup function so all duplicates are actually removed
             server_1_watched_filtered = copy.deepcopy(server_1_watched)
@@ -362,10 +367,10 @@ def main():
             logger(f"server 1 watched that needs to be synced to server 2:\n{server_1_watched_filtered}", 1)
             logger(f"server 2 watched that needs to be synced to server 1:\n{server_2_watched_filtered}", 1)
 
-            # Update watched status
+            args= [[server_1_connection.update_watched, server_2_watched_filtered, user_mapping, library_mapping, dryrun]
-            server_1_connection.update_watched(server_2_watched_filtered, user_mapping, library_mapping, dryrun)
+                , [server_2_connection.update_watched, server_1_watched_filtered, user_mapping, library_mapping, dryrun]]
-            server_2_connection.update_watched(server_1_watched_filtered, user_mapping, library_mapping, dryrun)
 
+            future_thread_executor(args)
 
 if __name__ == "__main__":
     sleep_timer = float(os.getenv("SLEEP_TIMER", "3600"))

src/functions.py

@@ -1,10 +1,12 @@
 import os
+from concurrent.futures import ThreadPoolExecutor
 from dotenv import load_dotenv
+
 load_dotenv(override=True)
 
 logfile = os.getenv("LOGFILE","log.log")
 
-def logger(message, log_type=0):
+def logger(message: str, log_type=0):
     debug = str_to_bool(os.getenv("DEBUG", "True"))
     debug_level = os.getenv("DEBUG_LEVEL", "info").lower()
 
@@ -114,3 +116,20 @@ def generate_library_guids_dict(user_list: dict, generate_output: int):
 
     return show_output_dict, episode_output_dict, movies_output_dict
 
+def future_thread_executor(args: list):
+    futures_list = []
+    results = []
+
+    with ThreadPoolExecutor() as executor:
+        for arg in args:
+            # * arg unpacks the list into actual arguments
+            futures_list.append(executor.submit(*arg))
+
+        for future in futures_list:
+            try:
+                result = future.result()
+                results.append(result)
+            except Exception as e:
+                raise Exception(e)
+    
+    return results

src/jellyfin.py

@@ -1,10 +1,11 @@
 import requests
-from src.functions import logger, search_mapping, str_to_bool, check_skip_logic, generate_library_guids_dict
+from src.functions import logger, search_mapping, str_to_bool, check_skip_logic, generate_library_guids_dict, future_thread_executor
 
 class Jellyfin():
     def __init__(self, baseurl, token):
         self.baseurl = baseurl
         self.token = token
+        self.session = requests.Session()
 
         if not self.baseurl:
             raise Exception("Jellyfin baseurl not set")
@@ -19,8 +20,12 @@ class Jellyfin():
         try:
             response = None
 
+            headers = {
+                "Accept": "application/json",
+                "X-Emby-Token": self.token
+            }
             if query_type == "get":
-                response = requests.get(self.baseurl + query, headers={"accept":"application/json", "X-Emby-Token": self.token})
+                response = self.session.get(self.baseurl + query, headers=headers)
 
             elif query_type == "post":
                 authorization = (
@@ -30,7 +35,8 @@ class Jellyfin():
                     'DeviceId="script", '
                     'Version="0.0.0"'
                 )
-                response = requests.post(self.baseurl + query, headers={"accept":"application/json", "X-Emby-Authorization": authorization, "X-Emby-Token": self.token})
+                headers["X-Emby-Authorization"] = authorization
+                response = self.session.post(self.baseurl + query, headers=headers)
 
             return response.json()
         except Exception as e:
@@ -50,8 +56,60 @@ class Jellyfin():
 
         return users
 
+    def get_user_watched(self, user_name, user_id, library_type, library_id, library_title):
+        user_watched = {}
+
+        logger(f"Jellyfin: Generating watched for {user_name} in library {library_title}", 0)
+        # Movies
+        if library_type == "Movie":
+            watched = self.query(f"/Users/{user_id}/Items?SortBy=SortName&SortOrder=Ascending&Recursive=true&ParentId={library_id}&Filters=IsPlayed&Fields=ItemCounts,ProviderIds", "get")
+            for movie in watched["Items"]:
+                if movie["UserData"]["Played"] == True:
+                    if movie["ProviderIds"]:
+                        if user_name not in user_watched:
+                            user_watched[user_name] = {}
+                        if library_title not in user_watched[user_name]:
+                            user_watched[user_name][library_title] = []
+                        # Lowercase movie["ProviderIds"] keys
+                        movie["ProviderIds"] = {k.lower(): v for k, v in movie["ProviderIds"].items()}
+                        user_watched[user_name][library_title].append(movie["ProviderIds"])
+
+        # TV Shows
+        if library_type == "Episode":
+            watched = self.query(f"/Users/{user_id}/Items?SortBy=SortName&SortOrder=Ascending&Recursive=true&ParentId={library_id}&Fields=ItemCounts,ProviderIds", "get")
+            watched_shows = [x for x in watched["Items"] if x["Type"] == "Series"]
+
+            for show in watched_shows:
+                show_guids = {k.lower(): v for k, v in show["ProviderIds"].items()}
+                show_guids["title"] = show["Name"]
+                show_guids = frozenset(show_guids.items())
+                seasons = self.query(f"/Shows/{show['Id']}/Seasons?userId={user_id}&Fields=ItemCounts,ProviderIds", "get")
+                if len(seasons["Items"]) > 0:
+                    for season in seasons["Items"]:
+                        episodes = self.query(f"/Shows/{show['Id']}/Episodes?seasonId={season['Id']}&userId={user_id}&Fields=ItemCounts,ProviderIds", "get")
+                        if len(episodes["Items"]) > 0:
+                            for episode in episodes["Items"]:
+                                if episode["UserData"]["Played"] == True:
+                                    if episode["ProviderIds"]:
+                                        if user_name not in user_watched:
+                                            user_watched[user_name] = {}
+                                        if library_title not in user_watched[user_name]:
+                                            user_watched[user_name][library_title] = {}
+                                        if show_guids not in user_watched[user_name][library_title]:
+                                            user_watched[user_name][library_title][show_guids] = {}
+                                        if season["Name"] not in user_watched[user_name][library_title][show_guids]:
+                                            user_watched[user_name][library_title][show_guids][season["Name"]] = []
+
+                                        # Lowercase episode["ProviderIds"] keys
+                                        episode["ProviderIds"] = {k.lower(): v for k, v in episode["ProviderIds"].items()}
+                                        user_watched[user_name][library_title][show_guids][season["Name"]].append(episode["ProviderIds"])
+
+        return user_watched
+
+
     def get_watched(self, users, blacklist_library, whitelist_library, blacklist_library_type, whitelist_library_type, library_mapping=None):
         users_watched = {}
+        args = []
         
         for user_name, user_id in users.items():
             # Get all libraries
@@ -76,102 +134,15 @@ class Jellyfin():
                     logger(f"Jellyfin: Skipping library {library_title} {skip_reason}", 1)
                     continue
 
-                logger(f"Jellyfin: Generating watched for {user_name} in library {library_title}", 0)
+                args.append([self.get_user_watched, user_name, user_id, library_type, library_id, library_title])
-                # Movies
-                if library_type == "Movie":
-                    watched = self.query(f"/Users/{user_id}/Items?SortBy=SortName&SortOrder=Ascending&Recursive=true&ParentId={library_id}&Filters=IsPlayed&Fields=ItemCounts,ProviderIds", "get")
-                    for movie in watched["Items"]:
-                        if movie["UserData"]["Played"] == True:
-                            if movie["ProviderIds"]:
-                                if user_name not in users_watched:
-                                    users_watched[user_name] = {}
-                                if library_title not in users_watched[user_name]:
-                                    users_watched[user_name][library_title] = []
-                                # Lowercase movie["ProviderIds"] keys
-                                movie["ProviderIds"] = {k.lower(): v for k, v in movie["ProviderIds"].items()}
-                                users_watched[user_name][library_title].append(movie["ProviderIds"])
 
-                # TV Shows
+        for user_watched in future_thread_executor(args):
-                if library_type == "Episode":
+            users_watched.update(user_watched)
-                    watched = self.query(f"/Users/{user_id}/Items?SortBy=SortName&SortOrder=Ascending&Recursive=true&ParentId={library_id}&Fields=ItemCounts,ProviderIds", "get")
-                    watched_shows = [x for x in watched["Items"] if x["Type"] == "Series"]
-
-                    for show in watched_shows:
-                        show_guids = {k.lower(): v for k, v in show["ProviderIds"].items()}
-                        show_guids["title"] = show["Name"]
-                        show_guids = frozenset(show_guids.items())
-                        seasons = self.query(f"/Shows/{show['Id']}/Seasons?userId={user_id}&Fields=ItemCounts,ProviderIds", "get")
-                        if len(seasons["Items"]) > 0:
-                            for season in seasons["Items"]:
-                                episodes = self.query(f"/Shows/{show['Id']}/Episodes?seasonId={season['Id']}&userId={user_id}&Fields=ItemCounts,ProviderIds", "get")
-                                if len(episodes["Items"]) > 0:
-                                    for episode in episodes["Items"]:
-                                        if episode["UserData"]["Played"] == True:
-                                            if episode["ProviderIds"]:
-                                                if user_name not in users_watched:
-                                                    users_watched[user_name] = {}
-                                                if library_title not in users_watched[user_name]:
-                                                    users_watched[user_name][library_title] = {}
-                                                if show_guids not in users_watched[user_name][library_title]:
-                                                    users_watched[user_name][library_title][show_guids] = {}
-                                                if season["Name"] not in users_watched[user_name][library_title][show_guids]:
-                                                    users_watched[user_name][library_title][show_guids][season["Name"]] = []
-
-                                                # Lowercase episode["ProviderIds"] keys
-                                                episode["ProviderIds"] = {k.lower(): v for k, v in episode["ProviderIds"].items()}
-                                                users_watched[user_name][library_title][show_guids][season["Name"]].append(episode["ProviderIds"])
 
         return users_watched
 
-    def update_watched(self, watched_list, user_mapping=None, library_mapping=None, dryrun=False):
-        for user, libraries in watched_list.items():
-            user_other = None
-            if user_mapping:
-                if user in user_mapping.keys():
-                    user_other = user_mapping[user]
-                elif user in user_mapping.values():
-                    user_other = search_mapping(user_mapping, user)
 
-            user_id = None
+    def update_user_watched(self, user, user_id, library, library_id, videos, dryrun):
-            for key in self.users.keys():
-                if user.lower() == key.lower():
-                    user_id = self.users[key]
-                    break
-                elif user_other and user_other.lower() == key.lower():
-                    user_id = self.users[key]
-                    break
-
-            if not user_id:
-                logger(f"{user} {user_other} not found in Jellyfin", 2)
-                break
-
-            jellyfin_libraries = self.query(f"/Users/{user_id}/Views", "get")["Items"]
-
-            for library, videos in libraries.items():
-                library_other = None
-                if library_mapping:
-                    if library in library_mapping.keys():
-                        library_other = library_mapping[library]
-                    elif library in library_mapping.values():
-                        library_other = search_mapping(library_mapping, library)
-
-
-                if library.lower() not in [x["Name"].lower() for x in jellyfin_libraries]:
-                    if library_other and library_other.lower() in [x["Name"].lower() for x in jellyfin_libraries]:
-                        logger(f"Plex: Library {library} not found, but {library_other} found, using {library_other}", 1)
-                        library = library_other
-                    else:
-                        logger(f"Library {library} {library_other} not found in Plex library list", 2)
-                        continue
-
-
-                library_id = None
-                for jellyfin_library in jellyfin_libraries:
-                    if jellyfin_library["Name"] == library:
-                        library_id = jellyfin_library["Id"]
-                        continue
-
-                if library_id:
         logger(f"Jellyfin: Updating watched for {user} in library {library}", 1)
         library_search = self.query(f"/Users/{user_id}/Items?SortBy=SortName&SortOrder=Ascending&Recursive=true&ParentId={library_id}&limit=1", "get")
         library_type = library_search["Items"][0]["Type"]
@@ -228,3 +199,58 @@ class Jellyfin():
 
                     if show_found:
                         break
+
+
+    def update_watched(self, watched_list, user_mapping=None, library_mapping=None, dryrun=False):
+        args = []
+        for user, libraries in watched_list.items():
+            user_other = None
+            if user_mapping:
+                if user in user_mapping.keys():
+                    user_other = user_mapping[user]
+                elif user in user_mapping.values():
+                    user_other = search_mapping(user_mapping, user)
+
+            user_id = None
+            for key in self.users.keys():
+                if user.lower() == key.lower():
+                    user_id = self.users[key]
+                    break
+                elif user_other and user_other.lower() == key.lower():
+                    user_id = self.users[key]
+                    break
+
+            if not user_id:
+                logger(f"{user} {user_other} not found in Jellyfin", 2)
+                continue
+
+            jellyfin_libraries = self.query(f"/Users/{user_id}/Views", "get")["Items"]
+
+            for library, videos in libraries.items():
+                library_other = None
+                if library_mapping:
+                    if library in library_mapping.keys():
+                        library_other = library_mapping[library]
+                    elif library in library_mapping.values():
+                        library_other = search_mapping(library_mapping, library)
+
+
+                if library.lower() not in [x["Name"].lower() for x in jellyfin_libraries]:
+                    if library_other and library_other.lower() in [x["Name"].lower() for x in jellyfin_libraries]:
+                        logger(f"Plex: Library {library} not found, but {library_other} found, using {library_other}", 1)
+                        library = library_other
+                    else:
+                        logger(f"Library {library} {library_other} not found in Plex library list", 2)
+                        continue
+
+
+                library_id = None
+                for jellyfin_library in jellyfin_libraries:
+                    if jellyfin_library["Name"] == library:
+                        library_id = jellyfin_library["Id"]
+                        continue
+
+                if library_id:
+                    args.append([self.update_user_watched, user, user_id, library, library_id, videos, dryrun])
+
+        future_thread_executor(args)

src/plex.py

@@ -1,9 +1,11 @@
 import re
 
-from src.functions import logger, search_mapping, check_skip_logic, generate_library_guids_dict
 from plexapi.server import PlexServer
 from plexapi.myplex import MyPlexAccount
 
+from src.functions import logger, search_mapping, check_skip_logic, generate_library_guids_dict, future_thread_executor
+
+
 # class plex accept base url and token and username and password but default with none
 class Plex:
     def __init__(self, baseurl=None, token=None, username=None, password=None, servername=None):
@@ -117,60 +119,20 @@ class Plex:
                 logger(f"Plex: Skipping library {library_title} {skip_reason}", 1)
                 continue
 
+            args = []
             for user in users:
                 logger(f"Plex: Generating watched for {user.title} in library {library_title}", 0)
                 user_name = user.title.lower()
-                watched = self.get_user_watched(user, library)
+                watched = args.append([self.get_user_watched, user, library])
-                if watched:
+
+            for user_watched in future_thread_executor(args):
+                if user_watched:
                     if user_name not in users_watched:
                         users_watched[user_name] = {}
-                    if library_title not in users_watched[user_name]:
+                    users_watched[user_name][library_title] = user_watched
-                        users_watched[user_name][library_title] = []
-                    users_watched[user_name][library_title] = watched
-
         return users_watched
 
-    def update_watched(self, watched_list, user_mapping=None, library_mapping=None, dryrun=False):
+    def update_user_watched (self, user, user_plex, library, videos, dryrun):
-        for user, libraries in watched_list.items():
-            user_other = None
-            # If type of user is dict
-            if user_mapping:
-                if user in user_mapping.keys():
-                    user_other = user_mapping[user]
-                elif user in user_mapping.values():
-                    user_other = search_mapping(user_mapping, user)
-
-            for index, value in enumerate(self.users):
-                if user.lower() == value.title.lower():
-                    user = self.users[index]
-                    break
-                elif user_other and user_other.lower() == value.title.lower():
-                    user = self.users[index]
-                    break
-
-            if self.admin_user == user:
-                user_plex = self.plex
-            else:
-                user_plex = PlexServer(self.baseurl, user.get_token(self.plex.machineIdentifier))
-
-            for library, videos in libraries.items():
-                library_other = None
-                if library_mapping:
-                    if library in library_mapping.keys():
-                        library_other = library_mapping[library]
-                    elif library in library_mapping.values():
-                        library_other = search_mapping(library_mapping, library)
-
-                # if library in plex library list
-                library_list = user_plex.library.sections()
-                if library.lower() not in [x.title.lower() for x in library_list]:
-                    if library_other and library_other.lower() in [x.title.lower() for x in library_list]:
-                        logger(f"Plex: Library {library} not found, but {library_other} found, using {library_other}", 1)
-                        library = library_other
-                    else:
-                        logger(f"Library {library} {library_other} not found in Plex library list", 2)
-                        continue
-
         logger(f"Plex: Updating watched for {user.title} in library {library}", 1)
         library_videos = user_plex.library.section(library)
 
@@ -226,3 +188,53 @@ class Plex:
 
                     if show_found:
                         break
+
+
+
+    def update_watched(self, watched_list, user_mapping=None, library_mapping=None, dryrun=False):
+        args = []
+
+        for user, libraries in watched_list.items():
+            user_other = None
+            # If type of user is dict
+            if user_mapping:
+                if user in user_mapping.keys():
+                    user_other = user_mapping[user]
+                elif user in user_mapping.values():
+                    user_other = search_mapping(user_mapping, user)
+
+            for index, value in enumerate(self.users):
+                if user.lower() == value.title.lower():
+                    user = self.users[index]
+                    break
+                elif user_other and user_other.lower() == value.title.lower():
+                    user = self.users[index]
+                    break
+
+            if self.admin_user == user:
+                user_plex = self.plex
+            else:
+                user_plex = PlexServer(self.baseurl, user.get_token(self.plex.machineIdentifier))
+
+            for library, videos in libraries.items():
+                library_other = None
+                if library_mapping:
+                    if library in library_mapping.keys():
+                        library_other = library_mapping[library]
+                    elif library in library_mapping.values():
+                        library_other = search_mapping(library_mapping, library)
+
+                # if library in plex library list
+                library_list = user_plex.library.sections()
+                if library.lower() not in [x.title.lower() for x in library_list]:
+                    if library_other and library_other.lower() in [x.title.lower() for x in library_list]:
+                        logger(f"Plex: Library {library} not found, but {library_other} found, using {library_other}", 1)
+                        library = library_other
+                    else:
+                        logger(f"Library {library} {library_other} not found in Plex library list", 2)
+                        continue
+
+
+                args.append([self.update_user_watched, user, user_plex, library, videos, dryrun])
+
+        future_thread_executor(args)