BaseModel to validate RSS feed url input

src/feed_errors.py

@@ -0,0 +1,35 @@
+class FeedError(Exception):
+	"""Base class for every feed problem the API can explain to a client."""
+	status_code = 400
+
+
+class InvalidFeedUrlError(FeedError, ValueError):
+	"""The URL is malformed or doesn't fit the chosen feed type.
+
+	Subclasses ValueError so Pydantic reports it as a normal 422 validation error.
+	"""
+	status_code = 422
+
+
+class WrongHostError(InvalidFeedUrlError):
+	def __init__(self, feed_type: str, expected_host: str):
+		super().__init__(f"A {feed_type} feed must be on {expected_host}")
+
+
+class UnsupportedPathError(InvalidFeedUrlError):
+	def __init__(self, hint: str):
+		super().__init__(f"Unrecognised feed path ({hint})")
+
+
+class FeedNotFoundError(FeedError):
+	status_code = 422
+
+	def __init__(self, url: str):
+		super().__init__(f"The feed does not exist: {url}")
+
+
+class FeedCheckFailedError(FeedError):
+	status_code = 502
+
+	def __init__(self, url: str):
+		super().__init__(f"Could not verify the feed right now, try again later: {url}")

src/feed_handler.py

@@ -1,11 +1,10 @@
 import time
-from dataclasses import dataclass
-from typing import Tuple
 
 import feedparser
 import requests
-from pydantic import BaseModel
 
+from feed_errors import FeedCheckFailedError, FeedNotFoundError
+from models import FeedUpdate
 from simple_logger_handler import setup_logger
 
 logger = setup_logger(__name__)
@@ -14,14 +13,6 @@ _GITEA_INSTANCE_BASE_URL="https://git.gansejunge.com"
 _GITEA_REPOSITORY_OWNER="utility"
 
 
-@dataclass(frozen=True)
-class RoyalRoadStory(BaseModel):
-	chapter_title: str
-	link: str
-	story_title: str
-	chapter_id: str
-
-
 def get_rss_feed_data(rss_feed_url: str, max_retries: int = 3) -> feedparser.FeedParserDict:
 	"""
 	Fetches information from any RSS feed, with retries on network-related errors.
@@ -62,26 +53,27 @@ 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) -> Tuple[str, str]:
+def grab_gitea_release_information(rss_feed_url: str) -> FeedUpdate:
 	"""
 	Fetches the latest release information from a Gitea releases RSS feed.
 
 	Returns:
-		A tuple: (release_title, release_link)
+		FeedUpdate:
 
 	Raises:
 		ValueError: If the feed has no entries.
 	"""
 	feed = get_rss_feed_data(rss_feed_url)
-	entry = feed.entries[0]
+	feed_update = FeedUpdate (
-	title = entry["title"]
+		entry_id = feed.entries[0]["link"],
-	link = entry["link"]
+		feed_title = feed.feed.title,
+		body_title = feed.entries[0]["title"],
+	)
+	logger.info(f"[Feed] Latest release for feed '{rss_feed_url}': {feed_update.feed_title}")
+	logger.debug(f"[Feed] Latest release for feed '{rss_feed_url}': {feed_update.feed_title} -> {feed_update.link}")
+	return feed_update
 
-	logger.info(f"[Feed] Latest release for feed '{rss_feed_url}': {title}")
+def grab_royal_road_chapter_information(rss_feed_url: str) -> FeedUpdate :
-	logger.debug(f"[Feed] Latest release for feed '{rss_feed_url}': {title} -> {link}")
-	return title, link
-
-def grab_royal_road_chapter_information(rss_feed_url: str) -> RoyalRoadStory :
 	"""
 	Fetches the latest chapter information from a Royalroad RSS feed.
 
@@ -89,7 +81,7 @@ def grab_royal_road_chapter_information(rss_feed_url: str) -> RoyalRoadStory :
 		rss_feed_url: URL of the Royalroad story's RSS feed.
 
 	Returns:
-		A tuple: (chapter_title, chapter_link, story_title, chapter_id)
+		FeedUpdate
 
 	Raises:
 		ValueError: If the feed has no entries.
@@ -97,19 +89,19 @@ def grab_royal_road_chapter_information(rss_feed_url: str) -> RoyalRoadStory :
 	feed = get_rss_feed_data(rss_feed_url)
 	latest_chapter_data = feed.entries[0]
 	chapter_title_with_story_title = latest_chapter_data["title"]
-	chapter_title = chapter_title_with_story_title.removeprefix(feed.feed.title).removeprefix(" - ")
+	body_title = chapter_title_with_story_title.removeprefix(feed.feed.title).removeprefix(" - ")
 
-	royal_road_story =  RoyalRoadStory (
+	feed_update =  FeedUpdate (
-		chapter_title = chapter_title,
+		entry_id = str(latest_chapter_data["id"]),
+		feed_title = feed.feed.title,
+		body_title = body_title,
 		link = latest_chapter_data["link"],
-		story_title = feed.feed.title,
-		chapter_id = str(latest_chapter_data["id"])
 	)
 
-	logger.info(f"[Feed] Latest chapter for story '{royal_road_story.story_title}': {royal_road_story.chapter_title}")
+	logger.info(f"[Feed] Latest chapter for story '{feed_update.feed_title}': {feed_update.body_title}")
-	logger.debug(f"[Feed] Latest chapter for story '{royal_road_story.story_title}' (RSS URL {rss_feed_url}):"
+	logger.debug(f"[Feed] Latest chapter for story '{feed_update.feed_title}' (RSS URL {rss_feed_url}):"
-				f"{royal_road_story.chapter_title} (ID {royal_road_story.chapter_id}) -> {royal_road_story.link}")
+				f"{feed_update.body_title} (ID {feed_update.entry_id}) -> {feed_update.link}")
-	return royal_road_story
+	return feed_update
 
 def build_gitea_runner(version: str, gitea_token: str) -> bool:
 	url = (
@@ -138,6 +130,17 @@ def build_gitea_runner(version: str, gitea_token: str) -> bool:
 		logger.warning(f"[Feed] Failed to run Gitea-Runner build: {e}")
 	return False
 
+def ensure_feed_exists(url: str, timeout: tuple[float, float] = (3.0, 5.0)) -> None:
+	try:
+		with requests.get(url, stream=True, timeout=timeout, allow_redirects=False) as response:
+			status = response.status_code
+	except requests.RequestException as e:
+		raise FeedCheckFailedError(url) from e
+	if status == 404:
+		raise FeedNotFoundError(url)
+	if status != 200:
+		raise FeedCheckFailedError(url)
+
 if __name__ == "__main__":
 	#rsssss="https://gitea.com/gitea/runner/releases.rss"
 	#print(get_rss_feed_data(rsssss))

src/models.py

@@ -0,0 +1,92 @@
+from dataclasses import dataclass
+from enum import StrEnum
+from typing import Self, assert_never
+from urllib.parse import urlsplit
+
+from pydantic import BaseModel, Field, field_validator, model_validator
+
+from feed_errors import InvalidFeedUrlError, UnsupportedPathError, WrongHostError
+from feed_handler import grab_gitea_release_information, grab_royal_road_chapter_information
+
+REPO_CHARS = frozenset("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_.-")
+
+@dataclass(frozen=True)
+class FeedUpdate():
+	entry_id: str
+	feed_title: str
+	body_title: str
+	link: str | None = None
+
+class FeedEntryType(StrEnum):
+	ROYALROAD = "royalroad"
+	GITEA = "gitea"
+
+def fetch_latest(feed_type: FeedEntryType, url: str) -> FeedUpdate:
+	match feed_type:
+		case FeedEntryType.ROYALROAD:
+			return grab_royal_road_chapter_information(url)
+		case FeedEntryType.GITEA:
+			return grab_gitea_release_information(url)
+		case _:
+			assert_never(feed_type)
+		
+
+def _path_segments(raw: str, feed_type: FeedEntryType, host: str) -> list[str]:
+	"""Check scheme and host, then return the non-empty path segments."""
+	raw = raw.strip()
+	if "://" not in raw:
+		raw = "https://" + raw  # allow "royalroad.com/fiction/118891"
+	try:
+		parts = urlsplit(raw)
+	except ValueError as e:
+		raise InvalidFeedUrlError("The URL could not be parsed") from e
+	if parts.scheme not in ("http", "https"):
+		raise InvalidFeedUrlError("The URL must use http or https")
+	# Comparing the whole netloc also rejects ports and user@host
+	if parts.netloc.lower().removeprefix("www.") != host:
+		raise WrongHostError(feed_type.value, host)
+	return [s for s in parts.path.split("/") if s]
+
+
+def _royalroad_feed(segments: list[str]) -> str:
+	# fiction/<id>[/<slug>]
+	if len(segments) >= 2 and segments[0] == "fiction":
+		story_id = segments[1]
+		if story_id.isascii() and story_id.isdigit():
+			return f"https://www.royalroad.com/fiction/syndication/{int(story_id)}"
+	raise UnsupportedPathError("expected e.g. royalroad.com/fiction/118891")
+
+
+def _gitea_feed(segments: list[str]) -> str:
+	# gitea/<repo>[/releases | /releases.rss]
+	if len(segments) >= 2 and segments[0] == "gitea":
+		repo, rest = segments[1], segments[2:]
+		repo_ok = repo[0] not in ".-" and set(repo) <= REPO_CHARS
+		if repo_ok and rest in ([], ["releases"], ["releases.rss"]):
+			return f"https://gitea.com/gitea/{repo}/releases.rss"
+	raise UnsupportedPathError("expected e.g. gitea.com/gitea/runner")
+
+
+# To support a new feed type: add an enum member, a function above, add a case tofetch_latest and a line here.
+_RULES = {
+	FeedEntryType.ROYALROAD: ("royalroad.com", _royalroad_feed),
+	FeedEntryType.GITEA: ("gitea.com", _gitea_feed),
+}
+
+
+class FeedEntry(BaseModel):
+	type: FeedEntryType
+	rss_feed_url: str = Field(max_length=512)
+	custom_name: str | None = Field(default=None, max_length=255)
+
+	@field_validator("custom_name")
+	@classmethod
+	def blank_name_is_none(cls, v: str | None) -> str | None:
+		return (v or "").strip() or None
+
+	@model_validator(mode="after")
+	def normalise_feed_url(self) -> Self:
+		host, build_feed_url = _RULES[self.type]
+		segments = _path_segments(self.rss_feed_url, self.type, host)
+		self.rss_feed_url = build_feed_url(segments)
+		return self