initial work to support adding/deleting presets from api

app/library/EventsSubscriber.py

@@ -41,6 +41,8 @@ class Events:
     TASK_FINISHED = "task_finished"
     TASK_ERROR = "task_error"
 
+    PRESETS_ADD = "preset_add"
+
 
 @dataclass(kw_only=True)
 class Event:

app/library/HttpAPI.py

@@ -29,6 +29,7 @@ from .ffprobe import ffprobe
 from .M3u8 import M3u8
 from .Notifications import Notification, NotificationEvents
 from .Playlist import Playlist
+from .Presets import Presets
 from .Segments import Segments
 from .Subtitle import Subtitle
 from .Tasks import Task, Tasks
@@ -578,6 +579,22 @@ class HttpAPI(Common):
             dumps=self.encoder.encode,
         )
 
+    @route("GET", "api/presets")
+    async def presets(self, _: Request) -> Response:
+        """
+        Get the presets.
+
+        Args:
+            _: The request object.
+
+        Returns:
+            Response: The response object.
+
+        """
+        return web.json_response(
+            data=Presets.get_instance().get_all(), status=web.HTTPOk.status_code, dumps=self.encoder.encode
+        )
+
     @route("GET", "api/tasks")
     async def tasks(self, _: Request) -> Response:
         """

app/library/Presets.py

@@ -0,0 +1,234 @@
+import asyncio
+import json
+import logging
+import os
+from dataclasses import dataclass, field
+from typing import Any
+
+from aiohttp import web
+
+from .config import Config
+from .Emitter import Emitter
+from .encoder import Encoder
+from .EventsSubscriber import Event, Events, EventsSubscriber
+from .Singleton import Singleton
+
+LOG = logging.getLogger("presets")
+
+
+@dataclass(kw_only=True)
+class Preset:
+    name: str
+    """The name of the preset."""
+
+    format: str
+    """The format of the preset."""
+
+    args: dict[str, list[str] | bool] | None = field(default_factory=dict)
+    """The arguments of the preset."""
+
+    postprocessors: list | None = field(default_factory=list)
+    """The postprocessors of the preset."""
+
+    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 Presets(metaclass=Singleton):
+    """
+    This class is used to manage the presets.
+    """
+
+    _presets: list[Preset] = []
+    """The list of presets."""
+
+    _instance = None
+    """The instance of the class."""
+
+    def __init__(
+        self,
+        file: str | None = None,
+        emitter: Emitter | None = None,
+        loop: asyncio.AbstractEventLoop | None = None,
+        config: Config | None = None,
+    ):
+        Presets._instance = self
+
+        config = config or Config.get_instance()
+
+        self._file: str = file or os.path.join(config.config_path, "presets.json")
+        self._loop: asyncio.AbstractEventLoop = loop or asyncio.get_event_loop()
+        self._emitter: Emitter = emitter or Emitter.get_instance()
+
+        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 handle_event(_, e: Event):
+            self.save(**e.data)
+
+        EventsSubscriber.get_instance().subscribe(Events.PRESETS_ADD, f"{__class__}.save", handle_event)
+
+    @staticmethod
+    def get_instance() -> "Presets":
+        """
+        Get the instance of the class.
+
+        Returns:
+            Presets: The instance of the class
+
+        """
+        if not Presets._instance:
+            Presets._instance = Presets()
+
+        return Presets._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[Preset]:
+        """Return the presets."""
+        return self._presets
+
+    def load(self) -> "Presets":
+        """
+        Load the Presets.
+
+        Returns:
+            Presets: 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 presets from '{self._file}'.")
+        try:
+            with open(self._file) as f:
+                presets = json.load(f)
+        except Exception as e:
+            LOG.error(f"Failed to parse presets from '{self._file}'. '{e}'.")
+            return self
+
+        if not presets or len(presets) < 1:
+            LOG.info(f"No presets were defined in '{self._file}'.")
+            return self
+
+        for i, preset in enumerate(presets):
+            try:
+                preset = Preset(**preset)
+                self._presets.append(preset)
+            except Exception as e:
+                LOG.error(f"Failed to parse preset at list position '{i}'. '{e!s}'.")
+                continue
+
+        return self
+
+    def clear(self) -> "Presets":
+        """
+        Clear all presets
+
+        Returns:
+            Presets: The current instance.
+
+        """
+        if len(self._presets) < 1:
+            return self
+
+        self._presets.clear()
+
+        return self
+
+    def validate(self, preset: Preset | dict) -> bool:
+        """
+        Validate the preset.
+
+        Args:
+            preset (Preset|dict): The preset to validate.
+
+        Returns:
+            bool: True if the preset is valid, False otherwise.
+
+        """
+        if not isinstance(preset, dict):
+            if not isinstance(preset, Preset):
+                msg = "Invalid preset type."
+                raise ValueError(msg)  # noqa: TRY004
+
+            preset = preset.serialize()
+
+        if not preset.get("name"):
+            msg = "No name found."
+            raise ValueError(msg)
+
+        if not preset.get("format"):
+            msg = "No format found."
+            raise ValueError(msg)
+
+        if preset.get("args") and not isinstance(preset.get("args"), dict):
+            msg = "Invalid args type. expected dict."
+            raise ValueError(msg)
+
+        if preset.get("postprocessors") and not isinstance(preset.get("postprocessors"), list):
+            msg = "Invalid postprocessors type. expected list."
+            raise ValueError(msg)
+
+        return True
+
+    def save(self, presets: list[Preset | dict]) -> "Presets":
+        """
+        Save the presets.
+
+        Args:
+            presets (list[Preset]): The presets to save.
+
+        Returns:
+            Presets: The current instance.
+
+        """
+        for i, preset in enumerate(presets):
+            try:
+                if not isinstance(preset, Preset):
+                    preset = Preset(**preset)
+                    presets[i] = preset
+            except Exception as e:
+                LOG.error(f"Failed to save preset '{i}' due to parsing error. '{e!s}'.")
+                continue
+
+            try:
+                self.validate(preset)
+            except ValueError as e:
+                LOG.error(f"Failed to validate preset '{i}: {preset.name}'. '{e}'.")
+                continue
+
+        try:
+            with open(self._file, "w") as f:
+                json.dump(obj=[preset.serialize() for preset in presets], fp=f, indent=4)
+
+            LOG.info(f"Presets saved to '{self._file}'.")
+        except Exception as e:
+            LOG.error(f"Failed to save presets to '{self._file}'. '{e!s}'.")
+
+        return self

app/main.py

@@ -17,6 +17,7 @@ from library.HttpAPI import HttpAPI
 from library.HttpSocket import HttpSocket
 from library.Notifications import Notification
 from library.PackageInstaller import PackageInstaller
+from library.Presets import Presets
 from library.Tasks import Tasks
 
 LOG = logging.getLogger("app")
@@ -96,6 +97,7 @@ class Main:
         self._http.attach(self._app)
         self._queue.attach(self._app)
         Tasks.get_instance().attach(self._app)
+        Presets.get_instance().attach(self._app)
 
         def started(_):
             LOG.info("=" * 40)