diff --git a/.vscode/settings.json b/.vscode/settings.json index 7f04cc32..c24dd4cd 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -133,6 +133,7 @@ "SUPPRESSHELP", "tebibytes", "testpaths", + "threadsafe", "tiktok", "timespec", "tmpfilename", @@ -170,5 +171,10 @@ "**/.venv": true, }, "eslint.format.enable": true, - "eslint.ignoreUntitled": true + "eslint.ignoreUntitled": true, + "python.testing.pytestArgs": [ + "app/tests" + ], + "python.testing.unittestEnabled": true, + "python.testing.pytestEnabled": true } diff --git a/app/library/Archiver.py b/app/library/Archiver.py index 977db85f..6b17f323 100644 --- a/app/library/Archiver.py +++ b/app/library/Archiver.py @@ -40,19 +40,15 @@ class Archiver(metaclass=ThreadSafe): """ def __init__(self) -> None: - if getattr(self, "_initialized", False): - return - self._cache: dict[str, _Entry] = {} self._locks: dict[str, threading.RLock] = {} self._global_lock = threading.RLock() self._stats_check: bool = True self._stats_ttl: float = 0.2 - self._initialized = True - @classmethod - def get_instance(cls) -> "Archiver": - return cls() + @staticmethod + def get_instance() -> "Archiver": + return Archiver() def _key(self, file: str | Path) -> str: """ diff --git a/app/library/BackgroundWorker.py b/app/library/BackgroundWorker.py index d7fa7139..48add1d3 100644 --- a/app/library/BackgroundWorker.py +++ b/app/library/BackgroundWorker.py @@ -22,15 +22,17 @@ class BackgroundWorker(metaclass=Singleton): It is designed to run in a separate thread and uses asyncio to handle asynchronous tasks. """ - _instance = None - """The instance of the Notification class.""" - - thread: threading.Thread - """The thread that runs the background worker.""" - def __init__(self): - self.queue = Queue() + self.queue: Queue = Queue() + "The queue to hold the tasks." self.running = True + "Whether the background worker is running or not." + self.thread: threading.Thread = None + "The thread that runs the background worker." + + @staticmethod + def get_instance() -> "BackgroundWorker": + return BackgroundWorker() def attach(self, app: web.Application): app.on_shutdown.append(self.on_shutdown) @@ -49,13 +51,6 @@ class BackgroundWorker(metaclass=Singleton): except Exception as e: LOG.error(f"Failed to shut down background worker: {e}") - @staticmethod - def get_instance() -> "BackgroundWorker": - if BackgroundWorker._instance is None: - BackgroundWorker._instance = BackgroundWorker() - - return BackgroundWorker._instance - def _run(self): asyncio.set_event_loop(asyncio.new_event_loop()) loop = asyncio.get_event_loop() diff --git a/app/library/DataStore.py b/app/library/DataStore.py index 55ff4bfd..a306adf6 100644 --- a/app/library/DataStore.py +++ b/app/library/DataStore.py @@ -11,12 +11,12 @@ from .Download import Download from .ItemDTO import ItemDTO from .Utils import init_class -LOG = logging.getLogger("datastore") +LOG: logging.Logger = logging.getLogger("datastore") class StoreType(str, Enum): - HISTORY = "done" - QUEUE = "queue" + HISTORY: str = "done" + QUEUE: str = "queue" @classmethod def all(cls) -> list[str]: @@ -53,19 +53,13 @@ class DataStore: Persistent queue. """ - _type: StoreType = None - """Type of the store, e.g., DONE, QUEUE, PENDING.""" - - _dict: OrderedDict[str, Download] = None - """Ordered dictionary to store Download objects.""" - - _connection: Connection - """SQLite connection to the database.""" - def __init__(self, type: StoreType, connection: Connection): - self._dict = OrderedDict() - self._type = type - self._connection = connection + self._dict: OrderedDict[str, Download] = OrderedDict() + "The dictionary of items." + self._type: StoreType = type + "The type of the datastore." + self._connection: Connection = connection + "The database connection." def load(self) -> None: for id, item in self.saved_items(): @@ -136,7 +130,7 @@ class DataStore: ) for row in cursor: - rowDate = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S") # noqa: DTZ007 + rowDate: datetime = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S") # noqa: DTZ007 data: dict = json.loads(row["data"]) data.pop("_id", None) item: ItemDTO = init_class(ItemDTO, data) diff --git a/app/library/Download.py b/app/library/Download.py index 00acd227..b18ad0d1 100644 --- a/app/library/Download.py +++ b/app/library/Download.py @@ -47,22 +47,8 @@ class Download: Download task. """ - id: str = None - download_dir: str = None - temp_dir: str = None - temp_disabled: bool = False - "Disable the temporary files feature." - template: str = None - template_chapter: str = None - info: ItemDTO = None - debug: bool = False temp_path: str = None - cancelled: bool = False - is_live: bool = False - info_dict: dict = None - "yt-dlp metadata dict." update_task = None - cancel_in_progress: bool = False final_update = False @@ -86,12 +72,6 @@ class Download: ) "Fields to be extracted from yt-dlp progress hook." - temp_keep: bool = False - "Keep temp folder after download." - - logs: list = [] - "Logs from yt-dlp." - def __init__(self, info: ItemDTO, info_dict: dict = None, logs: list | None = None): """ Initialize download task. @@ -104,29 +84,52 @@ class Download: """ config: Config = Config.get_instance() - self.download_dir = info.download_dir - self.temp_dir = info.temp_dir - self.template = info.template - self.template_chapter = info.template_chapter + self.download_dir: str = info.download_dir + "Download directory." + self.temp_dir: str | None = info.temp_dir + "Temporary directory." + self.template: str | None = info.template + "Filename template." + self.template_chapter: str | None = info.template_chapter + "Chapter filename template." self.download_info_expires = int(config.download_info_expires) - self.info = info - self.id = info._id + "Time in seconds before the download info is considered expired." + self.info: ItemDTO = info + "ItemDTO object." + self.id: str = info._id + "Download ID." self.debug = bool(config.debug) + "Debug mode." self.debug_ytdl = bool(config.ytdlp_debug) + "Debug mode for yt-dlp." self.cancelled = False + "Download cancelled." self.tmpfilename = None + "Temporary filename." self.status_queue = None + "Status queue." self.proc = None - self._notify = EventBus.get_instance() + "yt-dlp process." + self._notify: EventBus = EventBus.get_instance() + "Event bus instance." self.max_workers = int(config.max_workers) + "Maximum number of concurrent downloads." self.temp_keep = bool(config.temp_keep) + "Keep temp folder after download." self.temp_disabled = bool(config.temp_disabled) - self.is_live = bool(info.is_live) or info.live_in is not None - self.info_dict = info_dict + "Disable the temporary files feature." + self.is_live: bool = bool(info.is_live) or info.live_in is not None + "Is the download a live stream." + self.info_dict: dict = info_dict + "yt-dlp metadata dict." self.logger: logging.Logger = logging.getLogger(f"Download.{info.id if info.id else info._id}") + "Logger for the download task." self.started_time = 0 + "Time when the download started." self.queue_time: datetime = datetime.now(tz=UTC) + "Time when the download was queued." self.logs = logs if logs else [] + "Logs from yt-dlp." def _progress_hook(self, data: dict): if self.debug: diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index ee86c939..0aac9c68 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -47,44 +47,29 @@ class DownloadQueue(metaclass=Singleton): DownloadQueue class is a singleton class that manages the download queue and the download history. """ - paused: asyncio.Event - """Event to pause the download queue.""" - - event: asyncio.Event - """Event to signal the download queue to start downloading.""" - - _active: dict[str, Download] = {} - """Dictionary of active downloads.""" - - _instance = None - """Instance of the DownloadQueue.""" - - queue: DataStore - """DataStore for the download queue.""" - - done: DataStore - """DataStore for the completed downloads.""" - - workers: asyncio.Semaphore - """Semaphore to limit the number of concurrent downloads.""" - - processors: asyncio.Semaphore - """Semaphore to limit the number of concurrent processors.""" - def __init__(self, connection: Connection, config: Config | None = None): - DownloadQueue._instance = self - - self.config = config or Config.get_instance() - self._notify = EventBus.get_instance() + self.config: Config = config or Config.get_instance() + "Configuration instance." + self._notify: EventBus = EventBus.get_instance() + "Event bus instance." self.done = DataStore(type=StoreType.HISTORY, connection=connection) + "DataStore for the completed downloads." self.queue = DataStore(type=StoreType.QUEUE, connection=connection) + "DataStore for the download queue." + self.workers = asyncio.Semaphore(self.config.max_workers) + "Semaphore to limit the number of concurrent downloads." + self.processors = asyncio.Semaphore(self.config.playlist_items_concurrency) + "Semaphore to limit the number of concurrent processors." + self.paused = asyncio.Event() + "Event to pause the download queue." + self.event = asyncio.Event() + "Event to signal the download queue to start downloading." + self._active: dict[str, Download] = {} + """Dictionary of active downloads.""" + self.done.load() self.queue.load() - self.paused = asyncio.Event() self.paused.set() - self.event = asyncio.Event() - self.workers = asyncio.Semaphore(self.config.max_workers) - self.processors = asyncio.Semaphore(self.config.playlist_items_concurrency) @staticmethod def get_instance() -> "DownloadQueue": @@ -95,10 +80,7 @@ class DownloadQueue(metaclass=Singleton): DownloadQueue: The instance of the DownloadQueue """ - if not DownloadQueue._instance: - DownloadQueue._instance = DownloadQueue() - - return DownloadQueue._instance + return DownloadQueue() def attach(self, _: web.Application) -> None: """ @@ -1205,6 +1187,5 @@ class DownloadQueue(metaclass=Singleton): return if exc := task.exception(): - task_name: str = task.get_name() if task.get_name() else "unknown_task" LOG.error(f"Unhandled exception in background task '{task_name}': {exc!s}. {traceback.format_exc()}") diff --git a/app/library/Events.py b/app/library/Events.py index 2f2f2571..e9973182 100644 --- a/app/library/Events.py +++ b/app/library/Events.py @@ -194,14 +194,13 @@ class Event: class EventListener: - name: str - is_coroutine: bool = False - call_back: callable - def __init__(self, name: str, callback: callable): - self.name = name - self.call_back = callback - self.is_coroutine = asyncio.iscoroutinefunction(callback) + self.name: str = name + "The name of the listener." + self.call_back: callable = callback + "The callback function to call when the event is emitted." + self.is_coroutine: bool = asyncio.iscoroutinefunction(callback) + "Whether the callback is a coroutine function or not." async def handle(self, event: Event, **kwargs): if self.is_coroutine: @@ -215,20 +214,15 @@ class EventBus(metaclass=Singleton): This class is used to subscribe to and emit events to the registered listeners. """ - _instance = None - """the instance of the EventsSubscriber""" - - _listeners: dict[str, list[str, EventListener]] = {} - """The listeners for the events.""" - - debug: bool = False - """Whether to log debug messages or not.""" - - _offload: BackgroundWorker = None - """The background worker to offload tasks to.""" - def __init__(self): - EventBus._instance = self + self._listeners: dict[str, list[str, EventListener]] = {} + "The listeners for the events." + + self.debug: bool = False + "Whether to log debug messages or not." + + self._offload: BackgroundWorker = None + "The background worker to offload tasks to." @staticmethod def get_instance() -> "EventBus": @@ -239,10 +233,7 @@ class EventBus(metaclass=Singleton): EventsSubscriber: The instance of the EventsSubscriber """ - if not EventBus._instance: - EventBus._instance = EventBus() - - return EventBus._instance + return EventBus() def subscribe(self, event: str | list | tuple, callback: Awaitable, name: str | None = None) -> "EventBus": """ diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index 5c497633..543ccf2a 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -57,7 +57,7 @@ class HttpSocket: async def event_handler(e: Event, _, **kwargs): await self.sio.emit(event=e.event, data=encoder.encode(e), **kwargs) - services = Services.get_instance() + services: Services = Services.get_instance() services.add_all( { k: v diff --git a/app/library/LogWrapper.py b/app/library/LogWrapper.py index 2f19a9fb..149242a9 100644 --- a/app/library/LogWrapper.py +++ b/app/library/LogWrapper.py @@ -59,11 +59,9 @@ class LogWrapper: """ - targets: list[LogTarget] = [] - """A list of dictionaries where each dictionary represents a logging target with its level and type.""" - def __init__(self): self.targets: list[LogTarget] = [] + """A list of dictionaries where each dictionary represents a logging target with its level and type.""" def add_target(self, target: logging.Logger | Callable, level: int = logging.DEBUG, name: str | None = None): """ diff --git a/app/library/Notifications.py b/app/library/Notifications.py index 5ac676a1..e4b777e4 100644 --- a/app/library/Notifications.py +++ b/app/library/Notifications.py @@ -130,12 +130,6 @@ class NotificationEvents: class Notification(metaclass=Singleton): - _targets: list[Target] = [] - """Notification targets to send events to.""" - - _instance = None - """The instance of the Notification class.""" - def __init__( self, file: str | None = None, @@ -144,15 +138,23 @@ class Notification(metaclass=Singleton): config: Config | None = None, background_worker: BackgroundWorker | None = None, ): - Notification._instance = self + self._targets: list[Target] = [] + "Notification targets to send events to." + config: Config = config or Config.get_instance() self._debug: bool = config.debug + "Debug mode." self._file: Path = Path(file) if file else Path(config.config_path).joinpath("notifications.json") + "File to store notification targets." self._client: httpx.AsyncClient = client or httpx.AsyncClient() + "HTTP client to send requests." self._encoder: Encoder = encoder or Encoder() + "Encoder to encode data." self._version: str = config.app_version + "Application version." self._offload: BackgroundWorker = background_worker or BackgroundWorker.get_instance() + "Background worker to offload tasks to." if self._file.exists() and "600" != self._file.stat().st_mode: try: @@ -161,11 +163,19 @@ class Notification(metaclass=Singleton): pass @staticmethod - def get_instance() -> "Notification": - if Notification._instance is None: - Notification._instance = Notification() - - return Notification._instance + def get_instance( + file: str | None = None, + client: httpx.AsyncClient | None = None, + encoder: Encoder | None = None, + config: Config | None = None, + background_worker: BackgroundWorker | None = None, + ) -> "Notification": + """ + Get the instance of the class. + """ + return Notification( + file=file, client=client, encoder=encoder, config=config, background_worker=background_worker + ) def attach(self, _: web.Application): """ diff --git a/app/library/PackageInstaller.py b/app/library/PackageInstaller.py index c43ef67f..18c35fe5 100644 --- a/app/library/PackageInstaller.py +++ b/app/library/PackageInstaller.py @@ -37,9 +37,10 @@ class Packages: class PackageInstaller: - user_site: Path | None = None - def __init__(self, pkg_path: Path | None = None): + self.user_site: Path | None = None + "Where to install user packages." + if pkg_path: self.user_site = pkg_path diff --git a/app/library/Playlist.py b/app/library/Playlist.py index af648ebb..9fed2312 100644 --- a/app/library/Playlist.py +++ b/app/library/Playlist.py @@ -6,11 +6,11 @@ from .Utils import StreamingError, get_file_sidecar class Playlist: - _url: str = None - def __init__(self, download_path: Path, url: str): self.url: str = url + "The base URL for the playlist." self.download_path: Path = download_path + "The path where files are downloaded." async def make(self, file: Path) -> str: ref: str = Path(str(file.relative_to(self.download_path)).strip("/")) diff --git a/app/library/Presets.py b/app/library/Presets.py index 6aa1e99c..c6a83011 100644 --- a/app/library/Presets.py +++ b/app/library/Presets.py @@ -105,24 +105,20 @@ class Presets(metaclass=Singleton): This class is used to manage the presets. """ - _items: list[Preset] = [] - """The list of presets.""" - - _instance = None - """The instance of the class.""" - - _config: Config = None - """The config instance.""" - - _default: list[Preset] = [] - """The list of default presets.""" - def __init__(self, file: str | Path | None = None, config: Config | None = None): - Presets._instance = self + self._items: list[Preset] = [] + "The list of presets." + + self._config: Config = None + "The config instance." + + self._default: list[Preset] = [] + "The list of default presets." self._config = config or Config.get_instance() self._file: Path = Path(file) if file else Path(self._config.config_path).joinpath("presets.json") + "The path to the presets file." if self._file.exists() and "600" != self._file.stat().st_mode: try: @@ -139,18 +135,19 @@ class Presets(metaclass=Singleton): continue @staticmethod - def get_instance() -> "Presets": + def get_instance(file: str | Path | None = None, config: Config | None = None) -> "Presets": """ Get the instance of the class. + Args: + file (str|Path|None): The path to the presets file. + config (Config|None): The config instance. + Returns: Presets: The instance of the class """ - if not Presets._instance: - Presets._instance = Presets() - - return Presets._instance + return Presets(file=file, config=config) async def on_shutdown(self, _: web.Application): pass diff --git a/app/library/Scheduler.py b/app/library/Scheduler.py index 969eadc9..b3230454 100644 --- a/app/library/Scheduler.py +++ b/app/library/Scheduler.py @@ -15,25 +15,15 @@ class Scheduler(metaclass=Singleton): This class is used to manage the schedule. """ - _jobs: dict[str, Cron] = {} - """The scheduled jobs.""" - - _instance = None - """The instance of the class.""" - def __init__(self, loop: asyncio.AbstractEventLoop | None = None): - Scheduler._instance = self + self._jobs: dict[str, Cron] = {} + "The scheduled jobs." self._loop = loop or asyncio.get_event_loop() - - async def event_handler(data, _): - if data and data.data: - self.add(**data.data) - - EventBus.get_instance().subscribe(Events.SCHEDULE_ADD, event_handler, f"{__class__.__name__}.add") + "The event loop to use." @staticmethod - def get_instance() -> "Scheduler": + def get_instance(loop: asyncio.AbstractEventLoop | None = None) -> "Scheduler": """ Get the instance of the class. @@ -41,9 +31,7 @@ class Scheduler(metaclass=Singleton): Scheduler: The instance of the class """ - if not Scheduler._instance: - Scheduler._instance = Scheduler() - return Scheduler._instance + return Scheduler(loop=loop) async def on_shutdown(self, _: web.Application): """ @@ -70,6 +58,12 @@ class Scheduler(metaclass=Singleton): def attach(self, app: web.Application): app.on_shutdown.append(self.on_shutdown) + async def event_handler(data, _): + if data and data.data: + self.add(**data.data) + + EventBus.get_instance().subscribe(Events.SCHEDULE_ADD, event_handler, f"{__class__.__name__}.add") + def get_all(self) -> dict[str, Cron]: """Return the jobs.""" return self._jobs diff --git a/app/library/Segments.py b/app/library/Segments.py index 963ffda8..ef1edee7 100644 --- a/app/library/Segments.py +++ b/app/library/Segments.py @@ -11,22 +11,32 @@ from aiohttp import web from .config import Config from .ffprobe import ffprobe -LOG = logging.getLogger("player.segments") +LOG: logging.Logger = logging.getLogger("player.segments") class Segments: def __init__(self, download_path: str, index: int, duration: float, vconvert: bool, aconvert: bool): - config = Config.get_instance() - self.download_path = download_path + config: Config = Config.get_instance() + self.download_path: str = download_path + "The path where files are downloaded." self.index = int(index) + "The index of the segment." self.duration = float(duration) + "The duration of the segment." self.vconvert = bool(vconvert) + "Whether to convert video." self.aconvert = bool(aconvert) - self.vcodec = config.streamer_vcodec - self.acodec = config.streamer_acodec + "Whether to convert audio." + self.vcodec: str = config.streamer_vcodec + "The video codec to use." + self.acodec: str = config.streamer_acodec + "The audio codec to use." + # sadly due to unforeseen circumstances, we have to convert the video for now. self.vconvert = True + "Whether to convert video." self.aconvert = True + "Whether to convert audio." async def build_ffmpeg_args(self, file: Path) -> list[str]: try: diff --git a/app/library/Services.py b/app/library/Services.py index bf9d93cd..894c66aa 100644 --- a/app/library/Services.py +++ b/app/library/Services.py @@ -9,42 +9,37 @@ LOG: logging.Logger = logging.getLogger(__name__) class Services(metaclass=Singleton): - _dct: dict[str, T] = {} - - _instance = None - """The instance of the class.""" + def __init__(self): + self._services: dict[str, T] = {} @staticmethod def get_instance() -> "Services": - if Services._instance is None: - Services._instance = Services() - - return Services._instance + return Services() def add(self, name: str, service: T): - self._dct[name] = service + self._services[name] = service def add_all(self, services: dict[str, T]): for name, service in services.items(): self.add(name, service) def get(self, name: str) -> T | None: - return self._dct.get(name) + return self._services.get(name) def has(self, name: str) -> bool: - return name in self._dct + return name in self._services def remove(self, name: str): - if name not in self._dct: + if name not in self._services: return - self._dct.pop(name, None) + self._services.pop(name, None) def clear(self): - self._dct.clear() + self._services.clear() def get_all(self) -> dict[str, T]: - return self._dct.copy() + return self._services.copy() async def handle_async(self, handler: callable, **kwargs) -> Any: context = {**self.get_all(), **kwargs} diff --git a/app/library/Singleton.py b/app/library/Singleton.py index 4c9a9641..70c8631a 100644 --- a/app/library/Singleton.py +++ b/app/library/Singleton.py @@ -8,6 +8,7 @@ class Singleton(type): """ _instances: dict[type, Any] = {} + "The singleton instances." def __call__(cls, *args: Any, **kwargs: Any) -> Any: if cls not in cls._instances: @@ -30,7 +31,10 @@ class ThreadSafe(type): """ _instances: dict[type, Any] = {} + "The singleton instances." + _lock = threading.Lock() + "A lock to ensure thread-safe singleton creation." def __call__(cls, *args: Any, **kwargs: Any) -> Any: with cls._lock: diff --git a/app/library/Tasks.py b/app/library/Tasks.py index b440be26..26b50847 100644 --- a/app/library/Tasks.py +++ b/app/library/Tasks.py @@ -132,12 +132,6 @@ class Tasks(metaclass=Singleton): This class is used to manage the tasks. """ - _tasks: list[Task] = [] - """The tasks.""" - - _instance = None - """The instance of the Tasks.""" - def __init__( self, file: str | None = None, @@ -146,19 +140,28 @@ class Tasks(metaclass=Singleton): encoder: Encoder | None = None, scheduler: Scheduler | None = None, ): - Tasks._instance = self - + self._tasks: list[Task] = [] + "The tasks." config = config or Config.get_instance() self._debug: bool = config.debug + "Debug mode." self._default_preset: str = config.default_preset + "The default preset." self._file: Path = Path(file) if file else Path(config.config_path).joinpath("tasks.json") + "The tasks file." self._encoder: Encoder = encoder or Encoder() + "The JSON encoder." self._loop: asyncio.AbstractEventLoop = loop or asyncio.get_event_loop() + "The event loop." self._scheduler: Scheduler = scheduler or Scheduler.get_instance() + "The scheduler." self._notify: EventBus = EventBus.get_instance() + "The event bus." self._task_handler = HandleTask(self._scheduler, self, config) + "The task handler." self._downloadQueue = DownloadQueue.get_instance() + "The download queue." if self._file.exists() and "600" != self._file.stat().st_mode: try: @@ -167,7 +170,13 @@ class Tasks(metaclass=Singleton): pass @staticmethod - def get_instance() -> "Tasks": + def get_instance( + file: str | None = None, + loop: asyncio.AbstractEventLoop | None = None, + config: Config | None = None, + encoder: Encoder | None = None, + scheduler: Scheduler | None = None, + ) -> "Tasks": """ Get the instance of the class. @@ -175,10 +184,7 @@ class Tasks(metaclass=Singleton): Tasks: The instance of the class. """ - if not Tasks._instance: - Tasks._instance = Tasks() - - return Tasks._instance + return Tasks(file=file, loop=loop, config=config, encoder=encoder, scheduler=scheduler) async def on_shutdown(self, _: web.Application): self.clear(shutdown=True) @@ -454,17 +460,17 @@ class Tasks(metaclass=Singleton): class HandleTask: - _tasks: Tasks - _handlers: list[type] - _scheduler: Scheduler - _config: Config - _task_name: str - def __init__(self, scheduler: Scheduler, tasks: Tasks, config: Config) -> None: - self._tasks = tasks - self._scheduler = scheduler - self._config = config + self._handlers: list[type] = [] + "The available handlers." + self._tasks: Tasks = tasks + "The tasks manager." + self._scheduler: Scheduler = scheduler + "The scheduler." + self._config: Config = config + "The configuration." self._task_name: str = f"{__class__.__name__}._dispatcher" + "The task name for the scheduler." def load(self) -> None: self._handlers: list[type] = self._discover() @@ -496,24 +502,36 @@ class HandleTask: self._scheduler.remove(self._task_name) def _dispatcher(self): + s: dict[list[str]] = {"h": [], "d": [], "u": [], "f": []} + for task in self._tasks.get_all(): if not task.handler_enabled: + s["d"].append(task.name) continue if not task.get_ytdlp_opts().get_all().get("download_archive"): LOG.debug(f"Task '{task.name}' does not have an archive file configured.") + s["f"].append(task.name) continue try: handler = self._find_handler(task) if handler is None: + s["u"].append(task.name) continue coro = self.dispatch(task, handler=handler) t = asyncio.create_task(coro, name=f"task-{task.id}") t.add_done_callback(lambda fut, t=task: self._handle_exception(fut, t)) + s["h"].append(task.name) except Exception as e: LOG.error(f"Failed to handle task '{task.name}'. '{e!s}'.") + s["f"].append(task.name) + + if len(self._tasks.get_all()) > 0: + LOG.info( + f"Task Handler summary: Handled: {len(s['h'])}, Unhandled: {len(s['u'])}, Disabled: {len(s['d'])}, Failed: {len(s['f'])}." + ) def _handle_exception(self, fut: asyncio.Task, task: Task) -> None: if fut.cancelled(): diff --git a/app/library/Utils.py b/app/library/Utils.py index f6bbe50f..45021a7b 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -61,10 +61,13 @@ REMOVE_KEYS: list = [ "opt_list_impersonate_targets": "--list-impersonate-targets", }, ] +"Keys to remove from yt-dlp options at various levels." YTDLP_INFO_CLS: YTDLP = None +"Cached YTDLP info class." ALLOWED_SUBS_EXTENSIONS: set[str] = {".srt", ".vtt", ".ass"} +"Allowed subtitle file extensions." FILES_TYPE: list = [ {"rx": re.compile(r"\.(avi|ts|mkv|mp4|mp3|mpv|ogm|m4v|webm|m4b)$", re.IGNORECASE), "type": "video"}, @@ -74,11 +77,15 @@ FILES_TYPE: list = [ {"rx": re.compile(r"\.(txt|nfo|md|json|yml|yaml|plexmatch)$", re.IGNORECASE), "type": "text"}, {"rx": re.compile(r"\.(nfo|json|jpg|torrent|\.info\.json)$", re.IGNORECASE), "type": "metadata"}, ] +"File type patterns for sidecar files." TAG_REGEX: re.Pattern[str] = re.compile(r"%{([^:}]+)(?::([^}]*))?}c") +"Regex to find tags in templates." DT_PATTERN: re.Pattern[str] = re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:[+-]\d{2}:\d{2}))\s?") +"Regex to match ISO 8601 datetime strings." T = TypeVar("T") +"Generic type variable." class StreamingError(Exception): @@ -111,11 +118,11 @@ def calc_download_path(base_path: str | Path, folder: str | None = None, create_ folder = folder.removeprefix("/") - realBasePath = base_path.resolve() - download_path = Path(realBasePath).joinpath(folder).resolve(strict=False) + realBasePath: Path = base_path.resolve() + download_path: Path = Path(realBasePath).joinpath(folder).resolve(strict=False) if not str(download_path).startswith(str(realBasePath)): - msg = f'Folder "{folder}" must resolve inside the base download folder "{realBasePath}".' + msg: str = f'Folder "{folder}" must resolve inside the base download folder "{realBasePath}".' raise Exception(msg) try: @@ -165,7 +172,7 @@ def extract_info( } # Remove keys that are not needed for info extraction. - keys_to_remove = [key for key in params if str(key).startswith("write") or key in ["postprocessors"]] + keys_to_remove: list = [key for key in params if str(key).startswith("write") or key in ["postprocessors"]] for key in keys_to_remove: params.pop(key, None) @@ -175,8 +182,8 @@ def extract_info( params["quiet"] = True log_wrapper = LogWrapper() - idDict = get_archive_id(url=url) - archive_id = f".{idDict['id']}" if idDict.get("id") else None + idDict: dict[str, str | None] = get_archive_id(url=url) + archive_id: str | None = f".{idDict['id']}" if idDict.get("id") else None log_wrapper.add_target( target=logging.getLogger(f"yt-dlp{archive_id if archive_id else '.extract_info'}"), @@ -244,7 +251,7 @@ def _is_safe_key(key: any) -> bool: return False # Block only truly dangerous dunder patterns - key_stripped = key.strip() + key_stripped: str = key.strip() # Block dunder attributes (starts AND ends with __) return not (key_stripped.startswith("__") and key_stripped.endswith("__")) @@ -279,7 +286,7 @@ def merge_dict( # Prevent deep recursion DoS if _depth > max_depth: - msg = f"Recursion depth limit exceeded ({max_depth})" + msg: str = f"Recursion depth limit exceeded ({max_depth})" raise RecursionError(msg) # Initialize circular reference tracking @@ -294,10 +301,10 @@ def merge_dict( raise ValueError(msg) # Track current objects - current_seen = _seen | {source_id, dest_id} + current_seen: set[Any | int] = _seen | {source_id, dest_id} # Create a clean copy of destination with only safe keys - destination_copy = {} + destination_copy: dict = {} for k, v in destination.items(): if _is_safe_key(k): # Prevent memory DoS from large lists @@ -311,7 +318,7 @@ def merge_dict( if not _is_safe_key(key): continue - destination_value = destination_copy.get(key) + destination_value: Any | None = destination_copy.get(key) # Recursively merge dictionaries with safety checks if isinstance(value, dict) and isinstance(destination_value, dict): @@ -325,8 +332,8 @@ def merge_dict( combined_size = len(value) + len(destination_value) if combined_size > max_list_size: # Truncate to stay within limits - available_space = max_list_size - len(destination_value) - truncated_value = value[: max(0, available_space)] + available_space: int = max_list_size - len(destination_value) + truncated_value: list = value[: max(0, available_space)] destination_copy[key] = copy.deepcopy(destination_value) + copy.deepcopy(truncated_value) else: destination_copy[key] = copy.deepcopy(destination_value) + copy.deepcopy(value) @@ -354,11 +361,13 @@ def check_id(file: Path) -> bool | str: bool|str: False if no file found, else the file path. """ - match = re.search(r"(?<=\[)(?:youtube-)?(?P[a-zA-Z0-9\-_]{11})(?=\])", file.stem, re.IGNORECASE) + match: re.Match[str] | None = re.search( + r"(?<=\[)(?:youtube-)?(?P[a-zA-Z0-9\-_]{11})(?=\])", file.stem, re.IGNORECASE + ) if not match: return False - id = match.groupdict().get("id") + id: str | None = match.groupdict().get("id") try: for f in file.parent.iterdir(): @@ -378,8 +387,8 @@ def check_id(file: Path) -> bool | str: @lru_cache(maxsize=512) def is_private_address(hostname: str) -> bool: - ip = socket.gethostbyname(hostname) - ip_obj = ipaddress.ip_address(ip) + ip: str = socket.gethostbyname(hostname) + ip_obj: ipaddress.IPv4Address | ipaddress.IPv6Address = ipaddress.ip_address(ip) return ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_reserved or ip_obj.is_link_local @@ -545,7 +554,7 @@ def get_file_sidecar(file: Path) -> list[dict]: list: List of sidecar files. """ - files = {} + files: dict = {} for i, f in enumerate(file.parent.glob(f"{glob.escape(file.stem)}.*")): if f == file or f.is_file() is False or f.stem.startswith("."): @@ -557,15 +566,15 @@ def get_file_sidecar(file: Path) -> list[dict]: content_type = "Unknown" for pattern in FILES_TYPE: if pattern["rx"].search(f.name): - content_type = pattern["type"] + content_type: str = pattern["type"] break if content_type == "subtitle": if f.suffix not in ALLOWED_SUBS_EXTENSIONS: continue - lg = re.search(r"\.(?P\w{2,3})\.\w{3}$", f.name) - lang = lg.groupdict().get("lang") if lg else "und" - content = {"file": f, "lang": lang, "name": f"{f.suffix[1:].upper()} ({i}) - {lang}"} + lg: re.Match[str] | None = re.search(r"\.(?P\w{2,3})\.\w{3}$", f.name) + lang: str | None = lg.groupdict().get("lang") if lg else "und" + content: dict[str, Any] = {"file": f, "lang": lang, "name": f"{f.suffix[1:].upper()} ({i}) - {lang}"} else: content = {"file": f} @@ -574,7 +583,7 @@ def get_file_sidecar(file: Path) -> list[dict]: files[content_type].append(content) - images = get_possible_images(str(file.parent)) + images: list[dict] = get_possible_images(str(file.parent)) if len(images) > 0: if "image" not in files: files["image"] = [] @@ -586,7 +595,7 @@ def get_file_sidecar(file: Path) -> list[dict]: @lru_cache(maxsize=512) def get_possible_images(dir: str) -> list[dict]: - images = [] + images: list = [] path_loc = Path(dir, "test.jpg") @@ -615,7 +624,7 @@ def get_mime_type(metadata: dict, file_path: Path) -> str: format_name = metadata.get("format_name", "") # Define mappings for HTML5-compatible video types - format_to_mime = { + format_to_mime: dict[str, str] = { "matroska": "video/x-matroska", # Default for MKV "webm": "video/webm", # MKV can also be WebM "mp4": "video/mp4", @@ -628,7 +637,7 @@ def get_mime_type(metadata: dict, file_path: Path) -> str: for fmt in format_name.split(","): fmt = fmt.strip().lower() if fmt in format_to_mime: - selected = format_to_mime[fmt] + selected: str = format_to_mime[fmt] if selected: return selected @@ -668,7 +677,7 @@ def get_file(download_path: str | Path, file: str | Path) -> tuple[Path, int]: LOG.error(f"Error calculating download path. {e!s}") return (Path(file), 404) - possibleFile = check_id(file=realFile) + possibleFile: bool | str = check_id(file=realFile) if not possibleFile: return (realFile, 404) @@ -775,7 +784,7 @@ def get( return default() if callable(default) else default # Split the path by the separator and traverse the data structure. - segments = path.split(separator) + segments: list[str] = path.split(separator) for segment in segments: if isinstance(data, dict): if segment in data: @@ -873,7 +882,7 @@ def get_files(base_path: Path | str, dir: str | None = None): if not content_type: content_type = "download" - stat = file.stat() + stat: os.stat_result = file.stat() contents.append( { @@ -897,9 +906,6 @@ def get_files(base_path: Path | str, dir: str | None = None): def clean_item(item: dict, keys: list | tuple) -> tuple[dict, bool]: """ Remove given keys from a dictionary. - This function modifies the dictionary in place and returns a tuple - containing the modified dictionary and a boolean indicating - whether any keys were removed. Args: item (dict): The item to clean. @@ -913,6 +919,7 @@ def clean_item(item: dict, keys: list | tuple) -> tuple[dict, bool]: """ status = False + item = copy.deepcopy(item) if not isinstance(item, dict): msg = "Item must be a dictionary." @@ -946,7 +953,7 @@ def strip_newline(string: str) -> str: if not string: return "" - res = re.sub(r"(\r\n|\r|\n)", " ", string) + res: str = re.sub(r"(\r\n|\r|\n)", " ", string) return res.strip() if res else "" @@ -978,34 +985,34 @@ async def read_logfile(file: Path, offset: int = 0, limit: int = 50) -> dict: try: async with await open_file(file, "rb") as f: await f.seek(0, os.SEEK_END) - file_size = await f.tell() + file_size: int = await f.tell() block_size = 1024 - block_end = file_size - buffer = b"" - lines = [] + block_end: int = file_size + buffer: bytes = b"" + lines: list = [] - required_count = offset + limit + 1 + required_count: int = offset + limit + 1 while len(lines) < required_count and block_end > 0: - block_start = max(0, block_end - block_size) + block_start: int = max(0, block_end - block_size) await f.seek(block_start) - chunk = await f.read(block_end - block_start) - buffer = chunk + buffer # prepend the chunk + chunk: bytes = await f.read(block_end - block_start) + buffer: bytes = chunk + buffer # prepend the chunk lines = buffer.splitlines() block_end = block_start if len(lines) > offset + limit: - next_offset = offset + limit + next_offset: int = offset + limit end_is_reached = False else: next_offset = None end_is_reached = True for line in lines[-(offset + limit) : -offset] if offset else lines[-limit:]: - line_bytes = line if isinstance(line, bytes) else line.encode() - msg = line.decode(errors="replace") - dt_match = DT_PATTERN.match(msg) + line_bytes: bytes | str = line if isinstance(line, bytes) else line.encode() + msg: str = line.decode(errors="replace") + dt_match: re.Match[str] | None = DT_PATTERN.match(msg) result.append( { "id": sha256(line_bytes).hexdigest(), @@ -1041,13 +1048,13 @@ async def tail_log(file: Path, emitter: callable, sleep_time: float = 0.5): async with await open_file(file, "rb") as f: await f.seek(0, os.SEEK_END) while True: - line = await f.readline() + line: bytes = await f.readline() if not line: await asyncio_sleep(sleep_time) continue - msg = line.decode(errors="replace") - dt_match = DT_PATTERN.match(msg) + msg: str = line.decode(errors="replace") + dt_match: re.Match[str] | None = DT_PATTERN.match(msg) await emitter( { @@ -1101,7 +1108,7 @@ def get_archive_id(url: str) -> dict[str, str | None]: """ global YTDLP_INFO_CLS # noqa: PLW0603 - idDict = { + idDict: dict[str, None] = { "id": None, "ie_key": None, "archive_id": None, @@ -1158,7 +1165,7 @@ def dt_delta(delta: timedelta) -> str: hours, rem = divmod(rem, 3600) minutes, secs = divmod(rem, 60) - parts = [] + parts: list[str] = [] if days: parts.append(f"{days}d") if hours: @@ -1269,6 +1276,7 @@ def load_modules(root_path: Path, directory: Path): full_name: str = f"{package_name}.{name}" if name.startswith("_"): continue + try: importlib.import_module(full_name) except ImportError as e: @@ -1377,37 +1385,6 @@ def ytdlp_reject(entry: dict, yt_params: dict) -> tuple[bool, str]: return (True, "") -def find_unpickleable(obj, name="root", seen=None): - import pickle - - if seen is None: - seen = set() - - if id(obj) in seen: - return - - seen.add(id(obj)) - - try: - pickle.dumps(obj) - except Exception as e: - LOG.error(f"[UNPICKLEABLE] {name}: {e}") - - if isinstance(obj, dict): - for k, v in obj.items(): - find_unpickleable(v, f"{name}[{repr(k)!s}]", seen) - elif hasattr(obj, "__dict__"): - for attr in vars(obj): - try: - value = getattr(obj, attr) - find_unpickleable(value, f"{name}.{attr}", seen) - except Exception as ie: - LOG.error(f"[ERROR] Accessing {name}.{attr}: {ie}") - elif isinstance(obj, list | tuple | set): - for idx, item in enumerate(obj): - find_unpickleable(item, f"{name}[{idx}]", seen) - - def list_folders(path: Path, base: Path, depth_limit: int) -> list[str]: """ List all folders relative to a base path, up to a specified depth limit. diff --git a/app/library/YTDLPOpts.py b/app/library/YTDLPOpts.py index ab1b6239..67526352 100644 --- a/app/library/YTDLPOpts.py +++ b/app/library/YTDLPOpts.py @@ -9,20 +9,17 @@ LOG: logging.Logger = logging.getLogger("YTDLPOpts") class YTDLPOpts: - _item_opts: dict = {} - """The item options.""" - - _preset_opts: dict = {} - """The preset options.""" - - _item_cli: list = [] - """The command options for yt-dlp from item.""" - - _preset_cli: str = "" - """The command options for yt-dlp from preset.""" - def __init__(self): self._config: Config = Config.get_instance() + "The config instance." + self._item_opts: dict = {} + "The item options." + self._preset_opts: dict = {} + "The preset options." + self._item_cli: list = [] + "The command options for yt-dlp from item." + self._preset_cli: str = "" + "The command options for yt-dlp from preset." @staticmethod def get_instance() -> "YTDLPOpts": diff --git a/app/library/cache.py b/app/library/cache.py index 4fd32509..34aa7bbe 100644 --- a/app/library/cache.py +++ b/app/library/cache.py @@ -11,12 +11,8 @@ class Cache(metaclass=ThreadSafe): """ Initialize the Cache. """ - # Prevent reinitialization in singleton context. - if hasattr(self, "_initialized") and self._initialized: - return self._cache: dict[str, tuple[Any, float | None]] = {} self._lock = threading.Lock() - self._initialized = True def set(self, key: str, value: Any, ttl: float | None = None) -> None: """ diff --git a/app/library/conditions.py b/app/library/conditions.py index 98c45de9..adb813ed 100644 --- a/app/library/conditions.py +++ b/app/library/conditions.py @@ -14,7 +14,7 @@ from .mini_filter import match_str from .Singleton import Singleton from .Utils import arg_converter, init_class -LOG = logging.getLogger("conditions") +LOG: logging.Logger = logging.getLogger("conditions") @dataclass(kw_only=True) @@ -49,18 +49,14 @@ class Conditions(metaclass=Singleton): This class is used to manage the download conditions. """ - _items: list[Condition] = [] - """The list of items.""" - - _instance = None - """The instance of the class.""" - def __init__(self, file: Path | str | None = None, config: Config | None = None): - Conditions._instance = self + self._items: list[Condition] = [] + "The list of items." config = config or Config.get_instance() self._file: Path = Path(file) if file else Path(config.config_path) / "conditions.json" + "The path to the file where the items are stored." if self._file.exists() and "600" != self._file.stat().st_mode: try: @@ -68,14 +64,8 @@ class Conditions(metaclass=Singleton): except Exception: pass - async def event_handler(_, __): - msg = "Not implemented" - raise Exception(msg) - - EventBus.get_instance().subscribe(Events.CONDITIONS_ADD, event_handler, f"{__class__.__name__}.save") - @staticmethod - def get_instance() -> "Conditions": + def get_instance(file: Path | str | None = None, config: Config | None = None) -> "Conditions": """ Get the instance of the class. @@ -83,10 +73,7 @@ class Conditions(metaclass=Singleton): Conditions: The instance of the class """ - if not Conditions._instance: - Conditions._instance = Conditions() - - return Conditions._instance + return Conditions(file=file, config=config) async def on_shutdown(self, _: web.Application): pass @@ -104,6 +91,12 @@ class Conditions(metaclass=Singleton): """ self.load() + async def event_handler(_, __): + msg = "Not implemented" + raise Exception(msg) + + EventBus.get_instance().subscribe(Events.CONDITIONS_ADD, event_handler, f"{__class__.__name__}.save") + def get_all(self) -> list[Condition]: """Return the items.""" return self._items diff --git a/app/library/config.py b/app/library/config.py index 8a83578b..071194ee 100644 --- a/app/library/config.py +++ b/app/library/config.py @@ -12,6 +12,7 @@ from typing import TYPE_CHECKING import coloredlogs from dotenv import load_dotenv +from .Singleton import Singleton from .Utils import FileLogFormatter, arg_converter from .version import APP_BRANCH, APP_BUILD_DATE, APP_COMMIT_SHA, APP_VERSION @@ -19,7 +20,7 @@ if TYPE_CHECKING: from subprocess import CompletedProcess -class Config: +class Config(metaclass=Singleton): app_env: str = "production" """The application environment, can be 'production' or 'development'.""" @@ -131,9 +132,6 @@ class Config: app_branch: str = APP_BRANCH "The branch of the application." - __instance = None - "The instance of the class." - started: int = 0 "The time the application was started." @@ -206,7 +204,6 @@ class Config: "The variables that are set manually." _immutable: tuple = ( - "__instance", "ytdl_options", "started", "ytdlp_cli", @@ -284,8 +281,7 @@ class Config: @staticmethod def get_instance(is_native: bool = False) -> "Config": - """Static access method.""" - cls: Config = Config(is_native) if not Config.__instance else Config.__instance + cls = Config(is_native) cls.is_native = is_native or cls.is_native return cls @@ -297,13 +293,6 @@ class Config: return Config._manager def __init__(self, is_native: bool = False): - """Virtually private constructor.""" - if Config.__instance is not None: - msg = "This class is a singleton. Use Config.get_instance() instead." - raise Exception(msg) - - Config.__instance = self - baseDefaultPath: str = str(Path(__file__).parent.parent.parent.absolute()) self.is_native = is_native diff --git a/app/library/dl_fields.py b/app/library/dl_fields.py index b563581c..85abee56 100644 --- a/app/library/dl_fields.py +++ b/app/library/dl_fields.py @@ -100,21 +100,13 @@ class DLFields(metaclass=Singleton): This class is used to manage the DLFields. """ - _items: list[DLField] = [] - """The list of items.""" - - _instance = None - """The instance of the class.""" - - _config: Config = None - """The config instance.""" - def __init__(self, file: str | Path | None = None, config: Config | None = None): - DLFields._instance = self - - self._config = config or Config.get_instance() - - self._file: Path = Path(file) if file else Path(self._config.config_path).joinpath("dl_fields.json") + self._items: list[DLField] = [] + "The list of items." + config: Config = config or Config.get_instance() + "The configuration instance." + self._file: Path = Path(file) if file else Path(config.config_path).joinpath("dl_fields.json") + "The path to the file where the items are stored." if self._file.exists() and "600" != self._file.stat().st_mode: try: @@ -122,14 +114,8 @@ class DLFields(metaclass=Singleton): except Exception: pass - async def event_handler(_, __): - msg = "Not implemented" - raise Exception(msg) - - EventBus.get_instance().subscribe(Events.DLFIELDS_ADD, event_handler, f"{__class__.__name__}.add") - @staticmethod - def get_instance() -> "DLFields": + def get_instance(file: str | Path | None = None, config: Config | None = None) -> "DLFields": """ Get the instance of the class. @@ -137,10 +123,7 @@ class DLFields(metaclass=Singleton): DLFields: The instance of the class """ - if not DLFields._instance: - DLFields._instance = DLFields() - - return DLFields._instance + return DLFields(file=file, config=config) async def on_shutdown(self, _: web.Application): pass @@ -158,6 +141,12 @@ class DLFields(metaclass=Singleton): """ self.load() + async def event_handler(_, __): + msg = "Not implemented" + raise Exception(msg) + + EventBus.get_instance().subscribe(Events.DLFIELDS_ADD, event_handler, f"{__class__.__name__}.add") + def get_all(self) -> list[DLField]: """Return the items.""" return self._items diff --git a/app/library/ffprobe.py b/app/library/ffprobe.py index 30ef5800..54ba5404 100644 --- a/app/library/ffprobe.py +++ b/app/library/ffprobe.py @@ -277,7 +277,7 @@ async def ffprobe(file: str) -> FFProbeResult: msg = "ffprobe not found." raise OSError(msg) from e - args = ["-v", "quiet", "-of", "json", "-show_streams", "-show_format", str(f)] + args: list[str] = ["-v", "quiet", "-of", "json", "-show_streams", "-show_format", str(f)] p = await asyncio.create_subprocess_exec( "ffprobe", @@ -287,13 +287,13 @@ async def ffprobe(file: str) -> FFProbeResult: creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0, ) - exitCode = await p.wait() + exitCode: int = await p.wait() data, err = await p.communicate() if 0 == exitCode: parsed: dict = json.loads(data.decode("utf-8")) else: - msg = f"ffprobe returned with non-0 exit code. '{err.decode('utf-8')}'" + msg: str = f"ffprobe returned with non-0 exit code. '{err.decode('utf-8')}'" raise FFProbeError(msg) result = FFProbeResult() diff --git a/app/library/task_handlers/youtube.py b/app/library/task_handlers/youtube.py index 9d652793..1ac3a952 100644 --- a/app/library/task_handlers/youtube.py +++ b/app/library/task_handlers/youtube.py @@ -61,7 +61,7 @@ class YoutubeHandler(BaseHandler): params: dict = task.get_ytdlp_opts().get_all() feed_url: str = YoutubeHandler.FEED.format(type=parsed["type"], id=parsed["id"]) - LOG.info(f"'{task.name}': Fetching feed.") + LOG.debug(f"'{task.name}': Fetching feed.") items: list = [] @@ -106,7 +106,7 @@ class YoutubeHandler(BaseHandler): if real_count < 1: LOG.warning(f"'{task.name}': No entries found the RSS feed. URL: {feed_url}") else: - LOG.info(f"'{task.name}': Feed has '{real_count}' entries, all already downloaded/queued.") + LOG.debug(f"'{task.name}': Feed has '{real_count}' entries, all already downloaded/queued.") return filtered: list = [] @@ -134,7 +134,7 @@ class YoutubeHandler(BaseHandler): filtered.append(item) if len(filtered) < 1: - LOG.info(f"'{task.name}': Feed has '{real_count}' entries, all already downloaded/queued.") + LOG.debug(f"'{task.name}': Feed has '{real_count}' entries, all already downloaded/queued.") return LOG.info(f"'{task.name}': Found '{len(filtered)}/{real_count}' new items from feed.") diff --git a/app/library/ytdlp.py b/app/library/ytdlp.py index c67ce607..0e4caaf4 100644 --- a/app/library/ytdlp.py +++ b/app/library/ytdlp.py @@ -14,6 +14,7 @@ class _ArchiveProxy: def __init__(self, file: str | None): self._file: str | None = file + "The archive file path." def __contains__(self, item: str) -> bool: if not self._file or not item: diff --git a/app/tests/test_cache.py b/app/tests/test_cache.py index b8460434..20f72d68 100644 --- a/app/tests/test_cache.py +++ b/app/tests/test_cache.py @@ -29,9 +29,7 @@ class TestCache: def setup_method(self): """Set up test fixtures.""" - # Clear any existing cache instance - if hasattr(Cache, "_instances"): - Cache._instances.clear() + Cache._reset_singleton() self.cache = Cache() def test_singleton_behavior(self): diff --git a/app/tests/test_conditions.py b/app/tests/test_conditions.py index 7ea5a606..8a57e761 100644 --- a/app/tests/test_conditions.py +++ b/app/tests/test_conditions.py @@ -21,7 +21,6 @@ from unittest.mock import MagicMock, patch import pytest from app.library.conditions import Condition, Conditions -from app.library.Singleton import Singleton class TestCondition: @@ -112,11 +111,7 @@ class TestConditions: def setup_method(self): """Set up test fixtures by clearing singleton instances.""" - # Clear singleton instances before each test - Singleton._instances.clear() - # Reset class variable - Conditions._items = [] - Conditions._instance = None + Conditions._reset_singleton() def test_conditions_singleton(self): """Test that Conditions follows singleton pattern.""" diff --git a/app/tests/test_dl_fields.py b/app/tests/test_dl_fields.py index 72ff381e..d888c310 100644 --- a/app/tests/test_dl_fields.py +++ b/app/tests/test_dl_fields.py @@ -22,7 +22,6 @@ from unittest.mock import MagicMock, patch import pytest from app.library.dl_fields import DLField, DLFields, FieldType -from app.library.Singleton import Singleton class TestFieldType: @@ -156,12 +155,7 @@ class TestDLFields: def setup_method(self): """Set up test fixtures.""" - # Clear singleton instances before each test - Singleton._instances.clear() - if hasattr(DLFields, "_instances"): - DLFields._instances.clear() - # Also clear the class-level _items list - DLFields._items = [] + DLFields._reset_singleton() @pytest.fixture def temp_file(self): @@ -219,11 +213,9 @@ class TestDLFields: assert fields1 is fields2 @patch("app.library.dl_fields.Config") - @patch("app.library.dl_fields.EventBus") - def test_dl_fields_initialization(self, mock_event_bus, mock_config): + def test_dl_fields_initialization(self, mock_config): """Test DLFields initialization.""" mock_config.get_instance.return_value.config_path = "/tmp" - mock_event_bus.get_instance.return_value.subscribe = MagicMock() with tempfile.TemporaryDirectory() as temp_dir: temp_file = Path(temp_dir) / "test_fields.json" @@ -231,8 +223,6 @@ class TestDLFields: fields = DLFields(file=str(temp_file)) assert fields._file == temp_file - assert fields._config is not None - mock_event_bus.get_instance.return_value.subscribe.assert_called_once() @patch("app.library.dl_fields.Config") def test_dl_fields_load_empty_file(self, mock_config, temp_file): diff --git a/app/tests/test_events.py b/app/tests/test_events.py index 11079cef..a23ee1e1 100644 --- a/app/tests/test_events.py +++ b/app/tests/test_events.py @@ -627,10 +627,7 @@ class TestEventBus: def setup_method(self): """Clear EventBus listeners and reset singleton before each test.""" - # Reset singleton instance to allow mocking to work - EventBus._instance = None - bus = EventBus.get_instance() - bus.clear() + EventBus._reset_singleton() @patch("app.library.config.Config") @patch("app.library.BackgroundWorker.BackgroundWorker") @@ -656,9 +653,6 @@ class TestEventBus: def test_event_bus_initialization(self): """Test EventBus initialization with new clean design.""" - # Reset singleton to ensure fresh instance - EventBus._instance = None - # Create EventBus with clean initialization bus = EventBus() diff --git a/app/tests/test_presets.py b/app/tests/test_presets.py index 74f605ed..95464e8f 100644 --- a/app/tests/test_presets.py +++ b/app/tests/test_presets.py @@ -99,13 +99,6 @@ class TestPresets: def setup_method(self): """Set up test environment before each test.""" # Reset singleton completely - Presets._instance = None - - def teardown_method(self): - """Clean up after each test.""" - # Reset singleton completely - Presets.get_instance()._items.clear() - Presets._instance = None Presets._reset_singleton() @patch("app.library.Presets.Config") diff --git a/app/tests/test_services.py b/app/tests/test_services.py new file mode 100644 index 00000000..75850d3f --- /dev/null +++ b/app/tests/test_services.py @@ -0,0 +1,552 @@ +import logging +from unittest.mock import MagicMock, patch + +import pytest + +from app.library.Services import Services + + +class TestServices: + """Test the Services singleton class.""" + + def setup_method(self): + """Clear services before each test.""" + Services._reset_singleton() + + def test_singleton_behavior(self): + """Test that Services follows singleton pattern.""" + service1 = Services() + service2 = Services() + service3 = Services.get_instance() + + assert service1 is service2, "Multiple Services() calls should return same instance" + assert service1 is service3, "get_instance() should return same instance" + assert id(service1) == id(service2) == id(service3), "All references should point to same object" + + def test_add_and_get_service(self): + """Test adding and retrieving services.""" + services = Services() + test_service = "test_value" + + services.add("test_service", test_service) + retrieved = services.get("test_service") + + assert retrieved == test_service, "Should retrieve the same service that was added" + + def test_get_nonexistent_service(self): + """Test retrieving a service that doesn't exist.""" + services = Services() + result = services.get("nonexistent") + + assert result is None, "Should return None for nonexistent service" + + def test_has_service(self): + """Test checking if service exists.""" + services = Services() + services.add("existing", "value") + + assert services.has("existing") is True, "Should return True for existing service" + assert services.has("nonexistent") is False, "Should return False for nonexistent service" + + def test_remove_service(self): + """Test removing a service.""" + services = Services() + services.add("to_remove", "value") + + assert services.has("to_remove") is True, "Service should exist before removal" + services.remove("to_remove") + assert services.has("to_remove") is False, "Service should not exist after removal" + + def test_remove_nonexistent_service(self): + """Test removing a service that doesn't exist.""" + services = Services() + # Should not raise an exception + services.remove("nonexistent") + assert services.has("nonexistent") is False + + def test_clear_services(self): + """Test clearing all services.""" + services = Services() + services.add("service1", "value1") + services.add("service2", "value2") + + assert len(services.get_all()) == 2, "Should have 2 services before clear" + services.clear() + assert len(services.get_all()) == 0, "Should have 0 services after clear" + + def test_add_all_services(self): + """Test adding multiple services at once.""" + services = Services() + services_dict = { + "service1": "value1", + "service2": "value2", + "service3": "value3" + } + + services.add_all(services_dict) + + assert services.get("service1") == "value1" + assert services.get("service2") == "value2" + assert services.get("service3") == "value3" + assert len(services.get_all()) == 3 + + def test_get_all_returns_copy(self): + """Test that get_all returns a copy, not the original dict.""" + services = Services() + services.add("test", "value") + + all_services = services.get_all() + all_services["injected"] = "malicious" + + assert "injected" not in services.get_all(), "Modifying returned dict should not affect internal state" + + def test_handle_sync_with_matching_args(self): + """Test synchronous handler with matching arguments.""" + services = Services() + services.add("db", "database_connection") + services.add("logger", "logger_instance") + + def test_handler(db, logger): + return f"Handler called with {db} and {logger}" + + result = services.handle_sync(test_handler) + expected = "Handler called with database_connection and logger_instance" + assert result == expected + + def test_handle_sync_with_extra_kwargs(self): + """Test synchronous handler with additional kwargs.""" + services = Services() + services.add("db", "database_connection") + + def test_handler(db, user_id): + return f"Handler called with {db} and {user_id}" + + result = services.handle_sync(test_handler, user_id=123) + expected = "Handler called with database_connection and 123" + assert result == expected + + def test_handle_sync_with_missing_args(self): + """Test synchronous handler with missing arguments.""" + services = Services() + services.add("db", "database_connection") + + def test_handler(db_param, missing_service_param): # noqa: ARG001 + return "Should not reach here" + + with patch("app.library.Services.LOG") as mock_logger: + # The current implementation still calls the handler even with missing args + # This causes a TypeError, which is the actual current behavior + with pytest.raises(TypeError, match=r"missing .* required positional argument"): + services.handle_sync(test_handler) + + # Should still log error about missing arguments + mock_logger.error.assert_called_once() + error_call = mock_logger.error.call_args[0][0] + assert "Missing arguments" in error_call + assert "missing_service_param" in error_call + + def test_handle_sync_no_args_handler(self): + """Test synchronous handler that takes no arguments.""" + services = Services() + services.add("unused", "value") + + def test_handler(): + return "No args handler" + + result = services.handle_sync(test_handler) + assert result == "No args handler" + + @pytest.mark.asyncio + async def test_handle_async_with_matching_args(self): + """Test asynchronous handler with matching arguments.""" + services = Services() + services.add("db", "database_connection") + services.add("logger", "logger_instance") + + async def test_handler(db, logger): + return f"Async handler called with {db} and {logger}" + + result = await services.handle_async(test_handler) + expected = "Async handler called with database_connection and logger_instance" + assert result == expected + + @pytest.mark.asyncio + async def test_handle_async_with_extra_kwargs(self): + """Test asynchronous handler with additional kwargs.""" + services = Services() + services.add("db", "database_connection") + + async def test_handler(db, user_id): + return f"Async handler called with {db} and {user_id}" + + result = await services.handle_async(test_handler, user_id=456) + expected = "Async handler called with database_connection and 456" + assert result == expected + + @pytest.mark.asyncio + async def test_handle_async_with_missing_args(self): + """Test asynchronous handler with missing arguments.""" + services = Services() + services.add("db", "database_connection") + + async def test_handler(db_param, missing_service_param): # noqa: ARG001 + return "Should not reach here" + + with patch("app.library.Services.LOG") as mock_logger: + # The current implementation still calls the handler even with missing args + # This causes a TypeError, which is the actual current behavior + with pytest.raises(TypeError, match=r"missing .* required positional argument"): + await services.handle_async(test_handler) + + # Should still log error about missing arguments + mock_logger.error.assert_called_once() + error_call = mock_logger.error.call_args[0][0] + assert "Missing arguments" in error_call + assert "missing_service_param" in error_call + + @pytest.mark.asyncio + async def test_handle_async_no_args_handler(self): + """Test asynchronous handler that takes no arguments.""" + services = Services() + services.add("unused", "value") + + async def test_handler(): + return "No args async handler" + + result = await services.handle_async(test_handler) + assert result == "No args async handler" + + def test_handle_sync_kwargs_override_services(self): + """Test that kwargs override services with same name.""" + services = Services() + services.add("param", "service_value") + + def test_handler(param): + return f"Received: {param}" + + result = services.handle_sync(test_handler, param="override_value") + assert result == "Received: override_value" + + @pytest.mark.asyncio + async def test_handle_async_kwargs_override_services(self): + """Test that kwargs override services with same name in async handler.""" + services = Services() + services.add("param", "service_value") + + async def test_handler(param): + return f"Received: {param}" + + result = await services.handle_async(test_handler, param="override_value") + assert result == "Received: override_value" + + def test_handle_sync_with_complex_signature(self): + """Test synchronous handler with complex function signature.""" + services = Services() + services.add("db", "database") + services.add("cache", "redis") + + def complex_handler(db, cache, *args, **kwargs): + return f"db:{db}, cache:{cache}, args:{args}, kwargs:{kwargs}" + + result = services.handle_sync(complex_handler, extra="value") + expected = "db:database, cache:redis, args:(), kwargs:{}" + assert result == expected + + @pytest.mark.asyncio + async def test_handle_async_with_complex_signature(self): + """Test asynchronous handler with complex function signature.""" + services = Services() + services.add("db", "database") + services.add("cache", "redis") + + async def complex_handler(db, cache, *args, **kwargs): + return f"db:{db}, cache:{cache}, args:{args}, kwargs:{kwargs}" + + result = await services.handle_async(complex_handler, extra="value") + expected = "db:database, cache:redis, args:(), kwargs:{}" + assert result == expected + + def test_service_types_preserved(self): + """Test that different service types are preserved correctly.""" + services = Services() + + # Test various types + string_service = "string_value" + int_service = 42 + list_service = [1, 2, 3] + dict_service = {"key": "value"} + custom_object = MagicMock() + + services.add("string", string_service) + services.add("int", int_service) + services.add("list", list_service) + services.add("dict", dict_service) + services.add("object", custom_object) + + assert services.get("string") == string_service + assert services.get("int") == int_service + assert services.get("list") == list_service + assert services.get("dict") == dict_service + assert services.get("object") is custom_object + + def test_add_none_service(self): + """Test adding None as a service value.""" + services = Services() + services.add("none_service", None) + + assert services.has("none_service") is True + assert services.get("none_service") is None + + def test_service_name_edge_cases(self): + """Test edge cases for service names.""" + services = Services() + + # Empty string name + services.add("", "empty_name_value") + assert services.get("") == "empty_name_value" + + # Numeric string name + services.add("123", "numeric_name") + assert services.get("123") == "numeric_name" + + # Special characters in name + services.add("special-chars_123!@#", "special_value") + assert services.get("special-chars_123!@#") == "special_value" + + def test_overwrite_existing_service(self): + """Test overwriting an existing service.""" + services = Services() + services.add("service", "original_value") + services.add("service", "new_value") + + assert services.get("service") == "new_value" + + def test_singleton_persistence_across_operations(self): + """Test that singleton behavior persists across various operations.""" + # Get instance and add a service + services1 = Services() + services1.add("persistent", "value") + + # Get another instance and verify service exists + services2 = Services.get_instance() + assert services2.get("persistent") == "value" + + # Clear from one instance + services1.clear() + + # Verify cleared in other instance + assert services2.get("persistent") is None + + def test_handler_exception_propagation(self): + """Test that exceptions in handlers are properly propagated.""" + services = Services() + + def failing_handler(): + msg = "Handler failed" + raise ValueError(msg) + + with pytest.raises(ValueError, match="Handler failed"): + services.handle_sync(failing_handler) + + @pytest.mark.asyncio + async def test_async_handler_exception_propagation(self): + """Test that exceptions in async handlers are properly propagated.""" + services = Services() + + async def failing_async_handler(): + msg = "Async handler failed" + raise RuntimeError(msg) + + with pytest.raises(RuntimeError, match="Async handler failed"): + await services.handle_async(failing_async_handler) + + def test_handle_sync_with_callable_object(self): + """Test handle_sync with callable object instead of function.""" + services = Services() + services.add("data", "test_data") + + class CallableHandler: + def __call__(self, data): + return f"Callable received: {data}" + + handler = CallableHandler() + result = services.handle_sync(handler) + assert result == "Callable received: test_data" + + @pytest.mark.asyncio + async def test_handle_async_with_callable_object(self): + """Test handle_async with callable object instead of function.""" + services = Services() + services.add("data", "test_data") + + class AsyncCallableHandler: + async def __call__(self, data): + return f"Async callable received: {data}" + + handler = AsyncCallableHandler() + result = await services.handle_async(handler) + assert result == "Async callable received: test_data" + + def test_inspect_signature_edge_cases(self): + """Test that inspect.signature works correctly with edge cases.""" + services = Services() + + # Lambda function replacement + def lambda_handler(x): + return f"Lambda: {x}" + services.add("x", "lambda_value") + result = services.handle_sync(lambda_handler) + assert result == "Lambda: lambda_value" + + def test_logging_configuration(self): + """Test that logging is properly configured.""" + # This test verifies the module-level logger setup + from app.library.Services import LOG + + assert isinstance(LOG, logging.Logger) + assert LOG.name == "app.library.Services" + + def test_service_container_isolation(self): + """Test that services don't interfere with each other.""" + services = Services() + + # Add services with potentially conflicting names + services.add("data", {"type": "database"}) + services.add("data_backup", {"type": "backup"}) + + assert services.get("data")["type"] == "database" + assert services.get("data_backup")["type"] == "backup" + + # Remove one, other should remain + services.remove("data") + assert services.get("data") is None + assert services.get("data_backup")["type"] == "backup" + + def test_large_number_of_services(self): + """Test handling a large number of services.""" + services = Services() + + # Add many services + num_services = 1000 + for i in range(num_services): + services.add(f"service_{i}", f"value_{i}") + + # Verify all exist + assert len(services.get_all()) == num_services + + # Verify specific services + assert services.get("service_0") == "value_0" + assert services.get("service_500") == "value_500" + assert services.get("service_999") == "value_999" + + # Clear should work efficiently + services.clear() + assert len(services.get_all()) == 0 + + def test_add_all_overwrites_existing(self): + """Test that add_all overwrites existing services.""" + services = Services() + services.add("existing", "original") + + new_services = { + "existing": "overwritten", + "new": "value" + } + + services.add_all(new_services) + + assert services.get("existing") == "overwritten" + assert services.get("new") == "value" + + def test_add_all_empty_dict(self): + """Test add_all with empty dictionary.""" + services = Services() + services.add("existing", "value") + + services.add_all({}) + + # Should not affect existing services + assert services.get("existing") == "value" + assert len(services.get_all()) == 1 + + def test_type_var_generic_behavior(self): + """Test that TypeVar T is handled correctly.""" + services = Services() + + # Add different types and ensure they're returned correctly + services.add("string", "text") + services.add("number", 42) + services.add("boolean", True) # noqa: FBT003 + + # Type should be preserved (runtime check) + assert isinstance(services.get("string"), str) + assert isinstance(services.get("number"), int) + assert isinstance(services.get("boolean"), bool) + + def test_concurrent_access_safety(self): + """Test basic thread safety aspects of singleton.""" + import threading + import time + + results = [] + + def get_instance(): + time.sleep(0.01) # Small delay to increase chance of race condition + instance = Services() + results.append(id(instance)) + + # Create multiple threads + threads = [] + for _ in range(10): + thread = threading.Thread(target=get_instance) + threads.append(thread) + thread.start() + + # Wait for all threads + for thread in threads: + thread.join() + + # All should be the same instance + assert len(set(results)) == 1, "All threads should get the same singleton instance" + + def test_method_chaining_possibility(self): + """Test that methods can be potentially chained.""" + services = Services() + + # While current implementation doesn't return self, test the pattern works + services.add("test1", "value1") + services.add("test2", "value2") + services.remove("test1") + + assert services.get("test1") is None + assert services.get("test2") == "value2" + + def test_edge_case_empty_handler_name(self): + """Test handlers with minimal or no names.""" + services = Services() + services.add("param", "value") + + # Anonymous lambda + result = services.handle_sync(lambda param: f"anon: {param}") + assert result == "anon: value" + + # Function with minimal signature info + def minimal(param): return param + result = services.handle_sync(minimal) + assert result == "value" + + def test_services_state_isolation(self): + """Test that different Services instances share state properly.""" + # This test verifies the singleton behavior more thoroughly + s1 = Services() + s1.add("shared", "data") + + s2 = Services.get_instance() + assert s2.get("shared") == "data" + + s2.add("another", "value") + assert s1.get("another") == "value" + + # Clear from s1 affects s2 + s1.clear() + assert len(s2.get_all()) == 0 diff --git a/app/tests/test_utils.py b/app/tests/test_utils.py index 214f78f6..507a28a6 100644 --- a/app/tests/test_utils.py +++ b/app/tests/test_utils.py @@ -26,7 +26,6 @@ from app.library.Utils import ( encrypt_data, extract_info, extract_ytdlp_logs, - find_unpickleable, get, get_archive_id, get_file, @@ -1424,30 +1423,6 @@ class TestYtdlpReject: assert isinstance(message, str) -class TestFindUnpickleable: - """Test the find_unpickleable function.""" - - def test_find_unpickleable_simple(self): - """Test with simple pickleable object.""" - obj = {"key": "value", "number": 42} - - try: - find_unpickleable(obj) - # Should not find any unpickleable items - except Exception: - # Function might raise exceptions for complex objects - pass - - def test_find_unpickleable_complex(self): - """Test with complex object.""" - obj = {"func": lambda x: x} # Lambda is not pickleable - - try: - find_unpickleable(obj) - except Exception: - pass - - class TestInitClass: """Test the init_class function."""