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