show extract_info logs via notification
This commit is contained in:
ArabCoders
parent
7bdd548358
commit
1934815a41
2 changed files with 182 additions and 51 deletions
158
app/library/LogWrapper.py
Normal file
158
app/library/LogWrapper.py
Normal file
|
|
@ -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)
|
||||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in a new issue