diff --git a/app/library/LogWrapper.py b/app/library/LogWrapper.py new file mode 100644 index 00000000..3a8df8f1 --- /dev/null +++ b/app/library/LogWrapper.py @@ -0,0 +1,158 @@ +import logging +from dataclasses import dataclass + + +@dataclass(kw_only=True) +class LogTarget: + """ + A data class that represents a logging target with its level and type. + + Attributes: + target: The logging target, which can be a logging.Logger instance or a callable. + level (int): The logging level for the target. + logger (bool): True if the target is a logging.Logger instance, False otherwise. + type: The type of the target. + + """ + + target: "logging.Logger|callable" + level: int + logger: bool + + +class LogWrapper: + """ + A wrapper class for logging that allows adding multiple logging targets with different logging levels. + + Attributes: + targets (list[dict]): A list of dictionaries where each dictionary represents a logging target with its level and type. + + Methods: + add_target(target, level=logging.DEBUG): + Adds a new logging target with the specified logging level. + + has_targets(): + Checks if there are any logging targets added. + + _log(level, msg, *args, **kwargs): + Logs a message to all targets that have a logging level less than or equal to the specified level. + + debug(msg, *args, **kwargs): + Logs a debug message to all targets. + + info(msg, *args, **kwargs): + Logs an info message to all targets. + + warning(msg, *args, **kwargs): + Logs a warning message to all targets. + + error(msg, *args, **kwargs): + Logs an error message to all targets. + + critical(msg, *args, **kwargs): + Logs a critical message to all targets. + + """ + + targets: list[LogTarget] = [] + + def add_target(self, target: "logging.Logger|callable", level: int = logging.DEBUG): + """ + Adds a new logging target with the specified logging level. + + Args: + target (logging.Logger|callable): The logging target, which can be a logging.Logger instance or a callable. + level (int): The logging level for the target. Defaults to logging.DEBUG. + + """ + self.targets.append(LogTarget(target=target, level=level, logger=isinstance(target, logging.Logger))) + + def has_targets(self): + """ + Checks if there are any logging targets added. + + Returns: + bool: True if there are targets, False otherwise. + + """ + return len(self.targets) > 0 + + def _log(self, level, msg, *args, **kwargs): + """ + Logs a message to all targets that have a logging level less than or equal to the specified level. + + Args: + level (int): The logging level of the message. + msg (str): The message to log. + *args: Additional positional arguments to pass to the logging method. + **kwargs: Additional keyword arguments to pass to the logging method. + + """ + for target in self.targets: + if level < target.level: + continue + + if target.logger: + target.target.log(level, msg, *args, **kwargs) + else: + target.target(level, msg, *args, **kwargs) + + def debug(self, msg, *args, **kwargs): + """ + Logs a debug message to all targets. + + Args: + msg (str): The message to log. + *args: Additional positional arguments to pass to the logging method. + **kwargs: Additional keyword arguments to pass to the logging method. + + """ + self._log(logging.DEBUG, msg, *args, **kwargs) + + def info(self, msg, *args, **kwargs): + """ + Logs an info message to all targets. + + Args: + msg (str): The message to log. + *args: Additional positional arguments to pass to the logging method. + **kwargs: Additional keyword arguments to pass to the logging method. + + """ + self._log(logging.INFO, msg, *args, **kwargs) + + def warning(self, msg, *args, **kwargs): + """ + Logs a warning message to all targets. + + Args: + msg (str): The message to log. + *args: Additional positional arguments to pass to the logging method. + **kwargs: Additional keyword arguments to pass to the logging method. + + """ + self._log(logging.WARNING, msg, *args, **kwargs) + + def error(self, msg, *args, **kwargs): + """ + Logs an error message to all targets. + + Args: + msg (str): The message to log. + *args: Additional positional arguments to pass to the logging method. + **kwargs: Additional keyword arguments to pass to the logging method. + + """ + self._log(logging.ERROR, msg, *args, **kwargs) + + def critical(self, msg, *args, **kwargs): + """ + Logs a critical message to all targets. + + Args: + msg (str): The message to log. + *args: Additional positional arguments to pass to the logging method. + **kwargs: Additional keyword arguments to pass to the logging method. + + """ + self._log(logging.CRITICAL, msg, *args, **kwargs) diff --git a/app/library/Utils.py b/app/library/Utils.py index 1605e605..078ec1c3 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -8,13 +8,14 @@ import re import shlex import socket import uuid -from datetime import UTC, datetime from functools import lru_cache from typing import Any import yt_dlp from yt_dlp.networking.impersonate import ImpersonateTarget +from .LogWrapper import LogWrapper + LOG = logging.getLogger("Utils") IGNORED_KEYS: tuple[str] = ( @@ -137,6 +138,8 @@ def calc_download_path(base_path: str, folder: str | None = None, create_path: b def extract_info(config: dict, url: str, debug: bool = False) -> dict: + log_wrapper = LogWrapper() + params: dict = { "color": "no_color", "extract_flat": True, @@ -159,12 +162,29 @@ def extract_info(config: dict, url: str, debug: bool = False) -> dict: if key in params: params.pop(key) + log_wrapper.add_target(target=logging.getLogger("yt-dlp"), level=logging.DEBUG if debug else logging.WARNING) if debug: params["verbose"] = True - params["logger"] = logging.getLogger("YTPTube-ytdl") else: params["quiet"] = True + if "callback" in params: + # callback can be a function or dict with {level: level, func: target} + if isinstance(params["callback"], dict): + log_wrapper.add_target( + target=params["callback"]["func"], + level=params["callback"]["level"] or logging.ERROR, + ) + else: + log_wrapper.add_target(target=params["callback"], level=logging.ERROR) + del params["callback"] + + if log_wrapper.has_targets(): + if "logger" in params: + log_wrapper.add_target(target=params["logger"], level=logging.DEBUG) + + params["logger"] = log_wrapper + return yt_dlp.YoutubeDL(params=params).extract_info(url, download=False) @@ -198,7 +218,7 @@ def merge_config(config: dict, new_config: dict) -> dict: """ for key in IGNORED_KEYS: if key in new_config: - LOG.error(f"Key '{key}' is not allowed to be manually set.") + LOG.error(f"Key '{key}' is not allowed to be manually set via config.") del new_config[key] conf = merge_dict(new_config, config) @@ -224,7 +244,7 @@ def is_downloaded(archive_file: str, url: str) -> tuple[bool, dict[str | None, s idDict, ) - if not YTDLP_INFO_CLS: + if YTDLP_INFO_CLS is None: YTDLP_INFO_CLS = yt_dlp.YoutubeDL( params={ "color": "no_color", @@ -263,53 +283,6 @@ def is_downloaded(archive_file: str, url: str) -> tuple[bool, dict[str | None, s return (False, idDict) -def json_cookie(cookies: dict[dict[str, any]]) -> str | None: - """ - Converts JSON cookies to Netscape cookies - - Returns None if no cookies are found, otherwise returns a string of cookies in Netscape format. - """ - netscapeCookies = "# Netscape HTTP Cookie File\n# https://curl.haxx.se/docs/http-cookies.html\n# This file was generated by libcurl! Edit at your own risk.\n\n" - hasCookies: bool = False - - for domain in cookies: # noqa: PLC0206 - if not isinstance(cookies[domain], dict): - continue - - for subDomain in cookies[domain]: - if not isinstance(cookies[domain][subDomain], dict): - continue - - for cookie in cookies[domain][subDomain]: - if not isinstance(cookies[domain][subDomain][cookie], dict): - continue - - cookieDict = cookies[domain][subDomain][cookie] - - if 0 == int(cookieDict["expirationDate"]): - cookieDict["expirationDate"] = datetime.now(UTC).timestamp() + (86400 * 1000) - - hasCookies = True - netscapeCookies += ( - "\t".join( - [ - cookieDict["domain"] - if str(cookieDict["domain"]).startswith(".") - else "." + cookieDict["domain"], - "TRUE", - cookieDict["path"], - "TRUE" if cookieDict["secure"] else "FALSE", - str(int(cookieDict["expirationDate"])), - cookieDict["name"], - cookieDict["value"], - ] - ) - + "\n" - ) - - return netscapeCookies if hasCookies else None - - def load_file(file: str, check_type=None) -> tuple[dict | list, bool, str]: """ Load a JSON or JSON5 file and return the contents as a dictionary