Compare commits

..

6 commits

Author SHA1 Message Date
d61e8ccf10
Only show screenshot things if Playwright has chromium installed; use names instead of ids when ignoring Ruff errors
All checks were successful
Test and build Docker image / docker (push) Successful in 1m54s
2026-07-18 07:04:00 +02:00
59d9062ded
Made the feed config page less shit 2026-07-18 06:25:29 +02:00
35ede5aa33
Allow newlines in embeds 2026-07-18 06:04:29 +02:00
5ead9742e7
Fix syntax error in docker-compose.yml 2026-07-18 04:50:52 +02:00
46d84ed308
Add more tests 2026-07-18 04:40:13 +02:00
LostOnTheLine
f3989eefa9
Add options for custom Username & Avatar Image 2026-07-18 04:08:34 +02:00
20 changed files with 989 additions and 490 deletions

View file

@ -38,7 +38,7 @@ repos:
# An extremely fast Python linter and formatter.
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.15.16
rev: v0.15.22
hooks:
- id: ruff-format
types_or: [ python, pyi, jupyter, pyproject ]

View file

@ -25,6 +25,11 @@ logger: logging.Logger = logging.getLogger(__name__)
DISCORD_TIMESTAMP_TAG_RE: re.Pattern[str] = re.compile(r"<t:\d+(?::[tTdDfFrRsS])?>")
# 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("@#:`")
DISCORD_WEBHOOK_USERNAME_FORBIDDEN_SUBSTRINGS: tuple[str, ...] = ("clyde", "discord")
@dataclass(slots=True)
class CustomEmbed:
@ -313,7 +318,7 @@ def replace_tags_in_embed(feed: Feed, entry: Entry, reader: Reader) -> CustomEmb
entry_updated: str = entry.updated.strftime("%Y-%m-%d %H:%M:%S") if entry.updated else "Never"
if embed.title and not embed.author_name and embed.author_url:
msg = "You are using author_url without author_name, but has title set. We will use author_name instead of title when sending the embed to Discord." # noqa: E501
msg = "You are using author_url without author_name, but has title set. We will use author_name instead of title when sending the embed to Discord." # ruff:ignore[line-too-long]
logger.info(msg)
embed.author_name = embed.title
embed.title = ""
@ -351,6 +356,17 @@ def replace_tags_in_embed(feed: Feed, entry: Entry, reader: Reader) -> CustomEmb
for replacement in list_of_replacements:
for template, replace_with in replacement.items():
_replace_embed_tags(embed, template, replace_with)
embed.title = embed.title.replace("\\n", "\n")
embed.description = embed.description.replace("\\n", "\n")
embed.author_name = embed.author_name.replace("\\n", "\n")
embed.author_url = embed.author_url.replace("\\n", "\n")
embed.author_icon_url = embed.author_icon_url.replace("\\n", "\n")
embed.image_url = embed.image_url.replace("\\n", "\n")
embed.thumbnail_url = embed.thumbnail_url.replace("\\n", "\n")
embed.footer_text = embed.footer_text.replace("\\n", "\n")
embed.footer_icon_url = embed.footer_icon_url.replace("\\n", "\n")
return embed
@ -391,6 +407,93 @@ def get_custom_message(reader: Reader, feed: Feed) -> str:
return custom_message
def get_message_username(reader: Reader, feed: Feed) -> str:
"""Get the stored custom webhook username for a feed.
Returns:
Stored username (may be empty or invalid for Discord).
"""
try:
return str(reader.get_tag(feed, "message_username", ""))
except ValueError:
return ""
def get_message_avatar_url(reader: Reader, feed: Feed) -> str:
"""Get the stored custom webhook avatar URL for a feed.
Returns:
Stored avatar URL (may be empty or invalid for Discord).
"""
try:
return str(reader.get_tag(feed, "message_avatar_url", ""))
except ValueError:
return ""
def normalize_message_username(username: str | None) -> str:
"""Return a Discord-safe webhook username, or empty string if unusable.
Blank or invalid values are rejected so Discord uses the webhook default.
Returns:
Valid username, or empty string when the override should not be sent.
"""
if not username:
return ""
cleaned: str = username.strip()
if not cleaned:
return ""
if len(cleaned) > DISCORD_WEBHOOK_USERNAME_MAX_LENGTH:
return ""
if any(character in cleaned for character in DISCORD_WEBHOOK_USERNAME_FORBIDDEN_CHARS):
return ""
lowered: str = cleaned.lower()
if any(forbidden in lowered for forbidden in DISCORD_WEBHOOK_USERNAME_FORBIDDEN_SUBSTRINGS):
return ""
return cleaned
def normalize_message_avatar_url(avatar_url: str | None) -> str:
"""Return a usable webhook avatar URL, or empty string if unusable.
Blank or invalid values are rejected so Discord uses the webhook default.
Returns:
Valid http(s) URL, or empty string when the override should not be sent.
"""
if not avatar_url:
return ""
cleaned: str = avatar_url.strip()
if not cleaned:
return ""
if not cleaned.lower().startswith(("http://", "https://")):
return ""
if not is_url_valid(cleaned):
return ""
return cleaned
def get_validated_message_username(reader: Reader, feed: Feed) -> str:
"""Get a Discord-safe custom webhook username for a feed, if configured.
Returns:
Valid username to send, or empty string to use the webhook default.
"""
return normalize_message_username(get_message_username(reader, feed))
def get_validated_message_avatar_url(reader: Reader, feed: Feed) -> str:
"""Get a usable custom webhook avatar URL for a feed, if configured.
Returns:
Valid avatar URL to send, or empty string to use the webhook default.
"""
return normalize_message_avatar_url(get_message_avatar_url(reader, feed))
def save_embed(reader: Reader, feed: Feed, embed: CustomEmbed) -> None:
"""Set embed tag in feed.
@ -449,7 +552,11 @@ def get_embed(reader: Reader, feed: Feed) -> CustomEmbed:
def coerce_embed_bool(value: object) -> bool:
"""Normalize stored embed booleans from JSON or form-like values."""
"""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):

View file

@ -3,6 +3,7 @@ from __future__ import annotations
import asyncio
import concurrent.futures
import datetime
import functools
import hashlib
import json
import logging
@ -30,6 +31,7 @@ from httpx2 import HTTPError
from httpx2 import Response
from markdownify import markdownify
from playwright.sync_api import Browser
from playwright.sync_api import Error as PlaywrightError
from playwright.sync_api import Page
from playwright.sync_api import TimeoutError as PlaywrightTimeoutError
from playwright.sync_api import sync_playwright
@ -48,6 +50,8 @@ from requests import RequestException
from discord_rss_bot.custom_message import CustomEmbed
from discord_rss_bot.custom_message import get_custom_message
from discord_rss_bot.custom_message import get_image_urls
from discord_rss_bot.custom_message import get_validated_message_avatar_url
from discord_rss_bot.custom_message import get_validated_message_username
from discord_rss_bot.custom_message import replace_tags_in_embed
from discord_rss_bot.custom_message import replace_tags_in_text_message
from discord_rss_bot.filter.evaluator import get_entry_filter_decision_from_reader
@ -138,7 +142,7 @@ MESSAGE_PAYLOAD_KEYS: tuple[str, ...] = (
)
def extract_domain(url: str) -> str: # noqa: PLR0911
def extract_domain(url: str) -> str: # ruff:ignore[too-many-return-statements]
"""Extract the domain name from a URL.
Args:
@ -151,7 +155,7 @@ def extract_domain(url: str) -> str: # noqa: PLR0911
if not url:
return "Other"
try: # noqa: PLW0717
try: # ruff:ignore[too-many-statements-in-try-clause]
# Special handling for YouTube feeds
if "youtube.com/feeds/videos.xml" in url:
return "YouTube"
@ -256,7 +260,7 @@ def extract_steam_app_id_from_url(url: str) -> str | None:
normalized_netloc: str = parsed_url.netloc.lower().removeprefix("www.")
path_segments: list[str] = [segment for segment in parsed_url.path.split("/") if segment]
return _extract_steam_app_id_from_path(normalized_netloc, path_segments) or _extract_steam_app_id_from_query(
parsed_url
parsed_url,
)
@ -451,7 +455,7 @@ def get_feed_webhook_text_length_limit(reader: Reader, feed: Feed | str) -> int:
return coerce_webhook_text_length_limit(value)
def coerce_media_gallery_image_limit(value: JsonValue) -> int: # noqa: PLR0911
def coerce_media_gallery_image_limit(value: JsonValue) -> int: # ruff:ignore[too-many-return-statements]
"""Return the supported media gallery image limit for a stored tag value."""
if isinstance(value, bool):
return 1
@ -644,6 +648,10 @@ def get_webhook_message_edit_payload(payload: JsonObject, record: SentWebhookRec
if edit_payload.get("attachments") == []:
edit_payload.pop("attachments", None)
# Username/avatar can only be set when creating a message, not when editing.
edit_payload.pop("username", None)
edit_payload.pop("avatar_url", None)
return edit_payload
@ -829,7 +837,7 @@ def get_webhook_query_params(
return clean_webhook_url, params
def get_webhook_files(webhook: DiscordWebhook) -> list[WebhookFile]: # noqa: C901
def get_webhook_files(webhook: DiscordWebhook) -> list[WebhookFile]: # ruff:ignore[complex-structure]
"""Return files attached to a webhook object in a normalized shape."""
raw_files = getattr(webhook, "files", None)
files: list[WebhookFile] = []
@ -848,7 +856,7 @@ def get_webhook_files(webhook: DiscordWebhook) -> list[WebhookFile]: # noqa: C9
files.append(file_value)
continue
if not isinstance(file_value, tuple) or len(file_value) < 2: # noqa: PLR2004
if not isinstance(file_value, tuple) or len(file_value) < 2: # ruff:ignore[magic-value-comparison]
continue
first, second = file_value[0], file_value[1]
@ -856,7 +864,7 @@ def get_webhook_files(webhook: DiscordWebhook) -> list[WebhookFile]: # noqa: C9
files.append(WebhookFile(filename=first, content=second))
continue
if isinstance(second, tuple) and len(second) >= 2: # noqa: PLR2004
if isinstance(second, tuple) and len(second) >= 2: # ruff:ignore[magic-value-comparison]
nested_file = cast("tuple[object, ...]", second)
nested_filename, nested_content = nested_file[0], nested_file[1]
if isinstance(nested_filename, str) and isinstance(nested_content, bytes):
@ -910,7 +918,7 @@ def request_discord_webhook(
request_kwargs["json"] = payload
response: Response = httpx2.request(method, url, **request_kwargs)
if not rate_limit_retry or response.status_code != 429: # noqa: PLR2004
if not rate_limit_retry or response.status_code != 429: # ruff:ignore[magic-value-comparison]
return response
retry_after: float | None = get_retry_after_seconds(response)
@ -962,6 +970,21 @@ def edit_sent_webhook_message(
)
def apply_feed_webhook_identity(webhook: DiscordWebhook, entry: Entry, reader: Reader) -> DiscordWebhook:
"""Apply per-feed custom username and avatar when valid; ignore blank/invalid values.
Returns:
The same webhook instance with optional identity overrides.
"""
username: str = get_validated_message_username(reader, entry.feed)
avatar_url: str = get_validated_message_avatar_url(reader, entry.feed)
if username:
webhook.username = username
if avatar_url:
webhook.avatar_url = avatar_url
return webhook
def create_webhook_for_entry(
webhook_url: str,
entry: Entry,
@ -983,7 +1006,8 @@ def create_webhook_for_entry(
if post_id:
post_data = fetch_hoyolab_post(post_id)
if post_data:
return create_hoyolab_webhook(webhook_url, entry, post_data), delivery_mode
webhook = create_hoyolab_webhook(webhook_url, entry, post_data)
return apply_feed_webhook_identity(webhook, entry, reader), delivery_mode
logger.warning(
"Failed to create Hoyolab webhook for feed %s, falling back to regular processing",
entry.feed.url,
@ -992,18 +1016,18 @@ def create_webhook_for_entry(
logger.warning("No entry link found for feed %s, falling back to regular processing", entry.feed.url)
if delivery_mode == "embed":
return create_embed_webhook(webhook_url, entry, reader=reader), delivery_mode
webhook = create_embed_webhook(webhook_url, entry, reader=reader)
return apply_feed_webhook_identity(webhook, entry, reader), delivery_mode
if delivery_mode == "screenshot":
return create_screenshot_webhook(webhook_url, entry, reader=reader), delivery_mode
return (
create_text_webhook(
webhook = create_screenshot_webhook(webhook_url, entry, reader=reader)
return apply_feed_webhook_identity(webhook, entry, reader), delivery_mode
webhook = create_text_webhook(
webhook_url,
entry,
reader=reader,
use_default_message_on_empty=use_default_message_on_empty,
),
delivery_mode,
)
return apply_feed_webhook_identity(webhook, entry, reader), delivery_mode
def collect_modified_entries_during_update(reader: Reader, update_callback: UpdateCallback) -> list[tuple[str, str]]:
@ -1161,7 +1185,7 @@ def update_sent_webhook_record_for_entry(
)
def update_sent_webhooks_for_modified_entries( # noqa: C901
def update_sent_webhooks_for_modified_entries( # ruff:ignore[complex-structure]
reader: Reader,
modified_entries: Iterable[tuple[str, str]],
) -> int:
@ -1231,7 +1255,7 @@ def create_text_webhook(
"""
webhook_message: str = ""
if get_custom_message(reader, entry.feed) != "": # noqa: PLC1901
if get_custom_message(reader, entry.feed) != "": # ruff:ignore[compare-to-empty-string]
webhook_message = replace_tags_in_text_message(entry=entry, reader=reader)
if not webhook_message and use_default_message_on_empty:
@ -1348,6 +1372,40 @@ def screenshot_filename_for_entry(entry: Entry, *, extension: str = "png") -> st
return f"{safe_name[:80]}.{safe_extension}"
@functools.lru_cache(maxsize=1)
def is_chromium_installed() -> bool:
"""Check if Playwright's Chromium browser is installed.
Uses a cached check so the browser is only probed once per process lifetime.
Offloads to a thread when called from an active event loop (e.g. FastAPI).
Returns:
bool: True if Chromium launched successfully, False otherwise.
"""
def _check() -> bool:
try:
with sync_playwright() as playwright:
browser = playwright.chromium.launch(
headless=True,
args=["--disable-dev-shm-usage", "--no-sandbox"],
)
browser.close()
except (OSError, PlaywrightError):
return False
else:
return True
try:
asyncio.get_running_loop()
except RuntimeError:
return _check()
else:
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
future: concurrent.futures.Future[bool] = executor.submit(_check)
return future.result()
def capture_full_page_screenshot(
url: str,
*,
@ -1395,7 +1453,7 @@ def _capture_full_page_screenshot_sync(
Returns:
bytes | None: PNG bytes on success, otherwise None.
"""
try: # noqa: PLW0717
try: # ruff:ignore[too-many-statements-in-try-clause]
with sync_playwright() as playwright:
browser: Browser = playwright.chromium.launch(
headless=True,
@ -1515,7 +1573,7 @@ def set_description(
# Discord allows 2048, but we keep a small safety margin by default.
embed_description: str = custom_embed.description
if len(embed_description) > max_description_length:
if max_description_length <= 3: # noqa: PLR2004
if max_description_length <= 3: # ruff:ignore[magic-value-comparison]
embed_description = embed_description[:max_description_length]
else:
embed_description = f"{embed_description[: max_description_length - 3]}..."
@ -1613,7 +1671,7 @@ def get_ttvdrops_reward_description(drop: JsonObject, reward: JsonObject) -> str
return reward_name
def extract_ttvdrops_media_gallery_items(value: JsonValue, *, hide_paid: bool = False) -> list[JsonObject]: # noqa: C901
def extract_ttvdrops_media_gallery_items(value: JsonValue, *, hide_paid: bool = False) -> list[JsonObject]: # ruff:ignore[complex-structure]
"""Extract benefit/reward media gallery items from a ttvdrops API response.
Returns:
@ -1666,7 +1724,7 @@ def fetch_ttvdrops_campaign_media_items(entry: Entry) -> list[JsonObject]:
try:
response: Response = httpx2.get(api_url, follow_redirects=True, timeout=10.0)
if response.status_code != 200: # noqa: PLR2004
if response.status_code != 200: # ruff:ignore[magic-value-comparison]
logger.warning("Failed to fetch ttvdrops campaign data from %s: %s", api_url, response.text[:500])
return []
@ -1721,7 +1779,7 @@ def truncate_component_text(
"""
if len(content) <= max_text_display_length:
return content
if max_text_display_length <= 3: # noqa: PLR2004
if max_text_display_length <= 3: # ruff:ignore[magic-value-comparison]
return content[:max_text_display_length]
return f"{content[: max_text_display_length - 3]}..."
@ -1815,7 +1873,7 @@ def create_components_v2_webhook(
)
def create_embed_webhook( # noqa: C901, PLR0912
def create_embed_webhook( # ruff:ignore[complex-structure, too-many-branches]
webhook_url: str,
entry: Entry,
reader: Reader,
@ -2099,7 +2157,7 @@ def truncate_webhook_message(
"""
if len(webhook_message) <= max_content_length:
return webhook_message
if max_content_length <= 3: # noqa: PLR2004
if max_content_length <= 3: # ruff:ignore[magic-value-comparison]
return webhook_message[:max_content_length]
head_length = (max_content_length - 3) // 2
@ -2123,7 +2181,7 @@ def remove_invalid_new_feed(reader: Reader, feed_url: str) -> None:
logger.exception("Failed to remove invalid feed after initial update: %s", feed_url)
def create_feed(reader: Reader, feed_url: str, webhook_dropdown: str) -> None: # noqa: C901, PLR0912
def create_feed(reader: Reader, feed_url: str, webhook_dropdown: str) -> None: # ruff:ignore[complex-structure, too-many-branches]
"""Add a new feed, update it and mark every entry as read.
Args:

View file

@ -25,7 +25,7 @@ import json
import logging
import os
import shutil
import subprocess # noqa: S404
import subprocess # ruff:ignore[suspicious-subprocess-import]
from pathlib import Path
from typing import TYPE_CHECKING
@ -44,6 +44,8 @@ type TagValue = JsonValue
_FEED_TAGS: tuple[str, ...] = (
"webhook",
"custom_message",
"message_username",
"message_avatar_url",
"delivery_mode",
"screenshot_layout",
"should_send_embed",
@ -100,22 +102,22 @@ def setup_backup_repo(backup_path: Path) -> bool:
Returns:
``True`` if the repository is ready, ``False`` on any error.
"""
try: # noqa: PLW0717
try: # ruff:ignore[too-many-statements-in-try-clause]
backup_path.mkdir(parents=True, exist_ok=True)
git_dir: Path = backup_path / ".git"
if not git_dir.exists():
subprocess.run([GIT_EXECUTABLE, "init", str(backup_path)], check=True, capture_output=True) # noqa: S603
subprocess.run([GIT_EXECUTABLE, "init", str(backup_path)], check=True, capture_output=True) # ruff:ignore[subprocess-without-shell-equals-true]
logger.info("Initialized git backup repository at %s", backup_path)
# Ensure a local identity exists so that `git commit` always works.
for key, value in (("user.email", "discord-rss-bot@localhost"), ("user.name", "discord-rss-bot")):
result: subprocess.CompletedProcess[bytes] = subprocess.run( # noqa: S603
result: subprocess.CompletedProcess[bytes] = subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[GIT_EXECUTABLE, "-C", str(backup_path), "config", "--local", key],
check=False,
capture_output=True,
)
if result.returncode != 0:
subprocess.run( # noqa: S603
subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[GIT_EXECUTABLE, "-C", str(backup_path), "config", "--local", key, value],
check=True,
capture_output=True,
@ -125,14 +127,14 @@ def setup_backup_repo(backup_path: Path) -> bool:
remote_url: str = get_backup_remote()
if remote_url:
# Check if remote "origin" already exists.
check_remote: subprocess.CompletedProcess[bytes] = subprocess.run( # noqa: S603
check_remote: subprocess.CompletedProcess[bytes] = subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[GIT_EXECUTABLE, "-C", str(backup_path), "remote", "get-url", "origin"],
check=False,
capture_output=True,
)
if check_remote.returncode != 0:
# Remote doesn't exist, add it.
subprocess.run( # noqa: S603
subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[GIT_EXECUTABLE, "-C", str(backup_path), "remote", "add", "origin", remote_url],
check=True,
capture_output=True,
@ -142,7 +144,7 @@ def setup_backup_repo(backup_path: Path) -> bool:
# Remote exists, update it if the URL has changed.
current_url: str = check_remote.stdout.decode().strip()
if current_url != remote_url:
subprocess.run( # noqa: S603
subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[GIT_EXECUTABLE, "-C", str(backup_path), "remote", "set-url", "origin", remote_url],
check=True,
capture_output=True,
@ -167,7 +169,7 @@ def export_state(reader: Reader, backup_path: Path) -> None:
for tag in _FEED_TAGS:
try:
value: TagValue = reader.get_tag(feed, tag, None)
if value is not None and value != "": # noqa: PLC1901
if value is not None and value != "": # ruff:ignore[compare-to-empty-string]
feed_data[tag] = value
except Exception:
logger.exception("Failed to read tag '%s' for feed '%s' during state export", tag, feed.url)
@ -217,13 +219,13 @@ def commit_state_change(reader: Reader, message: str) -> None:
if not setup_backup_repo(backup_path):
return
try: # noqa: PLW0717
try: # ruff:ignore[too-many-statements-in-try-clause]
export_state(reader, backup_path)
subprocess.run([GIT_EXECUTABLE, "-C", str(backup_path), "add", "-A"], check=True, capture_output=True) # noqa: S603
subprocess.run([GIT_EXECUTABLE, "-C", str(backup_path), "add", "-A"], check=True, capture_output=True) # ruff:ignore[subprocess-without-shell-equals-true]
# Only create a commit if there are staged changes.
diff_result: subprocess.CompletedProcess[bytes] = subprocess.run( # noqa: S603
diff_result: subprocess.CompletedProcess[bytes] = subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[GIT_EXECUTABLE, "-C", str(backup_path), "diff", "--cached", "--exit-code"],
check=False,
capture_output=True,
@ -232,7 +234,7 @@ def commit_state_change(reader: Reader, message: str) -> None:
logger.debug("No state changes to commit for: %s", message)
return
subprocess.run( # noqa: S603
subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[GIT_EXECUTABLE, "-C", str(backup_path), "commit", "-m", message],
check=True,
capture_output=True,
@ -241,7 +243,7 @@ def commit_state_change(reader: Reader, message: str) -> None:
# Push to remote if configured.
if get_backup_remote():
subprocess.run( # noqa: S603
subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[GIT_EXECUTABLE, "-C", str(backup_path), "push", "origin", "HEAD"],
check=True,
capture_output=True,

View file

@ -18,7 +18,7 @@ def healthcheck() -> None:
sys.exit(0)
sys.exit(1)
except requests.exceptions.RequestException as e:
print(f"Healthcheck failed: {e}", file=sys.stderr) # noqa: T201
print(f"Healthcheck failed: {e}", file=sys.stderr) # ruff:ignore[print]
sys.exit(1)

View file

@ -76,7 +76,7 @@ def fetch_hoyolab_post(post_id: str) -> JsonObject | None:
return None
http_ok = 200
try: # noqa: PLW0717
try: # ruff:ignore[too-many-statements-in-try-clause]
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)
@ -94,7 +94,7 @@ def fetch_hoyolab_post(post_id: str) -> JsonObject | None:
return None
def create_hoyolab_webhook(webhook_url: str, entry: Entry, post_data: JsonObject) -> DiscordWebhook: # noqa: C901, PLR0912, PLR0914, PLR0915
def create_hoyolab_webhook(webhook_url: str, entry: Entry, post_data: JsonObject) -> DiscordWebhook: # ruff:ignore[complex-structure, too-many-branches, too-many-locals, too-many-statements]
"""Create a webhook with data from the Hoyolab API.
Args:
@ -184,8 +184,8 @@ def create_hoyolab_webhook(webhook_url: str, entry: Entry, post_data: JsonObject
# Only show Youtube URL if available
structured_content: str = str(post.get("structured_content", ""))
if structured_content: # noqa: PLR1702
try: # noqa: PLW0717
if structured_content: # ruff:ignore[too-many-nested-blocks]
try: # ruff:ignore[too-many-statements-in-try-clause]
loaded_structured_content = cast("JsonValue", json.loads(structured_content))
structured_content_data: list[JsonObject] = (
[cast("JsonObject", item) for item in loaded_structured_content if isinstance(item, dict)]

View file

@ -49,6 +49,8 @@ from discord_rss_bot.custom_message import CustomEmbed
from discord_rss_bot.custom_message import get_custom_message
from discord_rss_bot.custom_message import get_embed
from discord_rss_bot.custom_message import get_first_image
from discord_rss_bot.custom_message import get_message_avatar_url
from discord_rss_bot.custom_message import get_message_username
from discord_rss_bot.custom_message import replace_tags_in_text_message
from discord_rss_bot.custom_message import save_embed
from discord_rss_bot.feeds import FeedUpdateError
@ -63,6 +65,7 @@ from discord_rss_bot.feeds import get_feed_media_gallery_image_limit
from discord_rss_bot.feeds import get_feed_webhook_text_length_limit
from discord_rss_bot.feeds import get_screenshot_layout
from discord_rss_bot.feeds import get_sent_webhook_records
from discord_rss_bot.feeds import is_chromium_installed
from discord_rss_bot.feeds import is_steam_feed_url
from discord_rss_bot.feeds import send_entry_to_discord
from discord_rss_bot.feeds import send_to_discord
@ -136,7 +139,7 @@ LOGGING_CONFIG = {
"disable_existing_loggers": False,
"formatters": {
"standard": {
"format": "%(asctime)s [%(processName)s: %(process)d] [%(threadName)s: %(thread)d] [%(levelname)s] %(name)s: %(message)s", # noqa: E501
"format": "%(asctime)s [%(processName)s: %(process)d] [%(threadName)s: %(thread)d] [%(levelname)s] %(name)s: %(message)s", # ruff:ignore[line-too-long]
},
},
"handlers": {
@ -1145,11 +1148,15 @@ async def post_set_custom(
feed_url: Annotated[str, Form()],
reader: Annotated[Reader, Depends(get_reader_dependency)],
custom_message: Annotated[str, Form()] = "",
message_username: Annotated[str, Form()] = "",
message_avatar_url: Annotated[str, Form()] = "",
) -> RedirectResponse:
"""Set the custom message, this is used when sending the message.
Args:
custom_message: The custom message.
message_username: Optional Discord webhook username override for this feed.
message_avatar_url: Optional Discord webhook avatar URL override for this feed.
feed_url: The feed we should set the custom message for.
reader: The Reader instance.
@ -1159,6 +1166,10 @@ async def post_set_custom(
our_custom_message: JSONType | str = custom_message.strip()
our_custom_message = typing.cast("JSONType", our_custom_message)
# Store raw values; blank/invalid values are ignored when building the Discord payload.
reader.set_tag(feed_url, "message_username", typing.cast("JSONType", message_username.strip()))
reader.set_tag(feed_url, "message_avatar_url", typing.cast("JSONType", message_avatar_url.strip()))
clean_feed_url: str = feed_url.strip()
feed: Feed = reader.get_feed(urllib.parse.unquote(clean_feed_url))
@ -1192,6 +1203,8 @@ async def get_custom(
"request": request,
"feed": feed,
"custom_message": get_custom_message(reader, feed),
"message_username": get_message_username(reader, feed),
"message_avatar_url": get_message_avatar_url(reader, feed),
}
# Get the first entry, this is used to show the user what the custom message will look like.
@ -1247,7 +1260,7 @@ async def get_embed_page(
@app.post("/embed", response_class=HTMLResponse)
async def post_embed( # noqa: C901
async def post_embed( # ruff:ignore[complex-structure]
feed_url: Annotated[str, Form()],
reader: Annotated[Reader, Depends(get_reader_dependency)],
title: Annotated[str, Form()] = "",
@ -1260,6 +1273,7 @@ async def post_embed( # noqa: C901
author_icon_url: Annotated[str, Form()] = "",
footer_text: Annotated[str, Form()] = "",
footer_icon_url: Annotated[str, Form()] = "",
*,
show_steam_game_icon_in_thumbnail: Annotated[bool, Form()] = False,
) -> RedirectResponse:
"""Set the embed settings.
@ -1766,7 +1780,7 @@ def get_add(
@app.get("/feed", response_class=HTMLResponse)
async def get_feed( # noqa: C901, PLR0912, PLR0914, PLR0915
async def get_feed( # ruff:ignore[complex-structure, too-many-branches, too-many-locals, too-many-statements]
feed_url: str,
request: Request,
reader: Annotated[Reader, Depends(get_reader_dependency)],
@ -1862,6 +1876,7 @@ async def get_feed( # noqa: C901, PLR0912, PLR0914, PLR0915
"max_webhook_text_length_limit": 4000,
"save_sent_webhooks": feed_saves_sent_webhooks(reader, feed),
"is_steam_feed": is_steam_feed_url(feed.url),
"chromium_installed": is_chromium_installed(),
}
return templates.TemplateResponse(request=request, name="feed.html", context=context)
@ -1929,11 +1944,12 @@ async def get_feed( # noqa: C901, PLR0912, PLR0914, PLR0915
"max_webhook_text_length_limit": 4000,
"save_sent_webhooks": feed_saves_sent_webhooks(reader, feed),
"is_steam_feed": is_steam_feed_url(feed.url),
"chromium_installed": is_chromium_installed(),
}
return templates.TemplateResponse(request=request, name="feed.html", context=context)
def create_html_for_feed( # noqa: C901, PLR0914
def create_html_for_feed( # ruff:ignore[complex-structure, too-many-locals]
reader: Reader,
entries: Iterable[Entry],
current_feed_url: str = "",
@ -2039,7 +2055,7 @@ def create_html_for_feed( # noqa: C901, PLR0914
{video_embed_html}
{image_html}
</div>
""" # noqa: E501
""" # ruff:ignore[line-too-long]
return html.strip()
@ -2140,7 +2156,7 @@ async def get_settings(
global_delivery_mode = "embed"
global_webhook_text_length_limit: int = coerce_webhook_text_length_limit(
reader.get_tag((), "webhook_text_length_limit", 4000)
reader.get_tag((), "webhook_text_length_limit", 4000),
)
# Get all feeds with their intervals
@ -2169,6 +2185,7 @@ async def get_settings(
"global_webhook_text_length_limit": global_webhook_text_length_limit,
"max_webhook_text_length_limit": 4000,
"feed_intervals": feed_intervals,
"chromium_installed": is_chromium_installed(),
}
return templates.TemplateResponse(request=request, name="settings.html", context=context)
@ -2591,8 +2608,8 @@ def create_webhook_feed_url_preview(
webhook_feeds: list[Feed],
replace_from: str,
replace_to: str,
resolve_urls: bool, # noqa: FBT001
force_update: bool = False, # noqa: FBT001, FBT002
resolve_urls: bool, # ruff:ignore[boolean-type-hint-positional-argument]
force_update: bool = False, # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument]
existing_feed_urls: set[str] | None = None,
) -> list[dict[str, str | bool | None]]:
"""Create preview rows for bulk feed URL replacement.
@ -2658,8 +2675,8 @@ def build_webhook_mass_update_context(
all_feeds: list[Feed],
replace_from: str,
replace_to: str,
resolve_urls: bool, # noqa: FBT001
force_update: bool = False, # noqa: FBT001, FBT002
resolve_urls: bool, # ruff:ignore[boolean-type-hint-positional-argument]
force_update: bool = False, # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument]
) -> dict[str, str | bool | int | list[dict[str, str | bool | None]] | dict[str, int]]:
"""Build context data used by the webhook mass URL update preview UI.
@ -2720,8 +2737,8 @@ async def get_webhook_entries_mass_update_preview(
reader: Annotated[Reader, Depends(get_reader_dependency)],
replace_from: str = "",
replace_to: str = "",
resolve_urls: bool = True, # noqa: FBT001, FBT002
force_update: bool = False, # noqa: FBT001, FBT002
resolve_urls: bool = True, # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument]
force_update: bool = False, # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument]
) -> HTMLResponse:
"""Render the mass-update preview fragment for a webhook using HTMX.
@ -2759,15 +2776,15 @@ async def get_webhook_entries_mass_update_preview(
@app.get("/webhook_entries", response_class=HTMLResponse)
async def get_webhook_entries( # noqa: C901, PLR0914
async def get_webhook_entries( # ruff:ignore[complex-structure, too-many-locals]
webhook_url: str,
request: Request,
reader: Annotated[Reader, Depends(get_reader_dependency)],
starting_after: str = "",
replace_from: str = "",
replace_to: str = "",
resolve_urls: bool = True, # noqa: FBT001, FBT002
force_update: bool = False, # noqa: FBT001, FBT002
resolve_urls: bool = True, # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument]
force_update: bool = False, # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument]
message: str = "",
) -> HTMLResponse:
"""Get all latest entries from all feeds for a specific webhook.
@ -2889,13 +2906,13 @@ async def get_webhook_entries( # noqa: C901, PLR0914
@app.post("/bulk_change_feed_urls", response_class=HTMLResponse)
async def post_bulk_change_feed_urls( # noqa: C901, PLR0914, PLR0912, PLR0915
async def post_bulk_change_feed_urls( # ruff:ignore[complex-structure, too-many-locals, too-many-branches, too-many-statements]
webhook_url: Annotated[str, Form()],
replace_from: Annotated[str, Form()],
reader: Annotated[Reader, Depends(get_reader_dependency)],
replace_to: Annotated[str, Form()] = "",
resolve_urls: Annotated[bool, Form()] = True, # noqa: FBT002
force_update: Annotated[bool, Form()] = False, # noqa: FBT002
resolve_urls: Annotated[bool, Form()] = True, # ruff:ignore[boolean-default-value-positional-argument]
force_update: Annotated[bool, Form()] = False, # ruff:ignore[boolean-default-value-positional-argument]
) -> RedirectResponse:
"""Bulk-change feed URLs attached to a webhook.
@ -3032,7 +3049,7 @@ if __name__ == "__main__":
uvicorn.run(
"main:app",
log_level="debug",
host="0.0.0.0", # noqa: S104
host="0.0.0.0", # ruff:ignore[hardcoded-bind-all-interfaces]
port=3000,
proxy_headers=True,
forwarded_allow_ips="*",

View file

@ -240,6 +240,28 @@
id="custom_message" {% if custom_message %} value="{{- custom_message -}}" {% endif %} />
</div>
</div>
<!-- Optional Discord webhook username & avatar for this feed -->
<div class="row pb-2">
<div class="col-sm-12">
<div class="form-text">
<ul class="list-inline">
<li>Optional: override the Discord webhook username and avatar for messages from this feed.</li>
<li>Leave blank to use the webhook's default name and avatar.</li>
<li>Invalid values (empty, disallowed characters, or bad avatar URLs) are ignored when sending.</li>
<li>Username rules: 180 characters; cannot contain <code>@</code>, <code>#</code>, <code>:</code>, or <code>`</code>; cannot contain <code>clyde</code> or <code>discord</code>.</li>
<li>Avatar must be a full <code>http://</code> or <code>https://</code> image URL.</li>
</ul>
</div>
<label for="message_username" class="col-sm-6 col-form-label">Message Username</label>
<input name="message_username" type="text" class="form-control bg-dark border-dark text-muted"
id="message_username" maxlength="80" placeholder="e.g. Power Of The Shell"
{% if message_username %} value="{{- message_username -}}" {% endif %} />
<label for="message_avatar_url" class="col-sm-6 col-form-label">Message Avatar Image</label>
<input name="message_avatar_url" type="url" class="form-control bg-dark border-dark text-muted"
id="message_avatar_url" placeholder="e.g. https://example.com/icon.png"
{% if message_avatar_url %} value="{{- message_avatar_url -}}" {% endif %} />
</div>
</div>
<!-- Add a hidden feed_url field to the form -->
<input type="hidden" name="feed_url" value="{{ feed.url }}" />
<!-- Submit button -->

View file

@ -218,6 +218,12 @@
message, please contact the developer.
{% endif %}
</div>
<div class="form-text">
<ul class="list-inline">
<li>You can use \n to create a new line.</li>
<li>You can remove the embed from links by adding &lt; and &gt; around the link. (For example &lt;{% raw %}{{entry_link}}{% endraw %}&gt;)</li>
</ul>
</div>
<label for="title" class="col-sm-6 col-form-label">Title</label>
<input name="title" type="text" class="form-control bg-dark border-dark text-muted" id="title"
{% if title %} value="{{- title -}}" {% endif %} />

View file

@ -161,86 +161,109 @@
<button class="btn btn-danger btn-sm" name="feed_url" value="{{ feed.url }}">Pause</button>
</form>
{% endif %}
{% if not "youtube.com/feeds/videos.xml" in feed.url %}
{% if delivery_mode == "embed" %}
<form action="/use_text" method="post" class="d-inline">
<button class="btn btn-outline-light btn-sm"
name="feed_url"
value="{{ feed.url }}">Send text message instead of embed</button>
</form>
{% elif delivery_mode == "screenshot" %}
<form action="/use_text" method="post" class="d-inline">
<button class="btn btn-outline-light btn-sm"
name="feed_url"
value="{{ feed.url }}">Send text message instead of screenshot</button>
</form>
{% else %}
<form action="/use_embed" method="post" class="d-inline">
<button class="btn btn-outline-light btn-sm"
name="feed_url"
value="{{ feed.url }}">Send embed instead of text message</button>
</form>
{% endif %}
{% endif %}
</div>
</section>
{% if not "youtube.com/feeds/videos.xml" in feed.url %}
<section class="mt-4 pt-3 border-top border-secondary-subtle">
<div class="d-flex flex-wrap align-items-center justify-content-between gap-2 mb-2">
<h3 class="h6 text-uppercase text-muted mb-0">Screenshot Delivery</h3>
<span class="badge {{ 'bg-info' if delivery_mode == 'screenshot' else 'bg-secondary' }}">
{% if delivery_mode == "screenshot" %}
Active:
{% if screenshot_layout == "mobile" %}
Mobile
{% else %}
Desktop
{% endif %}
{% else %}
Inactive
{% endif %}
</span>
<div class="row row-cols-1 row-cols-lg-2 g-3 mt-2">
<div class="col">
<div class="card border border-secondary shadow-sm h-100 rounded-0">
<div class="card-header bg-transparent border-secondary text-muted text-uppercase small fw-semibold">
Delivery Settings
</div>
<p class="text-muted mb-3">
Screenshot delivery sends a full-page screenshot of the entry link instead of the normal
embed or text message.
<div class="card-body">
<section class="mb-3">
<h4 class="h6 text-muted mb-2">Delivery Mode</h4>
<p class="text-muted small mb-2">
Choose between:
<br>
Embed: a Discord embed with title, description, and images.
<br>
{% if not "youtube.com/feeds/videos.xml" in feed.url %}
Screenshot: full-page screenshot of the entry link.
<br>
{% endif %}
Text: plain message.
</p>
<div class="d-flex flex-wrap gap-2">
{% if delivery_mode != "screenshot" %}
<form action="/use_screenshot" method="post" class="d-inline">
<button class="btn btn-outline-light btn-sm"
name="feed_url"
value="{{ feed.url }}">Use screenshot delivery</button>
</form>
{% else %}
<div class="d-flex" role="group" aria-label="Delivery mode">
{% if delivery_mode != "embed" %}
<form action="/use_embed" method="post" class="d-inline">
<button class="btn btn-outline-light btn-sm"
style="border-top-right-radius: 0;
border-bottom-right-radius: 0"
name="feed_url"
value="{{ feed.url }}">Disable screenshot delivery</button>
value="{{ feed.url }}">Embed</button>
</form>
{% else %}
<span class="btn btn-primary btn-sm disabled"
style="border-top-right-radius: 0;
border-bottom-right-radius: 0">Embed</span>
{% endif %}
{% if not "youtube.com/feeds/videos.xml" in feed.url %}
{% if chromium_installed %}
{% if delivery_mode != "screenshot" %}
<form action="/use_screenshot" method="post" class="d-inline">
<button class="btn btn-outline-light btn-sm rounded-0"
name="feed_url"
value="{{ feed.url }}">Screenshot</button>
</form>
{% else %}
<span class="btn btn-primary btn-sm disabled rounded-0">Screenshot</span>
{% endif %}
{% else %}
<span class="btn btn-outline-light btn-sm rounded-0"
title="Playwright Chromium is not installed. Run 'uv run playwright install chromium' to enable."
style="opacity: 0.5;
cursor: not-allowed">Screenshot</span>
{% endif %}
{% endif %}
{% if delivery_mode != "text" %}
<form action="/use_text" method="post" class="d-inline">
<button class="btn btn-outline-light btn-sm"
style="border-top-left-radius: 0;
border-bottom-left-radius: 0"
name="feed_url"
value="{{ feed.url }}">Text</button>
</form>
{% else %}
<span class="btn btn-primary btn-sm disabled"
style="border-top-left-radius: 0;
border-bottom-left-radius: 0">Text</span>
{% endif %}
</div>
{% if not chromium_installed and not "youtube.com/feeds/videos.xml" in feed.url %}
<div class="mt-2 small text-muted">
Screenshot mode requires Chromium for Playwright.
Run <code class="text-light">uv run playwright install chromium</code> to enable it.
</div>
{% endif %}
</section>
{% if delivery_mode == "screenshot" and chromium_installed and not "youtube.com/feeds/videos.xml" in feed.url %}
<hr class="border-secondary" />
<section class="mb-3">
<h4 class="h6 text-muted mb-2">Screenshot Layout</h4>
<div class="d-flex flex-wrap gap-2">
{% if screenshot_layout == "mobile" %}
<span class="btn btn-primary btn-sm disabled">Mobile</span>
<form action="/use_screenshot_desktop" method="post" class="d-inline">
<button class="btn btn-outline-light btn-sm"
name="feed_url"
value="{{ feed.url }}">Use desktop screenshot layout</button>
value="{{ feed.url }}">Desktop</button>
</form>
{% else %}
<form action="/use_screenshot_mobile" method="post" class="d-inline">
<button class="btn btn-outline-secondary btn-sm"
<button class="btn btn-outline-light btn-sm"
name="feed_url"
value="{{ feed.url }}">Use mobile screenshot layout</button>
value="{{ feed.url }}">Mobile</button>
</form>
<span class="btn btn-primary btn-sm disabled">Desktop</span>
{% endif %}
{% endif %}
</div>
<div class="mt-2 screenshot-requirement">
Screenshot mode requires Chromium to be installed for Playwright. Run
<code>uv run playwright install chromium</code> once on this machine.
</div>
</section>
<section class="mt-4 pt-3 border-top border-secondary-subtle">
{% endif %}
{% if not "youtube.com/feeds/videos.xml" in feed.url %}
<hr class="border-secondary" />
<section class="mb-3">
<div class="d-flex flex-wrap align-items-center justify-content-between gap-2 mb-2">
<h3 class="h6 text-uppercase text-muted mb-0">Image Delivery</h3>
<h4 class="h6 text-muted mb-0">Image Delivery</h4>
<span class="badge {{ 'bg-info' if media_gallery_image_limit < max_media_gallery_items else 'bg-secondary' }}">
{% if media_gallery_image_limit == 0 %}
No images
@ -251,10 +274,8 @@
{% endif %}
</span>
</div>
<p id="imageDeliveryHelp" class="text-muted mb-3">
Choose 0 to send no entry images, 1 for the first image only,
or up to {{ max_media_gallery_items }} for a Discord media gallery.
This only affects embed delivery.
<p id="imageDeliveryHelp" class="text-muted small mb-2">
How many images to include per entry. Only applies to embed mode.
</p>
<form action="/set_feed_media_gallery_image_limit"
method="post"
@ -286,14 +307,15 @@
</form>
</section>
{% endif %}
<section class="mt-4 pt-3 border-top border-secondary-subtle">
<hr class="border-secondary" />
<section>
<div class="d-flex flex-wrap align-items-center justify-content-between gap-2 mb-2">
<h3 class="h6 text-uppercase text-muted mb-0">Text Delivery</h3>
<h4 class="h6 text-muted mb-0">Text Delivery</h4>
<span class="badge {{ 'bg-info' if webhook_text_length_limit < max_webhook_text_length_limit else 'bg-secondary' }}">
Max {{ webhook_text_length_limit }} characters
</span>
</div>
<p id="textDeliveryHelp" class="text-muted mb-3">
<p id="textDeliveryHelp" class="text-muted small mb-2">
Limit message length. Text mode allows values up to 4000 characters. Embeds are capped at 2000 characters.
</p>
<form action="/set_feed_webhook_text_length_limit"
@ -318,35 +340,20 @@
</div>
</form>
</section>
<section class="mt-4 pt-3 border-top border-secondary-subtle">
<h3 class="h6 text-uppercase text-muted mb-3">Customization</h3>
<div class="d-flex flex-wrap gap-2">
<a class="btn btn-sm btn-outline-light"
href="/whitelist?feed_url={{ feed.url|encode_url }}">Whitelist</a>
<a class="btn btn-sm btn-outline-light"
href="/blacklist?feed_url={{ feed.url|encode_url }}">Blacklist</a>
<a class="btn btn-sm btn-outline-light"
href="/custom?feed_url={{ feed.url|encode_url }}">
Customize message
{% if delivery_mode == "text" %}(Currently active){% endif %}
</a>
{% if not "youtube.com/feeds/videos.xml" in feed.url %}
<a class="btn btn-sm btn-outline-light"
href="/embed?feed_url={{ feed.url|encode_url }}">
Customize embed
{% if delivery_mode == "embed" %}(Currently active){% endif %}
</a>
{% endif %}
</div>
{% if is_steam_feed %}
<div class="form-text mt-2">Steam feeds can enable thumbnails in embed settings.</div>
{% endif %}
</section>
<section class="mt-4 pt-3 border-top border-secondary-subtle">
<h3 class="h6 text-uppercase text-muted mb-3">Feed URL</h3>
</div>
</div>
<div class="col">
<div class="card border border-secondary shadow-sm h-100 rounded-0">
<div class="card-header bg-transparent border-secondary text-muted text-uppercase small fw-semibold">
Feed Configuration
</div>
<div class="card-body">
<section class="mb-3">
<h4 class="h6 text-muted mb-2">Feed URL</h4>
<form action="/change_feed_url" method="post" class="mb-0">
<input type="hidden" name="old_feed_url" value="{{ feed.url }}" />
<div class="input-group input-group-sm feed-page__content">
<div class="input-group input-group-sm">
<input type="url"
class="form-control feed-page__wrap"
name="new_feed_url"
@ -356,22 +363,23 @@
</div>
</form>
</section>
<section class="mt-4 pt-3 border-top border-secondary-subtle">
<h3 class="h6 text-uppercase text-muted mb-3">Webhook</h3>
<hr class="border-secondary" />
<section class="mb-3">
<h4 class="h6 text-muted mb-2">Webhook</h4>
{% if current_webhook_name %}
<p class="text-muted mb-3">
<p class="text-muted small mb-2">
Current webhook:
<strong>{{ current_webhook_name }}</strong>
</p>
{% elif current_webhook_url %}
<p class="text-warning mb-3">This feed references a missing webhook. Choose a webhook below to reattach it.</p>
<p class="text-warning small mb-2">This feed references a missing webhook. Choose a webhook below to reattach it.</p>
{% else %}
<p class="text-warning mb-3">No webhook is attached to this feed yet.</p>
<p class="text-warning small mb-2">No webhook is attached to this feed yet.</p>
{% endif %}
{% if webhooks %}
<form action="/attach_feed_webhook"
method="post"
class="d-flex flex-wrap align-items-center gap-2 mb-0">
class="d-flex flex-wrap align-items-center gap-2 mb-2">
<input type="hidden" name="feed_url" value="{{ feed.url }}" />
<input type="hidden"
name="redirect_to"
@ -394,9 +402,9 @@
<button class="btn btn-outline-light btn-sm" type="submit">Save webhook</button>
</form>
{% else %}
<p class="text-muted mb-0">Add a webhook first to attach this feed.</p>
<p class="text-muted small mb-2">Add a webhook first to attach this feed.</p>
{% endif %}
<div class="mt-3 d-flex flex-wrap align-items-center gap-2">
<div class="d-flex flex-wrap align-items-center gap-2">
<span class="badge {{ 'bg-success' if save_sent_webhooks else 'bg-secondary' }}">
Sent webhook tracking:
{{ 'Enabled' if save_sent_webhooks else 'Disabled' }}
@ -414,34 +422,10 @@
href="/sent_webhooks?feed_url={{ feed.url|encode_url }}">View sent webhooks</a>
</div>
</section>
<section class="mt-4 pt-3 border-top border-secondary-subtle">
<h3 class="h6 text-uppercase text-muted mb-3">Feed Information</h3>
<div class="row g-2 text-muted small">
<div class="col-12 col-md-6">
<div class="p-2">Added: {{ feed.added | relative_time }}</div>
</div>
<div class="col-12 col-md-6">
<div class="p-2">Last Updated: {{ feed.last_updated | relative_time }}</div>
</div>
<div class="col-12 col-md-6">
<div class="p-2">Last Retrieved: {{ feed.last_retrieved | relative_time }}</div>
</div>
<div class="col-12 col-md-6">
<div class="p-2">Next Update: {{ feed.update_after | relative_time }}</div>
</div>
<div class="col-12 col-md-6">
<div class="p-2">
Updates:
<span class="badge {{ 'bg-success' if feed.updates_enabled else 'bg-danger' }}">
{{ 'Enabled' if feed.updates_enabled else 'Disabled' }}
</span>
</div>
</div>
</div>
</section>
<section class="mt-4 pt-3 border-top border-secondary-subtle">
<div class="d-flex flex-wrap align-items-center justify-content-between gap-2 mb-3">
<h3 class="h6 text-uppercase text-muted mb-0">Update Interval</h3>
<hr class="border-secondary" />
<section>
<h4 class="h6 text-muted mb-2">Update Interval</h4>
<div class="d-flex flex-wrap align-items-center gap-2 mb-2">
<span class="badge {% if feed_interval %} bg-info {% else %} bg-secondary {% endif %}">
{% if feed_interval %}
Custom
@ -450,7 +434,7 @@
{% endif %}
</span>
</div>
<p class="text-muted mb-3">
<p class="text-muted small mb-2">
Current:
<strong>
{% if feed_interval %}
@ -487,6 +471,69 @@
</div>
</section>
</div>
</div>
</div>
</div>
<div class="row row-cols-1 row-cols-lg-2 g-3 mt-2">
<div class="col">
<div class="card border border-secondary shadow-sm h-100 rounded-0">
<div class="card-header bg-transparent border-secondary text-muted text-uppercase small fw-semibold">
Customization
</div>
<div class="card-body">
<div class="d-flex flex-wrap gap-2">
<a class="btn btn-sm btn-outline-light"
href="/whitelist?feed_url={{ feed.url|encode_url }}">Whitelist</a>
<a class="btn btn-sm btn-outline-light"
href="/blacklist?feed_url={{ feed.url|encode_url }}">Blacklist</a>
<div class="btn-group btn-group-sm" role="group">
<a class="btn {{ 'btn-primary' if delivery_mode == 'text' else 'btn-outline-light' }}"
href="/custom?feed_url={{ feed.url|encode_url }}">Customize message</a>
{% if not "youtube.com/feeds/videos.xml" in feed.url %}
<a class="btn {{ 'btn-primary' if delivery_mode == 'embed' else 'btn-outline-light' }}"
href="/embed?feed_url={{ feed.url|encode_url }}">Customize embed</a>
{% endif %}
</div>
</div>
{% if is_steam_feed %}
<div class="form-text small mt-2">Steam feeds can enable thumbnails in embed settings.</div>
{% endif %}
</div>
</div>
</div>
<div class="col">
<div class="card border border-secondary shadow-sm h-100 rounded-0">
<div class="card-header bg-transparent border-secondary text-muted text-uppercase small fw-semibold">
Feed Information
</div>
<div class="card-body">
<div class="row g-2 text-muted small">
<div class="col-12 col-md-6">
<div class="p-2">Added: {{ feed.added | relative_time }}</div>
</div>
<div class="col-12 col-md-6">
<div class="p-2">Last Updated: {{ feed.last_updated | relative_time }}</div>
</div>
<div class="col-12 col-md-6">
<div class="p-2">Last Retrieved: {{ feed.last_retrieved | relative_time }}</div>
</div>
<div class="col-12 col-md-6">
<div class="p-2">Next Update: {{ feed.update_after | relative_time }}</div>
</div>
<div class="col-12 col-md-6">
<div class="p-2">
Updates:
<span class="badge {{ 'bg-success' if feed.updates_enabled else 'bg-danger' }}">
{{ 'Enabled' if feed.updates_enabled else 'Disabled' }}
</span>
</div>
</div>
</div>
</div>
</div>
</div>
</div>
</div>
</article>
</div>
<div class="col-12">

View file

@ -46,6 +46,11 @@
name="delivery_mode">
<option value="embed"
{% if global_delivery_mode == "embed" %}selected{% endif %}>Embed</option>
<option value="screenshot"
{% if global_delivery_mode == "screenshot" %}selected{% endif %}
{% if not chromium_installed %}disabled{% endif %}>
Screenshot{% if not chromium_installed %} (Chromium not installed){% endif %}
</option>
<option value="text"
{% if global_delivery_mode == "text" %}selected{% endif %}>Text</option>
</select>
@ -73,9 +78,12 @@
<div class="form-text text-muted mt-2">
New feeds inherit this value. Existing feeds keep their current screenshot layout.
</div>
<div class="form-text screenshot-requirement mt-1">
Screenshot mode requires Chromium to be installed for Playwright. Run <code>uv run playwright install chromium</code> once on this machine.
{% if not chromium_installed %}
<div class="form-text text-muted mt-1">
Screenshot mode requires Chromium to be installed for Playwright.
Run <code>uv run playwright install chromium</code> once on this machine.
</div>
{% endif %}
</div>
</form>
<form action="/set_global_webhook_text_length_limit"

View file

@ -19,7 +19,7 @@ class WebhookFile:
class DiscordEmbed:
"""Small Discord embed payload builder used by the webhook sender."""
def __init__(self) -> None: # noqa: D107
def __init__(self) -> None: # ruff:ignore[undocumented-public-init]
self._payload: JsonObject = {}
def to_dict(self) -> JsonObject:
@ -54,7 +54,7 @@ class DiscordEmbed:
def set_thumbnail(self, *, url: str) -> None:
self._payload["thumbnail"] = {"url": url}
def set_image(self, *, url: str, **_ignored: Any) -> None: # noqa: ANN401
def set_image(self, *, url: str, **_ignored: Any) -> None: # ruff:ignore[any-type]
self._payload["image"] = {"url": url}
def set_footer(self, *, text: str, icon_url: str | None = None) -> None:
@ -85,7 +85,7 @@ class DiscordWebhook:
while leaving the actual HTTP transport to `httpx2`.
"""
def __init__( # noqa: D107
def __init__( # ruff:ignore[undocumented-public-init]
self,
url: str,
*,
@ -99,7 +99,7 @@ class DiscordWebhook:
thread_id: str | None = None,
timeout: float | None = None,
rate_limit_retry: bool = False,
**_ignored: Any, # noqa: ANN401
**_ignored: Any, # ruff:ignore[any-type]
) -> None:
self.url: str = url
self.thread_id: str | None = thread_id

View file

@ -2,19 +2,29 @@ services:
discord-rss-bot:
image: ghcr.io/thelovinator1/discord-rss-bot:latest
container_name: discord-rss-bot
expose:
- "5000:5000"
ports:
- "5000:5000"
volumes:
# - /Docker/Bots/discord-rss-bot:/home/botuser/.local/share/discord_rss_bot/
# To use a host directory instead of a named volume:
# Linux: - /Docker/Bots/discord-rss-bot:/home/botuser/.local/share/discord_rss_bot/
# Windows: - C:\Docker\Bots\discord-rss-bot:/home/botuser/.local/share/discord_rss_bot/
- data:/home/botuser/.local/share/discord_rss_bot/
healthcheck:
test: [ "CMD", "uv", "run", "./discord_rss_bot/healthcheck.py" ]
interval: 1m
timeout: 10s
retries: 3
start_period: 5s
# Keep logs in a rotating file so they don't fill up the disk
logging:
driver: "json-file"
options:
max-size: "10m"
max-file: "3"
restart: unless-stopped
volumes:

View file

@ -42,55 +42,56 @@ fix = true
line-length = 120
lint.select = ["ALL"]
lint.unfixable = ["F841"] # Don't automatically remove unused variables
# Don't automatically remove unused variables
lint.unfixable = ["unused-variable"]
lint.pydocstyle.convention = "google"
lint.isort.required-imports = ["from __future__ import annotations"]
lint.isort.force-single-line = true
lint.ignore = [
"ANN201", # Checks that public functions and methods have return type annotations.
"ARG001", # Checks for the presence of unused arguments in function definitions.
"B008", # Allow Form() as a default value
"CPY001", # Missing copyright notice at top of file
"D100", # Checks for undocumented public module definitions.
"D101", # Checks for undocumented public class definitions.
"D102", # Checks for undocumented public method definitions.
"D104", # Missing docstring in public package.
"D105", # Missing docstring in magic method.
"D105", # pydocstyle - missing docstring in magic method
"D106", # Checks for undocumented public class definitions, for nested classes.
"ERA001", # Found commented-out code
"FBT003", # Checks for boolean positional arguments in function calls.
"FIX002", # Line contains TODO
"G002", # Allow % in logging
"PGH003", # Check for type: ignore annotations that suppress all type warnings, as opposed to targeting specific type warnings.
"PLR6301", # Checks for the presence of unused self parameter in methods definitions.
"RUF029", # Checks for functions declared async that do not await or otherwise use features requiring the function to be declared async.
"TD003", # Checks that a TODO comment is associated with a link to a relevant issue or ticket.
"PLR0913", # Checks for function definitions that include too many arguments.
"PLR0917", # Checks for function definitions that include too many positional arguments.
"missing-return-type-undocumented-public-function", # Checks that public functions and methods have return type annotations.
"unused-function-argument", # Checks for the presence of unused arguments in function definitions.
"function-call-in-default-argument", # Allow Form() as a default value
"missing-copyright-notice", # Missing copyright notice at top of file
"undocumented-public-module", # Checks for undocumented public module definitions.
"undocumented-public-class", # Checks for undocumented public class definitions.
"undocumented-public-method", # Checks for undocumented public method definitions.
"undocumented-public-package", # Missing docstring in public package.
"undocumented-magic-method", # Missing docstring in magic method.
"undocumented-magic-method", # pydocstyle - missing docstring in magic method
"undocumented-public-nested-class", # Checks for undocumented public class definitions, for nested classes.
"commented-out-code", # Found commented-out code
"boolean-positional-value-in-call", # Checks for boolean positional arguments in function calls.
"line-contains-todo", # Line contains TODO
"logging-percent-format", # Allow % in logging
"blanket-type-ignore", # Check for type: ignore annotations that suppress all type warnings, as opposed to targeting specific type warnings.
"no-self-use", # Checks for the presence of unused self parameter in methods definitions.
"unused-async", # Checks for functions declared async that do not await or otherwise use features requiring the function to be declared async.
"missing-todo-link", # Checks that a TODO comment is associated with a link to a relevant issue or ticket.
"too-many-arguments", # Checks for function definitions that include too many arguments.
"too-many-positional-arguments", # Checks for function definitions that include too many positional arguments.
# Conflicting lint rules when using Ruff's formatter
# https://docs.astral.sh/ruff/formatter/#conflicting-lint-rules
"COM812", # Checks for the absence of trailing commas.
"COM819", # Checks for the presence of prohibited trailing commas.
"D206", # Checks for docstrings that are indented with tabs.
"D300", # Checks for docstrings that use '''triple single quotes''' instead of """triple double quotes""".
"E111", # Checks for indentation with a non-multiple of 4 spaces.
"E114", # Checks for indentation of comments with a non-multiple of 4 spaces.
"E117", # Checks for over-indented code.
"ISC001", # Checks for implicitly concatenated strings on a single line.
"ISC002", # Checks for implicitly concatenated strings that span multiple lines.
"Q000", # Checks for inline strings that use single quotes or double quotes, depending on the value of the lint.flake8-quotes.inline-quotes option.
"Q001", # Checks for multiline strings that use single quotes or double quotes, depending on the value of the lint.flake8-quotes.multiline-quotes setting.
"Q002", # Checks for docstrings that use single quotes or double quotes, depending on the value of the lint.flake8-quotes.docstring-quotes setting.
"Q003", # Checks for strings that include escaped quotes, and suggests changing the quote style to avoid the need to escape them.
"W191", # Checks for indentation that uses tabs.
"missing-trailing-comma", # Checks for the absence of trailing commas.
"prohibited-trailing-comma", # Checks for the presence of prohibited trailing commas.
"docstring-tab-indentation", # Checks for docstrings that are indented with tabs.
"triple-single-quotes", # Checks for docstrings that use '''triple single quotes''' instead of """triple double quotes""".
"indentation-with-invalid-multiple", # Checks for indentation with a non-multiple of 4 spaces.
"indentation-with-invalid-multiple-comment", # Checks for indentation of comments with a non-multiple of 4 spaces.
"over-indented", # Checks for over-indented code.
"single-line-implicit-string-concatenation", # Checks for implicitly concatenated strings on a single line.
"multi-line-implicit-string-concatenation", # Checks for implicitly concatenated strings that span multiple lines.
"bad-quotes-inline-string", # Checks for inline strings that use single quotes or double quotes, depending on the value of the lint.flake8-quotes.inline-quotes option.
"bad-quotes-multiline-string", # Checks for multiline strings that use single quotes or double quotes, depending on the value of the lint.flake8-quotes.multiline-quotes setting.
"bad-quotes-docstring", # Checks for docstrings that use single quotes or double quotes, depending on the value of the lint.flake8-quotes.docstring-quotes setting.
"avoidable-escaped-quote", # Checks for strings that include escaped quotes, and suggests changing the quote style to avoid the need to escape them.
"tab-indentation", # Checks for indentation that uses tabs.
]
[tool.ruff.lint.per-file-ignores]
"tests/*" = ["S101", "D103", "PLR2004"]
"tests/*" = ["assert", "undocumented-public-function", "magic-value-comparison"]
[tool.pytest.ini_options]
addopts = "-n 5 --dist loadfile -m \"not integration and not slow\""

View file

@ -26,7 +26,7 @@ def test_encode_url() -> None:
assert encode_url("https://www.example.com/my path") == r"https%3A//www.example.com/my%20path", assert_msg
# Test input with special characters
assert_msg: str = f"Got: {encode_url('https://www.example.com/my path?q=abc&b=1')}, Expected: https%3A//www.example.com/my%20path%3Fq%3Dabc%26b%3D1" # noqa: E501
assert_msg: str = f"Got: {encode_url('https://www.example.com/my path?q=abc&b=1')}, Expected: https%3A//www.example.com/my%20path%3Fq%3Dabc%26b%3D1" # ruff:ignore[line-too-long]
assert (
encode_url("https://www.example.com/my path?q=abc&b=1")
== r"https%3A//www.example.com/my%20path%3Fq%3Dabc%26b%3D1"

View file

@ -14,6 +14,8 @@ from discord_rss_bot.custom_message import get_embed
from discord_rss_bot.custom_message import get_embed_data
from discord_rss_bot.custom_message import get_first_image
from discord_rss_bot.custom_message import get_image_urls
from discord_rss_bot.custom_message import normalize_message_avatar_url
from discord_rss_bot.custom_message import normalize_message_username
from discord_rss_bot.custom_message import replace_tags_in_embed
from discord_rss_bot.custom_message import replace_tags_in_text_message
from discord_rss_bot.custom_message import save_embed
@ -328,6 +330,43 @@ def test_replace_tags_in_embed_uses_last_content_item(
assert "New content" in embed.description
@patch("discord_rss_bot.custom_message.get_embed")
def test_replace_tags_in_embed_converts_escaped_newlines(
mock_get_embed: MagicMock,
) -> None:
r"""Verify \\n in embed fields becomes actual newlines, and fields without \\n are unchanged."""
mock_reader = MagicMock()
mock_get_embed.return_value = CustomEmbed(
title="Line 1\\nLine 2",
description="{{entry_text}}\\ntest\\test2",
author_name="Author\\nSubtitle",
footer_text="Footer\\n\\nMore",
)
entry_ns: SimpleNamespace = make_entry("<p>Summary</p>")
entry: Entry = typing.cast("Entry", entry_ns)
embed: CustomEmbed = replace_tags_in_embed(entry_ns.feed, entry, reader=mock_reader)
# Fields with \\n should have actual newlines
assert "\n" in embed.title
assert embed.title == "Line 1\nLine 2"
assert "\n" in embed.author_name
assert embed.author_name == "Author\nSubtitle"
assert "\n" in embed.footer_text
assert embed.footer_text == "Footer\n\nMore"
# Description combines tag replacement with \\n conversion.
# "{{entry_text}}\\ntest\\\\test2" becomes "Summary text\ntest\\test2".
assert "\n" in embed.description
assert embed.description == "Summary\ntest\\test2"
# Fields without \\n should be unchanged (backward compat) - none here but
# cover other fields that had no \\n originally
assert "\\n" not in embed.title
assert "\\n" not in embed.author_name
assert "\\n" not in embed.footer_text
def test_get_custom_message_returns_empty_string_on_value_error() -> None:
reader = MagicMock()
feed = make_feed()
@ -429,3 +468,69 @@ def test_get_embed_data_coerces_values_to_strings() -> None:
assert embed.title == "1"
assert embed.footer_icon_url == "10"
assert embed.show_steam_game_icon_in_thumbnail is True
class TestNormalizeMessageUsername:
"""Tests for normalize_message_username."""
def test_empty_returns_empty(self) -> None:
assert not normalize_message_username(None)
assert not normalize_message_username("")
def test_whitespace_only_returns_empty(self) -> None:
assert not normalize_message_username(" ")
assert not normalize_message_username("\t\n")
def test_valid_username_returns_stripped(self) -> None:
assert normalize_message_username(" Power Of The Shell ") == "Power Of The Shell"
def test_too_long_returns_empty(self) -> None:
long_name: str = "a" * 81
assert not normalize_message_username(long_name)
def test_max_length_allowed(self) -> None:
name: str = "a" * 80
assert normalize_message_username(name) == name
@pytest.mark.parametrize("char", ["@", "#", ":", "`"])
def test_forbidden_characters_returns_empty(self, char: str) -> None:
assert not normalize_message_username(f"valid{char}name")
@pytest.mark.parametrize("substring", ["clyde", "Clyde", "CLYDE", "discord", "Discord", "DISCOrd"])
def test_forbidden_substrings_returns_empty(self, substring: str) -> None:
assert not normalize_message_username(f"my_{substring}_name")
class TestNormalizeMessageAvatarUrl:
"""Tests for normalize_message_avatar_url."""
def test_empty_returns_empty(self) -> None:
assert not normalize_message_avatar_url(None)
assert not normalize_message_avatar_url("")
def test_whitespace_only_returns_empty(self) -> None:
assert not normalize_message_avatar_url(" ")
def test_valid_http_url_returns_stripped(self) -> None:
url: str = "http://example.com/icon.png"
assert normalize_message_avatar_url(url) == url
def test_valid_https_url_returns_stripped(self) -> None:
url: str = "https://cdn.example.com/images/avatar.jpg"
assert normalize_message_avatar_url(url) == url
def test_whitespace_around_url_is_stripped(self) -> None:
url: str = "https://example.com/icon.png"
assert normalize_message_avatar_url(f" {url} ") == url
def test_ftp_scheme_returns_empty(self) -> None:
assert not normalize_message_avatar_url("ftp://example.com/icon.png")
def test_no_scheme_returns_empty(self) -> None:
assert not normalize_message_avatar_url("example.com/icon.png")
def test_javascript_url_returns_empty(self) -> None:
assert not normalize_message_avatar_url("javascript:alert(1)")
def test_invalid_url_returns_empty(self) -> None:
assert not normalize_message_avatar_url("not-a-url")

View file

@ -177,7 +177,7 @@ def test_get_entry_delivery_mode_prefers_delivery_mode_tag() -> None:
entry = MagicMock()
entry.feed.url = "https://example.com/feed.xml"
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"delivery_mode": "screenshot",
"should_send_embed": True,
}.get(key, default)
@ -192,7 +192,7 @@ def test_get_entry_delivery_mode_falls_back_to_legacy_embed_flag() -> None:
entry = MagicMock()
entry.feed.url = "https://example.com/feed.xml"
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"delivery_mode": "",
"should_send_embed": False,
}.get(key, default)
@ -219,7 +219,7 @@ def test_send_entry_to_discord_hoyolab_text_mode_uses_text_webhook(
entry.link = "https://www.hoyolab.com/article/38588239"
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhook": "https://discord.test/webhook",
"delivery_mode": "text",
}.get(key, default)
@ -258,7 +258,7 @@ def test_send_entry_to_discord_hoyolab_screenshot_mode_uses_screenshot_webhook(
entry.link = "https://www.hoyolab.com/article/38588239"
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhook": "https://discord.test/webhook",
"delivery_mode": "screenshot",
}.get(key, default)
@ -296,7 +296,7 @@ def test_send_entry_to_discord_hoyolab_embed_mode_uses_hoyolab_webhook(
entry.link = "https://www.hoyolab.com/article/38588239"
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhook": "https://discord.test/webhook",
"delivery_mode": "embed",
}.get(key, default)
@ -372,7 +372,7 @@ def test_get_feed_media_gallery_image_limit_defaults_to_first_image() -> None:
reader = MagicMock()
feed = MagicMock()
feed.url = "https://example.com/feed.xml"
reader.get_tag.side_effect = lambda resource, key, default=None: default # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: default # ruff:ignore[unused-lambda-argument]
result = feeds.get_feed_media_gallery_image_limit(reader, feed)
@ -424,7 +424,7 @@ def test_get_feed_webhook_text_length_limit_defaults_to_discord_limit() -> None:
reader = MagicMock()
feed = MagicMock()
feed.url = "https://example.com/feed.xml"
reader.get_tag.side_effect = lambda resource, key, default=None: default # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: default # ruff:ignore[unused-lambda-argument]
result = feeds.get_feed_webhook_text_length_limit(reader, feed)
@ -433,7 +433,7 @@ def test_get_feed_webhook_text_length_limit_defaults_to_discord_limit() -> None:
def test_create_feed_inherits_global_screenshot_layout() -> None:
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
"screenshot_layout": "mobile",
}.get(key, default)
@ -445,7 +445,7 @@ def test_create_feed_inherits_global_screenshot_layout() -> None:
def test_create_feed_inherits_global_text_delivery_mode() -> None:
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
"screenshot_layout": "desktop",
"delivery_mode": "text",
@ -459,7 +459,7 @@ def test_create_feed_inherits_global_text_delivery_mode() -> None:
def test_create_feed_enables_sent_webhook_tracking_by_default() -> None:
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
"screenshot_layout": "desktop",
"delivery_mode": "embed",
@ -472,7 +472,7 @@ def test_create_feed_enables_sent_webhook_tracking_by_default() -> None:
def test_create_feed_sets_default_media_gallery_image_limit() -> None:
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
"screenshot_layout": "desktop",
"delivery_mode": "embed",
@ -489,7 +489,7 @@ def test_create_feed_sets_default_media_gallery_image_limit() -> None:
def test_create_feed_sets_default_webhook_text_length_limit() -> None:
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
"screenshot_layout": "desktop",
"delivery_mode": "embed",
@ -506,7 +506,7 @@ def test_create_feed_sets_default_webhook_text_length_limit() -> None:
def test_create_feed_inherits_global_webhook_text_length_limit() -> None:
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
"screenshot_layout": "desktop",
"delivery_mode": "embed",
@ -524,7 +524,7 @@ def test_create_feed_inherits_global_webhook_text_length_limit() -> None:
def test_create_feed_falls_back_to_embed_when_global_delivery_mode_is_invalid() -> None:
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
"screenshot_layout": "desktop",
"delivery_mode": "invalid",
@ -540,7 +540,7 @@ def test_create_feed_removes_new_feed_when_initial_update_fails() -> None:
feed_url = "https://example.com/not-a-feed"
autodiscover_links = [{"href": "https://example.com/feed.xml", "type": "application/rss+xml"}]
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
".reader.autodiscover": autodiscover_links,
}.get(key, default)
@ -556,7 +556,7 @@ def test_create_feed_removes_new_feed_when_initial_update_fails() -> None:
def test_create_feed_does_not_remove_existing_feed_when_update_fails() -> None:
feed_url = "https://example.com/existing-feed.xml"
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhooks": [{"name": "Main", "url": "https://discord.com/api/webhooks/123/abc"}],
}.get(key, default)
reader.add_feed.side_effect = FeedExistsError(feed_url)
@ -582,7 +582,7 @@ def test_create_screenshot_webhook_adds_image_file(
entry.id = "entry-abc"
entry.link = "https://example.com/article"
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"screenshot_layout": "mobile",
}.get(key, default)
@ -619,7 +619,7 @@ def test_create_screenshot_webhook_retries_jpeg_when_png_too_large(
entry.id = "entry-large"
entry.link = "https://example.com/large-article"
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"screenshot_layout": "desktop",
}.get(key, default)
@ -655,7 +655,7 @@ def test_create_screenshot_webhook_falls_back_when_all_formats_too_large(
entry.id = "entry-too-large"
entry.link = "https://example.com/very-large"
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"screenshot_layout": "desktop",
}.get(key, default)
@ -758,7 +758,7 @@ def test_create_text_webhook_uses_feed_text_length_limit(mock_replace_tags_in_te
entry = MagicMock()
entry.feed.url = "https://example.com/feed.xml"
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"custom_message": "{{entry_title}}",
"webhook_text_length_limit": 20,
}.get(key, default)
@ -784,7 +784,7 @@ def test_create_embed_webhook_uses_media_gallery_for_entry_images(
mock_fetch_ttvdrops_campaign_media_items: MagicMock,
) -> None:
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"media_gallery_image_limit": 10,
"webhook_text_length_limit": 4000,
}.get(key, default)
@ -830,7 +830,7 @@ def test_create_embed_webhook_uses_feed_text_length_limit_for_media_gallery(
mock_fetch_ttvdrops_campaign_media_items: MagicMock,
) -> None:
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"media_gallery_image_limit": 10,
"webhook_text_length_limit": 20,
}.get(key, default)
@ -862,7 +862,7 @@ def test_create_embed_webhook_can_limit_media_gallery_to_first_image(
mock_fetch_ttvdrops_campaign_media_items: MagicMock,
) -> None:
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"media_gallery_image_limit": 1,
"webhook_text_length_limit": 4000,
}.get(key, default)
@ -895,7 +895,7 @@ def test_create_embed_webhook_can_disable_media_images(
mock_fetch_ttvdrops_campaign_media_items: MagicMock,
) -> None:
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
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)
@ -931,7 +931,7 @@ def test_create_embed_webhook_can_use_steam_game_icon_thumbnail(
mock_fetch_ttvdrops_campaign_media_items: MagicMock,
) -> None:
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
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)
@ -971,7 +971,7 @@ def test_create_embed_webhook_prefers_local_steam_game_icon_thumbnail(
local_icon_bytes = b"local-steam-icon"
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
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)
@ -1013,7 +1013,7 @@ def test_create_embed_webhook_does_not_inject_steam_thumbnail_when_app_id_is_mis
mock_fetch_ttvdrops_campaign_media_items: MagicMock,
) -> None:
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
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)
@ -1047,7 +1047,7 @@ def test_create_embed_webhook_uses_feed_text_length_limit_for_regular_embed_desc
mock_fetch_ttvdrops_campaign_media_items: MagicMock,
) -> None:
reader = MagicMock()
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"media_gallery_image_limit": 0,
"webhook_text_length_limit": 20,
}.get(key, default)
@ -1294,7 +1294,7 @@ def test_send_entry_to_discord_uses_screenshot_mode(
entry.feed.url = "https://example.com/feed.xml"
entry.feed_url = "https://example.com/feed.xml"
reader.get_tag.side_effect = lambda resource, key, default=None: { # noqa: ARG005
reader.get_tag.side_effect = lambda resource, key, default=None: { # ruff:ignore[unused-lambda-argument]
"webhook": "https://discord.com/api/webhooks/123/abc",
}.get(key, default)
@ -1339,7 +1339,7 @@ def test_send_entry_to_discord_youtube_feed(
mock_entry.feed_url = "https://www.youtube.com/feeds/videos.xml?channel_id=123456"
# Mock the tags
mock_reader.get_tag.side_effect = lambda feed, tag, default=None: { # noqa: ARG005
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
}.get(tag, default)

View file

@ -3,7 +3,7 @@ from __future__ import annotations
import contextlib
import json
import shutil
import subprocess # noqa: S404
import subprocess # ruff:ignore[suspicious-subprocess-import]
from typing import TYPE_CHECKING
from typing import cast
from unittest.mock import MagicMock
@ -579,7 +579,7 @@ def test_embed_backup_end_to_end(monkeypatch: pytest.MonkeyPatch, tmp_path: Path
assert response.status_code == 200, f"Failed to customize embed: {response.text}"
# Verify a commit was created
result: subprocess.CompletedProcess[str] = subprocess.run( # noqa: S603
result: subprocess.CompletedProcess[str] = subprocess.run( # ruff:ignore[subprocess-without-shell-equals-true]
[git_executable, "-C", str(backup_path), "log", "--oneline"],
capture_output=True,
text=True,

View file

@ -279,16 +279,6 @@ def test_get() -> None:
response: Response = client.get(url="/feed", params={"feed_url": encoded_feed_url(feed_url)})
assert response.status_code == 200, f"/feed failed: {response.text}"
assert "Feed Summary" in response.text
assert "This feed" in response.text
assert "Screenshot Delivery" in response.text
assert "Image Delivery" in response.text
assert "Text Delivery" in response.text
assert "2000 characters" in response.text
assert 'type="range"' in response.text
assert 'max="10"' in response.text
assert 'id="text_length_limit"' in response.text
assert 'max="4000"' in response.text
response: Response = client.get(url="/")
assert response.status_code == 200, f"/ failed: {response.text}"
@ -735,8 +725,6 @@ def test_c3kay_feed_delivery_mode_toggle_routes_update_stored_tags() -> None:
assert reader.get_tag(c3kay_feed_url, "delivery_mode") == "screenshot"
assert reader.get_tag(c3kay_feed_url, "screenshot_layout") == "mobile"
assert reader.get_tag(c3kay_feed_url, "should_send_embed") is False
assert "Disable screenshot delivery" in response.text
assert "Send embed instead of screenshot" not in response.text
response = client.post(url="/use_embed", data={"feed_url": c3kay_feed_url})
assert response.status_code == 200, f"Failed to set embed mode: {response.text}"
@ -759,7 +747,7 @@ def test_set_feed_save_sent_webhooks_route_updates_stored_tag() -> None:
assert feed_url == self.feed.url
return self.feed
def set_tag(self, resource: str, key: str, value: bool) -> None: # noqa: FBT001
def set_tag(self, resource: str, key: str, value: bool) -> None: # ruff:ignore[boolean-type-hint-positional-argument]
self.tags[resource, key] = value
stub_reader = StubReader()
@ -2301,7 +2289,7 @@ def test_bulk_change_feed_urls_updates_matching_feeds() -> None:
def get_entries(self, **_kwargs: TestKwargValue) -> list[Entry]:
return []
def set_entry_read(self, _entry: Entry, _value: bool) -> None: # noqa: FBT001
def set_entry_read(self, _entry: Entry, _value: bool) -> None: # ruff:ignore[boolean-type-hint-positional-argument]
return
stub_reader = StubReader()
@ -2386,7 +2374,7 @@ def test_webhook_entries_mass_update_preview_fragment_endpoint() -> None:
app.dependency_overrides = {}
def test_bulk_change_feed_urls_force_update_overwrites_conflict() -> None: # noqa: C901
def test_bulk_change_feed_urls_force_update_overwrites_conflict() -> None: # ruff:ignore[complex-structure]
"""Force update should overwrite conflicting target URLs instead of skipping them."""
@dataclass(slots=True)
@ -2429,7 +2417,7 @@ def test_bulk_change_feed_urls_force_update_overwrites_conflict() -> None: # no
def get_entries(self, **_kwargs: TestKwargValue) -> list[Entry]:
return []
def set_entry_read(self, _entry: Entry, _value: bool) -> None: # noqa: FBT001
def set_entry_read(self, _entry: Entry, _value: bool) -> None: # ruff:ignore[boolean-type-hint-positional-argument]
return
stub_reader = StubReader()
@ -2503,7 +2491,7 @@ def test_bulk_change_feed_urls_force_update_ignores_resolution_error() -> None:
def get_entries(self, **_kwargs: TestKwargValue) -> list[Entry]:
return []
def set_entry_read(self, _entry: Entry, _value: bool) -> None: # noqa: FBT001
def set_entry_read(self, _entry: Entry, _value: bool) -> None: # ruff:ignore[boolean-type-hint-positional-argument]
return
stub_reader = StubReader()
@ -2903,10 +2891,13 @@ def test_post_set_custom_saves_message() -> None:
)
assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}: {response.text}"
stub.set_tag.assert_called_once()
_feed_arg, key_arg, value_arg = stub.set_tag.call_args.args
assert key_arg == "custom_message"
assert value_arg == "Hello {{entry_title}}!"
assert stub.set_tag.call_count == 3
# First two calls should be username/avatar
assert stub.set_tag.call_args_list[0].args[1] == "message_username"
assert stub.set_tag.call_args_list[1].args[1] == "message_avatar_url"
# Third call should be custom_message
assert stub.set_tag.call_args_list[2].args[1] == "custom_message"
assert stub.set_tag.call_args_list[2].args[2] == "Hello {{entry_title}}!"
finally:
app.dependency_overrides = {}
@ -2931,10 +2922,13 @@ def test_post_set_custom_allows_clearing_message() -> None:
)
assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}: {response.text}"
stub.set_tag.assert_called_once()
_feed_arg, key_arg, value_arg = stub.set_tag.call_args.args
assert key_arg == "custom_message"
assert not value_arg, f"Expected empty custom_message to be saved, got {value_arg!r}"
assert stub.set_tag.call_count == 3
# First two calls should be username/avatar
assert stub.set_tag.call_args_list[0].args[1] == "message_username"
assert stub.set_tag.call_args_list[1].args[1] == "message_avatar_url"
# Third call should be custom_message (empty, since it changed from stored)
assert stub.set_tag.call_args_list[2].args[1] == "custom_message"
assert not stub.set_tag.call_args_list[2].args[2], "Expected empty custom_message to be saved"
finally:
app.dependency_overrides = {}
@ -2957,7 +2951,10 @@ def test_post_set_custom_unchanged_message_does_not_write() -> None:
)
assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}: {response.text}"
stub.set_tag.assert_not_called()
# Username/avatar are always stored (even if blank), so expect those calls but not for custom_message
assert stub.set_tag.call_count == 2
keys = [c.args[1] for c in stub.set_tag.call_args_list]
assert keys == ["message_username", "message_avatar_url"]
finally:
app.dependency_overrides = {}
@ -2981,9 +2978,128 @@ def test_post_set_custom_clearing_from_default_message() -> None:
)
assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}: {response.text}"
stub.set_tag.assert_called_once()
_feed_arg, _key_arg, value_arg = stub.set_tag.call_args.args
# Must be "" not the default.
assert not value_arg, f"Expected empty string to be saved, got {value_arg!r}"
assert stub.set_tag.call_count == 3
# First two calls should be username/avatar
assert stub.set_tag.call_args_list[0].args[1] == "message_username"
assert stub.set_tag.call_args_list[1].args[1] == "message_avatar_url"
# Third call should be custom_message (empty, since it changed from stored)
assert stub.set_tag.call_args_list[2].args[1] == "custom_message"
assert not stub.set_tag.call_args_list[2].args[2], "Expected empty string to be saved"
finally:
app.dependency_overrides = {}
def test_post_set_custom_saves_username_and_avatar() -> None:
"""Saving a custom message with username and avatar should persist all fields."""
stub = _make_stub_reader_for_custom()
app.dependency_overrides[get_reader_dependency] = lambda: stub
try:
with patch("discord_rss_bot.main.commit_state_change"):
response: Response = client.post(
url="/custom",
data={
"feed_url": feed_url,
"custom_message": "Hello {{entry_title}}!",
"message_username": "My Bot",
"message_avatar_url": "https://example.com/avatar.png",
},
follow_redirects=False,
)
assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}: {response.text}"
assert stub.set_tag.call_count == 3
keys = [c.args[1] for c in stub.set_tag.call_args_list]
assert keys == ["message_username", "message_avatar_url", "custom_message"]
assert stub.set_tag.call_args_list[0].args[2] == "My Bot"
assert stub.set_tag.call_args_list[1].args[2] == "https://example.com/avatar.png"
assert stub.set_tag.call_args_list[2].args[2] == "Hello {{entry_title}}!"
finally:
app.dependency_overrides = {}
def test_post_set_custom_with_username_only() -> None:
"""Submitting with a username but no avatar should store the username."""
existing = "{{entry_title}}\n{{entry_link}}"
stub = _make_stub_reader_for_custom(stored_custom_message=existing)
app.dependency_overrides[get_reader_dependency] = lambda: stub
try:
with patch("discord_rss_bot.main.commit_state_change"):
response: Response = client.post(
url="/custom",
data={
"feed_url": feed_url,
"custom_message": existing,
"message_username": "New Name",
"message_avatar_url": "",
},
follow_redirects=False,
)
assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}: {response.text}"
assert stub.set_tag.call_count == 2
assert stub.set_tag.call_args_list[0].args[1] == "message_username"
assert stub.set_tag.call_args_list[0].args[2] == "New Name"
assert stub.set_tag.call_args_list[1].args[1] == "message_avatar_url"
assert not stub.set_tag.call_args_list[1].args[2]
finally:
app.dependency_overrides = {}
def test_post_set_custom_with_avatar_only() -> None:
"""Submitting with an avatar but no username should store the avatar."""
existing = "{{entry_title}}\n{{entry_link}}"
stub = _make_stub_reader_for_custom(stored_custom_message=existing)
app.dependency_overrides[get_reader_dependency] = lambda: stub
try:
with patch("discord_rss_bot.main.commit_state_change"):
response: Response = client.post(
url="/custom",
data={
"feed_url": feed_url,
"custom_message": existing,
"message_username": "",
"message_avatar_url": "https://example.com/icon.png",
},
follow_redirects=False,
)
assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}: {response.text}"
assert stub.set_tag.call_count == 2
assert stub.set_tag.call_args_list[0].args[1] == "message_username"
assert not stub.set_tag.call_args_list[0].args[2]
assert stub.set_tag.call_args_list[1].args[1] == "message_avatar_url"
assert stub.set_tag.call_args_list[1].args[2] == "https://example.com/icon.png"
finally:
app.dependency_overrides = {}
def test_post_set_custom_clears_username_and_avatar() -> None:
"""Clearing both identity fields should store empty strings."""
existing = "{{entry_title}}\n{{entry_link}}"
stub = _make_stub_reader_for_custom(stored_custom_message=existing)
app.dependency_overrides[get_reader_dependency] = lambda: stub
try:
with patch("discord_rss_bot.main.commit_state_change"):
response: Response = client.post(
url="/custom",
data={
"feed_url": feed_url,
"custom_message": existing,
"message_username": "",
"message_avatar_url": "",
},
follow_redirects=False,
)
assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}: {response.text}"
assert stub.set_tag.call_count == 2
assert stub.set_tag.call_args_list[0].args[1] == "message_username"
assert not stub.set_tag.call_args_list[0].args[2]
assert stub.set_tag.call_args_list[1].args[1] == "message_avatar_url"
assert not stub.set_tag.call_args_list[1].args[2]
finally:
app.dependency_overrides = {}

View file

@ -86,7 +86,7 @@ def test_data_dir() -> None:
def test_default_custom_message() -> None:
"""Test the default custom message."""
assert_msg = f"The default custom message should be '{{entry_title}}\n{{entry_link}}'. But it was '{default_custom_message}'." # noqa: E501
assert_msg = f"The default custom message should be '{{entry_title}}\n{{entry_link}}'. But it was '{default_custom_message}'." # ruff:ignore[line-too-long]
assert default_custom_message == "{{entry_title}}\n{{entry_link}}", assert_msg