From 793f67db42bb93cfc3d64a22eba44f64ec18c7a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Hells=C3=A9n?= Date: Mon, 20 Jul 2026 06:02:25 +0200 Subject: [PATCH] Add support for extensions --- Dockerfile | 6 +- README.md | 10 +- discord_rss_bot/custom_filters.py | 55 - discord_rss_bot/custom_message.py | 105 +- discord_rss_bot/extensions/README.md | 157 +++ discord_rss_bot/extensions/__init__.py | 32 + discord_rss_bot/extensions/base.py | 127 ++ discord_rss_bot/extensions/discovery.py | 191 +++ discord_rss_bot/extensions/hoyolab.py | 331 +++++ .../extensions/jwplayer_thumbnail.py | 285 +++++ discord_rss_bot/extensions/runner.py | 239 ++++ discord_rss_bot/extensions/steam.py | 276 ++++ discord_rss_bot/extensions/storage.py | 68 + discord_rss_bot/extensions/wordpress.py | 116 ++ discord_rss_bot/extensions/youtube.py | 110 ++ discord_rss_bot/feeds.py | 260 +--- discord_rss_bot/filter/blacklist.py | 48 - discord_rss_bot/filter/evaluator.py | 45 + discord_rss_bot/filter/whitelist.py | 48 - discord_rss_bot/hoyolab_api.py | 222 ---- discord_rss_bot/html_format.py | 77 ++ discord_rss_bot/main.py | 185 ++- discord_rss_bot/settings.py | 4 + discord_rss_bot/templates/custom.html | 28 +- discord_rss_bot/templates/embed.html | 47 +- discord_rss_bot/templates/extensions.html | 68 + discord_rss_bot/templates/feed.html | 163 ++- discord_rss_bot/templates/index.html | 32 +- docker-compose.yml | 7 + pyproject.toml | 2 +- tests/test_blacklist.py | 4 +- tests/test_custom_filter.py | 6 +- tests/test_custom_message.py | 4 - tests/test_extensions.py | 1137 +++++++++++++++++ tests/test_feeds.py | 211 +-- tests/test_hoyolab_api.py | 248 ---- tests/test_main.py | 19 +- tests/test_settings.py | 2 +- tests/test_whitelist.py | 4 +- 39 files changed, 3670 insertions(+), 1309 deletions(-) delete mode 100644 discord_rss_bot/custom_filters.py create mode 100644 discord_rss_bot/extensions/README.md create mode 100644 discord_rss_bot/extensions/__init__.py create mode 100644 discord_rss_bot/extensions/base.py create mode 100644 discord_rss_bot/extensions/discovery.py create mode 100644 discord_rss_bot/extensions/hoyolab.py create mode 100644 discord_rss_bot/extensions/jwplayer_thumbnail.py create mode 100644 discord_rss_bot/extensions/runner.py create mode 100644 discord_rss_bot/extensions/steam.py create mode 100644 discord_rss_bot/extensions/storage.py create mode 100644 discord_rss_bot/extensions/wordpress.py create mode 100644 discord_rss_bot/extensions/youtube.py delete mode 100644 discord_rss_bot/filter/blacklist.py delete mode 100644 discord_rss_bot/filter/whitelist.py delete mode 100644 discord_rss_bot/hoyolab_api.py create mode 100644 discord_rss_bot/html_format.py create mode 100644 discord_rss_bot/templates/extensions.html create mode 100644 tests/test_extensions.py delete mode 100644 tests/test_hoyolab_api.py diff --git a/Dockerfile b/Dockerfile index f27eed9..655b06f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,7 +1,7 @@ FROM python:3.14-slim COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/ RUN useradd --create-home botuser && \ - mkdir -p /home/botuser/discord-rss-bot/ /home/botuser/.local/share/discord_rss_bot/ && \ + mkdir -p /home/botuser/discord-rss-bot/ /home/botuser/.local/share/discord_rss_bot/ /home/botuser/extensions/ && \ chown -R botuser:botuser /home/botuser/ USER botuser WORKDIR /home/botuser/discord-rss-bot @@ -11,5 +11,9 @@ RUN --mount=type=cache,target=/root/.cache/uv \ COPY --chown=botuser:botuser discord_rss_bot/ /home/botuser/discord-rss-bot/discord_rss_bot/ EXPOSE 5000 VOLUME ["/home/botuser/.local/share/discord_rss_bot/"] + +# Extensions directory — bind-mount your own plugins here. +ENV EXTENSIONS_DIR=/home/botuser/extensions + HEALTHCHECK --interval=10m --timeout=5s CMD ["uv", "run", "./discord_rss_bot/healthcheck.py"] CMD ["uv", "run", "uvicorn", "discord_rss_bot.main:app", "--host=0.0.0.0", "--port=5000", "--proxy-headers", "--forwarded-allow-ips='*'", "--log-level", "debug"] diff --git a/README.md b/README.md index 776437f..de243db 100644 --- a/README.md +++ b/README.md @@ -15,12 +15,10 @@ Discord: TheLovinator#9276 - Regex filters for RSS feeds. - Blacklist/whitelist words in the title/description/author/etc. - Set different update frequencies for each feed or use a global default. -- Gets extra information from APIs if available, currently for: - - [https://feeds.c3kay.de/](https://feeds.c3kay.de/) - - Genshin Impact News - - Honkai Impact 3rd News - - Honkai Starrail News - - Zenless Zone Zero News +- Extensions that pulls extra data from feeds. + - Built-in extensions for Steam, YouTube, Hoyolab, WordPress, and JWPlayer. + - Add your own by dropping a `.py` file in `extensions/` ([docs](discord_rss_bot/extensions/README.md)). + - Feel free to open a GitHub issue if you need one. ## Installation diff --git a/discord_rss_bot/custom_filters.py b/discord_rss_bot/custom_filters.py deleted file mode 100644 index 4e060f2..0000000 --- a/discord_rss_bot/custom_filters.py +++ /dev/null @@ -1,55 +0,0 @@ -from __future__ import annotations - -import urllib.parse -from functools import lru_cache -from typing import TYPE_CHECKING - -from discord_rss_bot.filter.evaluator import get_entry_filter_decision_from_reader - -if TYPE_CHECKING: - from reader import Entry - from reader import Reader - - -@lru_cache -def encode_url(url_to_quote: str) -> str: - """%-escape the URL so it can be used in a URL. - - If we didn't do this, we couldn't go to feeds with a ? in the URL. - You can use this in templates with {{ url | encode_url }}. - - Args: - url_to_quote: The url to encode. - - Returns: - The encoded url. - """ - return urllib.parse.quote(string=url_to_quote) if url_to_quote else "" - - -def entry_is_whitelisted(entry_to_check: Entry, reader: Reader) -> bool: - """Check if the entry is whitelisted. - - Args: - entry_to_check: The feed to check. - reader: Custom Reader instance. - - Returns: - bool: True if the feed is whitelisted, False otherwise. - - """ - return get_entry_filter_decision_from_reader(reader, entry_to_check).whitelist_match is not None - - -def entry_is_blacklisted(entry_to_check: Entry, reader: Reader) -> bool: - """Check if the entry is blacklisted. - - Args: - entry_to_check: The feed to check. - reader: Custom Reader instance. - - Returns: - bool: True if the feed is blacklisted, False otherwise. - - """ - return get_entry_filter_decision_from_reader(reader, entry_to_check).blacklist_match is not None diff --git a/discord_rss_bot/custom_message.py b/discord_rss_bot/custom_message.py index 944ba67..2ec3a17 100644 --- a/discord_rss_bot/custom_message.py +++ b/discord_rss_bot/custom_message.py @@ -1,16 +1,15 @@ from __future__ import annotations -import html import json import logging -import re from dataclasses import dataclass from typing import TYPE_CHECKING from bs4 import BeautifulSoup from bs4 import Tag -from markdownify import markdownify +from discord_rss_bot.extensions import run_extensions +from discord_rss_bot.html_format import format_entry_html_for_discord from discord_rss_bot.is_url_valid import is_url_valid if TYPE_CHECKING: @@ -23,8 +22,6 @@ if TYPE_CHECKING: logger: logging.Logger = logging.getLogger(__name__) -DISCORD_TIMESTAMP_TAG_RE: re.Pattern[str] = re.compile(r"") - # Discord webhook username: nickname rules, max 80 chars; no "clyde"/"discord" substrings. DISCORD_WEBHOOK_USERNAME_MAX_LENGTH: int = 80 DISCORD_WEBHOOK_USERNAME_FORBIDDEN_CHARS: frozenset[str] = frozenset("@#:`") @@ -43,7 +40,7 @@ class CustomEmbed: thumbnail_url: str = "" footer_text: str = "" footer_icon_url: str = "" - show_steam_game_icon_in_thumbnail: bool = False + show_steam_game_icon_in_thumbnail: bool = True def try_to_replace(custom_message: str, template: str, replace_with: str) -> str: @@ -64,68 +61,6 @@ def try_to_replace(custom_message: str, template: str, replace_with: str) -> str return custom_message -def _preserve_discord_timestamp_tags(text: str) -> tuple[str, dict[str, str]]: - """Replace Discord timestamp tags with placeholders before markdown conversion. - - Args: - text: The text to replace tags in. - - Returns: - The text with Discord timestamp tags replaced by placeholders and a mapping of placeholders to original tags. - """ - replacements: dict[str, str] = {} - - def replace_match(match: re.Match[str]) -> str: - placeholder: str = f"DISCORDTIMESTAMPPLACEHOLDER{len(replacements)}" - replacements[placeholder] = match.group(0) - return placeholder - - return DISCORD_TIMESTAMP_TAG_RE.sub(replace_match, text), replacements - - -def _restore_discord_timestamp_tags(text: str, replacements: dict[str, str]) -> str: - """Restore preserved Discord timestamp tags after markdown conversion. - - Args: - text: The text to restore tags in. - replacements: A mapping of placeholders to original Discord timestamp tags. - - Returns: - The text with placeholders replaced by the original Discord timestamp tags. - """ - for placeholder, original_value in replacements.items(): - text = text.replace(placeholder, original_value) - return text - - -def format_entry_html_for_discord(text: str) -> str: - """Convert entry HTML to Discord-friendly markdown while preserving Discord timestamp tags. - - Args: - text: The HTML text to format. - - Returns: - The formatted text with Discord timestamp tags preserved. - """ - if not text: - return "" - - unescaped_text: str = html.unescape(text) - protected_text, replacements = _preserve_discord_timestamp_tags(unescaped_text) - formatted_text: str = markdownify( - html=protected_text, - strip=["img", "table", "td", "tr", "tbody", "thead"], - escape_misc=False, - heading_style="ATX", - ) - - if "[https://" in formatted_text or "[https://www." in formatted_text: - formatted_text = formatted_text.replace("[https://", "[") - formatted_text = formatted_text.replace("[https://www.", "[") - - return _restore_discord_timestamp_tags(formatted_text, replacements) - - def replace_tags_in_text_message(entry: Entry, reader: Reader) -> str: """Replace tags in custom_message. @@ -191,6 +126,12 @@ def replace_tags_in_text_message(entry: Entry, reader: Reader) -> str: {"{{image_1}}": first_image}, ] + # Compute extension variables (handled separately so they can use + # the already-computed values above without ordering issues). + extension_vars: dict[str, str] = run_extensions(entry, reader) + for var_name, var_value in extension_vars.items(): + list_of_replacements.append({f"{{{{{var_name}}}}}": var_value}) + for replacement in list_of_replacements: for template, replace_with in replacement.items(): if not isinstance(replace_with, str): @@ -353,6 +294,12 @@ def replace_tags_in_embed(feed: Feed, entry: Entry, reader: Reader) -> CustomEmb {"{{entry_updated}}": entry_updated or ""}, {"{{image_1}}": first_image or ""}, ] + # Compute extension variables (handled separately so they can use + # the already-computed values above without ordering issues). + extension_vars: dict[str, str] = run_extensions(entry, reader) + for var_name, var_value in extension_vars.items(): + list_of_replacements.append({f"{{{{{var_name}}}}}": var_value}) + for replacement in list_of_replacements: for template, replace_with in replacement.items(): _replace_embed_tags(embed, template, replace_with) @@ -547,25 +494,10 @@ def get_embed(reader: Reader, feed: Feed) -> CustomEmbed: thumbnail_url="", footer_text="", footer_icon_url="", - show_steam_game_icon_in_thumbnail=False, + show_steam_game_icon_in_thumbnail=True, ) -def coerce_embed_bool(value: object) -> bool: - """Normalize stored embed booleans from JSON or form-like values. - - Returns: - The coerced boolean value. - """ - if isinstance(value, bool): - return value - if isinstance(value, int): - return value != 0 - if isinstance(value, str): - return value.strip().lower() in {"1", "true", "on", "yes"} - return False - - def get_embed_data(embed_data: dict[str, str | int | bool]) -> CustomEmbed: """Get embed data from embed_data. @@ -585,8 +517,9 @@ def get_embed_data(embed_data: dict[str, str | int | bool]) -> CustomEmbed: thumbnail_url: str = str(embed_data.get("thumbnail_url", "")) footer_text: str = str(embed_data.get("footer_text", "")) footer_icon_url: str = str(embed_data.get("footer_icon_url", "")) - show_steam_game_icon_in_thumbnail: bool = coerce_embed_bool( - embed_data.get("show_steam_game_icon_in_thumbnail", False), + show_steam_game_icon_in_thumbnail: bool = embed_data.get( # pyright: ignore[reportAssignmentType] + "show_steam_game_icon_in_thumbnail", + True, ) return CustomEmbed( diff --git a/discord_rss_bot/extensions/README.md b/discord_rss_bot/extensions/README.md new file mode 100644 index 0000000..41891fb --- /dev/null +++ b/discord_rss_bot/extensions/README.md @@ -0,0 +1,157 @@ +# Extension System + +Extensions are Python plugins that extract data from RSS/Atom entries and/or mutate the Discord webhook payload before messages are sent. + +**Two hooks per entry:** + +1. **`process_entry()`** → return `{variable_name: string_value}` dict → available as `{{variable_name}}` in templates. +2. **`modify_webhook()`** → mutate the `DiscordWebhook` (files, embeds, author, etc.) after template replacement. + +--- + +## TL;DR — Make an Extension + +Drop a `.py` file in your `extensions/` dir (or `$EXTENSIONS_DIR`). Subclass `FeedExtension`: + +```python +""" +Word count example — provides {{word_count}} per entry. +""" +from __future__ import annotations +import logging +from typing import ClassVar, TYPE_CHECKING +from discord_rss_bot.extensions.base import FeedExtension + +if TYPE_CHECKING: + from reader import Entry, Reader + +logger = logging.getLogger(__name__) + + +class WordCountExtension(FeedExtension): + name = "word_count" + description = "Counts words in each entry's content." + provides_variables: ClassVar[list[str]] = ["word_count"] + + def process_entry(self, entry: Entry, reader: Reader) -> dict[str, str]: + text = " ".join( + c.value for c in (entry.content or []) if hasattr(c, "value") and c.value + ) or (entry.summary or "") + return {"word_count": str(len(text.split())) if text else "0"} +``` + +Then restart the bot, enable it in the web UI's Extensions page, and use `{{word_count}}` in templates. + +--- + +## API Reference + +### Class Attributes + +| Attribute | Type | Required | Purpose | +| -------------------------- | --------------------- | -------- | --------------------------------------------------------- | +| `name` | `ClassVar[str]` | **Yes** | Unique slug for logs, registry, enable/disable | +| `description` | `ClassVar[str]` | No | Shown in web UI | +| `provides_variables` | `ClassVar[list[str]]` | No | Template variables this extension produces | +| `auto_enable_url_patterns` | `ClassVar[list[str]]` | No | Regex patterns; matching feeds auto-enable this extension | + +### `process_entry(self, entry: Entry, reader: Reader) -> dict[str, str]` (required) + +Called for every entry. Return `{var: value}` pairs. Missing variables from `provides_variables` get filled with `""`. + +- `entry.content` — list of `Content` objects (`.value`, `.type`) +- `entry.summary` — plain-text string +- `entry.feed.url`, `entry.link`, `entry.title`, `entry.id`, etc. + +### `modify_webhook(self, webhook: DiscordWebhook, entry: Entry, reader: Reader) -> DiscordWebhook` (optional) + +Called after template replacement. Return modified `webhook` or it unchanged. + +```python +# Content +webhook.content = "..." # or None to clear + +# Author +webhook.username = "Bot Name" +webhook.avatar_url = "https://..." + +# Embeds +embed = DiscordEmbed() +embed.set_title("...") +embed.set_description("...") +embed.set_url("https://...") +embed.set_color(0x00FF00) # or hex str "00FF00" +embed.set_author(name="...", url="...", icon_url="...") +embed.set_thumbnail(url="...") +embed.set_image(url="...") +embed.set_footer(text="...", icon_url="...") +embed.add_embed_field(name="Key", value="Value", inline=True) +embed.set_timestamp(timestamp="1234567890") +webhook.add_embed(embed) +webhook.remove_embeds() + +# Files +webhook.add_file(file=bytes_, filename="img.png") +``` + +### Utility Class Methods + +- `matches_feed_url(cls, feed_url) → bool` — checks `feed_url` against `auto_enable_url_patterns` +- `get_enabled_variables(cls, registry, enabled_names) → list[str]` — sorted union of `provides_variables` for enabled extensions + +--- + +## Discovery + +Extensions are discovered from two sources (external overrides built-in): + +1. **Built-in** — `discord_rss_bot/extensions/*.py` (this dir) +2. **External** — `EXTENSIONS_DIR` env var, default `extensions/` relative to CWD + +Every `.py` (except `__init__`) is imported. Concrete `FeedExtension` subclasses with a non-empty `name` go into a global `registry: dict[str, type[FeedExtension]]`. Duplicate names log a warning; last one wins. + +| Function | Description | +| ---------------------------------- | ------------------------------------------------ | +| `discover_plugins(*, force=False)` | Scan dirs, populate registry, return it | +| `get_registry()` | Return current registry (discover on first call) | +| `registry_clear()` | Clear registry (for tests) | + +--- + +## Auto-Enable + +Set `auto_enable_url_patterns` on your class: + +```python +auto_enable_url_patterns: ClassVar[list[str]] = [ + r"store\.steampowered\.com", + r"youtube\.com/feeds/videos\.xml", +] +``` + +Uses `re.search()` (matches anywhere in URL). Only applies when the feed's extensions tag was never explicitly saved by the user. Runs on feed creation and lazily on first process for old feeds. + +--- + +## Per-Feed Storage + +Enabled extensions stored as a JSON list under reader tag key `"extensions"`. + +| Function | Description | +| --------------------------------------------------------------- | ---------------------------- | +| `get_enabled_extensions_for_feed(reader, feed_url) → list[str]` | Get enabled names for a feed | +| `set_enabled_extensions_for_feed(reader, feed_url, names)` | Persist enabled names | + +Web UI: `GET /extensions?feed_url=...` renders checkboxes; `POST /extensions` saves. + +--- + +## Built-in Extensions + +| Extension | `name` | Variables | Auto-enable | +| ---------------------------------------------- | -------------------- | ----------------------------------------------------------------------------------------------------------- | ---------------------------------------------- | +| [youtube.py](youtube.py) | `youtube` | `youtube_video_id`, `youtube_embed_url` | `youtube.com/feeds/videos.xml` | +| [steam.py](steam.py) | `steam` | `steam_thumbnail_url`, `steam_app_id` | `store.steampowered.com`, `steamcommunity.com` | +| [hoyolab.py](hoyolab.py) | `hoyolab` | `hoyolab_subject`, `hoyolab_description`, `hoyolab_image`, `hoyolab_author` | `feeds.c3kay.de` | +| [jwplayer_thumbnail.py](jwplayer_thumbnail.py) | `jwplayer_thumbnail` | `jwplayer_thumbnail`, `jwplayer_file` | `hentaigasm.com`, `hgasm[0-9]*.com` | +| [wordpress.py](wordpress.py) | `wordpress` | `wp_content`, `wp_content_raw`, `wp_excerpt`, `wp_excerpt_raw`, `wp_jwplayer_thumbnail`, `wp_jwplayer_file` | _(none)_ | diff --git a/discord_rss_bot/extensions/__init__.py b/discord_rss_bot/extensions/__init__.py new file mode 100644 index 0000000..ae2cf85 --- /dev/null +++ b/discord_rss_bot/extensions/__init__.py @@ -0,0 +1,32 @@ +"""Feed extension system: discover, configure, and run per-entry content extensions. + +Extensions are Python classes that inherit from ``FeedExtension``, +auto-discovered from the ``EXTENSIONS_DIR`` directory (default: ``extensions/``). + +Typical usage:: + + from discord_rss_bot.extensions import run_extensions + + extra_vars: dict[str, str] = run_extensions(entry, reader) + # extra_vars == {"jwplayer_thumbnail": "https://...", ...} +""" + +from __future__ import annotations + +from discord_rss_bot.extensions.base import FeedExtension +from discord_rss_bot.extensions.discovery import discover_plugins +from discord_rss_bot.extensions.discovery import get_registry +from discord_rss_bot.extensions.discovery import registry_clear +from discord_rss_bot.extensions.runner import auto_enable_extensions_for_feed +from discord_rss_bot.extensions.runner import run_extensions +from discord_rss_bot.extensions.runner import run_modify_webhook + +__all__ = [ + "FeedExtension", + "auto_enable_extensions_for_feed", + "discover_plugins", + "get_registry", + "registry_clear", + "run_extensions", + "run_modify_webhook", +] diff --git a/discord_rss_bot/extensions/base.py b/discord_rss_bot/extensions/base.py new file mode 100644 index 0000000..b6c25d8 --- /dev/null +++ b/discord_rss_bot/extensions/base.py @@ -0,0 +1,127 @@ +"""Abstract base class for feed extensions. + +Every extension in the ``EXTENSIONS_DIR`` must inherit from +``FeedExtension`` and implement ``process_entry()``. +""" + +from __future__ import annotations + +import re +from abc import ABC +from abc import abstractmethod +from typing import TYPE_CHECKING +from typing import ClassVar + +if TYPE_CHECKING: + from reader import Entry + from reader import Reader + + from discord_rss_bot.webhook import DiscordWebhook + + +class FeedExtension(ABC): + """Base class for per-feed content extensions. + + Subclasses are auto-discovered from the ``EXTENSIONS_DIR`` directory. + Each instance processes one entry and returns template variables + that become available as ``{{variable_name}}`` in Discord messages + and embeds. + + Class attributes set by subclasses: + + ``name``: + Unique identifier (used in logs and per-feed enable/disable config). + Should be a short slug, e.g. ``"jwplayer_thumbnail"``. + + ``description``: + Human-readable description shown in the web UI (optional). + """ + + name: ClassVar[str] = "" + description: ClassVar[str] = "" + + #: List of template variable names this extension provides. + #: Used by the web UI to show users what ``{{variable}}`` tags + #: are available when an extension is enabled. + provides_variables: ClassVar[list[str]] = [] + + #: List of regex patterns. When a feed URL matches any pattern, + #: this extension is automatically enabled for that feed. + auto_enable_url_patterns: ClassVar[list[str]] = [] + + @abstractmethod + def process_entry(self, entry: Entry, reader: Reader) -> dict[str, str]: + """Return template variable pairs extracted from *entry*. + + Args: + entry: The feed entry to process. + reader: The reader instance (useful for loading feed tags). + + Returns: + A dict of ``{variable_name: value}`` pairs. Return an empty + dict if this entry does not need any extra variables. + """ + + def modify_webhook( + self, + webhook: DiscordWebhook, + _entry: Entry, + _reader: Reader, + ) -> DiscordWebhook: + """Modify the Discord webhook before it is sent. + + Override this method when your extension needs to change the + webhook payload itself (e.g. to add files, replace embeds, set + author information, etc.). By default it returns the webhook + unchanged. + + Args: + webhook: The fully built webhook payload. + _entry: The feed entry being processed. + _reader: The reader instance. + + Returns: + The (possibly modified) ``DiscordWebhook``. Return the + original *webhook* unchanged if no modifications are needed. + """ + return webhook + + @classmethod + def get_enabled_variables(cls, registry: dict[str, type[FeedExtension]], enabled_names: list[str]) -> list[str]: + """Return the union of ``provides_variables`` for all enabled extensions. + + Args: + registry: The extension registry ``{name: class}``. + enabled_names: List of enabled extension names for a feed. + + Returns: + Sorted list of unique variable names. + """ + seen: set[str] = set() + result: list[str] = [] + for name in enabled_names: + ext_cls = registry.get(name) + if ext_cls is None: + continue + for var_name in ext_cls.provides_variables: + if var_name not in seen: + seen.add(var_name) + result.append(var_name) + result.sort() + return result + + @classmethod + def matches_feed_url(cls, feed_url: str) -> bool: + """Return ``True`` if *feed_url* matches any of the auto-enable patterns. + + Args: + feed_url: The feed URL to test. + + Returns: + ``True`` if a match is found, ``False`` otherwise (including when + no patterns are defined). + """ + if not cls.auto_enable_url_patterns: + return False + + return any(re.search(p, feed_url) for p in cls.auto_enable_url_patterns) diff --git a/discord_rss_bot/extensions/discovery.py b/discord_rss_bot/extensions/discovery.py new file mode 100644 index 0000000..2d8a2b3 --- /dev/null +++ b/discord_rss_bot/extensions/discovery.py @@ -0,0 +1,191 @@ +"""Plugin discovery — import ``.py`` files from ``EXTENSIONS_DIR``. + +All ``FeedExtension`` subclasses found in those files are collected into +a global registry (``dict[str, type[FeedExtension]]``). +""" + +from __future__ import annotations + +import importlib.util +import logging +import os +import sys +from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + import types + +from discord_rss_bot.extensions.base import FeedExtension + +logger: logging.Logger = logging.getLogger(__name__) + +#: Environment variable that points to the extensions directory. +ENV_EXTENSIONS_DIR: str = "EXTENSIONS_DIR" + +#: Fallback if the env var is not set. +DEFAULT_EXTENSIONS_DIR: str = "extensions" + +_registry: dict[str, type[FeedExtension]] = {} +_discovered: bool = False + + +def _get_extensions_dir() -> Path: + """Return the path to the extensions directory. + + Respects the ``EXTENSIONS_DIR`` environment variable. Falls back to + ``extensions/`` relative to the current working directory. + """ + raw: str = os.getenv(ENV_EXTENSIONS_DIR, "").strip() + if raw: + return Path(raw).resolve() + return Path.cwd() / DEFAULT_EXTENSIONS_DIR + + +def _try_import_module(module_name: str, filepath: Path) -> types.ModuleType | None: + """Create a module spec and execute it, returning the module or ``None``. + + Args: + module_name: The name to give the imported module. + filepath: Absolute path to a ``.py`` file. + + Returns: + The imported module, or ``None`` if the spec could not be created. + """ + spec = importlib.util.spec_from_file_location(module_name, filepath) + if spec is None or spec.loader is None: + logger.warning("Could not create module spec for %s", filepath) + return None + + module = importlib.util.module_from_spec(spec) + # Make the parent package importable so relative imports in plugins work. + sys.modules[module_name] = module + spec.loader.exec_module(module) + return module + + +def _import_module_from_path(filepath: Path) -> types.ModuleType | None: + """Import a single ``.py`` file as a module and return the module object. + + Args: + filepath: Absolute path to a ``.py`` file. + + Returns: + The imported module, or ``None`` on failure. + """ + module_name: str = f"_ext_plugin_{filepath.stem}" + + # Avoid re-importing if already loaded. + if module_name in sys.modules: + return sys.modules[module_name] + + try: + return _try_import_module(module_name, filepath) + except Exception: + logger.exception("Failed to import extension plugin from %s", filepath) + return None + + +def _collect_extensions_from_module(module: types.ModuleType) -> list[type[FeedExtension]]: + """Return ``FeedExtension`` subclasses defined in *module*. + + Only concrete (non-ABC) subclasses with a non-empty ``name`` are + returned. + + Args: + module: An imported module object. + + Returns: + List of extension classes found in the module. + """ + extensions: list[type[FeedExtension]] = [] + for attr_name in dir(module): + obj = getattr(module, attr_name) + if ( + isinstance(obj, type) + and issubclass(obj, FeedExtension) + and obj is not FeedExtension + and not getattr(obj, "__abstractmethods__", None) + ): + name: str = getattr(obj, "name", "") or attr_name + extensions.append(obj) + logger.debug("Discovered extension %r (class %s)", name, attr_name) + return extensions + + +def _scan_directory(ext_dir: Path) -> None: + """Scan a single directory and register any ``FeedExtension`` subclasses found.""" + if not ext_dir.is_dir(): + return + + py_files: list[Path] = sorted(p for p in ext_dir.iterdir() if p.suffix == ".py" and p.stem != "__init__") + for filepath in py_files: + module = _import_module_from_path(filepath) + if module is None: + continue + classes = _collect_extensions_from_module(module) + for cls in classes: + cls_name: str = cls.name or cls.__name__ + if cls_name in _registry: + logger.warning("Duplicate extension name %r — overwriting with %s", cls_name, filepath) + _registry[cls_name] = cls + + +def discover_plugins(*, force: bool = False) -> dict[str, type[FeedExtension]]: + """Scan built-in and external extension directories and register. + + All ``FeedExtension`` subclasses found in those directories are + collected into a global registry. + + Order of discovery: + 1. Built-in extensions shipped with the bot (``discord_rss_bot/extensions/``) + 2. External user-provided plugins from ``EXTENSIONS_DIR`` + + External plugins can override built-in ones by using the same ``name``. + This is called automatically on first import. Call with + ``force=True`` to re-scan (useful in tests). + + Args: + force: If ``True``, re-scan even if already discovered. + + Returns: + The global registry ``{name: class}``. + """ + global _discovered # ruff:ignore[global-statement] + if _discovered and not force: + return _registry + + # 1. Scan the built-in package directory first. + built_in_dir: Path = Path(__file__).resolve().parent + logger.debug("Scanning built-in extensions directory: %s", built_in_dir) + _scan_directory(built_in_dir) + + # 2. Scan the external user-provided extensions directory. + ext_dir: Path = _get_extensions_dir() + if ext_dir != built_in_dir: + logger.info("Scanning external extensions directory: %s", ext_dir) + _scan_directory(ext_dir) + else: + logger.debug("External extensions directory is the same as built-in — skipping duplicate scan") + + _discovered = True + logger.info("Discovered %d extension plugin(s): %s", len(_registry), list(_registry.keys())) + return _registry + + +def get_registry() -> dict[str, type[FeedExtension]]: + """Return the current extension registry, discovering plugins if needed. + + Returns: + The global registry ``{name: class}``. + """ + if not _discovered: + return discover_plugins() + return _registry + + +def registry_clear() -> None: + """Clear the registry (useful for testing).""" + global _discovered, _registry # ruff:ignore[global-statement] + _registry = {} + _discovered = False diff --git a/discord_rss_bot/extensions/hoyolab.py b/discord_rss_bot/extensions/hoyolab.py new file mode 100644 index 0000000..0f1abb3 --- /dev/null +++ b/discord_rss_bot/extensions/hoyolab.py @@ -0,0 +1,331 @@ +"""Hoyolab feed extension: fetches full post data from the Hoyolab API. + +Detects feeds from ``feeds.c3kay.de`` and replaces the Discord embed +with richer content fetched directly from Hoyolab (Genshin Impact, +Honkai Starrail, Honkai Impact 3rd, Zenless Zone Zero). + +Usage: + 1. Enable ``hoyolab`` on the feed's Extensions page. + 2. The extension automatically decorates the embed with the post + title, author avatar, game color, images, and more. +""" + +from __future__ import annotations + +import contextlib +import json +import logging +import re +from typing import TYPE_CHECKING +from typing import ClassVar +from typing import cast + +import requests + +from discord_rss_bot.extensions.base import FeedExtension +from discord_rss_bot.webhook import DiscordEmbed +from discord_rss_bot.webhook import DiscordWebhook + +if TYPE_CHECKING: + from reader import Entry + from reader import Reader + +logger: logging.Logger = logging.getLogger(__name__) + +type JsonValue = bool | int | float | str | list[JsonValue] | dict[str, JsonValue] | None +type JsonObject = dict[str, JsonValue] + + +def is_c3kay_feed_url(feed_url: str) -> bool: + """Return whether a feed URL is from feeds.c3kay.de.""" + return "feeds.c3kay.de" in (feed_url or "") + + +def extract_post_id(url: str) -> str | None: + """Extract the Hoyolab post ID from a URL. + + Args: + url: The URL to inspect (e.g. ``https://www.hoyolab.com/article/38588239``). + + Returns: + The numeric post ID if found, otherwise ``None``. + """ + if not url: + return None + try: + match = re.search(r"/article/(\d+)", url) + if match: + return match.group(1) + except (ValueError, AttributeError, TypeError) as exc: + logger.warning("Error extracting post ID from URL %s: %s", url, exc) + return None + + +def _try_fetch_post(post_id: str) -> JsonObject | None: + """Attempt to fetch a Hoyolab post, returning the payload or ``None``. + + Does not catch exceptions — the caller handles them. + + Returns: + The post payload dict, or ``None`` on failure. + """ + url: str = f"https://bbs-api-os.hoyolab.com/community/post/wapi/getPostFull?post_id={post_id}" + response: requests.Response = requests.get(url, timeout=10) + if response.status_code == http_ok: + data = cast("JsonObject", response.json()) + data_payload: JsonObject = cast("JsonObject", data.get("data", {})) + post_payload: JsonObject = cast("JsonObject", data_payload.get("post", {})) + if data.get("retcode") == 0 and post_payload: + return post_payload + logger.warning("Failed to fetch Hoyolab post %s: %s", post_id, response.text) + return None + + +def fetch_post(post_id: str) -> JsonObject | None: + """Fetch post data from the Hoyolab API. + + Args: + post_id: The numeric post ID. + + Returns: + The post payload dict, or ``None`` on failure. + """ + if not post_id: + return None + try: + return _try_fetch_post(post_id) + except (requests.RequestException, ValueError): + logger.exception("Error fetching Hoyolab post %s", post_id) + return None + + +http_ok: int = 200 + + +def _as_json_object(value: JsonValue) -> JsonObject: + return cast("JsonObject", value) if isinstance(value, dict) else {} + + +class HoyolabExtension(FeedExtension): + """Fetches full post data from the Hoyolab API for c3kay.de feeds. + + Replaces the Discord embed with richer content (title, author avatar, + game color, images, video, and more). + """ + + name = "hoyolab" + description = ( + "Detects c3kay.de / Hoyolab feeds and replaces the Discord embed " + "with full post data from the Hoyolab API " + "(Genshin Impact, Honkai Starrail, Honkai Impact 3rd, ZZZ)." + ) + provides_variables: ClassVar[list[str]] = [ + "hoyolab_subject", + "hoyolab_description", + "hoyolab_image", + "hoyolab_author", + ] + auto_enable_url_patterns: ClassVar[list[str]] = [ + r"feeds\.c3kay\.de", + ] + + def process_entry(self, entry: Entry, reader: Reader) -> dict[str, str]: # ruff:ignore[unused-method-argument] + """Provide Hoyolab post data as template variables. + + Args: + entry: The feed entry to process. + reader: The reader instance. + + Returns: + Dict with ``hoyolab_subject``, ``hoyolab_description``, + ``hoyolab_image`` and ``hoyolab_author`` if the data was + fetched successfully, otherwise empty dict. + """ + if not is_c3kay_feed_url(entry.feed.url): + return {} + + post_id: str | None = extract_post_id(str(entry.link or "")) + if not post_id: + return {} + + post_data: JsonObject | None = fetch_post(post_id) + if not post_data: + return {} + + post: JsonObject = _as_json_object(post_data.get("post")) + subject: str = str(post.get("subject", "")) + description: str = str(post.get("desc", "")) + + image_url: str = "" + image_list = post_data.get("image_list", []) + if isinstance(image_list, list) and image_list: + first_image = image_list[0] + if isinstance(first_image, dict): + image_url = str(first_image.get("url", "")) + + author: str = "" + user: JsonObject = _as_json_object(post_data.get("user")) + if user: + author = str(user.get("nickname", "")) + + return { + "hoyolab_subject": subject, + "hoyolab_description": description, + "hoyolab_image": image_url, + "hoyolab_author": author, + } + + def modify_webhook( + self, + webhook: DiscordWebhook, + entry: Entry, + reader: Reader, # ruff:ignore[unused-method-argument] + ) -> DiscordWebhook: + """Replace the embed with richer Hoyolab content when applicable. + + Args: + webhook: The fully built webhook payload. + entry: The feed entry being processed. + reader: The reader instance. + + Returns: + The modified webhook with Hoyolab data. + """ + if not is_c3kay_feed_url(entry.feed.url): + return webhook + + # Don't interfere with non-embed delivery modes. + if not webhook.json.get("embeds"): + return webhook + + post_id: str | None = extract_post_id(str(entry.link or "")) + if not post_id: + return webhook + + post_data: JsonObject | None = fetch_post(post_id) + if not post_data: + return webhook + + discord_embed = self._build_embed_from_post(post_data, entry.link or entry.feed.url) + + self._attach_video_if_present(webhook, entry, post_data) + self._apply_author_from_post(webhook, post_data) + self._apply_structured_content(webhook, post_data) + + webhook.remove_embeds() + webhook.add_embed(discord_embed) + return webhook + + def _build_embed_from_post(self, post_data: JsonObject, entry_link: str) -> DiscordEmbed: + """Build a ``DiscordEmbed`` populated with post data. + + Args: + post_data: Raw Hoyolab post payload. + entry_link: Fallback link URL. + + Returns: + A populated embed. + """ + post: JsonObject = _as_json_object(post_data.get("post")) + subject: str = str(post.get("subject", "")) + content_raw: str = str(post.get("content", "{}")) + + content_data: JsonObject = {} + with contextlib.suppress(json.JSONDecodeError, ValueError): + loaded = cast("JsonValue", json.loads(content_raw)) + content_data = _as_json_object(loaded) + + description: str = str(content_data.get("describe", "")) + if not description: + description = str(post.get("desc", "")) + + discord_embed = DiscordEmbed() + discord_embed.set_title(subject) + discord_embed.set_url(entry_link) + + # Image list + image_list_value: JsonValue = post_data.get("image_list", []) + image_list: list[JsonObject] = ( + [cast("JsonObject", item) for item in image_list_value if isinstance(item, dict)] + if isinstance(image_list_value, list) + else [] + ) + if image_list: + discord_embed.set_image(url=str(image_list[0].get("url", ""))) + + # Game colour + game: JsonObject = _as_json_object(post_data.get("game")) + if game and game.get("color"): + discord_embed.set_color(str(game.get("color", "")).removeprefix("#")) + + # Footer + classification: JsonObject = _as_json_object(post_data.get("classification")) + if classification and classification.get("name"): + discord_embed.set_footer(text=str(classification.get("name", ""))) + + # Event dates + event_start: str = str(post.get("event_start_date", "")) + if event_start and event_start != "0": + discord_embed.add_embed_field(name="Start", value=f"") + event_end: str = str(post.get("event_end_date", "")) + if event_end and event_end != "0": + discord_embed.add_embed_field(name="End", value=f"") + + created_at: str = str(post.get("created_at", "")) + if created_at and created_at != "0": + discord_embed.set_timestamp(timestamp=created_at) + + if description: + discord_embed.set_description(description) + + return discord_embed + + def _attach_video_if_present(self, webhook: DiscordWebhook, entry: Entry, post_data: JsonObject) -> None: + """Download and attach a Hoyolab video to the webhook if present.""" + video: JsonObject = _as_json_object(post_data.get("video")) + if not video or not video.get("url"): + return + video_url: str = str(video.get("url", "")) + with contextlib.suppress(requests.RequestException): + video_response: requests.Response = requests.get(video_url, stream=True, timeout=10) + if video_response.ok: + webhook.add_file(file=video_response.content, filename=f"{entry.id}.mp4") + + def _apply_author_from_post(self, webhook: DiscordWebhook, post_data: JsonObject) -> None: + """Set webhook author (username and avatar) from post user data.""" + user: JsonObject = _as_json_object(post_data.get("user")) + author_name: str = str(user.get("nickname", "")) + avatar_url: str = str(user.get("avatar_url", "")) + if author_name: + webhook.avatar_url = avatar_url + webhook.username = author_name + + def _apply_structured_content(self, webhook: DiscordWebhook, post_data: JsonObject) -> None: + """Parse structured content for YouTube embeds and add them to the webhook.""" + post: JsonObject = _as_json_object(post_data.get("post")) + structured_content: str = str(post.get("structured_content", "")) + if not structured_content: + return + with contextlib.suppress(json.JSONDecodeError, ValueError): + loaded_structured = cast("JsonValue", json.loads(structured_content)) + structured_items: list[JsonObject] = ( + [cast("JsonObject", item) for item in loaded_structured if isinstance(item, dict)] + if isinstance(loaded_structured, list) + else [] + ) + for item in structured_items: + self._apply_structured_item(webhook, item) + + def _apply_structured_item(self, webhook: DiscordWebhook, item: JsonObject) -> None: + """Apply a single structured content item (e.g. YouTube embed).""" + insert: JsonObject = _as_json_object(item.get("insert")) + if not insert: + return + video_url_str: str = str(insert.get("video", "")) + if not video_url_str: + return + video_id_match = re.search(r"embed/([a-zA-Z0-9_-]+)", video_url_str) + if not video_id_match: + return + webhook.content = f"https://www.youtube.com/watch?v={video_id_match.group(1)}" + webhook.remove_embeds() diff --git a/discord_rss_bot/extensions/jwplayer_thumbnail.py b/discord_rss_bot/extensions/jwplayer_thumbnail.py new file mode 100644 index 0000000..f5dffa6 --- /dev/null +++ b/discord_rss_bot/extensions/jwplayer_thumbnail.py @@ -0,0 +1,285 @@ +"""Extension: extract JWPlayer thumbnail/video URLs from WordPress sites. + +`JWPlayer `_ is a commercial video player +used by many sites to serve self-hosted or CDN-hosted video. The player +is embedded via a `` + """ + entry: SimpleNamespace = SimpleNamespace( + id="test", + content=[SimpleNamespace(value=raw_html)], + summary="", + feed=SimpleNamespace(url="https://example.com/feed.xml"), + ) + result: dict[str, str] = ext.process_entry(entry, MagicMock()) # type: ignore[arg-type] + assert result.get("jwplayer_thumbnail") == "https://example.com/thumbnail.jpg" + assert result.get("jwplayer_file") == "https://example.com/video.mp4" + + +def test_jwplayer_thumbnail_extension_returns_empty_without_content() -> None: + """If the entry has no content, the extension should return an empty dict.""" + ext = JWPlayerThumbnailExtension() + entry: SimpleNamespace = SimpleNamespace( + id="test", + content=[], + summary="", + link="https://example.com/video", + feed=SimpleNamespace(url="https://example.com/feed.xml"), + ) + result: dict[str, str] = ext.process_entry(entry, MagicMock()) # type: ignore[arg-type] + assert result == {} + + +def test_jwplayer_thumbnail_extension_returns_empty_without_match() -> None: + """If the HTML has no JWPlayer pattern, the extension should return empty.""" + ext = JWPlayerThumbnailExtension() + raw_html: str = "

No player here.

" + entry: SimpleNamespace = SimpleNamespace( + id="test", + content=[SimpleNamespace(value=raw_html)], + summary="", + link="https://example.com/video", + feed=SimpleNamespace(url="https://example.com/feed.xml"), + ) + result: dict[str, str] = ext.process_entry(entry, MagicMock()) # type: ignore[arg-type] + assert result == {} + + +def test_jwplayer_thumbnail_extension_matches_hentaigasm_format() -> None: + """The extension should extract URLs from the actual hentaigasm.com feed format.""" + ext = JWPlayerThumbnailExtension() + # This is the actual HTML structure from the main hentaigasm feed. + raw_html: str = """ +

HENTAIGASM EXCLUSIVE!

+ +
+ + + +
+ +
+ +DOWNLOAD +""" + entry: SimpleNamespace = SimpleNamespace( + id="test-hentai", + content=[SimpleNamespace(value=raw_html)], + summary="", + feed=SimpleNamespace(url="https://hentaigasm.com/feed/"), + ) + result: dict[str, str] = ext.process_entry(entry, MagicMock()) # type: ignore[arg-type] + assert result.get("jwplayer_thumbnail") == "https://hgasm1.com/thumbnail/Test%20Video%201%20Subbed.jpg" + assert result.get("jwplayer_file") == "https://hgasm2.com/Test%20Video%201%20Subbed.mp4" + + +# --------------------------------------------------------------------------- +# Tests: auto_enable_url_patterns +# --------------------------------------------------------------------------- + + +def test_auto_enable_by_url_pattern_matches( + mock_reader: MagicMock, + temp_extensions_dir: str, +) -> None: + """An extension with matching auto_enable_url_patterns should be auto-enabled. + + Check that ``auto_enable_extensions_for_feed`` activates the plugin. + """ + # Register a test plugin with a URL pattern. + plugin_code: str = """ +from discord_rss_bot.extensions.base import FeedExtension + +class AutoPlugin(FeedExtension): + name = "auto_test" + auto_enable_url_patterns = [r"example\\.com/feeds"] + + def process_entry(self, entry, reader): + return {"auto_var": "enabled"} +""" + plugin_path: Path = Path(temp_extensions_dir) / "auto_plugin.py" + plugin_path.write_text(plugin_code, encoding="utf-8") + discover_plugins(force=True) + + mock_feed_url: str = "https://example.com/feeds/test.xml" + result: list[str] = auto_enable_extensions_for_feed(mock_reader, mock_feed_url) + + assert "auto_test" in result + + +def test_auto_enable_by_url_pattern_does_not_match( + mock_reader: MagicMock, + temp_extensions_dir: str, +) -> None: + """An extension whose pattern does not match should NOT be auto-enabled.""" + # Register the same plugin as the matching test, so this test actually + # validates that a non-matching URL does not trigger auto-enable. + plugin_code: str = """ +from discord_rss_bot.extensions.base import FeedExtension + +class AutoPlugin(FeedExtension): + name = "auto_test" + auto_enable_url_patterns = [r"example\\.com/feeds"] + + def process_entry(self, entry, reader): + return {"auto_var": "enabled"} +""" + plugin_path: Path = Path(temp_extensions_dir) / "auto_plugin.py" + plugin_path.write_text(plugin_code, encoding="utf-8") + discover_plugins(force=True) + + mock_feed_url: str = "https://other-site.com/feed.xml" + result: list[str] = auto_enable_extensions_for_feed(mock_reader, mock_feed_url) + + assert "auto_test" not in result + + +def test_auto_enable_does_not_duplicate_explicitly_enabled( + mock_reader: MagicMock, +) -> None: + """If an extension is already explicitly enabled, auto-enable should not add it again.""" + mock_feed_url: str = "https://example.com/feeds/test.xml" + # Explicitly enable the extension first. + set_enabled_extensions_for_feed(mock_reader, mock_feed_url, ["auto_test"]) + + result: list[str] = auto_enable_extensions_for_feed(mock_reader, mock_feed_url) + # Should contain auto_test only once. + assert result == ["auto_test"] + + +def test_matches_feed_url_classmethod() -> None: + """The ``matches_feed_url`` classmethod should work correctly.""" + assert FeedExtension.matches_feed_url("http://example.com") is False, "Base class has no patterns" + + +def test_steam_extension_has_auto_enable_patterns() -> None: + """The built-in Steam extension should declare URL patterns.""" + assert len(SteamExtension.auto_enable_url_patterns) > 0 + assert SteamExtension.matches_feed_url("https://store.steampowered.com/feeds/news/app/570/") + assert SteamExtension.matches_feed_url("https://steamcommunity.com/games/570/") + assert not SteamExtension.matches_feed_url("https://example.com/feed.xml") + + +def test_youtube_extension_has_auto_enable_patterns() -> None: + """The built-in YouTube extension should declare URL patterns.""" + assert len(YouTubeExtension.auto_enable_url_patterns) > 0 + assert YouTubeExtension.matches_feed_url("https://www.youtube.com/feeds/videos.xml?channel_id=123") + assert not YouTubeExtension.matches_feed_url("https://example.com/feed.xml") + + +def test_hoyolab_extension_has_auto_enable_patterns() -> None: + """The built-in Hoyolab extension should declare URL patterns.""" + assert len(HoyolabExtension.auto_enable_url_patterns) > 0 + assert HoyolabExtension.matches_feed_url("https://feeds.c3kay.de/hoyolab.xml") + assert not HoyolabExtension.matches_feed_url("https://example.com/feed.xml") + + +# --------------------------------------------------------------------------- +# Integration test with a real WordPress comment RSS feed (the original issue) +# --------------------------------------------------------------------------- + + +_COMMENTS_FEED_XML: str = r""" + + +Comments on: Test Article +https://example.com/test-article/ + +Sat, 18 Jul 2026 05:11:01 +0000 +hourly +1 +https://wordpress.org/?v=5.8.2 + + By: Anonymous + https://example.com/test-article/comment-page-1/#comment-820928 + + Sat, 18 Jul 2026 05:11:01 +0000 + https://example.com/?p=5633#comment-820928 + + Am I the only one that saw “Bookmark us” message?

]]> + + + By: Unknown + https://example.com/test-article/comment-page-1/#comment-820900 + + Fri, 17 Jul 2026 14:29:42 +0000 + https://example.com/?p=5633#comment-820900 + + Please uncensor the video

]]> + + + By: mid + https://example.com/test-article/comment-page-1/#comment-820887 + + Fri, 17 Jul 2026 06:20:56 +0000 + https://example.com/?p=5633#comment-820887 + + that’s not uncensored

]]> + + + +""" + + +class _CommentsFeedHandler(BaseHTTPRequestHandler): + """Serves the WordPress comment RSS XML on any request.""" + + def do_GET(self) -> None: + self.send_response(200) + self.send_header("Content-Type", "application/xml") + self.end_headers() + self.wfile.write(_COMMENTS_FEED_XML.encode("utf-8")) + + def log_message(self, _format: str, *args: str | int) -> None: + pass + + +@contextmanager +def _serve_feed() -> Iterator[str]: + """Start a local HTTP server serving the comments feed XML. + + Yields: + The URL of the feed. + """ + with ThreadingHTTPServer(("127.0.0.1", 0), _CommentsFeedHandler) as server: + server_thread = Thread(target=server.serve_forever, daemon=True) + server_thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}/feed.xml" + finally: + server.shutdown() + server_thread.join() + + +@pytest.mark.slow +def test_wordpress_comments_feed_pipeline() -> None: + """End-to-end test using the WordPress comment RSS feed from the original issue. + + Validates that: + 1. The feed XML parses correctly via the reader library + 2. Entry content and summary are populated as expected + 3. The jwplayer_thumbnail extension gracefully returns empty (no script tags) + 4. Tag replacement does not crash with comment feed data + """ + with ( + tempfile.TemporaryDirectory() as tmpdir, + _serve_feed() as feed_url, + ): + db_path: Path = Path(tmpdir) / "test.sqlite" + reader: ReaderType = make_reader(url=str(db_path)) + + try: + # Add and update the feed. + reader.add_feed(feed_url) + reader.update_feed(feed_url) + + feed = reader.get_feed(feed_url) + entries: list[Entry] = list(reader.get_entries(feed=feed_url)) + + # The comments feed should have 3 entries. + assert len(entries) == 3, f"Expected 3 comments, got {len(entries)}" + + for entry in entries: + # Each entry should have a title (comment author). + assert entry.title is not None + assert entry.title.startswith("By:") + + # Each entry should have a link back to the comment. + assert entry.link is not None + assert "comment-page" in entry.link + + # Summary should be plain text (from ), + # content should be HTML (from ). + assert entry.summary is not None + assert "

" not in entry.summary + if entry.content: + assert "

" in entry.content[0].value + + # The jwplayer_thumbnail extension should return empty + # because this feed has no JWPlayer script blocks. + ext = JWPlayerThumbnailExtension() + result: dict[str, str] = ext.process_entry(entry, reader) # type: ignore[arg-type] + assert result == {}, f"JWPlayer extension should return empty for comment feed, got {result}" + + # Tag replacement should not crash. + replaced_text: str = replace_tags_in_text_message(entry, reader) + assert isinstance(replaced_text, str) + + # Embed replacement should not crash. + replaced_embed = replace_tags_in_embed(feed, entry, reader) + assert replaced_embed is not None + + # Verify content:encoded was parsed into the content field. + first_entry = entries[0] + assert first_entry.content is not None + raw_content: str = first_entry.content[0].value + assert "Bookmark us" in raw_content or "bookmark" in raw_content.lower() + assert "

" in raw_content + + finally: + reader.close() + + +# --------------------------------------------------------------------------- +# Main feed XML (hentaigasm-style, with JWPlayer video entries) +# --------------------------------------------------------------------------- + +_MAIN_FEED_XML: str = """ + + +Hentaigasm - Test Feed +https://example.com + +Sun, 19 Jul 2026 04:20:40 +0000 + + Test Video 1 Subbed + https://example.com/test-video-1/ + + Sun, 19 Jul 2026 04:00:49 +0000 + https://example.com/?p=5501 + + HENTAIGASM EXCLUSIVE!

+ +
+ + + +
+ +
+ +]]> + + + +""" + + +class _MainFeedHandler(BaseHTTPRequestHandler): + """Serves the main feed XML with JWPlayer video entries.""" + + def do_GET(self) -> None: + self.send_response(200) + self.send_header("Content-Type", "application/xml") + self.end_headers() + self.wfile.write(_MAIN_FEED_XML.encode("utf-8")) + + def log_message(self, _format: str, *args: str | int) -> None: + pass + + +@contextmanager +def _serve_main_feed() -> Iterator[str]: + """Start a local HTTP server serving the main feed XML with JWPlayer content. + + Yields: + The URL of the feed. + """ + with ThreadingHTTPServer(("127.0.0.1", 0), _MainFeedHandler) as server: + server_thread = Thread(target=server.serve_forever, daemon=True) + server_thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}/feed.xml" + finally: + server.shutdown() + server_thread.join() + + +@pytest.mark.slow +def test_hentaigasm_main_feed_jwplayer_extraction() -> None: + """End-to-end test using a hentaigasm-style main feed with JWPlayer content. + + This test confirms that feedparser strips ``" + ), + }, + }, + { + "slug": "other-post", + "content": {"rendered": "

No player here.

"}, + }, + ]) + + class _APIHandler(BaseHTTPRequestHandler): + def do_GET(self) -> None: + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(wp_json.encode("utf-8")) + + def log_message(self, _format: str, *args: str | int) -> None: + pass + + with ThreadingHTTPServer(("127.0.0.1", 0), _APIHandler) as server: + server_thread = Thread(target=server.serve_forever, daemon=True) + server_thread.start() + port: int = server.server_port + + try: + entry = SimpleNamespace( + id="test-wp-batch", + content=[], + summary="", + link=f"http://127.0.0.1:{port}/test-slug/", + feed=SimpleNamespace(url=f"http://127.0.0.1:{port}/feed/"), + ) + + ext = JWPlayerThumbnailExtension() + result: dict[str, str] = ext.process_entry(entry, MagicMock()) # type: ignore[arg-type] + + assert result.get("jwplayer_thumbnail") == "https://cdn.example.com/thumb.jpg", ( + f"Batch WP fallback should extract thumbnail, got {result}" + ) + assert result.get("jwplayer_file") == "https://cdn.example.com/video.mp4", ( + f"Batch WP fallback should extract file, got {result}" + ) + + # Second entry from the same site uses cache — no API call. + entry2 = SimpleNamespace( + id="other-post", + content=[], + summary="", + link=f"http://127.0.0.1:{port}/other-post/", + feed=SimpleNamespace(url=f"http://127.0.0.1:{port}/feed/"), + ) + result2: dict[str, str] = ext.process_entry(entry2, MagicMock()) # type: ignore[arg-type] + assert result2 == {}, "Entry without player should return empty" + + finally: + server.shutdown() + server_thread.join() + _SLUG_CACHE.clear() + + +# --------------------------------------------------------------------------- +# Tests: WordPress extension +# --------------------------------------------------------------------------- + + +def test_wordpress_extension_uses_shared_batch_cache() -> None: + """The WordPress extension should use the shared batch cache.""" + _SLUG_CACHE.clear() + content_html: str = ( + "

Test

" + "" + ) + # Pre-populate the shared cache with the new richer format. + _SLUG_CACHE["https://example.com"] = { + "test-slug": { + "content": content_html, + "excerpt": "

Test excerpt

", + "title": "Test Post", + }, + } + + ext = WordPressExtension() + entry = SimpleNamespace( + id="test", + link="https://example.com/test-slug/", + feed=SimpleNamespace(url="https://example.com/feed/"), + ) + result = ext.process_entry(entry, MagicMock()) # type: ignore[arg-type] + assert result.get("wp_jwplayer_thumbnail") == "https://cdn.example.com/thumb.jpg" + assert result.get("wp_jwplayer_file") == "https://cdn.example.com/v.mp4" + assert result.get("wp_content_raw") == content_html + assert result.get("wp_content") is not None + assert "Test" in result.get("wp_content", "") + assert result.get("wp_excerpt_raw") == "

Test excerpt

" + assert result.get("wp_excerpt") is not None + assert "Test excerpt" in result.get("wp_excerpt", "") + # No spaces in URL values. + for key, val in result.items(): + if key.startswith("wp_jwplayer"): + assert " " not in val, f"URL should be encoded, got spaces: {val}" + _SLUG_CACHE.clear() + + +def test_wordpress_extension_provides_correct_variables() -> None: + """The WordPress extension should declare the correct variables.""" + expected: set[str] = { + "wp_content", + "wp_content_raw", + "wp_excerpt", + "wp_excerpt_raw", + "wp_jwplayer_file", + "wp_jwplayer_thumbnail", + } + assert set(WordPressExtension.provides_variables) == expected diff --git a/tests/test_feeds.py b/tests/test_feeds.py index 80fb70d..7e6fefe 100644 --- a/tests/test_feeds.py +++ b/tests/test_feeds.py @@ -22,6 +22,8 @@ from reader import StorageError from reader import make_reader from discord_rss_bot import feeds +from discord_rss_bot.extensions.steam import extract_app_id +from discord_rss_bot.extensions.youtube import is_youtube_feed_url from discord_rss_bot.feeds import JsonObject from discord_rss_bot.feeds import capture_full_page_screenshot from discord_rss_bot.feeds import create_feed @@ -31,13 +33,11 @@ from discord_rss_bot.feeds import extract_domain from discord_rss_bot.feeds import get_entry_delivery_mode from discord_rss_bot.feeds import get_screenshot_layout from discord_rss_bot.feeds import get_webhook_url -from discord_rss_bot.feeds import is_youtube_feed from discord_rss_bot.feeds import screenshot_filename_for_entry from discord_rss_bot.feeds import send_discord_quest_notification from discord_rss_bot.feeds import send_entry_to_discord from discord_rss_bot.feeds import send_to_discord from discord_rss_bot.feeds import set_entry_as_read -from discord_rss_bot.feeds import should_send_embed_check from discord_rss_bot.feeds import truncate_webhook_message @@ -119,57 +119,15 @@ def test_truncate_webhook_message_long_message(): def test_is_youtube_feed(): - """Test the is_youtube_feed function.""" + """Test the is_youtube_feed_url function.""" # YouTube feed URLs - assert is_youtube_feed("https://www.youtube.com/feeds/videos.xml?channel_id=123456") is True - assert is_youtube_feed("https://www.youtube.com/feeds/videos.xml?user=username") is True + assert is_youtube_feed_url("https://www.youtube.com/feeds/videos.xml?channel_id=123456") + assert is_youtube_feed_url("https://www.youtube.com/feeds/videos.xml?user=username") # Non-YouTube feed URLs - assert is_youtube_feed("https://www.example.com/feed.xml") is False - assert is_youtube_feed("https://www.youtube.com/watch?v=123456") is False - assert is_youtube_feed("https://www.reddit.com/r/Python/.rss") is False - - -@patch("discord_rss_bot.feeds.logger") -def test_should_send_embed_check_youtube_feeds(mock_logger: MagicMock) -> None: - """Test should_send_embed_check returns False for YouTube feeds regardless of settings.""" - # Create mocks - mock_reader = MagicMock() - mock_entry = MagicMock() - - # Configure a YouTube feed - mock_entry.feed.url = "https://www.youtube.com/feeds/videos.xml?channel_id=123456" - - # Set reader to return True for should_send_embed (would normally create an embed) - mock_reader.get_tag.return_value = True - - # Result should be False, overriding the feed settings - result = should_send_embed_check(mock_reader, mock_entry) - assert result is False, "YouTube feeds should never use embeds" - - # Function should not even call get_tag for YouTube feeds - mock_reader.get_tag.assert_not_called() - - -@patch("discord_rss_bot.feeds.logger") -def test_should_send_embed_check_normal_feeds(mock_logger: MagicMock) -> None: - """Test should_send_embed_check returns feed settings for non-YouTube feeds.""" - # Create mocks - mock_reader = MagicMock() - mock_entry = MagicMock() - - # Configure a normal feed - mock_entry.feed.url = "https://www.example.com/feed.xml" - - # Test with should_send_embed set to True - mock_reader.get_tag.return_value = True - result = should_send_embed_check(mock_reader, mock_entry) - assert result is True, "Normal feeds should use embeds when enabled" - - # Test with should_send_embed set to False - mock_reader.get_tag.return_value = False - result = should_send_embed_check(mock_reader, mock_entry) - assert result is False, "Normal feeds should not use embeds when disabled" + assert not is_youtube_feed_url("https://www.example.com/feed.xml") + assert not is_youtube_feed_url("https://www.youtube.com/watch?v=123456") + assert not is_youtube_feed_url("https://www.reddit.com/r/Python/.rss") def test_get_entry_delivery_mode_prefers_delivery_mode_tag() -> None: @@ -204,11 +162,7 @@ def test_get_entry_delivery_mode_falls_back_to_legacy_embed_flag() -> None: @patch("discord_rss_bot.feeds.execute_webhook") @patch("discord_rss_bot.feeds.create_text_webhook") -@patch("discord_rss_bot.feeds.create_hoyolab_webhook") -@patch("discord_rss_bot.feeds.fetch_hoyolab_post") def test_send_entry_to_discord_hoyolab_text_mode_uses_text_webhook( - mock_fetch_hoyolab_post: MagicMock, - mock_create_hoyolab_webhook: MagicMock, mock_create_text_webhook: MagicMock, mock_execute_webhook: MagicMock, ) -> None: @@ -230,8 +184,6 @@ def test_send_entry_to_discord_hoyolab_text_mode_uses_text_webhook( result = send_entry_to_discord(entry, reader) assert result is None - mock_fetch_hoyolab_post.assert_not_called() - mock_create_hoyolab_webhook.assert_not_called() mock_create_text_webhook.assert_called_once_with( "https://discord.test/webhook", entry, @@ -243,11 +195,7 @@ def test_send_entry_to_discord_hoyolab_text_mode_uses_text_webhook( @patch("discord_rss_bot.feeds.execute_webhook") @patch("discord_rss_bot.feeds.create_screenshot_webhook") -@patch("discord_rss_bot.feeds.create_hoyolab_webhook") -@patch("discord_rss_bot.feeds.fetch_hoyolab_post") def test_send_entry_to_discord_hoyolab_screenshot_mode_uses_screenshot_webhook( - mock_fetch_hoyolab_post: MagicMock, - mock_create_hoyolab_webhook: MagicMock, mock_create_screenshot_webhook: MagicMock, mock_execute_webhook: MagicMock, ) -> None: @@ -269,8 +217,6 @@ def test_send_entry_to_discord_hoyolab_screenshot_mode_uses_screenshot_webhook( result = send_entry_to_discord(entry, reader) assert result is None - mock_fetch_hoyolab_post.assert_not_called() - mock_create_hoyolab_webhook.assert_not_called() mock_create_screenshot_webhook.assert_called_once_with( "https://discord.test/webhook", entry, @@ -281,14 +227,11 @@ def test_send_entry_to_discord_hoyolab_screenshot_mode_uses_screenshot_webhook( @patch("discord_rss_bot.feeds.execute_webhook") @patch("discord_rss_bot.feeds.create_embed_webhook") -@patch("discord_rss_bot.feeds.create_hoyolab_webhook") -@patch("discord_rss_bot.feeds.fetch_hoyolab_post") -def test_send_entry_to_discord_hoyolab_embed_mode_uses_hoyolab_webhook( - mock_fetch_hoyolab_post: MagicMock, - mock_create_hoyolab_webhook: MagicMock, +def test_send_entry_to_discord_hoyolab_embed_mode_uses_embed_webhook( mock_create_embed_webhook: MagicMock, mock_execute_webhook: MagicMock, ) -> None: + """Embed mode should use the standard embed pipeline, not a Hoyolab bypass.""" entry = MagicMock() entry.id = "entry-3" entry.feed.url = "https://feeds.c3kay.de/hoyolab.xml" @@ -301,21 +244,14 @@ def test_send_entry_to_discord_hoyolab_embed_mode_uses_hoyolab_webhook( "delivery_mode": "embed", }.get(key, default) - mock_fetch_hoyolab_post.return_value = {"post": {"subject": "News"}} - hoyolab_webhook = MagicMock() - mock_create_hoyolab_webhook.return_value = hoyolab_webhook + embed_webhook = MagicMock() + mock_create_embed_webhook.return_value = embed_webhook result = send_entry_to_discord(entry, reader) assert result is None - mock_fetch_hoyolab_post.assert_called_once_with("38588239") - mock_create_hoyolab_webhook.assert_called_once_with( - "https://discord.test/webhook", - entry, - {"post": {"subject": "News"}}, - ) - mock_create_embed_webhook.assert_not_called() - mock_execute_webhook.assert_called_once_with(hoyolab_webhook, entry, reader=reader) + mock_create_embed_webhook.assert_called_once() + mock_execute_webhook.assert_called_once_with(embed_webhook, entry, reader=reader) def test_get_screenshot_layout_prefers_mobile_tag() -> None: @@ -391,7 +327,7 @@ def test_get_feed_media_gallery_image_limit_defaults_to_first_image() -> None: ], ) def test_extract_steam_app_id_from_url(url: str, expected_app_id: str | None) -> None: - assert feeds.extract_steam_app_id_from_url(url) == expected_app_id + assert extract_app_id(url) == expected_app_id @pytest.mark.parametrize( @@ -942,102 +878,6 @@ def test_create_embed_webhook_can_use_steam_game_icon_thumbnail( entry.summary = "" entry.content = [] entry.feed.url = "https://store.steampowered.com/feeds/news/app/570/?cc=us&l=english" - mock_replace_tags_in_embed.return_value = feeds.CustomEmbed( - description="Steam news", - thumbnail_url="https://example.com/custom-thumb.jpg", - show_steam_game_icon_in_thumbnail=True, - ) - - with patch("discord_rss_bot.feeds.Path.is_file", return_value=False): - webhook = feeds.create_embed_webhook("https://discord.com/api/webhooks/123/abc", entry, reader) - - assert "components" not in webhook.json - embeds = webhook.json.get("embeds") - assert isinstance(embeds, list) - assert isinstance(embeds[0], dict) - assert embeds[0]["thumbnail"] == { - "url": "https://cdn.cloudflare.steamstatic.com/steam/apps/570/capsule_sm_120.jpg", - } - assert webhook.files == [] - mock_fetch_ttvdrops_campaign_media_items.assert_not_called() - - -@patch("discord_rss_bot.feeds.fetch_ttvdrops_campaign_media_items", return_value=[]) -@patch("discord_rss_bot.feeds.replace_tags_in_embed") -def test_create_embed_webhook_prefers_local_steam_game_icon_thumbnail( - mock_replace_tags_in_embed: MagicMock, - mock_fetch_ttvdrops_campaign_media_items: MagicMock, -) -> None: - local_icon_bytes = b"local-steam-icon" - - reader = MagicMock() - reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument] - "media_gallery_image_limit": 0, - "webhook_text_length_limit": 4000, - }.get(key, default) - entry = MagicMock() - entry.id = "entry-steam-local-1" - entry.title = "Dota 2 patch notes" - entry.link = "https://steamcommunity.com/games/570/announcements/detail/1234567890" - entry.summary = "" - entry.content = [] - entry.feed.url = "https://store.steampowered.com/feeds/news/app/570/?cc=us&l=english" - mock_replace_tags_in_embed.return_value = feeds.CustomEmbed( - description="Steam news", - thumbnail_url="https://example.com/custom-thumb.jpg", - show_steam_game_icon_in_thumbnail=True, - ) - - with ( - patch("discord_rss_bot.feeds.Path.is_file", return_value=True), - patch("discord_rss_bot.feeds.Path.read_bytes", return_value=local_icon_bytes), - ): - webhook = feeds.create_embed_webhook("https://discord.com/api/webhooks/123/abc", entry, reader) - - embeds = webhook.json.get("embeds") - assert isinstance(embeds, list) - assert isinstance(embeds[0], dict) - assert len(webhook.files) == 1 - uploaded_icon = webhook.files[0] - assert uploaded_icon.content == local_icon_bytes - assert uploaded_icon.filename.startswith("steam-app-570-") - assert uploaded_icon.filename.endswith(".png") - assert embeds[0]["thumbnail"] == {"url": f"attachment://{uploaded_icon.filename}"} - mock_fetch_ttvdrops_campaign_media_items.assert_not_called() - - -@patch("discord_rss_bot.feeds.fetch_ttvdrops_campaign_media_items", return_value=[]) -@patch("discord_rss_bot.feeds.replace_tags_in_embed") -def test_create_embed_webhook_does_not_inject_steam_thumbnail_when_app_id_is_missing( - mock_replace_tags_in_embed: MagicMock, - mock_fetch_ttvdrops_campaign_media_items: MagicMock, -) -> None: - reader = MagicMock() - reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument] - "media_gallery_image_limit": 0, - "webhook_text_length_limit": 4000, - }.get(key, default) - entry = MagicMock() - entry.id = "entry-steam-2" - entry.title = "Steam group post" - entry.link = "https://steamcommunity.com/groups/example/announcements/detail/1234567890" - entry.summary = "" - entry.content = [] - entry.feed.url = "https://steamcommunity.com/groups/example/rss/" - mock_replace_tags_in_embed.return_value = feeds.CustomEmbed( - description="Steam group news", - thumbnail_url="https://example.com/custom-thumb.jpg", - show_steam_game_icon_in_thumbnail=True, - ) - - webhook = feeds.create_embed_webhook("https://discord.com/api/webhooks/123/abc", entry, reader) - - assert "components" not in webhook.json - embeds = webhook.json.get("embeds") - assert isinstance(embeds, list) - assert isinstance(embeds[0], dict) - assert "thumbnail" not in embeds[0] - mock_fetch_ttvdrops_campaign_media_items.assert_not_called() @patch("discord_rss_bot.feeds.fetch_ttvdrops_campaign_media_items", return_value=[]) @@ -1338,10 +1178,11 @@ def test_send_entry_to_discord_youtube_feed( mock_entry.feed.url = "https://www.youtube.com/feeds/videos.xml?channel_id=123456" mock_entry.feed_url = "https://www.youtube.com/feeds/videos.xml?channel_id=123456" - # Mock the tags + # Mock the tags — with no delivery_mode tag and should_send_embed=True, + # the entry delivery mode will be "embed", which calls create_embed_webhook. mock_reader.get_tag.side_effect = lambda feed, tag, default=None: { # ruff:ignore[unused-lambda-argument] "webhook": "https://discord.com/api/webhooks/123/abc", - "should_send_embed": True, # This should be ignored for YouTube feeds + "should_send_embed": True, }.get(tag, default) # Mock custom message @@ -1355,23 +1196,19 @@ def test_send_entry_to_discord_youtube_feed( # Call the function send_entry_to_discord(mock_entry, mock_reader) - # Assertions - mock_create_embed.assert_not_called() - mock_discord_webhook.assert_called_once() - - # Check webhook was created with the right message - webhook_call_kwargs = mock_discord_webhook.call_args[1] - assert "content" in webhook_call_kwargs, "Webhook should have content" - assert webhook_call_kwargs["url"] == "https://discord.com/api/webhooks/123/abc" + # Assertions — YouTube feeds now follow standard delivery mode. + # With should_send_embed=True, embed mode is used. + mock_create_embed.assert_called_once() + mock_discord_webhook.assert_not_called() # Verify execute_webhook was called - mock_execute_webhook.assert_called_once_with(mock_webhook, mock_entry, reader=mock_reader) + mock_execute_webhook.assert_called_once() def test_extract_domain_youtube_feed() -> None: """Test extract_domain for YouTube feeds.""" url: str = "https://www.youtube.com/feeds/videos.xml?channel_id=123456" - assert extract_domain(url) == "YouTube", "YouTube feeds should return 'YouTube' as the domain." + assert extract_domain(url) == "YouTube" def test_extract_domain_reddit_feed() -> None: diff --git a/tests/test_hoyolab_api.py b/tests/test_hoyolab_api.py deleted file mode 100644 index 0649578..0000000 --- a/tests/test_hoyolab_api.py +++ /dev/null @@ -1,248 +0,0 @@ -from __future__ import annotations - -import json -import typing -from types import SimpleNamespace -from unittest.mock import MagicMock -from unittest.mock import patch - -import pytest -import requests - -from discord_rss_bot.hoyolab_api import create_hoyolab_webhook -from discord_rss_bot.hoyolab_api import extract_post_id_from_hoyolab_url -from discord_rss_bot.hoyolab_api import fetch_hoyolab_post -from discord_rss_bot.hoyolab_api import is_c3kay_feed - -if typing.TYPE_CHECKING: - from reader import Entry - - -class TestExtractPostIdFromHoyolabUrl: - def test_extract_post_id_from_article_url(self) -> None: - """Test extracting post ID from a direct article URL.""" - test_cases: list[str] = [ - "https://www.hoyolab.com/article/38588239", - "http://hoyolab.com/article/12345", - "https://www.hoyolab.com/article/987654321/comments", - ] - - expected_ids: list[str] = ["38588239", "12345", "987654321"] - - for url, expected_id in zip(test_cases, expected_ids, strict=False): - assert extract_post_id_from_hoyolab_url(url) == expected_id - - def test_url_without_post_id(self) -> None: - """Test with a URL that doesn't have a post ID.""" - test_cases: list[str] = [ - "https://www.hoyolab.com/community", - ] - - for url in test_cases: - assert extract_post_id_from_hoyolab_url(url) is None - - def test_edge_cases(self) -> None: - """Test edge cases like None, empty string, and malformed URLs.""" - test_cases: list[str | None] = [ - None, - "", - "not_a_url", - "http:/", # Malformed URL - ] - - for url in test_cases: - assert extract_post_id_from_hoyolab_url(url) is None # type: ignore - - -def make_entry(link: str | None = "https://www.hoyolab.com/article/38588239") -> SimpleNamespace: - feed: SimpleNamespace = SimpleNamespace(url="https://feeds.c3kay.de/hoyolab.xml") - return SimpleNamespace( - id="entry-123", - link=link, - feed=feed, - ) - - -class TestIsC3KayFeed: - def test_true_for_c3kay_feed(self) -> None: - assert is_c3kay_feed("https://feeds.c3kay.de/rss") is True - - def test_false_for_non_c3kay_feed(self) -> None: - assert is_c3kay_feed("https://example.com/rss") is False - - -class TestFetchHoyolabPost: - @patch("discord_rss_bot.hoyolab_api.requests.get") - def test_returns_none_for_empty_post_id(self, mock_get: MagicMock) -> None: - assert fetch_hoyolab_post("") is None - mock_get.assert_not_called() - - @patch("discord_rss_bot.hoyolab_api.requests.get") - def test_returns_post_data_for_success_response(self, mock_get: MagicMock) -> None: - mock_response = MagicMock() - mock_response.status_code = 200 - mock_response.json.return_value = { - "retcode": 0, - "data": { - "post": { - "post_id": "38588239", - "subject": "Event", - }, - }, - } - mock_get.return_value = mock_response - - result = fetch_hoyolab_post("38588239") - - assert result == {"post_id": "38588239", "subject": "Event"} - assert mock_get.call_args.args[0].endswith("post_id=38588239") - - @patch("discord_rss_bot.hoyolab_api.logger") - @patch("discord_rss_bot.hoyolab_api.requests.get") - def test_returns_none_and_logs_warning_for_non_success_payload( - self, - mock_get: MagicMock, - mock_logger: MagicMock, - ) -> None: - mock_response = MagicMock() - mock_response.status_code = 200 - mock_response.text = "bad payload" - mock_response.json.return_value = { - "retcode": -1, - "data": {}, - } - mock_get.return_value = mock_response - - result = fetch_hoyolab_post("38588239") - - assert result is None - mock_logger.warning.assert_called_once() - - @patch("discord_rss_bot.hoyolab_api.logger") - @patch("discord_rss_bot.hoyolab_api.requests.get") - def test_returns_none_and_logs_exception_on_request_error( - self, - mock_get: MagicMock, - mock_logger: MagicMock, - ) -> None: - mock_get.side_effect = requests.RequestException("network issue") - - result = fetch_hoyolab_post("38588239") - - assert result is None - mock_logger.exception.assert_called_once() - - -class TestCreateHoyolabWebhook: - @patch("discord_rss_bot.hoyolab_api.requests.get") - @patch("discord_rss_bot.hoyolab_api.DiscordEmbed") - @patch("discord_rss_bot.hoyolab_api.DiscordWebhook") - def test_builds_embed_webhook_with_full_post_data( - self, - mock_webhook_cls: MagicMock, - mock_embed_cls: MagicMock, - mock_requests_get: MagicMock, - ) -> None: - webhook_instance = MagicMock() - embed_instance = MagicMock() - mock_webhook_cls.return_value = webhook_instance - mock_embed_cls.return_value = embed_instance - - video_response = MagicMock() - video_response.ok = True - video_response.content = b"video-bytes" - mock_requests_get.return_value = video_response - - post_data = { - "post": { - "subject": "Update 4.0", - "content": json.dumps({"describe": "Patch notes"}), - "desc": "fallback description", - "structured_content": json.dumps( - [{"insert": {"video": "https://www.youtube.com/embed/abc123_XY"}}], - ), - "event_start_date": "1712000000", - "event_end_date": "1712600000", - "created_at": "1711000000", - }, - "image_list": [{"url": "https://img.example.com/hero.jpg", "height": 1080, "width": 1920}], - "video": {"url": "https://cdn.example.com/video.mp4"}, - "game": {"color": "#11AAFF"}, - "user": {"nickname": "Paimon", "avatar_url": "https://img.example.com/avatar.jpg"}, - "classification": {"name": "Official"}, - } - - entry = make_entry(link=None) - entry = typing.cast("Entry", entry) - webhook = create_hoyolab_webhook("https://discord.test/webhook", entry, post_data) # type: ignore - - assert webhook is webhook_instance - mock_webhook_cls.assert_called_once_with(url="https://discord.test/webhook", rate_limit_retry=True) - - embed_instance.set_title.assert_called_once_with("Update 4.0") - embed_instance.set_url.assert_called_once_with("https://feeds.c3kay.de/hoyolab.xml") - embed_instance.set_image.assert_called_once_with( - url="https://img.example.com/hero.jpg", - height=1080, - width=1920, - ) - embed_instance.set_color.assert_called_once_with("11AAFF") - embed_instance.set_footer.assert_called_once_with(text="Official") - embed_instance.add_embed_field.assert_any_call(name="Start", value="") - embed_instance.add_embed_field.assert_any_call(name="End", value="") - embed_instance.set_timestamp.assert_called_once_with(timestamp="1711000000") - - webhook_instance.add_file.assert_called_once_with(file=b"video-bytes", filename="entry-123.mp4") - webhook_instance.add_embed.assert_called_once_with(embed_instance) - assert webhook_instance.content == "https://www.youtube.com/watch?v=abc123_XY" - webhook_instance.remove_embeds.assert_called_once() - - @patch("discord_rss_bot.hoyolab_api.requests.get") - @patch("discord_rss_bot.hoyolab_api.DiscordEmbed") - @patch("discord_rss_bot.hoyolab_api.DiscordWebhook") - def test_handles_invalid_structured_content_without_removing_embeds( - self, - mock_webhook_cls: MagicMock, - mock_embed_cls: MagicMock, - mock_requests_get: MagicMock, - ) -> None: - webhook_instance = MagicMock() - embed_instance = MagicMock() - mock_webhook_cls.return_value = webhook_instance - mock_embed_cls.return_value = embed_instance - mock_requests_get.return_value = MagicMock(ok=False) - - post_data = { - "post": { - "subject": "News", - "content": "{}", - "structured_content": "not-json", - }, - } - - entry = make_entry() - entry = typing.cast("Entry", entry) - webhook = create_hoyolab_webhook("https://discord.test/webhook", entry, post_data) # type: ignore - - assert webhook is webhook_instance - webhook_instance.remove_embeds.assert_not_called() - - -def test_extract_post_id_with_querystring() -> None: - url = "https://www.hoyolab.com/article/38588239?utm_source=feed" - assert extract_post_id_from_hoyolab_url(url) == "38588239" - - -def test_extract_post_id_non_string_input_returns_none() -> None: - assert extract_post_id_from_hoyolab_url(None) is None # type: ignore[arg-type] - - -@pytest.mark.parametrize( - ("url", "expected"), - [ - ("https://feeds.c3kay.de/rss", True), - ("https://www.hoyolab.com/feed", False), - ], -) -def test_is_c3kay_feed_parametrized(*, url: str, expected: bool) -> None: - assert is_c3kay_feed(url) is expected diff --git a/tests/test_main.py b/tests/test_main.py index ee1b12a..967ffd3 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -32,6 +32,8 @@ if TYPE_CHECKING: from reader import Entry from reader import Reader + from discord_rss_bot.feeds import JsonValue + client: TestClient = TestClient(app) webhook_name: str = "Hello, I am a webhook!" webhook_url: str = "https://discord.com/api/webhooks/1234567890/abcdefghijklmnopqrstuvwxyz" @@ -318,7 +320,7 @@ def test_feed_page_shows_steam_thumbnail_hint_for_steam_feeds() -> None: assert feed_url == self.feed.url return self.feed - def get_tag(self, _resource: object, key: str, default: TestTagValue = None) -> TestTagValue: + def get_tag(self, _resource: str | DummyFeed, key: str, default: TestTagValue = None) -> TestTagValue: return { "webhooks": [], "delivery_mode": "embed", @@ -2645,7 +2647,7 @@ def test_export_opml_with_stub_reader() -> None: filename="reader-feeds-2026-07-18-12-00-00.opml", ) - def get_tag(self, _resource: tuple, _key: str, default: object = None) -> object: + def get_tag(self, _resource: tuple[()], _key: str, default: TestTagValue = None) -> TestTagValue: return default def get_feeds(self) -> list: @@ -2672,7 +2674,7 @@ class StubImportConfirmReader: self.added_urls: list[str] = [] self.tags: dict[tuple[str, str], str] = {} - def get_tag(self, _resource: tuple, _key: str, default: object = None) -> object: + def get_tag(self, _resource: tuple[()], _key: str, default: TestTagValue = None) -> TestTagValue: """Stub get_tag. Returns: @@ -2705,7 +2707,7 @@ class StubImportConfirmReaderWithExisting: self.added_urls: list[str] = [] self.tags: dict[tuple[str, str], str] = {} - def get_tag(self, _resource: tuple, _key: str, default: object = None) -> object: + def get_tag(self, _resource: tuple[()], _key: str, default: TestTagValue = None) -> TestTagValue: """Stub get_tag. Returns: @@ -2857,7 +2859,6 @@ def test_post_embed_saves_all_fields() -> None: "thumbnail_url": "https://example.com/thumb.png", "footer_text": "Footer Text", "footer_icon_url": "https://example.com/footer.png", - "show_steam_game_icon_in_thumbnail": "true", }, follow_redirects=False, ) @@ -2880,7 +2881,6 @@ def test_post_embed_saves_all_fields() -> None: assert saved["thumbnail_url"] == "https://example.com/thumb.png" assert saved["footer_text"] == "Footer Text" assert saved["footer_icon_url"] == "https://example.com/footer.png" - assert saved["show_steam_game_icon_in_thumbnail"] is True finally: app.dependency_overrides = {} @@ -2958,7 +2958,6 @@ def test_post_embed_allows_clearing_all_fields() -> None: "thumbnail_url": "https://old.example.com/thumb.png", "footer_text": "Old Footer", "footer_icon_url": "https://old.example.com/footer.png", - "show_steam_game_icon_in_thumbnail": True, }) stub = _make_stub_reader_for_embed(stored_embed=existing) app.dependency_overrides[get_reader_dependency] = lambda: stub @@ -2998,7 +2997,6 @@ def test_post_embed_allows_clearing_all_fields() -> None: assert not saved["thumbnail_url"] assert not saved["footer_text"] assert not saved["footer_icon_url"] - assert saved["show_steam_game_icon_in_thumbnail"] is False finally: app.dependency_overrides = {} @@ -3016,7 +3014,6 @@ def test_post_embed_untouched_fields_retain_values() -> None: "thumbnail_url": "", "footer_text": "Old Footer", "footer_icon_url": "", - "show_steam_game_icon_in_thumbnail": False, }) stub = _make_stub_reader_for_embed(stored_embed=existing) app.dependency_overrides[get_reader_dependency] = lambda: stub @@ -3056,7 +3053,6 @@ def test_post_embed_untouched_fields_retain_values() -> None: assert saved["author_name"] == "Author" assert saved["author_url"] == "https://a.example.com" assert saved["footer_text"] == "Old Footer" - assert saved["show_steam_game_icon_in_thumbnail"] is False finally: app.dependency_overrides = {} @@ -3094,7 +3090,6 @@ def test_post_embed_saves_empty_description_when_no_prior_embed_exists() -> None saved: dict[str, str] = json.loads(json_arg) assert saved["title"] == "Just a Title" assert not saved["description"], f"Expected empty description, got {saved['description']!r}" - assert saved["show_steam_game_icon_in_thumbnail"] is False finally: app.dependency_overrides = {} @@ -3412,7 +3407,7 @@ class _StubReaderForMass: return self._webhook_url return default - def set_tag(self, resource: str, key: str, value: object) -> None: # pyright: ignore[reportArgumentType] + def set_tag(self, resource: str, key: str, value: JsonValue) -> None: """Record tag set calls.""" self.set_tag_calls.append((resource, key, value)) diff --git a/tests/test_settings.py b/tests/test_settings.py index 3dcddf0..1815e0c 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -39,7 +39,7 @@ class _AutodiscoverHandler(BaseHTTPRequestHandler): b'type="application/rss+xml" title="Example">', ) - def log_message(self, _format: str, *_args: object) -> None: + def log_message(self, _format: str, *_args: str | int) -> None: """Suppress HTTP request logging during tests.""" diff --git a/tests/test_whitelist.py b/tests/test_whitelist.py index 360d688..e235ee5 100644 --- a/tests/test_whitelist.py +++ b/tests/test_whitelist.py @@ -11,8 +11,8 @@ from reader import make_reader from discord_rss_bot.filter.evaluator import evaluate_entry_filters from discord_rss_bot.filter.evaluator import get_filter_values_from_reader -from discord_rss_bot.filter.whitelist import has_white_tags -from discord_rss_bot.filter.whitelist import should_be_sent +from discord_rss_bot.filter.evaluator import has_white_tags +from discord_rss_bot.filter.evaluator import should_be_sent if TYPE_CHECKING: from collections.abc import Iterable