Merge pull request #41 from 31December99/0.x.x

Connect the igdb class to uploader class

common/clients/qbitt.py

@@ -32,7 +32,7 @@ def _check_connection(cls) -> typing.Optional[Client]:
 
             qb.login(username=config.QBIT_USER, password=config.QBIT_PASS)
             qb.torrents()
-            custom_console.bot_log(f"[QBITTORRENT]...... Online")
+            # custom_console.bot_log(f"[QBITTORRENT]...... Online")
             return qb
         except requests.exceptions.HTTPError:
             custom_console.bot_error_log(
@@ -83,21 +83,18 @@ def send_to_client(self):
             # Search for your torrent
             self.qbit(self.torrents)
 
-    def download(self, link: requests) -> typing.IO:
-
+    def download(self, tracker_torrent_url: requests) -> typing.IO:
         # Archive the torrent file if torrent_archive is set
         if self.torrent_archive:
-            full_path_origin = f"{self.torrent_path}.torrent"
             file_name = f"{os.path.basename(self.torrent_path)}.torrent"
             full_path_archive = os.path.join(self.torrent_archive, file_name)
-            os.replace(full_path_origin, full_path_archive)
         else:
             # Or save to the current path
             full_path_archive = f"{self.torrent_path}.torrent"
 
         # File archived
         with open(full_path_archive, "wb") as file:
-            file.write(link.content)
+            file.write(tracker_torrent_url.content)
 
         # Ready for seeding
         return open(full_path_archive, "rb")

common/config.py

@@ -3,11 +3,13 @@
 import os
 from common.custom_console import custom_console
 from pydantic_settings import BaseSettings
-from pydantic import Field, field_validator
+from pydantic import Field, field_validator, fields
 from dotenv import load_dotenv
 from urllib.parse import urlparse
 from pathlib import Path
 
+service_filename = "Unit3Dbot_service.env"
+
 
 def create_default_env_file(path):
     default_content = """
@@ -32,12 +34,15 @@ def create_default_env_file(path):
 # Number of screenshots we create
 NUMBER_OF_SCREENSHOTS=6
 
-# Level of compression for screenshot ( quality) ) 0 = Best quality
+# Level of compression for screenshot (quality) 0 = Best quality
 COMPRESS_SCSHOT=4
 
 # Path for each torrent file created
 TORRENT_ARCHIVE=
 
+# Torrent file comment (max 100 chars)
+TORRENT_COMMENT=
+
 # Preferred language. Discard videos with a language different from preferred_lang
 PREFERRED_LANG=
 
@@ -72,27 +77,24 @@ def create_default_env_file(path):
         f.write(default_content.strip())
 
 
-# Define the path for the configuration file ( Windows/Linux)
-if os.name == "nt":
-    default_env_path = Path(os.getenv("APPDATA", ".")) / "service.env"
-    torrent_archive_path = Path(os.getenv("APPDATA", ".")) / "torrent_archive"
-else:
-    default_env_path = Path.home() / "service.env"
+# Define the path for the configuration file
+if os.name == "nt":  # If on Windows
+    default_env_path = Path(os.getenv("LOCALAPPDATA", ".")) / f"{service_filename}"
+    torrent_archive_path = Path(os.getenv("LOCALAPPDATA", ".")) / "torrent_archive"
+else:  # If on Linux/macOS
+    default_env_path = Path.home() / f"{service_filename}"
     torrent_archive_path = Path.home() / "torrent_archive"
 
-# print the configuration path
-custom_console.bot_log(f"Default configuration path... {default_env_path}")
+custom_console.bot_question_log(f"Default configuration path: {default_env_path}\n")
 
 # Create the configuration file if it does not exist
 if not default_env_path.exists():
-    custom_console.bot_log(f"Create default configuration file... {default_env_path}")
+    print(f"Create default configuration file: {default_env_path}")
     create_default_env_file(default_env_path)
 
 # Create the directory for the torrent archive if it does not exist
 if not torrent_archive_path.exists():
-    custom_console.bot_log(
-        f"Create default torrent archive path... {torrent_archive_path}"
-    )
+    print(f"Create default torrent archive path: {torrent_archive_path}")
     os.makedirs(torrent_archive_path, exist_ok=True)
 
 # Load environment variables
@@ -102,7 +104,7 @@ def create_default_env_file(path):
 class Config(BaseSettings):
     # TRACKER
     ITT_APIKEY: str | None = Field(default=None, env="ITT_APIKEY")
-    ITT_URL: str | None = Field(default=None, env="ITT_URL")
+    ITT_URL: str = Field(default="https://itatorrents.xyz", env="ITT_URL")
 
     # EXTERNAL SERVICE
     TMDB_APIKEY: str | None = Field(default=None, env="TMDB_APIKEY")
@@ -120,14 +122,15 @@ class Config(BaseSettings):
     # TORRENT CLIENT
     QBIT_USER: str | None = Field(default=None, env="QBIT_USER")
     QBIT_PASS: str | None = Field(default=None, env="QBIT_PASS")
-    QBIT_URL: str = Field(default="http://localhost:8080", env="QBIT_URL")
+    QBIT_URL: str = Field(default="http://127.0.0.1", env="QBIT_URL")
     QBIT_PORT: str | None = Field(default="8080", env="QBIT_PORT")
 
     # USER PREFERENCES
-    DUPLICATE_ON: bool = Field(default=False, env="DUPLICATE_ON")
+    DUPLICATE_ON: str = Field(default=False, env="DUPLICATE_ON")
     NUMBER_OF_SCREENSHOTS: int = Field(default=6, env="NUMBER_OF_SCREENSHOTS")
     COMPRESS_SCSHOT: int = Field(default=4, env="COMPRESS_SCSHOT")
     TORRENT_ARCHIVE: str | None = Field(default=None, env="TORRENT_ARCHIVE")
+    TORRENT_COMMENT: str | None = Field(default=None, env="TORRENT_COMMENT")
     PREFERRED_LANG: str | None = Field(default=None, env="PREFERRED_LANG")
     SIZE_TH: int = Field(default=100, env="SIZE_TH")
     FTPX_LOCAL_PATH: str | None = Field(default=None, env="FTPX_LOCAL_PATH")
@@ -138,19 +141,35 @@ def __init__(self, **values: any):
         super().__init__(**values)
 
     @staticmethod
-    def validate_url(value: str | None, default_value: str) -> str:
+    def validate_url(value: str | None, field: str, fields: dict) -> str:
         if not value:
-            return default_value
+            return fields[field].default
         parsed_url = urlparse(value)
         if not (parsed_url.scheme and parsed_url.netloc):
-            raise ValueError(f"{value} Invalid Url")
+            custom_console.bot_error_log(
+                f"{value} is an invalid URL. Using default: {fields[field].default}"
+            )
+            return fields[field].default
         return value
 
+    @staticmethod
+    def validate_boolean(value: any, field: str, fields: dict) -> bool:
+        if isinstance(value, str):
+            lowered_value = value.lower()
+            if lowered_value in ["true", "1", "yes"]:
+                return True
+            elif lowered_value in ["false", "0", "no"]:
+                return False
+        custom_console.bot_error_log(
+            f"{value} is not a valid boolean for {field}. Using default: {fields[field].default}"
+        )
+        return fields[field].default
+
     @field_validator("ITT_URL")
     def validate_itt_url(cls, value):
         if not value:
             custom_console.bot_error_log("No ITT_URL provided")
-        return cls.validate_url(value, cls.__fields__["ITT_URL"].default)
+        return cls.validate_url(value, "ITT_URL", cls.__fields__)
 
     @field_validator("ITT_APIKEY")
     def validate_itt_apikey(cls, value):
@@ -162,13 +181,13 @@ def validate_itt_apikey(cls, value):
     def validate_qbit_url(cls, value):
         if not value:
             custom_console.bot_error_log("No QBIT_URL provided")
-        return cls.validate_url(value, cls.__fields__["QBIT_URL"].default)
+        return cls.validate_url(value, "QBIT_URL", cls.__fields__)
 
     @field_validator("PW_URL")
     def validate_pw_url(cls, value):
         # if not value:
         # custom_console.bot_question_log("[Optional] No PW_URL provided\n")
-        return cls.validate_url(value, cls.__fields__["PW_URL"].default)
+        return cls.validate_url(value, "PW_URL", cls.__fields__)
 
     @field_validator("PW_API_KEY")
     def validate_pw_apikey(cls, value):
@@ -214,10 +233,7 @@ def validate_qbit_port(cls, value):
 
     @field_validator("DUPLICATE_ON")
     def validate_duplicate_on(cls, value):
-        if not isinstance(value, bool):
-            custom_console.bot_error_log("DUPLICATE_ON should be a boolean")
-            return cls.__fields__["DUPLICATE_ON"].default
-        return value
+        return cls.validate_boolean(value, "DUPLICATE_ON", cls.__fields__)
 
     @field_validator("NUMBER_OF_SCREENSHOTS")
     def validate_n_screenshot(cls, value):
@@ -237,6 +253,12 @@ def validate_torrent_archive(cls, value):
             return cls.__fields__["TORRENT_ARCHIVE"].default
         return value
 
+    @field_validator("TORRENT_COMMENT")
+    def validate_torrent_comment(cls, value):
+        if not isinstance(value, str):
+            return cls.__fields__["TORRENT_COMMENT"].default
+        return value
+
     @field_validator("PREFERRED_LANG")
     def validate_preferred_lang(cls, value):
         if not isinstance(value, str):

common/custom_console.py

@@ -6,6 +6,7 @@
 from rich.panel import Panel
 from rich.text import Text
 from rich.table import Table
+from common.trackers.itt import itt_data
 
 from decouple import Config, RepositoryEnv, UndefinedValueError
 
@@ -100,19 +101,37 @@ def bot_error_log(self, message: str):
     def bot_question_log(self, message: str):
         console.print(message, end="", style=self.question_msg_color)
 
-    def bot_process_table_log(self, content: list):
+    @staticmethod
+    def get_key_by_value(tracker_data, category, value):
+        if category in tracker_data:
+            if isinstance(tracker_data[category], dict):
+                for k, v in tracker_data[category].items():
+                    if v == value:
+                        return k
+
+    @staticmethod
+    def bot_process_table_log(content: list):
 
         table = Table(
-            title="Selected Files List",
+            title="Here is your files list" if content else "There are no files here",
             border_style="bold blue",
             header_style="red blue",
         )
+
         table.add_column("Torrent Pack", style="dim")
+        table.add_column("Media", justify="left", style="bold green")
         table.add_column("Path", justify="left", style="bold green")
 
         for item in content:
             pack = "Yes" if item.torrent_pack else "No"
-            table.add_row(pack, item.torrent_path)
+            category_name = CustomConsole.get_key_by_value(itt_data, "CATEGORY", item.category)
+            if not category_name:
+                category_name = ''
+            table.add_row(
+                pack,
+                category_name,
+                item.torrent_path,
+            )
 
         console.print(Align.center(table))
 

common/external_services/Pw/core/models/indexers.py

@@ -52,7 +52,7 @@ class Field:
 class Indexer:
     added: str = ''
     appProfileId: int = 0
-    capabilities: Capability = Capability()
+    capabilities: Capability = field(default_factory=Capability)
     configContract: str = ''
     definitionName: str = ''
     description: str = ''

common/external_services/Pw/core/models/search.py

@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 
 @dataclass
@@ -9,25 +9,26 @@ class Search:
     Get a result from the endpoint /search?query
 
     """
+
     age: int = 0
     ageHours: float = 0.0
     ageMinutes: float = 0.0
-    categories: list[dict[str, any]] = None
-    downloadUrl: str = ""
-    fileName: str = ""
-    guid: str = ""
+    categories: list[dict[str, any]] = field(default_factory=list)
+    downloadUrl: None | str = None
+    fileName: None | str = None
+    guid: None | str = None
     imdbId: int = 0
-    indexer: str = ""
-    indexerFlags: list[str] = None
+    indexer: None | str = None
+    indexerFlags: list[str] = field(default_factory=list)
     indexerId: int = 0
-    infoUrl: str = ""
+    infoUrl: None | str = None
     leechers: int = 0
     protocol: str = "torrent"
-    publishDate: str = ""
+    publishDate: None | str = None
     seeders: int = 0
     size: int = 0
-    sortTitle: str = ""
-    title: str = ""
+    sortTitle: None | str = None
+    title: None | str = None
     tmdbId: int = 0
     tvMazeId: int = 0
     tvdbId: int = 0

common/external_services/Pw/core/models/torrent_client_config.py

@@ -28,15 +28,15 @@ class Field:
 @dataclass
 class TorrentClientConfig:
     categories: list[str] = field(default_factory=list)
-    configContract: str = ""
+    configContract: str | None = None
     enable: bool = True
     fields: list[Field] = field(default_factory=list)
     id: int = 0
-    implementation: str = ""
-    implementationName: str = ""
-    infoLink: str = ""
-    name: str = ""
+    implementation: str | None = None
+    implementationName: str | None = None
+    infoLink: str | None = None
+    name: str | None = None
     priority: int = 0
-    protocol: str = ""
+    protocol: str | None = None
     supportsCategories: bool = False
     tags: list[str] = field(default_factory=list)