diff --git a/src/feed_errors.py b/src/feed_errors.py new file mode 100644 index 0000000..fdb4f9a --- /dev/null +++ b/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}") \ No newline at end of file diff --git a/src/feed_handler.py b/src/feed_handler.py index 4127156..8d4b1a0 100644 --- a/src/feed_handler.py +++ b/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] - title = entry["title"] - link = entry["link"] + feed_update = FeedUpdate ( + entry_id = feed.entries[0]["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}") - 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 : +def grab_royal_road_chapter_information(rss_feed_url: str) -> FeedUpdate : """ 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 ( - chapter_title = chapter_title, + feed_update = FeedUpdate ( + 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.debug(f"[Feed] Latest chapter for story '{royal_road_story.story_title}' (RSS URL {rss_feed_url}):" - f"{royal_road_story.chapter_title} (ID {royal_road_story.chapter_id}) -> {royal_road_story.link}") - return royal_road_story + logger.info(f"[Feed] Latest chapter for story '{feed_update.feed_title}': {feed_update.body_title}") + logger.debug(f"[Feed] Latest chapter for story '{feed_update.feed_title}' (RSS URL {rss_feed_url}):" + f"{feed_update.body_title} (ID {feed_update.entry_id}) -> {feed_update.link}") + 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)) diff --git a/src/models.py b/src/models.py new file mode 100644 index 0000000..9afce99 --- /dev/null +++ b/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/[/] + 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/[/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 \ No newline at end of file