diff --git a/.vscode/settings.json b/.vscode/settings.json index 24590940..2e166d45 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -191,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/FAQ.md b/FAQ.md index 26ce8118..e8503381 100644 --- a/FAQ.md +++ b/FAQ.md @@ -145,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 b78b4350..dd82fb82 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,7 @@ includes features like scheduling downloads, sending notifications, and built-in * 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. 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/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/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_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_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_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 @@