TheLovinator/discord-rss-bot
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: TheLovinator:64a116d9478643141c6738c55361f8964bc83d52
Branches Tags
TheLovinator:master
TheLovinator:domain_filter
..
compare: TheLovinator:d61e8ccf1004b5951c5d640705501d1928d23ac5
Branches Tags
TheLovinator:master
TheLovinator:domain_filter

No commits in common. "64a116d9478643141c6738c55361f8964bc83d52" and "d61e8ccf1004b5951c5d640705501d1928d23ac5" have entirely different histories.
64a116d947 ... d61e8ccf10

6 changed files with 211 additions and 2200 deletions
Download patch file Download diff file Expand all files Collapse all files

664
discord_rss_bot/main.py
View file

@ -1,12 +1,9 @@
from __future__ import annotations
import concurrent.futures
import io
import json
import logging
import logging.config
import re
import tempfile
import typing
import urllib.parse
from contextlib import asynccontextmanager
@ -16,7 +13,6 @@ from datetime import datetime
from functools import lru_cache
from html import escape
from html import unescape
from pathlib import Path
from typing import TYPE_CHECKING
from typing import Annotated
from typing import TypedDict
@ -28,11 +24,9 @@ import uvicorn
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from fastapi import Depends
from fastapi import FastAPI
from fastapi import File
from fastapi import Form
from fastapi import HTTPException
from fastapi import Request
from fastapi import UploadFile
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
@ -43,14 +37,11 @@ from reader import Entry
from reader import EntryNotFoundError
from reader import Feed
from reader import FeedExistsError
from reader import FeedImportError
from reader import FeedNotFoundError
from reader import Reader
from reader import ReaderError
from reader import TagNotFoundError
from reader import opml
from starlette.responses import RedirectResponse
from starlette.responses import Response as StarletteResponse
from discord_rss_bot.custom_filters import entry_is_blacklisted
from discord_rss_bot.custom_filters import entry_is_whitelisted
@ -63,7 +54,6 @@ 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
from discord_rss_bot.feeds import JsonValue
from discord_rss_bot.feeds import SentWebhookRecord
from discord_rss_bot.feeds import coerce_media_gallery_image_limit
from discord_rss_bot.feeds import coerce_webhook_text_length_limit
@ -94,15 +84,12 @@ from discord_rss_bot.git_backup import commit_state_change
from discord_rss_bot.git_backup import get_backup_path
from discord_rss_bot.is_url_valid import is_url_valid
from discord_rss_bot.search import create_search_context
from discord_rss_bot.settings import data_dir
from discord_rss_bot.settings import default_custom_embed
from discord_rss_bot.settings import default_custom_message
from discord_rss_bot.settings import get_reader
from discord_rss_bot.settings import make_app_reader
if TYPE_CHECKING:
from collections.abc import AsyncGenerator
from collections.abc import Iterable
from pathlib import Path
from reader.types import JSONType
@ -287,196 +274,6 @@ templates.env.globals["get_backup_path"] = get_backup_path # pyright: ignore[re
templates.env.globals["has_webhooks"] = has_webhooks # pyright: ignore[reportArgumentType]
@app.get("/export_opml")
def export_opml(
reader: Annotated[Reader, Depends(get_reader_dependency)],
) -> StarletteResponse:
"""Export all feeds as an OPML subscription list.
Args:
reader: The Reader instance.
Returns:
StarletteResponse: The OPML file for download.
"""
export = reader.export_feeds()
return StarletteResponse(
content=export.content,
status_code=200,
headers={
"Content-Type": "application/xml",
"Content-Disposition": f'attachment; filename="{export.filename}"',
},
)
@app.post("/import_opml", response_model=None)
async def import_opml(
request: Request,
file: Annotated[UploadFile, File()],
reader: Annotated[Reader, Depends(get_reader_dependency)],
):
"""Upload an OPML file and show a preview of feeds to import.
Args:
request: The request object.
file: The uploaded OPML file.
reader: The Reader instance.
Returns:
HTMLResponse: The OPML import preview page.
RedirectResponse: Redirect to settings on error.
"""
if not file.filename or not file.filename.lower().endswith(".opml"):
return RedirectResponse(
url=f"/settings?message={urllib.parse.quote('Please upload a file with a .opml extension.')}",
status_code=303,
)
try:
content: bytes = await file.read()
feeds_to_import = opml.parse(io.BytesIO(content))
except FeedImportError as e:
return RedirectResponse(
url=f"/settings?message={urllib.parse.quote(f'Failed to parse OPML file: {e}')}",
status_code=303,
)
# Check which feeds already exist
existing_urls: set[str] = {feed.url for feed in reader.get_feeds()}
feed_list = [
{
"url": feed.url,
"title": feed.title or feed.url,
"already_exists": feed.url in existing_urls,
}
for feed in feeds_to_import
]
context = {
"request": request,
"feeds": feed_list,
"total": len(feed_list),
"new_count": sum(1 for f in feed_list if not f["already_exists"]),
"existing_count": sum(1 for f in feed_list if f["already_exists"]),
"webhooks": reader.get_tag((), "webhooks", []),
}
return templates.TemplateResponse(request=request, name="import_opml_preview.html", context=context)
@app.post("/import_opml_confirm")
async def import_opml_confirm(
request: Request,
reader: Annotated[Reader, Depends(get_reader_dependency)],
feed_urls: Annotated[list[str] | None, Form()] = None,
webhook_name: Annotated[str | None, Form()] = None,
) -> RedirectResponse:
"""Import the selected feeds from the OPML preview.
Args:
request: The request object.
reader: The Reader instance.
feed_urls: The selected feed URLs to import.
webhook_name: Optional webhook name to attach imported feeds to.
Returns:
RedirectResponse: Redirect to the settings page with a status message.
"""
if feed_urls is None:
feed_urls = []
if not feed_urls:
return RedirectResponse(
url="/settings?message=No%20feeds%20were%20selected%20for%20import.",
status_code=303,
)
webhook_url = _resolve_webhook_url(reader, webhook_name)
imported, updated_webhook, errors = _import_opml_feeds(reader, feed_urls, webhook_url)
message = _summarize_opml_import(imported, updated_webhook, errors)
logger.info("OPML import complete: %s", message)
commit_state_change(reader, f"OPML import: {imported} feeds")
return RedirectResponse(url=f"/settings?message={urllib.parse.quote(message)}", status_code=303)
def _resolve_webhook_url(reader: Reader, webhook_name: str | None) -> str:
"""Resolve a webhook name to its URL from reader storage.
Args:
reader: The Reader instance.
webhook_name: The webhook name to look up.
Returns:
The webhook URL, or empty string if not found or no name given.
"""
if not webhook_name:
return ""
hooks = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", [])))
for hook in hooks:
if hook.get("name") == webhook_name.strip():
return hook.get("url", "").strip()
return ""
def _import_opml_feeds(
reader: Reader,
feed_urls: list[str],
webhook_url: str,
) -> tuple[int, int, list[str]]:
"""Add feeds from an OPML import, optionally setting webhooks.
Args:
reader: The Reader instance.
feed_urls: The feed URLs to add.
webhook_url: Webhook URL to attach, or empty string.
Returns:
A tuple of (imported_count, updated_webhook_count, error_messages).
"""
imported: int = 0
updated_webhook: int = 0
errors: list[str] = []
for feed_url in feed_urls:
try:
reader.add_feed(feed_url)
if webhook_url:
reader.set_tag(feed_url, "webhook", webhook_url) # pyright: ignore[reportArgumentType]
imported += 1
except FeedExistsError:
if webhook_url:
reader.set_tag(feed_url, "webhook", webhook_url) # pyright: ignore[reportArgumentType]
updated_webhook += 1
except Exception as e:
errors.append(f"{feed_url}: {e}")
logger.exception("Failed to import feed: %s", feed_url)
return imported, updated_webhook, errors
def _summarize_opml_import(imported: int, updated_webhook: int, errors: list[str]) -> str:
"""Build a human-readable summary of an OPML import result.
Args:
imported: Number of newly imported feeds.
updated_webhook: Number of existing feeds whose webhook was updated.
errors: List of error strings.
Returns:
A summary string.
"""
parts: list[str] = []
if imported:
parts.append(f"Successfully imported {imported} feed{'s' if imported != 1 else ''}")
if updated_webhook:
parts.append(f"Updated webhook for {updated_webhook} existing feed{'s' if updated_webhook != 1 else ''}")
if errors:
parts.append(f"{len(errors)} error{'s' if len(errors) != 1 else ''}")
return ". ".join(parts) + "."
def get_global_delivery_mode(reader: Reader) -> str:
"""Return the normalized default delivery mode for new feeds.
@ -2332,14 +2129,12 @@ def get_data_from_hook_url(hook_name: str, hook_url: str) -> WebhookInfo:
async def get_settings(
request: Request,
reader: Annotated[Reader, Depends(get_reader_dependency)],
message: str = "",
):
"""Settings page.
Args:
request: The request object.
reader: The Reader instance.
message: Optional message to display to the user.
Returns:
HTMLResponse: The settings page.
@ -2391,7 +2186,6 @@ async def get_settings(
"max_webhook_text_length_limit": 4000,
"feed_intervals": feed_intervals,
"chromium_installed": is_chromium_installed(),
"messages": message or None,
}
return templates.TemplateResponse(request=request, name="settings.html", context=context)
@ -2597,33 +2391,6 @@ async def update_feed(
return RedirectResponse(url="/feed?feed_url=" + urllib.parse.quote(feed_url), status_code=303)
@app.get("/export")
def export_database(
reader: Annotated[Reader, Depends(get_reader_dependency)],
) -> StarletteResponse:
"""Export the entire database as a compressed SQLite file.
Args:
reader: The Reader instance.
Returns:
StarletteResponse: The exported database file for download.
"""
with tempfile.TemporaryDirectory() as tmpdir:
export_path: Path = reader._storage.export(tmpdir, "discord-rss-bot-export") # ruff:ignore[private-member-access]
filename: str = export_path.name
file_bytes: bytes = export_path.read_bytes()
return StarletteResponse(
content=file_bytes,
status_code=200,
headers={
"Content-Type": "application/gzip",
"Content-Disposition": f'attachment; filename="{filename}"',
},
)
@app.post("/backup")
async def manual_backup(
request: Request,
@ -2655,439 +2422,12 @@ async def manual_backup(
return RedirectResponse(url=f"/?message={urllib.parse.quote(message)}", status_code=303)
def _get_grouped_feeds(reader: Reader) -> list[dict[str, typing.Any]]:
"""Build a list of webhook groups with pre-computed indices for template use.
Each group dict contains:
name: The webhook name (or "Orphaned (no webhook)").
group_idx: 1-based index for the group.
feeds: List of dicts with:
feed: The Feed object.
feed_idx: 1-based index within the group.
Returns:
list[dict]: Grouped feeds with pre-computed indices for template rendering.
"""
hooks: list[dict[str, str]] = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", [])))
feeds_by_webhook: dict[str, list[Feed]] = {}
orphaned: list[Feed] = []
for feed in reader.get_feeds():
feed_webhook: str = str(reader.get_tag(feed.url, "webhook", ""))
hook_name: str = ""
for hook in hooks:
if hook["url"] == feed_webhook:
hook_name = hook["name"]
break
if hook_name:
feeds_by_webhook.setdefault(hook_name, []).append(feed)
else:
orphaned.append(feed)
grouped: list[dict[str, typing.Any]] = []
for group_idx, (name, feed_list) in enumerate(feeds_by_webhook.items(), start=1):
grouped.append({
"name": name,
"group_idx": group_idx,
"feeds": [{"feed": f, "feed_idx": idx} for idx, f in enumerate(feed_list, start=1)],
})
if orphaned:
grouped.append({
"name": "Orphaned (no webhook)",
"group_idx": len(grouped) + 1,
"feeds": [{"feed": f, "feed_idx": idx} for idx, f in enumerate(orphaned, start=1)],
})
return grouped
@app.get("/mass", response_class=HTMLResponse)
async def get_mass(
request: Request,
reader: Annotated[Reader, Depends(get_reader_dependency)],
active_tab: str = "create",
) -> HTMLResponse:
"""Mass operations page: create, delete, or modify feeds in bulk.
Args:
request: The request object.
reader: The Reader instance.
active_tab: The active tab (create, delete, modify).
Returns:
HTMLResponse: The mass operations page.
"""
hooks: list[dict[str, str]] = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", [])))
context: dict[str, typing.Any] = {
"request": request,
"webhooks": hooks,
"all_feeds_grouped": _get_grouped_feeds(reader),
"active_tab": active_tab,
}
return templates.TemplateResponse(request=request, name="mass.html", context=context)
def _create_and_tag_feed(reader: Reader, feed_url: str, webhook_url: str) -> str | None:
"""Add a feed and set its tags, without updating.
Returns:
The feed URL on success, or None if adding failed.
"""
clean_url: str = feed_url.strip()
try:
reader.add_feed(clean_url)
except FeedExistsError:
pass
except ReaderError:
return None
reader.set_tag(clean_url, "webhook", webhook_url) # pyright: ignore[reportArgumentType]
reader.set_tag(clean_url, "save_sent_webhooks", True) # pyright: ignore[reportArgumentType]
reader.set_tag(clean_url, "media_gallery_image_limit", cast("JSONType", 1))
global_webhook_text_length_limit: int = coerce_webhook_text_length_limit(
cast("JsonValue", reader.get_tag((), "webhook_text_length_limit", 4000)), # pyright: ignore[reportArgumentType]
)
reader.set_tag(clean_url, "webhook_text_length_limit", cast("JSONType", global_webhook_text_length_limit))
reader.set_tag(clean_url, "custom_message", default_custom_message) # pyright: ignore[reportArgumentType]
global_screenshot_layout: str = str(reader.get_tag((), "screenshot_layout", "desktop")).strip().lower()
if global_screenshot_layout not in {"desktop", "mobile"}:
global_screenshot_layout = "desktop"
reader.set_tag(clean_url, "screenshot_layout", global_screenshot_layout) # pyright: ignore[reportArgumentType]
global_delivery_mode: str = str(reader.get_tag((), "delivery_mode", "embed")).strip().lower()
if global_delivery_mode not in {"embed", "text"}:
global_delivery_mode = "embed"
reader.set_tag(clean_url, "delivery_mode", global_delivery_mode) # pyright: ignore[reportArgumentType]
reader.set_tag(clean_url, "should_send_embed", global_delivery_mode == "embed") # pyright: ignore[reportArgumentType]
reader.set_tag(clean_url, "embed", json.dumps(default_custom_embed)) # pyright: ignore[reportArgumentType]
return clean_url
def _modify_single_feed( # ruff:ignore[complex-structure, too-many-branches, too-many-statements]
reader: Reader,
url: str,
modify_action: str,
modify_value: str,
) -> dict[str, typing.Any]:
"""Apply a modification action to a single feed and return the result.
Args:
reader: The Reader instance.
url: The feed URL to modify.
modify_action: The action to perform.
modify_value: The value for the action.
Returns:
dict: Result with url, success, error, and action_taken keys.
"""
result: dict[str, typing.Any] = {
"url": url,
"success": False,
"error": "",
"action_taken": "",
}
try:
reader.get_feed(url)
except FeedNotFoundError:
result["error"] = "Feed not found"
return result
if modify_action == "pause":
reader.disable_feed_updates(url)
result["success"] = True
result["action_taken"] = "Paused"
elif modify_action == "unpause":
reader.enable_feed_updates(url)
result["success"] = True
result["action_taken"] = "Unpaused"
elif modify_action == "change_webhook":
webhooks: list[dict[str, str]] = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", [])))
webhook_url: str = ""
for hook in webhooks:
if hook["name"] == modify_value:
webhook_url = hook["url"]
break
if webhook_url:
reader.set_tag(url, "webhook", webhook_url) # pyright: ignore[reportArgumentType]
result["success"] = True
result["action_taken"] = f"Webhook changed to {modify_value}"
else:
result["error"] = f"Webhook '{modify_value}' not found"
elif modify_action == "delivery_mode":
if modify_value == "embed":
reader.set_tag(url, "delivery_mode", "embed") # pyright: ignore[reportArgumentType]
reader.set_tag(url, "should_send_embed", True) # pyright: ignore[reportArgumentType]
result["success"] = True
result["action_taken"] = "Delivery mode set to embed"
elif modify_value == "text":
reader.set_tag(url, "delivery_mode", "text") # pyright: ignore[reportArgumentType]
reader.set_tag(url, "should_send_embed", False) # pyright: ignore[reportArgumentType]
result["success"] = True
result["action_taken"] = "Delivery mode set to text"
elif modify_value == "screenshot_desktop":
reader.set_tag(url, "delivery_mode", "screenshot") # pyright: ignore[reportArgumentType]
reader.set_tag(url, "screenshot_layout", "desktop") # pyright: ignore[reportArgumentType]
result["success"] = True
result["action_taken"] = "Delivery mode set to screenshot (desktop)"
elif modify_value == "screenshot_mobile":
reader.set_tag(url, "delivery_mode", "screenshot") # pyright: ignore[reportArgumentType]
reader.set_tag(url, "screenshot_layout", "mobile") # pyright: ignore[reportArgumentType]
result["success"] = True
result["action_taken"] = "Delivery mode set to screenshot (mobile)"
else:
result["error"] = f"Unknown delivery mode: {modify_value}"
elif modify_action == "screenshot_layout":
if modify_value in {"desktop", "mobile"}:
reader.set_tag(url, "screenshot_layout", modify_value) # pyright: ignore[reportArgumentType]
result["success"] = True
result["action_taken"] = f"Screenshot layout set to {modify_value}"
else:
result["error"] = f"Unknown layout: {modify_value}"
elif modify_action == "update_interval":
try:
interval: int = int(modify_value)
except ValueError as e:
result["error"] = str(e)
return result
if interval < 1:
result["error"] = "Interval must be at least 1 minute"
else:
reader.set_tag(url, ".reader.update", {"interval": interval}) # pyright: ignore[reportArgumentType]
result["success"] = True
result["action_taken"] = f"Update interval set to {interval} minute(s)" # ruff:ignore[hardcoded-sql-expression]
else:
result["error"] = f"Unknown action: {modify_action}"
return result
def _update_and_mark_read(db_path: Path, feed_url: str) -> tuple[str, bool, str]:
"""Update a feed and mark entries as read, in its own reader instance.
Called from worker threads to parallelize HTTP fetches.
Returns:
(feed_url, success, error_message)
"""
worker_reader: Reader = make_app_reader(db_path)
try:
worker_reader.update_feed(feed_url)
for entry in worker_reader.get_entries(feed=feed_url, read=False):
worker_reader.set_entry_read(entry, True)
except ReaderError as e:
logger.warning("Failed to update feed %s: %s", feed_url, e)
return feed_url, False, str(e)[:200]
except Exception as e:
logger.exception("Unexpected error updating feed %s", feed_url)
return feed_url, False, str(e)[:200]
finally:
worker_reader.close()
return feed_url, True, ""
@app.post("/mass/create", response_class=HTMLResponse)
async def post_mass_create( # ruff:ignore[complex-structure]
request: Request,
feed_urls: Annotated[str, Form()],
webhook_dropdown: Annotated[str, Form()],
reader: Annotated[Reader, Depends(get_reader_dependency)],
) -> HTMLResponse:
"""Create multiple feeds at once.
Phase 1: Add feeds and set tags (sequential, fast, no HTTP).
Phase 2: Update feeds in parallel via a thread pool.
Args:
request: The request object.
feed_urls: Feed URLs (one per line).
webhook_dropdown: The webhook to attach feeds to.
reader: The Reader instance.
Returns:
HTMLResponse: The mass operations page with results.
Raises:
HTTPException: If the selected webhook is not found.
"""
urls: list[str] = [url.strip() for url in feed_urls.strip().split("\n") if url.strip()]
results: list[dict[str, typing.Any]] = []
webhooks: list[dict[str, str]] = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", [])))
# Resolve webhook name to URL once for all feeds
webhook_url: str = ""
for hook in webhooks:
if hook["name"] == webhook_dropdown:
webhook_url = hook["url"]
break
if not webhook_url:
raise HTTPException(status_code=404, detail="Webhook not found")
# Phase 1: Add feeds and set tags (sequential, fast)
urls_to_update: list[str] = []
for url in urls:
added_url: str | None = _create_and_tag_feed(reader, url, webhook_url)
if added_url:
urls_to_update.append(added_url)
results.append({"url": url, "success": False, "feed_url": None, "error": ""})
continue
results.append({"url": url, "success": False, "feed_url": None, "error": "Failed to add feed"})
# Phase 2: Update feeds in parallel
if urls_to_update:
db_path: Path = Path(data_dir) / "db.sqlite"
max_workers: int = min(10, len(urls_to_update))
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_url = {executor.submit(_update_and_mark_read, db_path, u): u for u in urls_to_update}
for future in concurrent.futures.as_completed(future_to_url):
original_url: str = future_to_url[future]
feed_url, update_success, error_msg = future.result()
# Find and update the matching result entry
for result in results:
if result["url"] == original_url:
result["success"] = update_success
result["feed_url"] = feed_url if update_success else None
result["error"] = error_msg
break
reader.update_search()
success_count: int = sum(1 for r in results if r["success"])
if success_count > 0:
commit_state_change(reader, f"Mass create {success_count} feed(s)")
context: dict[str, typing.Any] = {
"request": request,
"webhooks": webhooks,
"all_feeds_grouped": _get_grouped_feeds(reader),
"active_tab": "create",
"feed_urls": feed_urls,
"selected_webhook": webhook_dropdown,
"create_results": results,
"message": f"Created {success_count} of {len(results)} feed(s).",
}
return templates.TemplateResponse(request=request, name="mass.html", context=context)
@app.post("/mass/delete", response_class=HTMLResponse)
async def post_mass_delete(
request: Request,
reader: Annotated[Reader, Depends(get_reader_dependency)],
feed_urls: Annotated[list[str] | None, Form()] = None,
) -> HTMLResponse:
"""Delete multiple feeds at once.
Args:
request: The request object.
reader: The Reader instance.
feed_urls: List of feed URLs to delete.
Returns:
HTMLResponse: The mass operations page with results.
"""
if feed_urls is None:
feed_urls = []
results: list[dict[str, typing.Any]] = []
for url in feed_urls:
result: dict[str, typing.Any] = {"url": url, "success": False, "error": ""}
try:
reader.delete_feed(url)
result["success"] = True
except FeedNotFoundError:
result["error"] = "Feed not found"
except Exception as e: # ruff:ignore[blind-except]
result["error"] = str(e)[:200]
results.append(result)
deleted_count: int = sum(1 for r in results if r["success"])
if deleted_count > 0:
commit_state_change(reader, f"Mass delete {deleted_count} feed(s)")
webhooks: list[dict[str, str]] = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", [])))
failed_count: int = len(results) - deleted_count
context: dict[str, typing.Any] = {
"request": request,
"webhooks": webhooks,
"all_feeds_grouped": _get_grouped_feeds(reader),
"active_tab": "delete",
"delete_results": results,
"delete_summary": {"deleted": deleted_count, "failed": failed_count},
"message": f"Deleted {deleted_count} feed(s). {failed_count} failed.",
}
return templates.TemplateResponse(request=request, name="mass.html", context=context)
@app.post("/mass/modify", response_class=HTMLResponse)
async def post_mass_modify(
request: Request,
reader: Annotated[Reader, Depends(get_reader_dependency)],
feed_urls: Annotated[list[str] | None, Form()] = None,
modify_action: Annotated[str, Form()] = "",
modify_value: Annotated[str, Form()] = "",
) -> HTMLResponse:
"""Modify multiple feeds at once.
Args:
request: The request object.
reader: The Reader instance.
feed_urls: List of feed URLs to modify.
modify_action: The action to perform (pause, unpause, change_webhook, etc.).
modify_value: The value for the action.
Returns:
HTMLResponse: The mass operations page with results.
"""
feed_urls_list: list[str] = feed_urls if feed_urls is not None else []
results: list[dict[str, typing.Any]] = []
for url in feed_urls_list:
try:
result = _modify_single_feed(reader, url, modify_action, modify_value)
except Exception as e:
logger.exception("Failed to modify feed %s", url)
result = {"url": url, "success": False, "error": str(e)[:200], "action_taken": ""}
results.append(result)
modified_count: int = sum(1 for r in results if r["success"])
failed_count: int = sum(1 for r in results if not r["success"])
if modified_count > 0:
commit_state_change(reader, f"Mass modify {modified_count} feed(s)")
webhooks = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", [])))
context: dict[str, typing.Any] = {
"request": request,
"webhooks": webhooks,
"all_feeds_grouped": _get_grouped_feeds(reader),
"active_tab": "modify",
"modify_results": results,
"modify_action": modify_action,
"modify_value": modify_value,
"modify_summary": {
"modified": modified_count,
"failed": failed_count,
"skipped": 0,
},
"message": f"Modified {modified_count} feed(s). {failed_count} failed.",
}
return templates.TemplateResponse(request=request, name="mass.html", context=context)
@app.get("/search", response_class=HTMLResponse)
async def search(
request: Request,
query: str,
reader: Annotated[Reader, Depends(get_reader_dependency)],
) -> HTMLResponse:
):
"""Get entries matching a full-text search query.
Args:

117
discord_rss_bot/templates/import_opml_preview.html
View file

@ -1,117 +0,0 @@
{% extends "base.html" %}
{% block title %}
Import OPML Feeds | discord-rss-bot
{% endblock title %}
{% block description %}
Select which feeds to import from your OPML subscription list.
{% endblock description %}
{% block content %}
<section>
<div class="text-light">
<div class="d-flex flex-wrap justify-content-between align-items-center gap-2">
<h2 class="mb-0">Import OPML Feeds</h2>
</div>
<p class="text-muted mt-2 mb-4">
Found {{ total }} feed{{ 's' if total != 1 else '' }} in the OPML file.
{% if new_count %}
<span class="text-success">{{ new_count }} new</span>
{% endif %}
{% if existing_count %}
<span class="text-warning">{{ existing_count }} already exist{{ 's' if existing_count == 1 else '' }}</span>
{% endif %}
</p>
</div>
{% if feeds %}
<form action="/import_opml_confirm" method="post">
<!-- Webhook selector -->
{% if webhooks %}
<div class="mb-3">
<label for="webhook_name" class="form-label text-muted small mb-1">Attach imported feeds to webhook</label>
<select id="webhook_name"
name="webhook_name"
class="form-select bg-dark text-light border-secondary w-auto">
<option value="">No webhook (attach later)</option>
{% for hook in webhooks %}
<option value="{{ hook.name }}">{{ hook.name }}</option>
{% endfor %}
</select>
<div class="form-text text-muted mt-1">
New feeds will be added with this webhook. Existing feeds will have their webhook overwritten. You can change this later per-feed.
</div>
</div>
{% else %}
<div class="alert alert-warning mb-3" role="alert">
No webhooks configured. Imported feeds will need a webhook attached before they can send messages.
<a href="/add_webhook" class="alert-link">Add a webhook</a>
</div>
{% endif %}
<div class="table-responsive">
<table class="table table-dark table-hover">
<thead>
<tr>
<th>
<input type="checkbox"
id="select_all"
class="form-check-input"
checked
onchange="toggleCheckboxes(this)" />
</th>
<th>Feed</th>
<th>URL</th>
<th>Status</th>
</tr>
</thead>
<tbody>
{% for feed in feeds %}
<tr class="{{ 'text-muted' if feed.already_exists else '' }}">
<td>
<input type="checkbox"
name="feed_urls"
value="{{ feed.url }}"
class="form-check-input feed-checkbox"
checked />
</td>
<td>
<a href="{{ feed.url }}"
class="text-light text-decoration-none"
target="_blank"
rel="noopener noreferrer">{{ feed.title }}</a>
</td>
<td>
<a href="{{ feed.url }}"
class="text-muted small text-decoration-none"
target="_blank"
rel="noopener noreferrer"><code>{{ feed.url }}</code></a>
</td>
<td>
{% if feed.already_exists %}
<span class="badge bg-secondary">Already exists</span>
{% else %}
<span class="badge bg-success">New</span>
{% endif %}
</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
<div class="d-flex flex-wrap gap-2 mt-3">
<button type="submit" class="btn btn-primary px-4">Import Selected</button>
<a href="/settings" class="btn btn-outline-secondary px-4">Cancel</a>
</div>
</form>
{% else %}
<p class="text-muted">No feeds found in the OPML file.</p>
<a href="/settings" class="btn btn-primary px-4">Back to Settings</a>
{% endif %}
</section>
{% endblock content %}
{% block head %}
<script>
function toggleCheckboxes(selectAll) {
document.querySelectorAll('.feed-checkbox').forEach(function(checkbox) {
checkbox.checked = selectAll.checked;
});
}
</script>
{% endblock head %}

355
discord_rss_bot/templates/mass.html
View file

@ -1,355 +0,0 @@
{% extends "base.html" %}
{% block title %}
Mass operations | discord-rss-bot
{% endblock title %}
{% block description %}
Mass create, delete, or modify RSS feeds in bulk.
{% endblock description %}
{% block content %}
{% if message %}
<div class="alert alert-info alert-dismissible fade show" role="alert">
<pre>{{ message }}</pre>
<button type="button" class="btn-close" data-bs-dismiss="alert" aria-label="Close"></button>
</div>
{% endif %}
<ul class="nav nav-tabs nav-dark border-dark mb-3" id="massTabs" role="tablist">
<li class="nav-item" role="presentation">
<button class="nav-link text-muted {% if active_tab == 'create' %}active bg-dark text-light{% endif %}"
id="create-tab"
data-bs-toggle="tab"
data-bs-target="#create"
type="button"
role="tab"
aria-controls="create"
aria-selected="{% if active_tab == 'create' %}true{% else %}false{% endif %}">
Mass Create
</button>
</li>
<li class="nav-item" role="presentation">
<button class="nav-link text-muted {% if active_tab == 'delete' %}active bg-dark text-light{% endif %}"
id="delete-tab"
data-bs-toggle="tab"
data-bs-target="#delete"
type="button"
role="tab"
aria-controls="delete"
aria-selected="{% if active_tab == 'delete' %}true{% else %}false{% endif %}">
Mass Delete
</button>
</li>
<li class="nav-item" role="presentation">
<button class="nav-link text-muted {% if active_tab == 'modify' %}active bg-dark text-light{% endif %}"
id="modify-tab"
data-bs-toggle="tab"
data-bs-target="#modify"
type="button"
role="tab"
aria-controls="modify"
aria-selected="{% if active_tab == 'modify' %}true{% else %}false{% endif %}">
Mass Modify
</button>
</li>
</ul>
<div class="tab-content" id="massTabsContent">
{# ===== Mass Create ===== #}
<div class="tab-pane fade {% if active_tab == 'create' %}show active{% endif %}"
id="create"
role="tabpanel"
aria-labelledby="create-tab">
<div class="card border border-dark p-3 text-light">
<h3 class="h5">Create multiple feeds</h3>
<p class="text-muted small mb-3">Enter one feed URL per line. All feeds will be attached to the selected webhook.</p>
<form action="/mass/create" method="post">
<div class="mb-3">
<label for="feed_urls" class="form-label">Feed URLs (one per line)</label>
<textarea name="feed_urls"
id="feed_urls"
class="form-control bg-dark border-dark text-muted"
rows="10"
placeholder="https://example.com/feed.xml&#10;https://other.com/rss&#10;https://third.com/atom">{{ feed_urls or '' }}</textarea>
</div>
<div class="mb-3">
<label for="webhook_dropdown" class="form-label">Webhook</label>
<select class="form-select bg-dark border-dark text-muted"
id="webhook_dropdown"
name="webhook_dropdown"
required>
<option value="" {% if not selected_webhook %}selected{% endif %} disabled>Choose webhook...</option>
{% for hook in webhooks %}
<option value="{{ hook.name }}"
{% if hook.name == selected_webhook %}selected{% endif %}>{{ hook.name }}</option>
{% endfor %}
</select>
</div>
<button type="submit" class="btn btn-primary">Create all feeds</button>
</form>
</div>
{% if create_results %}
<div class="card border border-dark p-3 text-light mt-3">
<h3 class="h5">Results</h3>
<ul class="list-unstyled mb-0">
{% for result in create_results %}
<li class="mb-2 {% if result.success %}text-success{% else %}text-danger{% endif %}">
<strong>{{ result.url }}</strong>
{% if result.success %}
&check; Added
{% if result.feed_url %}
- <a class="text-success" href="/feed?feed_url={{ result.feed_url|encode_url }}">View feed</a>
{% endif %}
{% else %}
&cross; {{ result.error }}
{% endif %}
</li>
{% endfor %}
</ul>
</div>
{% endif %}
</div>
{# ===== Mass Delete ===== #}
<div class="tab-pane fade {% if active_tab == 'delete' %}show active{% endif %}"
id="delete"
role="tabpanel"
aria-labelledby="delete-tab">
<div class="card border border-dark p-3 text-light">
<h3 class="h5">Delete multiple feeds</h3>
<p class="text-muted small mb-3">Select feeds to delete. This action cannot be undone.</p>
{% if all_feeds_grouped %}
<form id="mass-delete-form" action="/mass/delete" method="post">
{% for group in all_feeds_grouped %}
<div class="mb-3 border border-dark p-2">
<div class="d-flex justify-content-between align-items-center mb-2">
<h4 class="h6 mb-0">{{ group.name }} ({{ group.feeds|length }})</h4>
<div>
<button type="button"
class="btn btn-outline-light btn-sm select-all"
data-group="{{ group.group_idx }}">Select all</button>
<button type="button"
class="btn btn-outline-secondary btn-sm deselect-all"
data-group="{{ group.group_idx }}">Deselect all</button>
</div>
</div>
<ul class="list-unstyled mb-0">
{% for item in group.feeds %}
{% set feed = item.feed %}
<li class="form-check">
<input class="form-check-input mass-delete-checkbox group-{{ group.group_idx }}"
type="checkbox"
name="feed_urls"
value="{{ feed.url }}"
id="delete-{{ group.group_idx }}-{{ item.feed_idx }}" />
<label class="form-check-label text-muted" for="delete-{{ group.group_idx }}-{{ item.feed_idx }}">
{% if feed.title %}
{{ feed.title }}
{% else %}
{{ feed.url }}
{% endif %}
{% if feed.title %}<span class="small">- {{ feed.url }}</span>{% endif %}
{% if not feed.updates_enabled %}<span class="text-warning small">Paused</span>{% endif %}
{% if feed.last_exception %}<span class="text-danger small">Error</span>{% endif %}
</label>
</li>
{% endfor %}
</ul>
</div>
{% endfor %}
<button type="submit"
class="btn btn-danger"
onclick="return confirm('Are you sure you want to delete the selected feeds? This cannot be undone.');">
Delete selected feeds
</button>
</form>
{% else %}
<p class="text-muted mb-0">No feeds found.</p>
{% endif %}
</div>
{% if delete_results %}
<div class="card border border-dark p-3 text-light mt-3">
<h3 class="h5">Results</h3>
<ul class="list-unstyled mb-0">
{% for result in delete_results %}
<li class="mb-2 {% if result.success %}text-success{% else %}text-danger{% endif %}">
<strong>{{ result.url }}</strong>
{% if result.success %}
&check; Deleted
{% else %}
&cross; {{ result.error }}
{% endif %}
</li>
{% endfor %}
<li class="mt-2 text-muted">{{ delete_summary.deleted }} deleted, {{ delete_summary.failed }} failed</li>
</ul>
</div>
{% endif %}
</div>
{# ===== Mass Modify ===== #}
<div class="tab-pane fade {% if active_tab == 'modify' %}show active{% endif %}"
id="modify"
role="tabpanel"
aria-labelledby="modify-tab">
<div class="card border border-dark p-3 text-light">
<h3 class="h5">Modify multiple feeds</h3>
<p class="text-muted small mb-3">Select feeds and choose a setting to update.</p>
{% if all_feeds_grouped %}
<form id="mass-modify-form" action="/mass/modify" method="post">
{% for group in all_feeds_grouped %}
<div class="mb-3 border border-dark p-2">
<div class="d-flex justify-content-between align-items-center mb-2">
<h4 class="h6 mb-0">{{ group.name }} ({{ group.feeds|length }})</h4>
<div>
<button type="button"
class="btn btn-outline-light btn-sm select-all"
data-group="modify-{{ group.group_idx }}">Select all</button>
<button type="button"
class="btn btn-outline-secondary btn-sm deselect-all"
data-group="modify-{{ group.group_idx }}">Deselect all</button>
</div>
</div>
<ul class="list-unstyled mb-0">
{% for item in group.feeds %}
{% set feed = item.feed %}
<li class="form-check">
<input class="form-check-input mass-modify-checkbox group-modify-{{ group.group_idx }}"
type="checkbox"
name="feed_urls"
value="{{ feed.url }}"
id="modify-{{ group.group_idx }}-{{ item.feed_idx }}" />
<label class="form-check-label text-muted"
for="modify-{{ group.group_idx }}-{{ item.feed_idx }}">
{% if feed.title %}
{{ feed.title }}
{% else %}
{{ feed.url }}
{% endif %}
{% if feed.title %}<span class="small">- {{ feed.url }}</span>{% endif %}
{% if not feed.updates_enabled %}<span class="text-warning small">Paused</span>{% endif %}
</label>
</li>
{% endfor %}
</ul>
</div>
{% endfor %}
<hr class="border-secondary" />
<div class="row g-3">
<div class="col-md-4">
<label for="modify_action" class="form-label">Action</label>
<select class="form-select bg-dark border-dark text-muted"
id="modify_action"
name="modify_action"
required>
<option value="" disabled {% if not modify_action %}selected{% endif %}>Choose action...</option>
<option value="pause" {% if modify_action == 'pause' %}selected{% endif %}>Pause (disable updates)</option>
<option value="unpause" {% if modify_action == 'unpause' %}selected{% endif %}>Unpause (enable updates)</option>
<option value="change_webhook" {% if modify_action == 'change_webhook' %}selected{% endif %}>Change webhook</option>
<option value="delivery_mode" {% if modify_action == 'delivery_mode' %}selected{% endif %}>Change delivery mode</option>
<option value="screenshot_layout" {% if modify_action == 'screenshot_layout' %}selected{% endif %}>Change screenshot layout</option>
<option value="update_interval" {% if modify_action == 'update_interval' %}selected{% endif %}>Change update interval (minutes)</option>
</select>
</div>
<div class="col-md-4" id="modify-value-group">
<label for="modify_value" class="form-label">Value</label>
<div id="modify-value-dynamic">
<input type="text"
name="modify_value"
id="modify_value"
class="form-control bg-dark border-dark text-muted"
value="{{ modify_value or '' }}"
placeholder="New value" />
</div>
</div>
</div>
<button type="submit" class="btn btn-warning mt-3">Apply to selected feeds</button>
</form>
{% else %}
<p class="text-muted mb-0">No feeds found.</p>
{% endif %}
</div>
{% if modify_results %}
<div class="card border border-dark p-3 text-light mt-3">
<h3 class="h5">Results</h3>
<ul class="list-unstyled mb-0">
{% for result in modify_results %}
<li class="mb-2 {% if result.success %}text-success{% else %}text-danger{% endif %}">
<strong>{{ result.url }}</strong>
{% if result.success %}
&check; {{ result.action_taken }}
{% else %}
&cross; {{ result.error }}
{% endif %}
</li>
{% endfor %}
<li class="mt-2 text-muted">
{{ modify_summary.modified }} modified, {{ modify_summary.failed }} failed, {{ modify_summary.skipped }} skipped
</li>
</ul>
</div>
{% endif %}
</div>
</div>
{% endblock content %}
{% block head %}
<script>
document.addEventListener('DOMContentLoaded', function() {
// Select/Deselect all for mass delete and modify checkboxes
document.querySelectorAll('.select-all').forEach(function(btn) {
btn.addEventListener('click', function() {
var group = this.getAttribute('data-group');
document.querySelectorAll('.group-' + group).forEach(function(cb) {
cb.checked = true;
});
});
});
document.querySelectorAll('.deselect-all').forEach(function(btn) {
btn.addEventListener('click', function() {
var group = this.getAttribute('data-group');
document.querySelectorAll('.group-' + group).forEach(function(cb) {
cb.checked = false;
});
});
});
// Dynamic value field for modify action
var actionSelect = document.getElementById('modify_action');
var valueDynamic = document.getElementById('modify-value-dynamic');
if (actionSelect && valueDynamic) {
function updateModifyValueField() {
var action = actionSelect.value;
var html = '';
if (action === 'pause' || action === 'unpause') {
html = '<p class="text-muted small mt-2">No additional value needed.</p>';
} else if (action === 'change_webhook') {
html = `<select name="modify_value" class="form-select bg-dark border-dark text-muted" required>
<option value="" disabled selected>Choose webhook...</option>
{% for hook in webhooks %}
<option value="{{ hook.name }}">{{ hook.name }}</option>
{% endfor %}
</select>`;
} else if (action === 'delivery_mode') {
html = `<select name="modify_value" class="form-select bg-dark border-dark text-muted" required>
<option value="" disabled selected>Choose mode...</option>
<option value="embed">Embed</option>
<option value="text">Text</option>
<option value="screenshot_desktop">Screenshot (Desktop)</option>
<option value="screenshot_mobile">Screenshot (Mobile)</option>
</select>`;
} else if (action === 'screenshot_layout') {
html = `<select name="modify_value" class="form-select bg-dark border-dark text-muted" required>
<option value="" disabled selected>Choose layout...</option>
<option value="desktop">Desktop</option>
<option value="mobile">Mobile</option>
</select>`;
} else if (action === 'update_interval') {
html = '<input type="number" name="modify_value" class="form-control bg-dark border-dark text-muted" required min="1" value="15" placeholder="Minutes" />';
} else {
html = '<input type="text" name="modify_value" class="form-control bg-dark border-dark text-muted" placeholder="Value" />';
}
valueDynamic.innerHTML = html;
}
actionSelect.addEventListener('change', updateModifyValueField);
// Init
if (actionSelect.value) {
updateModifyValueField();
}
}
});
</script>
{% endblock head %}

4
discord_rss_bot/templates/nav.html
View file

@ -18,10 +18,6 @@
</li>
<li class="nav-item nav-link d-none d-md-block">|</li>
{% endif %}
<li class="nav-item">
<a class="nav-link" href="/mass">Mass ops</a>
</li>
<li class="nav-item nav-link d-none d-md-block">|</li>
<li class="nav-item">
<a class="nav-link" href="/webhooks">Webhooks</a>
</li>

207
discord_rss_bot/templates/settings.html
View file

@ -6,47 +6,43 @@
Adjust default update intervals, delivery modes, screenshot layout, and webhook text limits for feeds managed by your bot.
{% endblock description %}
{% block content %}
<div class="row g-3 settings-page">
<!-- Global Defaults -->
<div class="col-12">
<article class="card border border-dark shadow-sm text-light rounded-0">
<div class="card-body p-3 p-md-4">
<div class="d-flex flex-wrap justify-content-between align-items-start gap-3">
<div>
<h2 class="h5 mb-0">Global Defaults</h2>
<section>
<div class="text-light">
<div class="d-flex flex-wrap justify-content-between align-items-center gap-2">
<h2 class="mb-0">Global Settings</h2>
</div>
</div>
<p class="text-muted small mt-2 mb-4">
These defaults apply to newly added feeds. Existing feeds keep their own per-feed settings.
<p class="text-muted mt-2 mb-4">
Set a default interval for all feeds. Individual feeds can still override this value.
</p>
<div class="row g-3">
<!-- Update Interval -->
<div class="col-md-6">
<form action="/set_global_update_interval" method="post">
<label for="interval_minutes" class="form-label text-muted small mb-1">Update interval (minutes)</label>
<div class="input-group input-group-sm">
<div class="mb-4">
<div>
Current default is {{ global_interval }} min.
Even though we check ETags and Last-Modified headers, choosing a very low interval may cause issues with some feeds or cause excessive load on the server hosting the feed. Remember to be kind.
</div>
</div>
</div>
<form action="/set_global_update_interval" method="post" class="mb-2">
<div class="settings-form-row mb-2">
<label for="interval_minutes" class="form-label mb-1">Default interval (minutes)</label>
<div class="input-group input-group-lg">
<input id="interval_minutes"
type="number"
class="form-control bg-dark text-light border-secondary"
class="form-control settings-input"
name="interval_minutes"
placeholder="Minutes"
min="1"
value="{{ global_interval }}"
required />
<button class="btn btn-primary" type="submit">Save</button>
<button class="btn btn-primary px-4" type="submit">Save</button>
</div>
<div class="form-text text-muted mt-1">
Currently {{ global_interval }} min. Low intervals may cause issues with some feeds.
</div>
</form>
</div>
<!-- Delivery Mode -->
<div class="col-md-6">
<form action="/set_global_delivery_mode" method="post">
<label for="delivery_mode" class="form-label text-muted small mb-1">Delivery mode</label>
<div class="input-group input-group-sm">
<form action="/set_global_delivery_mode" method="post" class="mt-4">
<div class="settings-form-row mb-2">
<label for="delivery_mode" class="form-label mb-1">Default delivery mode for new feeds</label>
<div class="input-group input-group-lg">
<select id="delivery_mode"
class="form-select bg-dark text-light border-secondary"
class="form-select settings-input"
name="delivery_mode">
<option value="embed"
{% if global_delivery_mode == "embed" %}selected{% endif %}>Embed</option>
@ -58,134 +54,73 @@
<option value="text"
{% if global_delivery_mode == "text" %}selected{% endif %}>Text</option>
</select>
<button class="btn btn-primary" type="submit">Save</button>
<button class="btn btn-primary px-4" type="submit">Save</button>
</div>
<div class="form-text text-muted mt-2">
New feeds inherit this value. Existing feeds keep their current delivery mode.
</div>
</div>
<div class="form-text text-muted mt-1">How new feeds send messages to Discord.</div>
</form>
</div>
<!-- Screenshot Layout -->
<div class="col-md-6">
<form action="/set_global_screenshot_layout" method="post">
<label for="screenshot_layout" class="form-label text-muted small mb-1">Screenshot layout</label>
<div class="input-group input-group-sm">
<form action="/set_global_screenshot_layout" method="post" class="mt-4">
<div class="settings-form-row mb-2">
<label for="screenshot_layout" class="form-label mb-1">Default screenshot layout for new feeds</label>
<div class="input-group input-group-lg">
<select id="screenshot_layout"
class="form-select bg-dark text-light border-secondary"
class="form-select settings-input"
name="screenshot_layout">
<option value="desktop"
{% if global_screenshot_layout == "desktop" %}selected{% endif %}>Desktop</option>
<option value="mobile"
{% if global_screenshot_layout == "mobile" %}selected{% endif %}>Mobile</option>
</select>
<button class="btn btn-primary" type="submit">Save</button>
<button class="btn btn-primary px-4" type="submit">Save</button>
</div>
<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 text-muted mt-1">
Default screenshot viewport for new feeds.
{% if not chromium_installed %}
<span class="d-block mt-1">
Requires Chromium. Run <code>uv run playwright install chromium</code>.
</span>
<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>
</div>
<!-- Text Length Limit -->
<div class="col-md-6">
<form action="/set_global_webhook_text_length_limit" method="post">
<label for="global_text_length_limit"
class="form-label text-muted small mb-1">Text length limit</label>
<div class="input-group input-group-sm">
<form action="/set_global_webhook_text_length_limit"
method="post"
class="mt-4">
<div class="settings-form-row mb-2">
<label for="global_text_length_limit" class="form-label mb-1">Default text limit for new feeds</label>
<div class="input-group input-group-lg">
<input id="global_text_length_limit"
type="number"
class="form-control bg-dark text-light border-secondary"
class="form-control settings-input"
name="text_length_limit"
min="1"
max="{{ max_webhook_text_length_limit }}"
value="{{ global_webhook_text_length_limit }}"
required />
<button class="btn btn-primary" type="submit">Save</button>
<button class="btn btn-primary px-4" type="submit">Save</button>
</div>
<div class="form-text text-muted mt-1">Max characters for text mode (4000). Embeds are capped at 2000.</div>
</form>
<div class="form-text text-muted mt-2">
New feeds inherit this value. Existing feeds keep their current
per-feed text limit. Text mode allows values up to 4000 characters. Embeds are capped at 2000 characters.
</div>
</div>
</div>
</article>
</div>
<!-- Data Management -->
<div class="col-12">
<article class="card border border-dark shadow-sm text-light rounded-0">
<div class="card-body p-3 p-md-4">
<div class="d-flex flex-wrap justify-content-between align-items-start gap-3">
<div>
<h2 class="h5 mb-0">Data Management</h2>
</div>
</div>
<p class="text-muted small mt-2 mb-4">Import and export your feeds and database.</p>
<div class="row g-3">
<!-- OPML Export -->
<div class="col-md-4">
<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">OPML Export</div>
<div class="card-body">
<p class="text-muted small mb-3">
Download all feeds as an OPML subscription list.
Supports title, links, and description.
</p>
<a href="/export_opml" class="btn btn-primary btn-sm" role="button">Export OPML</a>
</div>
</div>
</div>
<!-- OPML Import -->
<div class="col-md-4">
<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">OPML Import</div>
<div class="card-body">
<p class="text-muted small mb-3">Upload an OPML file to preview and select which feeds to import.</p>
<form action="/import_opml" method="post" enctype="multipart/form-data">
<div class="input-group input-group-sm">
<input type="file"
id="opml_file"
name="file"
accept=".opml"
class="form-control bg-dark text-light border-secondary"
required />
<button class="btn btn-primary" type="submit">Import</button>
</div>
</form>
</section>
<section class="mt-5">
<div class="text-light">
<div class="d-flex flex-wrap justify-content-between align-items-center gap-2">
<h2 class="mb-0">Feed Update Intervals</h2>
</div>
</div>
</div>
<!-- Database Export -->
<div class="col-md-4">
<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">
Database Export
</div>
<div class="card-body">
<p class="text-muted small mb-3">Download a compressed SQLite dump for backup or migration.</p>
<a href="/export" class="btn btn-primary btn-sm" role="button">Download Export</a>
</div>
</div>
</div>
</div>
</div>
</article>
</div>
<!-- Feed Update Intervals -->
<div class="col-12">
<article class="card border border-dark shadow-sm text-light rounded-0">
<div class="card-body p-3 p-md-4">
<div class="d-flex flex-wrap justify-content-between align-items-start gap-3">
<div>
<h2 class="h5 mb-0">Feed Update Intervals</h2>
</div>
</div>
<p class="text-muted small mt-2 mb-4">
Customize the update interval for individual feeds. Leave empty or reset to use the global default of {{ global_interval }} min.
<p class="text-muted mt-2 mb-4">
Customize the update interval for individual feeds. Leave empty or reset to use the global default.
</p>
</div>
{% if feed_intervals %}
<div class="table-responsive">
<table class="table table-dark table-hover table-sm mb-0">
<table class="table table-dark table-hover">
<thead>
<tr>
<th>Feed</th>
@ -193,10 +128,7 @@
<th>Status</th>
<th>Interval</th>
<th>Last Updated</th>
<th>
Next
Update
</th>
<th>Next Update</th>
<th>Set Interval (min)</th>
<th>Actions</th>
</tr>
@ -231,13 +163,13 @@
<small class="text-muted">{{ item.feed.update_after | relative_time }}</small>
</td>
<td>
<form action="/set_update_interval" method="post" class="d-flex gap-1">
<form action="/set_update_interval" method="post" class="d-flex gap-2">
<input type="hidden" name="feed_url" value="{{ item.feed.url }}" />
<input type="hidden" name="redirect_to" value="/settings" />
<input type="number"
class="form-control form-control-sm bg-dark text-light border-secondary w-auto d-inline-block"
class="form-control form-control-sm interval-input"
name="interval_minutes"
placeholder="Min"
placeholder="Minutes"
min="1"
value="{{ item.interval if item.interval else global_interval }}" />
<button class="btn btn-primary btn-sm" type="submit">Set</button>
@ -258,10 +190,7 @@
</table>
</div>
{% else %}
<p class="text-muted small mb-0">No feeds added yet.</p>
<p class="text-muted">No feeds added yet.</p>
{% endif %}
</div>
</article>
</div>
</div>
</section>
{% endblock content %}

838
tests/test_main.py
View file

@ -14,10 +14,7 @@ from typing import cast
from unittest.mock import MagicMock
from unittest.mock import patch
import pytest
from fastapi.testclient import TestClient
from reader import FeedExistsError
from reader import FeedNotFoundError
import discord_rss_bot.main as main_module
from discord_rss_bot import feeds
@ -28,6 +25,7 @@ from discord_rss_bot.main import get_reader_dependency
if TYPE_CHECKING:
from pathlib import Path
import pytest
from httpx2 import Response
from reader import Entry
from reader import Reader
@ -648,38 +646,44 @@ def test_author_templates_render_authors_str() -> None:
assert "Legacy Entry Author" not in filter_preview_html
def test_settings_page_loads() -> None:
"""The settings page should render without errors."""
def test_settings_page_shows_screenshot_layout_setting() -> None:
response: Response = client.get(url="/settings")
assert response.status_code == 200, f"/settings failed: {response.text}"
assert "Default delivery mode for new feeds" in response.text
assert "Default screenshot layout for new feeds" in response.text
assert "2000 characters" in response.text
assert 'id="global_text_length_limit"' in response.text
assert "uv run playwright install chromium" in response.text
def test_set_global_delivery_mode_stores_value() -> None:
"""POST /set_global_delivery_mode should persist the delivery mode in reader tags."""
def test_set_global_delivery_mode() -> None:
response: Response = client.post(url="/set_global_delivery_mode", data={"delivery_mode": "text"})
assert response.status_code == 200, f"Failed to set global delivery mode: {response.text}"
reader: Reader = get_reader_dependency()
assert reader.get_tag((), "delivery_mode", "") == "text"
response = client.get(url="/settings")
assert response.status_code == 200, f"/settings failed after setting delivery mode: {response.text}"
assert re.search(r"<option\s+value=\"text\"[^>]*\bselected\b", response.text)
def test_set_global_webhook_text_length_limit_stores_value() -> None:
"""POST /set_global_webhook_text_length_limit should persist the limit in reader tags."""
def test_set_global_webhook_text_length_limit() -> None:
response: Response = client.post(
url="/set_global_webhook_text_length_limit",
data={"text_length_limit": "2500"},
)
assert response.status_code == 200, f"Failed to set global webhook text length limit: {response.text}"
reader: Reader = get_reader_dependency()
assert reader.get_tag((), "webhook_text_length_limit", 0) == 2500
response = client.get(url="/settings")
assert response.status_code == 200, f"/settings failed after setting webhook text length limit: {response.text}"
assert 'value="2500"' in response.text
def test_set_global_delivery_mode_affects_add_page() -> None:
"""Setting the delivery mode should be reflected in the response from /add."""
reader: Reader = get_reader_dependency()
reader.set_tag((), "delivery_mode", "text") # pyright: ignore[reportArgumentType]
def test_add_page_shows_global_default_delivery_mode_hint() -> None:
response: Response = client.post(url="/set_global_delivery_mode", data={"delivery_mode": "text"})
assert response.status_code == 200, f"Failed to set global delivery mode: {response.text}"
response: Response = client.get(url="/add")
response = client.get(url="/add")
assert response.status_code == 200, f"/add failed: {response.text}"
assert "text" in response.text
def test_navbar_add_feed_visible_only_when_webhooks_exist() -> None:
@ -895,12 +899,13 @@ def test_sent_webhooks_view_shows_saved_records() -> None:
app.dependency_overrides = {}
def test_set_global_screenshot_layout_stores_value() -> None:
"""POST /set_global_screenshot_layout should persist the layout in reader tags."""
def test_set_global_screenshot_layout() -> None:
response: Response = client.post(url="/set_global_screenshot_layout", data={"screenshot_layout": "mobile"})
assert response.status_code == 200, f"Failed to set global screenshot layout: {response.text}"
reader: Reader = get_reader_dependency()
assert reader.get_tag((), "screenshot_layout", "") == "mobile"
response = client.get(url="/settings")
assert response.status_code == 200, f"/settings failed after setting layout: {response.text}"
assert re.search(r"<option\s+value=\"mobile\"[^>]*\bselected\b", response.text)
def test_pause_feed() -> None:
@ -2549,268 +2554,10 @@ def test_reader_dependency_override_is_used() -> None:
# ---------------------------------------------------------------------------
# Tests for OPML import / export
# Tests for post_embed — saving embed fields (including clearing to "")
# ---------------------------------------------------------------------------
def test_export_opml_returns_opml_content() -> None:
"""GET /export_opml should return an OPML XML file."""
reader: Reader = get_reader_dependency()
with contextlib.suppress(Exception):
reader.add_feed(feed_url)
response: Response = client.get("/export_opml")
assert response.status_code == 200, f"/export_opml failed: {response.text}"
assert response.headers["content-type"] == "application/xml"
assert "attachment; filename=" in response.headers["content-disposition"]
assert response.headers["content-disposition"].startswith("attachment")
assert ".opml" in response.headers["content-disposition"]
assert b"<opml" in response.content
assert b"<body>" in response.content
assert feed_url.encode() in response.content or b"lovinator.space" in response.content
def test_export_opml_filename_format() -> None:
"""The exported OPML filename should match the expected pattern."""
response: Response = client.get("/export_opml")
assert response.status_code == 200
disposition: str = response.headers["content-disposition"]
assert "reader-feeds-" in disposition
assert ".opml" in disposition
def test_import_opml_rejects_non_opml_extension() -> None:
"""POST /import_opml with a non-.opml file should redirect with an error."""
response: Response = client.post(
url="/import_opml",
files={"file": ("feeds.txt", b"not opml", "text/plain")},
follow_redirects=False,
)
assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}"
assert ".opml" in response.headers.get("location", "").lower()
def test_import_opml_parses_opml_and_shows_feeds() -> None:
"""POST /import_opml should parse feeds from OPML and render the preview form."""
opml_content: bytes = b"""<?xml version="1.0" encoding="UTF-8"?>
<opml version="1.0">
<head><title>test</title></head>
<body>
<outline type="rss" xmlUrl="https://example.com/feed.xml" title="Example Feed"/>
</body>
</opml>
"""
response: Response = client.post(
url="/import_opml",
files={"file": ("feeds.opml", opml_content, "application/xml")},
follow_redirects=False,
)
assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}"
assert "https://example.com/feed.xml" in response.text
assert 'action="/import_opml_confirm"' in response.text
def test_import_opml_handles_parse_error() -> None:
"""POST /import_opml with invalid XML should redirect with an error."""
response: Response = client.post(
url="/import_opml",
files={"file": ("bad.opml", b"not valid opml", "application/xml")},
follow_redirects=False,
)
assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}: {response.text}"
location = response.headers.get("location", "")
assert "Failed to parse OPML file" in urllib.parse.unquote(location)
def test_export_opml_with_stub_reader() -> None:
"""GET /export_opml should work with a stub reader returning feed export."""
@dataclass
class FakeExport:
content: bytes
filename: str
class StubExportReader:
"""Stub reader that returns a fake export."""
def export_feeds(self) -> FakeExport:
return FakeExport(
content=(
b'<?xml version="1.0" encoding="UTF-8"?>'
b'<opml version="1.0"><head><title>reader feeds</title></head><body></body></opml>'
),
filename="reader-feeds-2026-07-18-12-00-00.opml",
)
def get_tag(self, _resource: tuple, _key: str, default: object = None) -> object:
return default
def get_feeds(self) -> list:
return []
stub = StubExportReader()
app.dependency_overrides[get_reader_dependency] = lambda: stub
try:
response: Response = client.get("/export_opml")
assert response.status_code == 200, f"/export_opml with stub failed: {response.text}"
assert response.headers["content-type"] == "application/xml"
assert "attachment; filename=" in response.headers["content-disposition"]
assert ".opml" in response.headers["content-disposition"]
finally:
app.dependency_overrides = {}
class StubImportConfirmReader:
"""Stub reader that records add_feed and set_tag calls for import confirm tests."""
def __init__(self) -> None:
"""Initialize the stub with empty tracking lists."""
self.added_urls: list[str] = []
self.tags: dict[tuple[str, str], str] = {}
def get_tag(self, _resource: tuple, _key: str, default: object = None) -> object:
"""Stub get_tag.
Returns:
The default value.
"""
return default
def get_feeds(self) -> list:
"""Stub get_feeds.
Returns:
An empty list.
"""
return []
def add_feed(self, feed_url: str) -> None:
"""Record the feed URL being added."""
self.added_urls.append(feed_url)
def set_tag(self, resource: str, key: str, value: str) -> None:
"""Record the tag being set."""
self.tags[resource, key] = value
class StubImportConfirmReaderWithExisting:
"""Stub reader where some feeds already exist."""
def __init__(self) -> None:
"""Initialize the stub with empty tracking lists."""
self.added_urls: list[str] = []
self.tags: dict[tuple[str, str], str] = {}
def get_tag(self, _resource: tuple, _key: str, default: object = None) -> object:
"""Stub get_tag.
Returns:
The default value.
"""
if _key == "webhooks":
return [{"name": "Test Webhook", "url": "https://discord.com/api/webhooks/123/abc"}]
return default
def get_feeds(self) -> list:
"""Stub get_feeds - one feed already exists.
Returns:
A list with one existing feed.
"""
return [type("FakeFeed", (), {"url": "https://example.com/existing.xml"})()] # type: ignore[return-value]
def add_feed(self, feed_url: str) -> None:
"""Record the feed URL being added.
Raises:
FeedExistsError: If the feed URL already exists.
"""
if feed_url == "https://example.com/existing.xml":
raise FeedExistsError(feed_url)
self.added_urls.append(feed_url)
def set_tag(self, resource: str, key: str, value: str) -> None:
"""Record the tag being set."""
self.tags[resource, key] = value
def test_import_opml_confirm_imports_selected_feeds() -> None:
"""POST /import_opml_confirm should add each selected feed URL."""
stub = StubImportConfirmReader()
app.dependency_overrides[get_reader_dependency] = lambda: stub
try:
with patch("discord_rss_bot.main.commit_state_change"):
response: Response = client.post(
url="/import_opml_confirm",
data={"feed_urls": ["https://example.com/feed1.xml", "https://example.com/feed2.xml"]},
follow_redirects=False,
)
assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}: {response.text}"
assert stub.added_urls == ["https://example.com/feed1.xml", "https://example.com/feed2.xml"]
location = response.headers.get("location", "")
assert "Successfully imported 2 feeds" in urllib.parse.unquote(location)
finally:
app.dependency_overrides = {}
def test_import_opml_confirm_no_selection() -> None:
"""POST /import_opml_confirm with no selection should redirect with a message."""
stub = StubImportConfirmReader()
app.dependency_overrides[get_reader_dependency] = lambda: stub
try:
response: Response = client.post(
url="/import_opml_confirm",
data={},
follow_redirects=False,
)
assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}: {response.text}"
location = response.headers.get("location", "")
assert "No feeds were selected" in urllib.parse.unquote(location)
assert stub.added_urls == []
finally:
app.dependency_overrides = {}
def test_import_opml_confirm_updates_webhook_on_existing_feeds() -> None:
"""POST /import_opml_confirm should update webhook on feeds that already exist."""
stub = StubImportConfirmReaderWithExisting()
app.dependency_overrides[get_reader_dependency] = lambda: stub
try:
with patch("discord_rss_bot.main.commit_state_change"):
response: Response = client.post(
url="/import_opml_confirm",
data={
"feed_urls": ["https://example.com/existing.xml", "https://example.com/new.xml"],
"webhook_name": "Test Webhook",
},
follow_redirects=False,
)
assert response.status_code == 303, f"Expected 303 redirect, got {response.status_code}: {response.text}"
assert stub.added_urls == ["https://example.com/new.xml"]
# Existing feed should have its webhook updated
assert (
stub.tags.get(("https://example.com/existing.xml", "webhook")) == "https://discord.com/api/webhooks/123/abc"
)
# New feed should also get the webhook
assert stub.tags.get(("https://example.com/new.xml", "webhook")) == "https://discord.com/api/webhooks/123/abc"
location = response.headers.get("location", "")
decoded = urllib.parse.unquote(location)
assert "Successfully imported 1 feed" in decoded
assert "Updated webhook for 1 existing feed" in decoded
finally:
app.dependency_overrides = {}
def _make_stub_reader_for_embed(
*,
stored_embed: str | None = None,
@ -3356,532 +3103,3 @@ def test_post_set_custom_clears_username_and_avatar() -> None:
assert not stub.set_tag.call_args_list[1].args[2]
finally:
app.dependency_overrides = {}
# ---------------------------------------------------------------------------
# Mass operations tests
# ---------------------------------------------------------------------------
class _StubFeedForMass:
"""Minimal feed stub for mass operations tests."""
def __init__(self, url: str, title: str | None = None, updates_enabled: bool = True) -> None: # ruff:ignore[boolean-default-value-positional-argument, boolean-type-hint-positional-argument]
self.url: str = url
self.title: str | None = title
self.updates_enabled: bool = updates_enabled
self.last_exception: str | None = None
class _StubReaderForMass:
"""Stub reader with feeds grouped under a webhook, used by mass operations tests."""
def __init__(
self,
feeds: list[_StubFeedForMass] | None = None,
*,
webhook_name: str = "",
webhook_url: str = "",
delete_fail: bool = False,
) -> None:
self._feeds: list[_StubFeedForMass] = feeds or []
self._webhook_name: str = webhook_name
self._webhook_url: str = webhook_url
self._delete_fail: bool = delete_fail
self.set_tag_calls: list[tuple[str, str, object]] = []
self.disabled_feeds: set[str] = set()
self.enabled_feeds: set[str] = set()
def get_feeds(self) -> list[_StubFeedForMass]:
"""Return the list of feeds."""
return self._feeds
def get_tag(self, resource: tuple[()] | str, key: str, default: TestTagValue = None) -> TestTagValue: # ruff:ignore[too-many-return-statements]
"""Return stored tag values."""
if resource == () and key == "webhooks":
if self._webhook_name:
return [{"name": self._webhook_name, "url": self._webhook_url}]
return []
if resource == () and key == "delivery_mode":
return "embed"
if resource == () and key == "webhook_text_length_limit":
return 4000
if resource == () and key == "screenshot_layout":
return "desktop"
if isinstance(resource, str) and key == "webhook":
return self._webhook_url
return default
def set_tag(self, resource: str, key: str, value: object) -> None: # pyright: ignore[reportArgumentType]
"""Record tag set calls."""
self.set_tag_calls.append((resource, key, value))
def update_search(self) -> None:
"""No-op search index update."""
def delete_feed(self, url: str) -> None:
"""Delete a feed by URL.
Raises:
RuntimeError: If delete_fail was set in the constructor.
"""
if self._delete_fail:
msg = "Delete failed"
raise RuntimeError(msg)
self._feeds = [f for f in self._feeds if f.url != url]
def get_feed(self, url: str) -> _StubFeedForMass:
"""Get a feed by URL.
Returns:
The matching feed.
Raises:
FeedNotFoundError: If no feed matches the URL.
"""
for feed in self._feeds:
if feed.url == url:
return feed
raise FeedNotFoundError(url)
def disable_feed_updates(self, url: str) -> None:
"""Disable updates for a feed."""
self.disabled_feeds.add(url)
self.enabled_feeds.discard(url)
def enable_feed_updates(self, url: str) -> None:
"""Enable updates for a feed."""
self.enabled_feeds.add(url)
self.disabled_feeds.discard(url)
def add_feed(self, url: str) -> None:
"""Add a feed to the stub."""
self._feeds.append(_StubFeedForMass(url=url))
MASS_FEED_1_URL: str = "https://example.com/feed1.xml"
MASS_FEED_2_URL: str = "https://example.com/feed2.xml"
MASS_WEBHOOK_NAME: str = "Mass test webhook"
MASS_WEBHOOK_URL: str = "https://discord.com/api/webhooks/mass/abc"
@pytest.fixture
def mass_stub_reader() -> _StubReaderForMass:
"""Create a stub reader with feeds for mass operations tests.
Returns:
A configured stub reader instance.
"""
return _StubReaderForMass(
feeds=[
_StubFeedForMass(url=MASS_FEED_1_URL, title="Feed One"),
_StubFeedForMass(url=MASS_FEED_2_URL, title="Feed Two", updates_enabled=False),
],
webhook_name=MASS_WEBHOOK_NAME,
webhook_url=MASS_WEBHOOK_URL,
)
class TestMassPage:
"""Tests for the mass operations page."""
def test_mass_page_loads(self, mass_stub_reader: _StubReaderForMass) -> None:
"""GET /mass should render all three tabs."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
response: Response = client.get("/mass")
assert response.status_code == 200
assert "Mass Create" in response.text
assert "Mass Delete" in response.text
assert "Mass Modify" in response.text
finally:
app.dependency_overrides = {}
def test_mass_page_shows_feeds_when_webhook_exists(self, mass_stub_reader: _StubReaderForMass) -> None:
"""GET /mass should list feeds grouped by webhook."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
response: Response = client.get("/mass")
assert response.status_code == 200
assert MASS_WEBHOOK_NAME in response.text
assert "Feed One" in response.text
assert "Feed Two" in response.text
finally:
app.dependency_overrides = {}
def test_mass_page_no_feeds_shows_empty_state(self) -> None:
"""GET /mass with no feeds should show 'No feeds found'."""
stub = _StubReaderForMass(webhook_name=MASS_WEBHOOK_NAME, webhook_url=MASS_WEBHOOK_URL)
app.dependency_overrides[get_reader_dependency] = lambda: stub
try:
response: Response = client.get("/mass")
assert response.status_code == 200
assert "No feeds found" in response.text
finally:
app.dependency_overrides = {}
def test_mass_page_defaults_to_create_tab(self, mass_stub_reader: _StubReaderForMass) -> None:
"""GET /mass without active_tab should default to create tab."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
response: Response = client.get("/mass")
assert response.status_code == 200
assert 'id="create-tab"' in response.text
assert 'aria-selected="true"' in response.text
finally:
app.dependency_overrides = {}
def test_mass_page_respects_active_tab_param(self, mass_stub_reader: _StubReaderForMass) -> None:
"""GET /mass?active_tab=delete should show delete tab active."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
response: Response = client.get("/mass?active_tab=delete")
assert response.status_code == 200
assert "Delete multiple feeds" in response.text
finally:
app.dependency_overrides = {}
def test_mass_page_shows_webhooks_in_create_form(self, mass_stub_reader: _StubReaderForMass) -> None:
"""The mass create tab should list webhooks in the dropdown."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
response: Response = client.get("/mass")
assert response.status_code == 200
assert MASS_WEBHOOK_NAME in response.text
assert "Create all feeds" in response.text
finally:
app.dependency_overrides = {}
class TestMassDelete:
"""Tests for POST /mass/delete."""
def test_delete_selected_feeds(self, mass_stub_reader: _StubReaderForMass) -> None:
"""POST /mass/delete with feed URLs should delete them."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
with patch("discord_rss_bot.main.commit_state_change"):
response: Response = client.post(
"/mass/delete",
data={"feed_urls": [MASS_FEED_1_URL, MASS_FEED_2_URL]},
)
assert response.status_code == 200
assert "Deleted 2 feed(s)" in response.text
finally:
app.dependency_overrides = {}
def test_delete_no_selection(self, mass_stub_reader: _StubReaderForMass) -> None:
"""POST /mass/delete with no feed URLs should not crash."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
response: Response = client.post("/mass/delete")
assert response.status_code == 200
assert "Deleted 0 feed(s)" in response.text
finally:
app.dependency_overrides = {}
def test_delete_feed_error_does_not_crash(self) -> None:
"""POST /mass/delete should handle reader errors gracefully."""
stub = _StubReaderForMass(
feeds=[_StubFeedForMass(url=MASS_FEED_1_URL)],
delete_fail=True,
)
app.dependency_overrides[get_reader_dependency] = lambda: stub
try:
response: Response = client.post("/mass/delete", data={"feed_urls": [MASS_FEED_1_URL]})
assert response.status_code == 200
assert "0 deleted, 1 failed" in response.text
finally:
app.dependency_overrides = {}
class TestMassModify:
"""Tests for POST /mass/modify."""
def test_modify_pause_feed(self, mass_stub_reader: _StubReaderForMass) -> None:
"""POST /mass/modify with pause action should disable feed updates."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
with patch("discord_rss_bot.main.commit_state_change"):
response: Response = client.post(
"/mass/modify",
data={
"feed_urls": [MASS_FEED_1_URL],
"modify_action": "pause",
"modify_value": "",
},
)
assert response.status_code == 200
assert MASS_FEED_1_URL in response.text
assert "Modified 1 feed(s)" in response.text
assert MASS_FEED_1_URL in mass_stub_reader.disabled_feeds
finally:
app.dependency_overrides = {}
def test_modify_unpause_feed(self, mass_stub_reader: _StubReaderForMass) -> None:
"""POST /mass/modify with unpause action should enable feed updates."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
with patch("discord_rss_bot.main.commit_state_change"):
response: Response = client.post(
"/mass/modify",
data={
"feed_urls": [MASS_FEED_2_URL],
"modify_action": "unpause",
"modify_value": "",
},
)
assert response.status_code == 200
assert "Modified 1 feed(s)" in response.text
assert MASS_FEED_2_URL in mass_stub_reader.enabled_feeds
finally:
app.dependency_overrides = {}
def test_modify_nonexistent_feed(self, mass_stub_reader: _StubReaderForMass) -> None:
"""POST /mass/modify for a non-existent feed should report error."""
nonexistent = "https://example.com/nonexistent.xml"
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
response: Response = client.post(
"/mass/modify",
data={
"feed_urls": [nonexistent],
"modify_action": "pause",
"modify_value": "",
},
)
assert response.status_code == 200
assert "0 modified, 1 failed" in response.text
assert "Feed not found" in response.text
finally:
app.dependency_overrides = {}
def test_modify_unknown_action(self, mass_stub_reader: _StubReaderForMass) -> None:
"""POST /mass/modify with an unknown action should report error."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
response: Response = client.post(
"/mass/modify",
data={
"feed_urls": [MASS_FEED_1_URL],
"modify_action": "nonexistent_action",
"modify_value": "",
},
)
assert response.status_code == 200
assert "0 modified, 1 failed" in response.text
assert "Unknown action" in response.text
finally:
app.dependency_overrides = {}
def test_modify_change_webhook_updates_tag(self, mass_stub_reader: _StubReaderForMass) -> None:
"""POST /mass/modify with change_webhook should call set_tag with the new webhook URL."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
with patch("discord_rss_bot.main.commit_state_change"):
response: Response = client.post(
"/mass/modify",
data={
"feed_urls": [MASS_FEED_1_URL],
"modify_action": "change_webhook",
"modify_value": MASS_WEBHOOK_NAME,
},
)
assert response.status_code == 200
assert "Modified 1 feed(s)" in response.text
# Check that set_tag was called with the webhook tag
webhook_calls = [c for c in mass_stub_reader.set_tag_calls if c[1] == "webhook"]
assert len(webhook_calls) == 1
assert webhook_calls[0][2] == MASS_WEBHOOK_URL
finally:
app.dependency_overrides = {}
def test_modify_unknown_webhook_reports_error(self, mass_stub_reader: _StubReaderForMass) -> None:
"""POST /mass/modify with a non-existent webhook name should report error."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
response: Response = client.post(
"/mass/modify",
data={
"feed_urls": [MASS_FEED_1_URL],
"modify_action": "change_webhook",
"modify_value": "Nonexistent Hook",
},
)
assert response.status_code == 200
assert "0 modified, 1 failed" in response.text
assert "not found" in response.text
finally:
app.dependency_overrides = {}
def test_modify_delivery_mode_embed(self, mass_stub_reader: _StubReaderForMass) -> None:
"""POST /mass/modify with delivery_mode=embed should set delivery_mode and should_send_embed."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
with patch("discord_rss_bot.main.commit_state_change"):
response: Response = client.post(
"/mass/modify",
data={
"feed_urls": [MASS_FEED_1_URL],
"modify_action": "delivery_mode",
"modify_value": "embed",
},
)
assert response.status_code == 200
assert "Modified 1 feed(s)" in response.text
delivery_calls = [c for c in mass_stub_reader.set_tag_calls if c[1] == "delivery_mode"]
embed_calls = [c for c in mass_stub_reader.set_tag_calls if c[1] == "should_send_embed"]
assert len(delivery_calls) == 1
assert delivery_calls[0][2] == "embed"
assert len(embed_calls) == 1
assert embed_calls[0][2] is True
finally:
app.dependency_overrides = {}
def test_modify_delivery_mode_text(self, mass_stub_reader: _StubReaderForMass) -> None:
"""POST /mass/modify with delivery_mode=text should set should_send_embed to False."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
with patch("discord_rss_bot.main.commit_state_change"):
response: Response = client.post(
"/mass/modify",
data={
"feed_urls": [MASS_FEED_1_URL],
"modify_action": "delivery_mode",
"modify_value": "text",
},
)
assert response.status_code == 200
assert "Modified 1 feed(s)" in response.text
embed_calls = [c for c in mass_stub_reader.set_tag_calls if c[1] == "should_send_embed"]
assert len(embed_calls) == 1
assert embed_calls[0][2] is False
finally:
app.dependency_overrides = {}
def test_modify_update_interval(self, mass_stub_reader: _StubReaderForMass) -> None:
"""POST /mass/modify with update_interval should store the interval."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
with patch("discord_rss_bot.main.commit_state_change"):
response: Response = client.post(
"/mass/modify",
data={
"feed_urls": [MASS_FEED_1_URL],
"modify_action": "update_interval",
"modify_value": "30",
},
)
assert response.status_code == 200
assert "Modified 1 feed(s)" in response.text
update_calls = [c for c in mass_stub_reader.set_tag_calls if c[1] == ".reader.update"]
assert len(update_calls) == 1
assert update_calls[0][2] == {"interval": 30}
finally:
app.dependency_overrides = {}
def test_modify_update_interval_invalid(self, mass_stub_reader: _StubReaderForMass) -> None:
"""POST /mass/modify with non-numeric update_interval should report error."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
response: Response = client.post(
"/mass/modify",
data={
"feed_urls": [MASS_FEED_1_URL],
"modify_action": "update_interval",
"modify_value": "not-a-number",
},
)
assert response.status_code == 200
assert "0 modified, 1 failed" in response.text
finally:
app.dependency_overrides = {}
def test_modify_no_selection(self, mass_stub_reader: _StubReaderForMass) -> None:
"""POST /mass/modify with no feed URLs should not crash."""
app.dependency_overrides[get_reader_dependency] = lambda: mass_stub_reader
try:
response: Response = client.post(
"/mass/modify",
data={"modify_action": "pause", "modify_value": ""},
)
assert response.status_code == 200
assert "Modified 0 feed(s)" in response.text
finally:
app.dependency_overrides = {}
class TestMassCreate:
"""Tests for POST /mass/create."""
def test_create_rejects_missing_webhook(self) -> None:
"""POST /mass/create without a valid webhook should return 404."""
stub = _StubReaderForMass()
app.dependency_overrides[get_reader_dependency] = lambda: stub
try:
response: Response = client.post(
"/mass/create",
data={
"feed_urls": "https://example.com/feed.xml",
"webhook_dropdown": "nonexistent",
},
)
assert response.status_code == 404
assert "Webhook not found" in response.text
finally:
app.dependency_overrides = {}
def test_create_single_url(self) -> None:
"""POST /mass/create with a single URL should attempt to create it."""
stub = _StubReaderForMass(webhook_name=MASS_WEBHOOK_NAME, webhook_url=MASS_WEBHOOK_URL)
app.dependency_overrides[get_reader_dependency] = lambda: stub
try:
with (
patch("discord_rss_bot.main._update_and_mark_read") as mock_update,
patch("discord_rss_bot.main.commit_state_change"),
):
mock_update.return_value = ("https://example.com/new-feed.xml", True, "")
response: Response = client.post(
"/mass/create",
data={
"feed_urls": "https://example.com/new-feed.xml",
"webhook_dropdown": MASS_WEBHOOK_NAME,
},
)
assert response.status_code == 200
finally:
app.dependency_overrides = {}
def test_create_multiple_urls(self) -> None:
"""POST /mass/create with multiple URLs should attempt to create them."""
stub = _StubReaderForMass(webhook_name=MASS_WEBHOOK_NAME, webhook_url=MASS_WEBHOOK_URL)
app.dependency_overrides[get_reader_dependency] = lambda: stub
try:
with (
patch("discord_rss_bot.main._update_and_mark_read") as mock_update,
patch("discord_rss_bot.main.commit_state_change"),
):
mock_update.side_effect = [
("https://example.com/feed1.xml", True, ""),
("https://example.com/feed2.xml", True, ""),
]
response: Response = client.post(
"/mass/create",
data={
"feed_urls": "https://example.com/feed1.xml\nhttps://example.com/feed2.xml",
"webhook_dropdown": MASS_WEBHOOK_NAME,
},
)
assert response.status_code == 200
finally:
app.dependency_overrides = {}
class TestMassNav:
"""Tests for the mass operations navigation link."""
def test_mass_link_in_navbar(self) -> None:
"""The navbar should contain a link to the mass operations page."""
response: Response = client.get("/")
assert response.status_code == 200
assert "/mass" in response.text, f"Expected /mass link in navbar: {response.text}"