danvics/ytptube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

refactor: move presets to db model

Browse source
This commit is contained in:
arabcoders 2026-01-25 18:25:07 +03:00
parent fd4e755938
commit 7d6dc7ae76
37 changed files with 1984 additions and 1331 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
app/features/conditions/tests/test_conditions_repository.py
View file

@ -11,15 +11,11 @@ from app.library.sqlite_store import SqliteStore
@pytest_asyncio.fixture
async def repo(tmp_path):
"""Provide a fresh repository instance with initialized database for each test."""
# Reset singletons
async def repo():
ConditionsRepository._reset_singleton()
SqliteStore._reset_singleton()
# Initialize database with temp path
test_db_path = tmp_path / "test.db"
store = SqliteStore(db_path=str(test_db_path))
store = SqliteStore(db_path=":memory:")
await store.get_connection()
# Create repository

2
app/features/notifications/migration.py
View file

@ -7,8 +7,8 @@ from typing import TYPE_CHECKING, Any, cast
from app.features.core.migration import Migration as FeatureMigration
from app.features.notifications.schemas import NotificationEvents
from app.features.presets.service import Presets
from app.library.config import Config
from app.library.Presets import Presets
if TYPE_CHECKING:
from app.features.notifications.repository import NotificationsRepository

3
app/features/notifications/service.py
View file

@ -15,13 +15,14 @@ from app.features.notifications.schemas import (
NotificationRequestHeader,
NotificationRequestType,
)
from app.features.presets.schemas import Preset
from app.features.presets.service import Presets
from app.library.BackgroundWorker import BackgroundWorker
from app.library.config import Config
from app.library.encoder import Encoder
from app.library.Events import Event, EventBus, Events
from app.library.httpx_client import async_client
from app.library.ItemDTO import Item, ItemDTO
from app.library.Presets import Preset, Presets
from app.library.Singleton import Singleton
if TYPE_CHECKING:

1
app/features/presets/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Presets feature module."""

94
app/features/presets/defaults.py Normal file
View file

@ -0,0 +1,94 @@
from datetime import UTC, datetime
DEFAULT_PRESET_UPDATED_AT: datetime = datetime(2026, 1, 25, tzinfo=UTC)
DEFAULT_PRESETS: list[dict[str, object]] = [
{
"name": "default",
"default": True,
"cli": "--socket-timeout 30 --download-archive %(archive_file)s",
"description": "Default preset for yt-dlp. It will download whatever yt-dlp decides is the best quality for the video and audio.",
"folder": "",
"template": "",
"cookies": "",
"priority": 0,
"updated_at": DEFAULT_PRESET_UPDATED_AT,
},
{
"name": "mobile",
"cli": '--socket-timeout 30 --download-archive %(archive_file)s\n-t mp4 --merge-output-format mp4 --add-chapters --remux-video mp4 \n--embed-metadata --embed-thumbnail \n--postprocessor-args "-movflags +faststart"',
"default": True,
"description": "This preset is designed for mobile devices. It will download the best quality video and audio in mp4 format, merge them, and add chapters, metadata, and thumbnail.",
"folder": "",
"template": "",
"cookies": "",
"priority": 0,
"updated_at": DEFAULT_PRESET_UPDATED_AT,
},
{
"name": "1080p",
"cli": "--socket-timeout 30 --download-archive %(archive_file)s\n-S vcodec:h264 --format 'bv[height<=1080][ext=mp4]+ba[ext=m4a]/b[ext=mp4]/b[ext=webm]'",
"default": True,
"description": "Download the best quality video and audio in mp4 format for 1080p resolution.",
"folder": "",
"template": "",
"cookies": "",
"priority": 0,
"updated_at": DEFAULT_PRESET_UPDATED_AT,
},
{
"name": "720p",
"cli": "--socket-timeout 30 --download-archive %(archive_file)s\n-S vcodec:h264 --format 'bv[height<=720][ext=mp4]+ba[ext=m4a]/b[ext=mp4]/b[ext=webm]'",
"default": True,
"description": "Download the best quality video and audio in mp4 format for 720p resolution.",
"folder": "",
"template": "",
"cookies": "",
"priority": 0,
"updated_at": DEFAULT_PRESET_UPDATED_AT,
},
{
"name": "audio_only",
"cli": "--socket-timeout 30 --download-archive %(archive_file)s\n--extract-audio --add-chapters --embed-metadata --embed-thumbnail --format 'bestaudio/best'",
"default": True,
"description": "This preset is designed to download only the audio of the video. It will extract the audio, add chapters, metadata, and thumbnail.",
"folder": "",
"template": "",
"cookies": "",
"priority": 0,
"updated_at": DEFAULT_PRESET_UPDATED_AT,
},
{
"name": "info_reader_plugin",
"description": 'This preset generate specific filename format and metadata to work with yt-dlp info reader plugins for jellyfin/emby and plex and to make play state sync work for WatchState.\n\nThere is one more step you need to do via Other > Terminal if you have it enabled or directly from container shell\n\nyt-dlp -I0 --write-info-json --write-thumbnail --convert-thumbnails jpg --paths /downloads/youtube -o "%(channel|Unknown_title)s [%(channel_id|Unknown_id)s]/%(title).180B [%(channel_id|Unknown_id)s].%(ext)s" -- https://www.youtube.com/channel/UCClfFsWcT3N2I7VTXXyt84A\n\nChange the url to the channel you want to download.\n\nFor more information please visit \nhttps://github.com/arabcoders/watchstate/blob/master/FAQ.md#how-to-get-watchstate-working-with-youtube-contentlibrary',
"folder": "",
"template": "%(channel)s [%(channel_id|Unknown_id)s]/Season %(release_date>%Y,upload_date>%Y|Unknown)s/%(release_date>%Y%m%d,upload_date>%Y%m%d)s - %(title).180B [%(id)s].%(ext)s",
"cli": "--socket-timeout 30 --download-archive %(archive_file)s\n--windows-filenames --write-info-json --embed-info-json \n--convert-thumbnails jpg --write-thumbnail --embed-metadata",
"cookies": "",
"default": True,
"priority": 0,
"updated_at": DEFAULT_PRESET_UPDATED_AT,
},
{
"name": "nfo_maker_tv",
"description": "This preset generate specific filename format and metadata to work with media servers like jellyfin/emby and plex for episodes.",
"folder": "",
"template": "%(channel)s [%(channel_id|Unknown_id)s]/Season %(release_date>%Y,upload_date>%Y|Unknown)s/S%(release_date>%Y,upload_date>%Y)sE%(release_date>%m%d,upload_date>%m%d)s - %(title).100B [%(id)s].%(ext)s",
"cli": '--socket-timeout 30 --download-archive %(archive_file)s\n--use-postprocessor "NFOMakerPP:when=after_move" \n--windows-filenames --convert-thumbnails jpg --write-thumbnail',
"cookies": "",
"default": True,
"priority": 0,
"updated_at": DEFAULT_PRESET_UPDATED_AT,
},
{
"name": "nfo_maker_movie",
"description": "This preset generate specific filename format and metadata to work with media servers like jellyfin/emby and plex for movies.",
"folder": "",
"template": "%(channel)s [%(channel_id|Unknown_id)s]/Season %(release_date>%Y,upload_date>%Y|Unknown)s/S%(release_date>%Y,upload_date>%Y)sE%(release_date>%m%d,upload_date>%m%d)s - %(title).100B [%(id)s].%(ext)s",
"cli": '--socket-timeout 30 --download-archive %(archive_file)s\n--use-postprocessor "NFOMakerPP:when=after_move;mode=movie" \n--windows-filenames --convert-thumbnails jpg --write-thumbnail',
"cookies": "",
"default": True,
"priority": 0,
"updated_at": DEFAULT_PRESET_UPDATED_AT,
},
]

10
app/features/presets/deps.py Normal file
View file

@ -0,0 +1,10 @@
from app.features.presets.repository import PresetsRepository
from app.features.presets.service import Presets
def get_presets_repo() -> PresetsRepository:
return PresetsRepository.get_instance()
def get_presets_service() -> Presets:
return Presets.get_instance()

112
app/features/presets/migration.py Normal file
View file

@ -0,0 +1,112 @@
from __future__ import annotations
import json
import logging
from pathlib import Path
from typing import TYPE_CHECKING, Any
from app.features.core.migration import Migration as FeatureMigration
from app.features.presets.schemas import Preset
from app.features.presets.utils import preset_name
from app.library.config import Config
from app.library.Utils import arg_converter
if TYPE_CHECKING:
from app.features.presets.repository import PresetsRepository
LOG: logging.Logger = logging.getLogger(__name__)
class Migration(FeatureMigration):
name: str = "presets"
def __init__(self, repo: PresetsRepository, config: Config | None = None) -> None:
self._config: Config = config or Config.get_instance()
super().__init__(config=self._config)
self._repo: PresetsRepository = repo
self._source_file: Path = Path(self._config.config_path) / "presets.json"
async def should_run(self) -> bool:
return self._source_file.exists()
async def migrate(self) -> None:
try:
items: list[dict[str, Any]] | None = json.loads(self._source_file.read_text())
except Exception as exc:
LOG.exception("Failed to read %s: %s. Ignoring", self._source_file, exc)
await self._move_file(self._source_file)
return
if not items:
LOG.warning("No presets found in %s; skipping migration.", self._source_file)
await self._move_file(self._source_file)
return
inserted = 0
seen_names: dict[str, int] = {preset_name(preset.name): 1 for preset in await self._repo.list()}
for index, item in enumerate(items):
if not (normalized := self._normalize(item, index, seen_names)):
continue
try:
await self._repo.create(normalized)
inserted += 1
except Exception as exc:
LOG.exception("Failed to insert preset '%s': %s", normalized["name"], exc)
LOG.info("Migrated %s preset(s) from %s.", inserted, self._source_file)
await self._move_file(self._source_file)
def _normalize(self, item: Any, index: int, seen_names: dict[str, int]) -> dict[str, Any] | None:
if not isinstance(item, dict):
LOG.warning("Skipping preset at index %s due to invalid type.", index)
return None
name: str | None = item.get("name")
if not name or not isinstance(name, str):
LOG.warning("Skipping preset at index %s due to missing name.", index)
return None
if not (name := preset_name(name)):
LOG.warning("Skipping preset at index %s due to empty name.", index)
return None
name = self._unique_name(name, seen_names)
description: str = item.get("description") if isinstance(item.get("description"), str) else ""
folder: str = item.get("folder") if isinstance(item.get("folder"), str) else ""
template: str = item.get("template") if isinstance(item.get("template"), str) else ""
cookies: str = item.get("cookies") if isinstance(item.get("cookies"), str) else ""
cli: str = item.get("cli") if isinstance(item.get("cli"), str) else ""
priority = 0
if isinstance(item.get("priority"), int) and not isinstance(item.get("priority"), bool):
priority = int(item.get("priority"))
if not cli and isinstance(item.get("format"), str):
cli = f"--format {item.get('format')}"
if cli:
try:
arg_converter(args=cli, level=True)
except Exception as exc:
LOG.warning("Skipping preset '%s' due to invalid CLI: %s", name, exc)
return None
payload = {
"name": name,
"description": description,
"folder": folder,
"template": template,
"cookies": cookies,
"cli": cli,
"default": False,
"priority": priority,
}
try:
Preset.model_validate(payload)
except Exception as exc:
LOG.warning("Skipping preset '%s' due to validation error: %s", name, exc)
return None
return payload

27
app/features/presets/models.py Normal file
View file

@ -0,0 +1,27 @@
from datetime import datetime
from sqlalchemy import Boolean, Index, Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from app.features.core.models import Base, UTCDateTime, utcnow
class PresetModel(Base):
__tablename__: str = "presets"
__table_args__: tuple[Index, ...] = (
Index("ix_presets_name", "name"),
Index("ix_presets_is_default", "is_default"),
Index("ix_presets_priority", "priority"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(255), nullable=False, unique=True)
description: Mapped[str] = mapped_column(Text, nullable=False, default="")
folder: Mapped[str] = mapped_column(Text, nullable=False, default="")
template: Mapped[str] = mapped_column(Text, nullable=False, default="")
cookies: Mapped[str] = mapped_column(Text, nullable=False, default="")
cli: Mapped[str] = mapped_column(Text, nullable=False, default="")
default: Mapped[bool] = mapped_column("is_default", Boolean, nullable=False, default=False)
priority: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow, nullable=False)
updated_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow, onupdate=utcnow, nullable=False)

207
app/features/presets/repository.py Normal file
View file

@ -0,0 +1,207 @@
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any
from sqlalchemy import func, or_, select
from app.features.core.deps import get_session
from app.features.core.schemas import CEFeature, ConfigEvent
from app.features.presets.migration import Migration
from app.features.presets.models import PresetModel
from app.features.presets.utils import preset_name, seed_defaults
from app.library.config import Config
from app.library.Events import Event, EventBus, Events
from app.library.Services import Services
from app.library.Singleton import Singleton
if TYPE_CHECKING:
from collections.abc import AsyncGenerator
from aiohttp import web
from sqlalchemy.engine.result import Result
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql.elements import ColumnElement
from sqlalchemy.sql.selectable import Select
LOG: logging.Logger = logging.getLogger(__name__)
class PresetsRepository(metaclass=Singleton):
def __init__(self, session: AsyncGenerator[AsyncSession] | None = None) -> None:
self._migrated = False
self.session: AsyncGenerator[AsyncSession] = session or get_session
async def run_migrations(self) -> None:
if self._migrated:
return
self._migrated = True
await Migration(repo=self).run()
def attach(self, _: web.Application) -> None:
async def handle_event(_, __):
await seed_defaults(self)
await self.run_migrations()
await self._update_cache()
await self._ensure_default_preset()
async def handler(e: Event, __):
if isinstance(e.data, ConfigEvent) and CEFeature.PRESETS == e.data.feature:
LOG.debug("Refreshing presets cache due to configuration update.")
await self._update_cache()
Services.get_instance().add(__class__.__name__, self)
EventBus.get_instance().subscribe(
Events.STARTED, handle_event, f"{__class__.__name__}.run_migrations"
).subscribe(Events.CONFIG_UPDATE, handler, "Presets.refresh_cache")
async def _update_cache(self) -> None:
from app.features.presets.service import Presets
await Presets.get_instance().refresh_cache(await self.list())
@staticmethod
def get_instance() -> PresetsRepository:
return PresetsRepository()
async def _ensure_default_preset(self) -> None:
config = Config.get_instance()
if not (default_name := preset_name(config.default_preset)):
config.default_preset = "default"
default_name = config.default_preset
if await self.get_by_name(default_name) is None:
LOG.error("Default preset '%s' not found, using 'default' preset.", default_name)
config.default_preset = "default"
async def list(self) -> list[PresetModel]:
async with self.session() as session:
result: Result[tuple[PresetModel]] = await session.execute(
select(PresetModel).order_by(PresetModel.priority.desc(), PresetModel.name.asc())
)
return list(result.scalars().all())
async def list_paginated(self, page: int, per_page: int) -> tuple[list[PresetModel], int, int, int]:
async with self.session() as session:
total: int = await self.count()
total_pages: int = (total + per_page - 1) // per_page if total > 0 else 1
if page > total_pages and total > 0:
page = total_pages
query: Select[tuple[PresetModel]] = (
select(PresetModel)
.order_by(PresetModel.priority.desc(), PresetModel.name.asc())
.limit(per_page)
.offset((page - 1) * per_page)
)
result: Result[tuple[PresetModel]] = await session.execute(query)
return list(result.scalars().all()), total, page, total_pages
async def count(self) -> int:
async with self.session() as session:
result: Result[tuple[int]] = await session.execute(select(func.count()).select_from(PresetModel))
return int(result.scalar_one())
async def get(self, identifier: int | str) -> PresetModel | None:
async with self.session() as session:
if not identifier:
return None
if isinstance(identifier, int):
clause: ColumnElement[bool] = PresetModel.id == identifier
elif isinstance(identifier, str) and identifier.isdigit():
clause = or_(PresetModel.id == int(identifier), PresetModel.name == identifier)
else:
name = preset_name(identifier) if isinstance(identifier, str) else identifier
if not name:
return None
clause = PresetModel.name == name
result: Result[tuple[PresetModel]] = await session.execute(select(PresetModel).where(clause).limit(1))
return result.scalar_one_or_none()
async def get_by_name(self, name: str, exclude_id: int | None = None) -> PresetModel | None:
async with self.session() as session:
if not (name := preset_name(name)):
return None
query: Select[tuple[PresetModel]] = select(PresetModel).where(PresetModel.name == name)
if None is not exclude_id:
query = query.where(PresetModel.id != exclude_id)
result: Result[tuple[PresetModel]] = await session.execute(query.limit(1))
return result.scalar_one_or_none()
async def create(self, payload: PresetModel | dict) -> PresetModel:
async with self.session() as session:
model: PresetModel = PresetModel(**payload) if isinstance(payload, dict) else payload
if model.id is not None:
model.id = None
model.name = preset_name(model.name)
if await self.get_by_name(name=model.name) is not None:
msg: str = f"Preset with name '{model.name}' already exists."
raise ValueError(msg)
session.add(model)
await session.commit()
await session.refresh(model)
return model
async def update(self, identifier: int | str, payload: dict[str, Any]) -> PresetModel:
async with self.session() as session:
if isinstance(identifier, int):
clause: ColumnElement[bool] = PresetModel.id == identifier
elif isinstance(identifier, str) and identifier.isdigit():
clause = or_(PresetModel.id == int(identifier), PresetModel.name == identifier)
else:
clause = PresetModel.name == identifier
result: Result[tuple[PresetModel]] = await session.execute(select(PresetModel).where(clause).limit(1))
model: PresetModel | None = result.scalar_one_or_none()
if None is model:
msg: str = f"Preset '{identifier}' not found."
raise KeyError(msg)
assert None is not model
payload.pop("id", None)
payload.pop("created_at", None)
payload.pop("updated_at", None)
for key, value in payload.items():
if hasattr(model, key):
setattr(model, key, value)
model.name = preset_name(model.name)
if await self.get_by_name(name=model.name, exclude_id=model.id) is not None:
msg = f"Preset with name '{model.name}' already exists."
raise ValueError(msg)
await session.commit()
await session.refresh(model)
return model
async def delete(self, identifier: int | str) -> PresetModel:
async with self.session() as session:
if isinstance(identifier, int):
clause: ColumnElement[bool] = PresetModel.id == identifier
elif isinstance(identifier, str) and identifier.isdigit():
clause = or_(PresetModel.id == int(identifier), PresetModel.name == identifier)
else:
clause = PresetModel.name == preset_name(identifier)
result: Result[tuple[PresetModel]] = await session.execute(select(PresetModel).where(clause).limit(1))
if not (model := result.scalar_one_or_none()):
msg: str = f"Preset '{identifier}' not found."
raise KeyError(msg)
assert None is not model
await session.delete(model)
await session.commit()
return model

190
app/features/presets/router.py Normal file
View file

@ -0,0 +1,190 @@
from typing import Any
from aiohttp import web
from aiohttp.web import Request, Response
from pydantic import ValidationError
from app.features.core.schemas import CEAction, CEFeature, ConfigEvent, Pagination
from app.features.core.utils import build_pagination, format_validation_errors, normalize_pagination
from app.features.presets.repository import PresetsRepository
from app.features.presets.schemas import Preset, PresetList, PresetPatch
from app.library.encoder import Encoder
from app.library.Events import EventBus, Events
from app.library.router import route
def _model(model: Any) -> Preset:
return Preset.model_validate(model)
def _serialize(model: Any) -> dict:
return _model(model).model_dump()
@route("GET", "api/presets/", "presets")
async def presets_list(request: Request, encoder: Encoder, repo: PresetsRepository) -> Response:
page, per_page = normalize_pagination(request)
items, total, current_page, total_pages = await repo.list_paginated(page, per_page)
return web.json_response(
data=PresetList(
items=[_model(model) for model in items],
pagination=Pagination.model_validate(build_pagination(total, current_page, per_page, total_pages)),
),
status=web.HTTPOk.status_code,
dumps=encoder.encode,
)
@route("GET", r"api/presets/{id:\d+}", "presets_get")
async def presets_get(request: Request, encoder: Encoder, repo: PresetsRepository) -> Response:
if not (identifier := request.match_info.get("id")):
return web.json_response({"error": "ID required"}, status=web.HTTPBadRequest.status_code)
if not (model := await repo.get(identifier)):
return web.json_response({"error": "Preset not found"}, status=web.HTTPNotFound.status_code)
return web.json_response(data=_serialize(model), status=web.HTTPOk.status_code, dumps=encoder.encode)
@route("POST", "api/presets/", "presets_add")
async def presets_add(request: Request, encoder: Encoder, notify: EventBus, repo: PresetsRepository) -> Response:
data = await request.json()
if not isinstance(data, dict):
return web.json_response(
{"error": "Invalid request body expecting dict."},
status=web.HTTPBadRequest.status_code,
)
try:
item: Preset = Preset.model_validate(data)
except ValidationError as exc:
return web.json_response(
data={"error": "Failed to validate preset.", "detail": format_validation_errors(exc)},
status=web.HTTPBadRequest.status_code,
)
payload = item.model_dump(exclude_unset=True)
payload.pop("id", None)
payload["default"] = False
try:
saved = _serialize(await repo.create(payload))
except ValueError as exc:
return web.json_response({"error": str(exc)}, status=web.HTTPBadRequest.status_code)
notify.emit(
Events.CONFIG_UPDATE,
data=ConfigEvent(feature=CEFeature.PRESETS, action=CEAction.CREATE, data=saved),
)
return web.json_response(data=saved, status=web.HTTPOk.status_code, dumps=encoder.encode)
@route("PATCH", r"api/presets/{id:\d+}", "presets_patch")
async def presets_patch(request: Request, encoder: Encoder, notify: EventBus, repo: PresetsRepository) -> Response:
if not (identifier := request.match_info.get("id")):
return web.json_response({"error": "ID required"}, status=web.HTTPBadRequest.status_code)
if not (model := await repo.get(identifier)):
return web.json_response({"error": "Preset not found"}, status=web.HTTPNotFound.status_code)
if model.default:
return web.json_response(
{"error": "Default presets cannot be modified."}, status=web.HTTPBadRequest.status_code
)
data = await request.json()
if not isinstance(data, dict):
return web.json_response(
{"error": "Invalid request body expecting dict."},
status=web.HTTPBadRequest.status_code,
)
try:
validated = PresetPatch.model_validate(data)
except ValidationError as exc:
return web.json_response(
data={"error": "Failed to validate preset.", "detail": format_validation_errors(exc)},
status=web.HTTPBadRequest.status_code,
)
if validated.name and await repo.get_by_name(validated.name, exclude_id=model.id):
return web.json_response(
data={"error": f"Preset with name '{validated.name}' already exists."},
status=web.HTTPConflict.status_code,
)
payload = validated.model_dump(exclude_unset=True)
payload.pop("default", None)
updated = _serialize(await repo.update(model.id, payload))
notify.emit(
Events.CONFIG_UPDATE,
data=ConfigEvent(feature=CEFeature.PRESETS, action=CEAction.UPDATE, data=updated),
)
return web.json_response(data=updated, status=web.HTTPOk.status_code, dumps=encoder.encode)
@route("PUT", r"api/presets/{id:\d+}", "presets_update")
async def presets_update(request: Request, encoder: Encoder, notify: EventBus, repo: PresetsRepository) -> Response:
if not (identifier := request.match_info.get("id")):
return web.json_response({"error": "ID required"}, status=web.HTTPBadRequest.status_code)
if not (model := await repo.get(identifier)):
return web.json_response({"error": "Preset not found"}, status=web.HTTPNotFound.status_code)
if model.default:
return web.json_response(
{"error": "Default presets cannot be modified."}, status=web.HTTPBadRequest.status_code
)
data = await request.json()
if not isinstance(data, dict):
return web.json_response(
{"error": "Invalid request body expecting dict."},
status=web.HTTPBadRequest.status_code,
)
try:
validated = Preset.model_validate(data)
except ValidationError as exc:
return web.json_response(
data={"error": "Failed to validate preset.", "detail": format_validation_errors(exc)},
status=web.HTTPBadRequest.status_code,
)
if validated.name and await repo.get_by_name(validated.name, exclude_id=model.id):
return web.json_response(
data={"error": f"Preset with name '{validated.name}' already exists."},
status=web.HTTPConflict.status_code,
)
payload = validated.model_dump(exclude_unset=True)
payload.pop("default", None)
payload.pop("id", None)
updated = _serialize(await repo.update(model.id, payload))
notify.emit(
Events.CONFIG_UPDATE,
data=ConfigEvent(feature=CEFeature.PRESETS, action=CEAction.UPDATE, data=updated),
)
return web.json_response(data=updated, status=web.HTTPOk.status_code, dumps=encoder.encode)
@route("DELETE", r"api/presets/{id:\d+}", "presets_delete")
async def presets_delete(request: Request, encoder: Encoder, notify: EventBus, repo: PresetsRepository) -> Response:
if not (identifier := request.match_info.get("id")):
return web.json_response({"error": "ID required"}, status=web.HTTPBadRequest.status_code)
if not (model := await repo.get(identifier)):
return web.json_response({"error": "Preset not found"}, status=web.HTTPNotFound.status_code)
if model.default:
return web.json_response({"error": "Default presets cannot be deleted."}, status=web.HTTPBadRequest.status_code)
deleted = _serialize(await repo.delete(model.id))
notify.emit(
Events.CONFIG_UPDATE,
data=ConfigEvent(feature=CEFeature.PRESETS, action=CEAction.DELETE, data=deleted),
)
return web.json_response(data=deleted, status=web.HTTPOk.status_code, dumps=encoder.encode)

95
app/features/presets/schemas.py Normal file
View file

@ -0,0 +1,95 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, field_validator
from app.features.core.schemas import Pagination
from app.features.core.utils import parse_int
from app.features.presets.utils import preset_name
from app.library.config import Config
from app.library.Utils import arg_converter, create_cookies_file
class Preset(BaseModel):
model_config = ConfigDict(str_strip_whitespace=True, from_attributes=True)
id: int | None = None
name: str = Field(min_length=1)
description: str = ""
folder: str = ""
template: str = ""
cookies: str = ""
cli: str = ""
default: bool = False
priority: int = 0
_cookies_file: Path | None = PrivateAttr(default=None)
@field_validator("name", mode="before")
@classmethod
def _normalize_name(cls, value: Any) -> str:
if not isinstance(value, str):
msg: str = "Name must be a string."
raise ValueError(msg)
if not (value := preset_name(value)):
msg = "Name cannot be empty."
raise ValueError(msg)
return value
@field_validator("priority", mode="before")
@classmethod
def _normalize_priority(cls, value: Any) -> int:
return parse_int(value, field="Priority", minimum=0)
@field_validator("cli", mode="before")
@classmethod
def _validate_cli(cls, value: Any) -> str:
if not value:
return ""
if not isinstance(value, str):
msg: str = "CLI must be a string."
raise ValueError(msg)
value = value.strip()
if not value:
return ""
try:
arg_converter(args=value, level=True)
except Exception as e:
msg = f"Invalid command options for yt-dlp: {e!s}"
raise ValueError(msg) from e
return value
def get_cookies_file(self, config: Config | None = None) -> Path | None:
if None is not self._cookies_file:
return self._cookies_file
if not self.cookies or self.id is None:
return None
config = config or Config.get_instance()
self._cookies_file = create_cookies_file(self.cookies, Path(config.config_path) / "cookies" / f"{self.id}.txt")
return self._cookies_file
class PresetPatch(Preset):
model_config = ConfigDict(str_strip_whitespace=True, from_attributes=True)
name: str | None = None
description: str | None = None
folder: str | None = None
template: str | None = None
cookies: str | None = None
cli: str | None = None
priority: int | None = None
class PresetList(BaseModel):
model_config = ConfigDict(from_attributes=True)
items: list[Preset] = Field(default_factory=list)
pagination: Pagination

44
app/features/presets/service.py Normal file
View file

@ -0,0 +1,44 @@
from __future__ import annotations
from app.features.presets.models import PresetModel
from app.features.presets.repository import PresetsRepository
from app.features.presets.schemas import Preset
from app.features.presets.utils import preset_name
from app.library.Services import Services
from app.library.Singleton import Singleton
class Presets(metaclass=Singleton):
def __init__(self, repo: PresetsRepository | None = None) -> None:
self._repo: PresetsRepository = repo or PresetsRepository.get_instance()
self._cache: list[tuple[int, str, Preset]] = []
Services.get_instance().add(__class__.__name__, self)
@staticmethod
def get_instance() -> Presets:
return Presets()
async def refresh_cache(self, items: list[PresetModel]) -> None:
presets = [Preset.model_validate(item) for item in items]
self._cache = [(preset.id if preset.id is not None else -1, preset.name, preset) for preset in presets]
def get_all(self) -> list[Preset]:
return [preset for _, _, preset in self._cache]
def get(self, identifier: int | str) -> Preset | None:
if not identifier:
return None
if isinstance(identifier, str) and not (identifier := preset_name(identifier)):
return None
for id, name, preset in self._cache:
if isinstance(identifier, int) and id == identifier:
return preset
if isinstance(identifier, str) and name == identifier:
return preset
return None
def has(self, identifier: int | str) -> bool:
return self.get(identifier) is not None

71
app/features/presets/tests/test_presets_repository.py Normal file
View file

@ -0,0 +1,71 @@
from __future__ import annotations
import pytest
import pytest_asyncio
from app.features.presets.repository import PresetsRepository
from app.library.sqlite_store import SqliteStore
@pytest_asyncio.fixture
async def repo():
PresetsRepository._reset_singleton()
SqliteStore._reset_singleton()
store = SqliteStore(db_path=":memory:")
await store.get_connection()
repository = PresetsRepository.get_instance()
yield repository
if store._conn:
await store._conn.close()
if store._engine:
await store._engine.dispose()
PresetsRepository._reset_singleton()
SqliteStore._reset_singleton()
class TestPresetsRepository:
@pytest.mark.asyncio
async def test_create_and_get(self, repo):
preset = await repo.create({"name": "Custom", "cli": "--format best"})
assert preset.id is not None, "Should generate ID for new preset"
assert preset.name == "custom", "Should normalize preset name"
assert preset.cli == "--format best", "Should store preset cli"
fetched = await repo.get(preset.id)
assert fetched is not None, "Should fetch preset by id"
assert fetched.name == "custom", "Should return matching preset"
@pytest.mark.asyncio
async def test_create_normalizes_spaces(self, repo):
preset = await repo.create({"name": "My Preset"})
assert preset.name == "my_preset", "Should normalize spaces to underscores"
@pytest.mark.asyncio
async def test_list_orders_by_priority_then_name(self, repo):
await repo.create({"name": "B", "priority": 1})
await repo.create({"name": "A", "priority": 1})
await repo.create({"name": "C", "priority": 2})
items = await repo.list()
assert items[0].name == "c", "Highest priority should be first"
assert items[1].name == "a", "Same priority should sort by name"
assert items[2].name == "b", "Same priority should sort by name"
@pytest.mark.asyncio
async def test_list_paginated(self, repo):
for i in range(5):
await repo.create({"name": f"Item {i}", "priority": i})
items, total, page, total_pages = await repo.list_paginated(page=1, per_page=2)
assert len(items) == 2, "Should return 2 items per page"
assert total == 5, "Should report total count"
assert page == 1, "Should be on page 1"
assert total_pages == 3, "Should have 3 pages total"

55
app/features/presets/utils.py Normal file
View file

@ -0,0 +1,55 @@
import logging
import re
from datetime import UTC, datetime
NAME_WHITESPACE_PATTERN = re.compile(r"\s+")
LOG: logging.Logger = logging.getLogger(__name__)
async def seed_defaults(repo) -> None:
from .defaults import DEFAULT_PRESETS
for preset in DEFAULT_PRESETS:
try:
name = preset.get("name")
if not name or not isinstance(name, str):
LOG.warning("Skipping default preset with invalid name: %s", preset)
continue
if not (name := preset_name(name)):
LOG.warning("Skipping default preset with empty normalized name: %s", name)
continue
payload = {**preset}
payload.pop("id", None)
payload["default"] = True
payload["name"] = name
updated_at_value: datetime | None = None
updated_at_raw = payload.get("updated_at")
if isinstance(updated_at_raw, str):
updated_at_value = datetime.fromisoformat(updated_at_raw)
elif isinstance(updated_at_raw, datetime):
updated_at_value = updated_at_raw
if isinstance(updated_at_value, datetime) and updated_at_value.tzinfo is None:
updated_at_value = updated_at_value.replace(tzinfo=UTC)
payload["updated_at"] = updated_at_value
existing = await repo.get_by_name(name)
if None is existing:
await repo.create(payload)
continue
if existing.updated_at and updated_at_value and existing.updated_at >= updated_at_value:
if False is existing.default:
await repo.update(existing.id, {"default": True})
continue
await repo.update(existing.id, payload)
except Exception as exc:
LOG.exception("Failed to seed default preset '%s': %s", preset.get("name"), exc)
def preset_name(value: str) -> str:
return NAME_WHITESPACE_PATTERN.sub("_", value.strip().lower())

12
app/library/ItemDTO.py
View file

@ -20,7 +20,7 @@ from app.library.Utils import (
from app.library.YTDLPOpts import YTDLPOpts
if TYPE_CHECKING:
from app.library.Presets import Preset
from app.features.presets.schemas import Preset
LOG: logging.Logger = logging.getLogger("ItemDTO")
@ -184,7 +184,7 @@ class Item:
preset: str | None = item.get("preset")
if preset and isinstance(preset, str) and preset != Item._default_preset():
from .Presets import Presets
from app.features.presets.service import Presets
if not Presets.get_instance().has(preset):
msg: str = f"Preset '{preset}' does not exist."
@ -224,15 +224,15 @@ class Item:
return Item(**data)
def get_preset(self) -> "Preset":
def get_preset(self) -> "Preset | None":
"""
Get the preset for the item.
Returns:
Preset: The preset for the item. If not found, None.
Preset | None: The preset for the item. If not found, None.
"""
from .Presets import Presets
from app.features.presets.service import Presets
return Presets.get_instance().get(self.preset if self.preset else self._default_preset())
@ -558,7 +558,7 @@ class ItemDTO:
Preset | None: The preset for the item. If not found, None.
"""
from .Presets import Presets
from app.features.presets.service import Presets
return Presets.get_instance().get(self.preset if self.preset else "default")

412
app/library/Presets.py
View file

@ -1,412 +0,0 @@
import json
import logging
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from aiohttp import web
from .config import Config
from .Singleton import Singleton
from .Utils import arg_converter, create_cookies_file, init_class
LOG = logging.getLogger("presets")
DEFAULT_PRESETS: list[dict[int, dict[str, str | bool]]] = [
{
"id": "3e163c6c-64eb-4448-924f-814b629b3810",
"name": "default",
"default": True,
"cli": "--socket-timeout 30 --download-archive %(archive_file)s",
"description": "Default preset for yt-dlp. It will download whatever yt-dlp decides is the best quality for the video and audio.",
},
{
"id": "441675ed-b739-40f0-a0b0-1ecfcb9dc48d",
"name": "Mobile",
"cli": '--socket-timeout 30 --download-archive %(archive_file)s\n-t mp4 --merge-output-format mp4 --add-chapters --remux-video mp4 \n--embed-metadata --embed-thumbnail \n--postprocessor-args "-movflags +faststart"',
"default": True,
"description": "This preset is designed for mobile devices. It will download the best quality video and audio in mp4 format, merge them, and add chapters, metadata, and thumbnail.",
},
{
"id": "441675ed-b739-40f0-a0b0-1ecfcb9dc48b",
"name": "1080p",
"cli": "--socket-timeout 30 --download-archive %(archive_file)s\n-S vcodec:h264 --format 'bv[height<=1080][ext=mp4]+ba[ext=m4a]/b[ext=mp4]/b[ext=webm]'",
"default": True,
"description": "Download the best quality video and audio in mp4 format for 1080p resolution.",
},
{
"id": "9719fcc3-4cf2-4d88-b1e4-74dff3dba00e",
"name": "720p",
"cli": "--socket-timeout 30 --download-archive %(archive_file)s\n-S vcodec:h264 --format 'bv[height<=720][ext=mp4]+ba[ext=m4a]/b[ext=mp4]/b[ext=webm]'",
"default": True,
"description": "Download the best quality video and audio in mp4 format for 720p resolution.",
},
{
"id": "a6fd4b25-2b3e-458d-bb57-b75e41cc4330",
"name": "Audio Only",
"cli": "--socket-timeout 30 --download-archive %(archive_file)s\n--extract-audio --add-chapters --embed-metadata --embed-thumbnail --format 'bestaudio/best'",
"default": True,
"description": "This preset is designed to download only the audio of the video. It will extract the audio, add chapters, metadata, and thumbnail.",
},
{
"id": "2ade2c28-cad4-4a06-b7eb-2439fdf46f60",
"name": "Info Reader Plugin",
"description": 'This preset generate specific filename format and metadata to work with yt-dlp info reader plugins for jellyfin/emby and plex and to make play state sync work for WatchState.\n\nThere is one more step you need to do via Other > Terminal if you have it enabled or directly from container shell\n\nyt-dlp -I0 --write-info-json --write-thumbnail --convert-thumbnails jpg --paths /downloads/youtube -o "%(channel|Unknown_title)s [%(channel_id|Unknown_id)s]/%(title).180B [%(channel_id|Unknown_id)s].%(ext)s" -- https://www.youtube.com/channel/UCClfFsWcT3N2I7VTXXyt84A\n\nChange the url to the channel you want to download.\n\nFor more information please visit \nhttps://github.com/arabcoders/watchstate/blob/master/FAQ.md#how-to-get-watchstate-working-with-youtube-contentlibrary',
"template": "%(channel)s [%(channel_id|Unknown_id)s]/Season %(release_date>%Y,upload_date>%Y|Unknown)s/%(release_date>%Y%m%d,upload_date>%Y%m%d)s - %(title).180B [%(id)s].%(ext)s",
"cli": "--socket-timeout 30 --download-archive %(archive_file)s\n--windows-filenames --write-info-json --embed-info-json \n--convert-thumbnails jpg --write-thumbnail --embed-metadata",
"default": True,
},
{
"id": "2ade2c28-cad4-4a06-b7eb-2439fdf46f61",
"name": "NFO Maker TV",
"description": "This preset generate specific filename format and metadata to work with media servers like jellyfin/emby and plex for episodes.",
"template": "%(channel)s [%(channel_id|Unknown_id)s]/Season %(release_date>%Y,upload_date>%Y|Unknown)s/S%(release_date>%Y,upload_date>%Y)sE%(release_date>%m%d,upload_date>%m%d)s - %(title).100B [%(id)s].%(ext)s",
"cli": '--socket-timeout 30 --download-archive %(archive_file)s\n--use-postprocessor "NFOMakerPP:when=after_move" \n--windows-filenames --convert-thumbnails jpg --write-thumbnail',
"default": True,
},
{
"id": "2ade2c28-cad4-4a06-b7eb-2439fdf46f61",
"name": "NFO Maker Movie",
"description": "This preset generate specific filename format and metadata to work with media servers like jellyfin/emby and plex for movies.",
"template": "%(channel)s [%(channel_id|Unknown_id)s]/Season %(release_date>%Y,upload_date>%Y|Unknown)s/S%(release_date>%Y,upload_date>%Y)sE%(release_date>%m%d,upload_date>%m%d)s - %(title).100B [%(id)s].%(ext)s",
"cli": '--socket-timeout 30 --download-archive %(archive_file)s\n--use-postprocessor "NFOMakerPP:when=after_move;mode=movie" \n--windows-filenames --convert-thumbnails jpg --write-thumbnail',
"default": True,
},
]
@dataclass(kw_only=True)
class Preset:
id: str = field(default_factory=lambda: str(uuid.uuid4()))
"""The id of the preset."""
name: str
"""The name of the preset."""
description: str = ""
"""The description of the preset."""
folder: str = ""
"""The default download folder to use if non is given."""
template: str = ""
"""The default template to use if non is given."""
cookies: str = ""
"""The default cookies to use if non is given."""
cli: str = ""
"""command options for yt-dlp."""
default: bool = False
"""If True, the preset is a default preset."""
priority: int = 0
"""Priority of the preset."""
_cookies_file: Path | None = field(init=False, default=None)
"""The path to the cookies file."""
def serialize(self) -> dict:
return {k: v for k, v in self.__dict__.items() if not k.startswith("_")}
def json(self) -> str:
from .encoder import Encoder
return Encoder().encode(self.serialize())
def get(self, key: str, default: Any = None) -> Any:
return self.__dict__.get(key, default)
def get_cookies_file(self, config: Config | None = None) -> Path | None:
"""
Get the path to the cookies file.
Args:
config (Config|None): The config instance.
Returns:
Path|None: The path to the cookies file.
"""
if self._cookies_file:
return self._cookies_file
if not self.cookies or not self.id:
return None
if not config:
config = Config.get_instance()
self._cookies_file = create_cookies_file(self.cookies, Path(config.config_path) / "cookies" / f"{self.id}.txt")
return self._cookies_file
class Presets(metaclass=Singleton):
"""
This class is used to manage the presets.
"""
def __init__(self, file: str | Path | None = None, config: Config | None = None):
self._items: list[Preset] = []
"The list of presets."
self._config: Config = None
"The config instance."
self._default: list[Preset] = []
"The list of default presets."
self._config = config or Config.get_instance()
self._file: Path = Path(file) if file else Path(self._config.config_path).joinpath("presets.json")
"The path to the presets file."
if self._file.exists() and "600" != self._file.stat().st_mode:
try:
self._file.chmod(0o600)
except Exception:
pass
for i, preset in enumerate(DEFAULT_PRESETS):
try:
self.validate(preset)
self._default.append(init_class(Preset, preset))
except Exception as e:
LOG.error(f"Failed to parse default preset ':{i}'. '{e!s}'.")
continue
@staticmethod
def get_instance(file: str | Path | None = None, config: Config | None = None) -> "Presets":
"""
Get the instance of the class.
Args:
file (str|Path|None): The path to the presets file.
config (Config|None): The config instance.
Returns:
Presets: The instance of the class
"""
return Presets(file=file, config=config)
async def on_shutdown(self, _: web.Application):
pass
def attach(self, _: web.Application):
"""
Attach the class to the aiohttp application.
Args:
_ (web.Application): The aiohttp application.
Returns:
None
"""
self.load()
if not self.get(self._config.default_preset):
LOG.error(f"Default preset '{self._config.default_preset}' not found, using 'default' preset.")
self._config.default_preset = "default"
def get_all(self) -> list[Preset]:
"""Return the items."""
return sorted(self._default + self._items, key=lambda x: x.priority, reverse=True)
def load(self) -> "Presets":
"""
Load the items.
Returns:
Presets: The current instance.
"""
has: int = len(self._items)
self.clear()
if not self._file.exists() or self._file.stat().st_size < 10:
return self
try:
LOG.info(f"{'Reloading' if has else 'Loading'} '{self._file}'.")
presets: dict = json.loads(self._file.read_text())
except Exception as e:
LOG.error(f"Failed to parse '{self._file}'. '{e}'.")
return self
if not presets or len(presets) < 1:
return self
need_save = False
for i, preset in enumerate(presets):
try:
if "priority" not in preset:
preset["priority"] = 0
need_save = True
if "id" not in preset:
preset["id"] = str(uuid.uuid4())
need_save = True
if preset.get("format"):
if not preset.get("cli"):
preset.update({"cli": f"--format {preset['format']}"})
else:
preset["cli"] = f"--format '{preset['format']}'\n" + preset["cli"]
preset["cli"] = str(preset["cli"]).strip()
preset.pop("format")
need_save = True
preset: Preset = init_class(Preset, preset)
self._items.append(preset)
except Exception as e:
LOG.error(f"Failed to parse '{self._file}:{i}'. '{e!s}'.")
continue
if need_save:
LOG.warning("Saving presets due to schema changes.")
self.save(self._items)
return self
def clear(self) -> "Presets":
"""
Clear all items.
Returns:
Presets: The current instance.
"""
if len(self._items) < 1:
return self
self._items.clear()
return self
def validate(self, item: Preset | dict) -> bool:
"""
Validate the item.
Args:
item (Preset|dict): The item to validate.
Returns:
bool: True if valid
Raises:
ValueError: If the item is not valid.
"""
if not isinstance(item, dict):
if not isinstance(item, Preset):
msg = f"Unexpected '{type(item).__name__}' type was given."
raise ValueError(msg)
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 item.get("cli"):
try:
arg_converter(args=item.get("cli"))
except Exception as e:
msg = f"Invalid command options for yt-dlp. '{e!s}'."
raise ValueError(msg) from e
if item.get("priority") is not None:
priority = item.get("priority")
if not isinstance(priority, int):
msg = "Priority must be an integer."
raise ValueError(msg)
if priority < 0:
msg = "Priority must be >= 0."
raise ValueError(msg)
return True
def save(self, items: list[Preset | dict]) -> "Presets":
"""
Save the items.
Args:
items (list[Preset]): The items to save.
Returns:
Presets: The current instance.
"""
for i, preset in enumerate(items):
try:
if not isinstance(preset, Preset):
preset: Preset = init_class(Preset, preset)
items[i] = preset
except Exception as e:
LOG.error(f"Failed to save item '{i}' due to parsing error. '{e!s}'.")
continue
try:
self.validate(preset)
except ValueError as e:
LOG.error(f"Failed to validate item '{i}: {preset.name}'. '{e}'.")
continue
try:
self._file.write_text(
json.dumps(obj=[preset.serialize() for preset in items if preset.default is False], indent=4)
)
LOG.info(f"Saved '{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 item.
Returns:
bool: True if exists, False otherwise.
"""
return self.get(id_or_name) is not None
def get(self, id_or_name: str) -> Preset | None:
"""
Get the item by id or name.
Args:
id_or_name (str): The id or name of the item.
Returns:
Preset|None: The item if found, None otherwise.
"""
if not id_or_name:
return None
for preset in self.get_all():
if id_or_name not in (preset.id, preset.name):
continue
return preset
return None

14
app/library/YTDLPOpts.py
View file

@ -3,8 +3,9 @@ import shlex
from pathlib import Path
from typing import Any
from app.features.presets.schemas import Preset
from .config import Config
from .Presets import Preset, Presets
from .Utils import REMOVE_KEYS, arg_converter, calc_download_path, create_cookies_file, merge_dict
LOG: logging.Logger = logging.getLogger("YTDLPOpts")
@ -129,9 +130,16 @@ class YTDLPCli:
raise ValueError(msg)
self.item = item
self.preset: Preset = item.get_preset()
preset_name = item.preset if item.preset else self._config.default_preset
self.preset: Preset | None = YTDLPCli._get_presets().get(preset_name)
self._config: Config = config or Config.get_instance()
@staticmethod
def _get_presets():
from app.features.presets.service import Presets
return Presets.get_instance()
def build(self) -> tuple[str, dict]:
"""
Build the CLI command following make_command logic.
@ -306,7 +314,7 @@ class YTDLPOpts:
YTDLPOpts: The instance of the class
"""
preset: Preset | None = Presets.get_instance().get(name)
preset: Preset | None = YTDLPCli._get_presets().get(name)
if not preset:
return self

1
app/library/config.py
View file

@ -273,6 +273,7 @@ class Config(metaclass=Singleton):
"download_info_expires",
"auto_clear_history_days",
"default_pagination",
"extract_info_concurrency",
"flaresolverr_max_timeout",
"flaresolverr_client_timeout",
"flaresolverr_cache_ttl",

4
app/library/downloads/item_adder.py
View file

@ -8,9 +8,9 @@ from typing import TYPE_CHECKING
import yt_dlp.utils
from app.features.conditions.service import Conditions
from app.features.presets.service import Presets
from app.library.Events import Events
from app.library.ItemDTO import ItemDTO
from app.library.Presets import Presets
from app.library.Utils import (
archive_add,
archive_read,
@ -27,8 +27,8 @@ from .playlist_processor import process_playlist
from .video_processor import add_video
if TYPE_CHECKING:
from app.features.presets.schemas import Preset
from app.library.ItemDTO import Item
from app.library.Presets import Preset
from .queue_manager import DownloadQueue

4
app/main.py
View file

@ -17,6 +17,7 @@ from aiohttp import web
from app.features.conditions.service import Conditions
from app.features.dl_fields.service import DLFields
from app.features.notifications.service import Notifications
from app.features.presets.deps import get_presets_repo
from app.features.tasks.definitions.deps import get_task_definitions_repo
from app.features.tasks.service import Tasks
from app.library.BackgroundWorker import BackgroundWorker
@ -26,7 +27,6 @@ from app.library.downloads import DownloadQueue
from app.library.Events import EventBus, Events
from app.library.HttpAPI import HttpAPI
from app.library.HttpSocket import HttpSocket
from app.library.Presets import Presets
from app.library.Scheduler import Scheduler
from app.library.Services import Services
from app.library.sqlite_store import SqliteStore
@ -116,7 +116,7 @@ class Main:
self._socket.attach(self._app)
self._http.attach(self._app)
Presets.get_instance().attach(self._app)
get_presets_repo().attach(self._app)
Tasks.get_instance().attach(self._app)
Notifications.get_instance().attach(self._app)
Conditions.get_instance().attach(self._app)

44
app/migrations/20260125125900_add_presets_table.py Normal file
View file

@ -0,0 +1,44 @@
"""
This module contains a db migration.
Migration Name: add_presets_table
Migration Version: 20260125125900
"""
from sqlalchemy import text
async def upgrade(c):
sql: list[str] = [
"""
CREATE TABLE IF NOT EXISTS presets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
description TEXT NOT NULL DEFAULT '',
folder TEXT NOT NULL DEFAULT '',
template TEXT NOT NULL DEFAULT '',
cookies TEXT NOT NULL DEFAULT '',
cli TEXT NOT NULL DEFAULT '',
is_default INTEGER NOT NULL DEFAULT 0,
priority INTEGER NOT NULL DEFAULT 0,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
)
""",
'CREATE INDEX IF NOT EXISTS "ix_presets_name" ON "presets" ("name")',
'CREATE INDEX IF NOT EXISTS "ix_presets_is_default" ON "presets" ("is_default")',
'CREATE INDEX IF NOT EXISTS "ix_presets_priority" ON "presets" ("priority")',
]
for sql_stmt in sql:
await c.execute(text(sql_stmt))
async def downgrade(c):
sql: list[str] = [
'DROP INDEX IF EXISTS "ix_presets_name"',
'DROP INDEX IF EXISTS "ix_presets_is_default"',
'DROP INDEX IF EXISTS "ix_presets_priority"',
'DROP TABLE IF EXISTS "presets"',
]
for sql_stmt in sql:
await c.execute(text(sql_stmt))

2
app/routes/api/archiver.py
View file

@ -4,8 +4,8 @@ from collections.abc import Iterable
from aiohttp import web
from aiohttp.web import Request, Response
from app.features.presets.service import Presets
from app.library.Archiver import Archiver
from app.library.Presets import Presets
from app.library.router import route
from app.library.YTDLPOpts import YTDLPOpts

3
app/routes/api/history.py
View file

@ -5,13 +5,14 @@ from typing import TYPE_CHECKING, Any
from aiohttp import web
from aiohttp.web import Request, Response
from app.features.presets.schemas import Preset
from app.features.presets.service import Presets
from app.library.config import Config
from app.library.DataStore import StoreType
from app.library.downloads import Download, DownloadQueue
from app.library.encoder import Encoder
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
if TYPE_CHECKING:

109
app/routes/api/presets.py
View file

@ -1,108 +1 @@
import logging
import uuid
from aiohttp import web
from aiohttp.web import Request, Response
from app.features.core.schemas import CEAction, CEFeature, ConfigEvent
from app.library.encoder import Encoder
from app.library.Events import EventBus, Events
from app.library.Presets import Preset, Presets
from app.library.router import route
from app.library.Utils import init_class, validate_uuid
LOG: logging.Logger = logging.getLogger(__name__)
@route("GET", "api/presets/", "presets")
async def presets(request: Request, encoder: Encoder) -> Response:
"""
Get the presets.
Args:
request (Request): The request object.
encoder (Encoder): The encoder instance.
Returns:
Response: The response object.
"""
data: list[Preset] = Presets.get_instance().get_all()
filter_fields: str | None = request.query.get("filter", None)
if filter_fields:
fields: list[str] = [field.strip() for field in filter_fields.split(",")]
data = [{key: value for key, value in preset.serialize().items() if key in fields} for preset in data]
return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=encoder.encode)
@route("PUT", "api/presets/", "presets_add")
async def presets_add(request: Request, encoder: Encoder, notify: EventBus) -> Response:
"""
Add presets.
Args:
request (Request): The request object.
encoder (Encoder): The encoder instance.
notify (EventBus): The event bus instance.
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,
)
presets: list = []
cls = Presets.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("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 preset '{item.get('name')}'. '{e!s}'"},
status=web.HTTPBadRequest.status_code,
)
presets.append(init_class(Preset, item))
try:
presets = cls.save(items=presets).load().get_all()
except Exception as e:
LOG.exception(e)
return web.json_response(
{"error": "Failed to save presets.", "message": str(e)},
status=web.HTTPInternalServerError.status_code,
)
notify.emit(
Events.CONFIG_UPDATE,
data=ConfigEvent(
feature=CEFeature.PRESETS,
action=CEAction.REPLACE,
data=[preset.serialize() for preset in presets],
),
)
return web.json_response(data=presets, status=web.HTTPOk.status_code, dumps=encoder.encode)
import app.features.presets.router # noqa: F401

2
app/routes/api/system.py
View file

@ -12,11 +12,11 @@ from aiohttp.web import Request, Response
from aiohttp.web_runner import GracefulExit
from app.features.dl_fields.service import DLFields
from app.features.presets.service import Presets
from app.library.config import Config
from app.library.downloads import DownloadQueue
from app.library.encoder import Encoder
from app.library.Events import EventBus, Events
from app.library.Presets import Presets
from app.library.router import route
from app.library.UpdateChecker import UpdateChecker
from app.library.Utils import list_folders

2
app/routes/api/yt_dlp.py
View file

@ -7,11 +7,11 @@ from typing import Any
from aiohttp import web
from aiohttp.web import Request, Response
from app.features.presets.service import Presets
from app.library.cache import Cache
from app.library.config import Config
from app.library.encoder import Encoder
from app.library.ItemDTO import Item
from app.library.Presets import Presets
from app.library.router import route
from app.library.Utils import (
REMOVE_KEYS,

18
app/tests/test_itemdto.py
View file

@ -8,7 +8,7 @@ from app.library.ItemDTO import Item, ItemDTO
class TestItemFormatAndBasics:
@patch("app.library.Presets.Presets.get_instance")
@patch("app.features.presets.service.Presets.get_instance")
def test_format_validates_and_normalizes(self, mock_presets_get):
# Preset exists and is not default => allowed
mock_presets_get.return_value.has.return_value = True
@ -42,7 +42,7 @@ class TestItemFormatAndBasics:
assert item.requeued is True
assert item.cli == "--embed-metadata"
@patch("app.library.Presets.Presets.get_instance")
@patch("app.features.presets.service.Presets.get_instance")
def test_format_raises_for_missing_url_and_invalid_preset(self, mock_presets_get):
# Missing url
with pytest.raises(ValueError, match="url param is required"):
@ -282,14 +282,14 @@ class TestItemDTO:
def test_get_preset_returns_preset_instance(self):
"""Test ItemDTO.get_preset returns the Preset instance."""
from app.library.Presets import Preset
from app.features.presets.schemas import Preset
mock_preset = Preset(id="test-id", name="test-preset", cli="--test")
mock_preset = Preset(id=1, name="test-preset", cli="--format best")
with patch.object(ItemDTO, "__post_init__", lambda _: None):
dto = ItemDTO(id="vid", title="t", url="u", folder="f", preset="test-preset")
with patch("app.library.Presets.Presets.get_instance") as mock_presets:
with patch("app.features.presets.service.Presets.get_instance") as mock_presets:
mock_presets.return_value.get.return_value = mock_preset
result = dto.get_preset()
@ -300,14 +300,14 @@ class TestItemDTO:
def test_get_preset_uses_default_when_no_preset_set(self):
"""Test ItemDTO.get_preset uses 'default' when preset is empty."""
from app.library.Presets import Preset
from app.features.presets.schemas import Preset
mock_preset = Preset(id="default-id", name="default", cli="--default")
mock_preset = Preset(id=2, name="default", cli="--format best")
with patch.object(ItemDTO, "__post_init__", lambda _: None):
dto = ItemDTO(id="vid", title="t", url="u", folder="f", preset="")
with patch("app.library.Presets.Presets.get_instance") as mock_presets:
with patch("app.features.presets.service.Presets.get_instance") as mock_presets:
mock_presets.return_value.get.return_value = mock_preset
result = dto.get_preset()
@ -320,7 +320,7 @@ class TestItemDTO:
with patch.object(ItemDTO, "__post_init__", lambda _: None):
dto = ItemDTO(id="vid", title="t", url="u", folder="f", preset="nonexistent")
with patch("app.library.Presets.Presets.get_instance") as mock_presets:
with patch("app.features.presets.service.Presets.get_instance") as mock_presets:
mock_presets.return_value.get.return_value = None
result = dto.get_preset()

638
app/tests/test_presets.py
View file

@ -1,638 +0,0 @@
import json
import tempfile
import uuid
from pathlib import Path
from unittest.mock import Mock, patch
import pytest
from app.library.Presets import DEFAULT_PRESETS, Preset, Presets
class TestPreset:
"""Test the Preset dataclass."""
def test_preset_creation_with_defaults(self):
"""Test creating a preset with default values."""
preset = Preset(name="test_preset")
assert preset.name == "test_preset"
assert preset.description == ""
assert preset.folder == ""
assert preset.template == ""
assert preset.cookies == ""
assert preset.cli == ""
assert preset.default is False
assert len(preset.id) == 36, "ID should be auto-generated UUID" # UUID4 string length
assert "-" in preset.id
def test_preset_creation_with_all_fields(self):
"""Test creating a preset with all fields specified."""
preset_id = str(uuid.uuid4())
preset = Preset(
id=preset_id,
name="test_preset",
description="Test description",
folder="test_folder",
template="test_template",
cookies="test_cookies",
cli="--test-option",
default=True,
priority=10,
)
assert preset.id == preset_id
assert preset.name == "test_preset"
assert preset.description == "Test description"
assert preset.folder == "test_folder"
assert preset.template == "test_template"
assert preset.cookies == "test_cookies"
assert preset.cli == "--test-option"
assert preset.priority == 10
assert preset.default is True
def test_preset_serialize(self):
"""Test preset serialization to dictionary."""
preset = Preset(name="test_preset", description="Test description", cli="--test-option")
serialized = preset.serialize()
assert isinstance(serialized, dict)
assert serialized["name"] == "test_preset"
assert serialized["description"] == "Test description"
assert serialized["cli"] == "--test-option"
assert "id" in serialized
def test_preset_json(self):
"""Test preset JSON serialization."""
preset = Preset(name="test_preset", cli="--test-option")
json_str = preset.json()
assert isinstance(json_str, str)
# Should be valid JSON
parsed = json.loads(json_str)
assert parsed["name"] == "test_preset"
assert parsed["cli"] == "--test-option"
def test_preset_get_method(self):
"""Test preset get method for accessing fields."""
preset = Preset(name="test_preset", description="Test description")
assert preset.get("name") == "test_preset"
assert preset.get("description") == "Test description"
assert preset.get("nonexistent") is None
assert preset.get("nonexistent", "default_value") == "default_value"
def test_preset_id_generation(self):
"""Test that each preset gets a unique ID."""
preset1 = Preset(name="preset1")
preset2 = Preset(name="preset2")
assert preset1.id != preset2.id
assert len(preset1.id) == 36
assert len(preset2.id) == 36
class TestPresets:
"""Test the Presets singleton manager."""
def setup_method(self):
"""Set up test environment before each test."""
# Reset singleton completely
Presets._reset_singleton()
@patch("app.library.Presets.Config")
def test_presets_singleton(self, mock_config):
"""Test that Presets follows singleton pattern."""
mock_config.get_instance.return_value = Mock(config_path="/tmp", default_preset="default")
instance1 = Presets.get_instance()
instance2 = Presets.get_instance()
assert instance1 is instance2
assert isinstance(instance1, Presets)
@patch("app.library.Presets.Config")
def test_presets_initialization(self, mock_config):
"""Test Presets initialization with default presets."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
# Should have default presets loaded
default_presets = presets._default
assert len(default_presets) > 0
# Check that default presets are Preset instances
for preset in default_presets:
assert isinstance(preset, Preset)
assert preset.default is True
@patch("app.library.Presets.Config")
def test_presets_load_empty_file(self, mock_config):
"""Test loading presets from empty or non-existent file."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
presets.clear() # Ensure we start with empty items
result = presets.load()
assert result is presets
assert len(presets._items) == 0
assert len(presets._default) > 0 # Should still have default presets
@patch("app.library.Presets.Config")
@patch("app.library.Presets.arg_converter")
def test_presets_load_valid_file(self, mock_arg_converter, mock_config):
"""Test loading presets from valid JSON file."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
mock_arg_converter.return_value = [] # Mock successful CLI parsing
test_presets = [
{"id": str(uuid.uuid4()), "name": "test_preset_1", "description": "Test preset 1", "cli": "--format best"},
{"id": str(uuid.uuid4()), "name": "test_preset_2", "description": "Test preset 2", "cli": "--format worst"},
]
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets_file.write_text(json.dumps(test_presets, indent=2))
presets = Presets(file=presets_file, config=mock_config_instance)
presets._items.clear() # Ensure we start with empty items
result = presets.load()
assert result is presets
assert len(presets._items) == 2
assert presets._items[0].name == "test_preset_1"
assert presets._items[1].name == "test_preset_2"
@patch("app.library.Presets.Config")
def test_presets_load_invalid_json(self, mock_config):
"""Test loading presets from invalid JSON file."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets_file.write_text("invalid json content")
presets = Presets(file=presets_file, config=mock_config_instance)
# Should handle invalid JSON gracefully
result = presets.load()
assert result is presets
assert len(presets._items) == 0
@patch("app.library.Presets.Config")
@patch("app.library.Presets.arg_converter")
def test_presets_load_with_format_migration(self, mock_arg_converter, mock_config):
"""Test loading presets with old format that needs migration."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
mock_arg_converter.return_value = [] # Mock successful CLI parsing
# Old format preset with 'format' field instead of 'cli'
old_preset = {"name": "old_preset", "format": "best[height<=720]", "description": "Old format preset"}
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets_file.write_text(json.dumps([old_preset]))
presets = Presets(file=presets_file, config=mock_config_instance)
result = presets.load()
assert result is presets
assert len(presets._items) == 1
loaded_preset = presets._items[0]
assert loaded_preset.name == "old_preset"
assert "best[height<=720]" in loaded_preset.cli, "Should have migrated format to cli"
assert "--format" in loaded_preset.cli
# Should have generated ID
assert loaded_preset.id is not None
@patch("app.library.Presets.Config")
def test_presets_validate_valid_preset(self, mock_config):
"""Test validating a valid preset."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
valid_preset = {"id": str(uuid.uuid4()), "name": "valid_preset", "cli": "--format best"}
result = presets.validate(valid_preset)
assert result is True
@patch("app.library.Presets.Config")
def test_presets_validate_invalid_preset(self, mock_config):
"""Test validating invalid presets."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
# Test missing ID
with pytest.raises(ValueError, match="No id found"):
presets.validate({"name": "test"})
# Test missing name
with pytest.raises(ValueError, match="No name found"):
presets.validate({"id": str(uuid.uuid4())})
# Test wrong type
with pytest.raises(ValueError, match="Unexpected"):
presets.validate("invalid_type")
@patch("app.library.Presets.Config")
@patch("app.library.Presets.arg_converter")
def test_presets_save(self, mock_arg_converter, mock_config):
"""Test saving presets to file."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
mock_arg_converter.return_value = [] # Mock successful CLI parsing
test_presets = [
Preset(name="test1", cli="--format best"),
Preset(name="test2", cli="--format worst", default=True), # This should be excluded from save
]
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
result = presets.save(test_presets)
assert result is presets
assert presets_file.exists()
# Should only save non-default presets
saved_data = json.loads(presets_file.read_text())
assert len(saved_data) == 1
assert saved_data[0]["name"] == "test1"
@patch("app.library.Presets.Config")
def test_presets_get_all(self, mock_config):
"""Test getting all presets (default + custom)."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
# Add a custom preset
custom_preset = Preset(name="custom", cli="--custom")
presets._items.append(custom_preset)
all_presets = presets.get_all()
assert len(all_presets) > 1, "Should include both default and custom presets"
assert any(p.name == "custom" for p in all_presets)
assert any(p.default is True for p in all_presets)
@patch("app.library.Presets.Config")
def test_presets_get_by_id(self, mock_config):
"""Test getting preset by ID."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
test_id = str(uuid.uuid4())
test_preset = Preset(id=test_id, name="test_preset")
presets._items.append(test_preset)
found_preset = presets.get(test_id)
assert found_preset is not None
assert found_preset.id == test_id
assert found_preset.name == "test_preset"
@patch("app.library.Presets.Config")
def test_presets_get_by_name(self, mock_config):
"""Test getting preset by name."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
test_preset = Preset(name="test_preset", cli="--test")
presets._items.append(test_preset)
found_preset = presets.get("test_preset")
assert found_preset is not None
assert found_preset.name == "test_preset"
assert found_preset.cli == "--test"
@patch("app.library.Presets.Config")
def test_presets_get_nonexistent(self, mock_config):
"""Test getting non-existent preset returns None."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
found_preset = presets.get("nonexistent")
assert found_preset is None
@patch("app.library.Presets.Config")
def test_presets_get_empty_parameter(self, mock_config):
"""Test getting preset with empty string returns None."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
assert presets.get("") is None
assert presets.get(None) is None
@patch("app.library.Presets.Config")
def test_presets_has_method(self, mock_config):
"""Test has method for checking preset existence."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
test_preset = Preset(name="existing_preset")
presets._items.append(test_preset)
assert presets.has("existing_preset") is True
assert presets.has(test_preset.id) is True
assert presets.has("nonexistent") is False
@patch("app.library.Presets.Config")
def test_presets_clear(self, mock_config):
"""Test clearing all custom presets."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
# Add some custom presets
presets._items.append(Preset(name="custom1"))
presets._items.append(Preset(name="custom2"))
assert len(presets._items) == 2
result = presets.clear()
assert result is presets
assert len(presets._items) == 0
@patch("app.library.Presets.Config")
def test_presets_file_permissions(self, mock_config):
"""Test that presets file gets correct permissions."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
# Create file with different permissions
presets_file.write_text("{}")
presets_file.chmod(0o644)
# Creating Presets instance should attempt to fix permissions
presets = Presets(file=presets_file, config=mock_config_instance)
# Note: The actual chmod might fail in test environment,
# but we're testing that the code attempts it
assert presets is not None, "but we're testing that the code attempts it"
def test_default_presets_validation(self):
"""Test that all default presets are valid."""
# This test verifies that the DEFAULT_PRESETS constant contains valid presets
for i, preset_data in enumerate(DEFAULT_PRESETS):
assert "id" in preset_data, f"Default preset {i} missing id"
assert "name" in preset_data, f"Default preset {i} missing name"
assert "default" in preset_data, f"Default preset {i} missing default flag"
assert preset_data["default"] is True, f"Default preset {i} should have default=True"
# Should be able to create Preset instance
preset = Preset(
id=preset_data["id"],
name=preset_data["name"],
description=preset_data.get("description", ""),
folder=preset_data.get("folder", ""),
template=preset_data.get("template", ""),
cli=preset_data.get("cli", ""),
default=preset_data["default"],
)
assert isinstance(preset, Preset)
@patch("app.library.Presets.Config")
def test_presets_attach_method(self, mock_config):
"""Test the attach method for aiohttp integration."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
mock_app = Mock()
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
# Mock the get method to return a preset for default_preset check
with patch.object(presets, "get", return_value=Mock()) as mock_get:
presets.attach(mock_app)
mock_get.assert_called_once_with("default")
@patch("app.library.Presets.Config")
def test_presets_attach_default_preset_not_found(self, mock_config):
"""Test attach method when default preset is not found."""
mock_config_instance = Mock(config_path="/tmp", default_preset="nonexistent")
mock_config.get_instance.return_value = mock_config_instance
mock_app = Mock()
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
with patch.object(presets, "get", return_value=None):
# The actual config instance stored in presets should be checked
presets.attach(mock_app)
assert presets._config.default_preset == "default", "Should have reset default_preset to 'default'"
@pytest.mark.asyncio
@patch("app.library.Presets.Config")
async def test_presets_on_shutdown(self, mock_config):
"""Test the on_shutdown method."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
mock_app = Mock()
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
# Should not raise an exception
await presets.on_shutdown(mock_app)
@patch("app.library.Presets.Config")
@patch("app.library.Presets.LOG")
@patch("app.library.Presets.arg_converter")
def test_presets_load_invalid_preset_in_file(self, mock_arg_converter, mock_log, mock_config):
"""Test loading file with some invalid presets."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
mock_arg_converter.return_value = [] # Mock successful CLI parsing
# Mix of valid and invalid presets
test_presets = [
{"id": str(uuid.uuid4()), "name": "valid_preset", "cli": "--format best"},
{
# Missing required fields
"description": "Invalid preset"
},
]
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets_file.write_text(json.dumps(test_presets))
# Clear any existing items first
presets = Presets(file=presets_file, config=mock_config_instance)
presets.load()
assert len(presets._items) == 1, "Should only load the valid preset"
assert presets._items[0].name == "valid_preset", f"Expected 'valid_preset', got '{presets._items}'"
# Should have logged an error for the invalid preset
mock_log.error.assert_called()
@patch("app.library.Presets.Config")
@patch("app.library.Presets.LOG")
def test_presets_save_with_invalid_preset(self, mock_log, mock_config):
"""Test saving presets with some invalid ones."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
# Create preset with invalid CLI that will fail validation
valid_preset = Preset(name="valid", cli="--valid-option")
# Create invalid preset data
invalid_preset_data = {"name": "invalid"} # Missing ID
presets_to_save = [valid_preset, invalid_preset_data]
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
presets.save(presets_to_save)
# Should have logged errors for invalid presets
mock_log.error.assert_called()
@patch("app.library.Presets.Config")
def test_preset_priority_validation_integer(self, mock_config):
"""Test that priority must be an integer."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
# Test with non-integer priority
invalid_preset = {"id": str(uuid.uuid4()), "name": "test", "priority": "not_an_int"}
with pytest.raises(ValueError, match="Priority must be an integer"):
presets.validate(invalid_preset)
@patch("app.library.Presets.Config")
def test_preset_priority_validation_negative(self, mock_config):
"""Test that priority must be >= 0."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
# Test with negative priority
invalid_preset = {"id": str(uuid.uuid4()), "name": "test", "priority": -5}
with pytest.raises(ValueError, match="Priority must be >= 0"):
presets.validate(invalid_preset)
@patch("app.library.Presets.Config")
def test_preset_priority_sorting(self, mock_config):
"""Test that presets are sorted by priority in descending order."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
presets = Presets(file=presets_file, config=mock_config_instance)
# Create presets with different priorities
preset1 = Preset(name="low", priority=1)
preset2 = Preset(name="high", priority=10)
preset3 = Preset(name="medium", priority=5)
presets.save([preset1, preset2, preset3]).load()
all_presets = presets.get_all()
non_default = [p for p in all_presets if not p.default]
assert non_default[0].name == "high", "Should be sorted by priority descending"
assert non_default[0].priority == 10
assert non_default[1].name == "medium"
assert non_default[1].priority == 5
assert non_default[2].name == "low"
assert non_default[2].priority == 1
@patch("app.library.Presets.Config")
def test_preset_priority_default_zero(self, mock_config):
"""Test that priority defaults to 0 if not specified."""
preset = Preset(name="test")
assert preset.priority == 0
@patch("app.library.Presets.Config")
def test_preset_priority_migration(self, mock_config):
"""Test that old presets without priority get it added on load."""
mock_config_instance = Mock(config_path="/tmp", default_preset="default")
mock_config.get_instance.return_value = mock_config_instance
with tempfile.TemporaryDirectory() as temp_dir:
presets_file = Path(temp_dir) / "presets.json"
# Create a preset file without priority field
old_preset_data = [{"id": str(uuid.uuid4()), "name": "old_preset", "cli": ""}]
presets_file.write_text(json.dumps(old_preset_data))
# Load presets - should migrate by adding priority
presets = Presets(file=presets_file, config=mock_config_instance)
presets.load()
# Check that priority was added
loaded_data = json.loads(presets_file.read_text())
assert "priority" in loaded_data[0]
assert loaded_data[0]["priority"] == 0

42
app/tests/test_ytdlpopts.py
View file

@ -4,7 +4,7 @@ from unittest.mock import Mock, patch
import pytest
from app.library.Presets import Preset
from app.features.presets.schemas import Preset
from app.library.YTDLPOpts import YTDLPOpts
@ -116,7 +116,7 @@ class TestYTDLPOpts:
"""Test applying a valid preset."""
with (
patch("app.library.YTDLPOpts.Config") as mock_config,
patch("app.library.YTDLPOpts.Presets") as mock_presets,
patch("app.features.presets.service.Presets") as mock_presets,
patch("app.library.YTDLPOpts.arg_converter") as mock_converter,
):
# Mock config
@ -155,7 +155,7 @@ class TestYTDLPOpts:
def test_preset_with_nonexistent_preset(self):
"""Test applying a nonexistent preset returns self."""
with patch("app.library.YTDLPOpts.Config"), patch("app.library.YTDLPOpts.Presets") as mock_presets:
with patch("app.library.YTDLPOpts.Config"), patch("app.features.presets.service.Presets") as mock_presets:
mock_presets_instance = Mock()
mock_presets_instance.get.return_value = None
mock_presets.get_instance.return_value = mock_presets_instance
@ -171,7 +171,7 @@ class TestYTDLPOpts:
"""Test that preset with cookies creates cookie file."""
with (
patch("app.library.YTDLPOpts.Config") as mock_config,
patch("app.library.YTDLPOpts.Presets") as mock_presets,
patch("app.features.presets.service.Presets") as mock_presets,
):
# Mock config
mock_config_instance = Mock()
@ -209,7 +209,7 @@ class TestYTDLPOpts:
"""Test that preset with invalid CLI raises ValueError."""
with (
patch("app.library.YTDLPOpts.Config"),
patch("app.library.YTDLPOpts.Presets") as mock_presets,
patch("app.features.presets.service.Presets") as mock_presets,
patch("app.library.YTDLPOpts.arg_converter") as mock_converter,
):
mock_preset = Mock(spec=Preset)
@ -401,7 +401,7 @@ class TestYTDLPOpts:
with (
patch("app.library.YTDLPOpts.Config"),
patch("app.library.YTDLPOpts.arg_converter"),
patch("app.library.YTDLPOpts.Presets") as mock_presets,
patch("app.features.presets.service.Presets") as mock_presets,
):
mock_presets_instance = Mock()
mock_presets_instance.get.return_value = None
@ -441,7 +441,7 @@ class TestYTDLPOpts:
"""Test error handling when cookie loading fails."""
with (
patch("app.library.YTDLPOpts.Config") as mock_config,
patch("app.library.YTDLPOpts.Presets") as mock_presets,
patch("app.features.presets.service.Presets") as mock_presets,
patch("app.library.YTDLPOpts.LOG") as mock_log,
):
# Mock config
@ -741,7 +741,7 @@ class TestARGSMerger:
class TestYTDLPCli:
"""Test the YTDLPCli class."""
@patch("app.library.Presets.Presets")
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_constructor_with_valid_item(self, mock_config, mock_presets):
"""Test YTDLPCli constructor with valid Item."""
@ -770,7 +770,7 @@ class TestYTDLPCli:
with pytest.raises(ValueError, match="Expected Item instance"):
YTDLPCli(item="not an item") # type: ignore
@patch("app.library.Presets.Presets")
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_build_with_user_fields_only(self, mock_config, mock_presets):
"""Test build with only user-provided fields."""
@ -805,12 +805,12 @@ class TestYTDLPCli:
assert info["merged"]["save_path"] == "/downloads/myfolder"
assert "--format best" in command
@patch("app.library.Presets.Presets")
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_build_with_preset_fields_fallback(self, mock_config, mock_presets):
"""Test build falls back to preset fields when user doesn't provide them."""
from app.library.ItemDTO import Item
from app.library.Presets import Preset
from app.features.presets.schemas import Preset
from app.library.YTDLPOpts import YTDLPCli
mock_config_instance = Mock()
@ -839,12 +839,12 @@ class TestYTDLPCli:
assert info["merged"]["save_path"] == "/downloads/preset_folder"
assert "--format 720p" in command
@patch("app.library.Presets.Presets")
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_build_user_fields_override_preset(self, mock_config, mock_presets):
"""Test that user fields override preset fields."""
from app.library.ItemDTO import Item
from app.library.Presets import Preset
from app.features.presets.schemas import Preset
from app.library.YTDLPOpts import YTDLPCli
mock_config_instance = Mock()
@ -878,7 +878,7 @@ class TestYTDLPCli:
assert info["merged"]["save_path"] == "/downloads/user_folder"
assert "--format best" in command, "User CLI should appear after preset CLI in command"
@patch("app.library.Presets.Presets")
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_build_with_default_fallback(self, mock_config, mock_presets):
"""Test build falls back to defaults when neither user nor preset provide fields."""
@ -903,7 +903,7 @@ class TestYTDLPCli:
assert info["merged"]["template"] == "%(title)s.%(ext)s"
assert info["merged"]["save_path"] == "/default/downloads"
@patch("app.library.Presets.Presets")
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.create_cookies_file")
@patch("app.library.YTDLPOpts.Config")
def test_build_with_cookies_from_user(self, mock_config, mock_create_cookies, mock_presets):
@ -933,12 +933,12 @@ class TestYTDLPCli:
assert "/tmp/cookies.txt" in command
assert info["merged"]["cookie_file"] == "/tmp/cookies.txt"
@patch("app.library.Presets.Presets")
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_build_with_cookies_from_preset(self, mock_config, mock_presets):
"""Test build with cookies from preset when user doesn't provide."""
from app.library.ItemDTO import Item
from app.library.Presets import Preset
from app.features.presets.schemas import Preset
from app.library.YTDLPOpts import YTDLPCli
mock_config_instance = Mock()
@ -969,7 +969,7 @@ class TestYTDLPCli:
assert "--cookies" in command
assert "/preset/cookies.txt" in command
@patch("app.library.Presets.Presets")
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_build_with_absolute_folder_path(self, mock_config, mock_presets):
"""Test build with absolute folder path from user."""
@ -996,7 +996,7 @@ class TestYTDLPCli:
)
assert "--paths" in command
@patch("app.library.Presets.Presets")
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_build_includes_url_in_command(self, mock_config, mock_presets):
"""Test that build includes the URL in the final command."""
@ -1018,12 +1018,12 @@ class TestYTDLPCli:
assert "https://youtube.com/watch?v=test123" in command
@patch("app.library.Presets.Presets")
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_build_cli_args_priority_order(self, mock_config, mock_presets):
"""Test that CLI args are added in correct priority order (preset first, user last)."""
from app.library.ItemDTO import Item
from app.library.Presets import Preset
from app.features.presets.schemas import Preset
from app.library.YTDLPOpts import YTDLPCli
mock_config_instance = Mock()

8
ui/app/components/NewDownload.vue
View file

@ -38,7 +38,7 @@
<div class="field has-addons">
<div class="control" @click="show_description = !show_description">
<label class="button is-static">
<span class="icon"><i class="fa-solid fa-sliders" /></span>
<span class="icon"><i class="fa-solid fa-info-circle" /></span>
</label>
</div>
<div class="control is-expanded">
@ -48,12 +48,12 @@
v-tooltip.bottom="hasFormatInConfig ? 'Presets are disabled. Format key is present in the command options for yt-dlp.' : ''">
<optgroup label="Custom presets" v-if="config?.presets.filter(p => !p?.default).length > 0">
<option v-for="cPreset in filter_presets(false)" :key="cPreset.name" :value="cPreset.name">
{{ cPreset.name }}
{{ prettyName(cPreset.name) }}
</option>
</optgroup>
<optgroup label="Default presets">
<option v-for="dPreset in filter_presets(true)" :key="dPreset.name" :value="dPreset.name">
{{ dPreset.name }}
{{ prettyName(dPreset.name) }}
</option>
</optgroup>
</select>
@ -269,6 +269,7 @@ import TextareaAutocomplete from '~/components/TextareaAutocomplete.vue'
import TextDropzone from '~/components/TextDropzone.vue'
import type { item_request } from '~/types/item'
import type { AutoCompleteOptions } from '~/types/autocomplete'
import { prettyName } from '~/utils'
import { navigateTo } from '#app'
import { useDialog } from '~/composables/useDialog'
@ -803,4 +804,5 @@ watch(isMultiLineInput, async newValue => {
const inputElement = document.getElementById('url') as HTMLInputElement
inputElement?.focus()
})
</script>

27
ui/app/components/PresetForm.vue
View file

@ -36,12 +36,12 @@
<option value="" disabled>Select a preset</option>
<optgroup label="Custom presets" v-if="config?.presets.filter(p => !p?.default).length > 0">
<option v-for="item in filter_presets(false)" :key="item.name" :value="item.name">
{{ item.name }}
{{ prettyName(item.name) }}
</option>
</optgroup>
<optgroup label="Default presets">
<option v-for="item in filter_presets(true)" :key="item.name" :value="item.name">
{{ item.name }}
{{ prettyName(item.name) }}
</option>
</optgroup>
</select>
@ -97,7 +97,7 @@
</div>
<span class="help is-bold">
<span class="icon"><i class="fa-solid fa-info" /></span>
<span>The name to refers to this preset of settings.</span>
<span>Names are stored in lowercase with underscores (no spaces).</span>
</span>
</div>
</div>
@ -253,16 +253,18 @@
import { useStorage } from '@vueuse/core'
import TextareaAutocomplete from '~/components/TextareaAutocomplete.vue'
import TextDropzone from '~/components/TextDropzone.vue'
import type { ImportedItem } from '~/types';
import type { AutoCompleteOptions } from '~/types/autocomplete';
import type { Preset, PresetImport } from '~/types/presets'
import type { Preset } from '~/types/presets'
import { normalizePresetName, prettyName } from '~/utils'
const emitter = defineEmits<{
(event: 'cancel'): void
(event: 'submit', payload: { reference: string | null, preset: Preset }): void
(event: 'submit', payload: { reference: number | null, preset: Preset }): void
}>()
const props = defineProps<{
reference?: string | null
reference?: number | null
preset: Partial<Preset>
addInProgress?: boolean
presets?: Preset[]
@ -298,6 +300,13 @@ const checkInfo = async (): Promise<void> => {
}
}
const normalizedName = normalizePresetName(String(form.name))
if (!normalizedName) {
toast.error('The name field is required.')
return
}
form.name = normalizedName
if (form.folder) {
form.folder = form.folder.trim()
await nextTick()
@ -313,13 +322,13 @@ const checkInfo = async (): Promise<void> => {
const copy: Preset = JSON.parse(JSON.stringify(form))
let usedName = false
const name = String(form.name).trim().toLowerCase()
const name = normalizedName
props.presets?.forEach(p => {
if (p.id === props.reference) {
return
}
if (String(p.name).toLowerCase() === name) {
if (p.name === name) {
usedName = true
}
})
@ -366,7 +375,7 @@ const importItem = async (): Promise<void> => {
}
try {
const item = decode(val) as PresetImport
const item = decode(val) as Preset & ImportedItem
if (!item?._type || 'preset' !== item._type) {
toast.error(`Invalid import string. Expected type 'preset', got '${item._type ?? 'unknown'}'.`)

407
ui/app/composables/usePresets.ts Normal file
View file

@ -0,0 +1,407 @@
import { ref, readonly } from 'vue'
import { useNotification } from '~/composables/useNotification'
import { request, parse_list_response, parse_api_response, parse_api_error } from '~/utils'
import type { Preset, PresetRequest } from '~/types/presets'
import type { APIResponse, Pagination } from '~/types/responses'
/**
* List of all presets in memory.
*/
const presets = ref<Array<Preset>>([])
/**
* Pagination state for presets list.
*/
const pagination = ref<Pagination>({
page: 1,
per_page: 50,
total: 0,
total_pages: 0,
has_next: false,
has_prev: false,
})
/**
* Indicates if a request is in progress.
*/
const isLoading = ref<boolean>(false)
/**
* Indicates if an add/update operation is in progress.
*/
const addInProgress = ref<boolean>(false)
/**
* Stores the last error message, if any.
*/
const lastError = ref<string | null>(null)
/**
* If true, methods will throw errors instead of returning null/false (for testing)
*/
const throwInstead = ref(false)
/**
* Notification composable for showing success/error messages.
*/
const notify = useNotification()
/**
* Sorts presets by priority (descending), then name (A-Z).
* @param items Array of Preset
* @returns Sorted array of Preset
*/
const sortPresets = (items: Array<Preset>): Array<Preset> => {
return [...items].sort((a, b) => {
if (a.priority === b.priority) {
return a.name.localeCompare(b.name)
}
return b.priority - a.priority
})
}
/**
* Safely reads JSON from a Response, returns null on error.
* @param response Fetch Response object
* @returns Parsed JSON or null
*/
const readJson = async (response: Response): Promise<unknown> => {
try {
const clone = response.clone()
return await clone.json()
}
catch {
return null
}
}
/**
* Throws an error if the response is not OK, using API error message if available.
* @param response Fetch Response object
* @throws Error with message from API or status code
*/
const ensureSuccess = async (response: Response): Promise<void> => {
if (response.ok) {
return
}
const payload = await readJson(response)
const message = await parse_api_error(payload)
throw new Error(message)
}
/**
* Handles errors by updating lastError and showing a notification.
* @param error Error object or unknown
*/
const handleError = (error: unknown): void => {
const message = error instanceof Error ? error.message : 'Unexpected error occurred.'
lastError.value = message
notify.error(message)
}
/**
* Updates or adds a preset in the presets list, keeping sort order.
* Also increments pagination.total if it's a new preset.
* @param preset Preset to update/add
*/
const updatePresets = (preset: Preset): void => {
const isNew = !presets.value.some(item => item.id === preset.id)
presets.value = sortPresets([
...presets.value.filter(item => item.id !== preset.id),
preset,
])
if (isNew) {
pagination.value.total++
}
}
/**
* Removes a preset from the presets list by ID.
* Also decrements pagination.total.
* @param id Preset ID
*/
const removePreset = (id: number) => {
const initialLength = presets.value.length
presets.value = presets.value.filter(item => item.id !== id)
if (presets.value.length < initialLength) {
pagination.value.total = Math.max(0, pagination.value.total - 1)
}
}
/**
* Loads all presets from the API with pagination support.
* Updates presets, pagination, and lastError.
* @param page Page number
* @param perPage Items per page
*/
const loadPresets = async (page: number = 1, perPage: number | undefined = undefined): Promise<void> => {
isLoading.value = true
try {
let url = `/api/presets/?page=${page}`
if (perPage !== undefined) {
url += `&per_page=${perPage}`
}
const response = await request(url)
await ensureSuccess(response)
const json = await response.json()
const { items, pagination: paginationData } = await parse_list_response<Preset>(json)
presets.value = sortPresets(items)
pagination.value = paginationData
lastError.value = null
}
catch (error) {
handleError(error)
if (throwInstead.value) throw error
}
finally {
isLoading.value = false
}
}
/**
* Fetches a single preset by ID from the API.
* @param id Preset ID
* @returns Preset or null on error
*/
const getPreset = async (id: number): Promise<Preset | null> => {
try {
const response = await request(`/api/presets/${id}`)
await ensureSuccess(response)
const json = await response.json()
const preset = await parse_api_response<Preset>(json)
lastError.value = null
return preset
}
catch (error) {
handleError(error)
if (throwInstead.value) throw error
return null
}
}
/**
* Creates a new preset via API.
* @param preset PresetRequest to create
* @param callback Optional callback with APIResponse result
* @returns Created Preset or null on error
*/
const createPreset = async (
preset: PresetRequest,
callback?: (response: APIResponse<Preset>) => void,
): Promise<Preset | null> => {
addInProgress.value = true
try {
const response = await request('/api/presets/', {
method: 'POST',
body: JSON.stringify(preset),
})
await ensureSuccess(response)
const json = await response.json()
const created = await parse_api_response<Preset>(json)
updatePresets(created)
notify.success('Preset created.')
lastError.value = null
if (callback) {
callback({ success: true, error: null, detail: null, data: created })
}
return created
}
catch (error) {
const errorMessage = error instanceof Error ? error.message : 'Unexpected error occurred.'
handleError(error)
if (callback) {
callback({ success: false, error: errorMessage, detail: error, data: undefined })
}
if (throwInstead.value) throw error
return null
}
finally {
addInProgress.value = false
}
}
/**
* Updates an existing preset via API (PUT - full update).
* @param id Preset ID
* @param preset Updated Preset data
* @param callback Optional callback with APIResponse result
* @returns Updated Preset or null on error
*/
const updatePreset = async (
id: number,
preset: Preset,
callback?: (response: APIResponse<Preset>) => void,
): Promise<Preset | null> => {
addInProgress.value = true
try {
const payload = { ...preset }
if (payload.id) {
payload.id = undefined
}
if ('default' in payload) {
payload.default = false
}
const response = await request(`/api/presets/${id}`, {
method: 'PUT',
body: JSON.stringify(payload),
})
await ensureSuccess(response)
const json = await response.json()
const updated = await parse_api_response<Preset>(json)
updatePresets(updated)
notify.success(`Preset '${updated.name}' updated.`)
lastError.value = null
if (callback) {
callback({ success: true, error: null, detail: null, data: updated })
}
return updated
}
catch (error) {
const errorMessage = error instanceof Error ? error.message : 'Unexpected error occurred.'
handleError(error)
if (callback) {
callback({ success: false, error: errorMessage, detail: error, data: undefined })
}
if (throwInstead.value) throw error
return null
}
finally {
addInProgress.value = false
}
}
/**
* Partially updates an existing preset via API (PATCH).
* @param id Preset ID
* @param patch Partial Preset data
* @param callback Optional callback with APIResponse result
* @returns Updated Preset or null on error
*/
const patchPreset = async (
id: number,
patch: Partial<Preset>,
callback?: (response: APIResponse<Preset>) => void,
): Promise<Preset | null> => {
addInProgress.value = true
try {
const payload = { ...patch }
if (payload.id) {
payload.id = undefined
}
if ('default' in payload) {
payload.default = false
}
const response = await request(`/api/presets/${id}`, {
method: 'PATCH',
body: JSON.stringify(payload),
})
await ensureSuccess(response)
const json = await response.json()
const updated = await parse_api_response<Preset>(json)
updatePresets(updated)
notify.success(`Preset '${updated.name}' updated.`)
lastError.value = null
if (callback) {
callback({ success: true, error: null, detail: null, data: updated })
}
return updated
}
catch (error) {
const errorMessage = error instanceof Error ? error.message : 'Unexpected error occurred.'
handleError(error)
if (callback) {
callback({ success: false, error: errorMessage, detail: error, data: undefined })
}
if (throwInstead.value) throw error
return null
}
finally {
addInProgress.value = false
}
}
/**
* Deletes a preset by ID via API.
* @param id Preset ID
* @param callback Optional callback with APIResponse result
* @returns true if deleted, false on error
*/
const deletePreset = async (
id: number,
callback?: (response: APIResponse<boolean>) => void,
): Promise<boolean> => {
try {
const response = await request(`/api/presets/${id}`, { method: 'DELETE' })
await ensureSuccess(response)
removePreset(id)
notify.success('Preset deleted.')
lastError.value = null
if (callback) {
callback({ success: true, error: null, detail: null, data: true })
}
return true
}
catch (error) {
const errorMessage = error instanceof Error ? error.message : 'Unexpected error occurred.'
handleError(error)
if (callback) {
callback({ success: false, error: errorMessage, detail: error, data: false })
}
if (throwInstead.value) throw error
return false
}
}
/**
* Clears the last error message.
*/
const clearError = () => lastError.value = null
/**
* usePresets composable
*
* Returns reactive state and CRUD methods for presets.
* @returns Object with state and API methods
*/
export const usePresets = () => ({
presets: readonly(presets),
pagination: readonly(pagination),
isLoading: readonly(isLoading),
addInProgress: readonly(addInProgress),
lastError: readonly(lastError),
loadPresets,
getPreset,
createPreset,
updatePreset,
patchPreset,
deletePreset,
clearError,
throwInstead,
})

133
ui/app/pages/presets.vue
View file

@ -6,7 +6,7 @@
<span class="icon-text">
<template v-if="toggleForm">
<span class="icon"><i class="fa-solid" :class="{ 'fa-edit': presetRef, 'fa-plus': !presetRef }" /></span>
<span>{{ presetRef ? `Edit - ${preset.name}` : 'Add' }}</span>
<span>{{ presetRef ? `Edit - ${prettyName(preset.name || '')}` : 'Add' }}</span>
</template>
<template v-else>
<span class="icon"><i class="fa-solid fa-sliders" /></span>
@ -89,7 +89,7 @@
<tr v-for="item in filteredPresets" :key="item.id">
<td class="is-text-overflow is-vcentered">
<div class="is-text-overflow is-bold">
{{ item.name }}
{{ prettyName(item.name) }}
</div>
<div class="is-unselectable">
<span class="icon-text" :class="{ 'has-text-primary': item.cookies }">
@ -141,7 +141,7 @@
<header class="card-header">
<div class="card-header-title is-block is-clickable"
:class="{ 'is-text-overflow': !isExpanded(item.id, 'title') }" @click="toggleExpand(item.id, 'title')"
:title="!isExpanded(item.id, 'title') ? 'Click to expand' : 'Click to collapse'" v-text="item.name" />
:title="!isExpanded(item.id, 'title') ? 'Click to expand' : 'Click to collapse'" v-text="prettyName(item.name)" />
<div class="card-header-icon">
<div class="field is-grouped">
<div class="control" v-if="item.priority > 0">
@ -252,10 +252,12 @@
import { useStorage } from '@vueuse/core'
import type { Preset } from '~/types/presets'
import { useConfirm } from '~/composables/useConfirm'
import { usePresets } from '~/composables/usePresets'
import { prettyName } from '~/utils'
type PresetWithUI = Preset & { raw?: boolean, toggle_description?: boolean }
const toast = useNotification()
const presetsStore = usePresets()
const config = useConfigStore()
const box = useConfirm()
@ -265,12 +267,12 @@ const isMobile = useMediaQuery({ maxWidth: 1024 })
const query = ref<string>('')
const toggleFilter = ref(false)
const presets = ref<PresetWithUI[]>([])
const presets = presetsStore.presets as Ref<PresetWithUI[]>
const preset = ref<Partial<Preset>>({})
const presetRef = ref<string | null>('')
const presetRef = ref<number | null>(null)
const toggleForm = ref(false)
const isLoading = ref(true)
const addInProgress = ref(false)
const isLoading = presetsStore.isLoading
const addInProgress = presetsStore.addInProgress
const remove_keys = ['raw', 'toggle_description']
const expandedItems = ref<Record<string, Set<string>>>({})
@ -282,23 +284,25 @@ const filteredPresets = computed<PresetWithUI[]>(() => {
return presetsNoDefault.value.filter((item: PresetWithUI) => deepIncludes(item, q, new WeakSet()));
});
const toggleExpand = (itemId: string | undefined, field: string) => {
if (!itemId) return
const toggleExpand = (itemId: number | string | undefined, field: string) => {
if (itemId === undefined || itemId === null) return
const key = String(itemId)
if (!expandedItems.value[itemId]) {
expandedItems.value[itemId] = new Set()
if (!expandedItems.value[key]) {
expandedItems.value[key] = new Set()
}
if (expandedItems.value[itemId].has(field)) {
expandedItems.value[itemId].delete(field)
if (expandedItems.value[key].has(field)) {
expandedItems.value[key].delete(field)
} else {
expandedItems.value[itemId].add(field)
expandedItems.value[key].add(field)
}
}
const isExpanded = (itemId: string | undefined, field: string): boolean => {
if (!itemId) return false
return expandedItems.value[itemId]?.has(field) ?? false
const isExpanded = (itemId: number | string | undefined, field: string): boolean => {
if (itemId === undefined || itemId === null) return false
const key = String(itemId)
return expandedItems.value[key]?.has(field) ?? false
}
@ -308,114 +312,47 @@ watch(toggleFilter, (val) => {
}
})
const reloadContent = async (fromMounted = false) => {
try {
isLoading.value = true
const response = await request('/api/presets')
if (fromMounted && !response.ok) {
return
}
const data = await response.json()
if (0 === data.length) {
return
}
presets.value = data
} catch (e: any) {
if (fromMounted) {
return
}
console.error(e)
toast.error('Failed to fetch page content.')
} finally {
isLoading.value = false
}
const reloadContent = async () => {
await presetsStore.loadPresets(1, 1000)
}
const resetForm = (closeForm = false) => {
preset.value = {}
presetRef.value = null
addInProgress.value = false
if (closeForm) {
toggleForm.value = false
}
}
const updatePresets = async (items: Preset[]): Promise<boolean | undefined> => {
let data: any
try {
addInProgress.value = true
const response = await request('/api/presets', {
method: 'PUT',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(items.filter((t) => !t.default)),
})
data = await response.json()
if (200 !== response.status) {
toast.error(`Failed to update presets. ${data.error}`)
return false
}
presets.value = data
resetForm(true)
return true
} catch (e: any) {
toast.error(`Failed to update presets. ${data?.error}. ${e.message}`)
} finally {
addInProgress.value = false
}
}
const deleteItem = async (item: Preset) => {
if (true !== (await box.confirm(`Delete preset '${item.name}'?`))) {
return
}
const index = presets.value.findIndex((t) => t?.id === item.id)
if (-1 === index) {
toast.error('Preset not found.')
return
if (item.id) {
await presetsStore.deletePreset(item.id)
}
presets.value.splice(index, 1)
const status = await updatePresets(presets.value)
if (!status) {
return
}
toast.success('Preset deleted.')
}
const updateItem = async ({
reference,
preset: item,
}: {
reference: string | null
reference: number | null
preset: Preset
}) => {
item = cleanObject(item, remove_keys) as Preset
if (reference) {
const index = presets.value.findIndex((t) => t?.id === reference)
if (-1 !== index) {
presets.value[index] = item
const updated = await presetsStore.updatePreset(reference, item)
if (updated) {
resetForm(true)
}
} else {
presets.value.push(item)
}
const status = await updatePresets(presets.value)
if (!status) {
return
}
toast.success(`Preset ${reference ? 'updated' : 'added'}.`)
const created = await presetsStore.createPreset(item)
if (created) {
resetForm(true)
}
}
}
const filterItem = (item: Preset) => {
@ -432,7 +369,7 @@ const editItem = (item: Preset) => {
toggleForm.value = true
}
onMounted(async () => await reloadContent(true))
onMounted(async () => await reloadContent())
const exportItem = (item: Preset) => {
const excludedKeys = ['id', 'default', 'raw', 'cookies', 'toggle_description']

25
ui/app/stores/ConfigStore.ts
View file

@ -133,9 +133,30 @@ export const useConfigStore = defineStore('config', () => {
}
if ('presets' === feature) {
if ('replace' === action) {
state.presets = data as Array<Preset>
const item = data as Preset
const current = get(feature, []) as Array<Preset>
if ('create' === action) {
current.push(item)
return
}
if ('delete' === action) {
const index = current.findIndex(i => i.id === item.id)
if (-1 !== index) {
current.splice(index, 1)
}
return
}
if ('update' === action) {
const target = current.find(i => i.id === item.id)
if (target) {
Object.assign(target, item)
}
return
}
return
}

18
ui/app/types/presets.d.ts vendored
View file

@ -1,6 +1,6 @@
type Preset = {
/** Unique identifier for the preset */
id?: string
id?: number
/** Preset name, e.g. "default" */
name: string
/** Optional description for the preset */
@ -19,9 +19,17 @@ type Preset = {
priority: number
}
type PresetImport = Preset & {
_type: 'preset'
_version: string
/**
* Request payload for creating/updating preset
*/
type PresetRequest = {
name: string
description?: string
folder?: string
template?: string
cookies?: string
cli?: string
priority?: number
}
export type { Preset, PresetImport }
export type { Preset, PresetRequest }

6
ui/app/utils/index.ts
View file

@ -407,6 +407,10 @@ const sTrim = (str: string, delim: string): string => iTrim(str, delim, 'start')
*/
const ucFirst = (str: string): string => (!str) ? str : str.charAt(0).toUpperCase() + str.slice(1)
const normalizePresetName = (name: string): string => name.trim().toLowerCase().replace(/\s+/g, '_')
const prettyName = (name: string): string => name.replace(/_/g, ' ').replace(/\b\w/g, c => c.toUpperCase())
/**
* Get the name of a separator based on its value
*
@ -893,7 +897,7 @@ const parse_api_error = async (json: unknown): Promise<string> => {
}
export {
separators, convertCliOptions, getSeparatorsName, iTrim, eTrim, sTrim, ucFirst,
separators, convertCliOptions, getSeparatorsName, iTrim, eTrim, sTrim, ucFirst, normalizePresetName, prettyName,
getValue, ag, ag_set, awaitElement, r, copyText, dEvent, makePagination, encodePath,
request, removeANSIColors, dec2hex, makeId, basename, dirname, getQueryParams,
makeDownload, formatBytes, has_data, toggleClass, cleanObject, uri, formatTime,

465
ui/tests/composables/usePresets.test.ts Normal file
View file

@ -0,0 +1,465 @@
import * as utils from '~/utils/index'
import { describe, it, expect, beforeEach, vi } from 'vitest'
import { usePresets } from '~/composables/usePresets'
import type { Preset, PresetRequest } from '~/types/presets'
import type { Pagination } from '~/types/responses'
vi.mock('~/composables/useNotification', () => {
const success = vi.fn()
const error = vi.fn()
return {
useNotification: () => ({ success, error }),
default: () => ({ success, error }),
}
})
const mockPreset: Preset = {
id: 1,
name: 'Test Preset',
description: 'Test description',
folder: 'Downloads',
template: '%(title)s.%(ext)s',
cookies: 'cookie=value',
cli: '--format best',
default: false,
priority: 10,
}
const mockPagination: Pagination = {
page: 1,
per_page: 50,
total: 1,
total_pages: 1,
has_next: false,
has_prev: false,
}
function createMockResponse({ ok, status, jsonData }: { ok: boolean; status: number; jsonData: any }) {
return {
ok,
status,
headers: new Headers(),
redirected: false,
statusText: '',
type: 'basic',
url: '',
body: null,
bodyUsed: false,
clone() {
return this
},
async json() {
return jsonData
},
text: async () => JSON.stringify(jsonData),
arrayBuffer: async () => new ArrayBuffer(0),
blob: async () => new Blob(),
formData: async () => new FormData(),
} as Response
}
describe('usePresets', () => {
beforeEach(() => {
vi.clearAllMocks()
})
describe('loadPresets', () => {
it('loads presets with pagination successfully', async () => {
vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: true,
status: 200,
jsonData: {
items: [mockPreset],
pagination: mockPagination,
},
}),
)
const presets = usePresets()
await presets.loadPresets()
expect(presets.presets.value).toHaveLength(1)
expect(presets.presets.value[0]).toEqual(mockPreset)
expect(presets.pagination.value).toEqual(mockPagination)
expect(presets.lastError.value).toBeNull()
})
it('sorts presets by priority then name', async () => {
const items = [
{ ...mockPreset, id: 1, name: 'B', priority: 2 },
{ ...mockPreset, id: 2, name: 'A', priority: 2 },
{ ...mockPreset, id: 3, name: 'C', priority: 1 },
]
vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: true,
status: 200,
jsonData: {
items,
pagination: mockPagination,
},
}),
)
const presets = usePresets()
await presets.loadPresets()
const sorted = presets.presets.value
expect(sorted[0].priority).toBe(2)
expect(sorted[0].name).toBe('A')
expect(sorted[1].priority).toBe(2)
expect(sorted[1].name).toBe('B')
expect(sorted[2].priority).toBe(1)
expect(sorted[2].name).toBe('C')
})
it('handles empty presets list', async () => {
vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: true,
status: 200,
jsonData: {
items: [],
pagination: { ...mockPagination, total: 0 },
},
}),
)
const presets = usePresets()
await presets.loadPresets()
expect(presets.presets.value).toEqual([])
expect(presets.lastError.value).toBeNull()
})
it('handles errors during load', async () => {
vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: false,
status: 500,
jsonData: { error: 'Server error' },
}),
)
const presets = usePresets()
await presets.loadPresets()
expect(presets.lastError.value).toBeTruthy()
})
})
describe('getPreset', () => {
it('fetches a single preset successfully', async () => {
vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: true,
status: 200,
jsonData: mockPreset,
}),
)
const presets = usePresets()
const result = await presets.getPreset(1)
expect(result).toEqual(mockPreset)
expect(presets.lastError.value).toBeNull()
})
it('handles 404 not found', async () => {
vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: false,
status: 404,
jsonData: { error: 'Preset not found' },
}),
)
const presets = usePresets()
const result = await presets.getPreset(999)
expect(result).toBeNull()
expect(presets.lastError.value).toBeTruthy()
})
})
describe('createPreset', () => {
it('creates a preset successfully', async () => {
vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: true,
status: 200,
jsonData: mockPreset,
}),
)
const presets = usePresets()
const initialTotal = presets.pagination.value.total
const newPreset: PresetRequest = {
name: 'New Preset',
description: 'Desc',
cli: '--format best',
}
const result = await presets.createPreset(newPreset)
expect(result).toEqual(mockPreset)
expect(presets.presets.value).toContainEqual(mockPreset)
expect(presets.pagination.value.total).toBe(initialTotal + 1)
})
it('calls callback on success', async () => {
vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: true,
status: 200,
jsonData: mockPreset,
}),
)
const callback = vi.fn()
const presets = usePresets()
const newPreset: PresetRequest = {
name: 'New Preset',
description: 'Desc',
cli: '--format best',
}
await presets.createPreset(newPreset, callback)
expect(callback).toHaveBeenCalledWith({
success: true,
error: null,
detail: null,
data: mockPreset,
})
})
it('calls callback on error', async () => {
vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: false,
status: 400,
jsonData: { detail: [{ loc: ['name'], msg: 'Field required', type: 'value_error' }] },
}),
)
const callback = vi.fn()
const presets = usePresets()
const newPreset: PresetRequest = {
name: 'New Preset',
description: 'Desc',
cli: '--format best',
}
await presets.createPreset(newPreset, callback)
expect(callback).toHaveBeenCalledWith(
expect.objectContaining({
success: false,
error: expect.any(String),
}),
)
})
})
describe('updatePreset', () => {
it('updates a preset successfully', async () => {
vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: true,
status: 200,
jsonData: mockPreset,
}),
)
const presets = usePresets()
const result = await presets.updatePreset(1, mockPreset)
expect(result).toEqual(mockPreset)
})
it('strips id from update payload', async () => {
const requestSpy = vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: true,
status: 200,
jsonData: mockPreset,
}),
)
const presets = usePresets()
await presets.updatePreset(1, { ...mockPreset, default: true })
const requestBody = JSON.parse((requestSpy.mock.calls[0][1] as any).body)
expect(requestBody.id).toBeUndefined()
expect(requestBody.default).toBe(false)
})
})
describe('patchPreset', () => {
it('patches a preset successfully', async () => {
vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: true,
status: 200,
jsonData: { ...mockPreset, priority: 20 },
}),
)
const presets = usePresets()
const result = await presets.patchPreset(1, { priority: 20 })
expect(result?.priority).toBe(20)
})
it('strips id and default from patch payload', async () => {
const requestSpy = vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: true,
status: 200,
jsonData: mockPreset,
}),
)
const presets = usePresets()
await presets.patchPreset(1, { id: 10, default: true })
const requestBody = JSON.parse((requestSpy.mock.calls[0][1] as any).body)
expect(requestBody.id).toBeUndefined()
expect(requestBody.default).toBe(false)
})
})
describe('deletePreset', () => {
it('deletes a preset successfully', async () => {
vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: true,
status: 200,
jsonData: mockPreset,
}),
)
const presets = usePresets()
const initialTotal = presets.pagination.value.total
const result = await presets.deletePreset(mockPreset.id!)
expect(result).toBe(true)
expect(presets.pagination.value.total).toBe(Math.max(0, initialTotal - 1))
})
it('calls callback on delete success', async () => {
vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: true,
status: 200,
jsonData: mockPreset,
}),
)
const callback = vi.fn()
const presets = usePresets()
await presets.deletePreset(1, callback)
expect(callback).toHaveBeenCalledWith({
success: true,
error: null,
detail: null,
data: true,
})
})
it('calls callback on delete error', async () => {
vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: false,
status: 404,
jsonData: { error: 'Preset not found' },
}),
)
const callback = vi.fn()
const presets = usePresets()
await presets.deletePreset(999, callback)
expect(callback).toHaveBeenCalledWith(
expect.objectContaining({
success: false,
error: expect.any(String),
data: false,
}),
)
})
})
describe('error handling', () => {
it('throws when throwInstead is true', async () => {
vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: false,
status: 500,
jsonData: { error: 'Server error' },
}),
)
const presets = usePresets()
presets.throwInstead.value = true
await expect(presets.loadPresets()).rejects.toThrow()
})
it('clears error on clearError call', async () => {
vi.spyOn(utils, 'request').mockResolvedValueOnce(
createMockResponse({
ok: false,
status: 500,
jsonData: { error: 'Server error' },
}),
)
const presets = usePresets()
presets.throwInstead.value = false
await presets.loadPresets()
expect(presets.lastError.value).toBeTruthy()
presets.clearError()
expect(presets.lastError.value).toBeNull()
})
})
describe('addInProgress state', () => {
it('sets addInProgress during create operation', async () => {
let inProgressDuringCall = false
vi.spyOn(utils, 'request').mockImplementation(async () => {
const presets = usePresets()
inProgressDuringCall = presets.addInProgress.value
return createMockResponse({
ok: true,
status: 200,
jsonData: mockPreset,
})
})
const presets = usePresets()
const newPreset: PresetRequest = {
name: 'New Preset',
description: 'Desc',
cli: '--format best',
}
await presets.createPreset(newPreset)
expect(inProgressDuringCall).toBe(true)
expect(presets.addInProgress.value).toBe(false)
})
})
})