From da8358bad6630c639c06bc6d9be676a149806201 Mon Sep 17 00:00:00 2001 From: arabcoders Date: Fri, 29 Aug 2025 22:11:14 +0300 Subject: [PATCH] Further updates on how we handle archive status --- app/library/DataStore.py | 10 +- app/library/DownloadQueue.py | 23 ++-- app/library/HttpAPI.py | 5 +- app/library/HttpSocket.py | 7 +- app/library/ItemDTO.py | 223 ++++++++++++++++++++++++++++------ app/library/Presets.py | 2 +- app/library/Utils.py | 28 +++-- app/library/YTDLPOpts.py | 3 +- app/routes/api/history.py | 74 ++++++----- ui/app/components/History.vue | 40 +++--- ui/app/types/store.d.ts | 6 + 11 files changed, 284 insertions(+), 137 deletions(-) diff --git a/app/library/DataStore.py b/app/library/DataStore.py index 435d2bcd..c08c9722 100644 --- a/app/library/DataStore.py +++ b/app/library/DataStore.py @@ -121,8 +121,8 @@ class DataStore: return items - def put(self, value: Download) -> Download: - if "error" == value.info.status: + def put(self, value: Download, no_notify: bool = False) -> Download: + if "error" == value.info.status and not no_notify: from app.library.Events import EventBus, Events asyncio.create_task(EventBus.get_instance().emit(Events.ITEM_ERROR, value.info), name="emit_item_error") @@ -181,16 +181,18 @@ class DataStore: except AttributeError: pass + encoded: str = stored.json() + self._connection.execute( sqlStatement.strip(), ( stored._id, str(type), stored.url, - stored.json(), + encoded, str(type), stored.url, - stored.json(), + encoded, datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"), ), ) diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index f25fee31..b0040195 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -24,18 +24,15 @@ from .Presets import Presets from .Scheduler import Scheduler from .Singleton import Singleton from .Utils import ( - archive_read, arg_converter, calc_download_path, dt_delta, extract_info, extract_ytdlp_logs, - get_archive_id, load_cookies, str_to_dt, ytdlp_reject, ) -from .YTDLPOpts import YTDLPOpts if TYPE_CHECKING: from app.library.Presets import Preset @@ -659,22 +656,18 @@ class DownloadQueue(metaclass=Singleton): "level": logging.WARNING, "name": "callback-logger", }, - **YTDLPOpts.get_instance().preset(name=item.preset).add_cli(args=item.cli, from_user=True).get_all(), + **item.get_ytdlp_opts().get_all(), } if yt_conf.get("external_downloader"): LOG.warning(f"Using external downloader '{yt_conf.get('external_downloader')}' for '{item.url}'.") item.extras.update({"external_downloader": True}) - if archive_file := yt_conf.get("download_archive"): - idDict: dict = get_archive_id(item.url) - if (archive_id := idDict.get("archive_id")) and len(archive_read(archive_file, [archive_id])) > 0: - message: str = ( - f"'{idDict.get('id')}': The URL '{item.url}' is already downloaded and recorded in archive." - ) - LOG.error(message) - await self._notify.emit(Events.LOG_INFO, title="Already Downloaded", message=message) - return {"status": "error", "msg": message} + if item.is_archived(): + message: str = f"The URL '{item.url}' is already downloaded and recorded in archive." + LOG.error(message) + await self._notify.emit(Events.LOG_INFO, title="Already Downloaded", message=message) + return {"status": "error", "msg": message} started: float = time.perf_counter() @@ -1016,7 +1009,8 @@ class DownloadQueue(metaclass=Singleton): nMessage = f"Completed '{entry.info.title}' download." _tasks.append(self._notify.emit(Events.ITEM_COMPLETED, data=entry.info, title=nTitle, message=nMessage)) - self.done.put(value=entry) + await asyncio.sleep(0.2) + self.done.put(entry) _tasks.append( self._notify.emit( Events.ITEM_MOVED, @@ -1122,6 +1116,7 @@ class DownloadQueue(metaclass=Singleton): ) ) except Exception as e: + item.info.archive_status() self.done.put(item) LOG.exception(e) LOG.error(f"Failed to retry item '{item_ref}'. {e!s}") diff --git a/app/library/HttpAPI.py b/app/library/HttpAPI.py index 99479fac..b45d5bcb 100644 --- a/app/library/HttpAPI.py +++ b/app/library/HttpAPI.py @@ -97,7 +97,7 @@ class HttpAPI: app.on_shutdown.append(self.on_shutdown) - def add_routes(self, app: web.Application): + def add_routes(self, app: web.Application) -> None: """ Add the routes to the application. @@ -133,7 +133,8 @@ class HttpAPI: elif "" == base_path or not routePath.rstrip("/").startswith(base_path.rstrip("/")): route.path = f"{base_path}/{route.path.lstrip('/')}" - LOG.debug(f"Add ({route.name}) {route.method}: {route.path}.") + if self.config.debug: + LOG.debug(f"Add ({route.name}) {route.method}: {route.path}.") app.router.add_route(route.method, route.path, handler=_handle(route.handler), name=route.name) diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index 61c2105e..28c41297 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -110,9 +110,10 @@ class HttpSocket: load_modules(self.rootPath, self.rootPath / "routes" / "socket") for route in get_routes(RouteType.SOCKET).values(): - LOG.debug( - f"Add ({route.name}) {route.method.value if isinstance(route.method,RouteType) else route.method}: {route.path}." - ) + if self.config.debug: + LOG.debug( + f"Add ({route.name}) {route.method.value if isinstance(route.method, RouteType) else route.method}: {route.path}." + ) self.sio.on(route.path)(HttpSocket._injector(route.handler, route.path)) @staticmethod diff --git a/app/library/ItemDTO.py b/app/library/ItemDTO.py index 749b661c..eca4cab3 100644 --- a/app/library/ItemDTO.py +++ b/app/library/ItemDTO.py @@ -6,9 +6,10 @@ from dataclasses import dataclass, field from email.utils import formatdate from typing import Any -from app.library.Utils import clean_item, get_archive_id +from app.library.Utils import archive_add, archive_delete, archive_read, clean_item, get_archive_id +from app.library.YTDLPOpts import YTDLPOpts -LOG = logging.getLogger("ItemDTO") +LOG: logging.Logger = logging.getLogger("ItemDTO") @dataclass(kw_only=True) @@ -92,29 +93,6 @@ class Item: """ 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 and k not in ("auto_start"): - 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": """ @@ -137,16 +115,14 @@ class Item: msg = "url param is required." raise ValueError(msg) - data = { - "url": url, - } + data: dict[str, str] = {"url": url} - preset = item.get("preset") + preset: str | None = 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." + msg: str = f"Preset '{preset}' does not exist." raise ValueError(msg) data["preset"] = preset @@ -163,19 +139,19 @@ class Item: if "auto_start" in item and isinstance(item.get("auto_start"), bool): data["auto_start"] = bool(item.get("auto_start")) - extras = item.get("extras") + extras: dict | None = item.get("extras") if extras and isinstance(extras, dict) and len(extras) > 0: data["extras"] = extras if item.get("requeued") and isinstance(item.get("requeued"), bool): data["requeued"] = item.get("requeued") - cli = item.get("cli") + cli: str | None = item.get("cli") if cli and len(cli) > 2: from .Utils import arg_converter try: - removed_options = [] + removed_options: list = [] 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) @@ -187,6 +163,55 @@ class Item: return Item(**data) + def get_archive_id(self) -> str | None: + if not self.url: + return None + + idDict: dict = get_archive_id(self.url) + return idDict.get("archive_id") + + def get_ytdlp_opts(self) -> YTDLPOpts: + params: YTDLPOpts = YTDLPOpts.get_instance() + + if self.preset: + params = params.preset(name=self.preset) + + if self.cli: + params = params.add_cli(self.cli, from_user=True) + + return params + + def get_archive_file(self) -> str | None: + return self.get_ytdlp_opts().get_all().get("download_archive") + + def is_archived(self) -> bool: + archive_id: str | None = self.get_archive_id() + archive_file: str | None = self.get_archive_file() + return len(archive_read(archive_file, [archive_id])) > 0 if archive_file and archive_id else False + + def __repr__(self) -> str: + from .config import Config + from .Utils import calc_download_path, strip_newline + + data = {} + for k, v in self.serialize().items(): + if not v and k not in ("auto_start"): + 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(', ')})" + @dataclass(kw_only=True) class ItemDTO: @@ -196,30 +221,55 @@ class ItemDTO: """ _id: str = field(default_factory=lambda: str(uuid.uuid4()), init=False) - + """ Unique identifier for the item. """ error: str | None = None + """ Error message if the item failed. """ id: str + """ The ID of the item yt-dlp """ title: str + """ The title of the item. """ url: str - quality: str | None = None - format: str | None = None + """ The URL of the item. """ preset: str = "default" + """ The preset to be used for the item. """ folder: str + """ The folder to save the item to. """ download_dir: str | None = None + """ The full path to the download directory. """ temp_dir: str | None = None + """ The full path to the temporary directory. """ status: str | None = None + """ The status of the item. """ cookies: str | None = None + """ The cookies to be used for the item. """ template: str | None = None + """ The output template to be used for the item. """ template_chapter: str | None = None + """ The output template for chapters to be used for the item. """ timestamp: float = field(default_factory=lambda: time.time_ns()) + """ The timestamp of the item. """ is_live: bool | None = None + """ If the item is a live stream. """ datetime: str = field(default_factory=lambda: str(formatdate(time.time()))) + """ The datetime of the item. """ live_in: str | None = None + """ The time until the live stream starts. """ file_size: int | None = None + """ The file size of the item. """ options: dict = field(default_factory=dict) + """ The options used for the item. """ extras: dict = field(default_factory=dict) + """ Extra data associated with the item. """ cli: str = "" + """ The command options for yt-dlp to be used for this download. """ auto_start: bool = True + """ If the item should be started automatically. """ + is_archivable: bool | None = None + """ If the item can be archived. """ + is_archived: bool | None = None + """ If the item has been archived. """ + archive_id: str | None = None + """ The archive ID of the item. """ # yt-dlp injected fields. tmpfilename: str | None = None @@ -232,7 +282,13 @@ class ItemDTO: speed: str | None = None eta: str | None = None + _recomputed: bool = False + _archive_file: str | None = None + def serialize(self) -> dict: + if "finished" == self.status and not self._recomputed: + self.archive_status() + item, _ = clean_item(self.__dict__.copy(), ItemDTO.removed_fields()) return item @@ -256,11 +312,95 @@ class ItemDTO: str | None: The archive ID if available, None otherwise. """ - if not self.info: + if self.archive_id: + return self.archive_id + + if not self.url: return None idDict: dict = get_archive_id(self.url) - return idDict.get("archive_id") + self.archive_id = idDict.get("archive_id") + + return self.archive_id + + def get_ytdlp_opts(self) -> YTDLPOpts: + """ + Get the yt-dlp options for the item. + + Returns: + YTDLPOpts: The yt-dlp options for the item. + + """ + params: YTDLPOpts = YTDLPOpts.get_instance() + + if self.preset: + params = params.preset(name=self.preset) + + if self.cli: + params = params.add_cli(self.cli, from_user=True) + + return params + + def archive_status(self, force: bool = False) -> None: + if not force and (self._recomputed or not self.archive_id): + return + + if "finished" == self.status: + self._recomputed = True + + self.is_archivable = bool(self._archive_file) + if not self.is_archivable: + self.is_archived = False + else: + self.is_archived = len(archive_read(self._archive_file, [self.archive_id])) > 0 + + def get_archive_file(self) -> str | None: + """ + Get the archive file path from the yt-dlp options. + + Returns: + str | None: The archive file path if available, None otherwise. + + """ + if self._archive_file or self._recomputed or not self.archive_id: + return self._archive_file + + self._archive_file = self.get_ytdlp_opts().get_all().get("download_archive") + if self._archive_file: + self._archive_file = self._archive_file.strip() + + return self._archive_file + + def archive_add(self) -> bool: + """ + Archive the item by adding its archive ID to the download archive file. + + Returns: + bool: True if the item was archived, False otherwise. + + """ + if self.is_archived or not self.is_archivable or not self.archive_id or not self._archive_file: + return False + + self.is_archived = archive_add(self._archive_file, [self.archive_id]) + + return self.is_archived + + def archive_delete(self) -> bool: + """ + Remove the item's archive ID from the download archive file. + + Returns: + bool: True if the item was removed from the archive, False otherwise. + + """ + if not self.is_archivable or not self.is_archived or not self.archive_id or not self._archive_file: + return False + + archive_delete(self._archive_file, [self.archive_id]) + self.is_archived = False + + return True @staticmethod def removed_fields() -> tuple: @@ -275,4 +415,11 @@ class ItemDTO: "output_template_chapter", "config", "temp_path", + "_recomputed", + "_archive_file", ) + + def __post_init__(self): + self.get_archive_id() + self.get_archive_file() + self.archive_status() diff --git a/app/library/Presets.py b/app/library/Presets.py index ee6c149d..08d582d5 100644 --- a/app/library/Presets.py +++ b/app/library/Presets.py @@ -8,7 +8,6 @@ from typing import Any from aiohttp import web from .config import Config -from .encoder import Encoder from .Events import EventBus, Events from .Singleton import Singleton from .Utils import arg_converter, init_class @@ -92,6 +91,7 @@ class Preset: return self.__dict__ def json(self) -> str: + from .encoder import Encoder return Encoder().encode(self.serialize()) def get(self, key: str, default: Any = None) -> Any: diff --git a/app/library/Utils.py b/app/library/Utils.py index ce8c412e..07891ec5 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -13,7 +13,7 @@ from datetime import UTC, datetime, timedelta from functools import lru_cache from http.cookiejar import MozillaCookieJar from pathlib import Path -from typing import TypeVar +from typing import Any, TypeVar from Crypto.Cipher import AES from yt_dlp.utils import age_restricted, match_str @@ -203,7 +203,7 @@ def extract_info( if no_archive and "download_archive" in params: del params["download_archive"] - data = YTDLP(params=params).extract_info(url, download=False) + data: dict[str, Any] | None = YTDLP(params=params).extract_info(url, download=False) if data and follow_redirect and "_type" in data and "url" == data["_type"]: return extract_info( @@ -262,6 +262,7 @@ def merge_dict(source: dict, destination: dict) -> dict: return destination_copy + def check_id(file: Path) -> bool | str: """ Check if we are able to get an id from the file name. @@ -1185,14 +1186,11 @@ def load_modules(root_path: Path, directory: Path): package_name: str = str(directory.relative_to(root_path).as_posix()).replace("/", ".") - LOG.debug(f"Loading routes from '{directory}' with package name '{package_name}'.") - for _, name, _ in pkgutil.iter_modules([directory]): full_name: str = f"{package_name}.{name}" if name.startswith("_"): continue try: - LOG.debug(f"Loading module '{full_name}'.") importlib.import_module(full_name) except ImportError as e: LOG.error(f"Failed to import module '{full_name}': {e}") @@ -1367,7 +1365,7 @@ def archive_add(file: str | Path, ids: list[str], skip_check: bool = False) -> b skip_check (bool): If True, skip checking for existing IDs. """ - if not ids: + if not ids or not file: return False path: Path = Path(file) if not isinstance(file, Path) else file @@ -1423,12 +1421,15 @@ def archive_read(file: str | Path, ids: list[str] | None = None) -> list[str]: list[str]: List of ids found in the archive file filtered by `ids` if provided. """ + if not file: + return [] + path: Path = Path(file) if not isinstance(file, Path) else file - if not path.exists(): + if not file or not path.exists(): return [] ids_set: set[str] | None = ( - {s.strip() for s in ids if str(s).strip() and len(str(s).strip().split()) < 2} if ids else None + {s.strip() for s in ids if str(s).strip() and len(str(s).strip().split()) >= 2} if ids else None ) found: list[str] = [] @@ -1457,12 +1458,15 @@ def archive_delete(file: str | Path, ids: list[str]) -> bool: bool: True if deletion succeeded (or nothing to do), False on error. """ - path: Path = Path(file) if not isinstance(file, Path) else file - - if not path.exists() or not ids: + if not file or not ids: return False - remove_ids: set[str] = {x.strip() for x in ids if str(x).strip() and len(str(x).strip().split()) < 2} + path: Path = Path(file) if not isinstance(file, Path) else file + + if not path.exists(): + return False + + remove_ids: set[str] = {x.strip() for x in ids if str(x).strip() and len(str(x).strip().split()) >= 2} if not remove_ids: return True diff --git a/app/library/YTDLPOpts.py b/app/library/YTDLPOpts.py index fc3fafe7..cd33bfb1 100644 --- a/app/library/YTDLPOpts.py +++ b/app/library/YTDLPOpts.py @@ -200,7 +200,8 @@ class YTDLPOpts: if data["format"] == "-best": data["format"] = data["format"][1:] - LOG.debug(f"Final yt-dlp options: '{data!s}'.") + if self._config.debug: + LOG.debug(f"Final yt-dlp options: '{data!s}'.") return data diff --git a/app/routes/api/history.py b/app/routes/api/history.py index 57788f2c..2bd99639 100644 --- a/app/routes/api/history.py +++ b/app/routes/api/history.py @@ -14,7 +14,6 @@ from app.library.Events import EventBus, Events from app.library.ItemDTO import Item from app.library.Presets import Preset, Presets from app.library.router import route -from app.library.Utils import archive_add, archive_delete, archive_read, get_archive_id if TYPE_CHECKING: from library.Download import Download @@ -100,14 +99,8 @@ async def item_view(request: Request, queue: DownloadQueue, encoder: Encoder, co if not item.info: return web.json_response(data={"error": "item has no info."}, status=web.HTTPNotFound.status_code) - is_archived = False - params: dict = item.get_ytdlp_opts().get_all() - if (archive_file := params.get("download_archive")) and (archive_id := item.get_archive_id()): - is_archived: bool = len(archive_read(archive_file, [archive_id])) > 0 - - info = { + info: dict = { **item.info.serialize(), - "is_archived": is_archived, "ffprobe": {}, } @@ -267,14 +260,14 @@ async def items_add(request: Request, queue: DownloadQueue, encoder: Encoder) -> @route("POST", r"api/history/{id}/archive", "history.item.archive.add") -async def item_archive_add(request: Request, queue: DownloadQueue) -> Response: +async def item_archive_add(request: Request, queue: DownloadQueue, notify: EventBus) -> Response: """ Manually mark an item as archived. Args: request (Request): The request object. queue (DownloadQueue): The download queue instance. - config (Config): The configuration instance. + notify (EventBus): The event bus instance. Returns: Response: The response object. @@ -290,28 +283,33 @@ async def item_archive_add(request: Request, queue: DownloadQueue) -> Response: except KeyError: return web.json_response(data={"error": f"item '{id}' not found."}, status=web.HTTPNotFound.status_code) - params: dict = item.get_ytdlp_opts().get_all() - - if not (archive_file := params.get("download_archive")): + if not item.info.is_archivable: return web.json_response( data={"error": f"item '{item.info.title}' does not have an archive file."}, status=web.HTTPBadRequest.status_code, ) - idDict = get_archive_id(url=item.info.url) - if not (archive_id := idDict.get("archive_id")): + if not item.info.archive_id: return web.json_response( data={"error": f"item '{item.info.title}' does not have an archive ID."}, status=web.HTTPBadRequest.status_code, ) - if len(archive_read(archive_file, [archive_id])) > 0: + if item.info.is_archived: return web.json_response( data={"error": f"item '{item.info.title}' already archived."}, status=web.HTTPConflict.status_code, ) - archive_add(archive_file, [archive_id]) + if not item.info.archive_add(): + return web.json_response( + data={"error": f"item '{item.info.title}' could not be added to archive."}, + status=web.HTTPInternalServerError.status_code, + ) + + item.info.archive_status(force=True) + queue.done.put(item, no_notify=True) + await notify.emit(Events.ITEM_UPDATED, data=item.info) return web.json_response( data={"message": f"item '{item.info.title}' archived."}, @@ -320,13 +318,14 @@ async def item_archive_add(request: Request, queue: DownloadQueue) -> Response: @route("DELETE", r"api/history/{id}/archive", "history.item.archive.delete") -async def item_archive_delete(request: Request, queue: DownloadQueue) -> Response: +async def item_archive_delete(request: Request, queue: DownloadQueue, notify: EventBus) -> Response: """ Remove an item from the archive. Args: request (Request): The request object. queue (DownloadQueue): The download queue instance. + notify (EventBus): The event bus instance. Returns: Response: The response object. @@ -342,38 +341,35 @@ async def item_archive_delete(request: Request, queue: DownloadQueue) -> Respons except KeyError: return web.json_response(data={"error": f"item '{id}' not found."}, status=web.HTTPNotFound.status_code) - url: str = item.info.url - title: str = f" '{item.info.title}'" - params: dict = item.get_ytdlp_opts().get_all() - - if not (archive_file := params.get("download_archive")): + if not item.info.is_archivable: return web.json_response( - data={"error": "Archive file is not configured."}, + data={"error": f"item '{item.info.title}' does not have an archive file."}, status=web.HTTPBadRequest.status_code, ) - archive_file = Path(archive_file) - - if not archive_file.exists(): + if not item.info.archive_id: return web.json_response( - data={"error": f"Archive file '{archive_file}' does not exist."}, - status=web.HTTPNotFound.status_code, - ) - - idDict = get_archive_id(url=url) - if not (archive_id := idDict.get("archive_id")): - return web.json_response( - data={"error": "item does not have an archive ID."}, + data={"error": f"item '{item.info.title}' does not have an archive ID."}, status=web.HTTPBadRequest.status_code, ) - if not archive_delete(archive_file, [archive_id]): + if not item.info.is_archived: return web.json_response( - data={"error": f"item{title} not found in '{archive_file}' archive."}, - status=web.HTTPNotFound.status_code, + data={"error": f"item '{item.info.title}' not archived."}, + status=web.HTTPConflict.status_code, ) + if not item.info.archive_delete(): + return web.json_response( + data={"error": f"item '{item.info.title}' not found in archive file."}, + status=web.HTTPInternalServerError.status_code, + ) + + item.info.archive_status(force=True) + queue.done.put(item, no_notify=True) + await notify.emit(Events.ITEM_UPDATED, data=item.info) + return web.json_response( - data={"message": f"item{title} removed from '{archive_file}' archive."}, + data={"message": f"item '{item.info.title}' removed from archive."}, status=web.HTTPOk.status_code, ) diff --git a/ui/app/components/History.vue b/ui/app/components/History.vue index f3cb585a..67c3980b 100644 --- a/ui/app/components/History.vue +++ b/ui/app/components/History.vue @@ -203,25 +203,23 @@ Local Information - +
+ + + Add to download form + -