chore: apply black formatting

backend/controllers/default.py

@@ -13,7 +13,7 @@ async def root():
     return {
         "success": True,
         "message": "Iceburg is running!",
-        "version": settings_manager.settings.version
+        "version": settings_manager.settings.version,
     }
 
 
@@ -50,7 +50,4 @@ async def get_services(request: Request):
             if getattr(service, "sm", False):
                 for sub_service in service.sm.services:
                     data[sub_service.key] = sub_service.initialized
-    return {
-        "success": True,
-        "data": data
-    }
+    return {"success": True, "data": data}

backend/controllers/items.py

@@ -49,7 +49,4 @@ async def get_imdb_info(request: Request, imdb_id: str):
     item = request.app.program.media_items.get_item_by_imdb_id(imdb_id)
     if item is None:
         raise HTTPException(status_code=404, detail="Item not found")
-    return {
-        "success": True,
-        "item": item.to_extended_dict()
-    }
+    return {"success": True, "item": item.to_extended_dict()}

backend/controllers/settings.py

@@ -48,14 +48,14 @@ async def get_settings(paths: str):
     current_settings = settings_manager.settings.dict()
     data = {}
     for path in paths.split(","):
-        keys = path.split('.')
+        keys = path.split(".")
         current_obj = current_settings
 
         for k in keys:
             if k not in current_obj:
                 return None
             current_obj = current_obj[k]
-        
+
         data[path] = current_obj
 
     return {
@@ -67,32 +67,34 @@ async def get_settings(paths: str):
 @router.post("/set")
 async def set_settings(settings: List[SetSettings]):
     current_settings = settings_manager.settings.dict()
-    
+
     for setting in settings:
-        keys = setting.key.split('.')
+        keys = setting.key.split(".")
         current_obj = current_settings
-        
+
         # Navigate to the last key's parent object, similar to the getter.
         for k in keys[:-1]:
             if k not in current_obj:
                 # If a key in the path does not exist, raise an exception or optionally create a new dict.
-                raise HTTPException(status_code=400, detail=f"Path '{'.'.join(keys[:-1])}' does not exist.")
+                raise HTTPException(
+                    status_code=400,
+                    detail=f"Path '{'.'.join(keys[:-1])}' does not exist.",
+                )
             current_obj = current_obj[k]
-        
+
         # Set the value at the final key.
         if keys[-1] in current_obj:
             current_obj[keys[-1]] = setting.value
         else:
             # If the final key does not exist, raise an exception.
-            raise HTTPException(status_code=400, detail=f"Key '{keys[-1]}' does not exist in path '{'.'.join(keys[:-1])}'.")
-
+            raise HTTPException(
+                status_code=400,
+                detail=f"Key '{keys[-1]}' does not exist in path '{'.'.join(keys[:-1])}'.",
+            )
 
     settings_manager.load(settings_dict=current_settings)
-    
+
     # Notify observers about the update.
     settings_manager.notify_observers()
-
-    return {
-        "success": True,
-        "message": "Settings updated successfully."
-    }
+
+    return {"success": True, "message": "Settings updated successfully."}

backend/main.py

@@ -12,9 +12,10 @@
 import argparse
 
 parser = argparse.ArgumentParser()
-parser.add_argument('--dev', action='store_true', help='Enable development mode')
+parser.add_argument("--dev", action="store_true", help="Enable development mode")
 args = parser.parse_args()
 
+
 class Server(uvicorn.Server):
     def install_signal_handlers(self):
         pass
@@ -32,6 +33,7 @@ def run_in_thread(self):
             self.should_exit = True
             sys.exit(0)
 
+
 app = FastAPI()
 app.program = Program(args)
 

backend/program/__init__.py

@@ -24,8 +24,7 @@ def __init__(self, args):
         self.running = False
         self.startup_args = args
         logger.configure_logger(
-            debug=settings_manager.settings.debug,
-            log=settings_manager.settings.log
+            debug=settings_manager.settings.debug, log=settings_manager.settings.log
         )
 
     def start(self):
@@ -36,7 +35,9 @@ def start(self):
         if not self.startup_args.dev:
             self.pickly = Pickly(self.media_items, data_dir_path)
             self.pickly.start()
-        self.core_manager = ServiceManager(self.media_items, True, Content, Plex, Scraping, Debrid, Symlinker)
+        self.core_manager = ServiceManager(
+            self.media_items, True, Content, Plex, Scraping, Debrid, Symlinker
+        )
         if self.validate():
             logger.info("Iceberg started!")
         else:
@@ -66,4 +67,4 @@ def stop(self):
                 service.stop()
         self.pickly.stop()
         settings_manager.save()
-        self.running = False
+        self.running = False

backend/program/content/__init__.py

@@ -14,9 +14,13 @@ def __init__(self, media_items):
         self.initialized = False
         self.key = "content"
         self.running = False
-        self.sm = ServiceManager(media_items, False, Overseerr, PlexWatchlist, Listrr, Mdblist)
+        self.sm = ServiceManager(
+            media_items, False, Overseerr, PlexWatchlist, Listrr, Mdblist
+        )
         if not self.validate():
-            logger.error("You have no content services enabled, please enable at least one!")
+            logger.error(
+                "You have no content services enabled, please enable at least one!"
+            )
             return
         self._get_content()
         self.initialized = True

backend/program/content/base.py

@@ -14,13 +14,17 @@ def __init__(self, media_items: MediaItemContainer):
 
     def validate(self):
         """Validate the content provider settings."""
-        raise NotImplementedError("The 'validate' method must be implemented by subclasses.")
+        raise NotImplementedError(
+            "The 'validate' method must be implemented by subclasses."
+        )
 
     def run(self):
         """Fetch new media from the content provider."""
         raise NotImplementedError("The 'run' method must be implemented by subclasses.")
 
-    def process_items(self, items: MediaItemContainer, requested_by: str) -> MediaItemContainer:
+    def process_items(
+        self, items: MediaItemContainer, requested_by: str
+    ) -> MediaItemContainer:
         """Process fetched media items and log the results."""
         new_items = [item for item in items if self.is_valid_item(item)]
         if not new_items:
@@ -34,5 +38,7 @@ def process_items(self, items: MediaItemContainer, requested_by: str) -> MediaIt
 
     def is_valid_item(self, item: MediaItem) -> bool:
         """Check if an imdb_id is valid for processing and not already in media_items"""
-        is_unique = not any(existing_item.imdb_id == item for existing_item in self.media_items.items)
-        return item is not None and is_unique
+        is_unique = not any(
+            existing_item.imdb_id == item for existing_item in self.media_items.items
+        )
+        return item is not None and is_unique

backend/program/content/listrr.py

@@ -32,8 +32,10 @@ def validate(self) -> bool:
             logger.error("Listrr api key is not set or invalid.")
             return False
         valid_list_found = False
-        for list_name, content_list in [('movie_lists', self.settings.movie_lists), 
-                                        ('show_lists', self.settings.show_lists)]:
+        for list_name, content_list in [
+            ("movie_lists", self.settings.movie_lists),
+            ("show_lists", self.settings.show_lists),
+        ]:
             if content_list is None or not any(content_list):
                 continue
             for item in content_list:
@@ -46,7 +48,9 @@ def validate(self) -> bool:
         try:
             response = ping("https://listrr.pro/", additional_headers=self.headers)
             if not response.ok:
-                logger.error(f"Listrr ping failed - Status Code: {response.status_code}, Reason: {response.reason}")
+                logger.error(
+                    f"Listrr ping failed - Status Code: {response.status_code}, Reason: {response.reason}"
+                )
             return response.ok
         except Exception as e:
             logger.error(f"Listrr ping exception: {e}")
@@ -71,7 +75,9 @@ def run(self):
         elif length > 5:
             logger.info("Added %s items", length)
         if self.not_found_ids:
-            logger.debug("Failed to process %s items, skipping.", len(self.not_found_ids))
+            logger.debug(
+                "Failed to process %s items, skipping.", len(self.not_found_ids)
+            )
 
     def _get_items_from_Listrr(self, content_type, content_lists) -> MediaItemContainer:
         """Fetch unique IMDb IDs from Listrr for a given type and list of content."""
@@ -89,17 +95,17 @@ def _get_items_from_Listrr(self, content_type, content_lists) -> MediaItemContai
                     url = f"{self.url}/List/{content_type}/{list_id}/ReleaseDate/Descending/{page}"
                     response = get(url, additional_headers=self.headers).response
                     data = response.json()
-                    total_pages = data.get('pages', 1)
-                    for item in data.get('items', []):
-                        imdb_id = item.get('imDbId')
+                    total_pages = data.get("pages", 1)
+                    for item in data.get("items", []):
+                        imdb_id = item.get("imDbId")
                         if imdb_id:
                             unique_ids.add(imdb_id)
-                        elif content_type == "Movies" and item.get('tmDbId'):
-                            imdb_id = get_imdbid_from_tmdb(item['tmDbId'])
+                        elif content_type == "Movies" and item.get("tmDbId"):
+                            imdb_id = get_imdbid_from_tmdb(item["tmDbId"])
                             if imdb_id:
                                 unique_ids.add(imdb_id)
                         else:
-                            self.not_found_ids.append(item['id'])
+                            self.not_found_ids.append(item["id"])
                 except HTTPError as e:
                     if e.response.status_code in [400, 404, 429, 500]:
                         break

backend/program/content/mdblist.py

@@ -48,7 +48,9 @@ def run(self):
                 items = MediaItemContainer()
                 for list_id in self.settings.lists:
                     if list_id:
-                        items.extend(self._get_items_from_list(list_id, self.settings.api_key))
+                        items.extend(
+                            self._get_items_from_list(list_id, self.settings.api_key)
+                        )
                 added_items = self.process_items(items, "Mdblist")
                 if not added_items:
                     return

backend/program/content/overseerr.py

@@ -61,7 +61,9 @@ def run(self):
         elif length > 5:
             logger.info("Added %s items", length)
         if self.not_found_ids:
-            logger.debug("Failed to process %s items, skipping.", len(self.not_found_ids))
+            logger.debug(
+                "Failed to process %s items, skipping.", len(self.not_found_ids)
+            )
 
     def _get_items_from_overseerr(self, amount: int) -> MediaItemContainer:
         """Fetch media from overseerr"""
@@ -93,27 +95,34 @@ def get_imdb_id(self, overseerr_item) -> str:
         if f"{id_extension}{external_id}" in self.not_found_ids:
             return None
         response = get(
-            self.settings.url + f"/api/v1/{overseerr_item.mediaType}/{external_id}?language=en",
+            self.settings.url
+            + f"/api/v1/{overseerr_item.mediaType}/{external_id}?language=en",
             additional_headers=self.headers,
         )
         if not response.is_ok or not hasattr(response.data, "externalIds"):
-            logger.debug(f"Failed to fetch or no externalIds for {id_extension}{external_id}")
+            logger.debug(
+                f"Failed to fetch or no externalIds for {id_extension}{external_id}"
+            )
             return None
 
-        title = getattr(response.data, "title", None) or getattr(response.data, "originalName", None)
-        imdb_id = getattr(response.data.externalIds, 'imdbId', None)
+        title = getattr(response.data, "title", None) or getattr(
+            response.data, "originalName", None
+        )
+        imdb_id = getattr(response.data.externalIds, "imdbId", None)
         if imdb_id:
             return imdb_id
 
         # Try alternate IDs if IMDb ID is not available
         # alternate_ids = [('tvdbId', get_imdbid_from_tvdb), ('tmdbId', get_imdbid_from_tmdb)]
-        alternate_ids = [('tmdbId', get_imdbid_from_tmdb)]
+        alternate_ids = [("tmdbId", get_imdbid_from_tmdb)]
         for id_attr, fetcher in alternate_ids:
             external_id_value = getattr(response.data.externalIds, id_attr, None)
             if external_id_value:
                 new_imdb_id = fetcher(external_id_value)
                 if new_imdb_id:
-                    logger.debug(f"Found imdbId for {title} from {id_attr}: {external_id_value}")
+                    logger.debug(
+                        f"Found imdbId for {title} from {id_attr}: {external_id_value}"
+                    )
                     return new_imdb_id
 
         self.not_found_ids.append(f"{id_extension}{external_id}")

backend/program/content/plex_watchlist.py

@@ -34,9 +34,13 @@ def validate(self):
                 return True
             except HTTPError as e:
                 if e.response.status_code == 404:
-                    logger.warn("Plex RSS URL is Not Found. Please check your RSS URL in settings.")
+                    logger.warn(
+                        "Plex RSS URL is Not Found. Please check your RSS URL in settings."
+                    )
                 else:
-                    logger.warn(f"Plex RSS URL is not reachable (HTTP status code: {e.response.status_code}). Falling back to using user Watchlist.")
+                    logger.warn(
+                        f"Plex RSS URL is not reachable (HTTP status code: {e.response.status_code}). Falling back to using user Watchlist."
+                    )
                 return True
             except Exception as e:
                 logger.exception(f"Failed to validate Plex RSS URL: {e}")
@@ -64,8 +68,10 @@ def run(self):
         elif length > 5:
             logger.info("Added %s items", length)
         if self.not_found_ids:
-            logger.debug("Failed to process %s items, skipping.", len(self.not_found_ids))
-
+            logger.debug(
+                "Failed to process %s items, skipping.", len(self.not_found_ids)
+            )
+
     def _create_unique_list(self) -> MediaItemContainer:
         """Create a unique list of items from Plex RSS and Watchlist."""
         if not self.rss_enabled:
@@ -80,7 +86,9 @@ def _get_items_from_rss(self) -> list:
         try:
             response = get(self.settings.rss, timeout=60)
             if not response.is_ok:
-                logger.error(f"Failed to fetch Plex RSS feed: HTTP {response.status_code}")
+                logger.error(
+                    f"Failed to fetch Plex RSS feed: HTTP {response.status_code}"
+                )
                 return []
             imdb_ids = [
                 guid.split("//")[-1]
@@ -90,7 +98,9 @@ def _get_items_from_rss(self) -> list:
             ]
             return imdb_ids
         except Exception as e:
-            logger.error(f"An unexpected error occurred while fetching Plex RSS feed: {e}")
+            logger.error(
+                f"An unexpected error occurred while fetching Plex RSS feed: {e}"
+            )
             return []
 
     def _get_items_from_watchlist(self) -> list:
@@ -99,14 +109,27 @@ def _get_items_from_watchlist(self) -> list:
         url = f"https://metadata.provider.plex.tv/library/sections/watchlist/all?X-Plex-Token={self.token}&{filter_params}"
         response = get(url)
         if response.is_ok and hasattr(response.data, "MediaContainer"):
-            valid_items = filter(lambda item: hasattr(item, 'ratingKey') and item.ratingKey, response.data.MediaContainer.Metadata)
-            imdb_ids = list(filter(None, map(lambda item: self._ratingkey_to_imdbid(item.ratingKey), valid_items)))
+            valid_items = filter(
+                lambda item: hasattr(item, "ratingKey") and item.ratingKey,
+                response.data.MediaContainer.Metadata,
+            )
+            imdb_ids = list(
+                filter(
+                    None,
+                    map(
+                        lambda item: self._ratingkey_to_imdbid(item.ratingKey),
+                        valid_items,
+                    ),
+                )
+            )
             return imdb_ids
         return []
 
     def _ratingkey_to_imdbid(self, ratingKey: str) -> str:
         """Convert Plex rating key to IMDb ID"""
-        filter_params = "includeGuids=1&includeFields=guid,title,year&includeElements=Guid"
+        filter_params = (
+            "includeGuids=1&includeFields=guid,title,year&includeElements=Guid"
+        )
         url = f"https://metadata.provider.plex.tv/library/metadata/{ratingKey}?X-Plex-Token={self.token}&{filter_params}"
         response = get(url)
         if response.is_ok and hasattr(response.data, "MediaContainer"):