Add support for changing the update interval for feeds
Some checks failed
Test and build Docker image / docker (push) Has been cancelled

This commit is contained in:
Joakim Hellsén 2026-03-07 05:50:13 +01:00
commit 24d4d7a293
Signed by: Joakim Hellsén
SSH key fingerprint: SHA256:/9h/CsExpFp+PRhsfA0xznFx2CGfTT5R/kpuFfUgEQk
18 changed files with 803 additions and 119 deletions

View file

@ -14,6 +14,7 @@ Discord: TheLovinator#9276
- Choose between Discord embed or plain text. - Choose between Discord embed or plain text.
- Regex filters for RSS feeds. - Regex filters for RSS feeds.
- Blacklist/whitelist words in the title/description/author/etc. - Blacklist/whitelist words in the title/description/author/etc.
- Set different update frequencies for each feed or use a global default.
- Gets extra information from APIs if available, currently for: - Gets extra information from APIs if available, currently for:
- [https://feeds.c3kay.de/](https://feeds.c3kay.de/) - [https://feeds.c3kay.de/](https://feeds.c3kay.de/)
- Genshin Impact News - Genshin Impact News

View file

@ -98,7 +98,7 @@ def extract_domain(url: str) -> str: # noqa: PLR0911
return "Other" return "Other"
def send_entry_to_discord(entry: Entry, custom_reader: Reader | None = None) -> str | None: # noqa: PLR0912 def send_entry_to_discord(entry: Entry, custom_reader: Reader | None = None) -> str | None: # noqa: C901, PLR0912
"""Send a single entry to Discord. """Send a single entry to Discord.
Args: Args:
@ -240,7 +240,7 @@ def set_title(custom_embed: CustomEmbed, discord_embed: DiscordEmbed) -> None:
discord_embed.set_title(embed_title) if embed_title else None discord_embed.set_title(embed_title) if embed_title else None
def create_embed_webhook(webhook_url: str, entry: Entry) -> DiscordWebhook: def create_embed_webhook(webhook_url: str, entry: Entry) -> DiscordWebhook: # noqa: C901
"""Create a webhook with an embed. """Create a webhook with an embed.
Args: Args:
@ -341,7 +341,7 @@ def set_entry_as_read(reader: Reader, entry: Entry) -> None:
logger.exception("Error setting entry to read: %s", entry.id) logger.exception("Error setting entry to read: %s", entry.id)
def send_to_discord(custom_reader: Reader | None = None, feed: Feed | None = None, *, do_once: bool = False) -> None: # noqa: PLR0912 def send_to_discord(custom_reader: Reader | None = None, feed: Feed | None = None, *, do_once: bool = False) -> None: # noqa: C901, PLR0912
"""Send entries to Discord. """Send entries to Discord.
If response was not ok, we will log the error and mark the entry as unread, so it will be sent again next time. If response was not ok, we will log the error and mark the entry as unread, so it will be sent again next time.
@ -520,7 +520,7 @@ def truncate_webhook_message(webhook_message: str) -> str:
return webhook_message return webhook_message
def create_feed(reader: Reader, feed_url: str, webhook_dropdown: str) -> None: def create_feed(reader: Reader, feed_url: str, webhook_dropdown: str) -> None: # noqa: C901
"""Add a new feed, update it and mark every entry as read. """Add a new feed, update it and mark every entry as read.
Args: Args:

View file

@ -30,6 +30,8 @@ from pathlib import Path
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from typing import Any from typing import Any
from reader import TagNotFoundError
if TYPE_CHECKING: if TYPE_CHECKING:
from reader import Reader from reader import Reader
@ -66,6 +68,7 @@ _FEED_TAGS: tuple[str, ...] = (
"regex_whitelist_summary", "regex_whitelist_summary",
"regex_whitelist_content", "regex_whitelist_content",
"regex_whitelist_author", "regex_whitelist_author",
".reader.update",
) )
@ -178,10 +181,21 @@ def export_state(reader: Reader, backup_path: Path) -> None:
webhooks: list[str | int | float | bool | dict[str, Any] | list[Any] | None] = list( webhooks: list[str | int | float | bool | dict[str, Any] | list[Any] | None] = list(
reader.get_tag((), "webhooks", []) reader.get_tag((), "webhooks", [])
) )
except Exception: # noqa: BLE001 except TagNotFoundError:
webhooks = [] webhooks = []
# Export global update interval if set
global_update_interval: dict[str, Any] | None = None
try:
global_update_config = reader.get_tag((), ".reader.update", None)
if isinstance(global_update_config, dict):
global_update_interval = global_update_config
except TagNotFoundError:
pass
state: dict = {"feeds": feeds_state, "webhooks": webhooks} state: dict = {"feeds": feeds_state, "webhooks": webhooks}
if global_update_interval is not None:
state["global_update_interval"] = global_update_interval
state_file: Path = backup_path / "state.json" state_file: Path = backup_path / "state.json"
state_file.write_text(json.dumps(state, indent=2, default=str), encoding="utf-8") state_file.write_text(json.dumps(state, indent=2, default=str), encoding="utf-8")

View file

@ -4,10 +4,12 @@ import contextlib
import json import json
import logging import logging
import re import re
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING
from typing import Any
import requests import requests
from discord_webhook import DiscordEmbed, DiscordWebhook from discord_webhook import DiscordEmbed
from discord_webhook import DiscordWebhook
if TYPE_CHECKING: if TYPE_CHECKING:
from reader import Entry from reader import Entry

View file

@ -100,6 +100,46 @@ logging.config.dictConfig(LOGGING_CONFIG)
logger: logging.Logger = logging.getLogger(__name__) logger: logging.Logger = logging.getLogger(__name__)
reader: Reader = get_reader() reader: Reader = get_reader()
# Time constants for relative time formatting
SECONDS_PER_MINUTE = 60
SECONDS_PER_HOUR = 3600
SECONDS_PER_DAY = 86400
def relative_time(dt: datetime | None) -> str:
"""Convert a datetime to a relative time string (e.g., '2 hours ago', 'in 5 minutes').
Args:
dt: The datetime to convert (should be timezone-aware).
Returns:
A human-readable relative time string.
"""
if dt is None:
return "Never"
now = datetime.now(tz=UTC)
diff = dt - now
seconds = int(abs(diff.total_seconds()))
is_future = diff.total_seconds() > 0
# Determine the appropriate unit and value
if seconds < SECONDS_PER_MINUTE:
value = seconds
unit = "s"
elif seconds < SECONDS_PER_HOUR:
value = seconds // SECONDS_PER_MINUTE
unit = "m"
elif seconds < SECONDS_PER_DAY:
value = seconds // SECONDS_PER_HOUR
unit = "h"
else:
value = seconds // SECONDS_PER_DAY
unit = "d"
# Format based on future or past
return f"in {value}{unit}" if is_future else f"{value}{unit} ago"
@asynccontextmanager @asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None]: async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
@ -131,6 +171,7 @@ templates.env.filters["encode_url"] = lambda url: urllib.parse.quote(url) if url
templates.env.filters["entry_is_whitelisted"] = entry_is_whitelisted templates.env.filters["entry_is_whitelisted"] = entry_is_whitelisted
templates.env.filters["entry_is_blacklisted"] = entry_is_blacklisted templates.env.filters["entry_is_blacklisted"] = entry_is_blacklisted
templates.env.filters["discord_markdown"] = markdownify templates.env.filters["discord_markdown"] = markdownify
templates.env.filters["relative_time"] = relative_time
templates.env.globals["get_backup_path"] = get_backup_path templates.env.globals["get_backup_path"] = get_backup_path
@ -613,6 +654,102 @@ async def post_use_text(feed_url: Annotated[str, Form()]) -> RedirectResponse:
return RedirectResponse(url=f"/feed?feed_url={urllib.parse.quote(clean_feed_url)}", status_code=303) return RedirectResponse(url=f"/feed?feed_url={urllib.parse.quote(clean_feed_url)}", status_code=303)
@app.post("/set_update_interval")
async def post_set_update_interval(
feed_url: Annotated[str, Form()],
interval_minutes: Annotated[int | None, Form()] = None,
redirect_to: Annotated[str, Form()] = "",
) -> RedirectResponse:
"""Set the update interval for a feed.
Args:
feed_url: The feed to change.
interval_minutes: The update interval in minutes (None to reset to global default).
redirect_to: Optional redirect URL (defaults to feed page).
Returns:
RedirectResponse: Redirect to the specified page or feed page.
"""
clean_feed_url: str = feed_url.strip()
# If no interval specified, reset to global default
if interval_minutes is None:
try:
reader.delete_tag(clean_feed_url, ".reader.update")
commit_state_change(reader, f"Reset update interval to default for {clean_feed_url}")
except TagNotFoundError:
pass
else:
# Validate interval (minimum 1 minute, no maximum)
interval_minutes = max(interval_minutes, 1)
reader.set_tag(clean_feed_url, ".reader.update", {"interval": interval_minutes}) # pyright: ignore[reportArgumentType]
commit_state_change(reader, f"Set update interval to {interval_minutes} minutes for {clean_feed_url}")
# Update the feed immediately to recalculate update_after with the new interval
try:
reader.update_feed(clean_feed_url)
logger.info("Updated feed after interval change: %s", clean_feed_url)
except Exception:
logger.exception("Failed to update feed after interval change: %s", clean_feed_url)
if redirect_to:
return RedirectResponse(url=redirect_to, status_code=303)
return RedirectResponse(url=f"/feed?feed_url={urllib.parse.quote(clean_feed_url)}", status_code=303)
@app.post("/reset_update_interval")
async def post_reset_update_interval(
feed_url: Annotated[str, Form()],
redirect_to: Annotated[str, Form()] = "",
) -> RedirectResponse:
"""Reset the update interval for a feed to use the global default.
Args:
feed_url: The feed to change.
redirect_to: Optional redirect URL (defaults to feed page).
Returns:
RedirectResponse: Redirect to the specified page or feed page.
"""
clean_feed_url: str = feed_url.strip()
try:
reader.delete_tag(clean_feed_url, ".reader.update")
commit_state_change(reader, f"Reset update interval to default for {clean_feed_url}")
except TagNotFoundError:
# Tag doesn't exist, which is fine
pass
# Update the feed immediately to recalculate update_after with the new interval
try:
reader.update_feed(clean_feed_url)
logger.info("Updated feed after interval reset: %s", clean_feed_url)
except Exception:
logger.exception("Failed to update feed after interval reset: %s", clean_feed_url)
if redirect_to:
return RedirectResponse(url=redirect_to, status_code=303)
return RedirectResponse(url=f"/feed?feed_url={urllib.parse.quote(clean_feed_url)}", status_code=303)
@app.post("/set_global_update_interval")
async def post_set_global_update_interval(interval_minutes: Annotated[int, Form()]) -> RedirectResponse:
"""Set the global default update interval.
Args:
interval_minutes: The update interval in minutes.
Returns:
RedirectResponse: Redirect to the settings page.
"""
# Validate interval (minimum 1 minute, no maximum)
interval_minutes = max(interval_minutes, 1)
reader.set_tag((), ".reader.update", {"interval": interval_minutes}) # pyright: ignore[reportArgumentType]
commit_state_change(reader, f"Set global update interval to {interval_minutes} minutes")
return RedirectResponse(url="/settings", status_code=303)
@app.get("/add", response_class=HTMLResponse) @app.get("/add", response_class=HTMLResponse)
def get_add(request: Request): def get_add(request: Request):
"""Page for adding a new feed. """Page for adding a new feed.
@ -631,7 +768,7 @@ def get_add(request: Request):
@app.get("/feed", response_class=HTMLResponse) @app.get("/feed", response_class=HTMLResponse)
async def get_feed(feed_url: str, request: Request, starting_after: str = ""): async def get_feed(feed_url: str, request: Request, starting_after: str = ""): # noqa: C901, PLR0912, PLR0914, PLR0915
"""Get a feed by URL. """Get a feed by URL.
Args: Args:
@ -656,7 +793,7 @@ async def get_feed(feed_url: str, request: Request, starting_after: str = ""):
# Only show button if more than 10 entries. # Only show button if more than 10 entries.
total_entries: int = reader.get_entry_counts(feed=feed).total or 0 total_entries: int = reader.get_entry_counts(feed=feed).total or 0
show_more_entires_button: bool = total_entries > entries_per_page is_show_more_entries_button_visible: bool = total_entries > entries_per_page
# Get entries from the feed. # Get entries from the feed.
if starting_after: if starting_after:
@ -669,6 +806,27 @@ async def get_feed(feed_url: str, request: Request, starting_after: str = ""):
msg: str = f"{e}\n\n{[entry.id for entry in current_entries]}" msg: str = f"{e}\n\n{[entry.id for entry in current_entries]}"
html: str = create_html_for_feed(current_entries) html: str = create_html_for_feed(current_entries)
# Get feed and global intervals for error case too
feed_interval: int | None = None
try:
feed_update_config = reader.get_tag(feed, ".reader.update")
if isinstance(feed_update_config, dict) and "interval" in feed_update_config:
interval_value = feed_update_config["interval"]
if isinstance(interval_value, int):
feed_interval = interval_value
except TagNotFoundError:
pass
global_interval: int = 60
try:
global_update_config = reader.get_tag((), ".reader.update")
if isinstance(global_update_config, dict) and "interval" in global_update_config:
interval_value = global_update_config["interval"]
if isinstance(interval_value, int):
global_interval = interval_value
except TagNotFoundError:
pass
context = { context = {
"request": request, "request": request,
"feed": feed, "feed": feed,
@ -678,8 +836,10 @@ async def get_feed(feed_url: str, request: Request, starting_after: str = ""):
"should_send_embed": False, "should_send_embed": False,
"last_entry": None, "last_entry": None,
"messages": msg, "messages": msg,
"show_more_entires_button": show_more_entires_button, "is_show_more_entries_button_visible": is_show_more_entries_button_visible,
"total_entries": total_entries, "total_entries": total_entries,
"feed_interval": feed_interval,
"global_interval": global_interval,
} }
return templates.TemplateResponse(request=request, name="feed.html", context=context) return templates.TemplateResponse(request=request, name="feed.html", context=context)
@ -708,6 +868,29 @@ async def get_feed(feed_url: str, request: Request, starting_after: str = ""):
add_missing_tags(reader) add_missing_tags(reader)
should_send_embed: bool = bool(reader.get_tag(feed, "should_send_embed")) should_send_embed: bool = bool(reader.get_tag(feed, "should_send_embed"))
# Get the update interval for this feed
feed_interval: int | None = None
try:
feed_update_config = reader.get_tag(feed, ".reader.update")
if isinstance(feed_update_config, dict) and "interval" in feed_update_config:
interval_value = feed_update_config["interval"]
if isinstance(interval_value, int):
feed_interval = interval_value
except TagNotFoundError:
# No custom interval set for this feed, will use global default
pass
# Get the global default update interval
global_interval: int = 60 # Default to 60 minutes if not set
try:
global_update_config = reader.get_tag((), ".reader.update")
if isinstance(global_update_config, dict) and "interval" in global_update_config:
interval_value = global_update_config["interval"]
if isinstance(interval_value, int):
global_interval = interval_value
except TagNotFoundError:
pass
context = { context = {
"request": request, "request": request,
"feed": feed, "feed": feed,
@ -716,8 +899,10 @@ async def get_feed(feed_url: str, request: Request, starting_after: str = ""):
"html": html, "html": html,
"should_send_embed": should_send_embed, "should_send_embed": should_send_embed,
"last_entry": last_entry, "last_entry": last_entry,
"show_more_entires_button": show_more_entires_button, "is_show_more_entries_button_visible": is_show_more_entries_button_visible,
"total_entries": total_entries, "total_entries": total_entries,
"feed_interval": feed_interval,
"global_interval": global_interval,
} }
return templates.TemplateResponse(request=request, name="feed.html", context=context) return templates.TemplateResponse(request=request, name="feed.html", context=context)
@ -847,6 +1032,56 @@ def get_data_from_hook_url(hook_name: str, hook_url: str) -> WebhookInfo:
return our_hook return our_hook
@app.get("/settings", response_class=HTMLResponse)
async def get_settings(request: Request):
"""Settings page.
Args:
request: The request object.
Returns:
HTMLResponse: The settings page.
"""
# Get the global default update interval
global_interval: int = 60 # Default to 60 minutes if not set
try:
global_update_config = reader.get_tag((), ".reader.update")
if isinstance(global_update_config, dict) and "interval" in global_update_config:
interval_value = global_update_config["interval"]
if isinstance(interval_value, int):
global_interval = interval_value
except TagNotFoundError:
pass
# Get all feeds with their intervals
feeds: Iterable[Feed] = reader.get_feeds()
feed_intervals = []
for feed in feeds:
feed_interval: int | None = None
try:
feed_update_config = reader.get_tag(feed, ".reader.update")
if isinstance(feed_update_config, dict) and "interval" in feed_update_config:
interval_value = feed_update_config["interval"]
if isinstance(interval_value, int):
feed_interval = interval_value
except TagNotFoundError:
pass
feed_intervals.append({
"feed": feed,
"interval": feed_interval,
"effective_interval": feed_interval or global_interval,
"domain": extract_domain(feed.url),
})
context = {
"request": request,
"global_interval": global_interval,
"feed_intervals": feed_intervals,
}
return templates.TemplateResponse(request=request, name="settings.html", context=context)
@app.get("/webhooks", response_class=HTMLResponse) @app.get("/webhooks", response_class=HTMLResponse)
async def get_webhooks(request: Request): async def get_webhooks(request: Request):
"""Page for adding a new webhook. """Page for adding a new webhook.

View file

@ -6,6 +6,7 @@ from pathlib import Path
from platformdirs import user_data_dir from platformdirs import user_data_dir
from reader import Reader from reader import Reader
from reader import TagNotFoundError
from reader import make_reader from reader import make_reader
if typing.TYPE_CHECKING: if typing.TYPE_CHECKING:
@ -39,7 +40,12 @@ def get_reader(custom_location: Path | None = None) -> Reader:
reader: Reader = make_reader(url=str(db_location)) reader: Reader = make_reader(url=str(db_location))
# https://reader.readthedocs.io/en/latest/api.html#reader.types.UpdateConfig # https://reader.readthedocs.io/en/latest/api.html#reader.types.UpdateConfig
# Set the update interval to 15 minutes # Set the default update interval to 15 minutes if not already configured
# Users can change this via the Settings page or per-feed in the feed page
try:
reader.get_tag((), ".reader.update")
except TagNotFoundError:
# Set default
reader.set_tag((), ".reader.update", {"interval": 15}) reader.set_tag((), ".reader.update", {"interval": 15})
return reader return reader

View file

@ -13,3 +13,7 @@ body {
.form-text { .form-text {
color: #acabab; color: #acabab;
} }
.interval-input {
max-width: 120px;
}

View file

@ -1,90 +1,145 @@
{% extends "base.html" %} {% extends "base.html" %}
{% block title %} {% block title %}
| {{ feed.title }} | {{ feed.title }}
{% endblock title %} {% endblock title %}
{% block content %} {% block content %}
<div class="card mb-3 border border-dark p-3 text-light"> <div class="card mb-3 border border-dark p-3 text-light">
<!-- Feed Title --> <!-- Feed Title -->
<h2> <h2>
<a class="text-muted" href="{{ feed.url }}">{{ feed.title }}</a> ({{ total_entries }} entries) <a class="text-muted" href="{{ feed.url }}">{{ feed.title }}</a> ({{ total_entries }} entries)
</h2> </h2>
{% if not feed.updates_enabled %} {% if not feed.updates_enabled %}<span class="badge bg-danger">Disabled</span>{% endif %}
<span class="badge bg-danger">Disabled</span>
{% endif %}
{% if feed.last_exception %} {% if feed.last_exception %}
<div class="mt-3"> <div class="mt-3">
<h5 class="text-danger">{{ feed.last_exception.type_name }}:</h5> <h5 class="text-danger">{{ feed.last_exception.type_name }}:</h5>
<code class="d-block">{{ feed.last_exception.value_str }}</code> <code class="d-block">{{ feed.last_exception.value_str }}</code>
<button class="btn btn-secondary btn-sm mt-2" type="button" data-bs-toggle="collapse" <button class="btn btn-secondary btn-sm mt-2"
data-bs-target="#exceptionDetails" aria-expanded="false" aria-controls="exceptionDetails"> type="button"
Show Traceback data-bs-toggle="collapse"
</button> data-bs-target="#exceptionDetails"
aria-expanded="false"
aria-controls="exceptionDetails">Show Traceback</button>
<div class="collapse" id="exceptionDetails"> <div class="collapse" id="exceptionDetails">
<pre><code>{{ feed.last_exception.traceback_str }}</code></pre> <pre><code>{{ feed.last_exception.traceback_str }}</code></pre>
</div> </div>
</div> </div>
{% endif %} {% endif %}
<!-- Feed Actions --> <!-- Feed Actions -->
<div class="mt-3 d-flex flex-wrap gap-2"> <div class="mt-3 d-flex flex-wrap gap-2">
<a href="/update?feed_url={{ feed.url|encode_url }}" class="btn btn-primary btn-sm">Update</a> <a href="/update?feed_url={{ feed.url|encode_url }}"
class="btn btn-primary btn-sm">Update</a>
<form action="/remove" method="post" class="d-inline"> <form action="/remove" method="post" class="d-inline">
<button class="btn btn-danger btn-sm" name="feed_url" value="{{ feed.url }}" <button class="btn btn-danger btn-sm"
name="feed_url"
value="{{ feed.url }}"
onclick="return confirm('Are you sure you want to delete this feed?')">Remove</button> onclick="return confirm('Are you sure you want to delete this feed?')">Remove</button>
</form> </form>
{% if not feed.updates_enabled %} {% if not feed.updates_enabled %}
<form action="/unpause" method="post" class="d-inline"> <form action="/unpause" method="post" class="d-inline">
<button class="btn btn-secondary btn-sm" name="feed_url" value="{{ feed.url }}">Unpause</button> <button class="btn btn-secondary btn-sm"
name="feed_url"
value="{{ feed.url }}">Unpause</button>
</form> </form>
{% else %} {% else %}
<form action="/pause" method="post" class="d-inline"> <form action="/pause" method="post" class="d-inline">
<button class="btn btn-danger btn-sm" name="feed_url" value="{{ feed.url }}">Pause</button> <button class="btn btn-danger btn-sm" name="feed_url" value="{{ feed.url }}">Pause</button>
</form> </form>
{% endif %} {% endif %}
{% if not "youtube.com/feeds/videos.xml" in feed.url %} {% if not "youtube.com/feeds/videos.xml" in feed.url %}
{% if should_send_embed %} {% if should_send_embed %}
<form action="/use_text" method="post" class="d-inline"> <form action="/use_text" method="post" class="d-inline">
<button class="btn btn-dark btn-sm" name="feed_url" value="{{ feed.url }}"> <button class="btn btn-dark btn-sm" name="feed_url" value="{{ feed.url }}">Send text message instead of embed</button>
Send text message instead of embed
</button>
</form> </form>
{% else %} {% else %}
<form action="/use_embed" method="post" class="d-inline"> <form action="/use_embed" method="post" class="d-inline">
<button class="btn btn-dark btn-sm" name="feed_url" value="{{ feed.url }}"> <button class="btn btn-dark btn-sm" name="feed_url" value="{{ feed.url }}">Send embed instead of text message</button>
Send embed instead of text message
</button>
</form> </form>
{% endif %} {% endif %}
{% endif %} {% endif %}
</div> </div>
<!-- Additional Links --> <!-- Additional Links -->
<div class="mt-3"> <div class="mt-3">
<a class="text-muted d-block" href="/whitelist?feed_url={{ feed.url|encode_url }}">Whitelist</a> <a class="text-muted d-block"
<a class="text-muted d-block" href="/blacklist?feed_url={{ feed.url|encode_url }}">Blacklist</a> href="/whitelist?feed_url={{ feed.url|encode_url }}">Whitelist</a>
<a class="text-muted d-block" href="/custom?feed_url={{ feed.url|encode_url }}"> <a class="text-muted d-block"
Customize message {% if not should_send_embed %}(Currently active){% endif %} href="/blacklist?feed_url={{ feed.url|encode_url }}">Blacklist</a>
<a class="text-muted d-block"
href="/custom?feed_url={{ feed.url|encode_url }}">
Customize message
{% if not should_send_embed %}(Currently active){% endif %}
</a> </a>
{% if not "youtube.com/feeds/videos.xml" in feed.url %} {% if not "youtube.com/feeds/videos.xml" in feed.url %}
<a class="text-muted d-block" href="/embed?feed_url={{ feed.url|encode_url }}"> <a class="text-muted d-block"
Customize embed {% if should_send_embed %}(Currently active){% endif %} href="/embed?feed_url={{ feed.url|encode_url }}">
Customize embed
{% if should_send_embed %}(Currently active){% endif %}
</a> </a>
{% endif %} {% endif %}
</div> </div>
</div> <!-- Feed Metadata -->
<div class="mt-4 border-top border-secondary pt-3">
{# Rendered HTML content #} <h5 class="mb-3">Feed Information</h5>
<pre>{{ html|safe }}</pre> <div class="row text-muted">
<div class="col-md-6 mb-2">
{% if show_more_entires_button %} <small><strong>Added:</strong> {{ feed.added | relative_time }}</small>
<a class="btn btn-dark mt-3" </div>
<div class="col-md-6 mb-2">
<small><strong>Last Updated:</strong> {{ feed.last_updated | relative_time }}</small>
</div>
<div class="col-md-6 mb-2">
<small><strong>Last Retrieved:</strong> {{ feed.last_retrieved | relative_time }}</small>
</div>
<div class="col-md-6 mb-2">
<small><strong>Next Update:</strong> {{ feed.update_after | relative_time }}</small>
</div>
<div class="col-md-6 mb-2">
<small><strong>Updates:</strong> <span class="badge {{ 'bg-success' if feed.updates_enabled else 'bg-danger' }}">{{ 'Enabled' if feed.updates_enabled else 'Disabled' }}</span></small>
</div>
</div>
</div>
<!-- Update Interval Configuration -->
<div class="mt-4 border-top border-secondary pt-3">
<h5 class="mb-3">Update Interval</h5>
{% if feed_interval %}
<p class="text-muted mb-2">
Current: <strong>{{ feed_interval }} minutes</strong>
{% if feed_interval >= 60 %}({{ (feed_interval / 60) | round(1) }} hours){% endif %}
<span class="badge bg-info">Custom</span>
</p>
{% else %}
<p class="text-muted mb-2">
Current: <strong>{{ global_interval }} minutes</strong>
{% if global_interval >= 60 %}({{ (global_interval / 60) | round(1) }} hours){% endif %}
<span class="badge bg-secondary">Using global default</span>
</p>
{% endif %}
<form action="/set_update_interval" method="post" class="mb-2">
<input type="hidden" name="feed_url" value="{{ feed.url }}" />
<div class="input-group input-group-sm mb-2">
<input type="number"
class="form-control form-control-sm interval-input"
name="interval_minutes"
placeholder="Minutes"
min="1"
value="{{ feed_interval if feed_interval else global_interval }}"
required />
<button class="btn btn-primary" type="submit">Set Interval</button>
</div>
</form>
{% if feed_interval %}
<form action="/reset_update_interval" method="post" class="d-inline">
<input type="hidden" name="feed_url" value="{{ feed.url }}" />
<button class="btn btn-secondary btn-sm" type="submit">Reset to Global Default</button>
</form>
{% endif %}
</div>
</div>
{# Rendered HTML content #}
<pre>{{ html|safe }}</pre>
{% if is_show_more_entries_button_visible %}
<a class="btn btn-dark mt-3"
href="/feed?feed_url={{ feed.url|encode_url }}&starting_after={{ last_entry.id|encode_url }}"> href="/feed?feed_url={{ feed.url|encode_url }}&starting_after={{ last_entry.id|encode_url }}">
Show more entries Show more entries
</a> </a>
{% endif %} {% endif %}
{% endblock content %} {% endblock content %}

View file

@ -19,6 +19,10 @@
<li class="nav-item"> <li class="nav-item">
<a class="nav-link" href="/webhooks">Webhooks</a> <a class="nav-link" href="/webhooks">Webhooks</a>
</li> </li>
<li class="nav-item nav-link d-none d-md-block">|</li>
<li class="nav-item">
<a class="nav-link" href="/settings">Settings</a>
</li>
{% if get_backup_path() %} {% if get_backup_path() %}
<li class="nav-item nav-link d-none d-md-block">|</li> <li class="nav-item nav-link d-none d-md-block">|</li>
<li class="nav-item"> <li class="nav-item">

View file

@ -0,0 +1,122 @@
{% extends "base.html" %}
{% block title %}
| Settings
{% endblock title %}
{% block content %}
<section>
<div class="text-light">
<div class="d-flex flex-wrap justify-content-between align-items-center gap-2">
<h2 class="mb-0">Global Settings</h2>
</div>
<p class="text-muted mt-2 mb-4">
Set a default interval for all feeds. Individual feeds can still override this value.
</p>
<div class="mb-4">
<div>
Current default is {{ global_interval }} min.
Even though we check ETags and Last-Modified headers, choosing a very low interval may cause issues with some feeds or cause excessive load on the server hosting the feed. Remember to be kind.
</div>
</div>
</div>
<form action="/set_global_update_interval" method="post" class="mb-2">
<div class="settings-form-row mb-2">
<label for="interval_minutes" class="form-label mb-1">Default interval (minutes)</label>
<div class="input-group input-group-lg">
<input id="interval_minutes"
type="number"
class="form-control settings-input"
name="interval_minutes"
placeholder="Minutes"
min="1"
value="{{ global_interval }}"
required />
<button class="btn btn-primary px-4" type="submit">Save</button>
</div>
</div>
</form>
</section>
<section class="mt-5">
<div class="text-light">
<div class="d-flex flex-wrap justify-content-between align-items-center gap-2">
<h2 class="mb-0">Feed Update Intervals</h2>
</div>
<p class="text-muted mt-2 mb-4">
Customize the update interval for individual feeds. Leave empty or reset to use the global default.
</p>
</div>
{% if feed_intervals %}
<div class="table-responsive">
<table class="table table-dark table-hover">
<thead>
<tr>
<th>Feed</th>
<th>Domain</th>
<th>Status</th>
<th>Interval</th>
<th>Last Updated</th>
<th>Next Update</th>
<th>Set Interval (min)</th>
<th>Actions</th>
</tr>
</thead>
<tbody>
{% for item in feed_intervals %}
<tr>
<td>
<a href="/feed?feed_url={{ item.feed.url|encode_url }}"
class="text-light text-decoration-none">{{ item.feed.title }}</a>
</td>
<td>
<span class="text-muted small">{{ item.domain }}</span>
</td>
<td>
<span class="badge {{ 'bg-success' if item.feed.updates_enabled else 'bg-danger' }}">
{{ 'Enabled' if item.feed.updates_enabled else 'Disabled' }}
</span>
</td>
<td>
<span>{{ item.effective_interval }} min</span>
{% if item.interval %}
<span class="badge bg-info ms-1">Custom</span>
{% else %}
<span class="badge bg-secondary ms-1">Global</span>
{% endif %}
</td>
<td>
<small class="text-muted">{{ item.feed.last_updated | relative_time }}</small>
</td>
<td>
<small class="text-muted">{{ item.feed.update_after | relative_time }}</small>
</td>
<td>
<form action="/set_update_interval" method="post" class="d-flex gap-2">
<input type="hidden" name="feed_url" value="{{ item.feed.url }}" />
<input type="hidden" name="redirect_to" value="/settings" />
<input type="number"
class="form-control form-control-sm interval-input"
name="interval_minutes"
placeholder="Minutes"
min="1"
value="{{ item.interval if item.interval else global_interval }}" />
<button class="btn btn-primary btn-sm" type="submit">Set</button>
</form>
</td>
<td>
{% if item.interval %}
<form action="/reset_update_interval" method="post" class="d-inline">
<input type="hidden" name="feed_url" value="{{ item.feed.url }}" />
<input type="hidden" name="redirect_to" value="/settings" />
<button class="btn btn-outline-secondary btn-sm" type="submit">Reset</button>
</form>
{% endif %}
</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
{% else %}
<p class="text-muted">No feeds added yet.</p>
{% endif %}
</section>
{% endblock content %}

View file

@ -86,15 +86,7 @@ lint.ignore = [
[tool.ruff.lint.per-file-ignores] [tool.ruff.lint.per-file-ignores]
"tests/*" = ["S101", "D103", "PLR2004"] "tests/*" = ["S101", "D103", "PLR2004"]
[tool.ruff.lint.mccabe]
max-complexity = 15 # Don't judge lol
[tool.pytest.ini_options] [tool.pytest.ini_options]
python_files = ["test_*.py"]
log_cli = true
log_cli_level = "DEBUG"
log_cli_format = "%(asctime)s [%(levelname)8s] %(message)s (%(filename)s:%(lineno)s)"
log_cli_date_format = "%Y-%m-%d %H:%M:%S"
filterwarnings = [ filterwarnings = [
"ignore::bs4.GuessedAtParserWarning", "ignore::bs4.GuessedAtParserWarning",
"ignore:functools\\.partial will be a method descriptor in future Python versions; wrap it in staticmethod\\(\\) if you want to preserve the old behavior:FutureWarning", "ignore:functools\\.partial will be a method descriptor in future Python versions; wrap it in staticmethod\\(\\) if you want to preserve the old behavior:FutureWarning",

View file

@ -4,9 +4,13 @@ import tempfile
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from reader import Entry, Feed, Reader, make_reader from reader import Entry
from reader import Feed
from reader import Reader
from reader import make_reader
from discord_rss_bot.filter.blacklist import entry_should_be_skipped, feed_has_blacklist_tags from discord_rss_bot.filter.blacklist import entry_should_be_skipped
from discord_rss_bot.filter.blacklist import feed_has_blacklist_tags
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Iterable from collections.abc import Iterable

View file

@ -5,7 +5,9 @@ import tempfile
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from discord_rss_bot.custom_filters import encode_url, entry_is_blacklisted, entry_is_whitelisted from discord_rss_bot.custom_filters import encode_url
from discord_rss_bot.custom_filters import entry_is_blacklisted
from discord_rss_bot.custom_filters import entry_is_whitelisted
from discord_rss_bot.settings import get_reader from discord_rss_bot.settings import get_reader
if TYPE_CHECKING: if TYPE_CHECKING:

View file

@ -4,19 +4,20 @@ import os
import tempfile import tempfile
from pathlib import Path from pathlib import Path
from typing import LiteralString from typing import LiteralString
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock
from unittest.mock import patch
import pytest import pytest
from reader import Feed, Reader, make_reader from reader import Feed
from reader import Reader
from reader import make_reader
from discord_rss_bot.feeds import ( from discord_rss_bot.feeds import extract_domain
extract_domain, from discord_rss_bot.feeds import is_youtube_feed
is_youtube_feed, from discord_rss_bot.feeds import send_entry_to_discord
send_entry_to_discord, from discord_rss_bot.feeds import send_to_discord
send_to_discord, from discord_rss_bot.feeds import should_send_embed_check
should_send_embed_check, from discord_rss_bot.feeds import truncate_webhook_message
truncate_webhook_message,
)
from discord_rss_bot.missing_tags import add_missing_tags from discord_rss_bot.missing_tags import add_missing_tags

View file

@ -1,5 +1,6 @@
from __future__ import annotations from __future__ import annotations
import re
import urllib.parse import urllib.parse
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
@ -289,3 +290,152 @@ def test_backup_endpoint_returns_error_when_not_configured(monkeypatch: pytest.M
assert "Git backup is not configured" in response.text or "GIT_BACKUP_PATH" in response.text, ( assert "Git backup is not configured" in response.text or "GIT_BACKUP_PATH" in response.text, (
"Error message about backup not being configured should be shown" "Error message about backup not being configured should be shown"
) )
def test_show_more_entries_button_visible_when_many_entries() -> None:
"""Test that the 'Show more entries' button is visible when there are more than 20 entries."""
# Add the webhook first
client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
response: Response = client.post(
url="/add_webhook",
data={"webhook_name": webhook_name, "webhook_url": webhook_url},
)
assert response.status_code == 200, f"Failed to add webhook: {response.text}"
# Remove the feed if it already exists
feeds: Response = client.get(url="/")
if feed_url in feeds.text:
client.post(url="/remove", data={"feed_url": feed_url})
# Add the feed
response: Response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name})
assert response.status_code == 200, f"Failed to add feed: {response.text}"
# Get the feed page
response: Response = client.get(url="/feed", params={"feed_url": feed_url})
assert response.status_code == 200, f"Failed to get /feed: {response.text}"
# Check if the feed has more than 20 entries by looking at the response
# The button should be visible if there are more than 20 entries
# We check for both the button text and the link structure
if "Show more entries" in response.text:
# Button is visible - verify it has the correct structure
assert "starting_after=" in response.text, "Show more entries button should contain starting_after parameter"
# The button should be a link to the feed page with pagination
assert (
f'href="/feed?feed_url={urllib.parse.quote(feed_url)}' in response.text
or f'href="/feed?feed_url={encoded_feed_url(feed_url)}' in response.text
), "Show more entries button should link back to the feed page"
def test_show_more_entries_button_not_visible_when_few_entries() -> None:
"""Test that the 'Show more entries' button is not visible when there are 20 or fewer entries."""
# Use a feed with very few entries
small_feed_url = "https://lovinator.space/rss_test_small.xml"
# Clean up if exists
client.post(url="/remove", data={"feed_url": small_feed_url})
# Add a small feed (this may not exist, so this test is conditional)
response: Response = client.post(url="/add", data={"feed_url": small_feed_url, "webhook_dropdown": webhook_name})
if response.status_code == 200:
# Get the feed page
response: Response = client.get(url="/feed", params={"feed_url": small_feed_url})
assert response.status_code == 200, f"Failed to get /feed: {response.text}"
# If the feed has 20 or fewer entries, the button should not be visible
# We check the total entry count in the page
if "0 entries" in response.text or " entries)" in response.text:
# Extract entry count and verify button visibility
match: re.Match[str] | None = re.search(r"\((\d+) entries\)", response.text)
if match:
entry_count = int(match.group(1))
if entry_count <= 20:
assert "Show more entries" not in response.text, (
f"Show more entries button should not be visible when there are {entry_count} entries"
)
# Clean up
client.post(url="/remove", data={"feed_url": small_feed_url})
def test_show_more_entries_pagination_works() -> None:
"""Test that pagination with starting_after parameter works correctly."""
# Add the webhook first
client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
response: Response = client.post(
url="/add_webhook",
data={"webhook_name": webhook_name, "webhook_url": webhook_url},
)
assert response.status_code == 200, f"Failed to add webhook: {response.text}"
# Remove the feed if it already exists
feeds: Response = client.get(url="/")
if feed_url in feeds.text:
client.post(url="/remove", data={"feed_url": feed_url})
# Add the feed
response: Response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name})
assert response.status_code == 200, f"Failed to add feed: {response.text}"
# Get the first page
response: Response = client.get(url="/feed", params={"feed_url": feed_url})
assert response.status_code == 200, f"Failed to get /feed: {response.text}"
# Check if pagination is available
if "Show more entries" in response.text and "starting_after=" in response.text:
# Extract the starting_after parameter from the button link
match: re.Match[str] | None = re.search(r'starting_after=([^"&]+)', response.text)
if match:
starting_after_id: str = match.group(1)
# Request the second page
response: Response = client.get(
url="/feed", params={"feed_url": feed_url, "starting_after": starting_after_id}
)
assert response.status_code == 200, f"Failed to get paginated feed: {response.text}"
# Verify we got a valid response (the page should contain entries)
assert "entries)" in response.text, "Paginated page should show entry count"
def test_show_more_entries_button_context_variable() -> None:
"""Test that the button visibility variable is correctly passed to the template context."""
# Add the webhook first
client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
response: Response = client.post(
url="/add_webhook",
data={"webhook_name": webhook_name, "webhook_url": webhook_url},
)
assert response.status_code == 200, f"Failed to add webhook: {response.text}"
# Remove the feed if it already exists
feeds: Response = client.get(url="/")
if feed_url in feeds.text:
client.post(url="/remove", data={"feed_url": feed_url})
# Add the feed
response: Response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name})
assert response.status_code == 200, f"Failed to add feed: {response.text}"
# Get the feed page
response: Response = client.get(url="/feed", params={"feed_url": feed_url})
assert response.status_code == 200, f"Failed to get /feed: {response.text}"
# Extract the total entries count from the page
match: re.Match[str] | None = re.search(r"\((\d+) entries\)", response.text)
if match:
entry_count = int(match.group(1))
# If more than 20 entries, button should be visible
if entry_count > 20:
assert "Show more entries" in response.text, (
f"Button should be visible when there are {entry_count} entries (more than 20)"
)
# If 20 or fewer entries, button should not be visible
else:
assert "Show more entries" not in response.text, (
f"Button should not be visible when there are {entry_count} entries (20 or fewer)"
)

View file

@ -4,7 +4,9 @@ import tempfile
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from reader import Feed, Reader, make_reader from reader import Feed
from reader import Reader
from reader import make_reader
from discord_rss_bot.search import create_search_context from discord_rss_bot.search import create_search_context

View file

@ -6,7 +6,9 @@ from pathlib import Path
from reader import Reader from reader import Reader
from discord_rss_bot.settings import data_dir, default_custom_message, get_reader from discord_rss_bot.settings import data_dir
from discord_rss_bot.settings import default_custom_message
from discord_rss_bot.settings import get_reader
def test_reader() -> None: def test_reader() -> None:

View file

@ -0,0 +1,88 @@
from __future__ import annotations
import urllib.parse
from typing import TYPE_CHECKING
from fastapi.testclient import TestClient
from discord_rss_bot.main import app
if TYPE_CHECKING:
from httpx import Response
client: TestClient = TestClient(app)
webhook_name: str = "Test Webhook for Update Interval"
webhook_url: str = "https://discord.com/api/webhooks/1234567890/test_update_interval"
feed_url: str = "https://lovinator.space/rss_test.xml"
def test_global_update_interval() -> None:
"""Test setting the global update interval."""
# Set global update interval to 30 minutes
response: Response = client.post("/set_global_update_interval", data={"interval_minutes": "30"})
assert response.status_code == 200, f"Failed to set global interval: {response.text}"
# Check that the settings page shows the new interval
response = client.get("/settings")
assert response.status_code == 200, f"Failed to get settings page: {response.text}"
assert "30" in response.text, "Global interval not updated on settings page"
def test_per_feed_update_interval() -> None:
"""Test setting per-feed update interval."""
# Clean up any existing feed/webhook
client.post(url="/delete_webhook", data={"webhook_url": webhook_url})
client.post(url="/remove", data={"feed_url": feed_url})
# Add webhook and feed
response: Response = client.post(
url="/add_webhook",
data={"webhook_name": webhook_name, "webhook_url": webhook_url},
)
assert response.status_code == 200, f"Failed to add webhook: {response.text}"
response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name})
assert response.status_code == 200, f"Failed to add feed: {response.text}"
# Set feed-specific update interval to 15 minutes
response = client.post("/set_update_interval", data={"feed_url": feed_url, "interval_minutes": "15"})
assert response.status_code == 200, f"Failed to set feed interval: {response.text}"
# Check that the feed page shows the custom interval
encoded_url = urllib.parse.quote(feed_url)
response = client.get(f"/feed?feed_url={encoded_url}")
assert response.status_code == 200, f"Failed to get feed page: {response.text}"
assert "15" in response.text, "Feed interval not displayed on feed page"
assert "Custom" in response.text, "Custom badge not shown for feed-specific interval"
def test_reset_feed_update_interval() -> None:
"""Test resetting feed update interval to global default."""
# First set a custom interval
response: Response = client.post("/set_update_interval", data={"feed_url": feed_url, "interval_minutes": "15"})
assert response.status_code == 200, f"Failed to set feed interval: {response.text}"
# Reset to global default
response = client.post("/reset_update_interval", data={"feed_url": feed_url})
assert response.status_code == 200, f"Failed to reset feed interval: {response.text}"
# Check that the feed page shows global default
encoded_url = urllib.parse.quote(feed_url)
response = client.get(f"/feed?feed_url={encoded_url}")
assert response.status_code == 200, f"Failed to get feed page: {response.text}"
assert "Using global default" in response.text, "Global default badge not shown after reset"
def test_update_interval_validation() -> None:
"""Test that update interval validation works."""
# Try to set an interval below minimum (should be clamped to 1)
response: Response = client.post("/set_global_update_interval", data={"interval_minutes": "0"})
assert response.status_code == 200, f"Failed to handle minimum interval: {response.text}"
# Try to set an interval above maximum (should be clamped to 10080)
response = client.post("/set_global_update_interval", data={"interval_minutes": "20000"})
assert response.status_code == 200, f"Failed to handle maximum interval: {response.text}"
# Clean up
client.post(url="/remove", data={"feed_url": feed_url})
client.post(url="/delete_webhook", data={"webhook_url": webhook_url})