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 %} +
+ + {% if webhooks %} +
+ + +
+ New feeds will be added with this webhook. Existing feeds will have their webhook overwritten. You can change this later per-feed. +
+
+ {% else %} + + {% endif %} +
+ + + + + + + + + + + {% for feed in feeds %} + + + + + + + {% endfor %} + +
+ + FeedURLStatus
+ + + {{ feed.title }} + + {{ feed.url }} + + {% if feed.already_exists %} + Already exists + {% else %} + New + {% endif %} +
+
+
+ + Cancel +
+
+ {% 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 ===== #} +
+
+

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 }})

+
+ + +
+
+
    + {% for item in group.feeds %} + {% set feed = item.feed %} +
  • + + +
  • + {% endfor %} +
+
+ {% endfor %} + +
+ {% 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 }})

+
+ + +
+
+
    + {% for item in group.feeds %} + {% set feed = item.feed %} +
  • + + +
  • + {% endfor %} +
+
+ {% endfor %} +
+
+
+ + +
+
+ +
+ +
+
+
+ +
+ {% 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 %} + + 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. -
-
-
-
-
- -
- - -
-
-
-
-
- -
- - -
-
- New feeds inherit this value. Existing feeds keep their current delivery mode. -
-
-
-
-
- -
- - -
-
- New feeds inherit this value. Existing feeds keep their current screenshot layout. -
- {% if not chromium_installed %} -
- Screenshot mode requires Chromium to be installed for Playwright. - Run uv run playwright install chromium once on this machine. +
+ +
+
+
+
+
+

Global Defaults

+
+
+

+ These defaults apply to newly added feeds. Existing feeds keep their own per-feed settings. +

+
+ +
+ + +
+ + +
+
+ Currently {{ global_interval }} min. Low intervals may cause issues with some feeds. +
+ +
+ +
+
+ +
+ + +
+
How new feeds send messages to Discord.
+
+
+ +
+
+ +
+ + +
+
+ Default screenshot viewport for new feeds. + {% if not chromium_installed %} + + Requires Chromium. Run uv run playwright install chromium. + + {% endif %} +
+
+
+ +
+
+ +
+ + +
+
Max characters for text mode (4000). Embeds are capped at 2000.
+
+
- {% endif %} -
- -
-
- -
- -
-
- 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. -
-
-
-
-
-
-
-

Feed Update Intervals

-
-

- Customize the update interval for individual feeds. Leave empty or reset to use the global default. -

+
- {% if feed_intervals %} -
- - - - - - - - - - - - - - - {% for item in feed_intervals %} - - - - - - - - - - - {% endfor %} - -
FeedDomainStatusIntervalLast UpdatedNext UpdateSet Interval (min)Actions
- {{ item.feed.title }} - - {{ item.domain }} - - - {{ 'Enabled' if item.feed.updates_enabled else 'Disabled' }} - - - {{ item.effective_interval }} min - {% if item.interval %} - Custom - {% else %} - Global - {% endif %} - - {{ item.feed.last_updated | relative_time }} - - {{ item.feed.update_after | relative_time }} - -
- - - - + +
+
+
+
+
+

Data Management

+
+
+

Import and export your feeds and database.

+
+ +
+
+
OPML Export
+
+

+ Download all feeds as an OPML subscription list. + Supports title, links, and description. +

+ Export OPML +
+
+
+ +
+
+
OPML Import
+
+

Upload an OPML file to preview and select which feeds to import.

+ +
+ + +
-
- {% if item.interval %} -
- - - -
- {% endif %} -
-
- {% else %} -

No feeds added yet.

- {% endif %} -
+ + + + +
+
+
+ Database Export +
+
+

Download a compressed SQLite dump for backup or migration.

+ Download Export +
+
+
+ + + + + +
+
+
+
+
+

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 %} +
+ + + + + + + + + + + + + + + {% for item in feed_intervals %} + + + + + + + + + + + {% endfor %} + +
FeedDomainStatusIntervalLast Updated + Next + Update + Set Interval (min)Actions
+ {{ item.feed.title }} + + {{ item.domain }} + + + {{ 'Enabled' if item.feed.updates_enabled else 'Disabled' }} + + + {{ item.effective_interval }} min + {% if item.interval %} + Custom + {% else %} + Global + {% endif %} + + {{ item.feed.last_updated | relative_time }} + + {{ item.feed.update_after | relative_time }} + +
+ + + + +
+
+ {% if item.interval %} +
+ + + +
+ {% endif %} +
+
+ {% 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}"