diff --git a/.vscode/settings.json b/.vscode/settings.json index 5c75efb6..2e166d45 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -34,12 +34,14 @@ "Cfmrc", "choco", "cifs", + "connectionpool", "consoletitle", "continuedl", "cookiesfrombrowser", "copyts", "creationflags", "cronsim", + "currsize", "dailymotion", "datas", "dateparser", @@ -189,5 +191,13 @@ "app/tests" ], "python.testing.unittestEnabled": true, - "python.testing.pytestEnabled": true + "python.testing.pytestEnabled": true, + "json.schemas": [ + { + "fileMatch": [ + "**/var/config/tasks/*.json" + ], + "url": "./app/library/task_handlers/task_definition.schema.json" + } + ] } diff --git a/API.md b/API.md index 963c251a..454218c6 100644 --- a/API.md +++ b/API.md @@ -32,6 +32,7 @@ This document describes the available endpoints and their usage. All endpoints r - [DELETE /api/archiver](#delete-apiarchiver) - [GET /api/tasks](#get-apitasks) - [PUT /api/tasks](#put-apitasks) + - [POST /api/tasks/inspect](#post-apitasksinspect) - [POST /api/tasks/{id}/mark](#post-apitasksidmark) - [GET /api/player/playlist/{file:.\*}.m3u8](#get-apiplayerplaylistfilem3u8) - [GET /api/player/m3u8/{mode}/{file:.\*}.m3u8](#get-apiplayerm3u8modefilem3u8) @@ -551,6 +552,62 @@ or on error --- +### POST /api/tasks/inspect +**Purpose**: Preview which handler matches a URL and review the items it would queue without dispatching downloads. + +**Body**: +```json +{ + "url": "https://example.com/feed", // required, validated URL + "preset": "news-preset", // optional preset override - In real scenarios, the preset come from the task. + "handler": "GenericTaskHandler" // optional explicit handler class name, in real scenarios, the handler is resolved automatically. +} +``` + +**Response (success)**: +```json +{ + "matched": true, + "handler": "GenericTaskHandler", + "supported": true, + "items": [ + { + "url": "https://example.com/video/1", + "title": "Example title", + "archive_id": "generic 1", + "metadata": { + "source_task": "...", + "source_handler": "GenericTaskHandler" + } + } + ], + "metadata": { + "raw_count": 3 + } +} +``` + +- Returns `200 OK` when inspection succeeds. +- The `metadata` object is optional and may contain handler-specific keys. + +**Response (failure)**: +```json +{ + "matched": false, + "handler": null, + "error": "No handler matched the supplied URL.", + "message": "No handler matched the supplied URL.", + "metadata": { + "supported": true + } +} +``` + +- Returns `400 Bad Request` when the handler cannot process the URL or inspection fails. +- When a handler is specified but does not support manual inspection, `supported` is `false` and the `message` explains the limitation. + +--- + ### POST /api/tasks/{id}/mark **Purpose**: Mark all entries associated with a scheduled task as downloaded. diff --git a/FAQ.md b/FAQ.md index e3dbcef2..e8503381 100644 --- a/FAQ.md +++ b/FAQ.md @@ -85,10 +85,6 @@ the `yt-dlp` output and attempt to download the media directly to your iOS devic starting point for those who want to download media directly to their iOS device. We provide no support for this use case other than the shortcut itself. this shortcut missing support for parsing the http_headers, it's only parse the cookies. - - - - # Authentication To enable basic authentication, set the `YTP_AUTH_USERNAME` and `YTP_AUTH_PASSWORD` environment variables. And restart @@ -97,11 +93,6 @@ As this is a simple basic authentication, if your browser doesn't show the promp `http://username:password@your_ytptube_url:port` -# Basic mode - -What does the basic mode do? it basically strips down the interface to the bare minimum, -to make it easier for non-technical users to use it. - # I cant download anything If you are receiving errors like: @@ -154,6 +145,125 @@ YTP_YTDLP_VERSION=2025.07.21 or master or nightly Then restart the container to apply the changes. +# How can I monitor sites without RSS feeds? + +YTPTube includes a **generic task handler** that turns JSON definition files into site-specific scrapers. You can use it +to watch pages that do not expose RSS or public APIs and automatically enqueue new links into the download queue. + +1. Create definition files under `/config/tasks/*.json` (for Docker this is the mounted `config/tasks/` folder). +2. Keep your scheduled task in `tasks.json` pointing at the page you want to monitor and make sure it uses a preset that + enables a download archive (`--download-archive`). +3. When the task runs, the handler scans the JSON files, picks the first definition whose `match` rule covers the task + URL, fetches the page, extracts items, and queues the unseen ones. + +### Definition schema + +Each file must contain a single JSON object with the following keys: + +```json5 +{ + "name": "example", // Friendly identifier shown in logs + "match": [ + "https://example.com/articles/*", // Glob strings, or objects with {"regex": "..."} or {"glob": "..."} + { "regex": "https://example.com/post/[0-9]+" } + ], + "engine": { // Optional, defaults to HTTPX + "type": "httpx", // "httpx" (default) or "selenium" + "options": { + "url": "http://selenium:4444/wd/hub", // Selenium-only: remote hub URL + "arguments": ["--headless", "--disable-gpu"], + "wait_for": { "type": "css", "expression": ".article" }, + "wait_timeout": 15, + "page_load_timeout": 60 + } + }, + "request": { // Optional HTTP settings + "method": "GET", // GET or POST + "url": "https://example.com/articles/latest", // Override the task URL if needed + "headers": { "User-Agent": "MyAgent/1.0" }, + "params": { "page": 1 }, + "data": null, + "json": null, + "timeout": 30 + }, + "response": { // Optional: how to interpret the body + "type": "html" // "html" (default) or "json" + }, + "parse": { + "items": { // Optional container for per-item extraction + "selector": ".columns .card", // Defaults to CSS; set "type": "xpath" to use XPath + "fields": { + "link": { // Required inside fields: the per-item URL + "type": "css", + "expression": ".card-header a", + "attribute": "href" + }, + "title": { + "type": "css", + "expression": ".card-header a", + "attribute": "text" + }, + "poet": { + "type": "css", + "expression": "footer .card-footer-item:first-child a", + "attribute": "text" + } + } + }, + "page_title": { // Optional global field outside the container + "type": "css", + "expression": "title", + "attribute": "text" + } + } +} +``` + +For JSON endpoints, switch the response format and use `jsonpath` selectors: + +```json5 +{ + "response": { "type": "json" }, + "parse": { + "items": { + "type": "jsonpath", + "selector": "items", + "fields": { + "link": { "type": "jsonpath", "expression": "url" }, + "title": { "type": "jsonpath", "expression": "title" } + } + } + } +} +``` + +### Parsing rules + +- Every definition must provide a `link` field either at the top level or inside `parse.items.fields`. Other fields are optional metadata attached to the queued item. +- CSS and XPath rules may specify `attribute`: + - `text` / `inner_text` applies `normalize-space()`. + - `html` / `outer_html` returns the raw HTML fragment. + - Any other value reads that attribute from the element. When omitted, the handler uses text and, for `link`, falls + back to `href` automatically. +- Regex rules scan the HTML fragment associated with the current scope (page-level or container). Set `attribute` to a named/numbered capture group or rely on the first group. +- `post_filter` lets you run a final regex on the extracted value and pick a named (`value`) group. +- When you declare `parse.items`, each matching container is processed independently so missing fields in one card do not shift values for the rest. +- For JSON responses (`response.type = "json"`), set the container `type` and field `type` to `jsonpath` and supply [JMESPath](https://jmespath.org/) expressions. Relative values are resolved against each container object and converted to strings automatically. + +### Fetch engines + +- **httpx (default)**: supports custom headers, query params, JSON/body payloads, proxy inherited from the task preset, + and optional timeout. +- **selenium**: uses a remote Chrome session. Provide the hub URL under `engine.options.url`; only Chrome is supported at + the moment. Optional keys: `arguments` (list or string), `wait_for` (type `css`/`xpath` + `expression`), `wait_timeout`, + and `page_load_timeout`. + +Definitions are reloaded automatically when files change, so you can tweak them without restarting YTPTube. Check +`var/config/tasks/01-*.json` for sample files. + +> [!NOTE] +> A machine-readable schema is available at `app/library/task_handlers/task_definition.schema.json` if you want to validate your JSON with editors or CI tools. + # How to generate POT tokens? You need a pot provider server we already have the extractor `bgutil-ytdlp-pot-provider` pre-installed in the container. diff --git a/README.md b/README.md index ead86449..dd82fb82 100644 --- a/README.md +++ b/README.md @@ -5,8 +5,8 @@ ![Docker pull](https://ghcr-badge.elias.eu.org/shield/arabcoders/ytptube/ytptube) **YTPTube** is a web-based GUI for [yt-dlp](https://github.com/yt-dlp/yt-dlp), designed to make downloading videos from -YouTube and other video platforms easier and more user-friendly. It supports downloading playlists, channels, and -live streams, and includes features like scheduling downloads, sending notifications, and a built-in video player. +video platforms easier and user-friendly. It supports downloading playlists, channels, live streams and +includes features like scheduling downloads, sending notifications, and built-in video player. ![Short screenshot](https://raw.githubusercontent.com/ArabCoders/ytptube/master/sc_short.png) @@ -16,6 +16,7 @@ live streams, and includes features like scheduling downloads, sending notificat * Random beautiful background. * Handles live and upcoming streams. * Schedule channels or playlists to be downloaded automatically. +* Create your own custom task handler feeds for downloads, See [Feeds documentation](FAQ.md#how-can-i-monitor-sites-without-rss-feeds). * Send notification to targets based on selected events. includes [Apprise](https://github.com/caronc/apprise?tab=readme-ov-file#readme) support. * Support per link options. * Support for limits per extractor and overall global limit. @@ -38,7 +39,7 @@ Please read the [FAQ](FAQ.md) for more information. ## Run using docker command ```bash -mkdir -p ./{config,downloads} && docker run -d --rm --user "$UID:${GID-$UID}" --name ytptube \ +mkdir -p ./{config,downloads} && docker run -d --rm --user "${UID}:${UID}" --name ytptube \ -p 8081:8081 -v ./config:/config:rw -v ./downloads:/downloads:rw \ ghcr.io/arabcoders/ytptube:latest ``` @@ -93,7 +94,7 @@ For simple API documentation, you can refer to the [API documentation](API.md). # Disclaimer -This project is not affiliated with YouTube, yt-dlp, or any other service. It's a personal project that was created to +This project is not affiliated with yt-dlp, or any other service. It's a personal project that was created to make downloading videos from the internet easier. It's not intended to be used for piracy or any other illegal activities. # Social contact diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index 06d5cd9f..1ec7fe5d 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -465,6 +465,7 @@ class DownloadQueue(metaclass=Singleton): dl = ItemDTO( id=str(entry.get("id")), title=str(entry.get("title")), + description=str(entry.get("description", "")), url=str(entry.get("webpage_url") or entry.get("url")), preset=item.preset, folder=item.folder, @@ -982,6 +983,7 @@ class DownloadQueue(metaclass=Singleton): items["queue"][k] = self._active[k].info if k in self._active else v for k, v in self.done.saved_items(): + v.get_file_sidecar() items["done"][k] = v for k, v in self.queue.items(): @@ -989,8 +991,11 @@ class DownloadQueue(metaclass=Singleton): items["queue"][k] = self._active[k].info if k in self._active else v for k, v in self.done.items(): - if k not in items["done"]: - items["done"][k] = v.info + if k in items["done"]: + continue + + v.info.get_file_sidecar() + items["done"][k] = v.info return items @@ -1251,6 +1256,9 @@ class DownloadQueue(metaclass=Singleton): if task.cancelled(): return - if exc := task.exception(): - task_name: str = task.get_name() if task.get_name() else "unknown_task" - LOG.error(f"Unhandled exception in background task '{task_name}': {exc!s}. {traceback.format_exc()}") + if not (exc := task.exception()): + return + + task_name: str = task.get_name() if task.get_name() else "unknown_task" + tb = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__)) + LOG.error(f"Unhandled exception in background task '{task_name}': {exc!s}. {tb}") diff --git a/app/library/ItemDTO.py b/app/library/ItemDTO.py index 12449961..d638d45c 100644 --- a/app/library/ItemDTO.py +++ b/app/library/ItemDTO.py @@ -1,13 +1,22 @@ -import json import logging import re import time import uuid from dataclasses import dataclass, field from email.utils import formatdate +from pathlib import Path from typing import Any -from app.library.Utils import archive_add, archive_delete, archive_read, clean_item, get_archive_id +from app.library.encoder import Encoder +from app.library.Utils import ( + archive_add, + archive_delete, + archive_read, + clean_item, + get_archive_id, + get_file, + get_file_sidecar, +) from app.library.YTDLPOpts import YTDLPOpts LOG: logging.Logger = logging.getLogger("ItemDTO") @@ -47,12 +56,37 @@ class Item: """If the item should be started automatically.""" def serialize(self) -> dict: + """ + Serialize the item to a dictionary. + + Returns: + dict: The serialized item. + + """ return self.__dict__.copy() def json(self) -> str: - return json.dumps(self.serialize(), default=lambda o: o.__dict__, sort_keys=True, indent=4) + """ + Convert the item to a JSON string. + + Returns: + str: The JSON string representation of the item. + + """ + return Encoder(sort_keys=True, indent=4).encode(self.serialize()) def get(self, key: str, default: Any = None) -> Any: + """ + Get a value from the item by key. + + Args: + key (str): The key to get the value for. + default (Any): The default value to return if the key is not found. + + Returns: + Any: The value for the key, or the default value if the key is not found + + """ return self.__dict__.get(key, default) def has_extras(self) -> bool: @@ -77,6 +111,13 @@ class Item: @staticmethod def _default_preset() -> str: + """ + Get the default preset from the configuration. + + Returns: + str: The default preset name. + + """ from .config import Config return Config.get_instance().default_preset @@ -167,12 +208,33 @@ class Item: return Item(**data) def get_archive_id(self) -> str | None: + """ + Get the archive ID for the download URL. + + Returns: + str | None: The archive ID if available, None otherwise. + + """ return get_archive_id(self.url).get("archive_id") if self.url else None def get_extractor(self) -> str | None: + """ + Get the extractor key for the download URL. + + Returns: + str | None: The extractor key if available, None otherwise. + + """ return get_archive_id(self.url).get("ie_key") if self.url else None def get_ytdlp_opts(self) -> YTDLPOpts: + """ + Get the yt-dlp options for the item. + + Returns: + YTDLPOpts: The yt-dlp options for the item. + + """ params: YTDLPOpts = YTDLPOpts.get_instance() if self.preset: @@ -184,14 +246,35 @@ class Item: return params def get_archive_file(self) -> str | None: + """ + Get the archive file path from the yt-dlp options. + + Returns: + str | None: The archive file path if available, None otherwise. + + """ return self.get_ytdlp_opts().get_all().get("download_archive") def is_archived(self) -> bool: + """ + Check if the item has been archived. + + Returns: + bool: True if the item has been archived, False otherwise. + + """ archive_id: str | None = self.get_archive_id() archive_file: str | None = self.get_archive_file() return len(archive_read(archive_file, [archive_id])) > 0 if archive_file and archive_id else False def __repr__(self) -> str: + """ + Get a short string representation of the item. + + Returns: + str: A short string representation of the item. + + """ from .config import Config from .Utils import calc_download_path, strip_newline @@ -230,6 +313,8 @@ class ItemDTO: """ The ID of the item yt-dlp """ title: str """ The title of the item. """ + description: str = "" + """ The description of the item. """ url: str """ The URL of the item. """ preset: str = "default" @@ -272,6 +357,8 @@ class ItemDTO: """ If the item has been archived. """ archive_id: str | None = None """ The archive ID of the item. """ + sidecar: dict = field(default_factory=dict) + """ Sidecar data associated with the item. """ # yt-dlp injected fields. tmpfilename: str | None = None @@ -288,24 +375,101 @@ class ItemDTO: _archive_file: str | None = None def serialize(self) -> dict: - if "finished" == self.status and not self._recomputed: - self.archive_status() + """ + Serialize the item to a dictionary. + + Returns: + dict: The serialized item. + + """ + if "finished" == self.status: + if not self._recomputed: + self.archive_status() + + self.get_file_sidecar() item, _ = clean_item(self.__dict__.copy(), ItemDTO.removed_fields()) return item def json(self) -> str: - return json.dumps(self.serialize(), default=lambda o: o.__dict__, sort_keys=True, indent=4) + """ + Convert the item to a JSON string. + + Returns: + str: The JSON string representation of the item. + + """ + return Encoder(sort_keys=True, indent=4).encode(self.serialize()) def get(self, key: str, default: Any = None) -> Any: + """ + Get a value from the item by key. + + Args: + key (str): The key to get the value for. + default (Any): The default value to return if the key is not found. + + Returns: + Any: The value for the key, or the default value if the key is not found + + """ return self.__dict__.get(key, default) def get_id(self) -> str: + """ + Get the unique identifier for the item. + + Returns: + str: The unique identifier for the item. + + """ return self._id def name(self) -> str: + """ + Get a short string representation of the item. + + Returns: + str: A short string representation of the item. + + """ return f'id="{self.id}", title="{self.title}"' + def get_file(self, download_path: Path | None = None) -> Path | None: + """ + Get the file path of the downloaded item. + + Args: + download_path (Path | None): The base download path. If None, it will be taken from the config. + + Returns: + Path | None: The file path of the downloaded item, or None if not available. + + """ + if not self.filename: + return None + + if not download_path: + from .config import Config + + base_path = Path(Config.get_instance().download_path) + else: + base_path = download_path if isinstance(download_path, Path) else Path(download_path) + + filename = base_path + + if self.folder: + filename: Path = filename / self.folder + + filename = filename / self.filename + + try: + realFile, status = get_file(download_path=base_path, file=str(filename.relative_to(base_path))) + except Exception: + return None + + return Path(realFile) if status in (200, 302) else None + def get_archive_id(self) -> str | None: """ Get the archive ID for the download URL. @@ -326,6 +490,13 @@ class ItemDTO: return self.archive_id def get_extractor(self) -> str | None: + """ + Get the extractor key for the download URL. + + Returns: + str | None: The extractor key if available, None otherwise. + + """ if self.archive_id: return self.archive_id.split(" ")[0] @@ -353,6 +524,13 @@ class ItemDTO: return params def archive_status(self, force: bool = False) -> None: + """ + Recompute the archive status of the item. + + Args: + force (bool): If True, force recomputation even if already computed. + + """ if not force and (self._recomputed or not self.archive_id): return @@ -413,9 +591,28 @@ class ItemDTO: return True + def get_file_sidecar(self) -> dict: + """ + Get sidecar files associated with the item. + + Returns: + dict: A dictionary with sidecar files categorized by type. + + """ + if filename := self.get_file(): + self.sidecar = get_file_sidecar(filename) + + return self.sidecar + @staticmethod def removed_fields() -> tuple: - """Fields that once existed but are no longer used.""" + """ + Fields that once existed but are no longer used. + + Returns: + tuple: A tuple of field names that are no longer used. + + """ return ( "thumbnail", "quality", @@ -431,6 +628,10 @@ class ItemDTO: ) def __post_init__(self): + """ + Post-initialization to compute archive status if applicable. + """ self.get_archive_id() self.get_archive_file() self.archive_status() + self.get_file_sidecar() diff --git a/app/library/Notifications.py b/app/library/Notifications.py index e4b777e4..490ac303 100644 --- a/app/library/Notifications.py +++ b/app/library/Notifications.py @@ -1,9 +1,9 @@ import asyncio import json import logging +import traceback from collections.abc import Awaitable from dataclasses import dataclass, field -from datetime import datetime from pathlib import Path from typing import Any @@ -464,36 +464,36 @@ class Notification(metaclass=Singleton): try: LOG.info(f"Sending Notification event '{ev.event}: {ev.id}' to '{target.name}'.") - reqBody: dict[str, Any] = { - "method": target.request.method.upper(), - "url": target.request.url, - "headers": { - "User-Agent": f"YTPTube/{self._version}", - "X-Event-Id": ev.id, - "X-Event": ev.event, - "Content-Type": "application/json" - if "json" == target.request.type.lower() - else "application/x-www-form-urlencoded", - }, + headers: dict[str, str] = { + "User-Agent": f"YTPTube/{self._version}", + "X-Event-Id": ev.id, + "X-Event": ev.event, + "Content-Type": "application/json" + if "json" == target.request.type.lower() + else "application/x-www-form-urlencoded", } if len(target.request.headers) > 0: - for h in target.request.headers: - reqBody["headers"][h.key] = h.value + headers.update({h.key: h.value for h in target.request.headers if h.key and h.value}) - body_key: str = "json" if "json" == target.request.type.lower() else "data" - reqBody[body_key] = self._deep_unpack(ev.serialize()) + payload: dict = ev.serialize() if "data" != target.request.data_key: - reqBody[body_key][target.request.data_key] = reqBody[body_key]["data"] - reqBody[body_key].pop("data", None) + payload[target.request.data_key] = payload["data"] + payload.pop("data", None) if "form" == target.request.type.lower(): - reqBody[body_key][target.request.data_key] = self._encoder.encode( - reqBody[body_key][target.request.data_key] - ) + payload[target.request.data_key] = self._encoder.encode(payload[target.request.data_key]) + else: + payload = self._encoder.encode(payload) - response = await self._client.request(**reqBody) + response = await self._client.request( + method=target.request.method.upper(), + url=target.request.url, + headers=headers, + data=payload if "form" == target.request.type.lower() else None, + content=payload if "json" == target.request.type.lower() else None, + ) respData: dict[str, Any] = { "url": target.request.url, @@ -512,7 +512,9 @@ class Notification(metaclass=Singleton): err_msg = str(e) if not err_msg: err_msg: str = type(e).__name__ - LOG.error(f"Error sending Notification event '{ev.event}: {ev.id}' to '{target.name}'. '{err_msg!s}'.") + + tb = "".join(traceback.format_exception(type(e), e, e.__traceback__)) + LOG.error(f"Error sending Notification event '{ev.event}: {ev.id}' to '{target.name}'. '{err_msg!s}'. {tb}") return {"url": target.request.url, "status": 500, "text": str(ev)} def emit(self, e: Event, _, **__) -> None: @@ -522,18 +524,5 @@ class Notification(metaclass=Singleton): self._offload.submit(self.send, e) return - def _deep_unpack(self, data: dict) -> dict: - for k, v in data.items(): - if isinstance(v, dict): - data[k] = self._deep_unpack(v) - if isinstance(v, list): - data[k] = [self._deep_unpack(i) for i in v] - if isinstance(v, datetime): - data[k] = v.isoformat() - if isinstance(v, object) and hasattr(v, "serialize"): - data[k] = v.serialize() - - return data - async def noop(self) -> None: return None diff --git a/app/library/SegmentEncoders.py b/app/library/SegmentEncoders.py index 1040b3db..fb333fc3 100644 --- a/app/library/SegmentEncoders.py +++ b/app/library/SegmentEncoders.py @@ -7,9 +7,6 @@ import sys from pathlib import Path from typing import Any, Protocol -SUPPORTED_CODECS: tuple[str] = ("h264_qsv", "h264_nvenc", "h264_amf", "h264_videotoolbox", "h264_vaapi", "libx264") -"Supported encoder names in order of preference." - def has_dri_devices() -> bool: """ @@ -40,6 +37,7 @@ def ffmpeg_encoders() -> set[str]: set[str]: A set of available ffmpeg encoder names. """ + from .config import SUPPORTED_CODECS try: result: subprocess.CompletedProcess[str] = subprocess.run( ["ffmpeg", "-hide_banner", "-loglevel", "error", "-encoders"], # noqa: S607 @@ -75,6 +73,7 @@ def select_encoder(configured: str) -> str: str: The selected concrete encoder name. """ + from .config import SUPPORTED_CODECS configured = (configured or "").strip() avail: set[str] = ffmpeg_encoders() diff --git a/app/library/Segments.py b/app/library/Segments.py index 5c75b08d..c6dd78a4 100644 --- a/app/library/Segments.py +++ b/app/library/Segments.py @@ -10,15 +10,9 @@ from typing import TYPE_CHECKING, Any, ClassVar from aiohttp import web -from .config import Config +from .config import SUPPORTED_CODECS, Config from .ffprobe import ffprobe -from .SegmentEncoders import ( - SUPPORTED_CODECS, - encoder_fallback_chain, - get_builder_for_codec, - has_dri_devices, - select_encoder, -) +from .SegmentEncoders import encoder_fallback_chain, get_builder_for_codec, has_dri_devices, select_encoder if TYPE_CHECKING: from asyncio.subprocess import Process diff --git a/app/library/Tasks.py b/app/library/Tasks.py index 26b50847..b3318520 100644 --- a/app/library/Tasks.py +++ b/app/library/Tasks.py @@ -4,7 +4,8 @@ import json import logging import pkgutil import time -from dataclasses import dataclass +import uuid +from dataclasses import asdict, dataclass, field from datetime import UTC, datetime from pathlib import Path from typing import Any @@ -15,11 +16,11 @@ from .config import Config from .DownloadQueue import DownloadQueue from .encoder import Encoder from .Events import EventBus, Events -from .ItemDTO import Item +from .ItemDTO import Item, ItemDTO from .Scheduler import Scheduler from .Services import Services from .Singleton import Singleton -from .Utils import archive_add, archive_delete, extract_info, init_class, validate_url +from .Utils import archive_add, archive_delete, archive_read, extract_info, init_class, validate_url from .YTDLPOpts import YTDLPOpts LOG: logging.Logger = logging.getLogger("tasks") @@ -127,6 +128,95 @@ class Task: return {"file": archive_file, "items": items} +def _split_inspect_metadata(metadata: dict[str, Any] | None) -> tuple[dict[str, Any], dict[str, Any]]: + """ + Split commonly consumed metadata keys from the rest. + + Args: + metadata (dict[str, Any]|None): The metadata to split. + + Returns: + tuple[dict[str, Any], dict[str, Any]]: The primary and extra metadata. + + """ + metadata = dict(metadata or {}) + primary: dict[str, Any] = {} + + for key in ("matched", "handler", "supported"): + if key in metadata: + primary[key] = metadata.pop(key) + + return primary, metadata + + +@dataclass(slots=True) +class TaskItem: + url: str + "The URL of the item." + title: str | None = None + "The title of the item." + archive_id: str | None = None + "The archive ID of the item." + metadata: dict[str, Any] = field(default_factory=dict) + "Additional metadata related to the item." + + +@dataclass(slots=True) +class TaskResult: + items: list[TaskItem] = field(default_factory=list) + "The list of items." + metadata: dict[str, Any] = field(default_factory=dict) + "Additional metadata related to the result." + + def serialize(self) -> dict[str, Any]: + """ + Serialize the task result. + + Returns: + dict[str, Any]: The serialized task result. + + """ + primary, extra = _split_inspect_metadata(self.metadata) + payload: dict[str, Any] = {**primary, "items": [asdict(item) for item in self.items]} + + if extra: + payload["metadata"] = extra + + return payload + + +@dataclass(slots=True) +class TaskFailure: + message: str + "A human-readable message describing the failure." + error: str | None = None + "An optional error code or string." + metadata: dict[str, Any] = field(default_factory=dict) + "Additional metadata related to the failure." + + def serialize(self) -> dict[str, Any]: + """ + Serialize the task failure. + + Returns: + dict[str, Any]: The serialized task failure. + + """ + primary, extra = _split_inspect_metadata(self.metadata) + payload: dict[str, Any] = dict(primary) + + if self.error: + payload["error"] = self.error + + if self.message and (not self.error or self.message != self.error): + payload["message"] = self.message + + if extra: + payload["metadata"] = extra + + return payload + + class Tasks(metaclass=Singleton): """ This class is used to manage the tasks. @@ -199,6 +289,7 @@ class Tasks(metaclass=Singleton): """ self.load() + Services.get_instance().add("tasks", self) async def event_handler(data, _): if data and data.data: @@ -211,6 +302,10 @@ class Tasks(metaclass=Singleton): """Return the tasks.""" return self._tasks + def get_handler(self) -> "HandleTask": + """Expose the handle task helper.""" + return self._task_handler + def get(self, task_id: str) -> Task | None: """ Get a task by its ID. @@ -471,8 +566,21 @@ class HandleTask: "The configuration." self._task_name: str = f"{__class__.__name__}._dispatcher" "The task name for the scheduler." + self._queued: dict[str, set[str]] = {} + "Queued archive IDs per handler." + self._failure_count: dict[str, dict[str, int]] = {} + "Failure counts per handler and archive ID." + + EventBus.get_instance().subscribe( + Events.ITEM_ERROR, + self._handle_item_error, + f"{__class__.__name__}.item_error", + ) def load(self) -> None: + """ + Load the available handlers and schedule the dispatcher. + """ self._handlers: list[type] = self._discover() timer: str = self._config.tasks_handler_timer @@ -551,7 +659,7 @@ class HandleTask: return None - async def dispatch(self, task: Task, handler: type | None = None, **kwargs) -> Any | None: + async def dispatch(self, task: Task, handler: type | None = None, **kwargs) -> TaskResult | TaskFailure | None: # noqa: ARG002 """ Dispatch a task to the appropriate handler. @@ -561,7 +669,7 @@ class HandleTask: **kwargs: Additional context to pass to the handler. Returns: - Any|None: The result of the handler's execution, or None if no handler found. + TaskResult|TaskFailure|None: The extraction outcome, or None if no handler matched. """ if not handler: @@ -569,12 +677,225 @@ class HandleTask: if handler is None: return None + services: Services = Services.get_instance() + try: - return await Services.get_instance().handle_async(handler=handler.handle, task=task, **kwargs) - except Exception as e: - LOG.exception(e) + extraction: TaskResult | TaskFailure = await services.handle_async( + handler=handler.extract, task=task, config=self._config + ) + except NotImplementedError: + LOG.error(f"Handler '{handler.__name__}' does not implement extract().") + return TaskFailure(message="Handler does not support extraction.") + except Exception as exc: + LOG.exception(exc) raise + if isinstance(extraction, TaskFailure): + LOG.error(f"Handler '{handler.__name__}' failed to extract items: {extraction.message}") + return extraction + + if not isinstance(extraction, TaskResult): + LOG.error( + f"Handler '{handler.__name__}' returned unexpected result type '{type(extraction).__name__}'.", + ) + return TaskFailure( + message="Handler returned invalid result type.", metadata={"type": type(extraction).__name__} + ) + + raw_items: list[TaskItem] = extraction.items or [] + metadata: dict[str, Any] = extraction.metadata or {} + + handler_name: str = handler.__name__ + queued: set[str] = self._queued.setdefault(handler_name, set()) + failures: dict[str, int] = self._failure_count.setdefault(handler_name, {}) + + params: dict = task.get_ytdlp_opts().get_all() + archive_file: str | None = params.get("download_archive") + + download_queue: DownloadQueue = services.get("queue") or DownloadQueue.get_instance() + notify: EventBus = services.get("notify") or EventBus.get_instance() + + archive_ids: list[str] = [ + item.archive_id for item in raw_items if isinstance(item, TaskItem) and item.archive_id + ] + downloaded: list[str] = archive_read(archive_file, archive_ids) if archive_file else [] + + filtered: list[TaskItem] = [] + + for item in raw_items: + if not isinstance(item, TaskItem): + LOG.warning("Handler '{handler.__name__}' produced non-TaskItem entry: {item!r}") + continue + + url: str = item.url + if not url: + continue + + archive_id: str | None = item.archive_id + if not archive_id: + LOG.warning(f"'{task.name}': Item with URL '{url}' is missing an archive ID. Skipping.") + continue + + if archive_id in queued: + continue + + queued.add(archive_id) + + if archive_file and archive_id in downloaded: + continue + + if download_queue.queue.exists(url=url): + continue + + try: + done = download_queue.done.get(url=url) + if "error" != done.info.status: + continue + except KeyError: + pass + + if archive_id not in failures: + failures[archive_id] = 0 + + filtered.append(item) + + if not filtered: + if raw_items: + LOG.debug( + f"Handler '{handler.__name__}' produced '{len(raw_items)}' for '{task.name}' items, none queued after filtering." + ) + return TaskResult(items=[], metadata=metadata) + + LOG.info( + f"Handler '{handler.__name__}' Found '{len(filtered)}' new items for '{task.name}' (raw={len(raw_items)})." + ) + + base_item = Item.format( + { + "url": task.url, + "preset": task.preset or self._config.default_preset, + "folder": task.folder or "", + "template": task.template or "", + "cli": task.cli or "", + "auto_start": task.auto_start, + "extras": {"source_task": task.id, "source_handler": handler.__name__}, + } + ) + + for item in filtered: + metadata_entry: dict[str, Any] = item.metadata if isinstance(item.metadata, dict) else {} + extras: dict[str, Any] = base_item.extras.copy() + if metadata_entry: + extras["metadata"] = metadata_entry + + notify.emit( + Events.ADD_URL, + data=base_item.new_with(url=item.url, extras=extras).serialize(), + ) + + return TaskResult(items=filtered, metadata=metadata) + + async def inspect( + self, + url: str, + preset: str | None = None, + handler_name: str | None = None, + ) -> TaskResult | TaskFailure: + if not self._handlers: + self._handlers = self._discover() + + task = Task( + id=str(uuid.uuid4()), + name="Inspector", + url=url, + preset=preset or self._config.default_preset, + auto_start=False, + ) + + services = Services.get_instance() + + handler_cls: type | None + if handler_name: + handler_cls = next((cls for cls in self._handlers if cls.__name__.lower() == handler_name.lower()), None) + if handler_cls is None: + message: str = f"Handler '{handler_name}' not found." + return TaskFailure( + message=message, + error=message, + metadata={"matched": False, "handler": handler_name}, + ) + + try: + matched = services.handle_sync(handler=handler_cls.can_handle, task=task) + except Exception as exc: # pragma: no cover - defensive + LOG.exception(exc) + message = str(exc) + return TaskFailure( + message=message, + error=message, + metadata={"matched": False, "handler": handler_cls.__name__}, + ) + + if not matched: + return TaskFailure( + message="Handler cannot process the supplied URL.", + metadata={"matched": False, "handler": handler_cls.__name__}, + ) + else: + handler_cls = self._find_handler(task) + if handler_cls is None: + message = "No handler matched the supplied URL." + return TaskFailure( + message=message, + error=message, + metadata={"matched": False, "handler": None}, + ) + + base_metadata: dict[str, Any] = {"matched": True, "handler": handler_cls.__name__} + + try: + extraction: TaskResult | TaskFailure = await services.handle_async( + handler=handler_cls.extract, task=task, config=self._config + ) + except NotImplementedError: + return TaskFailure( + message="Handler does not support manual inspection.", + metadata={**base_metadata, "supported": False}, + ) + except Exception as exc: + LOG.exception(exc) + message = str(exc) + return TaskFailure( + message=message, + error=message, + metadata={**base_metadata, "supported": True}, + ) + + if isinstance(extraction, TaskFailure): + combined_failure_metadata: dict[str, Any] = {**base_metadata, "supported": True} + if extraction.metadata: + combined_failure_metadata.update(extraction.metadata) + + failure_error = extraction.error if extraction.error else extraction.message + + return TaskFailure( + message=extraction.message, + error=failure_error, + metadata=combined_failure_metadata, + ) + + if not isinstance(extraction, TaskResult): + LOG.error( + f"Handler '{handler_cls.__name__}' returned unexpected result type '{type(extraction).__name__}' during inspection.", + ) + extraction = TaskResult() + + combined_metadata: dict[str, Any] = {**base_metadata, "supported": True} + if extraction.metadata: + combined_metadata.update(extraction.metadata) + + return TaskResult(items=list(extraction.items), metadata=combined_metadata) + def _discover(self) -> list[type]: import importlib @@ -591,7 +912,28 @@ class HandleTask: if cls.__module__ != module.__name__: continue - if callable(getattr(cls, "can_handle", None)) and callable(getattr(cls, "handle", None)): + if callable(getattr(cls, "can_handle", None)) and callable(getattr(cls, "extract", None)): handlers.append(cls) return handlers + + async def _handle_item_error(self, event, _name, **_kwargs): + item = getattr(event, "data", None) + if not isinstance(item, ItemDTO): + return + + extras = getattr(item, "extras", {}) or {} + handler_name = extras.get("source_handler") + if not handler_name: + return + + archive_id = item.archive_id + if not archive_id: + return + + queued = self._queued.get(handler_name) + if queued: + queued.discard(archive_id) + + failures = self._failure_count.setdefault(handler_name, {}) + failures[archive_id] = failures.get(archive_id, 0) + 1 diff --git a/app/library/Utils.py b/app/library/Utils.py index 8411827b..b6c4a992 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -8,9 +8,10 @@ import os import re import shlex import socket +import time import uuid from datetime import UTC, datetime, timedelta -from functools import lru_cache +from functools import lru_cache, wraps from http.cookiejar import MozillaCookieJar from pathlib import Path from typing import Any, TypeVar @@ -87,8 +88,6 @@ DT_PATTERN: re.Pattern[str] = re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2} T = TypeVar("T") "Generic type variable." -ARCHIVE_IDS_CACHE: dict[str, dict] = {} -"Cache for archive IDs." class StreamingError(Exception): """Raised when an error occurs during streaming.""" @@ -99,6 +98,112 @@ class FileLogFormatter(logging.Formatter): return datetime.fromtimestamp(record.created).astimezone().isoformat(timespec="milliseconds") +def timed_lru_cache(ttl_seconds: int, max_size: int = 128): + """ + Decorator that applies an LRU cache with a time-to-live (TTL) to a function. + Supports both synchronous and asynchronous functions. + + Args: + ttl_seconds (int): Time-to-live in seconds. + max_size (int): Maximum size of the cache. + + Returns: + Decorated function with caching capabilities. + + """ + import inspect + + def decorator(func): + is_async = inspect.iscoroutinefunction(func) + + if is_async: + # For async functions, we need to cache the actual result, not the coroutine + cache = {} + cache_expiry = {} + + @wraps(func) + async def async_wrapper(*args, **kwargs): + key = str(args) + str(sorted(kwargs.items())) + current_time = time.monotonic() + + # Check if we have a cached result that hasn't expired + if key in cache and key in cache_expiry and current_time < cache_expiry[key]: + cached_result = cache[key] + try: + return copy.deepcopy(cached_result) + except Exception: + return cached_result + + # Call the function and cache the result + result = await func(*args, **kwargs) + try: + cached_value = copy.deepcopy(result) + except Exception: + cached_value = result + + cache[key] = cached_value + cache_expiry[key] = current_time + ttl_seconds + + # Limit cache size + if len(cache) > max_size: + # Remove oldest entries + oldest_keys = sorted(cache_expiry.keys(), key=lambda k: cache_expiry[k])[: len(cache) - max_size] + for old_key in oldest_keys: + cache.pop(old_key, None) + cache_expiry.pop(old_key, None) + + try: + return copy.deepcopy(cached_value) + except Exception: + return cached_value + + # Expose cache management methods + def cache_clear(): + cache.clear() + cache_expiry.clear() + + def cache_info(): + # Use functools._CacheInfo for compatibility with standard lru_cache + from functools import _CacheInfo + + return _CacheInfo(hits=0, misses=len(cache), maxsize=max_size, currsize=len(cache)) + + async_wrapper.cache_clear = cache_clear + async_wrapper.cache_info = cache_info + return async_wrapper + + # For sync functions, use the original implementation + + @lru_cache(maxsize=max_size) + def cached(*args, **kwargs): + result = func(*args, **kwargs) + try: + return copy.deepcopy(result) + except Exception: + return result + + cached_expiry = time.monotonic() + ttl_seconds + + @wraps(func) + def sync_wrapper(*args, **kwargs): + nonlocal cached_expiry + if time.monotonic() >= cached_expiry: + cached.cache_clear() + cached_expiry = time.monotonic() + ttl_seconds + result = cached(*args, **kwargs) + try: + return copy.deepcopy(result) + except Exception: + return result + + # expose cache_clear, cache_info + sync_wrapper.cache_clear = cached.cache_clear + sync_wrapper.cache_info = cached.cache_info + return sync_wrapper + + return decorator + + def calc_download_path(base_path: str | Path, folder: str | None = None, create_path: bool = True) -> str: """ Calculates download path and prevents folder traversal. @@ -387,7 +492,7 @@ def check_id(file: Path) -> bool | str: return False -@lru_cache(maxsize=512) +@timed_lru_cache(ttl_seconds=300, max_size=256) def is_private_address(hostname: str) -> bool: ip: str = socket.gethostbyname(hostname) ip_obj: ipaddress.IPv4Address | ipaddress.IPv6Address = ipaddress.ip_address(ip) @@ -545,19 +650,23 @@ def validate_uuid(uuid_str: str, version: int = 4) -> bool: return False -def get_file_sidecar(file: Path) -> list[dict]: +@timed_lru_cache(ttl_seconds=60, max_size=256) +def get_file_sidecar(file: Path | None = None) -> dict[dict]: """ Get sidecar files for the given file. Args: - file (Path): The video file. + file (Path|None): The video file. Returns: - list: List of sidecar files. + dict: A dictionary with sidecar files categorized by type. """ files: dict = {} + if not file: + return files + for i, f in enumerate(file.parent.glob(f"{glob.escape(file.stem)}.*")): if f == file or f.is_file() is False or f.stem.startswith("."): continue @@ -595,7 +704,6 @@ def get_file_sidecar(file: Path) -> list[dict]: return files -@lru_cache(maxsize=512) def get_possible_images(dir: str) -> list[dict]: images: list = [] @@ -1093,6 +1201,7 @@ def load_cookies(file: str | Path) -> tuple[bool, MozillaCookieJar]: raise ValueError(msg) from e +@timed_lru_cache(ttl_seconds=300, max_size=256) def get_archive_id(url: str) -> dict[str, str | None]: """ Get the archive ID for a given URL. @@ -1116,9 +1225,6 @@ def get_archive_id(url: str) -> dict[str, str | None]: "archive_id": None, } - if url in ARCHIVE_IDS_CACHE: - return ARCHIVE_IDS_CACHE[url] - if YTDLP_INFO_CLS is None: YTDLP_INFO_CLS = YTDLP( params={ @@ -1151,7 +1257,6 @@ def get_archive_id(url: str) -> dict[str, str | None]: LOG.exception(e) LOG.error(f"Error getting archive ID: {e}") - ARCHIVE_IDS_CACHE.update({url: idDict}) return idDict diff --git a/app/library/config.py b/app/library/config.py index d3f099ec..65f5ac1e 100644 --- a/app/library/config.py +++ b/app/library/config.py @@ -19,6 +19,9 @@ from .version import APP_BRANCH, APP_BUILD_DATE, APP_COMMIT_SHA, APP_VERSION if TYPE_CHECKING: from subprocess import CompletedProcess +SUPPORTED_CODECS: tuple[str] = ("h264_qsv", "h264_nvenc", "h264_amf", "h264_videotoolbox", "h264_vaapi", "libx264") +"Supported encoder names in order of preference." + class Config(metaclass=Singleton): app_env: str = "production" @@ -417,6 +420,11 @@ class Config(metaclass=Singleton): msg: str = f"Invalid app environment '{self.app_env}' specified. Must be 'production' or 'development'." raise ValueError(msg) + if self.streamer_vcodec and self.streamer_vcodec not in SUPPORTED_CODECS: + supported = ", ".join(SUPPORTED_CODECS) + msg: str = f"Invalid video codec '{self.streamer_vcodec}' specified. Supported: '{supported}'." + raise ValueError(msg) + if "dev-master" == self.app_version: self._version_via_git() diff --git a/app/library/ffprobe.py b/app/library/ffprobe.py index 54ba5404..78d0d142 100644 --- a/app/library/ffprobe.py +++ b/app/library/ffprobe.py @@ -10,12 +10,10 @@ import operator import os import subprocess # qa: ignore from pathlib import Path -from typing import TYPE_CHECKING import anyio -if TYPE_CHECKING: - from app.library.cache import Cache +from app.library.Utils import timed_lru_cache LOG: logging.Logger = logging.getLogger(__name__) @@ -238,7 +236,8 @@ class FFProbeResult: } -async def ffprobe(file: str) -> FFProbeResult: +@timed_lru_cache(ttl_seconds=300, max_size=128) +async def ffprobe(file: Path|str) -> FFProbeResult: """ Run ffprobe on a file and return the parsed data as a dictionary. @@ -249,21 +248,12 @@ async def ffprobe(file: str) -> FFProbeResult: dict: A dictionary containing the parsed data. """ - from app.library.Services import Services - - f = Path(file) + f = Path(file) if isinstance(file, str) else file if not f.exists(): msg = f"No such media file '{file}'." raise OSError(msg) - cache: Cache | None = Services.get_instance().get("cache") - cache_key: str = f"ffprobe:{f!s}:{f.stat().st_size}" - - if cache and (cached := cache.get(cache_key)): - LOG.debug(f"ffprobe cache hit for '{cache_key}'") - return cached - try: async with await anyio.open_file(os.devnull, "w") as tempf: await asyncio.create_subprocess_exec( @@ -310,7 +300,4 @@ async def ffprobe(file: str) -> FFProbeResult: elif stream.is_attachment(): result.attachment.append(stream) - if cache: - cache.set(cache_key, result, ttl=300) - return result diff --git a/app/library/task_handlers/_base_handler.py b/app/library/task_handlers/_base_handler.py index 4d36e585..b81ce9d5 100644 --- a/app/library/task_handlers/_base_handler.py +++ b/app/library/task_handlers/_base_handler.py @@ -1,73 +1,30 @@ # flake8: noqa: ARG004 -import logging from typing import Any import httpx from yt_dlp.utils.networking import random_user_agent from app.library.config import Config -from app.library.DownloadQueue import DownloadQueue -from app.library.Events import EventBus, Events -from app.library.ItemDTO import ItemDTO -from app.library.Tasks import Task - -LOG: logging.Logger = logging.getLogger(__name__) +from app.library.Tasks import Task, TaskFailure, TaskResult class BaseHandler: - queued: set[str] = set() - failure_count: dict[str, int] = {} - - def __init_subclass__(cls, **kwargs): - """Ensure each subclass has its own state containers.""" - super().__init_subclass__(**kwargs) - if "queued" not in cls.__dict__: - cls.queued = set() - if "failure_count" not in cls.__dict__: - cls.failure_count = {} - - async def event_handler(data, _): - if data and data.data: - await cls.on_error(data.data) - - EventBus.get_instance().subscribe(Events.ITEM_ERROR, event_handler, f"{cls.__name__}.on_error") - @staticmethod def can_handle(task: Task) -> bool: return False @staticmethod - async def handle(task: Task, notify: EventBus, config: Config, queue: DownloadQueue): - pass + async def extract(task: Task, config: Config | None = None) -> TaskResult | TaskFailure: + raise NotImplementedError + + @classmethod + async def inspect(cls, task: Task, config: Config | None = None) -> TaskResult | TaskFailure: + return await cls.extract(task=task, config=config) @staticmethod def parse(url: str) -> Any | None: return None - @classmethod - async def on_error(cls, item: ItemDTO) -> None: - """ - Handle errors by logging them and removing the queued ID if it exists. - - Args: - item (ItemDTO): The error data containing the URL and other information. - - """ - if not item or not isinstance(item, ItemDTO): - return - - if not item.archive_id or not cls.failure_count.get(item.archive_id, None): - LOG.debug(f"Item '{item.name()}' not queued by the handler.") - return - - failCount: int = int(cls.failure_count.get(item.archive_id, 0)) - - LOG.info(f"Removing '{item.name()}' from queued IDs due to error. Failure count: '{failCount + 1}'.") - if item.archive_id in cls.queued: - cls.queued.remove(item.archive_id) - - cls.failure_count[item.archive_id] = 1 + failCount - @staticmethod def tests() -> list[tuple[str, bool]]: return [] diff --git a/app/library/task_handlers/generic.py b/app/library/task_handlers/generic.py new file mode 100644 index 00000000..96d8d32e --- /dev/null +++ b/app/library/task_handlers/generic.py @@ -0,0 +1,1280 @@ +"""Generic task handler driven by JSON definitions.""" + +from __future__ import annotations + +import asyncio +import fnmatch +import hashlib +import json +import logging +import re +from collections.abc import Iterable, Mapping, MutableMapping +from dataclasses import dataclass, field +from pathlib import Path +from typing import TYPE_CHECKING, Any, Literal +from urllib.parse import urljoin + +import httpx +import jmespath +from parsel import Selector +from parsel.selector import SelectorList +from yt_dlp.utils.networking import random_user_agent + +from app.library.config import Config +from app.library.Tasks import Task, TaskFailure, TaskItem, TaskResult +from app.library.Utils import get_archive_id + +from ._base_handler import BaseHandler + +if TYPE_CHECKING: + from parsel.selector import SelectorList + +LOG: logging.Logger = logging.getLogger(__name__) + + +@dataclass(slots=True) +class MatchRule: + """Represents a single URL matcher compiled to regex.""" + + source: str + """Original source pattern (regex or glob).""" + + regex: re.Pattern[str] + """Compiled regex pattern.""" + + @classmethod + def from_value(cls, value: str | Mapping[str, Any]) -> MatchRule | None: + """ + Create a MatchRule from a string or mapping. + + Args: + value (str|Mapping[str, Any]): A string (treated as glob) or a mapping with 'regex' or 'glob' keys. + + Returns: + (MatchRule|None): A MatchRule instance if successful, None otherwise. + + """ + if isinstance(value, Mapping): + pattern: str | None = value.get("regex") + glob_pattern: str | None = value.get("glob") + raw: str | None = None + + if isinstance(pattern, str) and pattern: + raw = pattern + try: + compiled: re.Pattern[str] = re.compile(pattern) + except re.error as exc: + LOG.error(f"Invalid regex pattern '{pattern}': {exc}") + return None + + return cls(source=raw, regex=compiled) + + if isinstance(glob_pattern, str) and glob_pattern: + raw = glob_pattern + compiled = re.compile(fnmatch.translate(glob_pattern)) + return cls(source=raw, regex=compiled) + + LOG.error("Matcher mapping must include 'regex' or 'glob' key with a string value.") + return None + + if not isinstance(value, str) or not value: + LOG.error(f"Matcher value must be a non-empty string, got '{value!r}'.") + return None + + # Treat plain string as glob pattern for convenience. + compiled = re.compile(fnmatch.translate(value)) + return cls(source=value, regex=compiled) + + def matches(self, url: str) -> bool: + """ + Check if the given URL matches this rule. + + Args: + url (str): The URL to check. + + Returns: + (bool): True if the URL matches, False otherwise. + + """ + return bool(self.regex.match(url)) + + +@dataclass(slots=True) +class PostFilter: + """Regex post-filter applied on extracted values.""" + + pattern: re.Pattern[str] + """Compiled regex pattern.""" + + value_key: str | None = None + """Optional group name or index to extract from the match.""" + + @classmethod + def from_mapping(cls, mapping: Mapping[str, Any]) -> PostFilter | None: + """ + Create a PostFilter from a mapping. + + Args: + mapping (Mapping[str,Any]): A mapping with 'filter' (regex pattern) and optional 'value' (group name or index) keys. + + Returns: + (PostFilter|None): A PostFilter instance if successful, None otherwise. + + """ + pattern: str | None = mapping.get("filter") + if not isinstance(pattern, str) or not pattern: + LOG.error("post_filter requires a non-empty 'filter' string.") + return None + + try: + compiled: re.Pattern[str] = re.compile(pattern) + except re.error as exc: + LOG.error(f"Invalid post_filter regex '{pattern}': {exc}") + return None + + value_key: str | None = mapping.get("value") + if value_key is not None and not isinstance(value_key, str): + LOG.error("post_filter 'value' must be a string if provided.") + return None + + return cls(pattern=compiled, value_key=value_key) + + def apply(self, candidate: str) -> str | None: + """ + Apply the post-filter to the candidate string. + + Args: + candidate (str): The string to filter. + + Returns: + (str|None): The filtered value if matched, None otherwise. + + """ + match: re.Match[str] | None = self.pattern.search(candidate) + if not match: + return None + + if self.value_key: + try: + return match.group(self.value_key) + except IndexError: + LOG.warning( + f"post_filter value index '{self.value_key}' not present in pattern {self.pattern.pattern!r}." + ) + except KeyError: + LOG.warning( + f"post_filter value key '{self.value_key}' not present in pattern {self.pattern.pattern!r}." + ) + return None + + if match.groupdict(): + # Prefer first named group when available. + key, value = next(iter(match.groupdict().items())) + if value is not None: + LOG.debug(f"post_filter using named group '{key}'.") + return value + + if match.groups(): + return match.group(1) + + return match.group(0) + + +@dataclass(slots=True) +class ExtractionRule: + """Single field extraction description.""" + + type: Literal["css", "xpath", "regex"] + """Type of extraction to perform.""" + + expression: str + """CSS selector, XPath expression or regex pattern.""" + + attribute: str | None = None + """Optional attribute to extract (e.g. 'href', 'src', 'text', etc.).""" + + post_filter: PostFilter | None = None + """Optional post-filter to apply on extracted values.""" + + +@dataclass(slots=True) +class EngineConfig: + """Engine selection to fetch the page.""" + + type: Literal["httpx", "selenium"] = "httpx" + """Engine type to use.""" + + options: dict[str, Any] = field(default_factory=dict) + """Engine-specific options.""" + + +@dataclass(slots=True) +class RequestConfig: + """HTTP request configuration.""" + + method: str = "GET" + """HTTP method to use.""" + headers: dict[str, str] = field(default_factory=dict) + """HTTP headers to include.""" + params: dict[str, Any] = field(default_factory=dict) + """Query parameters to include.""" + + data: Any | None = None + """Request body data to include.""" + json: Any | None = None + """Request body JSON data to include.""" + timeout: float | None = None + """Request timeout in seconds.""" + url: str | None = None + """Optional URL to use instead of the task URL.""" + + def normalized_method(self) -> str: + """ + Get the HTTP method in uppercase. + + Returns: + (str): The HTTP method in uppercase. + + """ + return self.method.upper() if isinstance(self.method, str) else "GET" + + +@dataclass(slots=True) +class ResponseConfig: + """Defines how to interpret the response body returned by the fetch engine.""" + + format: Literal["html", "json"] = "html" + """Body format. Defaults to HTML.""" + + +@dataclass(slots=True) +class ContainerDefinition: + """Defines a repeating element with nested field extraction.""" + + selector_type: Literal["css", "xpath", "jsonpath"] + """Type of selector to use for locating container elements.""" + + selector: str + """Selector expression for locating container elements.""" + + fields: dict[str, ExtractionRule] + """Field extraction rules relative to the container.""" + + +@dataclass(slots=True) +class TaskDefinition: + """Full task definition as loaded from disk.""" + + name: str + """Human-readable name of the task definition.""" + source: Path + """Path to the source JSON file.""" + matchers: list[MatchRule] + """List of URL matchers.""" + engine: EngineConfig + """Engine configuration.""" + request: RequestConfig + """Request configuration.""" + parsers: dict[str, ExtractionRule] + """Field extraction rules.""" + container: ContainerDefinition | None = None + """Optional container definition for repeating elements.""" + response: ResponseConfig = field(default_factory=ResponseConfig) + """Response configuration.""" + + def matches(self, url: str) -> bool: + """ + Check if the given URL matches any of the defined matchers. + + Args: + url (str): The URL to check. + + Returns: + (bool): True if any matcher matches the URL, False otherwise. + + """ + return any(rule.matches(url) for rule in self.matchers) + + +def _build_extraction_rule(field: str, raw: Mapping[str, Any], *, source: Path) -> ExtractionRule | None: + """ + Build an ExtractionRule from a raw mapping. + + Args: + field (str): The name of the field being defined. + raw (Mapping[str, Any]): The raw mapping defining the extraction rule. + source (Path): Path to the source JSON file for logging context. + + Returns: + (ExtractionRule|None): An ExtractionRule instance if successful, None otherwise. + + """ + type_value: str | None = raw.get("type") + expression: str | None = raw.get("expression") + + if not isinstance(type_value, str): + LOG.error(f"[{source.name}] Field '{field}' is missing a valid 'type'.") + return None + + if type_value not in {"css", "xpath", "regex", "jsonpath"}: + LOG.error(f"[{source.name}] Field '{field}' has unsupported type '{type_value}'.") + return None + + if not isinstance(expression, str) or not expression: + LOG.error(f"[{source.name}] Field '{field}' requires non-empty 'expression'.") + return None + + attribute: str | None = raw.get("attribute") + if attribute is not None and not isinstance(attribute, str): + LOG.error(f"[{source.name}] Field '{field}' attribute must be a string if provided.") + return None + + post_filter: PostFilter | None = None + if isinstance(raw.get("post_filter"), Mapping): + post_filter = PostFilter.from_mapping(raw["post_filter"]) + + return ExtractionRule(type=type_value, expression=expression, attribute=attribute, post_filter=post_filter) + + +def _build_matchers(raw_match: Iterable[Any], *, source: Path) -> list[MatchRule]: + """ + Build a list of MatchRule instances from raw match definitions. + + Args: + raw_match (Iterable[Any]): An iterable of raw match definitions (strings or mappings). + source (Path): Path to the source JSON file for logging context. + + Returns: + (list[MatchRule]): A list of MatchRule instances. + + """ + matchers: list[MatchRule] = [] + for value in raw_match: + rule: MatchRule | None = MatchRule.from_value(value) + if rule: + matchers.append(rule) + + if not matchers: + LOG.error(f"[{source.name}] No valid match rules found.") + + return matchers + + +def _normalize_mapping(value: Any) -> MutableMapping[str, Any]: + """ + Ensure the value is a mutable mapping. + + Args: + value (Any): The value to check. + + Returns: + (MutableMapping[str, Any]): The value if it's a mutable mapping. + + """ + if isinstance(value, MutableMapping): + return value + + msg = "Expected a mapping for parse/request/engine sections." + raise ValueError(msg) + + +def load_task_definitions(config: Config | None = None) -> list[TaskDefinition]: + """ + Load JSON task definitions from the configured tasks directory. + + Args: + config (Config|None): Optional Config instance. If None, the singleton instance is used. + + Returns: + (list[TaskDefinition]): A list of loaded TaskDefinition instances. + + """ + cfg: Config = config or Config.get_instance() + tasks_dir: Path = Path(cfg.config_path) / "tasks" + + if not tasks_dir.exists(): + return [] + + definitions: list[TaskDefinition] = [] + + for path in sorted(tasks_dir.glob("*.json")): + try: + content: str = path.read_text(encoding="utf-8") + except Exception as exc: + LOG.error(f"Failed to read task configuration '{path}': {exc}") + continue + + try: + raw = json.loads(content) + except Exception as exc: + LOG.error(f"Failed to parse JSON for '{path}': {exc}") + continue + + if not isinstance(raw, Mapping): + LOG.error(f"Task definition in '{path}' must be a JSON object.") + continue + + name: str | None = raw.get("name") + if not isinstance(name, str) or not name.strip(): + LOG.error(f"Task definition '{path}' missing a valid 'name'.") + continue + + match_value: list[str] | None = raw.get("match") + if not isinstance(match_value, Iterable) or isinstance(match_value, (str, bytes)): + LOG.error(f"[{path.name}] 'match' must be a list of patterns.") + continue + + matchers: list[MatchRule] = _build_matchers(match_value, source=path) + if not matchers: + continue + + engine_raw: Any = raw.get("engine", {}) + try: + engine_map: MutableMapping[str, Any] = _normalize_mapping(engine_raw) + except ValueError: + LOG.error(f"[{path.name}] 'engine' must be a JSON object when provided.") + continue + + engine_type: str | None = engine_map.get("type", "httpx") + if engine_type not in ("httpx", "selenium"): + LOG.error(f"[{path.name}] Unsupported engine type '{engine_type}'.") + continue + + engine_options: Any = engine_map.get("options") if isinstance(engine_map.get("options"), Mapping) else {} + engine = EngineConfig(type=engine_type, options=dict(engine_options)) + + request_raw: Any = raw.get("request", {}) + try: + request_map: MutableMapping[str, Any] = _normalize_mapping(request_raw) + except ValueError: + LOG.error(f"[{path.name}] 'request' must be a JSON object when provided.") + continue + + request = RequestConfig( + method=str(request_map.get("method", "GET")), + headers=dict(request_map.get("headers", {})) if isinstance(request_map.get("headers"), Mapping) else {}, + params=dict(request_map.get("params", {})) if isinstance(request_map.get("params"), Mapping) else {}, + data=request_map.get("data"), + json=request_map.get("json"), + timeout=float(request_map.get("timeout")) if request_map.get("timeout") is not None else None, + url=str(request_map.get("url")) if isinstance(request_map.get("url"), str) else None, + ) + + response_raw: Any = raw.get("response", {}) + response_config = ResponseConfig() + if response_raw: + if not isinstance(response_raw, Mapping): + LOG.error(f"[{path.name}] 'response' must be an object when provided.") + continue + + response_type: str = str(response_raw.get("type", "html")).lower() + if response_type not in ("html", "json"): + LOG.error(f"[{path.name}] Unsupported response type '{response_type}'.") + continue + + response_config = ResponseConfig(format=response_type) # type: ignore[arg-type] + + parse_raw: Mapping | None = raw.get("parse") + if not isinstance(parse_raw, Mapping): + LOG.error(f"[{path.name}] 'parse' must be a JSON object mapping fields to instructions.") + continue + + container_definition: ContainerDefinition | None = None + parsers: dict[str, ExtractionRule] = {} + + items_block: Mapping | None = parse_raw.get("items") + if isinstance(items_block, Mapping): + raw_fields: Mapping | None = items_block.get("fields") + if not isinstance(raw_fields, Mapping): + LOG.error(f"[{path.name}] 'items.fields' must be a mapping of field definitions.") + continue + + container_fields: dict[str, ExtractionRule] = {} + for _field, rule in raw_fields.items(): + if not isinstance(_field, str): + LOG.error(f"[{path.name}] Container field names must be strings, got {_field!r}.") + continue + + if not isinstance(rule, Mapping): + LOG.error(f"[{path.name}] Container definition for '{_field}' must be an object.") + continue + + extraction_rule: ExtractionRule | None = _build_extraction_rule(_field, rule, source=path) + if extraction_rule: + container_fields[_field] = extraction_rule + + if "link" not in container_fields: + LOG.error(f"[{path.name}] Container definition is missing required 'link' field.") + continue + + selector_value: str | None = items_block.get("selector") or items_block.get("expression") + if not isinstance(selector_value, str) or not selector_value: + LOG.error(f"[{path.name}] 'items.selector' must be a non-empty string.") + continue + + selector_type = str(items_block.get("type", "css")) + if selector_type not in ("css", "xpath", "jsonpath"): + LOG.error(f"[{path.name}] Unsupported container selector type '{selector_type}'.") + continue + + container_definition = ContainerDefinition( + selector_type=selector_type, + selector=selector_value, + fields=container_fields, + ) + + for _field, rule in parse_raw.items(): + if "items" == _field: + continue + + if not isinstance(_field, str): + LOG.error(f"[{path.name}] Parser field names must be strings, got {_field!r}.") + continue + + if not isinstance(rule, Mapping): + LOG.error(f"[{path.name}] Parser definition for '{_field}' must be an object.") + continue + + extraction_rule = _build_extraction_rule(_field, rule, source=path) + if extraction_rule: + parsers[_field] = extraction_rule + + if container_definition is None and "link" not in parsers: + LOG.error(f"[{path.name}] Missing required 'link' parser definition.") + continue + + definition = TaskDefinition( + name=name.strip(), + source=path, + matchers=matchers, + engine=engine, + request=request, + parsers=parsers, + container=container_definition, + response=response_config, + ) + + definitions.append(definition) + + return definitions + + +class GenericTaskHandler(BaseHandler): + """Handler that scrapes arbitrary web pages based on JSON task definitions.""" + + _definitions: list[TaskDefinition] = [] + """Cached loaded task definitions.""" + + _sources_mtime: dict[Path, float] = {} + """Modification times of source files to detect changes.""" + + @classmethod + def _tasks_dir(cls) -> Path: + """ + Get the path to the tasks directory. + + Returns: + (Path): Path to the tasks directory. + + """ + return Path(Config.get_instance().config_path) / "tasks" + + @classmethod + def _refresh_definitions(cls, force: bool = False) -> None: + """ + Refresh the cached task definitions if source files have changed. + + Args: + force (bool): If True, force reload even if no changes detected. + + """ + tasks_dir: Path = cls._tasks_dir() + + if not tasks_dir.exists(): + if cls._definitions or cls._sources_mtime: + cls._definitions = [] + cls._sources_mtime = {} + return + + current: dict[Path, float] = {} + for path in tasks_dir.glob("*.json"): + try: + current[path] = path.stat().st_mtime + except OSError: + LOG.warning(f"Unable to stat task definition '{path}'.") + + if force or not cls._definitions or current != cls._sources_mtime: + cls._definitions = load_task_definitions() + cls._sources_mtime = current + + @classmethod + def _find_definition(cls, url: str) -> TaskDefinition | None: + """ + Find a task definition that matches the given URL. + + Args: + url (str): The URL to match. + + Returns: + (TaskDefinition|None): A matching TaskDefinition if found, None otherwise. + + """ + cls._refresh_definitions() + + for definition in cls._definitions: + try: + if definition.matches(url): + return definition + except Exception as exc: + LOG.error(f"Error while matching definition '{definition.name}': {exc}") + + return None + + @staticmethod + def can_handle(task: Task) -> bool: + """ + Determine if this handler can process the given task. + + Args: + task (Task): The task to check. + + Returns: + (bool): True if the handler can process the task, False otherwise. + + """ + definition: TaskDefinition | None = GenericTaskHandler._find_definition(task.url) + if definition: + LOG.debug(f"'{task.name}': Matched generic task definition '{definition.name}'.") + return True + + return False + + @staticmethod + async def extract(task: Task, config: Config | None = None) -> TaskResult | TaskFailure: # noqa: ARG004 + definition: TaskDefinition | None = GenericTaskHandler._find_definition(task.url) + if not definition: + return TaskFailure(message="No generic task definition matched the provided URL.") + + ytdlp_opts: dict[str, Any] = task.get_ytdlp_opts().get_all() + target_url: str = definition.request.url or task.url + + LOG.debug(f"{task.name!r}: Fetching '{target_url}' using engine '{definition.engine.type}'.") + + try: + body_text, json_data = await GenericTaskHandler._fetch_content( + url=target_url, definition=definition, ytdlp_opts=ytdlp_opts + ) + except Exception as exc: + LOG.exception(exc) + return TaskFailure(message="Failed to fetch target URL.", error=str(exc)) + + if "json" == definition.response.format and json_data is None: + return TaskFailure(message="Expected JSON response but decoding failed.") + + if "json" != definition.response.format and not body_text: + return TaskFailure(message="Received empty response body.") + + raw_items: list[dict[str, str]] = GenericTaskHandler._parse_items( + definition=definition, html=body_text or "", base_url=target_url, json_data=json_data + ) + + task_items: list[TaskItem] = [] + + for entry in raw_items: + if not isinstance(entry, dict): + continue + + if not (url := entry.get("link") or entry.get("url")): + continue + + idDict: str | None = get_archive_id(url=url) + archive_id: str | None = idDict.get("archive_id") + if not archive_id: + LOG.warning( + f"[{definition.name}]: '{task.name}': Could not compute archive ID for video '{url}' in feed. generating one." + ) + archive_id = f"generic {hashlib.sha256(url.encode()).hexdigest()[:16]}" + + metadata: dict[str, str] = { + k: v for k, v in entry.items() if k not in {"link", "url", "title", "published", "archive_id"} + } + + task_items.append( + TaskItem( + url=url, + title=entry.get("title"), + archive_id=archive_id, + metadata={"published": entry.get("published"), **metadata}, + ) + ) + + return TaskResult( + items=task_items, + metadata={ + "definition": definition.name, + "response_format": definition.response.format, + }, + ) + + @staticmethod + async def _fetch_content( + url: str, + definition: TaskDefinition, + ytdlp_opts: dict[str, Any], + ) -> tuple[str | None, Any | None]: + """ + Fetch the content of the given URL using the specified engine. + + Args: + url (str): The URL to fetch. + definition (TaskDefinition): The task definition specifying the engine and request details. + ytdlp_opts (dict[str, Any]): yt-dlp options that may influence fetching + + Returns: + (str|None): The fetched HTML content if successful, None otherwise. + + """ + if "selenium" == definition.engine.type: + return await GenericTaskHandler._fetch_with_selenium(url=url, definition=definition) + + return await GenericTaskHandler._fetch_with_httpx(url=url, definition=definition, ytdlp_opts=ytdlp_opts) + + @staticmethod + async def _fetch_with_httpx( + url: str, + definition: TaskDefinition, + ytdlp_opts: dict[str, Any], + ) -> tuple[str | None, Any | None]: + """ + Fetch the content using httpx. + + Args: + url (str): The URL to fetch. + definition (TaskDefinition): The task definition specifying the request details. + ytdlp_opts (dict[str, Any]): yt-dlp options that may influence fetching + + Returns: + (str|None): The fetched HTML content if successful, None otherwise. + + """ + headers: dict[str, str] = {**definition.request.headers} + client_options: dict[str, Any] = { + "headers": { + "User-Agent": random_user_agent(), + } + } + + try: + from httpx_curl_cffi import AsyncCurlTransport, CurlOpt + + client_options["transport"] = AsyncCurlTransport( + impersonate="chrome", + default_headers=True, + curl_options={CurlOpt.FRESH_CONNECT: True}, + ) + client_options["headers"].pop("User-Agent", None) + except Exception: + pass + + if headers: + client_options["headers"].update(headers) + + if proxy := ytdlp_opts.get("proxy"): + client_options["proxy"] = proxy + + timeout_value: float | Any = definition.request.timeout or ytdlp_opts.get("socket_timeout", 120) + + async with httpx.AsyncClient(**client_options) as client: + response: httpx.Response = await client.request( + method=definition.request.normalized_method(), + url=url, + params=definition.request.params or None, + data=definition.request.data, + json=definition.request.json, + timeout=timeout_value, + ) + response.raise_for_status() + + if "json" == definition.response.format: + try: + json_data: dict[str, Any] = response.json() + except Exception as exc: + LOG.error(f"Failed to decode JSON response from '{url}': {exc}") + return response.text, None + + return response.text, json_data + + return response.text, None + + @staticmethod + async def _fetch_with_selenium( + url: str, + definition: TaskDefinition, + ) -> tuple[str | None, Any | None]: + """ + Fetch the content using a Selenium WebDriver. + + Args: + url (str): The URL to fetch. + definition (TaskDefinition): The task definition specifying the engine options. + + Returns: + (str|None): The fetched HTML content if successful, None otherwise. + + """ + try: + from selenium import webdriver + from selenium.webdriver.chrome.options import Options as ChromeOptions + from selenium.webdriver.common.by import By + from selenium.webdriver.support import expected_conditions as EC + from selenium.webdriver.support.ui import WebDriverWait + except ImportError as exc: + LOG.error(f"Selenium engine requested but selenium is not installed: {exc!s}.") + return None + + options_map: dict[str, Any] = definition.engine.options + command_executor: str | None = options_map.get("url", "http://localhost:4444/wd/hub") + browser: str = str(options_map.get("browser", "chrome")).lower() + + if "chrome" != browser: + LOG.error(f"Unsupported selenium browser '{browser}'. Only 'chrome' is supported.") + return None + + arguments: list[str] | str = options_map.get("arguments", ["--headless", "--disable-gpu"]) + if isinstance(arguments, str): + arguments = [arguments] + + wait_for: Mapping | None = ( + options_map.get("wait_for") if isinstance(options_map.get("wait_for"), Mapping) else None + ) + wait_timeout = float(options_map.get("wait_timeout", 15)) + page_load_timeout = float(options_map.get("page_load_timeout", 60)) + + def load_page() -> str | None: + chrome_options = ChromeOptions() + for arg in arguments: + chrome_options.add_argument(str(arg)) + + driver = webdriver.Remote(command_executor=command_executor, options=chrome_options) + try: + driver.set_page_load_timeout(page_load_timeout) + driver.get(url) + + if wait_for: + wait_type: str = str(wait_for.get("type", "css")).lower() + expression: str | None = wait_for.get("expression") or wait_for.get("value") + if isinstance(expression, str) and expression: + locator = None + locator = (By.XPATH, expression) if "xpath" == wait_type else (By.CSS_SELECTOR, expression) + WebDriverWait(driver, wait_timeout).until(EC.presence_of_element_located(locator)) + + return driver.page_source + finally: + driver.quit() + + html: str | None = await asyncio.to_thread(load_page) + return html, None + + @staticmethod + def _parse_items( + definition: TaskDefinition, + html: str, + base_url: str, + json_data: Any | None = None, + ) -> list[dict[str, str]]: + """ + Parse the HTML content and extract items based on the definition. + + Args: + definition (TaskDefinition): The task definition specifying the parsers. + html (str): The HTML content to parse. + base_url (str): The base URL to resolve relative links. + json_data (Any|None): The JSON data to parse if applicable. + + Returns: + (list[dict[str, str]]): A list of extracted items as dictionaries. + + """ + if "json" == definition.response.format: + return GenericTaskHandler._parse_json_items(definition, json_data, base_url) + + selector = Selector(text=html) + + if definition.container: + return GenericTaskHandler._parse_with_container( + definition=definition, + selector=selector, + html=html, + base_url=base_url, + ) + + extracted: dict[str, list[str]] = {} + + for _field, rule in definition.parsers.items(): + values: list[str] = GenericTaskHandler._execute_rule(field=_field, selector=selector, html=html, rule=rule) + extracted[_field] = values + + link_values: list[str] = extracted.get("link", []) + if not link_values: + LOG.debug(f"Definition '{definition.name}' produced no link values.") + return [] + + total_items: int = len(link_values) + items: list[dict[str, str]] = [] + + for index in range(total_items): + entry: dict[str, str] = {} + link_value: str = link_values[index] + if not link_value: + continue + + entry["link"] = urljoin(base_url, link_value) + + for _field, values in extracted.items(): + if "link" == _field: + continue + + value: str | None = values[index] if index < len(values) else None + if value is None: + continue + + entry[_field] = value + + items.append(entry) + + return items + + @staticmethod + def _parse_json_items( + definition: TaskDefinition, + json_data: Any | None, + base_url: str, + ) -> list[dict[str, str]]: + if json_data is None: + LOG.debug(f"Definition '{definition.name}' expects JSON but no data was parsed.") + return [] + + if definition.container: + return GenericTaskHandler._parse_json_with_container(definition, json_data, base_url) + + items: list[dict[str, str]] = [] + entry: dict[str, str] = {} + + for _field, rule in definition.parsers.items(): + values: list[str] = GenericTaskHandler._execute_json_rule(_field, json_data, rule) + if values: + if "link" == _field: + entry["link"] = urljoin(base_url, values[0]) + else: + entry[_field] = values[0] + + if "link" in entry: + items.append(entry) + + return items + + @staticmethod + def _parse_with_container( + definition: TaskDefinition, + selector: Selector, + html: str, + base_url: str, + ) -> list[dict[str, str]]: + container: ContainerDefinition | None = definition.container + if not container: + return [] + + if "jsonpath" == container.selector_type: + LOG.error( + f"Container selector type 'jsonpath' requires response type 'json'. Definition '{definition.name}'." + ) + return [] + + selection: SelectorList[Selector] = ( + selector.css(container.selector) if "css" == container.selector_type else selector.xpath(container.selector) + ) + + items: list[dict[str, str]] = [] + + for node in selection: + node_html: Any | str = node.get() or html + entry: dict[str, str] = {} + + for _field, rule in container.fields.items(): + values: list[str] = GenericTaskHandler._execute_rule( + field=_field, + selector=node, + html=node_html, + rule=rule, + ) + + value: str | None = values[0] if values else None + if value is None: + continue + + if "link" == _field: + entry["link"] = urljoin(base_url, value) + else: + entry[_field] = value + + if "link" not in entry: + continue + + items.append(entry) + + return items + + @staticmethod + def _parse_json_with_container( + definition: TaskDefinition, + json_data: Any, + base_url: str, + ) -> list[dict[str, str]]: + container: ContainerDefinition | None = definition.container + if not container: + return [] + + if "jsonpath" != container.selector_type: + LOG.error(f"JSON response requires container selector type 'jsonpath'. Definition '{definition.name}'.") + return [] + + nodes: Any = GenericTaskHandler._json_search(json_data, container.selector) + if nodes is None: + return [] + + if not isinstance(nodes, list): + nodes = [nodes] + + items: list[dict[str, str]] = [] + + for node in nodes: + entry: dict[str, str] = {} + + for _field, rule in container.fields.items(): + values: list[str] = GenericTaskHandler._execute_json_rule(_field, node, rule) + if not values: + continue + + if "link" == _field: + entry["link"] = urljoin(base_url, values[0]) + else: + entry[_field] = values[0] + + if "link" not in entry: + continue + + items.append(entry) + + return items + + @staticmethod + def _execute_json_rule(field: str, data: Any, rule: ExtractionRule) -> list[str]: + values: list[str] = [] + + if "jsonpath" == rule.type: + result: Any = GenericTaskHandler._json_search(data, rule.expression) + candidates: list | list[Any] = result if isinstance(result, list) else [result] + for candidate in candidates: + if candidate is None: + continue + + text: str = GenericTaskHandler._coerce_to_string(candidate) + processed: str | None = GenericTaskHandler._apply_post_filter(text, rule) + if processed is not None: + values.append(processed) + + return values + + if "regex" == rule.type: + target: str = GenericTaskHandler._coerce_to_string(data) + try: + pattern: re.Pattern[str] = re.compile(rule.expression, re.MULTILINE | re.DOTALL) + except re.error as exc: + LOG.error(f"Invalid regex expression '{rule.expression}': {exc}") + return values + + for match in pattern.finditer(target): + raw: str | None = GenericTaskHandler._regex_value(match=match, attribute=rule.attribute) + processed = GenericTaskHandler._apply_post_filter(raw, rule) + if processed is not None: + values.append(processed) + + return values + + LOG.error(f"Unsupported extraction type '{rule.type}' for JSON data in field '{field}'.") + return values + + @staticmethod + def _json_search(data: Any, expression: str) -> Any: + try: + return jmespath.search(expression, data) + except Exception as exc: + LOG.error(f"JSONPath search failed for expression '{expression}': {exc}") + return None + + @staticmethod + def _coerce_to_string(value: Any) -> str: + if isinstance(value, str): + return value + if isinstance(value, (int, float, bool)) or value is None: + return "" if value is None else str(value) + try: + return json.dumps(value, ensure_ascii=False) + except Exception: + return str(value) + + @staticmethod + def _execute_rule(field: str, selector: Selector, html: str, rule: ExtractionRule) -> list[str]: + """ + Execute a single extraction rule and return the list of extracted values. + + Args: + field (str): The name of the field being extracted. + selector (Selector): The parsel Selector for the HTML content. + html (str): The raw HTML content. + rule (ExtractionRule): The extraction rule to execute. + + Returns: + (list[str]): A list of extracted values. + + """ + values: list[str] = [] + + if "regex" == rule.type: + try: + pattern: re.Pattern[str] = re.compile(rule.expression, re.MULTILINE | re.DOTALL) + except re.error as exc: + LOG.error(f"Invalid regex expression '{rule.expression}': {exc}") + return values + + for match in pattern.finditer(html): + raw: str | None = GenericTaskHandler._regex_value(match=match, attribute=rule.attribute) + processed: str | None = GenericTaskHandler._apply_post_filter(raw, rule) + if processed is not None: + values.append(processed) + + return values + + if "jsonpath" == rule.type: + LOG.error("Extraction type 'jsonpath' is only valid for JSON responses.") + return values + + selection: SelectorList[Selector] = ( + selector.css(rule.expression) if "css" == rule.type else selector.xpath(rule.expression) + ) + + for sel in selection: + raw = GenericTaskHandler._selector_value(field, sel, rule.attribute) + processed = GenericTaskHandler._apply_post_filter(raw, rule) + if processed is not None: + values.append(processed) + + return values + + @staticmethod + def _regex_value(match: re.Match[str], attribute: str | None) -> str | None: + """ + Extract a value from a regex match based on the attribute. + + Args: + match (re.Match[str]): The regex match object. + attribute (str|None): Optional group name or index to extract. + + Returns: + (str|None): The extracted value if found, None otherwise. + + """ + if attribute: + try: + return match.group(attribute) + except (IndexError, KeyError): + LOG.debug(f"Regex group '{attribute}' not found in pattern '{match.re.pattern}'.") + return None + + if match.groupdict(): + return next((value for value in match.groupdict().values() if value), None) + + if match.groups(): + return match.group(1) + + return match.group(0) + + @staticmethod + def _selector_value(field: str, sel: Selector, attribute: str | None) -> str | None: + """ + Extract a value from a parsel Selector based on the attribute. + + Args: + field (str): The name of the field being extracted. + sel (Selector): The parsel Selector object. + attribute (str|None): Optional attribute to extract (e.g. 'href', 'src', 'text', etc.). + + Returns: + (str|None): The extracted value if found, None otherwise. + + """ + attr: str | None = attribute.lower() if isinstance(attribute, str) else None + + if attr in {"text", "inner_text"}: + return sel.xpath("normalize-space()").get() + + if attr in {"html", "outer_html"}: + value: Any = sel.get() + return value if value is not None else None + + if attr and attr not in {"html", "outer_html", "text", "inner_text"}: + try: + attributes: dict[str, str] = sel.attrib + except AttributeError: + attributes = None + + if attributes and attr in attributes: + return attributes.get(attr) + + attr_value: str | None = sel.xpath(f"@{attr}").get() + if attr_value is not None: + return attr_value + + if attr is None and "link" == field.lower(): + href = None + try: + attributes = sel.attrib + except AttributeError: + attributes = None + + if attributes and "href" in attributes: + href: str | None = attributes.get("href") + if not href: + href: str | None = sel.xpath("@href").get() + if href: + return href + + if attr is None: + text_value: str | None = sel.xpath("normalize-space()").get() + if text_value: + return text_value + + value = sel.get() + return value if value is not None else None + + @staticmethod + def _apply_post_filter(value: str | None, rule: ExtractionRule) -> str | None: + """ + Apply the post-filter to the extracted value if defined. + + Args: + value (str|None): The extracted value to filter. + rule (ExtractionRule): The extraction rule containing the post-filter. + + Returns: + (str|None): The filtered value if applicable, None otherwise. + + """ + if value is None: + return None + + cleaned: str = value.strip() + if rule.post_filter: + return rule.post_filter.apply(cleaned) + + return cleaned or None diff --git a/app/library/task_handlers/task_definition.schema.json b/app/library/task_handlers/task_definition.schema.json new file mode 100644 index 00000000..711b06d3 --- /dev/null +++ b/app/library/task_handlers/task_definition.schema.json @@ -0,0 +1,456 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "$id": "https://ytptube.app/schemas/task-definition.json", + "title": "YTPTube Generic Task Definition", + "type": "object", + "required": [ + "name", + "match", + "parse" + ], + "additionalProperties": false, + "properties": { + "name": { + "type": "string", + "minLength": 1, + "description": "Human-readable identifier for this definition." + }, + "match": { + "type": "array", + "minItems": 1, + "items": { + "oneOf": [ + { + "type": "string", + "minLength": 1, + "description": "Glob pattern matched against task URLs." + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "regex": { + "type": "string", + "minLength": 1, + "description": "Regular expression applied to task URLs." + }, + "glob": { + "type": "string", + "minLength": 1, + "description": "Glob pattern applied to task URLs." + } + }, + "anyOf": [ + { + "required": [ + "regex" + ] + }, + { + "required": [ + "glob" + ] + } + ] + } + ] + }, + "description": "Patterns that determine which tasks use this definition." + }, + "engine": { + "type": "object", + "additionalProperties": false, + "properties": { + "type": { + "type": "string", + "enum": [ + "httpx", + "selenium" + ], + "default": "httpx", + "description": "Fetch engine to use (HTTPX or Selenium)." + }, + "options": { + "type": "object", + "description": "Engine-specific configuration.", + "additionalProperties": true, + "properties": { + "url": { + "type": "string", + "format": "uri", + "description": "Remote Selenium hub URL." + }, + "browser": { + "type": "string", + "enum": [ + "chrome" + ], + "description": "Selenium browser name (currently only chrome)." + }, + "arguments": { + "oneOf": [ + { + "type": "array", + "items": { + "type": "string" + } + }, + { + "type": "string" + } + ], + "description": "Browser launch arguments for Selenium." + }, + "wait_for": { + "$ref": "#/definitions/waitFor" + }, + "wait_timeout": { + "type": "number", + "minimum": 0, + "description": "Seconds to wait for the wait_for selector." + }, + "page_load_timeout": { + "type": "number", + "minimum": 0, + "description": "Seconds to allow page load to complete." + } + } + } + }, + "description": "Optional engine configuration (defaults to HTTPX)." + }, + "request": { + "type": "object", + "additionalProperties": false, + "properties": { + "method": { + "type": "string", + "enum": [ + "GET", + "POST" + ], + "default": "GET", + "description": "HTTP method to use when fetching the page." + }, + "url": { + "type": "string", + "format": "uri", + "description": "Override URL to fetch instead of the task URL." + }, + "headers": { + "$ref": "#/definitions/stringMap", + "description": "Additional request headers." + }, + "params": { + "$ref": "#/definitions/stringMap", + "description": "Query string parameters." + }, + "data": { + "description": "Form payload for POST requests.", + "oneOf": [ + { + "$ref": "#/definitions/stringMap" + }, + { + "type": "string" + }, + { + "type": "null" + } + ] + }, + "json": { + "description": "JSON payload for POST requests.", + "type": [ + "object", + "array", + "string", + "number", + "boolean", + "null" + ] + }, + "timeout": { + "type": "number", + "minimum": 0, + "description": "Timeout in seconds for the request." + } + }, + "description": "Optional HTTP request overrides." + }, + "response": { + "type": "object", + "additionalProperties": false, + "properties": { + "type": { + "type": "string", + "enum": [ + "html", + "json" + ], + "default": "html", + "description": "Body format returned by the target URL." + } + } + }, + "parse": { + "type": "object", + "minProperties": 1, + "description": "Field extraction rules and optional container definition.", + "additionalProperties": false, + "properties": { + "items": { + "$ref": "#/definitions/container" + } + }, + "patternProperties": { + "^(?!items$).+$": { + "$ref": "#/definitions/extractionRule" + } + }, + "allOf": [ + { + "if": { + "not": { + "required": [ + "items" + ] + } + }, + "then": { + "required": [ + "link" + ] + } + } + ] + } + }, + "definitions": { + "stringMap": { + "type": "object", + "additionalProperties": { + "type": [ + "string", + "number", + "boolean", + "null" + ] + } + }, + "waitFor": { + "type": "object", + "additionalProperties": false, + "properties": { + "type": { + "type": "string", + "enum": [ + "css", + "xpath" + ], + "description": "Selector type to wait for." + }, + "expression": { + "type": "string", + "minLength": 1, + "description": "Selector expression." + }, + "value": { + "type": "string", + "minLength": 1, + "description": "Alias for expression for backwards compatibility." + } + }, + "anyOf": [ + { + "required": [ + "expression" + ] + }, + { + "required": [ + "value" + ] + } + ] + }, + "postFilter": { + "type": "object", + "additionalProperties": false, + "required": [ + "filter" + ], + "properties": { + "filter": { + "type": "string", + "minLength": 1, + "description": "Regular expression applied after extraction." + }, + "value": { + "type": "string", + "minLength": 1, + "description": "Named or numbered capture group to return." + } + } + }, + "containerFields": { + "type": "object", + "minProperties": 1, + "additionalProperties": false, + "patternProperties": { + "^.+$": { + "$ref": "#/definitions/extractionRule" + } + }, + "required": [ + "link" + ] + }, + "container": { + "type": "object", + "additionalProperties": false, + "properties": { + "type": { + "type": "string", + "enum": [ + "css", + "xpath", + "jsonpath" + ], + "default": "css", + "description": "Selector type to use for locating repeatable elements." + }, + "selector": { + "type": "string", + "minLength": 1, + "description": "CSS/XPath selector identifying each item container." + }, + "expression": { + "type": "string", + "minLength": 1, + "description": "Alias for selector." + }, + "fields": { + "$ref": "#/definitions/containerFields" + } + }, + "required": [ + "fields" + ], + "anyOf": [ + { + "required": [ + "selector" + ] + }, + { + "required": [ + "expression" + ] + } + ] + }, + "extractionRule": { + "type": "object", + "additionalProperties": false, + "required": [ + "type", + "expression" + ], + "properties": { + "type": { + "type": "string", + "enum": [ + "css", + "xpath", + "regex", + "jsonpath" + ], + "description": "Extraction strategy." + }, + "expression": { + "type": "string", + "minLength": 1, + "description": "Selector or pattern used to extract values." + }, + "attribute": { + "type": "string", + "minLength": 1, + "description": "Optional attribute or group to return." + }, + "post_filter": { + "$ref": "#/definitions/postFilter" + } + } + } + }, + "examples": [ + { + "name": "example", + "match": [ + "https://example.com/articles/*", + { + "regex": "https://example.com/post/[0-9]+" + } + ], + "engine": { + "type": "httpx" + }, + "request": { + "method": "GET", + "headers": { + "User-Agent": "MyCustomAgent/1.0" + } + }, + "parse": { + "link": { + "type": "css", + "expression": ".article a.link", + "attribute": "href" + }, + "title": { + "type": "css", + "expression": ".article .title", + "attribute": "text" + }, + "id": { + "type": "regex", + "expression": "id=(?P[0-9]+)", + "post_filter": { + "filter": "(?P[0-9]+)", + "value": "id" + } + } + } + }, + { + "name": "container-example", + "match": [ + "https://example.com/list" + ], + "parse": { + "items": { + "type": "css", + "selector": ".cards .card", + "fields": { + "link": { + "type": "css", + "expression": ".card-header a", + "attribute": "href" + }, + "title": { + "type": "css", + "expression": ".card-header a", + "attribute": "text" + }, + "poet": { + "type": "css", + "expression": "footer .card-footer-item:first-child a", + "attribute": "text" + } + } + } + } + } + ] +} diff --git a/app/library/task_handlers/twitch.py b/app/library/task_handlers/twitch.py index be48f4ed..1426aaac 100644 --- a/app/library/task_handlers/twitch.py +++ b/app/library/task_handlers/twitch.py @@ -3,19 +3,14 @@ import re from typing import TYPE_CHECKING from xml.etree.ElementTree import Element -from app.library.DownloadQueue import DownloadQueue -from app.library.Events import EventBus, Events -from app.library.ItemDTO import Item -from app.library.Tasks import Task -from app.library.Utils import archive_read, get_archive_id +from app.library.Tasks import Task, TaskFailure, TaskItem, TaskResult +from app.library.Utils import get_archive_id from ._base_handler import BaseHandler if TYPE_CHECKING: from xml.etree.ElementTree import Element - from app.library.Download import Download - LOG: logging.Logger = logging.getLogger(__name__) @@ -30,40 +25,23 @@ class TwitchHandler(BaseHandler): return TwitchHandler.parse(task.url) is not None @staticmethod - async def handle(task: Task, notify: EventBus, queue: DownloadQueue): - """ - Fetch the RSS feed for a Twitch channel VODs, parse entries, - and enqueue new items that are not in the archive/queue already. - - Args: - task (Task): The task containing the Twitch channel URL. - notify (EventBus): The event bus for notifications. - queue (DownloadQueue): The download queue instance. - - """ + async def _collect_feed( + task: Task, + params: dict, + handle_name: str, + ) -> tuple[str, list[dict[str, str]], bool]: from defusedxml.ElementTree import fromstring - handleName: str | None = TwitchHandler.parse(task.url) - if not handleName: - LOG.error(f"Cannot parse '{task.name}' URL: {task.url}") - return - - params: dict = task.get_ytdlp_opts().get_all() - archive_file: str | None = params.get("download_archive") - if not archive_file: - LOG.error(f"Task '{task.name}' does not have an archive file.") - return - - feed_url: str = TwitchHandler.FEED.format(handle=handleName) + feed_url: str = TwitchHandler.FEED.format(handle=handle_name) LOG.debug(f"Fetching '{task.name}' feed.") response = await TwitchHandler.request(url=feed_url, ytdlp_opts=params) response.raise_for_status() - items: list = [] + root: Element[str] = fromstring(response.text) + items: list[dict[str, str]] = [] has_items = False - root: Element[str] = fromstring(response.text) for entry in root.findall("channel/item"): link_elem: Element[str] | None = entry.find("link") url: str = link_elem.text.strip() if link_elem is not None and link_elem.text else "" @@ -71,12 +49,14 @@ class TwitchHandler(BaseHandler): LOG.warning(f"Entry in '{task.name}' feed is missing URL. Skipping entry.") continue - m: re.Match[str] | None = re.search(r"^https?://(?:www\.)?twitch\.tv/videos/(?P\d+)(?:[/?].*)?$", url) - if not m: + match: re.Match[str] | None = re.search( + r"^https?://(?:www\.)?twitch\.tv/videos/(?P\d+)(?:[/?].*)?$", url + ) + if not match: LOG.warning(f"URL in '{task.name}' feed does not look like a VOD link: {url}") continue - vid: str = m.group("id") + vid: str = match.group("id") title_elem: Element[str] | None = entry.find("title") title: str = title_elem.text.strip() if title_elem is not None and title_elem.text else "" @@ -89,68 +69,34 @@ class TwitchHandler(BaseHandler): LOG.warning(f"Could not compute archive ID for video '{vid}' in '{task.name}' feed. Skipping entry.") continue - if archive_id in TwitchHandler.queued: - continue - items.append({"id": vid, "url": url, "title": title, "archive_id": archive_id}) - if len(items) < 1: - if not has_items: - LOG.warning(f"No entries found in '{task.name}' feed. URL: {feed_url}") - else: - LOG.debug(f"No new items found in '{task.name}' feed.") - return + return feed_url, items, has_items - filtered: list = [] + @staticmethod + async def extract(task: Task) -> TaskResult | TaskFailure: + handle_name: str | None = TwitchHandler.parse(task.url) + if not handle_name: + return TaskFailure(message="Unrecognized Twitch channel URL.") - downloaded: list[str] = archive_read(archive_file, [item["archive_id"] for item in items]) - - for item in items: - TwitchHandler.queued.add(item["archive_id"]) - - if item["archive_id"] in downloaded: - continue - - if queue.queue.exists(url=item["url"]): - continue - - try: - done: Download = queue.done.get(url=item["url"]) - if "error" != done.info.status: - continue - except KeyError: - pass - - if item["archive_id"] not in TwitchHandler.failure_count: - TwitchHandler.failure_count[item["archive_id"]] = 0 - - filtered.append(item) - - if len(filtered) < 1: - LOG.debug(f"No new items found in '{task.name}' feed.") - return - - LOG.info(f"Found '{len(filtered)}' new items from '{task.name}' feed.") - - rItem: Item = Item.format( - { - "url": feed_url, - "preset": task.preset, - "folder": task.folder if task.folder else "", - "template": task.template if task.template else "", - "cli": task.cli if task.cli else "", - "auto_start": task.auto_start, - "extras": {"source_task": task.id}, - } - ) + params: dict = task.get_ytdlp_opts().get_all() try: - for item in filtered: - notify.emit(Events.ADD_URL, data=rItem.new_with(url=item["url"]).serialize()) - except Exception as e: - LOG.exception(e) - LOG.error(f"Error while adding items from '{task.name}'. {e!s}") - return + feed_url, items, has_items = await TwitchHandler._collect_feed(task, params, handle_name) + except Exception as exc: + LOG.exception(exc) + return TaskFailure(message="Failed to fetch Twitch feed.", error=str(exc)) + + task_items: list[TaskItem] = [] + + for entry in items: + if not (url := entry.get("url")): + continue + + archive_id: str = entry.get("archive_id") + task_items.append(TaskItem(url=url, title=entry.get("title"), archive_id=archive_id)) + + return TaskResult( items=task_items, metadata={ "feed_url": feed_url, "has_entries": has_items } ) @staticmethod def parse(url: str) -> str | None: diff --git a/app/library/task_handlers/youtube.py b/app/library/task_handlers/youtube.py index 1ac3a952..63cba520 100644 --- a/app/library/task_handlers/youtube.py +++ b/app/library/task_handlers/youtube.py @@ -1,21 +1,16 @@ import logging import re -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from xml.etree.ElementTree import Element -from app.library.DownloadQueue import DownloadQueue -from app.library.Events import EventBus, Events -from app.library.ItemDTO import Item -from app.library.Tasks import Task -from app.library.Utils import archive_read, get_archive_id +from app.library.Tasks import Task, TaskFailure, TaskItem, TaskResult +from app.library.Utils import get_archive_id from ._base_handler import BaseHandler if TYPE_CHECKING: from xml.etree.ElementTree import Element - from app.library.Download import Download - LOG: logging.Logger = logging.getLogger(__name__) @@ -32,39 +27,28 @@ class YoutubeHandler(BaseHandler): @staticmethod def can_handle(task: Task) -> bool: - if not task.get_ytdlp_opts().get_all().get("download_archive"): - LOG.debug(f"'{task.name}': Task does not have an archive file configured.") - return False - LOG.debug(f"'{task.name}': Checking if task URL is parsable YouTube URL: {task.url}") return YoutubeHandler.parse(task.url) is not None @staticmethod - async def handle(task: Task, notify: EventBus, queue: DownloadQueue): + async def _get(task: Task, params: dict, parsed: dict[str, str]) -> tuple[str, list[dict[str, str]], int]: """ - Fetch the Atom feed for a YouTube channel or playlist, parse entries, - and return a list of videos with metadata. + Fetch the feed and return raw entries. Args: task (Task): The task containing the YouTube URL. - notify (EventBus): The event bus for notifications. - queue (DownloadQueue): The download queue instance. + params (dict): The ytdlp options. + parsed (dict): The parsed URL components. + + Returns: + tuple[str, list[dict[str, str]], int]: The feed URL, list of """ from defusedxml.ElementTree import fromstring - parsed: dict[str, str] | None = YoutubeHandler.parse(task.url) - if not parsed: - LOG.error(f"'{task.name}': Cannot parse task URL: {task.url}") - return - - params: dict = task.get_ytdlp_opts().get_all() - feed_url: str = YoutubeHandler.FEED.format(type=parsed["type"], id=parsed["id"]) LOG.debug(f"'{task.name}': Fetching feed.") - items: list = [] - response = await YoutubeHandler.request(url=feed_url, ytdlp_opts=params) response.raise_for_status() @@ -74,10 +58,12 @@ class YoutubeHandler(BaseHandler): "yt": "http://www.youtube.com/xml/schemas/2015", } - real_count: int = 0 + items: list[dict[str, str]] = [] + real_count = 0 + for entry in root.findall("atom:entry", ns): vid_elem: Element[str] | None = entry.find("yt:videoId", ns) - vid: str | None = vid_elem.text if vid_elem is not None else "" + vid: str = vid_elem.text if vid_elem is not None and vid_elem.text else "" if not vid: LOG.warning(f"'{task.name}': Entry in the feed is missing a video ID. Skipping.") continue @@ -91,73 +77,43 @@ class YoutubeHandler(BaseHandler): continue title_elem: Element[str] | None = entry.find("atom:title", ns) - title: str | None = title_elem.text if title_elem is not None else "" + title: str = title_elem.text if title_elem is not None and title_elem.text else "" pub_elem: Element[str] | None = entry.find("atom:published", ns) - published: str | None = pub_elem.text if pub_elem is not None else "" - real_count += 1 + published: str = pub_elem.text if pub_elem is not None and pub_elem.text else "" - if archive_id in YoutubeHandler.queued: - continue + real_count += 1 items.append({"id": vid, "url": url, "title": title, "published": published, "archive_id": archive_id}) - if len(items) < 1: - if real_count < 1: - LOG.warning(f"'{task.name}': No entries found the RSS feed. URL: {feed_url}") - else: - LOG.debug(f"'{task.name}': Feed has '{real_count}' entries, all already downloaded/queued.") - return + return feed_url, items, real_count - filtered: list = [] + @staticmethod + async def extract(task: Task) -> TaskResult | TaskFailure: + parsed: dict[str, str] | None = YoutubeHandler.parse(task.url) + if not parsed: + return TaskFailure(message="Unrecognized YouTube channel or playlist URL.") - downloaded: list[str] = archive_read(params.get("download_archive"), [item["archive_id"] for item in items]) - - for item in items: - YoutubeHandler.queued.add(item["archive_id"]) - if item["archive_id"] in downloaded: - continue - - if queue.queue.exists(url=item["url"]): - continue - - try: - done: Download = queue.done.get(url=item["url"]) - if "error" != done.info.status: - continue - except KeyError: - pass - - if item["archive_id"] not in YoutubeHandler.failure_count: - YoutubeHandler.failure_count[item["archive_id"]] = 0 - - filtered.append(item) - - if len(filtered) < 1: - LOG.debug(f"'{task.name}': Feed has '{real_count}' entries, all already downloaded/queued.") - return - - LOG.info(f"'{task.name}': Found '{len(filtered)}/{real_count}' new items from feed.") - - rItem: Item = Item.format( - { - "url": feed_url, - "preset": task.preset, - "folder": task.folder if task.folder else "", - "template": task.template if task.template else "", - "cli": task.cli if task.cli else "", - "auto_start": task.auto_start, - "extras": {"source_task": task.id}, - } - ) + params: dict = task.get_ytdlp_opts().get_all() try: - for item in filtered: - notify.emit(Events.ADD_URL, data=rItem.new_with(url=item["url"]).serialize()) - except Exception as e: - LOG.exception(e) - LOG.error(f"'{task.name}': Error while adding items from task feed. {e!s}") - return + feed_url, items, real_count = await YoutubeHandler._get(task, params, parsed) + except Exception as exc: + LOG.exception(exc) + return TaskFailure(message="Failed to fetch YouTube feed.", error=str(exc)) + + task_items: list[TaskItem] = [] + + for entry in items: + if not (url := entry.get("url")): + continue + + archive_id: str = entry.get("archive_id") + metadata: dict[str, Any] = {"published": entry.get("published")} + + task_items.append(TaskItem(url=url, title=entry.get("title"), archive_id=archive_id, metadata=metadata)) + + return TaskResult(items=task_items, metadata={"feed_url": feed_url, "entry_count": real_count}) @staticmethod def parse(url: str) -> dict[str, str] | None: diff --git a/app/routes/api/history.py b/app/routes/api/history.py index 2e0929d3..00ce125f 100644 --- a/app/routes/api/history.py +++ b/app/routes/api/history.py @@ -1,12 +1,10 @@ import asyncio import logging -from pathlib import Path from typing import TYPE_CHECKING from aiohttp import web from aiohttp.web import Request, Response -from app.library.config import Config from app.library.Download import Download from app.library.DownloadQueue import DownloadQueue from app.library.encoder import Encoder @@ -73,7 +71,7 @@ async def item_delete(request: Request, queue: DownloadQueue, encoder: Encoder) @route("GET", "api/history/{id}", "item_view") -async def item_view(request: Request, queue: DownloadQueue, encoder: Encoder, config: Config) -> Response: +async def item_view(request: Request, queue: DownloadQueue, encoder: Encoder) -> Response: """ Update an item in the history. @@ -82,7 +80,6 @@ async def item_view(request: Request, queue: DownloadQueue, encoder: Encoder, co queue (DownloadQueue): The download queue instance. encoder (Encoder): The encoder instance. notify (EventBus): The event bus instance. - config (Config): The configuration instance. Returns: Response: The response object. @@ -104,23 +101,11 @@ async def item_view(request: Request, queue: DownloadQueue, encoder: Encoder, co "ffprobe": {}, } - if item.info.filename: + if "finished" == item.info.status and (filename := item.info.get_file()): + from app.library.ffprobe import ffprobe + try: - from app.library.ffprobe import ffprobe - from app.library.Utils import get_file - - filename = Path(config.download_path) - if item.info.folder: - filename: Path = filename / item.info.folder - - filename: Path = filename / item.info.filename - - if filename.exists(): - realFile, status = get_file( - download_path=config.download_path, file=str(filename.relative_to(config.download_path)) - ) - if status in (web.HTTPOk.status_code, web.HTTPFound.status_code): - info["ffprobe"] = await ffprobe(str(realFile)) + info["ffprobe"] = await ffprobe(filename) except Exception: pass diff --git a/app/routes/api/tasks.py b/app/routes/api/tasks.py index 203fb4d5..2c52ff6d 100644 --- a/app/routes/api/tasks.py +++ b/app/routes/api/tasks.py @@ -6,12 +6,45 @@ from aiohttp.web import Request, Response from app.library.encoder import Encoder from app.library.router import route -from app.library.Tasks import Task, Tasks -from app.library.Utils import init_class, validate_uuid +from app.library.Tasks import Task, TaskFailure, TaskResult, Tasks +from app.library.Utils import init_class, validate_url, validate_uuid LOG: logging.Logger = logging.getLogger(__name__) +@route("POST", "api/tasks/inspect", "task_handler_inspect") +async def task_handler_inspect(request: Request, tasks: Tasks, encoder: Encoder) -> Response: + data = await request.json() + + url: str | None = data.get("url") if isinstance(data, dict) else None + if not url: + return web.json_response({"error": "url is required."}, status=web.HTTPBadRequest.status_code) + try: + validate_url(url) + except ValueError as e: + return web.json_response({"error": str(e)}, status=web.HTTPBadRequest.status_code) + + preset: str = data.get("preset", "") if isinstance(data, dict) else "" + handler_name: str | None = data.get("handler") if isinstance(data, dict) else None + + try: + result: TaskResult | TaskFailure = await tasks.get_handler().inspect( + url=url, preset=preset, handler_name=handler_name + ) + except Exception as e: + LOG.exception(e) + return web.json_response( + {"error": "Failed to inspect handler.", "message": str(e)}, + status=web.HTTPInternalServerError.status_code, + ) + + return web.json_response( + data=result, + status=web.HTTPBadRequest.status_code if isinstance(result, TaskFailure) else web.HTTPOk.status_code, + dumps=encoder.encode, + ) + + @route("GET", "api/tasks/", "tasks") async def tasks(encoder: Encoder) -> Response: """ diff --git a/app/tests/test_ffprobe.py b/app/tests/test_ffprobe.py new file mode 100644 index 00000000..c896a320 --- /dev/null +++ b/app/tests/test_ffprobe.py @@ -0,0 +1,207 @@ +import tempfile +from pathlib import Path +from unittest.mock import AsyncMock, patch + +import pytest + + +class TestFFProbe: + """Test the ffprobe module functionality.""" + + def setup_method(self): + """Set up test files.""" + self.temp_dir = tempfile.mkdtemp() + self.test_file = Path(self.temp_dir) / "test_video.mp4" + self.test_file.touch() + + def teardown_method(self): + """Clean up test files.""" + import shutil + shutil.rmtree(self.temp_dir, ignore_errors=True) + + @pytest.mark.asyncio + async def test_ffprobe_with_existing_file(self): + """Test ffprobe with an existing file.""" + from app.library.ffprobe import FFProbeResult, ffprobe + + # Mock subprocess to avoid actual ffprobe execution + with patch("asyncio.create_subprocess_exec") as mock_subprocess: + # Mock the subprocess result + mock_process = AsyncMock() + mock_process.wait.return_value = 0 + mock_process.communicate.return_value = (b'{"format": {"duration": "10.0"}}', b"") + mock_subprocess.return_value = mock_process + + with patch("anyio.open_file") as mock_open_file: + mock_file = AsyncMock() + mock_open_file.return_value.__aenter__.return_value = mock_file + + result = await ffprobe(str(self.test_file)) + assert isinstance(result, FFProbeResult) + + @pytest.mark.asyncio + async def test_ffprobe_with_nonexistent_file(self): + """Test ffprobe with a non-existent file.""" + from app.library.ffprobe import ffprobe + + nonexistent_file = Path(self.temp_dir) / "does_not_exist.mp4" + + with pytest.raises(OSError, match="No such media file"): + await ffprobe(str(nonexistent_file)) + + @pytest.mark.asyncio + async def test_ffprobe_caching_behavior(self): + """Test that ffprobe results are cached with enhanced async timed_lru_cache.""" + from app.library.ffprobe import ffprobe + + # Test that the function has been decorated with caching + assert hasattr(ffprobe, "cache_clear"), "ffprobe should have cache_clear method from timed_lru_cache" + assert hasattr(ffprobe, "cache_info"), "ffprobe should have cache_info method from timed_lru_cache" + + # Clear cache to start fresh + ffprobe.cache_clear() + + # Mock subprocess to avoid actual ffprobe execution + call_count = 0 + with patch("asyncio.create_subprocess_exec") as mock_subprocess: + def create_mock_process(*_args, **_kwargs): + nonlocal call_count + call_count += 1 + mock_process = AsyncMock() + mock_process.wait.return_value = 0 + mock_process.communicate.return_value = (b'{"format": {"duration": "10.0"}, "streams": []}', b"") + return mock_process + + mock_subprocess.side_effect = create_mock_process + + with patch("anyio.open_file") as mock_open_file: + mock_file = AsyncMock() + mock_open_file.return_value.__aenter__.return_value = mock_file + + # First call should execute the function + result1 = await ffprobe(str(self.test_file)) + assert result1 is not None + assert isinstance(result1.metadata, dict) + first_call_count = call_count + + # Second call with same argument should use cached result + result2 = await ffprobe(str(self.test_file)) + assert result2 is not None + assert isinstance(result2.metadata, dict) + + # The subprocess should not be called again for the actual ffprobe execution + # (it may be called for the -h check, but the main execution should be cached) + assert call_count == first_call_count, "Second call should use cached result" + + # Results should be equivalent (same data, may not be same object due to async nature) + assert result1.metadata == result2.metadata + + @pytest.mark.asyncio + async def test_ffprobe_with_path_object(self): + """Test ffprobe with Path object input.""" + from app.library.ffprobe import ffprobe + + # Mock subprocess to avoid actual ffprobe execution + with patch("asyncio.create_subprocess_exec") as mock_subprocess: + mock_process = AsyncMock() + mock_process.wait.return_value = 0 + mock_process.communicate.return_value = (b'{"format": {"duration": "10.0"}}', b"") + mock_subprocess.return_value = mock_process + + with patch("anyio.open_file") as mock_open_file: + mock_file = AsyncMock() + mock_open_file.return_value.__aenter__.return_value = mock_file + + # Test with Path object + result = await ffprobe(self.test_file) + assert result is not None + + def test_ffprobe_result_properties(self): + """Test FFProbeResult object properties.""" + from app.library.ffprobe import FFProbeResult, FFStream + + result = FFProbeResult() + + # Test empty result + assert result.video == [] + assert result.audio == [] + assert result.subtitle == [] + assert result.attachment == [] + assert result.metadata == {} + + # Test adding streams + video_stream = FFStream({ + "index": 0, + "codec_type": "video", + "codec_name": "h264", + "width": 1920, + "height": 1080 + }) + + audio_stream = FFStream({ + "index": 1, + "codec_type": "audio", + "codec_name": "aac", + "channels": 2 + }) + + result.video.append(video_stream) + result.audio.append(audio_stream) + + assert len(result.video) == 1 + assert len(result.audio) == 1 + assert result.video[0].is_video() + assert result.audio[0].is_audio() + + def test_stream_object_methods(self): + """Test Stream object methods.""" + from app.library.ffprobe import FFStream + + # Test video stream + video_stream = FFStream({ + "codec_type": "video", + "codec_name": "h264", + "width": 1920, + "height": 1080, + "duration": "10.5" + }) + + assert video_stream.is_video() + assert not video_stream.is_audio() + assert not video_stream.is_subtitle() + assert not video_stream.is_attachment() + + # Test audio stream + audio_stream = FFStream({ + "codec_type": "audio", + "codec_name": "aac", + "channels": 2, + "sample_rate": "44100" + }) + + assert not audio_stream.is_video() + assert audio_stream.is_audio() + assert not audio_stream.is_subtitle() + assert not audio_stream.is_attachment() + + # Test subtitle stream + subtitle_stream = FFStream({ + "codec_type": "subtitle", + "codec_name": "subrip" + }) + + assert not subtitle_stream.is_video() + assert not subtitle_stream.is_audio() + assert subtitle_stream.is_subtitle() + assert not subtitle_stream.is_attachment() + + # Test attachment stream + attachment_stream = FFStream({ + "codec_type": "attachment", + "codec_name": "ttf" + }) + + assert not attachment_stream.is_video() + assert not attachment_stream.is_audio() + assert not attachment_stream.is_subtitle() + assert attachment_stream.is_attachment() diff --git a/app/tests/test_generic_task_handler.py b/app/tests/test_generic_task_handler.py new file mode 100644 index 00000000..6654545e --- /dev/null +++ b/app/tests/test_generic_task_handler.py @@ -0,0 +1,356 @@ +import json +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from app.library.task_handlers.generic import ( + ContainerDefinition, + EngineConfig, + ExtractionRule, + GenericTaskHandler, + RequestConfig, + ResponseConfig, + TaskDefinition, + load_task_definitions, +) +from app.library.Tasks import Task, TaskFailure, TaskResult + + +@pytest.fixture(autouse=True) +def reset_generic_handler(monkeypatch): + monkeypatch.setattr(GenericTaskHandler, "_definitions", []) + monkeypatch.setattr(GenericTaskHandler, "_sources_mtime", {}) + + +def test_load_task_definitions_parses_valid_file(tmp_path: Path): + tasks_dir = tmp_path / "tasks" + tasks_dir.mkdir() + + definition_content = { + "name": "example", + "match": ["https://example.com/articles/*"], + "parse": { + "link": {"type": "css", "expression": ".article a.link::attr(href)"}, + "title": {"type": "css", "expression": ".article .title", "attribute": "text"}, + }, + } + + (tasks_dir / "01-example.json").write_text(json.dumps(definition_content), encoding="utf-8") + + config = SimpleNamespace(config_path=str(tmp_path)) + definitions = load_task_definitions(config=config) + + assert len(definitions) == 1 + definition = definitions[0] + assert definition.name == "example" + assert definition.matchers[0].matches("https://example.com/articles/123") + assert definition.parsers["link"].expression == ".article a.link::attr(href)" + + +def test_load_task_definitions_handles_container(tmp_path: Path): + tasks_dir = tmp_path / "tasks" + tasks_dir.mkdir() + + definition_content = { + "name": "container", + "match": ["https://example.com/cards"], + "parse": { + "items": { + "selector": ".cards .card", + "fields": { + "link": {"type": "css", "expression": ".card-header a", "attribute": "href"}, + "title": {"type": "css", "expression": ".card-header a", "attribute": "text"}, + }, + } + }, + } + + (tasks_dir / "02-container.json").write_text(json.dumps(definition_content), encoding="utf-8") + + config = SimpleNamespace(config_path=str(tmp_path)) + definitions = load_task_definitions(config=config) + + assert len(definitions) == 1 + definition = definitions[0] + assert definition.container is not None + assert definition.container.selector == ".cards .card" + assert "link" in definition.container.fields + assert definition.parsers == {} + + +def test_load_task_definitions_handles_json(tmp_path: Path): + tasks_dir = tmp_path / "tasks" + tasks_dir.mkdir() + + definition_content = { + "name": "json-def", + "match": ["https://example.com/api"], + "response": {"type": "json"}, + "parse": { + "items": { + "type": "jsonpath", + "selector": "items", + "fields": { + "link": {"type": "jsonpath", "expression": "url"}, + "title": {"type": "jsonpath", "expression": "title"}, + }, + } + }, + } + + (tasks_dir / "03-json.json").write_text(json.dumps(definition_content), encoding="utf-8") + + config = SimpleNamespace(config_path=str(tmp_path)) + definitions = load_task_definitions(config=config) + + assert len(definitions) == 1 + definition = definitions[0] + assert definition.response.format == "json" + assert definition.container is not None + assert definition.container.selector_type == "jsonpath" + assert definition.container.fields["link"].type == "jsonpath" + + +def test_parse_items_extracts_values(): + definition = TaskDefinition( + name="example", + source=Path("example.json"), + matchers=[], + engine=EngineConfig(), + request=RequestConfig(), + parsers={ + "link": ExtractionRule(type="css", expression=".article a.link::attr(href)", attribute=None), + "title": ExtractionRule(type="css", expression=".article .title", attribute="text"), + "id": ExtractionRule(type="css", expression=".article", attribute="data-id"), + }, + ) + + html = """ +
+ First + First Title +
+
+ Second + Second Title +
+ """ + + items = GenericTaskHandler._parse_items(definition, html, "https://example.com/base/") + + assert len(items) == 2 + assert items[0]["link"] == "https://example.com/article-101" + assert items[0]["title"] == "First Title" + assert items[0]["id"] == "101" + assert items[1]["link"] == "https://example.com/article-102" + + +def test_parse_items_handles_nested_card_layout(): + definition = TaskDefinition( + name="nested", + source=Path("nested.json"), + matchers=[], + engine=EngineConfig(), + request=RequestConfig(), + parsers={}, + container=ContainerDefinition( + selector_type="css", + selector=".columns .card", + fields={ + "link": ExtractionRule( + type="css", + expression=".card-header a[href]", + attribute="href", + ), + "title": ExtractionRule( + type="css", + expression=".card-header a[href]", + attribute="text", + ), + "poet": ExtractionRule( + type="css", + expression="footer .card-footer-item:first-child a", + attribute="text", + ), + "category": ExtractionRule( + type="css", + expression="footer .card-footer-item:nth-child(2) a", + attribute="text", + ), + }, + ), + ) + + html = """ +
+
+
+
+

+ First Poem +

+
+ +
+
+
+
+
+

+ Second Poem +

+
+ +
+
+
+ """ + + items = GenericTaskHandler._parse_items(definition, html, "https://example.com") + + assert len(items) == 2 + assert items[0]["link"] == "https://example.com/poems/view/111" + assert items[0]["title"] == "First Poem" + assert items[0]["poet"] == "Poet Alpha" + assert items[0]["category"] == "Category One" + + assert items[1]["link"] == "https://example.com/poems/view/222" + assert items[1]["title"] == "Second Poem" + assert items[1]["poet"] == "Poet Beta" + assert "category" not in items[1] + + +def test_parse_items_handles_json_container(): + definition = TaskDefinition( + name="json", + source=Path("json.json"), + matchers=[], + engine=EngineConfig(), + request=RequestConfig(), + parsers={}, + container=ContainerDefinition( + selector_type="jsonpath", + selector="entries", + fields={ + "link": ExtractionRule(type="jsonpath", expression="url"), + "title": ExtractionRule(type="jsonpath", expression="title"), + "id": ExtractionRule(type="jsonpath", expression="id"), + }, + ), + response=ResponseConfig(format="json"), + ) + + payload = { + "entries": [ + {"url": "/video/1", "title": "First", "id": 1}, + {"url": "https://example.com/video/2", "title": "Second", "id": 2}, + {"title": "Missing Link", "id": 3}, + ] + } + + items = GenericTaskHandler._parse_items( + definition=definition, + html="", + base_url="https://example.com", + json_data=payload, + ) + + assert len(items) == 2 + assert items[0]["link"] == "https://example.com/video/1" + assert items[0]["title"] == "First" + assert items[0]["id"] == "1" + + assert items[1]["link"] == "https://example.com/video/2" + assert items[1]["title"] == "Second" + assert items[1]["id"] == "2" + + +@pytest.mark.asyncio +async def test_generic_task_handler_inspect(monkeypatch): + definition = TaskDefinition( + name="json-inspect", + source=Path("json-inspect.json"), + matchers=[], + engine=EngineConfig(), + request=RequestConfig(), + parsers={}, + container=ContainerDefinition( + selector_type="jsonpath", + selector="items", + fields={ + "link": ExtractionRule(type="jsonpath", expression="url"), + "title": ExtractionRule(type="jsonpath", expression="title"), + }, + ), + response=ResponseConfig(format="json"), + ) + + monkeypatch.setattr( + GenericTaskHandler, + "_find_definition", + classmethod(lambda cls, url: definition), # noqa: ARG005 + ) + + async def fake_fetch_content(url, definition, ytdlp_opts): # noqa: ARG001 + return "", {"items": [{"url": "/video/1", "title": "First"}]} + + monkeypatch.setattr(GenericTaskHandler, "_fetch_content", staticmethod(fake_fetch_content)) + + task = Task(id="inspect", name="Inspect", url="https://example.com/api") + result: TaskResult | TaskFailure = await GenericTaskHandler.extract(task) + + assert isinstance(result, TaskResult) + assert len(result.items) == 1 + item = result.items[0] + assert item.url == "https://example.com/video/1" + assert item.title == "First" + + +def test_parse_items_handles_json_top_level_list(): + definition = TaskDefinition( + name="json-list", + source=Path("json-list.json"), + matchers=[], + engine=EngineConfig(), + request=RequestConfig(), + parsers={}, + container=ContainerDefinition( + selector_type="jsonpath", + selector="[]", + fields={ + "link": ExtractionRule(type="jsonpath", expression="url"), + "title": ExtractionRule(type="jsonpath", expression="title"), + }, + ), + response=ResponseConfig(format="json"), + ) + + payload = [ + {"url": "/video/1", "title": "First"}, + {"url": "/video/2", "title": "Second"}, + ] + + items = GenericTaskHandler._parse_items( + definition=definition, + html="", + base_url="https://example.com", + json_data=payload, + ) + + assert len(items) == 2 + assert items[0]["link"] == "https://example.com/video/1" + assert items[0]["title"] == "First" + assert items[1]["link"] == "https://example.com/video/2" + assert items[1]["title"] == "Second" diff --git a/app/tests/test_itemdto.py b/app/tests/test_itemdto.py index d777e251..796bbbbe 100644 --- a/app/tests/test_itemdto.py +++ b/app/tests/test_itemdto.py @@ -1,4 +1,5 @@ import json +from pathlib import Path from unittest.mock import patch import pytest @@ -165,3 +166,116 @@ class TestItemDTO: ok2 = dto.archive_delete() assert ok2 is True m_del.assert_called_once() + + def test_get_file_method(self): + """Test ItemDTO get_file method returns correct path.""" + # Create ItemDTO with filename but no folder + dto = ItemDTO( + id="test-id", + title="Test Video", + url="https://youtube.com/watch?v=test123", + folder="", + status="finished", + filename="test_video.mp4", + ) + + # Test with no filename returns None + dto_no_file = ItemDTO( + id="test-id-2", + title="Test Video 2", + url="https://youtube.com/watch?v=test456", + folder="", + status="finished", + ) + assert dto_no_file.get_file() is None + + # Mock get_file function to return success (without custom download_path) + with patch("app.library.ItemDTO.get_file") as mock_get_file, patch("app.library.config.Config") as mock_config: + mock_get_file.return_value = ("/downloads/test_video.mp4", 200) + mock_config.get_instance.return_value.download_path = "/downloads" + + result = dto.get_file() + assert result == Path("/downloads/test_video.mp4") + + # Test with folder + dto_with_folder = ItemDTO( + id="test-id-3", + title="Test Video 3", + url="https://youtube.com/watch?v=test789", + folder="media", + status="finished", + filename="test_video.mp4", + ) + + with patch("app.library.ItemDTO.get_file") as mock_get_file, patch("app.library.config.Config") as mock_config: + mock_get_file.return_value = ("/downloads/media/test_video.mp4", 200) + mock_config.get_instance.return_value.download_path = "/downloads" + + result = dto_with_folder.get_file() + assert result == Path("/downloads/media/test_video.mp4") + + # Test with file not found + with patch("app.library.ItemDTO.get_file") as mock_get_file, patch("app.library.config.Config") as mock_config: + mock_get_file.return_value = ("/downloads/test_video.mp4", 404) + mock_config.get_instance.return_value.download_path = "/downloads" + + result = dto.get_file() + assert result is None + + # Test with exception during file access + with patch("app.library.ItemDTO.get_file") as mock_get_file, patch("app.library.config.Config") as mock_config: + mock_get_file.side_effect = ValueError("File path error") + mock_config.get_instance.return_value.download_path = "/downloads" + + result = dto.get_file() + assert result is None + + # Test with custom download_path parameter (Config not imported in this case) + with patch("app.library.ItemDTO.get_file") as mock_get_file: + mock_get_file.return_value = ("/custom/test_video.mp4", 200) + + result = dto.get_file(download_path=Path("/custom")) + assert result == Path("/custom/test_video.mp4") + + def test_get_file_sidecar_populates_from_utils(self): + with patch.object(ItemDTO, "__post_init__", lambda _: None): + dto = ItemDTO(id="sidecar", title="Title", url="u", folder="f") + + expected_sidecar = { + "subtitle": [ + { + "file": Path("/downloads/video.en.srt"), + "lang": "en", + "name": "SRT (0) - en", + } + ] + } + + with ( + patch("app.library.ItemDTO.ItemDTO.get_file", autospec=True, return_value=Path("/downloads/video.mp4")) as mock_get_file, + patch("app.library.ItemDTO.get_file_sidecar", return_value=expected_sidecar) as mock_utils_sidecar, + ): + result = dto.get_file_sidecar() + + mock_get_file.assert_called_once_with(dto) + mock_utils_sidecar.assert_called_once_with(Path("/downloads/video.mp4")) + assert result is expected_sidecar + assert dto.sidecar is expected_sidecar + + def test_get_file_sidecar_returns_existing_when_no_file(self): + with patch.object(ItemDTO, "__post_init__", lambda _: None): + dto = ItemDTO(id="sidecar-none", title="Title", url="u", folder="f") + + existing = {"existing": []} + dto.sidecar = existing + + with ( + patch("app.library.ItemDTO.ItemDTO.get_file", autospec=True, return_value=None) as mock_get_file, + patch("app.library.ItemDTO.get_file_sidecar") as mock_utils_sidecar, + ): + result = dto.get_file_sidecar() + + mock_get_file.assert_called_once_with(dto) + mock_utils_sidecar.assert_not_called() + assert result is existing + assert dto.sidecar is existing diff --git a/app/tests/test_notifications.py b/app/tests/test_notifications.py index b67e4103..8d59b1a6 100644 --- a/app/tests/test_notifications.py +++ b/app/tests/test_notifications.py @@ -1003,87 +1003,6 @@ class TestNotification: assert result is None mock_worker_instance.submit.assert_called_once_with(notification.send, event) - def test_deep_unpack_simple_dict(self): - """Test _deep_unpack method with simple dictionary.""" - with patch("app.library.Notifications.Config") as mock_config, \ - patch("app.library.Notifications.BackgroundWorker") as mock_worker: - - mock_config.get_instance.return_value = Mock( - debug=False, config_path="/tmp", app_version="1.0.0" - ) - mock_worker.get_instance.return_value = Mock() - - notification = Notification.get_instance() - - data = {"key1": "value1", "key2": 123} - result = notification._deep_unpack(data) - - assert result == {"key1": "value1", "key2": 123} - - def test_deep_unpack_nested_dict(self): - """Test _deep_unpack method with nested dictionary.""" - with patch("app.library.Notifications.Config") as mock_config, \ - patch("app.library.Notifications.BackgroundWorker") as mock_worker: - - mock_config.get_instance.return_value = Mock( - debug=False, config_path="/tmp", app_version="1.0.0" - ) - mock_worker.get_instance.return_value = Mock() - - notification = Notification.get_instance() - - data = { - "level1": { - "level2": { - "value": "nested" - } - } - } - result = notification._deep_unpack(data) - - assert result["level1"]["level2"]["value"] == "nested" - - def test_deep_unpack_with_datetime(self): - """Test _deep_unpack method with datetime objects.""" - from datetime import UTC, datetime - - with patch("app.library.Notifications.Config") as mock_config, \ - patch("app.library.Notifications.BackgroundWorker") as mock_worker: - - mock_config.get_instance.return_value = Mock( - debug=False, config_path="/tmp", app_version="1.0.0" - ) - mock_worker.get_instance.return_value = Mock() - - notification = Notification.get_instance() - - test_datetime = datetime.now(tz=UTC) - data = {"timestamp": test_datetime} - result = notification._deep_unpack(data) - - assert result["timestamp"] == test_datetime.isoformat() - - def test_deep_unpack_with_serializable_object(self): - """Test _deep_unpack method with object having serialize method.""" - with patch("app.library.Notifications.Config") as mock_config, \ - patch("app.library.Notifications.BackgroundWorker") as mock_worker: - - mock_config.get_instance.return_value = Mock( - debug=False, config_path="/tmp", app_version="1.0.0" - ) - mock_worker.get_instance.return_value = Mock() - - notification = Notification.get_instance() - - # Mock object with serialize method - mock_obj = Mock() - mock_obj.serialize.return_value = {"serialized": "data"} - - data = {"object": mock_obj} - result = notification._deep_unpack(data) - - assert result["object"] == {"serialized": "data"} - @pytest.mark.asyncio async def test_noop(self): """Test noop method.""" diff --git a/app/tests/test_twitch_handler.py b/app/tests/test_twitch_handler.py new file mode 100644 index 00000000..821eb605 --- /dev/null +++ b/app/tests/test_twitch_handler.py @@ -0,0 +1,57 @@ +import pytest + +from app.library.task_handlers.twitch import TwitchHandler +from app.library.Tasks import Task, TaskResult + + +class DummyResponse: + def __init__(self, text: str): + self.text = text + + def raise_for_status(self) -> None: + return None + + +class DummyOpts: + def __init__(self, data): + self._data = data + + def get_all(self): + return self._data + + +@pytest.mark.asyncio +async def test_twitch_handler_inspect(monkeypatch): + feed = """ + + + https://www.twitch.tv/videos/111 + First VOD + + + https://www.twitch.tv/videos/222 + Second VOD + + + """.strip() + + async def fake_request(**kwargs): # noqa: ARG001 + return DummyResponse(feed) + + monkeypatch.setattr(TwitchHandler, "request", staticmethod(fake_request)) + monkeypatch.setattr(Task, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": "/tmp/archive"})) # noqa: ARG005 + + task = Task( + id="inspect", + name="Inspect", + url="https://www.twitch.tv/testchannel", + preset="default", + ) + + result = await TwitchHandler.extract(task) + + assert isinstance(result, TaskResult) + assert len(result.items) == 2 + assert result.items[0].url == "https://www.twitch.tv/videos/111" + assert result.items[0].title == "First VOD" + assert result.metadata.get("has_entries") is True diff --git a/app/tests/test_utils.py b/app/tests/test_utils.py index 6284a583..aeeda927 100644 --- a/app/tests/test_utils.py +++ b/app/tests/test_utils.py @@ -44,6 +44,7 @@ from app.library.Utils import ( str_to_dt, strip_newline, tail_log, + timed_lru_cache, validate_url, validate_uuid, ytdlp_reject, @@ -65,6 +66,233 @@ class TestStreamingError: assert isinstance(error, Exception) +class TestTimedLruCache: + """Test the timed_lru_cache decorator.""" + + def test_timed_lru_cache_basic_functionality(self): + """Test that timed_lru_cache caches function results.""" + call_count = 0 + + @timed_lru_cache(ttl_seconds=60, max_size=10) + def test_function(x): + nonlocal call_count + call_count += 1 + return x * 2 + + # First call should execute the function + result1 = test_function(5) + assert result1 == 10 + assert call_count == 1 + + # Second call with same argument should return cached result + result2 = test_function(5) + assert result2 == 10 + assert call_count == 1 # Should not increment + + # Different argument should execute the function again + result3 = test_function(10) + assert result3 == 20 + assert call_count == 2 + + def test_timed_lru_cache_expiration(self): + """Test that cache expires after TTL.""" + import time + + call_count = 0 + + @timed_lru_cache(ttl_seconds=1, max_size=10) # 1 second TTL + def test_function(x): + nonlocal call_count + call_count += 1 + return x * 2 + + # First call + result1 = test_function(5) + assert result1 == 10 + assert call_count == 1 + + # Second call immediately should be cached + result2 = test_function(5) + assert result2 == 10 + assert call_count == 1 + + # Wait for cache to expire + time.sleep(1.1) + + # Call after expiration should execute function again + result3 = test_function(5) + assert result3 == 10 + assert call_count == 2 + + def test_timed_lru_cache_methods_exposed(self): + """Test that cache_clear and cache_info methods are exposed.""" + from app.library.Utils import timed_lru_cache + + @timed_lru_cache(ttl_seconds=60, max_size=10) + def test_function(x): + return x * 2 + + # Test that methods exist + assert hasattr(test_function, "cache_clear") + assert hasattr(test_function, "cache_info") + + # Call function to populate cache + test_function(5) + + # Get cache info + info = test_function.cache_info() + assert info.hits == 0 + assert info.misses == 1 + + # Call again to test hit + test_function(5) + info = test_function.cache_info() + assert info.hits == 1 + assert info.misses == 1 + + # Clear cache + test_function.cache_clear() + info = test_function.cache_info() + assert info.hits == 0 + assert info.misses == 0 + + def test_timed_lru_cache_max_size(self): + """Test that cache respects max_size limit.""" + from app.library.Utils import timed_lru_cache + + @timed_lru_cache(ttl_seconds=60, max_size=2) + def test_function(x): + return x * 2 + + # Fill cache to max size + test_function(1) + test_function(2) + + info = test_function.cache_info() + assert info.misses == 2 + + # Adding another item should not exceed max size + test_function(3) + info = test_function.cache_info() + assert info.misses == 3 + + # Test that earlier entries may be evicted + test_function(1) # This might be a cache miss due to LRU eviction + + +class TestAsyncTimedLruCache: + """Test async functionality of timed_lru_cache decorator.""" + + @pytest.mark.asyncio + async def test_async_timed_lru_cache_basic(self): + """Test basic async caching functionality.""" + from app.library.Utils import timed_lru_cache + + call_count = 0 + + @timed_lru_cache(ttl_seconds=300, max_size=128) + async def async_test_func(x): + nonlocal call_count + call_count += 1 + return x * 2 + + # First call should execute the function + result1 = await async_test_func(5) + assert result1 == 10 + assert call_count == 1 + + # Second call should use cached result + result2 = await async_test_func(5) + assert result2 == 10 + assert call_count == 1 # Should not increment + + # Different argument should execute function again + result3 = await async_test_func(3) + assert result3 == 6 + assert call_count == 2 + + @pytest.mark.asyncio + async def test_async_timed_lru_cache_expiry(self): + """Test that async cache entries expire after TTL.""" + from app.library.Utils import timed_lru_cache + + call_count = 0 + + @timed_lru_cache(ttl_seconds=0.1, max_size=128) # 100ms TTL + async def async_expire_func(x): + nonlocal call_count + call_count += 1 + return x * 3 + + # First call + result1 = await async_expire_func(2) + assert result1 == 6 + assert call_count == 1 + + # Immediate second call should use cache + result2 = await async_expire_func(2) + assert result2 == 6 + assert call_count == 1 + + # Wait for cache to expire + await asyncio.sleep(0.15) + + # Third call should execute function again due to expiry + result3 = await async_expire_func(2) + assert result3 == 6 + assert call_count == 2 + + @pytest.mark.asyncio + async def test_async_cache_methods(self): + """Test that async cached functions expose cache management methods.""" + from app.library.Utils import timed_lru_cache + + @timed_lru_cache(ttl_seconds=300, max_size=128) + async def async_method_test(x): + return x + 1 + + # Test that cache methods exist + assert hasattr(async_method_test, "cache_clear") + assert hasattr(async_method_test, "cache_info") + + # Test cache_info + info = async_method_test.cache_info() + assert hasattr(info, "hits") + assert hasattr(info, "misses") + assert hasattr(info, "maxsize") + assert hasattr(info, "currsize") + assert info.maxsize == 128 + + # Test cache_clear + await async_method_test(1) + async_method_test.cache_clear() + info_after_clear = async_method_test.cache_info() + assert info_after_clear.currsize == 0 + + @pytest.mark.asyncio + async def test_async_cache_max_size(self): + """Test that async cache respects max_size parameter.""" + from app.library.Utils import timed_lru_cache + + @timed_lru_cache(ttl_seconds=300, max_size=2) + async def async_limited_func(x): + return x * 4 + + # Fill cache beyond max_size + result1 = await async_limited_func(1) + result2 = await async_limited_func(2) + result3 = await async_limited_func(3) # Should evict oldest entry + + # Verify results + assert result1 == 4 + assert result2 == 8 + assert result3 == 12 + + # Check cache size is limited + info = async_limited_func.cache_info() + assert info.currsize <= 2 + + class TestFileLogFormatter: """Test the FileLogFormatter class.""" diff --git a/app/tests/test_youtube_handler.py b/app/tests/test_youtube_handler.py new file mode 100644 index 00000000..26ee9c91 --- /dev/null +++ b/app/tests/test_youtube_handler.py @@ -0,0 +1,61 @@ +import pytest + +from app.library.task_handlers.youtube import YoutubeHandler +from app.library.Tasks import Task, TaskResult + + +class DummyResponse: + def __init__(self, text: str): + self.text = text + + def raise_for_status(self) -> None: + return None + + +class DummyOpts: + def __init__(self, data): + self._data = data + + def get_all(self): + return self._data + + +@pytest.mark.asyncio +async def test_youtube_handler_inspect(monkeypatch): + feed = """ + + + abc123 + First Video + 2024-01-01T00:00:00Z + + + def456 + Second Video + 2024-01-02T00:00:00Z + + + """.strip() + + async def fake_request(**kwargs): # noqa: ARG001 + return DummyResponse(feed) + + monkeypatch.setattr(YoutubeHandler, "request", staticmethod(fake_request)) + monkeypatch.setattr(Task, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": "/tmp/archive"})) # noqa: ARG005 + + task = Task( + id="inspect", + name="Inspect", + url="https://www.youtube.com/channel/UCabcdefghijklmnopqrstuv", + preset="default", + ) + + result = await YoutubeHandler.extract(task) + + assert isinstance(result, TaskResult) + assert len(result.items) == 2 + first = result.items[0] + assert first.url == "https://www.youtube.com/watch?v=abc123" + assert first.title == "First Video" + assert first.metadata.get("published") == "2024-01-01T00:00:00Z" + assert result.metadata.get("entry_count") == 2 diff --git a/pyproject.toml b/pyproject.toml index 9fe72038..534abeaa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,6 +40,9 @@ dependencies = [ "bgutil-ytdlp-pot-provider>=1.2.1", "pycryptodome>=3.23.0", "httpx-curl-cffi>=0.1.4", + "selenium>=4.35.0", + "parsel>=1.10.0", + "jmespath>=1.0.1", ] [tool.ruff] diff --git a/ui/app/components/Dialog.vue b/ui/app/components/Dialog.vue index a303932c..74fda850 100644 --- a/ui/app/components/Dialog.vue +++ b/ui/app/components/Dialog.vue @@ -44,7 +44,7 @@

-
+
@@ -61,8 +61,8 @@