feat: update RTN to latest

pyproject.toml

@@ -20,7 +20,6 @@ uvicorn = {extras = ["standard"], version = "^0.30.6"}
 apscheduler = "^3.10.4"
 regex = "^2023.12.25"
 coverage = "^7.5.4"
-rank-torrent-name = "0.2.23"
 cachetools = "^5.3.3"
 loguru = "^0.7.2"
 rich = "^13.7.1"
@@ -34,6 +33,7 @@ alembic = "^1.13.2"
 psycopg2-binary = "^2.9.9"
 apprise = "^1.8.1"
 subliminal = "^2.2.1"
+rank-torrent-name = "^1.0.2"
 
 [tool.poetry.group.dev.dependencies]
 pyright = "^1.1.352"

src/program/downloaders/alldebrid.py

@@ -17,7 +17,7 @@
 from requests import ConnectTimeout
 from RTN.exceptions import GarbageTorrent
 from RTN.parser import parse
-from RTN.patterns import extract_episodes
+from RTN.extras import extract_episodes
 from utils.logger import logger
 from utils.ratelimiter import RateLimiter
 from utils.request import get, ping, post
@@ -266,13 +266,12 @@ def _is_wanted_movie(self, file: dict, item: Movie) -> bool:
         if not isinstance(file, dict) or file.get("s", 0) < min_size or file.get("s", 0) > max_size or splitext(file.get("n", "").lower())[1] not in WANTED_FORMATS:
             return False
 
-        with contextlib.suppress(GarbageTorrent, TypeError):
-            parsed_file = parse(file["n"], remove_trash=True)
-            if parsed_file and parsed_file.type == "movie":
-                item.set("folder", item.active_stream.get("name"))
-                item.set("alternative_folder", item.active_stream.get("alternative_name", None))
-                item.set("file", file["n"])
-                return True
+        parsed_file = parse(file["n"])
+        if parsed_file and parsed_file.type == "movie":
+            item.set("folder", item.active_stream.get("name"))
+            item.set("alternative_folder", item.active_stream.get("alternative_name", None))
+            item.set("file", file["n"])
+            return True
         return False
 
     def _is_wanted_episode(self, file: dict, item: Episode) -> bool:
@@ -289,13 +288,12 @@ def _is_wanted_episode(self, file: dict, item: Episode) -> bool:
 
         one_season = len(item.parent.parent.seasons) == 1
 
-        with contextlib.suppress(GarbageTorrent, TypeError):
-            parsed_file = parse(file["n"], remove_trash=True)
-            if parsed_file and item.number in parsed_file.episode and (item.parent.number in parsed_file.season or one_season):
-                item.set("folder", item.active_stream.get("name"))
-                item.set("alternative_folder", item.active_stream.get("alternative_name"))
-                item.set("file", file["n"])
-                return True
+        parsed_file = parse(file["n"])
+        if parsed_file and item.number in parsed_file.episodes and (item.parent.number in parsed_file.seasons or one_season):
+            item.set("folder", item.active_stream.get("name"))
+            item.set("alternative_folder", item.active_stream.get("alternative_name"))
+            item.set("file", file["n"])
+            return True
         return False
 
     def _is_wanted_season(self, files: list, item: Season) -> bool:
@@ -324,12 +322,11 @@ def _is_wanted_season(self, files: list, item: Season) -> bool:
         season_num = item.number
 
         for file in filenames:
-            with contextlib.suppress(GarbageTorrent, TypeError):
-                parsed_file = parse(file["n"], remove_trash=True)
-                if parsed_file and (season_num in parsed_file.season or one_season):
-                    for ep_num in parsed_file.episode:
-                        if ep_num in needed_episodes:
-                            matched_files[ep_num] = file["n"]
+            parsed_file = parse(file["n"])
+            if parsed_file and (season_num in parsed_file.seasons or one_season):
+                for ep_num in parsed_file.episodes:
+                    if ep_num in needed_episodes:
+                        matched_files[ep_num] = file["n"]
 
         if not matched_files:
             return False
@@ -375,15 +372,14 @@ def _is_wanted_show(self, files: list, item: Show) -> bool:
 
         matched_files = {}
         for file in filenames:
-            with contextlib.suppress(GarbageTorrent, TypeError):
-                parsed_file = parse(file["n"], remove_trash=True)
-                if parsed_file:
-                    for season_number, episodes in needed_episodes.items():
-                        if season_number in parsed_file.season:
-                            for episode_number in list(episodes):
-                                if episode_number in parsed_file.episode:
-                                    matched_files[(season_number, episode_number)] = file
-                                    episodes.remove(episode_number)
+            parsed_file = parse(file["n"])
+            if parsed_file:
+                for season_number, episodes in needed_episodes.items():
+                    if season_number in parsed_file.seasons:
+                        for episode_number in list(episodes):
+                            if episode_number in parsed_file.episodes:
+                                matched_files[(season_number, episode_number)] = file
+                                episodes.remove(episode_number)
 
         if not matched_files:
             return False

src/program/downloaders/realdebrid.py

@@ -2,15 +2,13 @@
 from requests import ConnectTimeout
 from program.media.item import MediaItem
 from utils.ratelimiter import RateLimiter
-from .shared import FileFinder
+from .shared import FileFinder, VIDEO_EXTENSIONS
 import utils.request as request
 from loguru import logger
 from program.settings.manager import settings_manager as settings
 
 BASE_URL = "https://api.real-debrid.com/rest/1.0"
-VIDEO_EXTENSIONS = [
-    'mp4', 'mkv', 'avi', 'mov', 'wmv', 'flv', 'm4v', 'webm', 'mpg', 'mpeg', 'm2ts', 'ts'
-]
+
 torrent_limiter = RateLimiter(1, 1)
 overall_limiter = RateLimiter(60, 60)
 

src/program/downloaders/shared.py

@@ -2,8 +2,18 @@
 
 from program.media.item import MediaItem
 from program.media.state import States
+from program.settings.manager import settings_manager
+
+
+DEFAULT_VIDEO_EXTENSIONS = ["mp4", "mkv", "avi"]
+ALLOWED_VIDEO_EXTENSIONS = ["mp4", "mkv", "avi", "mov", "wmv", "flv", "m4v", "webm", "mpg", "mpeg", "m2ts", "ts"]
+
+VIDEO_EXTENSIONS = settings_manager.settings.downloaders.video_extensions or DEFAULT_VIDEO_EXTENSIONS
+VIDEO_EXTENSIONS = [ext for ext in VIDEO_EXTENSIONS if ext in ALLOWED_VIDEO_EXTENSIONS]
+
+if not VIDEO_EXTENSIONS:
+    VIDEO_EXTENSIONS = DEFAULT_VIDEO_EXTENSIONS
 
-WANTED_FORMATS = {".mkv", ".mp4", ".avi"}
 
 class FileFinder:
     """
@@ -26,7 +36,7 @@ def get_cached_container(self, needed_media: dict[int, list[int]], break_pointer
     def filename_matches_show(self, filename):
         try:
             parsed_data = parse(filename)
-            return parsed_data.season[0], parsed_data.episode
+            return parsed_data.seasons[0], parsed_data.episodes
         except Exception:
             return None, None
 

src/program/downloaders/torbox.py

@@ -146,12 +146,12 @@ def find_required_files(self, item, container):
             and splitext(file["name"].lower())[1] in WANTED_FORMATS
         ]
 
+        parsed_file = parse(file["name"])
+
         if item.type == "movie":
             for file in files:
-                with contextlib.suppress(GarbageTorrent, TypeError):
-                    parsed_file = parse(file["name"], remove_trash=True)
-                    if parsed_file.type == "movie":
-                        return [file]
+                if parsed_file.type == "movie":
+                    return [file]
         if item.type == "show":
             # Create a dictionary to map seasons and episodes needed
             needed_episodes = {}
@@ -179,22 +179,17 @@ def find_required_files(self, item, container):
             # the season and episode within the show
             matched_files = []
             for file in files:
-                with contextlib.suppress(GarbageTorrent, TypeError):
-                    parsed_file = parse(file["name"], remove_trash=True)
-                    if (
-                        not parsed_file
-                        or not parsed_file.parsed_title
-                        or 0 in parsed_file.season
-                    ):
-                        continue
-                    # Check each season and episode to find a match
-                    for season_number, episodes in needed_episodes.items():
-                        if season_number in parsed_file.season:
-                            for episode_number in list(episodes):
-                                if episode_number in parsed_file.episode:
-                                    # Store the matched file for this episode
-                                    matched_files.append(file)
-                                    episodes.remove(episode_number)
+                if not parsed_file.seasons or parsed_file.seasons == [0]:
+                    continue
+
+                # Check each season and episode to find a match
+                for season_number, episodes in needed_episodes.items():
+                    if season_number in parsed_file.season:
+                        for episode_number in list(episodes):
+                            if episode_number in parsed_file.episode:
+                                # Store the matched file for this episode
+                                matched_files.append(file)
+                                episodes.remove(episode_number)
             if not matched_files:
                 return False
 
@@ -217,19 +212,13 @@ def find_required_files(self, item, container):
             for file in files:
                 if not file or not file.get("name"):
                     continue
-                with contextlib.suppress(GarbageTorrent, TypeError):
-                    parsed_file = parse(file["name"], remove_trash=True)
-                    if (
-                        not parsed_file
-                        or not parsed_file.episode
-                        or 0 in parsed_file.season
-                    ):
-                        continue
-                    # Check if the file's season matches the item's season or if there's only one season
-                    if season_num in parsed_file.season or one_season:
-                        for ep_num in parsed_file.episode:
-                            if ep_num in needed_episodes:
-                                matched_files.append(file)
+                if not parsed_file.seasons or parsed_file.seasons == [0]: # skip specials
+                    continue
+                # Check if the file's season matches the item's season or if there's only one season
+                if season_num in parsed_file.seasons or one_season:
+                    for ep_num in parsed_file.episodes:
+                        if ep_num in needed_episodes:
+                            matched_files.append(file)
             if not matched_files:
                 return False
 
@@ -240,13 +229,11 @@ def find_required_files(self, item, container):
             for file in files:
                 if not file or not file.get("name"):
                     continue
-                with contextlib.suppress(GarbageTorrent, TypeError):
-                    parsed_file = parse(file["name"], remove_trash=True)
-                    if (
-                        item.number in parsed_file.episode
-                        and item.parent.number in parsed_file.season
-                    ):
-                        return [file]
+                if (
+                    item.number in parsed_file.episodes
+                    and item.parent.number in parsed_file.seasons
+                ):
+                    return [file]
 
         return []
 

src/program/media/item.py

@@ -650,7 +650,7 @@ def get_file_episodes(self) -> List[int]:
         if not self.file or not isinstance(self.file, str):
             raise ValueError("The file attribute must be a non-empty string.")
         # return list of episodes
-        return parse(self.file).episode
+        return parse(self.file).episodes
 
     @property
     def log_string(self):

src/program/media/stream.py

@@ -1,10 +1,14 @@
+from typing import TYPE_CHECKING
 from RTN import Torrent
 from sqlalchemy import Index
 
 from program.db.db import db
 import sqlalchemy
 from sqlalchemy.orm import Mapped, mapped_column, relationship
-from loguru import logger
+
+if TYPE_CHECKING:
+    from program.media.item import MediaItem
+
 
 class StreamRelation(db.Model):
     __tablename__ = "StreamRelation"

src/program/media/subtitle.py

@@ -2,6 +2,10 @@
 from program.db.db import db
 from sqlalchemy import Integer, String, ForeignKey, Index
 from sqlalchemy.orm import Mapped, mapped_column, relationship
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from program.media.item import MediaItem
 
 
 class Subtitle(db.Model):

src/program/scrapers/__init__.py

@@ -114,7 +114,7 @@ def run_service(service, item,):
 
         if sorted_streams and (log and settings_manager.settings.debug):
             item_type = item.type.title()
-            top_results = sorted(sorted_streams.values(), key=lambda x: x.rank, reverse=True)[:10]
+            top_results = list(sorted_streams.values())[:10]
             for sorted_tor in top_results:
                 if isinstance(item, (Movie, Show)):
                     logger.debug(f"[{item_type}] Parsed '{sorted_tor.parsed_title}' with rank {sorted_tor.rank} and ratio {sorted_tor.lev_ratio:.2f}: '{sorted_tor.raw_title}'")