82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
from dataclasses import dataclass
|
|
from enum import StrEnum
|
|
from typing import Self
|
|
from urllib.parse import urlsplit
|
|
|
|
from pydantic import BaseModel, Field, field_validator, model_validator
|
|
|
|
from feed_errors import InvalidFeedUrlError, UnsupportedPathError, WrongHostError
|
|
|
|
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 _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 |