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

Merge pull request #515 from arabcoders/dev

Browse source 
Refactor: swap sqlite to aiosqlite and add queued db updates
This commit is contained in:
Abdulmohsen 2025-12-19 22:11:36 +03:00 committed by GitHub
commit 47ca5461f1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
33 changed files with 3001 additions and 1839 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
.vscode/settings.json vendored
View file

@ -16,6 +16,8 @@
"ahas",
"ahash",
"aiocron",
"aiosqlite",
"alnum",
"anyio",
"Archiver",
"arrowdown",
@ -73,6 +75,8 @@
"falsey",
"faststart",
"Fetc",
"fetchall",
"fetchone",
"filterwarnings",
"finaldir",
"flac",
@ -91,6 +95,7 @@
"hookspath",
"httpx",
"imagetools",
"isinstance",
"jmespath",
"jsonschema",
"kibibytes",
@ -112,7 +117,10 @@
"mebibytes",
"MEIPASS",
"merch",
"metaclass",
"Metaclasses",
"metadataparser",
"mgmt",
"Microformat",
"microformats",
"mkvtoolsnix",
@ -128,6 +136,7 @@
"noninteractive",
"noprogress",
"onefile",
"oneline",
"parsel",
"pathex",
"pickleable",
@ -178,6 +187,7 @@
"Unraid",
"upgrader",
"urandom",
"urlparse",
"urlsafe",
"usegmt",
"usermod",
@ -190,6 +200,7 @@
"xerror",
"youtu",
"youtubepot",
"zstd",
"русский",
"العربية"
],

2
app/library/BackgroundWorker.py
View file

@ -6,6 +6,7 @@ from queue import Empty, Queue
from aiohttp import web
from .Services import Services
from .Singleton import Singleton
LOG: logging.Logger = logging.getLogger("BackgroundWorker")
@ -35,6 +36,7 @@ class BackgroundWorker(metaclass=Singleton):
return BackgroundWorker()
def attach(self, app: web.Application):
Services.get_instance().add("background_worker", self)
app.on_shutdown.append(self.on_shutdown)
LOG.debug("Starting background worker...")

333
app/library/DataStore.py
View file

@ -1,27 +1,20 @@
import copy
import json
import logging
from collections import OrderedDict
from dataclasses import fields
from datetime import UTC, datetime
from email.utils import formatdate
from collections.abc import Iterable
from enum import Enum
from sqlite3 import Connection
from .Download import Download
from .ItemDTO import ItemDTO
from .operations import matches_condition
from .Utils import init_class
from .sqlite_store import SqliteStore
LOG: logging.Logger = logging.getLogger("datastore")
ITEM_DTO_FIELDS = {f.name for f in fields(ItemDTO)}
"""Caching ItemDTO fields for performance."""
LOG = logging.getLogger("datastore")
class StoreType(str, Enum):
HISTORY: str = "done"
QUEUE: str = "queue"
HISTORY = "done"
QUEUE = "queue"
@classmethod
def all(cls) -> list[str]:
@ -29,48 +22,46 @@ class StoreType(str, Enum):
@classmethod
def from_value(cls, value: str) -> "StoreType":
"""
Returns the StoreType enum member corresponding to the given value.
Args:
value (str): The value to match against the enum members.
Returns:
StoreType: The enum member that matches the value.
Raises:
ValueError: If the value does not match any member.
"""
for member in cls:
if member.value == value:
return member
msg = f"Invalid StoreType value: {value}"
raise ValueError(msg)
def __str__(self) -> str:
def __str__(self) -> str: # pragma: no cover
return self.value
def _strip_transient_fields(item: ItemDTO) -> ItemDTO:
stored = copy.deepcopy(item)
if hasattr(stored, "datetime"):
try:
delattr(stored, "datetime")
except AttributeError:
pass
if stored.status == "finished" and hasattr(stored, "live_in"):
try:
delattr(stored, "live_in")
except AttributeError:
pass
return stored
class DataStore:
"""
Persistent queue.
"""
def __init__(self, type: StoreType, connection: Connection):
def __init__(self, type: StoreType, connection: SqliteStore):
self._type = type
self._connection: SqliteStore = connection
self._dict: OrderedDict[str, Download] = OrderedDict()
"The dictionary of items."
self._type: StoreType = type
"The type of the datastore."
self._connection: Connection = connection
"The database connection."
def load(self) -> None:
for id, item in self.saved_items():
self._dict.update({id: Download(info=item)})
async def load(self) -> None:
saved = await self._connection.fetch_saved(str(self._type))
for key, item in saved:
self._dict[key] = Download(info=item)
def exists(self, key: str | None = None, url: str | None = None) -> bool:
async def saved_items(self) -> list[tuple[str, ItemDTO]]:
return await self._connection.fetch_saved(str(self._type))
async def exists(self, key: str | None = None, url: str | None = None) -> bool:
if not key and not url:
msg = "key or url must be provided."
raise KeyError(msg)
@ -78,11 +69,12 @@ class DataStore:
if key and key in self._dict:
return True
return any(
(key and self._dict[i].info._id == key) or (url and self._dict[i].info.url == url) for i in self._dict
)
if any((key and self._dict[i].info._id == key) or (url and self._dict[i].info.url == url) for i in self._dict):
return True
def get(self, key: str | None = None, url: str | None = None) -> Download:
return StoreType.HISTORY == self._type and await self._connection.exists(str(self._type), key=key, url=url)
async def get(self, key: str | None = None, url: str | None = None) -> Download:
if not key and not url:
msg = "key or url must be provided."
raise KeyError(msg)
@ -91,37 +83,14 @@ class DataStore:
if (key and self._dict[i].info._id == key) or (url and self._dict[i].info.url == url):
return self._dict[i]
if StoreType.HISTORY == self._type and (item := await self._connection.get(str(self._type), key=key, url=url)):
self._dict[item._id] = Download(info=item)
return self._dict[item._id]
msg: str = f"{key=} or {url=} not found."
raise KeyError(msg)
def get_item(self, **kwargs) -> Download | None:
"""
Get a specific item from the datastore based on provided attributes with optional operations.
Args:
**kwargs: Arbitrary keyword arguments representing attributes of the ItemDTO.
Each value can be either:
- A direct value (defaults to EQUAL operation): {"title": "test"}
- A tuple of (Operation, value): {"title": (Operation.CONTAIN, "test")}
If no attributes are provided, the method returns None.
If any attribute matches, the corresponding Download object is returned.
Returns:
Download | None: The requested item if found, otherwise None.
Examples:
# Direct equality check (default)
store.get_item(title="Video 1")
# Using operations
store.get_item(title=(Operation.CONTAIN, "test"))
store.get_item(id=(Operation.EQUAL, "123"), status=(Operation.NOT_EQUAL, "error"))
# Mixed usage
store.get_item(title=(Operation.CONTAIN, "test"), folder="downloads")
"""
async def get_item(self, **kwargs) -> Download | None:
if not kwargs:
return None
@ -130,118 +99,62 @@ class DataStore:
continue
info = self._dict[i].info.__dict__
if any(matches_condition(key, value, info) for key, value in kwargs.items()):
return self._dict[i]
if StoreType.HISTORY == self._type and (item := await self._connection.get_item(str(self._type), **kwargs)):
self._dict[item._id] = Download(info=item)
return self._dict[item._id]
return None
def get_by_id(self, id: str) -> Download | None:
return self._dict.get(id, None)
async def get_by_id(self, id: str) -> Download | None:
val = self._dict.get(id)
if val or self._type == StoreType.QUEUE:
return val
def items(self) -> OrderedDict[tuple[str, Download]]:
if item := await self._connection.get_by_id(str(self._type), id):
self._dict[item._id] = Download(info=item)
return self._dict[id]
return None
def items(self):
return self._dict.items()
def saved_items(self) -> list[tuple[str, ItemDTO]]:
items: list[tuple[str, ItemDTO]] = []
cursor = self._connection.execute(
'SELECT "id", "data", "created_at" FROM "history" WHERE "type" = ? ORDER BY "created_at" ASC',
(str(self._type),),
)
for row in cursor:
rowDate: datetime = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S") # noqa: DTZ007
data: dict = json.loads(row["data"])
data.pop("_id", None)
item: ItemDTO = init_class(ItemDTO, data, ITEM_DTO_FIELDS)
item._id = row["id"]
item.datetime = formatdate(rowDate.replace(tzinfo=UTC).timestamp())
items.append((row["id"], item))
return items
def put(self, value: Download, no_notify: bool = False) -> Download:
if "error" == value.info.status and not no_notify:
from app.library.Events import EventBus, Events
EventBus.get_instance().emit(Events.ITEM_ERROR, value.info)
self._dict.update({value.info._id: value})
self._update_store_item(self._type, value.info)
async def put(self, value: Download, no_notify: bool = False) -> Download:
_ = no_notify
self._dict[value.info._id] = value
await self._connection.enqueue_upsert(str(self._type), _strip_transient_fields(value.info))
return self._dict[value.info._id]
def delete(self, key: str) -> None:
async def delete(self, key: str) -> None:
self._dict.pop(key, None)
self._delete_store_item(key)
await self._connection.enqueue_delete(str(self._type), key)
def next(self) -> tuple[str, Download]:
def next(self):
return next(iter(self._dict.items()))
def empty(self):
return not bool(self._dict)
def has_downloads(self):
if 0 == len(self._dict):
return False
def empty(self) -> bool:
return 0 == len(self._dict)
def has_downloads(self) -> bool:
return any(self._dict[key].info.auto_start and self._dict[key].started() is False for key in self._dict)
def get_next_download(self) -> Download:
def get_next_download(self) -> Download | None:
for key in self._dict:
ref = self._dict[key]
if ref.info.auto_start and ref.started() is False and ref.is_cancelled() is False:
if ref.info.auto_start and not ref.started() and not ref.is_cancelled():
return ref
return None
async def test(self) -> bool:
self._connection.execute('SELECT "id" FROM "history" LIMIT 1').fetchone()
return True
async def get_total_count(self, status_filter: str | None = None) -> int:
return await self._connection.count(str(self._type), status_filter=status_filter)
def get_total_count(self) -> int:
"""
Get the total count of items in the datastore.
Returns:
int: The total number of items in the datastore.
"""
cursor = self._connection.execute(
'SELECT COUNT(*) as count FROM "history" WHERE "type" = ?',
(str(self._type),),
)
row = cursor.fetchone()
return row["count"] if row else 0
def get_items_paginated(
self,
page: int = 1,
per_page: int = 50,
order: str = "DESC",
status_filter: str | None = None,
) -> tuple[list[tuple[str, ItemDTO]], int, int, int]:
"""
Get paginated items from the datastore.
Args:
page (int): The page number (1-indexed). Defaults to 1.
per_page (int): Number of items per page. Defaults to 50.
order (str): Sort order - 'ASC' or 'DESC'. Defaults to 'DESC' (newest first).
status_filter (str | None): Optional status filter. Can be a status value (e.g., 'finished')
to include only that status, or prefixed with '!' (e.g., '!finished') to exclude that status.
Returns:
tuple[list[tuple[str, ItemDTO]], int, int, int]: A tuple containing:
- List of (id, ItemDTO) tuples for the requested page
- Total number of items
- Current page number
- Total number of pages
Raises:
ValueError: If page < 1 or per_page < 1
"""
async def get_items_paginated(
self, page: int = 1, per_page: int = 50, order: str = "DESC", status_filter: str | None = None
) -> tuple[list[tuple[str, Download]], int, int, int]:
if page < 1:
msg = "page must be >= 1"
raise ValueError(msg)
@ -255,92 +168,18 @@ class DataStore:
msg = f"order must be 'ASC' or 'DESC', got '{order}'"
raise ValueError(msg)
order = "ASC" if order == "ASC" else "DESC"
# Build SQL query with status filter if provided
where_clauses = ['"type" = ?']
query_params: list = [str(self._type)]
if status_filter:
# Check if it's an exclusion filter (starts with !)
if status_filter.startswith("!"):
status_value = status_filter[1:]
where_clauses.append("json_extract(data, '$.status') != ?")
query_params.append(status_value)
else:
where_clauses.append("json_extract(data, '$.status') = ?")
query_params.append(status_filter)
where_clause = " AND ".join(where_clauses)
# Get total count with filter
count_query = f'SELECT COUNT(*) as count FROM "history" WHERE {where_clause}' # noqa: S608
count_result = self._connection.execute(count_query, tuple(query_params)).fetchone()
total_items: int = count_result["count"] if count_result else 0
total_pages: int = (total_items + per_page - 1) // per_page if total_items > 0 else 1
# Ensure page is within valid range.
if page > total_pages and total_items > 0:
page = total_pages
offset = (page - 1) * per_page
items: list[tuple[str, ItemDTO]] = []
query_params.extend([per_page, offset])
cursor = self._connection.execute(
f'SELECT "id", "data", "created_at" FROM "history" WHERE {where_clause} ORDER BY "created_at" {order} LIMIT ? OFFSET ?', # noqa: S608
tuple(query_params),
items, total_items, current_page, total_pages = await self._connection.paginate(
str(self._type), page, per_page, order, status_filter
)
for row in cursor:
rowDate: datetime = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S") # noqa: DTZ007
data: dict = json.loads(row["data"])
data.pop("_id", None)
item: ItemDTO = init_class(ItemDTO, data, ITEM_DTO_FIELDS)
item._id = row["id"]
item.datetime = formatdate(rowDate.replace(tzinfo=UTC).timestamp())
items.append((row["id"], item))
return [(item_id, Download(info=item)) for item_id, item in items], total_items, current_page, total_pages
return items, total_items, page, total_pages
async def bulk_delete(self, ids: Iterable[str]) -> int:
deleted = await self._connection.bulk_delete(str(self._type), ids)
for _id in ids:
self._dict.pop(_id, None)
return deleted
def _update_store_item(self, type: StoreType, item: ItemDTO) -> None:
sqlStatement = """
INSERT INTO "history" ("id", "type", "url", "data")
VALUES (?, ?, ?, ?)
ON CONFLICT DO UPDATE SET "type" = ?, "url" = ?, "data" = ?, created_at = ?
"""
stored: ItemDTO = copy.deepcopy(item)
if hasattr(stored, "datetime"):
try:
delattr(stored, "datetime")
except AttributeError:
pass
if hasattr(stored, "live_in") and stored.status == "finished":
try:
delattr(stored, "live_in")
except AttributeError:
pass
encoded: str = stored.json()
self._connection.execute(
sqlStatement.strip(),
(
stored._id,
str(type),
stored.url,
encoded,
str(type),
stored.url,
encoded,
datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"),
),
)
def _delete_store_item(self, key: str) -> None:
self._connection.execute('DELETE FROM "history" WHERE "type" = ? AND "id" = ?', (str(self._type), key))
async def test(self) -> bool:
await self._connection.count(str(self._type))
return True

81
app/library/DownloadQueue.py
View file

@ -9,7 +9,6 @@ import uuid
from datetime import UTC, datetime, timedelta
from email.utils import formatdate
from pathlib import Path
from sqlite3 import Connection
from typing import TYPE_CHECKING, Any
import yt_dlp.utils
@ -24,7 +23,9 @@ from .Events import EventBus, Events
from .ItemDTO import Item, ItemDTO
from .Presets import Presets
from .Scheduler import Scheduler
from .Services import Services
from .Singleton import Singleton
from .sqlite_store import SqliteStore
from .Utils import (
archive_add,
arg_converter,
@ -49,14 +50,14 @@ class DownloadQueue(metaclass=Singleton):
DownloadQueue class is a singleton class that manages the download queue and the download history.
"""
def __init__(self, connection: Connection, config: Config | None = None):
def __init__(self, config: Config | None = None):
self.config: Config = config or Config.get_instance()
"Configuration instance."
self._notify: EventBus = EventBus.get_instance()
"Event bus instance."
self.done = DataStore(type=StoreType.HISTORY, connection=connection)
self.done = DataStore(type=StoreType.HISTORY, connection=SqliteStore.get_instance())
"DataStore for the completed downloads."
self.queue = DataStore(type=StoreType.QUEUE, connection=connection)
self.queue = DataStore(type=StoreType.QUEUE, connection=SqliteStore.get_instance())
"DataStore for the download queue."
self.workers = asyncio.Semaphore(self.config.max_workers)
"Semaphore to limit the number of concurrent downloads."
@ -71,12 +72,10 @@ class DownloadQueue(metaclass=Singleton):
self._active: dict[str, Download] = {}
"""Dictionary of active downloads."""
self.done.load()
self.queue.load()
self.paused.set()
@staticmethod
def get_instance() -> "DownloadQueue":
def get_instance(config: Config | None = None) -> "DownloadQueue":
"""
Get the instance of the DownloadQueue.
@ -84,7 +83,7 @@ class DownloadQueue(metaclass=Singleton):
DownloadQueue: The instance of the DownloadQueue
"""
return DownloadQueue()
return DownloadQueue(config=config)
def _get_limit(self, extractor: str) -> asyncio.Semaphore:
"""
@ -123,6 +122,7 @@ class DownloadQueue(metaclass=Singleton):
_ (web.Application): The application to attach the download queue to.
"""
Services.get_instance().add("queue", self)
async def event_handler(_, __):
await self.initialize()
@ -166,6 +166,7 @@ class DownloadQueue(metaclass=Singleton):
"""
Initialize the download queue.
"""
await self.queue.load()
LOG.info(
f"Using '{self.config.max_workers}' workers for downloading and '{self.config.max_workers_per_extractor}' per extractor."
)
@ -187,7 +188,7 @@ class DownloadQueue(metaclass=Singleton):
for item_id in ids:
try:
item: Download = self.queue.get(key=item_id)
item: Download = await self.queue.get(key=item_id)
except KeyError as e:
status[item_id] = f"not found: {e!s}"
status["status"] = "error"
@ -199,7 +200,7 @@ class DownloadQueue(metaclass=Singleton):
continue
item.info.auto_start = True
updated: Download = self.queue.put(item)
updated: Download = await self.queue.put(item)
self._notify.emit(Events.ITEM_UPDATED, data=updated.info)
self._notify.emit(
Events.ITEM_RESUMED,
@ -230,7 +231,7 @@ class DownloadQueue(metaclass=Singleton):
for item_id in ids:
try:
item: Download = self.queue.get(key=item_id)
item: Download = await self.queue.get(key=item_id)
except KeyError as e:
status[item_id] = f"not found: {e!s}"
status["status"] = "error"
@ -246,7 +247,7 @@ class DownloadQueue(metaclass=Singleton):
continue
item.info.auto_start = False
updated: Download = self.queue.put(item)
updated: Download = await self.queue.put(item)
self._notify.emit(Events.ITEM_UPDATED, data=updated.info)
self._notify.emit(
Events.ITEM_PAUSED,
@ -443,7 +444,7 @@ class DownloadQueue(metaclass=Singleton):
LOG.debug(f"Entry id '{entry.get('id')}' url '{entry.get('webpage_url')} - {entry.get('url')}'.")
try:
_item: Download = self.done.get(key=entry.get("id"), url=entry.get("webpage_url") or entry.get("url"))
_item: Download = await self.done.get(key=entry.get("id"), url=entry.get("webpage_url") or entry.get("url"))
err_msg: str = f"Removing {_item.info.name()} from history list."
LOG.warning(err_msg)
await self.clear([_item.info._id], remove_file=False)
@ -451,7 +452,7 @@ class DownloadQueue(metaclass=Singleton):
pass
try:
_item: Download = self.queue.get(
_item: Download = await self.queue.get(
key=str(entry.get("id")), url=str(entry.get("webpage_url") or entry.get("url"))
)
err_msg: str = f"Item {_item.info.name()} is already in download queue."
@ -534,7 +535,7 @@ class DownloadQueue(metaclass=Singleton):
message=nMessage,
)
itemDownload: Download = self.done.put(dlInfo)
itemDownload: Download = await self.done.put(dlInfo)
elif not hasFormats:
ava: str = entry.get("availability", "public")
nTitle = "Download Error"
@ -547,7 +548,7 @@ class DownloadQueue(metaclass=Singleton):
dlInfo.info.error = nMessage.replace(f" for '{dl.title}'.", ".") + text_logs
dlInfo.info.status = "error"
itemDownload = self.done.put(dlInfo)
itemDownload = await self.done.put(dlInfo)
self._notify.emit(
Events.LOG_WARNING,
@ -580,12 +581,12 @@ class DownloadQueue(metaclass=Singleton):
if _requeue:
nEvent = Events.ITEM_ADDED
itemDownload = self.queue.put(dlInfo)
itemDownload = await self.queue.put(dlInfo)
if item.auto_start:
self.event.set()
else:
dlInfo.info.status = "not_live"
itemDownload = self.done.put(dlInfo)
itemDownload = await self.done.put(dlInfo)
nStore = "history"
nEvent = Events.ITEM_MOVED
nTitle = "Premiering right now"
@ -596,7 +597,7 @@ class DownloadQueue(metaclass=Singleton):
nEvent = Events.ITEM_ADDED
nTitle = "Item Added"
nMessage = f"Item '{dlInfo.info.title}' has been added to the download queue."
itemDownload = self.queue.put(dlInfo)
itemDownload = await self.queue.put(dlInfo)
if item.auto_start:
self.event.set()
else:
@ -712,7 +713,7 @@ class DownloadQueue(metaclass=Singleton):
if item.is_archived():
if archive_id:
store_type, _ = self.get_item(archive_id=archive_id)
store_type, _ = await self.get_item(archive_id=archive_id)
if not store_type:
dlInfo = Download(
info=ItemDTO(
@ -732,7 +733,7 @@ class DownloadQueue(metaclass=Singleton):
if archive_file := dlInfo.info.get_ytdlp_opts().get_all().get("download_archive"):
dlInfo.info.msg += f" Found in archive '{archive_file}'."
self.done.put(dlInfo)
await self.done.put(dlInfo)
self._notify.emit(
Events.ITEM_MOVED,
@ -800,7 +801,7 @@ class DownloadQueue(metaclass=Singleton):
log_message = f"Ignoring download of '{item_title}' as per condition '{condition.name}'{extra_msg}."
store_type, _ = self.get_item(archive_id=archive_id)
store_type, _ = await self.get_item(archive_id=archive_id)
if not store_type:
dlInfo = Download(
info=ItemDTO(
@ -816,7 +817,7 @@ class DownloadQueue(metaclass=Singleton):
extras=item.extras,
)
)
self.done.put(dlInfo)
await self.done.put(dlInfo)
LOG.info(log_message)
self._notify.emit(Events.LOG_INFO, data={}, title="Ignored Download", message=log_message)
@ -885,7 +886,7 @@ class DownloadQueue(metaclass=Singleton):
for id in ids:
try:
item = self.queue.get(key=id)
item = await self.queue.get(key=id)
except KeyError as e:
status[id] = str(e)
status["status"] = "error"
@ -902,7 +903,7 @@ class DownloadQueue(metaclass=Singleton):
else:
await item.close()
LOG.debug(f"Deleting from queue {item_ref}")
self.queue.delete(id)
await self.queue.delete(id)
self._notify.emit(
Events.ITEM_CANCELLED,
data=item.info,
@ -910,7 +911,7 @@ class DownloadQueue(metaclass=Singleton):
message=f"Download '{item.info.title}' has been cancelled.",
)
item.info.status = "cancelled"
self.done.put(item)
await self.done.put(item)
self._notify.emit(
Events.ITEM_MOVED,
data={"to": "history", "preset": item.info.preset, "item": item.info},
@ -939,7 +940,7 @@ class DownloadQueue(metaclass=Singleton):
for id in ids:
try:
item: Download = self.done.get(key=id)
item: Download = await self.done.get(key=id)
except KeyError as e:
status[id] = str(e)
status["status"] = "error"
@ -984,7 +985,7 @@ class DownloadQueue(metaclass=Singleton):
except Exception as e:
LOG.error(f"Unable to remove '{itemRef}' local file '{filename}'. {e!s}")
self.done.delete(id)
await self.done.delete(id)
_status: str = "Removed" if removed_files > 0 else "Cleared"
self._notify.emit(
@ -1003,7 +1004,7 @@ class DownloadQueue(metaclass=Singleton):
return status
def get(self, mode: str = "all") -> dict[str, list[dict[str, ItemDTO]]]:
async def get(self, mode: str = "all") -> dict[str, list[dict[str, ItemDTO]]]:
"""
Get the download queue and the download history.
@ -1017,12 +1018,11 @@ class DownloadQueue(metaclass=Singleton):
items = {"queue": {}, "done": {}}
if mode in ("all", "queue"):
for k, v in self.queue.saved_items():
for k, v in await self.queue.saved_items():
items["queue"][k] = self._active[k].info if k in self._active else v
if mode in ("all", "done"):
for k, v in self.done.saved_items():
v.get_file_sidecar()
for k, v in await self.done.saved_items():
items["done"][k] = v
if mode in ("all", "queue"):
@ -1035,12 +1035,11 @@ class DownloadQueue(metaclass=Singleton):
if k in items["done"]:
continue
v.info.get_file_sidecar()
items["done"][k] = v.info
return items
def get_item(self, **kwargs) -> tuple[StoreType, Download] | tuple[None, None]:
async def get_item(self, **kwargs) -> tuple[StoreType, Download] | tuple[None, None]:
"""
Get a specific item from the download queue or history.
@ -1051,10 +1050,10 @@ class DownloadQueue(metaclass=Singleton):
(StoreType, Download) | None: The requested item if found, otherwise None.
"""
if item := self.queue.get_item(**kwargs):
if item := await self.queue.get_item(**kwargs):
return (StoreType.QUEUE, item)
if item := self.done.get_item(**kwargs):
if item := await self.done.get_item(**kwargs):
return (StoreType.HISTORY, item)
return (None, None)
@ -1165,9 +1164,9 @@ class DownloadQueue(metaclass=Singleton):
await entry.close()
if self.queue.exists(key=id):
if await self.queue.exists(key=id):
LOG.debug(f"Download Task '{id}' is completed. Removing from queue.")
self.queue.delete(key=id)
await self.queue.delete(key=id)
nTitle: str | None = None
nMessage: str | None = None
@ -1186,7 +1185,7 @@ class DownloadQueue(metaclass=Singleton):
self._notify.emit(Events.ITEM_COMPLETED, data=entry.info, title=nTitle, message=nMessage)
self.done.put(entry)
await self.done.put(entry)
self._notify.emit(
Events.ITEM_MOVED,
data={"to": "history", "preset": entry.info.preset, "item": entry.info},
@ -1288,7 +1287,7 @@ class DownloadQueue(metaclass=Singleton):
)
)
except Exception as e:
self.done.put(item)
await self.done.put(item)
LOG.exception(e)
LOG.error(f"Failed to retry item '{item_ref}'. {e!s}")
@ -1329,7 +1328,7 @@ class DownloadQueue(metaclass=Singleton):
message=f"'{item_name}' record removed from history.",
)
titles.append(item_name)
self.done.delete(key)
await self.done.delete(key)
if titles:
LOG.info(f"Automatically cleared '{', '.join(titles)}' from download history due to age.")

2
app/library/Events.py
View file

@ -25,6 +25,7 @@ class Events:
CONNECTED: str = "connected"
CONFIGURATION: str = "configuration"
ACTIVE_QUEUE: str = "active_queue"
LOG_INFO: str = "log_info"
LOG_WARNING: str = "log_warning"
@ -94,6 +95,7 @@ class Events:
return [
Events.CONFIGURATION,
Events.CONNECTED,
Events.ACTIVE_QUEUE,
Events.LOG_INFO,
Events.LOG_WARNING,
Events.LOG_ERROR,

7
app/library/HttpAPI.py
View file

@ -14,7 +14,6 @@ from app.library.Services import Services
from .cache import Cache
from .config import Config
from .DownloadQueue import DownloadQueue
from .encoder import Encoder
from .Events import EventBus
from .router import RouteType, get_routes
@ -24,8 +23,7 @@ LOG: logging.Logger = logging.getLogger("http_api")
class HttpAPI:
def __init__(self, root_path: Path, queue: DownloadQueue):
self.queue: DownloadQueue = queue or DownloadQueue.get_instance()
def __init__(self, root_path: Path):
self.encoder: Encoder = Encoder()
self.config: Config = Config.get_instance()
self._notify: EventBus = EventBus.get_instance()
@ -33,12 +31,11 @@ class HttpAPI:
self.cache = Cache()
self.app: web.Application | None = None
services = Services.get_instance()
services: Services = Services.get_instance()
services.add_all(
{
k: v
for k, v in {
"queue": self.queue,
"encoder": self.encoder,
"config": self.config,
"notify": self._notify,

7
app/library/HttpSocket.py
View file

@ -11,7 +11,6 @@ from app.library.Services import Services
from app.library.Utils import load_modules
from .config import Config
from .DownloadQueue import DownloadQueue
from .encoder import Encoder
from .Events import Event, EventBus, Events
from .ItemDTO import Item
@ -26,19 +25,16 @@ class HttpSocket:
config: Config
sio: socketio.AsyncServer
queue: DownloadQueue
di_context: dict[str, Any] = {}
def __init__(
self,
root_path: Path,
queue: DownloadQueue | None = None,
encoder: Encoder | None = None,
config: Config | None = None,
sio: socketio.AsyncServer | None = None,
):
self.config = config or Config.get_instance()
self.queue = queue or DownloadQueue.get_instance()
self._notify = EventBus.get_instance()
self.sio = sio or socketio.AsyncServer(
@ -63,7 +59,6 @@ class HttpSocket:
k: v
for k, v in {
"config": self.config,
"queue": self.queue,
"sio": self.sio,
"encoder": encoder,
"notify": self._notify,
@ -104,7 +99,7 @@ class HttpSocket:
async def event_handler(data: Event, _):
if data and data.data:
await self.queue.add(item=Item.format(data.data))
await Services.get_instance().get("queue").add(item=Item.format(data.data))
self._notify.subscribe(Events.ADD_URL, event_handler, f"{__class__.__name__}.add")

8
app/library/ItemDTO.py
View file

@ -397,11 +397,8 @@ class ItemDTO:
dict: The serialized item.
"""
if "finished" == self.status:
if not self._recomputed:
self.archive_status()
self.get_file_sidecar()
if "finished" == self.status and not self._recomputed:
self.archive_status()
item, _ = clean_item(self.__dict__.copy(), ItemDTO.removed_fields())
return item
@ -661,4 +658,3 @@ class ItemDTO:
self.get_archive_id()
self.get_archive_file()
self.archive_status()
self.get_file_sidecar()

4
app/library/Tasks.py
View file

@ -828,11 +828,11 @@ class HandleTask:
if archive_file and archive_id in downloaded:
continue
if download_queue.queue.exists(url=url):
if await download_queue.queue.exists(url=url):
continue
try:
done = download_queue.done.get(url=url)
done = await download_queue.done.get(url=url)
if "error" != done.info.status:
continue
except KeyError:

15
app/library/config.py
View file

@ -311,20 +311,20 @@ class Config(metaclass=Singleton):
def __init__(self, is_native: bool = False):
baseDefaultPath: str = str(Path(__file__).parent.parent.parent.absolute())
self.is_native = is_native
self.temp_path = os.environ.get("YTP_TEMP_PATH", None) or str(Path(baseDefaultPath) / "var" / "tmp")
self.config_path = os.environ.get("YTP_CONFIG_PATH", None) or str(Path(baseDefaultPath) / "var" / "config")
self.download_path = os.environ.get("YTP_DOWNLOAD_PATH", None) or str(
Path(baseDefaultPath) / "var" / "downloads"
)
self.app_path = Path(__file__).parent.parent.absolute()
envFile: str = Path(self.config_path) / ".env"
if envFile.exists():
logging.info(f"Loading environment variables from '{envFile}'.")
load_dotenv(envFile)
self.is_native = is_native
self.temp_path = os.environ.get("YTP_TEMP_PATH", None) or str(Path(baseDefaultPath) / "var" / "tmp")
self.download_path = os.environ.get("YTP_DOWNLOAD_PATH", None) or str(
Path(baseDefaultPath) / "var" / "downloads"
)
self.app_path = Path(__file__).parent.parent.absolute()
for k, v in self._get_attributes().items():
if k.startswith("_") or k in self._manual_vars:
continue
@ -441,6 +441,7 @@ class Config(metaclass=Singleton):
("urllib3.connectionpool", logging.WARNING),
("apprise", logging.WARNING),
("httpcore", logging.INFO),
("aiosqlite", logging.INFO),
)
for _tool, _level in _log_levels:
logging.getLogger(_tool).setLevel(_level)

321
app/library/migrate.py Normal file
View file

@ -0,0 +1,321 @@
# flake8: noqa: S608
import contextlib
import datetime
import glob
import os.path
import traceback
from collections.abc import AsyncIterator, Iterable, Sequence
from importlib.machinery import ModuleSpec, SourceFileLoader
from importlib.util import module_from_spec, spec_from_loader
from typing import TYPE_CHECKING
import aiosqlite
if TYPE_CHECKING:
from types import ModuleType
UTC_LENGTH = 14
__version__ = "1.0.0"
class Error(Exception):
"""Base class for all Caribou errors."""
class InvalidMigrationError(Error):
"""Thrown when a client migration contains an error."""
class InvalidNameError(Error):
"""Thrown when a client migration has an invalid filename."""
def __init__(self, filename: str):
msg: str = (
f"Migration filenames must start with a UTC timestamp. The following file has an invalid name: {filename}"
)
super().__init__(msg)
@contextlib.asynccontextmanager
async def execute(
conn: aiosqlite.Connection, sql: str, params: Sequence[object] | None = None
) -> AsyncIterator[aiosqlite.Cursor]:
params = [] if params is None else params
cursor = await conn.execute(sql, params)
try:
yield cursor
finally:
await cursor.close()
@contextlib.asynccontextmanager
async def transaction(conn: aiosqlite.Connection) -> AsyncIterator[None]:
try:
yield
await conn.commit()
except Exception:
await conn.rollback()
raise
class Migration:
"""This class represents a migration version."""
def __init__(self, path: str):
self.path: str = path
self.filename: str = os.path.basename(path)
self.module_name, _ = os.path.splitext(self.filename)
self.get_version() # will assert the filename is valid
self.name: str = self.module_name[UTC_LENGTH:]
while self.name.startswith("_"):
self.name: str = self.name[1:]
loader = SourceFileLoader(self.module_name, path)
spec: ModuleSpec | None = spec_from_loader(self.module_name, loader)
if spec is None:
msg = f"Cannot create spec for {path}"
raise ImportError(msg)
try:
module: ModuleType = module_from_spec(spec)
loader.exec_module(module)
self.module: ModuleType = module
except Exception as e:
msg: str = f"Invalid migration {path}: {traceback.format_exc()}"
raise InvalidMigrationError(msg) from e
targets: list[str] = ["upgrade", "downgrade"]
missing: list[str] = [m for m in targets if not self.has_method(m)]
if missing:
msg = f"Migration '{self.path}' is missing required methods: {', '.join(missing)}."
raise InvalidMigrationError(msg)
def get_version(self) -> str:
if len(self.filename) < UTC_LENGTH:
raise InvalidNameError(self.filename)
timestamp: str = self.filename[:UTC_LENGTH]
if not timestamp.isdigit():
raise InvalidNameError(self.filename)
return timestamp
async def upgrade(self, conn: aiosqlite.Connection) -> None:
await self.module.upgrade(conn)
async def downgrade(self, conn: aiosqlite.Connection) -> None:
await self.module.downgrade(conn)
def has_method(self, name: str) -> bool:
return callable(getattr(self.module, name, None))
def __repr__(self) -> str:
return f"Migration(path={self.filename})"
class Database:
def __init__(self, db_url: aiosqlite.Connection | str, version_table: str = "migration_version"):
if not db_url:
msg = "Database requires db_url."
raise ValueError(msg)
self.version_table: str = version_table
self._owns_connection = bool(isinstance(db_url, str))
if self._owns_connection:
self.conn: aiosqlite.Connection | None = None
self.db_url: str = db_url
self._ensure_connection()
else:
self.conn: aiosqlite.Connection = db_url
async def __aenter__(self) -> "Database":
if self._owns_connection:
await self._ensure_connection()
return self
async def __aexit__(self, exc_type, exc, tb) -> None:
if not self._owns_connection:
return
await self.close()
async def close(self) -> None:
if self._owns_connection and self.conn:
await self.conn.close()
self.conn = None
async def _ensure_connection(self) -> None:
if self.conn:
return
self.conn = await aiosqlite.connect(self.db_url)
self.conn.row_factory = aiosqlite.Row
async def is_version_controlled(self) -> bool:
await self._ensure_connection()
assert self.conn
sql = """SELECT * FROM sqlite_master WHERE type = 'table' AND name = ?"""
async with execute(self.conn, sql, [self.version_table]) as cursor:
return bool(await cursor.fetchall())
async def upgrade(self, migrations: list[Migration], target_version: str | None = None) -> None:
await self._ensure_connection()
assert self.conn
if target_version:
_assert_migration_exists(migrations, target_version)
migrations.sort(key=lambda x: x.get_version())
database_version = await self.get_version()
for migration in migrations:
current_version: str = migration.get_version()
if database_version is not None and current_version <= database_version:
continue
if target_version and current_version > target_version:
break
await migration.upgrade(self.conn)
new_version: str = migration.get_version()
await self.update_version(new_version)
database_version: str = new_version
async def downgrade(self, migrations: list[Migration], target_version: str | int) -> None:
await self._ensure_connection()
assert self.conn
if target_version not in (0, "0"):
_assert_migration_exists(migrations, target_version)
migrations.sort(key=lambda x: x.get_version(), reverse=True)
database_version: str | None = await self.get_version()
for i, migration in enumerate(migrations):
current_version: str = migration.get_version()
if database_version is not None and current_version > database_version:
continue
if current_version <= target_version:
break
await migration.downgrade(self.conn)
next_version: str | int = 0
if i < len(migrations) - 1:
next_migration: Migration = migrations[i + 1]
next_version = next_migration.get_version()
await self.update_version(str(next_version))
database_version = str(next_version)
async def get_version(self) -> str | None:
"""
Return the database's version, or None if it is not under version
control.
"""
await self._ensure_connection()
assert self.conn
if not await self.is_version_controlled():
return None
sql: str = f"SELECT version FROM {self.version_table}"
async with execute(self.conn, sql) as cursor:
result = await cursor.fetchall()
return result[0][0] if result else "0"
async def update_version(self, version: str) -> None:
await self._ensure_connection()
assert self.conn
sql: str = f"UPDATE {self.version_table} SET version = ?"
async with transaction(self.conn):
await self.conn.execute(sql, [version])
async def initialize_version_control(self) -> None:
await self._ensure_connection()
assert self.conn
sql: str = f"""CREATE TABLE IF NOT EXISTS {self.version_table} ( version TEXT ) """
async with transaction(self.conn):
await self.conn.execute(sql)
await self.conn.execute(f"INSERT INTO {self.version_table} VALUES (0)")
def __repr__(self) -> str:
return f'Database("{self.db_url if self._owns_connection else "external_connection"}")'
def _assert_migration_exists(migrations: Iterable[Migration], version: str | int) -> None:
if version not in (m.get_version() for m in migrations):
msg: str = f"No migration with version {version} exists."
raise Error(msg)
def load_migrations(directory: str) -> list[Migration]:
"""Return the migrations contained in the given directory."""
directory = str(directory)
if not os.path.exists(directory) or not os.path.isdir(directory):
msg: str = f"{directory} is not a directory."
raise Error(msg)
return [Migration(f) for f in glob.glob(os.path.join(directory, "*.py"))]
async def upgrade(db_url: aiosqlite.Connection | str, migration_dir: str, version: str | None = None) -> None:
"""
Upgrade the given database with the migrations contained in the
migrations directory. If a version is not specified, upgrade
to the most recent version.
"""
async with Database(db_url) as db:
if not await db.is_version_controlled():
await db.initialize_version_control()
await db.upgrade(load_migrations(migration_dir), version)
async def downgrade(db_url: str | aiosqlite.Connection, migration_dir: str, version: str) -> None:
"""
Downgrade the database to the given version with the migrations
contained in the given migration directory.
"""
async with Database(db_url) as db:
if not await db.is_version_controlled():
msg = f"The database {db_url} is not version controlled."
raise Error(msg)
migrations = load_migrations(migration_dir)
await db.downgrade(migrations, version)
async def get_version(db_url: aiosqlite.Connection | str) -> str | None:
"""Return the migration version of the given database."""
async with Database(db_url) as db:
return await db.get_version()
def create_migration(name: str, directory: str | None = None) -> str:
"""
Create a migration with the given name. If no directory is specified,
the current working directory will be used.
"""
directory = directory if directory else "."
if not os.path.exists(directory) or not os.path.isdir(directory):
msg: str = f"{directory} is not a directory."
raise Error(msg)
now: datetime.datetime = datetime.datetime.now(tz=datetime.UTC)
version: str = now.strftime("%Y%m%d%H%M%S")
contents: str = MIGRATION_TEMPLATE % {"name": name, "version": version}
sanitized: str = name.replace(" ", "_")
filename: str = f"{version}_{sanitized}.py"
path: str = os.path.join(directory, filename)
with open(path, "w") as migration_file:
migration_file.write(contents)
return path
MIGRATION_TEMPLATE = """\
\"\"\"
This module contains a db migration.
Migration Name: %(name)s
Migration Version: %(version)s
\"\"\"
async def upgrade(c):
# add your upgrade step here
await c.execute("SELECT 1")
async def downgrade(c):
# add your downgrade step here
await c.execute("SELECT 1")
"""

466
app/library/sqlite_store.py Normal file
View file

@ -0,0 +1,466 @@
import asyncio
import contextlib
import json
import logging
import os
from collections.abc import Iterable
from dataclasses import fields
from datetime import UTC, datetime
from email.utils import formatdate
import aiosqlite
from aiohttp import web
from .ItemDTO import ItemDTO
from .operations import Operation, matches_condition
from .Services import Services
from .Singleton import ThreadSafe
from .Utils import init_class
LOG: logging.Logger = logging.getLogger(__name__)
ITEM_DTO_FIELDS: set[str] = {f.name for f in fields(ItemDTO)}
class Terminator:
pass
class _Op:
"""Queued write operation."""
__slots__ = ("item", "key", "keys", "op", "type_value")
def __init__(
self, op: Terminator | str, type_value: str, item: ItemDTO | None, key: str | None, keys: list[str] | None
):
self.op = op
self.type_value = type_value
self.item = item
self.key = key
self.keys = keys
class SqliteStore(metaclass=ThreadSafe):
@staticmethod
def get_instance(db_path: str | None = None) -> "SqliteStore":
return SqliteStore(db_path=db_path)
def attach(self, app: web.Application):
Services.get_instance().add("sqlite_store", self)
app.on_shutdown.append(self.on_shutdown)
async def on_shutdown(self, _: web.Application):
LOG.debug("Shutting down SqliteStore...")
await self.close()
LOG.debug("SqliteStore shut down complete.")
def __init__(self, db_path: str, *, max_pending: int = 200, flush_interval: float = 0.05):
self._db_path: str = db_path
self._conn: aiosqlite.Connection | None = None
self._queue: asyncio.Queue[_Op] | None = None
self._task: asyncio.Task | None = None
self._flush_interval: float = flush_interval
self._max_pending: int = max_pending
self._lock = asyncio.Lock()
async def __aenter__(self) -> "SqliteStore":
await self._ensure_conn()
return self
async def __aexit__(self, exc_type, exc, tb) -> None:
await self.close()
async def fetch_saved(self, type_value: str) -> list[tuple[str, ItemDTO]]:
await self._ensure_conn()
cursor = await self._conn.execute(
'SELECT "id", "data", "created_at" FROM "history" WHERE "type" = ? ORDER BY "created_at" ASC',
(type_value,),
)
items: list[tuple[str, ItemDTO]] = []
async with cursor:
async for row in cursor:
row_date = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S") # noqa: DTZ007
data = json.loads(row["data"])
data.pop("_id", None)
item = init_class(ItemDTO, data, ITEM_DTO_FIELDS)
item._id = row["id"]
item.datetime = formatdate(row_date.replace(tzinfo=UTC).timestamp())
items.append((row["id"], item))
return items
async def exists(self, type_value: str, key: str | None = None, url: str | None = None) -> bool:
return await self.get(type_value, key=key, url=url) is not None
async def get(self, type_value: str, key: str | None = None, url: str | None = None) -> bool:
if not key and not url:
msg = "key or url must be provided."
raise KeyError(msg)
await self._ensure_conn()
clauses: list[str] = []
params: list[str] = []
if key:
clauses.append('"id" = ?')
params.append(key)
if url:
clauses.append("json_extract(data, '$.url') = ?")
params.append(url)
where_clause = " OR ".join(clauses)
query = f'SELECT "id", "data", "created_at" FROM "history" WHERE "type" = ? AND ({where_clause}) LIMIT 1' # noqa: S608
cursor = await self._conn.execute(query, (type_value, *params))
row = await cursor.fetchone()
if not row:
return None
row_date = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S") # noqa: DTZ007
data = json.loads(row["data"])
data.pop("_id", None)
item = init_class(ItemDTO, data, ITEM_DTO_FIELDS)
item._id = row["id"]
item.datetime = formatdate(row_date.replace(tzinfo=UTC).timestamp())
return item
async def get_by_id(self, type_value: str, id: str) -> ItemDTO | None:
await self._ensure_conn()
cursor = await self._conn.execute(
'SELECT "id", "data", "created_at" FROM "history" WHERE "type" = ? AND "id" = ?',
(type_value, id),
)
row = await cursor.fetchone()
if not row:
return None
row_date = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S") # noqa: DTZ007
data = json.loads(row["data"])
data.pop("_id", None)
item = init_class(ItemDTO, data, ITEM_DTO_FIELDS)
item._id = row["id"]
item.datetime = formatdate(row_date.replace(tzinfo=UTC).timestamp())
return item
async def get_item(self, type_value: str, **kwargs) -> ItemDTO | None:
"""
Return first item of type matching *any* condition.
Mirrors :meth:`DataStore.get_item` semantics: if any provided condition
matches (OR logic) return the first row by creation time. Returns None
when kwargs is empty or no row matches.
"""
if not kwargs:
return None
await self._ensure_conn()
clauses: list[str] = []
params: list[str | float | int] = []
def _safe_key(key: str) -> str | None:
return key if key.replace("_", "").isalnum() else None
for key, raw_value in kwargs.items():
safe_key = _safe_key(key)
if not safe_key:
continue
if isinstance(raw_value, tuple) and len(raw_value) == 2:
operation, value = raw_value
else:
operation, value = Operation.EQUAL, raw_value
if isinstance(operation, str):
try:
operation = Operation(operation)
except ValueError:
operation = Operation.EQUAL
path = f"$.{safe_key}"
json_extract = f"json_extract(data, '{path}')"
if Operation.EQUAL == operation:
clauses.append(f"{json_extract} = ?")
params.append(value)
elif Operation.NOT_EQUAL == operation:
clauses.append(f"{json_extract} != ?")
params.append(value)
elif Operation.CONTAIN == operation:
clauses.append(f"{json_extract} LIKE ? ESCAPE '\\'")
params.append(f"%{value}%")
elif Operation.NOT_CONTAIN == operation:
clauses.append(f"({json_extract} IS NULL OR {json_extract} NOT LIKE ? ESCAPE '\\')")
params.append(f"%{value}%")
elif Operation.STARTS_WITH == operation:
clauses.append(f"{json_extract} LIKE ? ESCAPE '\\'")
params.append(f"{value}%")
elif Operation.ENDS_WITH == operation:
clauses.append(f"{json_extract} LIKE ? ESCAPE '\\'")
params.append(f"%{value}")
elif Operation.GREATER_THAN == operation:
clauses.append(f"{json_extract} > ?")
params.append(value)
elif Operation.LESS_THAN == operation:
clauses.append(f"{json_extract} < ?")
params.append(value)
elif Operation.GREATER_EQUAL == operation:
clauses.append(f"{json_extract} >= ?")
params.append(value)
elif Operation.LESS_EQUAL == operation:
clauses.append(f"{json_extract} <= ?")
params.append(value)
if not clauses:
return None
where_clause = " OR ".join(f"({clause})" for clause in clauses)
query = f'SELECT "id", "data", "created_at" FROM "history" WHERE "type" = ? AND ({where_clause}) ORDER BY "created_at" ASC LIMIT 1' # noqa: S608
cursor = await self._conn.execute(query, (type_value, *params))
row = await cursor.fetchone()
if not row:
return None
row_date = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S") # noqa: DTZ007
data = json.loads(row["data"])
data.pop("_id", None)
item = init_class(ItemDTO, data, ITEM_DTO_FIELDS)
item._id = row["id"]
item.datetime = formatdate(row_date.replace(tzinfo=UTC).timestamp())
return item if any(matches_condition(k, v, item.__dict__) for k, v in kwargs.items()) else None
async def count(self, type_value: str, status_filter: str | None = None) -> int:
await self._ensure_conn()
where_clauses = ['"type" = ?']
query_params: list[str] = [type_value]
if status_filter:
if status_filter.startswith("!"):
status_value = status_filter[1:]
where_clauses.append("json_extract(data, '$.status') != ?")
query_params.append(status_value)
else:
where_clauses.append("json_extract(data, '$.status') = ?")
query_params.append(status_filter)
where_clause = " AND ".join(where_clauses)
count_query = f'SELECT COUNT(*) as count FROM "history" WHERE {where_clause}' # noqa: S608
cursor = await self._conn.execute(count_query, tuple(query_params))
row = await cursor.fetchone()
return row["count"] if row else 0
async def paginate(
self,
type_value: str,
page: int,
per_page: int,
order: str,
status_filter: str | None = None,
) -> tuple[list[tuple[str, ItemDTO]], int, int, int]:
await self._ensure_conn()
where_clauses = ['"type" = ?']
query_params: list[str | int] = [type_value]
if status_filter:
if status_filter.startswith("!"):
status_value = status_filter[1:]
where_clauses.append("json_extract(data, '$.status') != ?")
query_params.append(status_value)
else:
where_clauses.append("json_extract(data, '$.status') = ?")
query_params.append(status_filter)
where_clause = " AND ".join(where_clauses)
count_query = f'SELECT COUNT(*) as count FROM "history" WHERE {where_clause}' # noqa: S608
count_cursor = await self._conn.execute(count_query, tuple(query_params))
count_row = await count_cursor.fetchone()
total_items = count_row["count"] if count_row else 0
total_pages = (total_items + per_page - 1) // per_page if total_items > 0 else 1
if page > total_pages and total_items > 0:
page = total_pages
offset = (page - 1) * per_page
query_params.extend([per_page, offset])
cursor = await self._conn.execute(
f'SELECT "id", "data", "created_at" FROM "history" WHERE {where_clause} ORDER BY "created_at" {order} LIMIT ? OFFSET ?', # noqa: S608
tuple(query_params),
)
items: list[tuple[str, ItemDTO]] = []
async with cursor:
async for row in cursor:
row_date = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S") # noqa: DTZ007
data = json.loads(row["data"])
data.pop("_id", None)
item = init_class(ItemDTO, data, ITEM_DTO_FIELDS)
item._id = row["id"]
item.datetime = formatdate(row_date.replace(tzinfo=UTC).timestamp())
items.append((row["id"], item))
return items, total_items, page, total_pages
async def upsert(self, type_value: str, item: ItemDTO) -> None:
await self._ensure_conn()
await self._upsert_now(type_value, item)
async def delete(self, type_value: str, key: str) -> None:
await self._ensure_conn()
await self._conn.execute('DELETE FROM "history" WHERE "type" = ? AND "id" = ?', (type_value, key))
async def bulk_delete(self, type_value: str, keys: Iterable[str]) -> int:
await self._ensure_conn()
keys_list = list(keys)
if not keys_list:
return 0
placeholders = ",".join("?" for _ in keys_list)
params = [type_value, *keys_list]
cur = await self._conn.execute(
f'DELETE FROM "history" WHERE "type" = ? AND "id" IN ({placeholders})', # noqa: S608
tuple(params),
)
return cur.rowcount if cur else 0
async def commit(self) -> None:
if self._conn:
await self._conn.commit()
async def enqueue_upsert(self, type_value: str, item: ItemDTO) -> None:
await self._enqueue(_Op("upsert", type_value, item, None, None))
async def enqueue_delete(self, type_value: str, key: str) -> None:
await self._enqueue(_Op("delete", type_value, None, key, None))
async def enqueue_bulk_delete(self, type_value: str, keys: Iterable[str]) -> None:
await self._enqueue(_Op("bulk_delete", type_value, None, None, list(keys)))
async def flush(self) -> None:
if self._queue:
await self._queue.join()
if self._conn:
await self._conn.commit()
async def shutdown(self) -> None:
"""Flush pending writes, stop writer task, and commit."""
if self._queue:
try:
await self._queue.put(_Op(Terminator(), "", None, None, None))
LOG.debug("Waiting for SqliteStore queue to drain...")
await asyncio.wait_for(self._queue.join(), timeout=2)
except TimeoutError:
LOG.warning("SqliteStore queue did not drain within timeout; forcing writer shutdown.")
if self._task:
LOG.debug("Waiting for SqliteStore writer task to finish...")
if not self._task.done():
self._task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await asyncio.wait_for(self._task, timeout=2)
if self._conn:
LOG.debug("Committing final changes to SqliteStore...")
await self._conn.commit()
self._task = None
self._queue = None
async def close(self) -> None:
await self.shutdown()
if self._conn:
LOG.debug("Closing SqliteStore connection...")
await self._conn.close()
self._conn = None
async def _enqueue(self, op: _Op) -> None:
self._ensure_worker()
await self._queue.put(op)
def _ensure_worker(self) -> None:
if self._queue is None:
self._queue = asyncio.Queue(maxsize=self._max_pending)
self._task = asyncio.create_task(self._writer(), name="sqlite-store-writer")
async def _writer(self) -> None:
while True:
op = await self._queue.get()
if isinstance(op.op, Terminator):
self._queue.task_done()
break
try:
async with self._lock:
await self._apply(op)
except Exception as ex:
LOG.exception(ex)
finally:
self._queue.task_done()
await asyncio.sleep(self._flush_interval)
async def _apply(self, op: _Op) -> None:
await self._ensure_conn()
if op.op == "upsert" and op.item:
await self._upsert_now(op.type_value, op.item)
elif op.op == "delete" and op.key:
await self._conn.execute('DELETE FROM "history" WHERE "type" = ? AND "id" = ?', (op.type_value, op.key))
elif op.op == "bulk_delete" and op.keys:
placeholders = ",".join("?" for _ in op.keys)
params = [op.type_value, *op.keys]
await self._conn.execute(
f'DELETE FROM "history" WHERE "type" = ? AND "id" IN ({placeholders})', # noqa: S608
tuple(params),
)
await self._conn.commit()
async def _upsert_now(self, type_value: str, item: ItemDTO) -> None:
await self._ensure_conn()
sql = """
INSERT INTO "history" ("id", "type", "url", "data")
VALUES (?, ?, ?, ?)
ON CONFLICT DO UPDATE SET "type" = ?, "url" = ?, "data" = ?, created_at = ?
"""
encoded = item.json()
await self._conn.execute(
sql.strip(),
(
item._id,
type_value,
item.url,
encoded,
type_value,
item.url,
encoded,
datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"),
),
)
async def _ensure_conn(self) -> None:
if self._conn:
return
if not self._db_path:
msg = "No database path specified for SqliteStore."
raise RuntimeError(msg)
from app.library import migrate
from app.main import ROOT_PATH
if not self._db_path.startswith(":memory"):
os.makedirs(os.path.dirname(self._db_path) or ".", exist_ok=True)
self._conn = await aiosqlite.connect(database=self._db_path, isolation_level=None)
self._conn.row_factory = aiosqlite.Row
version = await migrate.get_version(self._conn)
if version:
LOG.debug(f"DB Version: '{version}'.")
await migrate.upgrade(self._conn, ROOT_PATH / "migrations")
if not version:
version = await migrate.get_version(self._conn)
LOG.debug(f"DB Version after initial migration: '{version}'.")
await self._conn.execute("PRAGMA journal_mode=wal")
await self._conn.execute("PRAGMA busy_timeout=5000")
await self._conn.execute("PRAGMA foreign_keys=ON")
await self._conn.commit()

34
app/main.py
View file

@ -9,10 +9,8 @@ if APP_ROOT not in sys.path:
import asyncio
import logging
import sqlite3
from pathlib import Path
import caribou
import magic
from aiohttp import web
@ -28,6 +26,7 @@ from app.library.Notifications import Notification
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
from app.library.TaskDefinitions import TaskDefinitions
from app.library.Tasks import Tasks
@ -43,25 +42,11 @@ class Main:
self._config.set_app_path(str(ROOT_PATH))
self._app = web.Application()
self._app.on_shutdown.append(self.on_shutdown)
self._background_worker = BackgroundWorker()
Services.get_instance().add("app", self._app)
Services.get_instance().add("background_worker", self._background_worker)
self._check_folders()
LOG.debug(f"DB Version: '{caribou.get_version(self._config.db_file)}'.")
caribou.upgrade(self._config.db_file, ROOT_PATH / "migrations")
connection = sqlite3.connect(database=self._config.db_file, isolation_level=None)
connection.row_factory = sqlite3.Row
connection.execute("PRAGMA journal_mode=wal")
async def _close_connection(_):
LOG.debug("Closing database connection.")
connection.close()
LOG.debug("Database connection closed.")
try:
db_file = Path(self._config.db_file)
if "600" != oct(db_file.stat().st_mode)[-3:]:
@ -69,15 +54,12 @@ class Main:
except Exception:
pass
self._queue = DownloadQueue(connection=connection)
self._http = HttpAPI(root_path=ROOT_PATH, queue=self._queue)
self._socket = HttpSocket(root_path=ROOT_PATH, queue=self._queue)
self._app.on_cleanup.append(_close_connection)
self._http = HttpAPI(root_path=ROOT_PATH)
self._socket = HttpSocket(root_path=ROOT_PATH)
def _check_folders(self):
"""Check if the required folders exist and create them if they do not."""
folders = (self._config.download_path, self._config.temp_path, self._config.config_path)
folders: tuple[str, str, str] = (self._config.download_path, self._config.temp_path, self._config.config_path)
for folder in folders:
folder = Path(folder)
@ -124,19 +106,20 @@ class Main:
if self._config.debug:
EventBus.get_instance().debug_enable()
SqliteStore.get_instance(db_path=self._config.db_file).attach(self._app)
BackgroundWorker.get_instance().attach(self._app)
Scheduler.get_instance().attach(self._app)
self._socket.attach(self._app)
self._http.attach(self._app)
self._queue.attach(self._app)
Tasks.get_instance().attach(self._app)
Presets.get_instance().attach(self._app)
Tasks.get_instance().attach(self._app)
Notification.get_instance().attach(self._app)
Conditions.get_instance().attach(self._app)
DLFields.get_instance().attach(self._app)
TaskDefinitions.get_instance().attach(self._app)
self._background_worker.attach(self._app)
DownloadQueue.get_instance().attach(self._app)
EventBus.get_instance().emit(
Events.LOADED,
@ -148,6 +131,7 @@ class Main:
def started(_):
LOG.info("=" * 40)
LOG.info(f"YTPTube {self._config.app_version} - started on http://{host}:{port}{self._config.base_path}")
LOG.info(f"Download path: {self._config.download_path}")
if self._config.is_native:
LOG.info("Running in native mode.")
LOG.info("=" * 40)

16
app/migrations/20231115152938_initial.py
View file

@ -1,12 +1,12 @@
"""
This module contains a Caribou migration.
This module contains a db migration.
Migration Name: initial
Migration Version: 20231115152938
"""
def upgrade(connection):
async def upgrade(c):
sql = """
CREATE TABLE "history" (
"id" TEXT PRIMARY KEY UNIQUE NOT NULL,
@ -16,24 +16,24 @@ def upgrade(connection):
"created_at" TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
"""
connection.execute(sql)
await c.execute(sql)
sql = """
CREATE INDEX "history_type" ON "history" ("type");
"""
connection.execute(sql)
await c.execute(sql)
sql = """
CREATE UNIQUE INDEX "history_url" ON "history" ("url");
"""
connection.execute(sql)
await c.execute(sql)
connection.commit()
await c.commit()
def downgrade(connection):
async def downgrade(c):
sql = """
DROP TABLE IF EXISTS "history";
"""
connection.execute(sql)
await c.execute(sql)

10
app/migrations/20251116173731_add_status_index.py
View file

@ -1,11 +1,11 @@
"""
This module contains a Caribou migration.
This module contains a db migration.
Migration Name: add_status_index
Migration Version: 20251116173731
"""
def upgrade(connection):
async def upgrade(connection):
"""
Add index on json_extract(data, '$.status') for better query performance.
@ -15,13 +15,13 @@ def upgrade(connection):
sql = """
CREATE INDEX IF NOT EXISTS "history_status" ON "history" (json_extract("data", '$.status'));
"""
connection.execute(sql)
await connection.execute(sql)
def downgrade(connection):
async def downgrade(connection):
"""
Remove the status index.
"""
sql = """
DROP INDEX IF EXISTS "history_status";
"""
connection.execute(sql)
await connection.execute(sql)

8
app/routes/api/browser.py
View file

@ -397,12 +397,12 @@ async def path_actions(request: Request, config: Config, queue: DownloadQueue, n
for old_sidecar, new_sidecar in sidecar_renamed:
record(old_sidecar, ok=True, action=action, extra={"new_path": new_sidecar})
if item := queue.done.get_item(filename=str(path.relative_to(config.download_path))):
if item := await queue.done.get_item(filename=str(path.relative_to(config.download_path))):
item.info.filename = str(renamed.relative_to(config.download_path))
if sidecar_renamed:
item.info.get_file_sidecar()
queue.done.put(item)
await queue.done.put(item)
notify.emit(Events.ITEM_UPDATED, data=item.info)
if "delete" == action:
@ -503,12 +503,12 @@ async def path_actions(request: Request, config: Config, queue: DownloadQueue, n
for old_sidecar, new_sidecar in sidecar_moved:
record(old_sidecar, ok=True, action=action, extra={"new_path": new_sidecar})
if item := queue.done.get_item(filename=str(path.relative_to(config.download_path))):
if item := await queue.done.get_item(filename=str(path.relative_to(config.download_path))):
item.info.filename = str(moved.relative_to(config.download_path))
if sidecar_moved:
item.info.get_file_sidecar()
queue.done.put(item)
await queue.done.put(item)
notify.emit(Events.ITEM_UPDATED, data=item.info)
return web.json_response(data=operations_status, status=web.HTTPOk.status_code)

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

@ -47,30 +47,21 @@ async def items_list(request: Request, queue: DownloadQueue, encoder: Encoder, c
"""
from app.library.DataStore import StoreType
store_type = request.query.get("type", "all").lower()
stores: list[str] = ["all", StoreType.QUEUE.value, StoreType.HISTORY.value]
if store_type not in stores:
store_type: str = request.query.get("type", "queue").lower()
try:
store_type: StoreType = StoreType.from_value(store_type)
except ValueError:
return web.json_response(
data={"error": f"type must be one of {', '.join(stores)}."},
data={"error": f"type must be one of {', '.join(StoreType.all())}."},
status=web.HTTPBadRequest.status_code,
)
# Legacy behavior: return all items without pagination, will be removed in future.
if "all" == store_type:
data: dict = {"queue": [], "history": []}
q = queue.get()
data["queue"].extend([q.get("queue", {}).get(k) for k in q.get("queue", {})])
data["history"].extend([q.get("done", {}).get(k) for k in q.get("done", {})])
return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=encoder.encode)
ds = queue.queue if store_type == StoreType.QUEUE.value else queue.done
ds = queue.queue if store_type == StoreType.QUEUE else queue.done
try:
page = int(request.query.get("page", 1))
per_page = int(request.query.get("per_page", config.default_pagination))
order = request.query.get("order", "DESC").upper()
order: str = request.query.get("order", "DESC").upper()
except ValueError:
return web.json_response(
data={"error": "page and per_page must be valid integers."},
@ -92,25 +83,38 @@ async def items_list(request: Request, queue: DownloadQueue, encoder: Encoder, c
status=web.HTTPBadRequest.status_code,
)
# Parse status filter
status_filter = request.query.get("status", None)
items, total, current_page, total_pages = ds.get_items_paginated(
items, total, current_page, total_pages = await ds.get_items_paginated(
page=page, per_page=per_page, order=order, status_filter=status_filter
)
data = {
"pagination": {
"page": current_page,
"per_page": per_page,
"total": total,
"total_pages": total_pages,
"has_next": current_page < total_pages,
"has_prev": current_page > 1,
},
"items": [item for _, item in items],
}
return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=encoder.encode)
if store_type == StoreType.HISTORY:
for _, download in items:
if not download.info:
continue
try:
download.info.sidecar = download.get_file_sidecar()
except Exception:
download.info.sidecar = {}
return web.json_response(
data={
"type": store_type.value,
"pagination": {
"page": current_page,
"per_page": per_page,
"total": total,
"total_pages": total_pages,
"has_next": current_page < total_pages,
"has_prev": current_page > 1,
},
"items": [download.info for _, download in items],
},
status=web.HTTPOk.status_code,
dumps=encoder.encode,
)
@route("DELETE", "api/history/", "items_delete")
@ -137,24 +141,26 @@ async def items_delete(request: Request, queue: DownloadQueue, encoder: Encoder)
ids = data.get("ids")
remove_file: bool = bool(data.get("remove_file", True))
_type = data.get("type", data.get("where"))
storeType = data.get("type", data.get("where"))
if not _type:
if not storeType:
return web.json_response(data={"error": "Type is required."}, status=web.HTTPBadRequest.status_code)
if _type not in [StoreType.QUEUE.value, StoreType.HISTORY.value]:
try:
storeType: StoreType = StoreType.from_value(storeType)
except ValueError:
return web.json_response(
data={"error": f"type must be '{StoreType.QUEUE.value}' or '{StoreType.HISTORY.value}'."},
data={"error": f"type must be one of '{StoreType.all()}'."},
status=web.HTTPBadRequest.status_code,
)
ds = queue.queue if _type == StoreType.QUEUE.value else queue.done
ds = queue.queue if storeType == StoreType.QUEUE else queue.done
if ids:
return web.json_response(
data={
"items": await (
queue.cancel(ids) if _type == StoreType.QUEUE.value else queue.clear(ids, remove_file=remove_file)
queue.cancel(ids) if storeType == StoreType.QUEUE else queue.clear(ids, remove_file=remove_file)
),
"deleted": len(ids),
},
@ -174,7 +180,7 @@ async def items_delete(request: Request, queue: DownloadQueue, encoder: Encoder)
per_page = 1000
while True:
items, _, current_page, total_pages = ds.get_items_paginated(
items, _, current_page, total_pages = await ds.get_items_paginated(
page=page, per_page=per_page, order="DESC", status_filter=status_filter
)
@ -192,7 +198,7 @@ async def items_delete(request: Request, queue: DownloadQueue, encoder: Encoder)
data={
"items": await (
queue.cancel(items_to_delete)
if _type == StoreType.QUEUE.value
if storeType == StoreType.QUEUE
else queue.clear(items_to_delete, remove_file=remove_file)
),
"deleted": len(items_to_delete),
@ -221,7 +227,7 @@ async def item_view(request: Request, queue: DownloadQueue, encoder: Encoder) ->
if not id:
return web.json_response(data={"error": "id is required."}, status=web.HTTPBadRequest.status_code)
item: Download | None = queue.done.get_by_id(id) or queue.queue.get_by_id(id)
item: Download | None = await queue.done.get_by_id(id) or await queue.queue.get_by_id(id)
if not item:
return web.json_response(data={"error": "item not found."}, status=web.HTTPNotFound.status_code)
@ -231,6 +237,7 @@ async def item_view(request: Request, queue: DownloadQueue, encoder: Encoder) ->
info: dict = {
**item.info.serialize(),
"ffprobe": {},
"sidecar": {},
}
if "finished" == item.info.status and (filename := item.info.get_file()):
@ -241,6 +248,11 @@ async def item_view(request: Request, queue: DownloadQueue, encoder: Encoder) ->
except Exception:
pass
try:
info["sidecar"] = item.info.get_file_sidecar()
except Exception:
pass
return web.json_response(data=info, status=web.HTTPOk.status_code, dumps=encoder.encode)
@ -263,7 +275,7 @@ async def item_update(request: Request, queue: DownloadQueue, encoder: Encoder,
if not id:
return web.json_response(data={"error": "id is required."}, status=web.HTTPBadRequest.status_code)
item: Download | None = queue.done.get_by_id(id)
item: Download | None = await queue.done.get_by_id(id)
if not item:
return web.json_response(data={"error": "item not found."}, status=web.HTTPNotFound.status_code)
@ -285,7 +297,7 @@ async def item_update(request: Request, queue: DownloadQueue, encoder: Encoder,
LOG.debug(f"Updated '{k}' to '{v}' for '{item.info.id}'")
if updated:
queue.done.put(item)
await queue.done.put(item)
notify.emit(Events.ITEM_UPDATED, data=item.info)
return web.json_response(
@ -394,7 +406,7 @@ async def item_archive_add(request: Request, queue: DownloadQueue, notify: Event
return web.json_response(data={"error": "id is required."}, status=web.HTTPBadRequest.status_code)
try:
item: Download | None = queue.done.get_by_id(id)
item: Download | None = await queue.done.get_by_id(id)
if not item:
return web.json_response(data={"error": f"item '{id}' not found."}, status=web.HTTPNotFound.status_code)
except KeyError:
@ -425,7 +437,7 @@ async def item_archive_add(request: Request, queue: DownloadQueue, notify: Event
)
item.info.archive_status(force=True)
queue.done.put(item, no_notify=True)
await queue.done.put(item, no_notify=True)
notify.emit(Events.ITEM_UPDATED, data=item.info)
return web.json_response(
@ -452,7 +464,7 @@ async def item_archive_delete(request: Request, queue: DownloadQueue, notify: Ev
return web.json_response(data={"error": "id is required."}, status=web.HTTPBadRequest.status_code)
try:
item: Download | None = queue.done.get_by_id(id)
item: Download | None = await queue.done.get_by_id(id)
if not item:
return web.json_response(data={"error": f"item '{id}' not found."}, status=web.HTTPNotFound.status_code)
except KeyError:
@ -483,7 +495,7 @@ async def item_archive_delete(request: Request, queue: DownloadQueue, notify: Ev
)
item.info.archive_status(force=True)
queue.done.put(item, no_notify=True)
await queue.done.put(item, no_notify=True)
notify.emit(Events.ITEM_UPDATED, data=item.info)
return web.json_response(

12
app/routes/socket/connection.py
View file

@ -44,14 +44,22 @@ async def connect(config: Config, queue: DownloadQueue, notify: EventBus, sid: s
base=Path(config.download_path),
depth_limit=config.download_path_depth - 1,
),
"history_count": queue.done.get_total_count(),
"queue": queue.get("queue")["queue"],
"history_count": await queue.done.get_total_count(),
"queue": (await queue.get("queue"))["queue"],
},
title="Sending initial download data",
message=f"Sending initial download data to client '{sid}'.",
to=sid,
)
notify.emit(
Events.ACTIVE_QUEUE,
data={"queue": (await queue.get("queue"))["queue"]},
title="Sending initial active queue data",
message=f"Sending active queue data to client '{sid}'.",
to=sid,
)
@route(RouteType.SOCKET, "disconnect", "socket_disconnect")
async def disconnect(sio: socketio.AsyncServer, sid: str, data: str = None):

147
app/scripts/create_migration.py Normal file
View file

@ -0,0 +1,147 @@
#!/usr/bin/env python3
"""
Caribou migrations CLI.
Provides commands to create, list, and apply database migrations.
"""
from __future__ import annotations
import argparse
import sys
import traceback
from pathlib import Path
from typing import TYPE_CHECKING
APP_ROOT: Path = Path(__file__).resolve().parents[2]
if str(APP_ROOT) not in sys.path:
sys.path.insert(0, str(APP_ROOT))
from app.library import migrate
if TYPE_CHECKING:
from collections.abc import Callable
def e_print(msg: str = "") -> None:
sys.stderr.write(f"{msg}\n")
def o_print(msg: str = "") -> None:
sys.stdout.write(f"{msg}\n")
def cmd_info(_: argparse.Namespace) -> None:
o_print(f"Version: {migrate.__version__}")
def cmd_create(args: argparse.Namespace) -> None:
path = migrate.create_migration(args.name, args.migration_dir)
o_print(f"created migration {path}")
def cmd_version(args: argparse.Namespace) -> None:
version = migrate.get_version(args.database_path)
if version:
o_print(f"the db [{args.database_path}] is at version {version}")
else:
o_print(f"the db [{args.database_path}] is not under version control")
def cmd_upgrade(args: argparse.Namespace) -> None:
if args.version:
o_print(f"upgrading db [{args.database_path}] to version [{args.version}]")
else:
o_print(f"upgrading db [{args.database_path}] to most recent version")
migrate.upgrade(args.database_path, args.migration_dir, args.version)
new_version = migrate.get_version(args.database_path)
if args.version is not None:
assert new_version == args.version
o_print(f"upgraded [{args.database_path}] successfully to version [{new_version}]")
def cmd_downgrade(args: argparse.Namespace) -> None:
o_print(f"downgrading db [{args.database_path}] to version [{args.version}]")
migrate.downgrade(args.database_path, args.migration_dir, args.version)
o_print(f"downgraded [{args.database_path}] successfully to version [{args.version}]")
def cmd_list(args: argparse.Namespace) -> None:
o_print(f"Migrations in [{args.migration_dir}]:\n")
for m in migrate.load_migrations(args.migration_dir):
o_print(f"{m.get_version()}\t{m.name}\t{m.path}")
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(prog="caribou")
sub = p.add_subparsers(dest="command", required=True)
info = sub.add_parser("info", help="print library version")
info.set_defaults(func=cmd_info)
create = sub.add_parser("create", help="create a new migration file")
create.add_argument("name", help="the name of migration")
create.add_argument("-d", "--migration-dir", default=".", help="the migration directory")
create.set_defaults(func=cmd_create)
ver = sub.add_parser("version", help="return the migration version of the database")
ver.add_argument("database_path", help="path to the sqlite database")
ver.set_defaults(func=cmd_version)
up = sub.add_parser(
"upgrade",
help="upgrade the db (if version isn't specified, upgrade to the most recent)",
)
up.add_argument("database_path", help="path to the sqlite database")
up.add_argument("migration_dir", help="the migration directory")
up.add_argument("-v", "--version", type=int, default=None, help="the target migration version")
up.set_defaults(func=cmd_upgrade)
down = sub.add_parser(
"downgrade",
help="downgrade the db to a particular version (use 0 to rollback all changes)",
)
down.add_argument("database_path", help="path to the sqlite database")
down.add_argument("migration_dir", help="the migration directory")
down.add_argument("version", type=int, help="the target migration version")
down.set_defaults(func=cmd_downgrade)
lst = sub.add_parser("list", help="list the migration versions")
lst.add_argument("migration_dir", help="the migration directory")
lst.set_defaults(func=cmd_list)
return p
def run(argv: list[str]) -> int:
parser = build_parser()
args = parser.parse_args(argv)
try:
func: Callable[[argparse.Namespace], None] = args.func
func(args)
return 0
except migrate.InvalidMigrationError:
e_print(traceback.format_exc())
return 1
except migrate.Error as err:
e_print(f"Error: {err}")
return 1
except Exception:
e_print("an unexpected error occurred:")
e_print(traceback.format_exc())
return 1
def main() -> int:
return run(sys.argv[1:])
if __name__ == "__main__":
raise SystemExit(main())

15
app/scripts/seed_db.py
View file

@ -28,6 +28,7 @@ LOG = logging.getLogger("seed_db")
USED_IDS: set[str] = set()
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Seed history/queue with synthetic records for performance testing.")
parser.add_argument(
@ -141,13 +142,21 @@ def _build_row(
"live_in": None,
"file_size": stat.st_size,
"options": {},
"extras": {},
"extras": {
"source_name": "Tester",
"source_id": "0ddc40f3-9227-4ffa-9f3c-c52bcf33ef2f",
"source_handler": "YoutubeHandler",
"metadata": {
"published": formatdate(timestamp.timestamp()),
},
"is_video": True,
"is_audio": True,
},
"cli": "",
"auto_start": True,
"is_archivable": True,
"is_archived": True,
"archive_id": archive_id,
"sidecar": {},
"tmpfilename": None,
"filename": filename,
"total_bytes": stat.st_size,
@ -157,8 +166,6 @@ def _build_row(
"percent": 100,
"speed": None,
"eta": None,
"_recomputed": True,
"_archive_file": None,
}
created_at = timestamp.strftime("%Y-%m-%d %H:%M:%S")

695
app/tests/test_datastore.py
View file

File diff suppressed because it is too large Load diff

537
app/tests/test_datastore_pagination.py
View file

@ -1,257 +1,212 @@
import json
import sqlite3
from datetime import UTC, datetime
from pathlib import Path
from tempfile import TemporaryDirectory
import pytest
import pytest_asyncio
from app.library.DataStore import DataStore, StoreType
from app.library.Download import Download
from app.library.ItemDTO import ItemDTO
from app.library.sqlite_store import SqliteStore
async def make_db(data: int = 100) -> SqliteStore:
"""Create a temporary database with test data."""
SqliteStore._reset_singleton()
ins = SqliteStore.get_instance(db_path=":memory:")
await ins._ensure_conn()
base_time = datetime.now(UTC)
for i in range(data):
created_at = base_time.replace(hour=(i // 4) % 24, minute=(i * 15) % 60, second=i % 60).strftime(
"%Y-%m-%d %H:%M:%S"
)
item_data = {
"url": f"https://example.com/video{i}",
"title": f"Test Video {i}",
"id": f"video{i}",
"folder": "/downloads",
"status": "finished",
}
await ins._conn.execute(
'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)',
(
f"test-id-{i}",
str(StoreType.HISTORY),
item_data["url"],
json.dumps(item_data),
created_at,
),
)
if data > 0:
await ins._conn.commit()
return ins
@pytest.mark.asyncio
class TestDataStorePagination:
"""Test pagination functionality of DataStore."""
@pytest.fixture
def temp_db(self):
"""Create a temporary database with test data."""
with TemporaryDirectory() as tmpdir:
db_path = Path(tmpdir) / "test.db"
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row
@pytest_asyncio.fixture
async def make_db(self, data: int = 100):
"""Fixture to provide a temporary database with test data."""
return await make_db(data=data)
# Create the history table
conn.execute(
"""
CREATE TABLE IF NOT EXISTS "history" (
"id" TEXT PRIMARY KEY,
"type" TEXT NOT NULL,
"url" TEXT NOT NULL,
"data" TEXT NOT NULL,
"created_at" TEXT NOT NULL
)
"""
)
# Insert test data
base_time = datetime.now(UTC)
for i in range(100):
created_at = base_time.replace(
hour=(i // 4) % 24, minute=(i * 15) % 60, second=i % 60
).strftime("%Y-%m-%d %H:%M:%S")
item_data = {
"url": f"https://example.com/video{i}",
"title": f"Test Video {i}",
"id": f"video{i}",
"folder": "/downloads",
"status": "finished",
}
conn.execute(
'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)',
(
f"test-id-{i}",
str(StoreType.HISTORY),
item_data["url"],
json.dumps(item_data),
created_at,
),
)
conn.commit()
yield conn
conn.close()
def test_get_total_count(self, temp_db):
async def test_get_total_count(self):
"""Test getting total count of items."""
datastore = DataStore(type=StoreType.HISTORY, connection=temp_db)
count = datastore.get_total_count()
assert count == 100
db = await make_db(data=100)
try:
datastore = DataStore(type=StoreType.HISTORY, connection=db)
count = await datastore.get_total_count()
assert count == 100
finally:
await db.close()
def test_pagination_basic(self, temp_db):
async def test_pagination_basic(self):
"""Test basic pagination functionality."""
datastore = DataStore(type=StoreType.HISTORY, connection=temp_db)
db = await make_db(data=100)
try:
datastore = DataStore(type=StoreType.HISTORY, connection=db)
items, total, page, total_pages = await datastore.get_items_paginated(page=1, per_page=10)
assert len(items) == 10
assert total == 100
assert page == 1
assert total_pages == 10
for _, item in items:
assert isinstance(item, Download)
assert isinstance(item.info, ItemDTO)
assert item.info._id.startswith("test-id-")
finally:
await db.close()
# Get first page
items, total, page, total_pages = datastore.get_items_paginated(page=1, per_page=10)
assert len(items) == 10
assert total == 100
assert page == 1
assert total_pages == 10
# Verify items are ItemDTO instances
for _item_id, item in items:
assert isinstance(item, ItemDTO)
assert item._id.startswith("test-id-")
def test_pagination_last_page(self, temp_db):
async def test_pagination_last_page(self):
"""Test pagination on last page."""
datastore = DataStore(type=StoreType.HISTORY, connection=temp_db)
db = await make_db(data=100)
try:
datastore = DataStore(type=StoreType.HISTORY, connection=db)
items, total, page, total_pages = await datastore.get_items_paginated(page=10, per_page=10)
assert len(items) == 10
assert total == 100
assert page == 10
assert total_pages == 10
finally:
await db.close()
# Get last page
items, total, page, total_pages = datastore.get_items_paginated(page=10, per_page=10)
assert len(items) == 10
assert total == 100
assert page == 10
assert total_pages == 10
def test_pagination_partial_page(self, temp_db):
async def test_pagination_partial_page(self):
"""Test pagination with partial last page."""
datastore = DataStore(type=StoreType.HISTORY, connection=temp_db)
db = await make_db(data=100)
try:
datastore = DataStore(type=StoreType.HISTORY, connection=db)
items, total, page, total_pages = await datastore.get_items_paginated(page=4, per_page=30)
assert len(items) == 10 # 100 items / 30 per page = 3 full pages + 10 items
assert total == 100
assert page == 4
assert total_pages == 4
finally:
await db.close()
# Get items with per_page that doesn't divide evenly
items, total, page, total_pages = datastore.get_items_paginated(page=4, per_page=30)
assert len(items) == 10 # 100 items / 30 per page = 3 full pages + 10 items
assert total == 100
assert page == 4
assert total_pages == 4
def test_pagination_out_of_range(self, temp_db):
async def test_pagination_out_of_range(self):
"""Test pagination with page number out of range."""
datastore = DataStore(type=StoreType.HISTORY, connection=temp_db)
db = await make_db(data=100)
try:
datastore = DataStore(type=StoreType.HISTORY, connection=db)
items, total, page, total_pages = await datastore.get_items_paginated(page=999, per_page=10)
assert len(items) == 10
assert total == 100
assert page == 10 # Adjusted to last page
assert total_pages == 10
finally:
await db.close()
# Request page beyond total pages - should return last page
items, total, page, total_pages = datastore.get_items_paginated(page=999, per_page=10)
assert len(items) == 10
assert total == 100
assert page == 10 # Adjusted to last page
assert total_pages == 10
def test_pagination_order_desc(self, temp_db):
async def test_pagination_order_desc(self):
"""Test pagination with descending order (newest first)."""
datastore = DataStore(type=StoreType.HISTORY, connection=temp_db)
db = await make_db(data=100)
try:
datastore = DataStore(type=StoreType.HISTORY, connection=db)
items, _, _, _ = await datastore.get_items_paginated(page=1, per_page=5, order="DESC")
assert len(items) == 5
finally:
await db.close()
items, _, _, _ = datastore.get_items_paginated(page=1, per_page=5, order="DESC")
assert len(items) == 5
# Verify order - should be newest to oldest
# (note: exact order depends on the timestamp generation in fixture)
def test_pagination_order_asc(self, temp_db):
async def test_pagination_order_asc(self):
"""Test pagination with ascending order (oldest first)."""
datastore = DataStore(type=StoreType.HISTORY, connection=temp_db)
db = await make_db(data=100)
try:
datastore = DataStore(type=StoreType.HISTORY, connection=db)
items, _, _, _ = await datastore.get_items_paginated(page=1, per_page=5, order="ASC")
assert len(items) == 5
finally:
await db.close()
items, _, _, _ = datastore.get_items_paginated(page=1, per_page=5, order="ASC")
assert len(items) == 5
def test_pagination_invalid_page(self, temp_db):
async def test_pagination_invalid_page(self):
"""Test pagination with invalid page number."""
datastore = DataStore(type=StoreType.HISTORY, connection=temp_db)
db = await make_db(data=100)
try:
datastore = DataStore(type=StoreType.HISTORY, connection=db)
with pytest.raises(ValueError, match="page must be >= 1"):
await datastore.get_items_paginated(page=0, per_page=10)
with pytest.raises(ValueError, match="page must be >= 1"):
datastore.get_items_paginated(page=0, per_page=10)
with pytest.raises(ValueError, match="page must be >= 1"):
await datastore.get_items_paginated(page=-1, per_page=10)
finally:
await db.close()
with pytest.raises(ValueError, match="page must be >= 1"):
datastore.get_items_paginated(page=-1, per_page=10)
def test_pagination_invalid_per_page(self, temp_db):
async def test_pagination_invalid_per_page(self):
"""Test pagination with invalid per_page value."""
datastore = DataStore(type=StoreType.HISTORY, connection=temp_db)
db = await make_db(data=100)
try:
datastore = DataStore(type=StoreType.HISTORY, connection=db)
with pytest.raises(ValueError, match="per_page must be >= 1"):
datastore.get_items_paginated(page=1, per_page=0)
with pytest.raises(ValueError, match="per_page must be >= 1"):
await datastore.get_items_paginated(page=1, per_page=0)
with pytest.raises(ValueError, match="per_page must be >= 1"):
datastore.get_items_paginated(page=1, per_page=-10)
with pytest.raises(ValueError, match="per_page must be >= 1"):
await datastore.get_items_paginated(page=1, per_page=-10)
finally:
await db.close()
def test_pagination_invalid_order(self, temp_db):
async def test_pagination_invalid_order(self):
"""Test pagination with invalid order parameter."""
datastore = DataStore(type=StoreType.HISTORY, connection=temp_db)
db = await make_db(data=100)
try:
datastore = DataStore(type=StoreType.HISTORY, connection=db)
with pytest.raises(ValueError, match="order must be 'ASC' or 'DESC'"):
await datastore.get_items_paginated(page=1, per_page=10, order="INVALID")
finally:
await db.close()
with pytest.raises(ValueError, match="order must be 'ASC' or 'DESC'"):
datastore.get_items_paginated(page=1, per_page=10, order="INVALID")
def test_pagination_empty_store(self):
async def test_pagination_empty_store(self):
"""Test pagination with empty datastore."""
with TemporaryDirectory() as tmpdir:
db_path = Path(tmpdir) / "empty.db"
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row
# Create empty table
conn.execute(
"""
CREATE TABLE IF NOT EXISTS "history" (
"id" TEXT PRIMARY KEY,
"type" TEXT NOT NULL,
"url" TEXT NOT NULL,
"data" TEXT NOT NULL,
"created_at" TEXT NOT NULL
)
"""
)
conn.commit()
datastore = DataStore(type=StoreType.HISTORY, connection=conn)
items, total, page, total_pages = datastore.get_items_paginated(page=1, per_page=10)
db = await make_db(data=0)
try:
datastore = DataStore(type=StoreType.HISTORY, connection=db)
items, total, page, total_pages = await datastore.get_items_paginated(page=1, per_page=10)
assert len(items) == 0
assert total == 0
assert page == 1
assert total_pages == 1
finally:
await db.close()
conn.close()
def test_pagination_single_item(self):
async def test_pagination_single_item(self):
"""Test pagination with single item."""
with TemporaryDirectory() as tmpdir:
db_path = Path(tmpdir) / "single.db"
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row
# Create table with single item
conn.execute(
"""
CREATE TABLE IF NOT EXISTS "history" (
"id" TEXT PRIMARY KEY,
"type" TEXT NOT NULL,
"url" TEXT NOT NULL,
"data" TEXT NOT NULL,
"created_at" TEXT NOT NULL
)
"""
)
item_data = {
"url": "https://example.com/single",
"title": "Single Video",
"id": "single-video",
"folder": "/downloads",
"status": "finished",
}
conn.execute(
'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)',
(
"single-id",
str(StoreType.HISTORY),
item_data["url"],
json.dumps(item_data),
datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"),
),
)
conn.commit()
datastore = DataStore(type=StoreType.HISTORY, connection=conn)
items, total, page, total_pages = datastore.get_items_paginated(page=1, per_page=10)
db = await make_db(data=1)
try:
datastore = DataStore(type=StoreType.HISTORY, connection=db)
items, total, page, total_pages = await datastore.get_items_paginated(page=1, per_page=10)
assert len(items) == 1
assert total == 1
assert page == 1
assert total_pages == 1
finally:
await db.close()
conn.close()
def test_pagination_status_filter_include(self, temp_db):
async def test_pagination_status_filter_include(self):
"""Test pagination with status filter (inclusion)."""
# Add some items with different status values
item_data_pending = {
@ -269,50 +224,51 @@ class TestDataStorePagination:
"status": "downloading",
}
temp_db.execute(
'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)',
(
"pending-id",
str(StoreType.HISTORY),
item_data_pending["url"],
json.dumps(item_data_pending),
datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"),
),
)
temp_db.execute(
'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)',
(
"downloading-id",
str(StoreType.HISTORY),
item_data_downloading["url"],
json.dumps(item_data_downloading),
datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"),
),
)
temp_db.commit()
db = await make_db(data=100)
try:
await db._conn.execute(
'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)',
(
"pending-id",
str(StoreType.HISTORY),
item_data_pending["url"],
json.dumps(item_data_pending),
datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"),
),
)
await db._conn.execute(
'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)',
(
"downloading-id",
str(StoreType.HISTORY),
item_data_downloading["url"],
json.dumps(item_data_downloading),
datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"),
),
)
await db._conn.commit()
datastore = DataStore(type=StoreType.HISTORY, connection=db)
datastore = DataStore(type=StoreType.HISTORY, connection=temp_db)
# Filter for finished items only
items, total, _, __ = await datastore.get_items_paginated(page=1, per_page=50, status_filter="finished")
# Filter for finished items only
items, total, _page, _total_pages = datastore.get_items_paginated(
page=1, per_page=50, status_filter="finished"
)
assert len(items) == 50 # First page of finished items
assert total == 100 # Only 100 finished items in fixture
for _item_id, item in items:
assert item.info.status == "finished"
assert len(items) == 50 # First page of finished items
assert total == 100 # Only 100 finished items in fixture
for _item_id, item in items:
assert item.status == "finished"
# Filter for pending items only
items, total, _page, _total_pages = await datastore.get_items_paginated(
page=1, per_page=50, status_filter="pending"
)
# Filter for pending items only
items, total, _page, _total_pages = datastore.get_items_paginated(
page=1, per_page=50, status_filter="pending"
)
assert len(items) == 1
assert total == 1
assert items[0][1].info.status == "pending"
finally:
await db.close()
assert len(items) == 1
assert total == 1
assert items[0][1].status == "pending"
def test_pagination_status_filter_exclude(self, temp_db):
async def test_pagination_status_filter_exclude(self):
"""Test pagination with status filter (exclusion)."""
# Add some items with different status values
item_data_pending = {
@ -329,55 +285,62 @@ class TestDataStorePagination:
"folder": "/downloads",
"status": "error",
}
db = await make_db(data=0)
try:
await db._conn.execute(
'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)',
(
"pending-id-2",
str(StoreType.HISTORY),
item_data_pending["url"],
json.dumps(item_data_pending),
datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"),
),
)
await db._conn.execute(
'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)',
(
"error-id",
str(StoreType.HISTORY),
item_data_error["url"],
json.dumps(item_data_error),
datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"),
),
)
await db._conn.commit()
temp_db.execute(
'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)',
(
"pending-id-2",
str(StoreType.HISTORY),
item_data_pending["url"],
json.dumps(item_data_pending),
datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"),
),
)
temp_db.execute(
'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)',
(
"error-id",
str(StoreType.HISTORY),
item_data_error["url"],
json.dumps(item_data_error),
datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"),
),
)
temp_db.commit()
datastore = DataStore(type=StoreType.HISTORY, connection=db)
datastore = DataStore(type=StoreType.HISTORY, connection=temp_db)
# Exclude finished items
items, total, _page, _total_pages = await datastore.get_items_paginated(
page=1, per_page=50, status_filter="!finished"
)
# Exclude finished items
items, total, _page, _total_pages = datastore.get_items_paginated(
page=1, per_page=50, status_filter="!finished"
)
assert total == 2 # Only 2 non-finished items
assert len(items) == 2
for _item_id, item in items:
assert item.info.status != "finished"
assert total == 2 # Only 2 non-finished items
assert len(items) == 2
for _item_id, item in items:
assert item.status != "finished"
# Verify we have pending and error
statuses = {item.info.status for _, item in items}
assert statuses == {"pending", "error"}
finally:
await db.close()
# Verify we have pending and error
statuses = {item.status for _, item in items}
assert statuses == {"pending", "error"}
def test_pagination_status_filter_none_matching(self, temp_db):
async def test_pagination_status_filter_none_matching(self):
"""Test pagination with status filter that matches no items."""
datastore = DataStore(type=StoreType.HISTORY, connection=temp_db)
db = await make_db(data=0)
try:
datastore = DataStore(type=StoreType.HISTORY, connection=db)
# Filter for a status that doesn't exist
items, total, page, total_pages = datastore.get_items_paginated(
page=1, per_page=50, status_filter="nonexistent"
)
# Filter for a status that doesn't exist
items, total, page, total_pages = await datastore.get_items_paginated(
page=1, per_page=50, status_filter="nonexistent"
)
assert len(items) == 0
assert total == 0
assert page == 1
assert total_pages == 1
assert len(items) == 0
assert total == 0
assert page == 1
assert total_pages == 1
finally:
await db.close()

256
app/tests/test_sqlite_store.py Normal file
View file

@ -0,0 +1,256 @@
from datetime import UTC, datetime, timedelta
import pytest
from app.library.ItemDTO import ItemDTO
from app.library.operations import Operation
from app.library.sqlite_store import SqliteStore
async def make_store() -> SqliteStore:
"""Create an isolated in-memory SqliteStore instance with schema."""
SqliteStore._reset_singleton()
store = SqliteStore.get_instance(db_path=":memory:")
await store._ensure_conn()
assert store._conn is not None
return store
def make_item(idx: int, *, status: str = "finished") -> ItemDTO:
return ItemDTO(
id=f"vid{idx}",
title=f"Video {idx}",
url=f"https://example.com/{idx}",
folder="/downloads",
status=status,
)
@pytest.mark.asyncio
async def test_enqueue_upsert_and_fetch_saved():
store = await make_store()
item = make_item(1)
await store.enqueue_upsert("queue", item)
await store.flush()
saved = await store.fetch_saved("queue")
assert len(saved) == 1
key, loaded = saved[0]
assert key == item._id
assert loaded.title == item.title
await store.close()
@pytest.mark.asyncio
async def test_delete_and_bulk_delete():
store = await make_store()
items = [make_item(i) for i in range(3)]
for itm in items:
await store.enqueue_upsert("queue", itm)
await store.flush()
await store.enqueue_delete("queue", items[0]._id)
await store.enqueue_bulk_delete("queue", [items[1]._id])
await store.flush()
remaining = await store.fetch_saved("queue")
assert len(remaining) == 1
assert remaining[0][0] == items[2]._id
await store.close()
@pytest.mark.asyncio
async def test_count_with_status_filters():
store = await make_store()
finished_items = [make_item(i, status="finished") for i in range(2)]
pending_items = [make_item(i + 10, status="pending") for i in range(3)]
for itm in finished_items + pending_items:
await store.enqueue_upsert("history", itm)
await store.flush()
assert await store.count("history", status_filter="finished") == 2
assert await store.count("history", status_filter="pending") == 3
assert await store.count("history", status_filter="!finished") == 3
await store.close()
@pytest.mark.asyncio
async def test_paginate_order_and_bounds():
store = await make_store()
base = datetime(2024, 1, 1, tzinfo=UTC)
_list: list[ItemDTO] = []
for i in range(15):
itm = make_item(i)
encoded = itm.json()
created_at = (base + timedelta(minutes=i)).strftime("%Y-%m-%d %H:%M:%S")
_list.append(itm)
await store._conn.execute(
'INSERT INTO "history" ("id", "type", "url", "data", "created_at") VALUES (?, ?, ?, ?, ?)',
(itm._id, "history", itm.url, encoded, created_at),
)
await store._conn.commit()
items, total, page, total_pages = await store.paginate("history", page=1, per_page=5, order="ASC")
assert total == 15
assert page == 1
assert total_pages == 3
assert items[0][0] == _list[0]._id
items_desc, *_ = await store.paginate("history", page=1, per_page=3, order="DESC")
assert len(items_desc) == 3
assert items_desc[0][0] == _list[14]._id
await store.close()
@pytest.mark.asyncio
async def test_get_by_id():
store = await make_store()
item = make_item(99)
await store.enqueue_upsert("queue", item)
await store.flush()
assert item._id == (await store.get_by_id("queue", item._id))._id
await store.close()
@pytest.mark.asyncio
async def test_shutdown_closes_worker_queue():
store = await make_store()
await store.enqueue_upsert("queue", make_item(99))
await store.shutdown()
assert store._queue is None
await store.close()
@pytest.mark.asyncio
async def test_enqueue_delete_removes_row():
store = await make_store()
item = make_item(5)
await store.enqueue_upsert("queue", item)
await store.flush()
await store.enqueue_delete("queue", item._id)
await store.flush()
rows = await store.fetch_saved("queue")
assert rows == []
await store.close()
@pytest.mark.asyncio
async def test_enqueue_bulk_delete_returns_count_and_bulk_path():
store = await make_store()
items = [make_item(i) for i in range(3)]
for itm in items:
await store.enqueue_upsert("history", itm)
await store.flush()
await store.enqueue_bulk_delete("history", [items[0]._id, items[1]._id])
await store.flush()
remaining = await store.fetch_saved("history")
assert len(remaining) == 1
assert remaining[0][0] == items[2]._id
# direct bulk_delete should report affected rows
count = await store.bulk_delete("history", [items[2]._id])
assert count == 1
await store.close()
@pytest.mark.asyncio
async def test_paginate_out_of_range_returns_last_page():
store = await make_store()
for i in range(7):
await store.enqueue_upsert("history", make_item(i))
await store.flush()
items, total, page, total_pages = await store.paginate("history", page=10, per_page=3, order="DESC")
assert total == 7
assert total_pages == 3
assert page == 3
assert len(items) == 1
await store.close()
@pytest.mark.asyncio
async def test_get_item_returns_none_without_kwargs():
store = await make_store()
await store.enqueue_upsert("queue", make_item(1))
await store.flush()
result = await store.get_item("queue")
assert result is None
await store.close()
@pytest.mark.asyncio
async def test_get_item_matches_conditions():
store = await make_store()
first = make_item(1, status="finished")
second = make_item(2, status="pending")
await store.enqueue_upsert("queue", first)
await store.enqueue_upsert("queue", second)
await store.flush()
# equality match
match_equal = await store.get_item("queue", title=first.title)
assert match_equal is not None
assert match_equal._id == first._id
# NOT_EQUAL should find the second item
match_not_equal = await store.get_item("queue", status=(Operation.NOT_EQUAL, "finished"))
assert match_not_equal is not None
assert match_not_equal._id == second._id
# CONTAIN should find first created (first)
match_contain = await store.get_item("queue", title=(Operation.CONTAIN, "Video"))
assert match_contain is not None
assert match_contain._id == first._id
# No match returns None
no_match = await store.get_item("queue", title=(Operation.CONTAIN, "does-not-exist"))
assert no_match is None
await store.close()
@pytest.mark.asyncio
async def test_on_shutdown_closes_connection():
store = await make_store()
await store.enqueue_upsert("queue", make_item(1))
await store.flush()
await store.on_shutdown(None)
assert store._conn is None
@pytest.mark.asyncio
async def test_exists_and_get_by_key_and_url():
store = await make_store()
item = make_item(42)
await store.enqueue_upsert("queue", item)
await store.flush()
assert await store.exists("queue", key=item._id) is True
assert await store.exists("queue", url=item.url) is True
fetched_by_key = await store.get("queue", key=item._id)
assert fetched_by_key is not None
assert fetched_by_key._id == item._id
fetched_by_url = await store.get("queue", url=item.url)
assert fetched_by_url is not None
assert fetched_by_url._id == item._id
assert await store.exists("queue", key="missing") is False
assert await store.get("queue", key="missing") is None
await store.close()
@pytest.mark.asyncio
async def test_exists_and_get_raise_without_key_or_url():
store = await make_store()
with pytest.raises(KeyError):
await store.exists("queue")
with pytest.raises(KeyError):
await store.get("queue")
await store.close()

4
pyproject.toml
View file

@ -15,7 +15,6 @@ dependencies = [
# Cross‐platform
"python-socketio>=5.11.1",
"aiohttp>=3.9.3",
"caribou>=0.3.0",
"coloredlogs>=15.0.1",
"aiocron>=1.8",
"python-dotenv>=1.0.1",
@ -23,7 +22,7 @@ dependencies = [
"httpx[brotli,http2,socks,zstd]",
"async-timeout",
"pyjson5",
"curl_cffi>=0.13",
"curl_cffi==0.13",
"pysubs2",
"regex",
"mutagen",
@ -44,6 +43,7 @@ dependencies = [
"parsel>=1.10.0",
"jmespath>=1.0.1",
"jsonschema>=4.23.0",
"aiosqlite>=0.22.0",
]
[tool.ruff]

14
ui/app/components/History.vue
View file

@ -271,17 +271,17 @@
<figure :class="['image', thumbnail_ratio]">
<span v-if="'finished' === item.status && item.filename" @click="playVideo(item)" class="play-overlay">
<div class="play-icon"></div>
<img @load="(e: Event) => pImg(e)" :src="getImage(item)" v-if="getImage(item)" />
<img @load="pImg" @error="onImgError" :src="getImage(item)" v-if="getImage(item)" />
<img v-else src="/images/placeholder.png" />
</span>
<span v-else-if="isEmbedable(item.url)" @click="embed_url = getEmbedable(item.url) as string"
class="play-overlay">
<div class="play-icon embed-icon"></div>
<img @load="(e: Event) => pImg(e)" :src="getImage(item)" v-if="getImage(item)" />
<img @load="pImg" @error="onImgError" :src="getImage(item)" v-if="getImage(item)" />
<img v-else src="/images/placeholder.png" />
</span>
<template v-else>
<img @load="(e: Event) => pImg(e)" v-if="getImage(item)" :src="getImage(item)" />
<img @load="pImg" @error="onImgError" v-if="getImage(item)" :src="getImage(item)" />
<img v-else src="/images/placeholder.png" />
</template>
</figure>
@ -847,6 +847,14 @@ const pImg = (e: Event) => {
}
}
const onImgError = (e: Event) => {
const target = e.target as HTMLImageElement
if (target.src.endsWith('/images/placeholder.png')) {
return
}
target.src = '/images/placeholder.png'
}
watch(video_item, (v) => {
if (!bg_enable.value) {
return

12
ui/app/components/Queue.vue
View file

@ -193,13 +193,13 @@
<span v-if="isEmbedable(item.url)" @click="embed_url = getEmbedable(item.url) as string"
class="play-overlay">
<div class="play-icon embed-icon"></div>
<img @load="(e: Event) => pImg(e)"
<img @load="pImg" @error="onImgError"
:src="uri('/api/thumbnail?id=' + item._id + '&url=' + encodePath(item.extras.thumbnail))"
v-if="item.extras?.thumbnail" />
<img v-else src="/images/placeholder.png" />
</span>
<template v-else>
<img @load="(e: Event) => pImg(e)" v-if="item.extras?.thumbnail"
<img @load="pImg" @error="onImgError" v-if="item.extras?.thumbnail"
:src="uri('/api/thumbnail?id=' + item._id + '&url=' + encodePath(item.extras.thumbnail))" />
<img v-else src="/images/placeholder.png" />
</template>
@ -579,6 +579,14 @@ const pImg = (e: Event) => {
}
}
const onImgError = (e: Event) => {
const target = e.target as HTMLImageElement
if (target.src.endsWith('/images/placeholder.png')) {
return
}
target.src = '/images/placeholder.png'
}
watch(embed_url, v => {
if (!bg_enable.value) {
return

26
ui/app/components/Simple.vue
View file

@ -89,7 +89,8 @@
<figure class="media-left">
<figure class="image is-16by9 queue-thumb" :class="{ 'is-clickable': isEmbedable(entry.item.url) }"
role="presentation" @click="openPlayer(entry.item)">
<img :src="resolveThumbnail(entry)" :alt="entry.item.title || 'Video thumbnail'" loading="lazy">
<img :src="resolveThumbnail(entry)" :alt="entry.item.title || 'Video thumbnail'" loading="lazy"
@error="onImgError">
<span v-if="getDurationLabel(entry.item)" class="queue-thumb__badge">
{{ getDurationLabel(entry.item) }}
</span>
@ -688,18 +689,19 @@ const loadMoreHistory = async (): Promise<void> => {
}
}
// Setup intersection observer for infinite scroll
useIntersectionObserver(
loadMoreTrigger,
([entry]) => {
if (entry?.isIntersecting && !paginationInfo.value.isLoading && paginationInfo.value.page < paginationInfo.value.total_pages) {
loadMoreHistory()
}
},
{
threshold: 0.5,
const onImgError = (e: Event) => {
const target = e.target as HTMLImageElement
if (target.src.endsWith('/images/placeholder.png')) {
return
}
)
target.src = '/images/placeholder.png'
}
useIntersectionObserver(loadMoreTrigger, ([entry]) => {
if (entry?.isIntersecting && !paginationInfo.value.isLoading && paginationInfo.value.page < paginationInfo.value.total_pages) {
loadMoreHistory()
}
}, { threshold: 0.5 })
</script>

12
ui/app/stores/SocketStore.ts
View file

@ -118,10 +118,6 @@ export const useSocketStore = defineStore('socket', () => {
config.add('folders', json.data.folders)
}
if (json.data?.queue) {
stateStore.addAll('queue', json.data.queue || {})
}
if (typeof json.data?.history_count === 'number') {
stateStore.setHistoryCount(json.data.history_count)
}
@ -129,6 +125,14 @@ export const useSocketStore = defineStore('socket', () => {
error.value = null;
})
on('active_queue', stream => {
const json = JSON.parse(stream);
if (!json?.data?.queue) {
return;
}
stateStore.addAll('queue', json.data.queue || {})
})
on('item_added', stream => {
const json = JSON.parse(stream);
stateStore.add('queue', json.data._id, json.data);

5
ui/nuxt.config.ts
View file

@ -68,7 +68,7 @@ export default defineNuxtConfig({
'@pinia/nuxt',
'@vueuse/nuxt',
'floating-vue/nuxt',
process.env.NODE_ENV === 'development' ? '@nuxt/eslint' : '',
'development' === process.env.NODE_ENV ? '@nuxt/eslint' : '',
].filter(Boolean),
nitro: {
@ -80,6 +80,9 @@ export default defineNuxtConfig({
vite: {
server: {
allowedHosts: true,
},
build: {
chunkSizeWarningLimit: 2000,
}
},
telemetry: false,

16
ui/package.json
View file

@ -32,11 +32,11 @@
"marked-base-url": "^1.1.8",
"marked-gfm-heading-id": "^4.1.3",
"moment": "^2.30.1",
"nuxt": "^4.2.1",
"nuxt": "^4.2.2",
"pinia": "^3.0.4",
"socket.io-client": "^4.8.1",
"vue": "^3.5.25",
"vue-router": "^4.6.3",
"vue-router": "^4.6.4",
"vue-toastification": "2.0.0-rc.5"
},
"pnpm": {
@ -52,13 +52,13 @@
"ansi-regex": "6.2.0"
},
"devDependencies": {
"@nuxt/eslint": "^1.11.0",
"@nuxt/eslint-config": "^1.11.0",
"@typescript-eslint/parser": "^8.48.1",
"eslint": "^9.39.1",
"@nuxt/eslint": "^1.12.1",
"@nuxt/eslint-config": "^1.12.1",
"@typescript-eslint/parser": "^8.50.0",
"eslint": "^9.39.2",
"typescript": "^5.9.3",
"vitest": "^4.0.15",
"vitest": "^4.0.16",
"vue-eslint-parser": "^10.2.0",
"vue-tsc": "^3.1.5"
"vue-tsc": "^3.1.8"
}
}

1558
ui/pnpm-lock.yaml
View file

File diff suppressed because it is too large Load diff

104
uv.lock
View file

@ -105,6 +105,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/fb/76/641ae371508676492379f16e2fa48f4e2c11741bd63c48be4b12a6b09cba/aiosignal-1.4.0-py3-none-any.whl", hash = "sha256:053243f8b92b990551949e63930a839ff0cf0b0ebbe0597b0f3fb19e1a0fe82e", size = 7490 },
]
[[package]]
name = "aiosqlite"
version = "0.22.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/3a/0d/449c024bdabd0678ae07d804e60ed3b9786facd3add66f51eee67a0fccea/aiosqlite-0.22.0.tar.gz", hash = "sha256:7e9e52d72b319fcdeac727668975056c49720c995176dc57370935e5ba162bb9", size = 14707 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/ef/39/b2181148075272edfbbd6d87e6cd78cc71dca243446fa3b381fd4116950b/aiosqlite-0.22.0-py3-none-any.whl", hash = "sha256:96007fac2ce70eda3ca1bba7a3008c435258a592b8fbf2ee3eeaa36d33971a09", size = 17263 },
]
[[package]]
name = "altgraph"
version = "0.17.5"
@ -224,15 +233,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/07/6b/6e92009df3b8b7272f85a0992b306b61c34b7ea1c4776643746e61c380ac/brotlicffi-1.2.0.0-cp38-abi3-win_amd64.whl", hash = "sha256:f139a7cdfe4ae7859513067b736eb44d19fae1186f9e99370092f6915216451b", size = 378586 },
]
[[package]]
name = "caribou"
version = "0.4.1"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/6c/74/de2c20c5a24811d69a2a6150d5cd841f2af21ffc65ff964a73f46738bb68/caribou-0.4.1.tar.gz", hash = "sha256:13a66fc9ef9b1c9e9ef220876d59a44ee5eff9e579655252271b7514fc0ad787", size = 14019 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/b5/91/795eccdad6abd41b3634502d0971cb696d2e5c9cede67f8b38ebe30d2b1e/caribou-0.4.1-py2.py3-none-any.whl", hash = "sha256:5c7a036584b34021011f1512620152272924e55ca87c7c805073aeef31c5baed", size = 7203 },
]
[[package]]
name = "certifi"
version = "2025.11.12"
@ -416,19 +416,19 @@ wheels = [
[[package]]
name = "debugpy"
version = "1.8.17"
version = "1.8.19"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/15/ad/71e708ff4ca377c4230530d6a7aa7992592648c122a2cd2b321cf8b35a76/debugpy-1.8.17.tar.gz", hash = "sha256:fd723b47a8c08892b1a16b2c6239a8b96637c62a59b94bb5dab4bac592a58a8e", size = 1644129 }
sdist = { url = "https://files.pythonhosted.org/packages/73/75/9e12d4d42349b817cd545b89247696c67917aab907012ae5b64bbfea3199/debugpy-1.8.19.tar.gz", hash = "sha256:eea7e5987445ab0b5ed258093722d5ecb8bb72217c5c9b1e21f64efe23ddebdb", size = 1644590 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/50/76/597e5cb97d026274ba297af8d89138dfd9e695767ba0e0895edb20963f40/debugpy-1.8.17-cp313-cp313-macosx_15_0_universal2.whl", hash = "sha256:857c1dd5d70042502aef1c6d1c2801211f3ea7e56f75e9c335f434afb403e464", size = 2538386 },
{ url = "https://files.pythonhosted.org/packages/5f/60/ce5c34fcdfec493701f9d1532dba95b21b2f6394147234dce21160bd923f/debugpy-1.8.17-cp313-cp313-manylinux_2_34_x86_64.whl", hash = "sha256:3bea3b0b12f3946e098cce9b43c3c46e317b567f79570c3f43f0b96d00788088", size = 4292100 },
{ url = "https://files.pythonhosted.org/packages/e8/95/7873cf2146577ef71d2a20bf553f12df865922a6f87b9e8ee1df04f01785/debugpy-1.8.17-cp313-cp313-win32.whl", hash = "sha256:e34ee844c2f17b18556b5bbe59e1e2ff4e86a00282d2a46edab73fd7f18f4a83", size = 5277002 },
{ url = "https://files.pythonhosted.org/packages/46/11/18c79a1cee5ff539a94ec4aa290c1c069a5580fd5cfd2fb2e282f8e905da/debugpy-1.8.17-cp313-cp313-win_amd64.whl", hash = "sha256:6c5cd6f009ad4fca8e33e5238210dc1e5f42db07d4b6ab21ac7ffa904a196420", size = 5319047 },
{ url = "https://files.pythonhosted.org/packages/de/45/115d55b2a9da6de812696064ceb505c31e952c5d89c4ed1d9bb983deec34/debugpy-1.8.17-cp314-cp314-macosx_15_0_universal2.whl", hash = "sha256:045290c010bcd2d82bc97aa2daf6837443cd52f6328592698809b4549babcee1", size = 2536899 },
{ url = "https://files.pythonhosted.org/packages/5a/73/2aa00c7f1f06e997ef57dc9b23d61a92120bec1437a012afb6d176585197/debugpy-1.8.17-cp314-cp314-manylinux_2_34_x86_64.whl", hash = "sha256:b69b6bd9dba6a03632534cdf67c760625760a215ae289f7489a452af1031fe1f", size = 4268254 },
{ url = "https://files.pythonhosted.org/packages/86/b5/ed3e65c63c68a6634e3ba04bd10255c8e46ec16ebed7d1c79e4816d8a760/debugpy-1.8.17-cp314-cp314-win32.whl", hash = "sha256:5c59b74aa5630f3a5194467100c3b3d1c77898f9ab27e3f7dc5d40fc2f122670", size = 5277203 },
{ url = "https://files.pythonhosted.org/packages/b0/26/394276b71c7538445f29e792f589ab7379ae70fd26ff5577dfde71158e96/debugpy-1.8.17-cp314-cp314-win_amd64.whl", hash = "sha256:893cba7bb0f55161de4365584b025f7064e1f88913551bcd23be3260b231429c", size = 5318493 },
{ url = "https://files.pythonhosted.org/packages/b0/d0/89247ec250369fc76db477720a26b2fce7ba079ff1380e4ab4529d2fe233/debugpy-1.8.17-py2.py3-none-any.whl", hash = "sha256:60c7dca6571efe660ccb7a9508d73ca14b8796c4ed484c2002abba714226cfef", size = 5283210 },
{ url = "https://files.pythonhosted.org/packages/71/3d/388035a31a59c26f1ecc8d86af607d0c42e20ef80074147cd07b180c4349/debugpy-1.8.19-cp313-cp313-macosx_15_0_universal2.whl", hash = "sha256:91e35db2672a0abaf325f4868fcac9c1674a0d9ad9bb8a8c849c03a5ebba3e6d", size = 2538859 },
{ url = "https://files.pythonhosted.org/packages/4a/19/c93a0772d0962294f083dbdb113af1a7427bb632d36e5314297068f55db7/debugpy-1.8.19-cp313-cp313-manylinux_2_34_x86_64.whl", hash = "sha256:85016a73ab84dea1c1f1dcd88ec692993bcbe4532d1b49ecb5f3c688ae50c606", size = 4292575 },
{ url = "https://files.pythonhosted.org/packages/5c/56/09e48ab796b0a77e3d7dc250f95251832b8bf6838c9632f6100c98bdf426/debugpy-1.8.19-cp313-cp313-win32.whl", hash = "sha256:b605f17e89ba0ecee994391194285fada89cee111cfcd29d6f2ee11cbdc40976", size = 5286209 },
{ url = "https://files.pythonhosted.org/packages/fb/4e/931480b9552c7d0feebe40c73725dd7703dcc578ba9efc14fe0e6d31cfd1/debugpy-1.8.19-cp313-cp313-win_amd64.whl", hash = "sha256:c30639998a9f9cd9699b4b621942c0179a6527f083c72351f95c6ab1728d5b73", size = 5328206 },
{ url = "https://files.pythonhosted.org/packages/f6/b9/cbec520c3a00508327476c7fce26fbafef98f412707e511eb9d19a2ef467/debugpy-1.8.19-cp314-cp314-macosx_15_0_universal2.whl", hash = "sha256:1e8c4d1bd230067bf1bbcdbd6032e5a57068638eb28b9153d008ecde288152af", size = 2537372 },
{ url = "https://files.pythonhosted.org/packages/88/5e/cf4e4dc712a141e10d58405c58c8268554aec3c35c09cdcda7535ff13f76/debugpy-1.8.19-cp314-cp314-manylinux_2_34_x86_64.whl", hash = "sha256:d40c016c1f538dbf1762936e3aeb43a89b965069d9f60f9e39d35d9d25e6b809", size = 4268729 },
{ url = "https://files.pythonhosted.org/packages/82/a3/c91a087ab21f1047db328c1d3eb5d1ff0e52de9e74f9f6f6fa14cdd93d58/debugpy-1.8.19-cp314-cp314-win32.whl", hash = "sha256:0601708223fe1cd0e27c6cce67a899d92c7d68e73690211e6788a4b0e1903f5b", size = 5286388 },
{ url = "https://files.pythonhosted.org/packages/17/b8/bfdc30b6e94f1eff09f2dc9cc1f9cd1c6cde3d996bcbd36ce2d9a4956e99/debugpy-1.8.19-cp314-cp314-win_amd64.whl", hash = "sha256:8e19a725f5d486f20e53a1dde2ab8bb2c9607c40c00a42ab646def962b41125f", size = 5327741 },
{ url = "https://files.pythonhosted.org/packages/25/3e/e27078370414ef35fafad2c06d182110073daaeb5d3bf734b0b1eeefe452/debugpy-1.8.19-py2.py3-none-any.whl", hash = "sha256:360ffd231a780abbc414ba0f005dad409e71c78637efe8f2bd75837132a41d38", size = 5292321 },
]
[[package]]
@ -1246,15 +1246,15 @@ wheels = [
[[package]]
name = "python-socketio"
version = "5.15.0"
version = "5.15.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "bidict" },
{ name = "python-engineio" },
]
sdist = { url = "https://files.pythonhosted.org/packages/72/a8/5f7c805dd6d0d6cba91d3ea215b4b88889d1b99b71a53c932629daba53f1/python_socketio-5.15.0.tar.gz", hash = "sha256:d0403ababb59aa12fd5adcfc933a821113f27bd77761bc1c54aad2e3191a9b69", size = 126439 }
sdist = { url = "https://files.pythonhosted.org/packages/5c/b5/56d070ade9ae60ed90ce2cdb41da927791cdae31f1059aab4b6b60d223b3/python_socketio-5.15.1.tar.gz", hash = "sha256:54fe3e5580ea06a1b29b541e8ef32fe956846c99a76059e343e43aada754efdd", size = 127172 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/cd/fa/1ef2f8537272a2f383d72b9301c3ef66a49710b3bb7dcb2bd138cf2920d1/python_socketio-5.15.0-py3-none-any.whl", hash = "sha256:e93363102f4da6d8e7a8872bf4908b866c40f070e716aa27132891e643e2687c", size = 79451 },
{ url = "https://files.pythonhosted.org/packages/5c/47/45a805fc1e4c3104df1193a78aeb98734497e32931efd1dfe9897c19188b/python_socketio-5.15.1-py3-none-any.whl", hash = "sha256:abc3528803563ed9a2010bc76829afe21d7a308a1e5651171fdb582d12e2ace0", size = 79561 },
]
[[package]]
@ -1484,28 +1484,28 @@ wheels = [
[[package]]
name = "ruff"
version = "0.14.8"
version = "0.14.10"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/ed/d9/f7a0c4b3a2bf2556cd5d99b05372c29980249ef71e8e32669ba77428c82c/ruff-0.14.8.tar.gz", hash = "sha256:774ed0dd87d6ce925e3b8496feb3a00ac564bea52b9feb551ecd17e0a23d1eed", size = 5765385 }
sdist = { url = "https://files.pythonhosted.org/packages/57/08/52232a877978dd8f9cf2aeddce3e611b40a63287dfca29b6b8da791f5e8d/ruff-0.14.10.tar.gz", hash = "sha256:9a2e830f075d1a42cd28420d7809ace390832a490ed0966fe373ba288e77aaf4", size = 5859763 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/48/b8/9537b52010134b1d2b72870cc3f92d5fb759394094741b09ceccae183fbe/ruff-0.14.8-py3-none-linux_armv6l.whl", hash = "sha256:ec071e9c82eca417f6111fd39f7043acb53cd3fde9b1f95bbed745962e345afb", size = 13441540 },
{ url = "https://files.pythonhosted.org/packages/24/00/99031684efb025829713682012b6dd37279b1f695ed1b01725f85fd94b38/ruff-0.14.8-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:8cdb162a7159f4ca36ce980a18c43d8f036966e7f73f866ac8f493b75e0c27e9", size = 13669384 },
{ url = "https://files.pythonhosted.org/packages/72/64/3eb5949169fc19c50c04f28ece2c189d3b6edd57e5b533649dae6ca484fe/ruff-0.14.8-py3-none-macosx_11_0_arm64.whl", hash = "sha256:2e2fcbefe91f9fad0916850edf0854530c15bd1926b6b779de47e9ab619ea38f", size = 12806917 },
{ url = "https://files.pythonhosted.org/packages/c4/08/5250babb0b1b11910f470370ec0cbc67470231f7cdc033cee57d4976f941/ruff-0.14.8-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a9d70721066a296f45786ec31916dc287b44040f553da21564de0ab4d45a869b", size = 13256112 },
{ url = "https://files.pythonhosted.org/packages/78/4c/6c588e97a8e8c2d4b522c31a579e1df2b4d003eddfbe23d1f262b1a431ff/ruff-0.14.8-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:2c87e09b3cd9d126fc67a9ecd3b5b1d3ded2b9c7fce3f16e315346b9d05cfb52", size = 13227559 },
{ url = "https://files.pythonhosted.org/packages/23/ce/5f78cea13eda8eceac71b5f6fa6e9223df9b87bb2c1891c166d1f0dce9f1/ruff-0.14.8-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:1d62cb310c4fbcb9ee4ac023fe17f984ae1e12b8a4a02e3d21489f9a2a5f730c", size = 13896379 },
{ url = "https://files.pythonhosted.org/packages/cf/79/13de4517c4dadce9218a20035b21212a4c180e009507731f0d3b3f5df85a/ruff-0.14.8-py3-none-manylinux_2_17_ppc64.manylinux2014_ppc64.whl", hash = "sha256:1af35c2d62633d4da0521178e8a2641c636d2a7153da0bac1b30cfd4ccd91344", size = 15372786 },
{ url = "https://files.pythonhosted.org/packages/00/06/33df72b3bb42be8a1c3815fd4fae83fa2945fc725a25d87ba3e42d1cc108/ruff-0.14.8-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:25add4575ffecc53d60eed3f24b1e934493631b48ebbc6ebaf9d8517924aca4b", size = 14990029 },
{ url = "https://files.pythonhosted.org/packages/64/61/0f34927bd90925880394de0e081ce1afab66d7b3525336f5771dcf0cb46c/ruff-0.14.8-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:4c943d847b7f02f7db4201a0600ea7d244d8a404fbb639b439e987edcf2baf9a", size = 14407037 },
{ url = "https://files.pythonhosted.org/packages/96/bc/058fe0aefc0fbf0d19614cb6d1a3e2c048f7dc77ca64957f33b12cfdc5ef/ruff-0.14.8-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:cb6e8bf7b4f627548daa1b69283dac5a296bfe9ce856703b03130732e20ddfe2", size = 14102390 },
{ url = "https://files.pythonhosted.org/packages/af/a4/e4f77b02b804546f4c17e8b37a524c27012dd6ff05855d2243b49a7d3cb9/ruff-0.14.8-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:7aaf2974f378e6b01d1e257c6948207aec6a9b5ba53fab23d0182efb887a0e4a", size = 14230793 },
{ url = "https://files.pythonhosted.org/packages/3f/52/bb8c02373f79552e8d087cedaffad76b8892033d2876c2498a2582f09dcf/ruff-0.14.8-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:e5758ca513c43ad8a4ef13f0f081f80f08008f410790f3611a21a92421ab045b", size = 13160039 },
{ url = "https://files.pythonhosted.org/packages/1f/ad/b69d6962e477842e25c0b11622548df746290cc6d76f9e0f4ed7456c2c31/ruff-0.14.8-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:f74f7ba163b6e85a8d81a590363bf71618847e5078d90827749bfda1d88c9cdf", size = 13205158 },
{ url = "https://files.pythonhosted.org/packages/06/63/54f23da1315c0b3dfc1bc03fbc34e10378918a20c0b0f086418734e57e74/ruff-0.14.8-py3-none-musllinux_1_2_i686.whl", hash = "sha256:eed28f6fafcc9591994c42254f5a5c5ca40e69a30721d2ab18bb0bb3baac3ab6", size = 13469550 },
{ url = "https://files.pythonhosted.org/packages/70/7d/a4d7b1961e4903bc37fffb7ddcfaa7beb250f67d97cfd1ee1d5cddb1ec90/ruff-0.14.8-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:21d48fa744c9d1cb8d71eb0a740c4dd02751a5de9db9a730a8ef75ca34cf138e", size = 14211332 },
{ url = "https://files.pythonhosted.org/packages/5d/93/2a5063341fa17054e5c86582136e9895db773e3c2ffb770dde50a09f35f0/ruff-0.14.8-py3-none-win32.whl", hash = "sha256:15f04cb45c051159baebb0f0037f404f1dc2f15a927418f29730f411a79bc4e7", size = 13151890 },
{ url = "https://files.pythonhosted.org/packages/02/1c/65c61a0859c0add13a3e1cbb6024b42de587456a43006ca2d4fd3d1618fe/ruff-0.14.8-py3-none-win_amd64.whl", hash = "sha256:9eeb0b24242b5bbff3011409a739929f497f3fb5fe3b5698aba5e77e8c833097", size = 14537826 },
{ url = "https://files.pythonhosted.org/packages/6d/63/8b41cea3afd7f58eb64ac9251668ee0073789a3bc9ac6f816c8c6fef986d/ruff-0.14.8-py3-none-win_arm64.whl", hash = "sha256:965a582c93c63fe715fd3e3f8aa37c4b776777203d8e1d8aa3cc0c14424a4b99", size = 13634522 },
{ url = "https://files.pythonhosted.org/packages/60/01/933704d69f3f05ee16ef11406b78881733c186fe14b6a46b05cfcaf6d3b2/ruff-0.14.10-py3-none-linux_armv6l.whl", hash = "sha256:7a3ce585f2ade3e1f29ec1b92df13e3da262178df8c8bdf876f48fa0e8316c49", size = 13527080 },
{ url = "https://files.pythonhosted.org/packages/df/58/a0349197a7dfa603ffb7f5b0470391efa79ddc327c1e29c4851e85b09cc5/ruff-0.14.10-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:674f9be9372907f7257c51f1d4fc902cb7cf014b9980152b802794317941f08f", size = 13797320 },
{ url = "https://files.pythonhosted.org/packages/7b/82/36be59f00a6082e38c23536df4e71cdbc6af8d7c707eade97fcad5c98235/ruff-0.14.10-py3-none-macosx_11_0_arm64.whl", hash = "sha256:d85713d522348837ef9df8efca33ccb8bd6fcfc86a2cde3ccb4bc9d28a18003d", size = 12918434 },
{ url = "https://files.pythonhosted.org/packages/a6/00/45c62a7f7e34da92a25804f813ebe05c88aa9e0c25e5cb5a7d23dd7450e3/ruff-0.14.10-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:6987ebe0501ae4f4308d7d24e2d0fe3d7a98430f5adfd0f1fead050a740a3a77", size = 13371961 },
{ url = "https://files.pythonhosted.org/packages/40/31/a5906d60f0405f7e57045a70f2d57084a93ca7425f22e1d66904769d1628/ruff-0.14.10-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:16a01dfb7b9e4eee556fbfd5392806b1b8550c9b4a9f6acd3dbe6812b193c70a", size = 13275629 },
{ url = "https://files.pythonhosted.org/packages/3e/60/61c0087df21894cf9d928dc04bcd4fb10e8b2e8dca7b1a276ba2155b2002/ruff-0.14.10-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:7165d31a925b7a294465fa81be8c12a0e9b60fb02bf177e79067c867e71f8b1f", size = 14029234 },
{ url = "https://files.pythonhosted.org/packages/44/84/77d911bee3b92348b6e5dab5a0c898d87084ea03ac5dc708f46d88407def/ruff-0.14.10-py3-none-manylinux_2_17_ppc64.manylinux2014_ppc64.whl", hash = "sha256:c561695675b972effb0c0a45db233f2c816ff3da8dcfbe7dfc7eed625f218935", size = 15449890 },
{ url = "https://files.pythonhosted.org/packages/e9/36/480206eaefa24a7ec321582dda580443a8f0671fdbf6b1c80e9c3e93a16a/ruff-0.14.10-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:4bb98fcbbc61725968893682fd4df8966a34611239c9fd07a1f6a07e7103d08e", size = 15123172 },
{ url = "https://files.pythonhosted.org/packages/5c/38/68e414156015ba80cef5473d57919d27dfb62ec804b96180bafdeaf0e090/ruff-0.14.10-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:f24b47993a9d8cb858429e97bdf8544c78029f09b520af615c1d261bf827001d", size = 14460260 },
{ url = "https://files.pythonhosted.org/packages/b3/19/9e050c0dca8aba824d67cc0db69fb459c28d8cd3f6855b1405b3f29cc91d/ruff-0.14.10-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:59aabd2e2c4fd614d2862e7939c34a532c04f1084476d6833dddef4afab87e9f", size = 14229978 },
{ url = "https://files.pythonhosted.org/packages/51/eb/e8dd1dd6e05b9e695aa9dd420f4577debdd0f87a5ff2fedda33c09e9be8c/ruff-0.14.10-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:213db2b2e44be8625002dbea33bb9c60c66ea2c07c084a00d55732689d697a7f", size = 14338036 },
{ url = "https://files.pythonhosted.org/packages/6a/12/f3e3a505db7c19303b70af370d137795fcfec136d670d5de5391e295c134/ruff-0.14.10-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:b914c40ab64865a17a9a5b67911d14df72346a634527240039eb3bd650e5979d", size = 13264051 },
{ url = "https://files.pythonhosted.org/packages/08/64/8c3a47eaccfef8ac20e0484e68e0772013eb85802f8a9f7603ca751eb166/ruff-0.14.10-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:1484983559f026788e3a5c07c81ef7d1e97c1c78ed03041a18f75df104c45405", size = 13283998 },
{ url = "https://files.pythonhosted.org/packages/12/84/534a5506f4074e5cc0529e5cd96cfc01bb480e460c7edf5af70d2bcae55e/ruff-0.14.10-py3-none-musllinux_1_2_i686.whl", hash = "sha256:c70427132db492d25f982fffc8d6c7535cc2fd2c83fc8888f05caaa248521e60", size = 13601891 },
{ url = "https://files.pythonhosted.org/packages/0d/1e/14c916087d8598917dbad9b2921d340f7884824ad6e9c55de948a93b106d/ruff-0.14.10-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:5bcf45b681e9f1ee6445d317ce1fa9d6cba9a6049542d1c3d5b5958986be8830", size = 14336660 },
{ url = "https://files.pythonhosted.org/packages/f2/1c/d7b67ab43f30013b47c12b42d1acd354c195351a3f7a1d67f59e54227ede/ruff-0.14.10-py3-none-win32.whl", hash = "sha256:104c49fc7ab73f3f3a758039adea978869a918f31b73280db175b43a2d9b51d6", size = 13196187 },
{ url = "https://files.pythonhosted.org/packages/fb/9c/896c862e13886fae2af961bef3e6312db9ebc6adc2b156fe95e615dee8c1/ruff-0.14.10-py3-none-win_amd64.whl", hash = "sha256:466297bd73638c6bdf06485683e812db1c00c7ac96d4ddd0294a338c62fdc154", size = 14661283 },
{ url = "https://files.pythonhosted.org/packages/74/31/b0e29d572670dca3674eeee78e418f20bdf97fa8aa9ea71380885e175ca0/ruff-0.14.10-py3-none-win_arm64.whl", hash = "sha256:e51d046cf6dda98a4633b8a8a771451107413b0f07183b2bef03f075599e44e6", size = 13729839 },
]
[[package]]
@ -1624,11 +1624,11 @@ wheels = [
[[package]]
name = "tzdata"
version = "2025.2"
version = "2025.3"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/95/32/1a225d6164441be760d75c2c42e2780dc0873fe382da3e98a2e1e48361e5/tzdata-2025.2.tar.gz", hash = "sha256:b60a638fcc0daffadf82fe0f57e53d06bdec2f36c4df66280ae79bce6bd6f2b9", size = 196380 }
sdist = { url = "https://files.pythonhosted.org/packages/5e/a7/c202b344c5ca7daf398f3b8a477eeb205cf3b6f32e7ec3a6bac0629ca975/tzdata-2025.3.tar.gz", hash = "sha256:de39c2ca5dc7b0344f2eba86f49d614019d29f060fc4ebc8a417896a620b56a7", size = 196772 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/5c/23/c7abc0ca0a1526a0774eca151daeb8de62ec457e77262b66b359c3c7679e/tzdata-2025.2-py2.py3-none-any.whl", hash = "sha256:1a403fada01ff9221ca8044d701868fa132215d84beb92242d9acd2147f667a8", size = 347839 },
{ url = "https://files.pythonhosted.org/packages/c7/b0/003792df09decd6849a5e39c28b513c06e84436a54440380862b5aeff25d/tzdata-2025.3-py2.py3-none-any.whl", hash = "sha256:06a47e5700f3081aab02b2e513160914ff0694bce9947d6b76ebd6bf57cfc5d1", size = 348521 },
]
[[package]]
@ -1645,11 +1645,11 @@ wheels = [
[[package]]
name = "urllib3"
version = "2.6.0"
version = "2.6.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/1c/43/554c2569b62f49350597348fc3ac70f786e3c32e7f19d266e19817812dd3/urllib3-2.6.0.tar.gz", hash = "sha256:cb9bcef5a4b345d5da5d145dc3e30834f58e8018828cbc724d30b4cb7d4d49f1", size = 432585 }
sdist = { url = "https://files.pythonhosted.org/packages/1e/24/a2a2ed9addd907787d7aa0355ba36a6cadf1768b934c652ea78acbd59dcd/urllib3-2.6.2.tar.gz", hash = "sha256:016f9c98bb7e98085cb2b4b17b87d2c702975664e4f060c6532e64d1c1a5e797", size = 432930 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/56/1a/9ffe814d317c5224166b23e7c47f606d6e473712a2fad0f704ea9b99f246/urllib3-2.6.0-py3-none-any.whl", hash = "sha256:c90f7a39f716c572c4e3e58509581ebd83f9b59cced005b7db7ad2d22b0db99f", size = 131083 },
{ url = "https://files.pythonhosted.org/packages/6d/b9/4095b668ea3678bf6a0af005527f39de12fb026516fb3df17495a733b7f8/urllib3-2.6.2-py3-none-any.whl", hash = "sha256:ec21cddfe7724fc7cb4ba4bea7aa8e2ef36f607a4bab81aa6ce42a13dc3f03dd", size = 131182 },
]
[package.optional-dependencies]
@ -1823,13 +1823,13 @@ source = { virtual = "." }
dependencies = [
{ name = "aiocron" },
{ name = "aiohttp" },
{ name = "aiosqlite" },
{ name = "anyio" },
{ name = "apprise" },
{ name = "async-timeout" },
{ name = "bgutil-ytdlp-pot-provider" },
{ name = "brotli" },
{ name = "brotlicffi" },
{ name = "caribou" },
{ name = "coloredlogs" },
{ name = "curl-cffi" },
{ name = "dateparser" },
@ -1874,15 +1874,15 @@ dev = [
requires-dist = [
{ name = "aiocron", specifier = ">=1.8" },
{ name = "aiohttp", specifier = ">=3.9.3" },
{ name = "aiosqlite", specifier = ">=0.22.0" },
{ name = "anyio" },
{ name = "apprise", specifier = ">=1.9.3" },
{ name = "async-timeout" },
{ name = "bgutil-ytdlp-pot-provider", specifier = ">=1.2.1" },
{ name = "brotli" },
{ name = "brotlicffi" },
{ name = "caribou", specifier = ">=0.3.0" },
{ name = "coloredlogs", specifier = ">=15.0.1" },
{ name = "curl-cffi", specifier = ">=0.13" },
{ name = "curl-cffi", specifier = "==0.13" },
{ name = "dateparser", specifier = ">=1.2.1" },
{ name = "debugpy", specifier = ">=1.8.1" },
{ name = "defusedxml", specifier = ">=0.7.1" },

0
var/downloads/.gitkeep
View file