Added subclasses that denote which notification type should be sent
This commit is contained in:
1 parent
55fad17310
commit
f5b58170ec
2 files changed
+14
-7
No files matched your search
+6
-6
@@ -5,7 +5,7 @@ import feedparser
|
||||
import requests
|
||||
|
||||
from feed_errors import FeedCheckFailedError, FeedNotFoundError
|
||||
from models import FeedEntryType, FeedUpdate
|
||||
from models import FeedEntryType, GiteaUpdate, RoyalRoadUpdate
|
||||
from simple_logger_handler import setup_logger
|
||||
|
||||
logger = setup_logger(__name__)
|
||||
@@ -54,7 +54,7 @@ def get_rss_feed_data(rss_feed_url: str, max_retries: int = 3) -> feedparser.Fee
|
||||
logger.debug(f"[Feed] Parsed {len(feed.entries)} entries from {rss_feed_url}")
|
||||
return feed
|
||||
|
||||
def grab_gitea_release_information(rss_feed_url: str) -> FeedUpdate:
|
||||
def grab_gitea_release_information(rss_feed_url: str) -> GiteaUpdate:
|
||||
"""
|
||||
Fetches the latest release information from a Gitea releases RSS feed.
|
||||
|
||||
@@ -65,7 +65,7 @@ def grab_gitea_release_information(rss_feed_url: str) -> FeedUpdate:
|
||||
ValueError: If the feed has no entries.
|
||||
"""
|
||||
feed = get_rss_feed_data(rss_feed_url)
|
||||
feed_update = FeedUpdate (
|
||||
feed_update = GiteaUpdate (
|
||||
entry_id = feed.entries[0]["link"],
|
||||
feed_title = feed.feed.title,
|
||||
body_title = feed.entries[0]["title"],
|
||||
@@ -75,7 +75,7 @@ def grab_gitea_release_information(rss_feed_url: str) -> FeedUpdate:
|
||||
logger.debug(f"[Feed] Latest release for feed '{rss_feed_url}': {feed_update.feed_title} -> {feed_update.link}")
|
||||
return feed_update
|
||||
|
||||
def grab_royal_road_chapter_information(rss_feed_url: str) -> FeedUpdate :
|
||||
def grab_royal_road_chapter_information(rss_feed_url: str) -> RoyalRoadUpdate :
|
||||
"""
|
||||
Fetches the latest chapter information from a Royalroad RSS feed.
|
||||
|
||||
@@ -93,7 +93,7 @@ def grab_royal_road_chapter_information(rss_feed_url: str) -> FeedUpdate :
|
||||
chapter_title_with_story_title = latest_chapter_data["title"]
|
||||
body_title = chapter_title_with_story_title.removeprefix(feed.feed.title).removeprefix(" - ")
|
||||
|
||||
feed_update = FeedUpdate (
|
||||
feed_update = RoyalRoadUpdate (
|
||||
entry_id = str(latest_chapter_data["id"]),
|
||||
feed_title = feed.feed.title,
|
||||
body_title = body_title,
|
||||
@@ -143,7 +143,7 @@ def ensure_feed_exists(url: str, timeout: tuple[float, float] = (3.0, 5.0)) -> N
|
||||
if status != 200:
|
||||
raise FeedCheckFailedError(url)
|
||||
|
||||
def fetch_latest(feed_type: FeedEntryType, url: str) -> FeedUpdate:
|
||||
def fetch_latest(feed_type: FeedEntryType, url: str) -> RoyalRoadUpdate | GiteaUpdate:
|
||||
match feed_type:
|
||||
case FeedEntryType.ROYALROAD:
|
||||
return grab_royal_road_chapter_information(url)
|
||||
|
||||
+8
-1
@@ -1,6 +1,6 @@
|
||||
from dataclasses import dataclass
|
||||
from enum import StrEnum
|
||||
from typing import Self
|
||||
from typing import ClassVar, Self
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
from pydantic import BaseModel, Field, field_validator, model_validator
|
||||
@@ -14,12 +14,19 @@ class FeedUpdate():
|
||||
entry_id: str
|
||||
feed_title: str
|
||||
body_title: str
|
||||
notification_type: ClassVar[str]
|
||||
link: str | None = None
|
||||
|
||||
class FeedEntryType(StrEnum):
|
||||
ROYALROAD = "royalroad"
|
||||
GITEA = "gitea"
|
||||
|
||||
class RoyalRoadUpdate(FeedUpdate):
|
||||
notification_type = "royal-road"
|
||||
|
||||
|
||||
class GiteaUpdate(FeedUpdate):
|
||||
notification_type = "utility"
|
||||
|
||||
def _path_segments(raw: str, feed_type: FeedEntryType, host: str) -> list[str]:
|
||||
"""Check scheme and host, then return the non-empty path segments."""
|
||||
|
||||
Reference in new issue
Block a user