import json import logging import time import uuid from dataclasses import dataclass, field from email.utils import formatdate from typing import Any LOG = logging.getLogger("ItemDTO") @dataclass(kw_only=True) class Item: """ This class represents a single download request for data transfer purposes. """ url: str """The URL of the item to be downloaded.""" preset: str = field(default_factory=lambda: Item._default_preset()) """The preset to be used for the download.""" folder: str = "" """The folder to save the download to.""" cookies: str = "" """The cookies to be used for the download.""" template: str = "" """The template to be used for the download.""" cli: str = "" """The yt-dlp cli options to be used for the download.""" extras: dict = field(default_factory=dict) """Extra data to be added to the download.""" def serialize(self) -> dict: return self.__dict__.copy() def json(self) -> str: return json.dumps(self.serialize(), default=lambda o: o.__dict__, sort_keys=True, indent=4) def get(self, key: str, default: Any = None) -> Any: return self.__dict__.get(key, default) def has_extras(self) -> bool: """ Check if the item has any extras data associated with it. Returns: bool: True if the item has extras, False otherwise. """ return self.extras and len(self.extras) > 0 def has_cli(self) -> bool: """ Check if the item has any yt-dlp cli options associated with it. Returns: bool: True if the item has yt-dlp cli options, False otherwise. """ return self.cli and len(self.cli) > 2 def _default_preset() -> str: from .config import Config return Config.get_instance().default_preset def new_with(self, **kwargs) -> "Item": """ Create a new instance of Item with the given parameters. Args: **kwargs: The parameters to be used for creating the new instance. Returns: Item: A new instance of Item with the given parameters. """ return Item.format({**self.serialize(), **kwargs}) def __repr__(self): from .config import Config from .Utils import calc_download_path, strip_newline data = {} for k, v in self.serialize().items(): if not v: continue if k == "cli": data[k] = strip_newline(v) elif k == "extras": data[k] = f"{len(v)} items" elif k == "cookies": data[k] = f"{len(v)}/chars" elif k == "folder": data[k] = calc_download_path(base_path=Config.get_instance().download_path, folder=v) else: data[k] = v items = "".join(f'{k}="{v}", ' for k, v in data.items() if v) return f"Item({items.strip(', ')})" @staticmethod def format(item: dict) -> "Item": """ Format the item to be added to the download queue. Args: item (dict): The item to be formatted. Raises: ValueError: If the url is not provided. ValueError: If the yt-cli command line arguments are not valid. Returns: Item: The formatted item. """ url: str = item.get("url") if not url or not isinstance(url, str): msg = "url param is required." raise ValueError(msg) data = { "url": url, } preset = item.get("preset") if preset and isinstance(preset, str) and preset != Item._default_preset(): from .Presets import Presets if not Presets.get_instance().has(preset): msg = f"Preset '{preset}' does not exist." raise ValueError(msg) data["preset"] = preset if item.get("folder") and isinstance(item.get("folder"), str): data["folder"] = item.get("folder") if item.get("cookies") and isinstance(item.get("cookies"), str): data["cookies"] = item.get("cookies") if item.get("template") and isinstance(item.get("template"), str): data["template"] = item.get("template") extras = item.get("extras") if extras and isinstance(extras, dict) and len(extras) > 0: data["extras"] = extras cli = item.get("cli") if cli and len(cli) > 2: from .Utils import arg_converter try: removed_options = [] arg_converter(args=cli, level=True, removed_options=removed_options) if len(removed_options) > 0: LOG.warning("Removed the following options '%s' for '%s'.", ", ".join(removed_options), url) data["cli"] = cli except Exception as e: msg = f"Failed to parse yt-dlp cli options. {e!s}" raise ValueError(msg) from e return Item(**data) @dataclass(kw_only=True) class ItemDTO: """ ItemDTO is a data transfer object that represents a single item in the queue. It contains all the required information for both the frontend and the backend to process the item. """ _id: str = field(default_factory=lambda: str(uuid.uuid4()), init=False) error: str | None = None id: str title: str url: str quality: str | None = None format: str | None = None preset: str = "default" folder: str download_dir: str | None = None temp_dir: str | None = None status: str | None = None cookies: str | None = None template: str | None = None template_chapter: str | None = None timestamp: float = field(default_factory=lambda: time.time_ns()) is_live: bool | None = None datetime: str = field(default_factory=lambda: str(formatdate(time.time()))) live_in: str | None = None file_size: int | None = None options: dict = field(default_factory=dict) extras: dict = field(default_factory=dict) cli: str = "" # yt-dlp injected fields. tmpfilename: str | None = None filename: str | None = None total_bytes: int | None = None total_bytes_estimate: int | None = None downloaded_bytes: int | None = None msg: str | None = None percent: int | None = None speed: str | None = None eta: str | None = None def serialize(self) -> dict: return self.__dict__.copy() def json(self) -> str: return json.dumps(self.serialize(), default=lambda o: o.__dict__, sort_keys=True, indent=4) def get(self, key: str, default: Any = None) -> Any: return self.__dict__.get(key, default) def get_id(self) -> str: return self._id @staticmethod def removed_fields() -> tuple: """Fields that once existed but are no longer used.""" return ( "thumbnail", "quality", "format", "ytdlp_cookies", "ytdlp_config", "output_template", "output_template_chapter", "config", )