Added info dict condition checker

app/library/DownloadQueue.py

@@ -15,6 +15,7 @@ import yt_dlp
 from aiohttp import web
 
 from .AsyncPool import AsyncPool
+from .conditions import Conditions
 from .config import Config
 from .DataStore import DataStore
 from .Download import Download
@@ -227,7 +228,7 @@ class DownloadQueue(metaclass=Singleton):
         eventType = entry.get("_type") or "video"
 
         if "playlist" == eventType:
-            LOG.info("Processing playlist")
+            LOG.info(f"Processing playlist '{entry.get('id')}: {entry.get('title')}'.")
             entries = entry.get("entries", [])
             playlistCount = int(entry.get("playlist_count", len(entries)))
             results = []
@@ -479,6 +480,13 @@ class DownloadQueue(metaclass=Singleton):
             if not entry:
                 return {"status": "error", "msg": "Unable to extract info." + "\n".join(logs)}
 
+            if not item.requeued:
+                condition = Conditions.get_instance().match(info=entry)
+                if condition is not None:
+                    already.pop()
+                    LOG.info(f"Condition '{condition}' matched for '{item.url}'.")
+                    return await self.add(item=item.new_with(requeued=True, cli=condition.cli),already=already)
+
             end_time = time.perf_counter() - started
             LOG.debug(f"extract_info: for 'URL: {item.url}' is done in '{end_time:.3f}'. Length: '{len(entry)}/keys'.")
         except yt_dlp.utils.ExistingVideoReached as exc:

app/library/Events.py

@@ -128,6 +128,9 @@ class Events:
     PRESETS_UPDATE = "presets_update"
     SCHEDULE_ADD = "schedule_add"
 
+    CONDITIONS_ADD = "conditions_add"
+    CONDITIONS_UPDATE = "conditions_update"
+
     SUBSCRIBED = "subscribed"
     UNSUBSCRIBED = "unsubscribed"
 

app/library/HttpAPI.py

@@ -22,6 +22,7 @@ from yt_dlp.cookies import LenientSimpleCookie
 
 from .cache import Cache
 from .common import Common
+from .conditions import Condition, Conditions
 from .config import Config
 from .DownloadQueue import DownloadQueue
 from .encoder import Encoder
@@ -771,6 +772,94 @@ class HttpAPI(Common):
             dumps=self.encoder.encode,
         )
 
+    @route("GET", "api/conditions")
+    async def conditions(self, _: Request) -> Response:
+        """
+        Get the conditions
+
+        Args:
+            _ (Request): The request object.
+
+        Returns:
+            Response: The response object.
+
+        """
+        return web.json_response(
+            data=Conditions.get_instance().get_all(),
+            status=web.HTTPOk.status_code,
+            dumps=self.encoder.encode,
+        )
+
+    @route("PUT", "api/conditions")
+    async def conditions_add(self, request: Request) -> Response:
+        """
+        Save Conditions.
+
+        Args:
+            request (Request): The request object.
+
+        Returns:
+            Response: The response object
+
+        """
+        data = await request.json()
+
+        if not isinstance(data, list):
+            return web.json_response(
+                {"error": "Invalid request body expecting list with dicts."},
+                status=web.HTTPBadRequest.status_code,
+            )
+
+        items: list = []
+
+        cls = Conditions.get_instance()
+
+        for item in data:
+            if not isinstance(item, dict):
+                return web.json_response(
+                    {"error": "Invalid request body expecting list with dicts."},
+                    status=web.HTTPBadRequest.status_code,
+                )
+
+            if not item.get("name"):
+                return web.json_response(
+                    {"error": "name is required.", "data": item}, status=web.HTTPBadRequest.status_code
+                )
+
+            if not item.get("filter"):
+                return web.json_response(
+                    {"error": "filter is required.", "data": item}, status=web.HTTPBadRequest.status_code
+                )
+
+            if not item.get("cli"):
+                return web.json_response(
+                    {"error": "CLI arguments is required.", "data": item}, status=web.HTTPBadRequest.status_code
+                )
+
+            if not item.get("id", None) or not validate_uuid(item.get("id"), version=4):
+                item["id"] = str(uuid.uuid4())
+
+            try:
+                cls.validate(item)
+            except ValueError as e:
+                return web.json_response(
+                    {"error": f"Failed to validate condition '{item.get('name')}'. '{e!s}'"},
+                    status=web.HTTPBadRequest.status_code,
+                )
+
+            items.append(Condition(**item))
+        try:
+            items = cls.save(items=items).load().get_all()
+        except Exception as e:
+            LOG.exception(e)
+            return web.json_response(
+                {"error": "Failed to save conditions.", "message": str(e)},
+                status=web.HTTPInternalServerError.status_code,
+            )
+
+        await self._notify.emit(Events.CONDITIONS_UPDATE, data=items)
+        return web.json_response(data=items, status=web.HTTPOk.status_code, dumps=self.encoder.encode)
+
     @route("GET", "api/presets")
     async def presets(self, request: Request) -> Response:
         """

app/library/ItemDTO.py

@@ -36,6 +36,9 @@ class Item:
     extras: dict = field(default_factory=dict)
     """Extra data to be added to the download."""
 
+    requeued: bool = False
+    """If the item has been re-queued already via conditions."""
+
     def serialize(self) -> dict:
         return self.__dict__.copy()
 
@@ -155,6 +158,9 @@ class Item:
         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")
         if cli and len(cli) > 2:
             from .Utils import arg_converter

app/library/conditions.py

@@ -0,0 +1,316 @@
+import json
+import logging
+import os
+import uuid
+from dataclasses import dataclass, field
+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
+
+LOG = logging.getLogger("presets")
+
+
+@dataclass(kw_only=True)
+class Condition:
+    id: str = field(default_factory=lambda: str(uuid.uuid4()))
+    """The id of the condition."""
+
+    name: str
+    """The name of the condition."""
+
+    filter: str
+    """The filter to run on info dict."""
+
+    cli: str
+    """If matched append this to the download request and re-queue."""
+
+    def serialize(self) -> dict:
+        return self.__dict__
+
+    def json(self) -> str:
+        return Encoder().encode(self.serialize())
+
+    def get(self, key: str, default: Any = None) -> Any:
+        return self.serialize().get(key, default)
+
+
+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: str | None = None, config: Config | None = None):
+        Conditions._instance = self
+
+        config = config or Config.get_instance()
+
+        self._file: str = file or os.path.join(config.config_path, "conditions.json")
+
+        if os.path.exists(self._file) and "600" != oct(os.stat(self._file).st_mode)[-3:]:
+            try:
+                os.chmod(self._file, 0o600)
+            except Exception:
+                pass
+
+        def event_handler(_, __):
+            msg = "Not implemented"
+            raise Exception(msg)
+
+        EventBus.get_instance().subscribe(Events.CONDITIONS_ADD, event_handler, f"{__class__.__name__}.add")
+
+    @staticmethod
+    def get_instance() -> "Conditions":
+        """
+        Get the instance of the class.
+
+        Returns:
+            Conditions: The instance of the class
+
+        """
+        if not Conditions._instance:
+            Conditions._instance = Conditions()
+
+        return Conditions._instance
+
+    async def on_shutdown(self, _: web.Application):
+        pass
+
+    def attach(self, _: web.Application):
+        """
+        Attach the work to the aiohttp application.
+
+        Args:
+            _ (web.Application): The aiohttp application.
+
+        Returns:
+            None
+
+        """
+        self.load()
+
+    def get_all(self) -> list[Condition]:
+        """Return the items."""
+        return self._items
+
+    def load(self) -> "Conditions":
+        """
+        Load the items.
+
+        Returns:
+            Conditions: The current instance.
+
+        """
+        self.clear()
+
+        if not os.path.exists(self._file) or os.path.getsize(self._file) < 10:
+            return self
+
+        LOG.info(f"Loading '{self._file}'.")
+        try:
+            with open(self._file) as f:
+                items = json.load(f)
+        except Exception as e:
+            LOG.error(f"Failed to parse '{self._file}'. '{e}'.")
+            return self
+
+        if not items or len(items) < 1:
+            LOG.debug(f"No items were found in '{self._file}'.")
+            return self
+
+        need_save = False
+
+        for i, item in enumerate(items):
+            try:
+                if "id" not in item:
+                    item["id"] = str(uuid.uuid4())
+                    need_save = True
+
+                item = Condition(**item)
+
+                self._items.append(item)
+            except Exception as e:
+                LOG.error(f"Failed to parse failure condition at list position '{i}'. '{e!s}'.")
+                continue
+
+        if need_save:
+            LOG.info("Saving failure conditions due to format, or id change.")
+            self.save(self._items)
+
+        return self
+
+    def clear(self) -> "Conditions":
+        """
+        Clear all items
+
+        Returns:
+            conditions: The current instance.
+
+        """
+        if len(self._items) < 1:
+            return self
+
+        self._items.clear()
+
+        return self
+
+    def validate(self, item: Condition | dict) -> bool:
+        """
+        Validate item.
+
+        Args:
+            item (Condition|dict): The item to validate.
+
+        Returns:
+            bool: True if valid, False otherwise.
+
+        """
+        from yt_dlp.utils import match_str
+
+        if not isinstance(item, dict):
+            if not isinstance(item, Condition):
+                msg = f"Unexpected '{type(item).__name__}' item type."
+                raise ValueError(msg)  # noqa: TRY004
+
+            item = item.serialize()
+
+        if not item.get("id"):
+            msg = "No id found."
+            raise ValueError(msg)
+
+        if not item.get("name"):
+            msg = "No name found."
+            raise ValueError(msg)
+
+        if not item.get("filter"):
+            msg = "No filter found."
+            raise ValueError(msg)
+
+        try:
+            match_str(item.get("filter"), {})
+        except Exception as e:
+            msg = f"Invalid filter. '{e!s}'."
+            raise ValueError(msg) from e
+
+        if not item.get("cli"):
+            msg = "No CLI arguments were found."
+            raise ValueError(msg)
+
+        try:
+            arg_converter(args=item.get("cli"))
+        except Exception as e:
+            msg = f"Invalid cli options. '{e!s}'."
+            raise ValueError(msg) from e
+
+        return True
+
+    def save(self, items: list[Condition | dict]) -> "Conditions":
+        """
+        Save the items.
+
+        Args:
+            items (list[Condition]): The items to save.
+
+        Returns:
+            Conditions: The current instance.
+
+        """
+        for i, item in enumerate(items):
+            try:
+                if not isinstance(item, Condition):
+                    item = Condition(**item)
+                    items[i] = item
+            except Exception as e:
+                LOG.error(f"Failed to save '{i}' due to parsing error. '{e!s}'.")
+                continue
+
+            try:
+                self.validate(item)
+            except ValueError as e:
+                LOG.error(f"Failed to validate '{i}: {item.name}'. '{e!s}'.")
+                continue
+
+        try:
+            with open(self._file, "w") as f:
+                json.dump(obj=[item.serialize() for item in items], fp=f, indent=4)
+
+            LOG.info(f"Updated '{self._file}'.")
+        except Exception as e:
+            LOG.error(f"Failed to save '{self._file}'. '{e!s}'.")
+
+        return self
+
+    def has(self, id_or_name: str) -> bool:
+        """
+        Check if the item exists by id or name.
+
+        Args:
+            id_or_name (str): The id or name of the preset.
+
+        Returns:
+            bool: True if the item exists, False otherwise.
+
+        """
+        return self.get(id_or_name) is not None
+
+    def get(self, id_or_name: str) -> Condition | None:
+        """
+        Get the item by id or name.
+
+        Args:
+            id_or_name (str): The id or name of the preset.
+
+        Returns:
+            Condition|None: The item if found, None otherwise.
+
+        """
+        if not id_or_name:
+            return None
+
+        for condition in self.get_all():
+            if id_or_name not in (condition.id, condition.name):
+                continue
+
+            return condition
+
+        return None
+
+    def match(self, info: dict) -> Condition | None:
+        """
+        Check if any condition matches the info dict.
+
+        Args:
+            info (dict): The info dict to check.
+
+        Returns:
+            Condition|None: The condition if found, None otherwise.
+
+        """
+        from yt_dlp.utils import match_str
+
+        for item in self.get_all():
+            if not item.filter:
+                LOG.error(f"Filter is empty for '{item.name}'.")
+                continue
+
+            try:
+                if match_str(item.filter, info):
+                    LOG.debug(f"Matched '{item.name}' with filter '{item.filter}'.")
+
+                    return item
+            except Exception as e:
+                LOG.error(f"Failed to evaluate '{item.name}'. '{e!s}'.")
+                continue
+
+        return None

app/main.py

@@ -9,6 +9,7 @@ from pathlib import Path
 import caribou
 import magic
 from aiohttp import web
+from library.conditions import Conditions
 from library.config import Config
 from library.DownloadQueue import DownloadQueue
 from library.Events import EventBus, Events
@@ -167,6 +168,7 @@ class Main:
         Tasks.get_instance().attach(self._app)
         Presets.get_instance().attach(self._app)
         Notification.get_instance().attach(self._app)
+        Conditions.get_instance().attach(self._app)
 
         EventBus.get_instance().sync_emit(Events.LOADED, data={"app": self._app})