diff --git a/discord_rss_bot/main.py b/discord_rss_bot/main.py
index 70b8de6..6693c82 100644
--- a/discord_rss_bot/main.py
+++ b/discord_rss_bot/main.py
@@ -1,9 +1,12 @@
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
@@ -13,6 +16,7 @@ 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
@@ -24,9 +28,11 @@ 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
@@ -37,11 +43,14 @@ 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
@@ -54,6 +63,7 @@ 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
@@ -84,12 +94,15 @@ 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
@@ -274,6 +287,196 @@ 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.
@@ -2129,12 +2332,14 @@ 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.
@@ -2186,6 +2391,7 @@ 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)
@@ -2391,6 +2597,33 @@ 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,
@@ -2422,12 +2655,439 @@ 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:
diff --git a/discord_rss_bot/templates/import_opml_preview.html b/discord_rss_bot/templates/import_opml_preview.html
new file mode 100644
index 0000000..ef901db
--- /dev/null
+++ b/discord_rss_bot/templates/import_opml_preview.html
@@ -0,0 +1,117 @@
+{% 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 %}
+
+
+
+
Import OPML Feeds
+
+
+ Found {{ total }} feed{{ 's' if total != 1 else '' }} in the OPML file.
+ {% if new_count %}
+ {{ new_count }} new
+ {% endif %}
+ {% if existing_count %}
+ {{ existing_count }} already exist{{ 's' if existing_count == 1 else '' }}
+ {% endif %}
+
+
+ {% if feeds %}
+
+ {% else %}
+ No feeds found in the OPML file.
+ Back to Settings
+ {% endif %}
+
+{% endblock content %}
+{% block head %}
+
+{% endblock head %}
diff --git a/discord_rss_bot/templates/mass.html b/discord_rss_bot/templates/mass.html
new file mode 100644
index 0000000..88a6af8
--- /dev/null
+++ b/discord_rss_bot/templates/mass.html
@@ -0,0 +1,355 @@
+{% 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 %}
+
+ {% endif %}
+
+
+
+ Mass Create
+
+
+
+
+ Mass Delete
+
+
+
+
+ Mass Modify
+
+
+
+
+ {# ===== Mass Create ===== #}
+
+
+
Create multiple feeds
+
Enter one feed URL per line. All feeds will be attached to the selected webhook.
+
+
+ {% if create_results %}
+
+
Results
+
+ {% for result in create_results %}
+
+ {{ result.url }}
+ {% if result.success %}
+ ✓ Added
+ {% if result.feed_url %}
+ - View feed
+ {% endif %}
+ {% else %}
+ ✗ {{ result.error }}
+ {% endif %}
+
+ {% endfor %}
+
+
+ {% endif %}
+
+ {# ===== Mass Delete ===== #}
+
+
+
Delete multiple feeds
+
Select feeds to delete. This action cannot be undone.
+ {% if all_feeds_grouped %}
+
+ {% for group in all_feeds_grouped %}
+
+
+
{{ group.name }} ({{ group.feeds|length }})
+
+ Select all
+ Deselect all
+
+
+
+
+ {% endfor %}
+
+ Delete selected feeds
+
+
+ {% else %}
+
No feeds found.
+ {% endif %}
+
+ {% if delete_results %}
+
+
Results
+
+ {% for result in delete_results %}
+
+ {{ result.url }}
+ {% if result.success %}
+ ✓ Deleted
+ {% else %}
+ ✗ {{ result.error }}
+ {% endif %}
+
+ {% endfor %}
+ {{ delete_summary.deleted }} deleted, {{ delete_summary.failed }} failed
+
+
+ {% endif %}
+
+ {# ===== Mass Modify ===== #}
+
+
+
Modify multiple feeds
+
Select feeds and choose a setting to update.
+ {% if all_feeds_grouped %}
+
+ {% for group in all_feeds_grouped %}
+
+
+
{{ group.name }} ({{ group.feeds|length }})
+
+ Select all
+ Deselect all
+
+
+
+
+ {% endfor %}
+
+
+
+ Action
+
+ Choose action...
+ Pause (disable updates)
+ Unpause (enable updates)
+ Change webhook
+ Change delivery mode
+ Change screenshot layout
+ Change update interval (minutes)
+
+
+
+
+ Apply to selected feeds
+
+ {% else %}
+
No feeds found.
+ {% endif %}
+
+ {% if modify_results %}
+
+
Results
+
+ {% for result in modify_results %}
+
+ {{ result.url }}
+ {% if result.success %}
+ ✓ {{ result.action_taken }}
+ {% else %}
+ ✗ {{ result.error }}
+ {% endif %}
+
+ {% endfor %}
+
+ {{ modify_summary.modified }} modified, {{ modify_summary.failed }} failed, {{ modify_summary.skipped }} skipped
+
+
+
+ {% endif %}
+
+
+{% endblock content %}
+{% block head %}
+
+{% endblock head %}
diff --git a/discord_rss_bot/templates/nav.html b/discord_rss_bot/templates/nav.html
index 1fb1728..d0a5cdf 100644
--- a/discord_rss_bot/templates/nav.html
+++ b/discord_rss_bot/templates/nav.html
@@ -18,6 +18,10 @@
|
{% endif %}
+
+ Mass ops
+
+ |
Webhooks
diff --git a/discord_rss_bot/templates/settings.html b/discord_rss_bot/templates/settings.html
index df4b5ee..32437ea 100644
--- a/discord_rss_bot/templates/settings.html
+++ b/discord_rss_bot/templates/settings.html
@@ -6,191 +6,262 @@
Adjust default update intervals, delivery modes, screenshot layout, and webhook text limits for feeds managed by your bot.
{% endblock description %}
{% block content %}
-
-
-
-
Global Settings
-
-
- Set a default interval for all feeds. Individual feeds can still override this value.
-
-
-
- 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.
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
Feed Update Intervals
+
+
+
+ Customize the update interval for individual feeds. Leave empty or reset to use the global default of {{ global_interval }} min.
+
+ {% if feed_intervals %}
+
+ {% else %}
+
No feeds added yet.
+ {% endif %}
+
+
+
+
{% endblock content %}
diff --git a/tests/test_main.py b/tests/test_main.py
index 9c28d64..ee1b12a 100644
--- a/tests/test_main.py
+++ b/tests/test_main.py
@@ -14,7 +14,10 @@ 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
@@ -25,7 +28,6 @@ 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
@@ -646,44 +648,38 @@ def test_author_templates_render_authors_str() -> None:
assert "Legacy Entry Author" not in filter_preview_html
-def test_settings_page_shows_screenshot_layout_setting() -> None:
+def test_settings_page_loads() -> None:
+ """The settings page should render without errors."""
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() -> None:
+def test_set_global_delivery_mode_stores_value() -> None:
+ """POST /set_global_delivery_mode should persist the delivery mode in reader tags."""
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 = client.get(url="/settings")
- assert response.status_code == 200, f"/settings failed after setting delivery mode: {response.text}"
- assert re.search(r"]*\bselected\b", response.text)
+ reader: Reader = get_reader_dependency()
+ assert reader.get_tag((), "delivery_mode", "") == "text"
-def test_set_global_webhook_text_length_limit() -> None:
+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."""
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}"
-
- 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
+ reader: Reader = get_reader_dependency()
+ assert reader.get_tag((), "webhook_text_length_limit", 0) == 2500
-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}"
+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]
- response = client.get(url="/add")
+ response: 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:
@@ -899,13 +895,12 @@ def test_sent_webhooks_view_shows_saved_records() -> None:
app.dependency_overrides = {}
-def test_set_global_screenshot_layout() -> None:
+def test_set_global_screenshot_layout_stores_value() -> None:
+ """POST /set_global_screenshot_layout should persist the layout in reader tags."""
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}"
-
- response = client.get(url="/settings")
- assert response.status_code == 200, f"/settings failed after setting layout: {response.text}"
- assert re.search(r" ]*\bselected\b", response.text)
+ reader: Reader = get_reader_dependency()
+ assert reader.get_tag((), "screenshot_layout", "") == "mobile"
def test_pause_feed() -> None:
@@ -2554,10 +2549,268 @@ def test_reader_dependency_override_is_used() -> None:
# ---------------------------------------------------------------------------
-# Tests for post_embed — saving embed fields (including clearing to "")
+# Tests for OPML import / export
# ---------------------------------------------------------------------------
+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"" 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"""
+
+ test
+
+
+
+
+"""
+
+ 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''
+ b'reader feeds '
+ ),
+ 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,
@@ -3103,3 +3356,532 @@ 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}"