Add Atom and RSS feeds

scripts/build_with_projects.py

@@ -13,6 +13,7 @@ import sys
 import uuid
 
 from project_sources import ProjectSourceError, sync_projects
+from feed_generator import prepare_feed_assets
 
 
 IGNORED_NAMES = {
@@ -34,7 +35,7 @@ def copy_site_source(site_root: Path, workspace: Path) -> None:
             shutil.copytree(
                 source,
                 destination,
-                ignore=shutil.ignore_patterns("__pycache__", "*.pyc"),
+                ignore=shutil.ignore_patterns(*IGNORED_NAMES, "*.pyc"),
             )
         elif source.is_file():
             shutil.copy2(source, destination)
@@ -99,6 +100,7 @@ def main() -> int:
             manifest["site_commit"] = site_commit
             manifest["site_dirty"] = dirty
             project_file = find_project_file(workspace)
+            prepare_feed_assets(project_file, workspace)
 
             if args.serve:
                 command = [

scripts/feed_generator.py

@@ -0,0 +1,488 @@
+#!/usr/bin/env python3
+"""Generate Labyricorn's Atom and RSS feeds from one shared feed catalog."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import date, datetime, time, timezone
+from email.utils import format_datetime as format_rfc2822_datetime
+from html import escape
+from html.parser import HTMLParser
+from pathlib import Path
+import re
+from typing import Any, Iterable
+from urllib.parse import urljoin, urlparse
+import xml.etree.ElementTree as ET
+
+
+CANONICAL_BASE_URL = "https://www.labyricorn.com/"
+DEFAULT_AUTHOR = "Christopher Chambers"
+GENERAL_FEED_LIMIT = 50
+ATOM_NS = "http://www.w3.org/2005/Atom"
+CONTENT_NS = "http://purl.org/rss/1.0/modules/content/"
+DC_NS = "http://purl.org/dc/elements/1.1/"
+XML_NS = "http://www.w3.org/XML/1998/namespace"
+EMPTY_FEED_DATE = datetime(1970, 1, 1, tzinfo=timezone.utc)
+INVALID_XML_CHARACTERS = re.compile(
+    "[\x00-\x08\x0b\x0c\x0e-\x1f\ud800-\udfff\ufffe\uffff]"
+)
+
+
+@dataclass(frozen=True)
+class FeedEntry:
+    title: str
+    canonical_url: str
+    stable_id: str
+    summary: str
+    content_html: str
+    published: datetime
+    updated: datetime
+    author: str | None
+
+
+@dataclass(frozen=True)
+class Feed:
+    title: str
+    description: str
+    canonical_url: str
+    atom_url: str
+    rss_url: str
+    entries: tuple[FeedEntry, ...]
+    author: str = DEFAULT_AUTHOR
+
+    @property
+    def updated(self) -> datetime:
+        return max((entry.updated for entry in self.entries), default=EMPTY_FEED_DATE)
+
+
+def _field(record: Any, name: str, default: Any = None) -> Any:
+    try:
+        value = record[name]
+    except (KeyError, TypeError):
+        return default
+    if value is None or value.__class__.__name__ == "Undefined":
+        return default
+    return value
+
+
+def _text(value: Any, default: str = "") -> str:
+    if value is None or value.__class__.__name__ == "Undefined":
+        return default
+    return str(value)
+
+
+def _datetime(value: Any, fallback: Any = None) -> datetime:
+    if value is None or value.__class__.__name__ == "Undefined":
+        value = fallback
+    if hasattr(value, "datetime") and isinstance(value.datetime, datetime):
+        value = value.datetime
+    if isinstance(value, datetime):
+        result = value
+    elif isinstance(value, date):
+        result = datetime.combine(value, time.min)
+    elif value:
+        candidate = str(value).strip().replace("Z", "+00:00")
+        try:
+            result = datetime.fromisoformat(candidate)
+        except ValueError:
+            result = datetime.combine(date.fromisoformat(candidate[:10]), time.min)
+    else:
+        result = EMPTY_FEED_DATE
+    if result.tzinfo is None:
+        result = result.replace(tzinfo=timezone.utc)
+    return result.astimezone(timezone.utc)
+
+
+def _public(record: Any) -> bool:
+    return not bool(record.is_hidden) and not bool(record.is_undiscoverable)
+
+
+def _query_records(query: Any, *order_by: str) -> list[Any]:
+    if hasattr(query, "include_hidden"):
+        query = query.include_hidden(False)
+    if order_by:
+        query = query.order_by(*order_by)
+    return [record for record in query if _public(record)]
+
+
+def select_recent_activity(pad: Any) -> list[Any]:
+    """Return the homepage Recent Activity records in homepage display order."""
+    records: list[Any] = []
+    for section_path in ("/articles", "/blog"):
+        section = pad.get(section_path)
+        if section is not None:
+            records.extend(_query_records(section.children))
+
+    projects = pad.get("/projects")
+    if projects is not None:
+        for project in _query_records(projects.children):
+            if _text(_field(project, "_model")) != "project":
+                continue
+            devlog = pad.get(f"{project.path}/devlog")
+            if devlog is None:
+                continue
+            devlog_entries = _query_records(
+                devlog.children, "-date", "-source_commit_time"
+            )
+            if devlog_entries:
+                records.append(devlog_entries[0])
+
+    return sorted(
+        records,
+        key=lambda record: _datetime(_field(record, "date")),
+        reverse=True,
+    )
+
+
+def _canonical_url(record: Any) -> str:
+    return urljoin(CANONICAL_BASE_URL, record.url_path.lstrip("/"))
+
+
+def _published(record: Any) -> datetime:
+    return _datetime(_field(record, "date", _field(record, "started")))
+
+
+def _updated(record: Any) -> datetime:
+    model = _text(_field(record, "_model"))
+    if model == "project":
+        value = _field(
+            record,
+            "repository_commit_date",
+            _field(record, "synchronized_at", _field(record, "date")),
+        )
+    elif model == "devlog-entry":
+        value = _field(record, "source_commit_time", _field(record, "date"))
+    else:
+        value = _field(record, "updated", _field(record, "date"))
+    return _datetime(value, _field(record, "date"))
+
+
+class _AbsoluteURLHTMLParser(HTMLParser):
+    URL_ATTRIBUTES = {"action", "href", "poster", "src"}
+
+    def __init__(self, base_url: str) -> None:
+        super().__init__(convert_charrefs=False)
+        self.base_url = base_url
+        self.parts: list[str] = []
+
+    def _absolute(self, value: str) -> str:
+        parsed = urlparse(value)
+        if parsed.scheme == "javascript":
+            return "#"
+        if parsed.scheme in {"data", "mailto", "tel"}:
+            return value
+        return urljoin(self.base_url, value)
+
+    def _attributes(self, attrs: list[tuple[str, str | None]]) -> str:
+        rendered: list[str] = []
+        for name, value in attrs:
+            if value is None:
+                rendered.append(name)
+                continue
+            if name.lower() in self.URL_ATTRIBUTES:
+                value = self._absolute(value)
+            elif name.lower() == "srcset":
+                candidates = []
+                for candidate in value.split(","):
+                    pieces = candidate.strip().split(None, 1)
+                    if pieces:
+                        pieces[0] = self._absolute(pieces[0])
+                    candidates.append(" ".join(pieces))
+                value = ", ".join(candidates)
+            rendered.append(f'{name}="{escape(value, quote=True)}"')
+        return (" " + " ".join(rendered)) if rendered else ""
+
+    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        self.parts.append(f"<{tag}{self._attributes(attrs)}>")
+
+    def handle_startendtag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        self.parts.append(f"<{tag}{self._attributes(attrs)} />")
+
+    def handle_endtag(self, tag: str) -> None:
+        self.parts.append(f"</{tag}>")
+
+    def handle_data(self, data: str) -> None:
+        self.parts.append(data)
+
+    def handle_entityref(self, name: str) -> None:
+        self.parts.append(f"&{name};")
+
+    def handle_charref(self, name: str) -> None:
+        self.parts.append(f"&#{name};")
+
+    def handle_comment(self, data: str) -> None:
+        self.parts.append(f"<!--{data}-->")
+
+    def handle_decl(self, decl: str) -> None:
+        self.parts.append(f"<!{decl}>")
+
+    def unknown_decl(self, data: str) -> None:
+        self.parts.append(f"<![{data}]>")
+
+
+def absolute_html_urls(html: str, base_url: str) -> str:
+    parser = _AbsoluteURLHTMLParser(base_url)
+    parser.feed(html)
+    parser.close()
+    return "".join(parser.parts)
+
+
+def _render_body(record: Any, canonical_url: str) -> str:
+    body = _field(record, "body")
+    if body is None:
+        return ""
+    from lektor.context import Context
+
+    context = Context(pad=record.pad)
+    context.source = record
+    context._forced_base_url = record.url_path
+    with context:
+        rendered = str(body)
+    return absolute_html_urls(rendered, canonical_url)
+
+
+def _entry(record: Any) -> FeedEntry:
+    canonical_url = _canonical_url(record)
+    author = _text(_field(record, "author")) or None
+    published = _published(record)
+    return FeedEntry(
+        title=_text(_field(record, "title"), "Untitled"),
+        canonical_url=canonical_url,
+        stable_id=canonical_url,
+        summary=_text(_field(record, "summary")),
+        content_html=_render_body(record, canonical_url),
+        published=published,
+        updated=max(published, _updated(record)),
+        author=author,
+    )
+
+
+def _general_feed(
+    title: str,
+    description: str,
+    canonical_path: str,
+    feed_path: str,
+    rss_path: str,
+    records: Iterable[Any],
+) -> Feed:
+    return Feed(
+        title=title,
+        description=description,
+        canonical_url=urljoin(CANONICAL_BASE_URL, canonical_path.lstrip("/")),
+        atom_url=urljoin(CANONICAL_BASE_URL, feed_path.lstrip("/")),
+        rss_url=urljoin(CANONICAL_BASE_URL, rss_path.lstrip("/")),
+        entries=tuple(_entry(record) for record in list(records)[:GENERAL_FEED_LIMIT]),
+    )
+
+
+def build_feed_catalog(pad: Any) -> tuple[list[Any], tuple[Feed, ...]]:
+    recent_records = select_recent_activity(pad)
+    articles = pad.get("/articles")
+    blog = pad.get("/blog")
+    article_records = _query_records(articles.children, "-date") if articles else []
+    blog_records = _query_records(blog.children, "-date") if blog else []
+
+    feeds: list[Feed] = [
+        _general_feed(
+            "Labyricorn - Recent Activity",
+            "Recent articles, blog dispatches, and the latest public devlog update from each project.",
+            "/",
+            "/feed.xml",
+            "/rss.xml",
+            recent_records,
+        ),
+        _general_feed(
+            "Labyricorn - Articles",
+            _text(_field(articles, "summary"), "Published Labyricorn articles."),
+            "/articles/",
+            "/articles/feed.xml",
+            "/articles/rss.xml",
+            article_records,
+        ),
+        _general_feed(
+            "Labyricorn - Blog",
+            _text(_field(blog, "summary"), "Published Labyricorn blog entries."),
+            "/blog/",
+            "/blog/feed.xml",
+            "/blog/rss.xml",
+            blog_records,
+        ),
+    ]
+
+    projects = pad.get("/projects")
+    if projects is not None:
+        for project in _query_records(projects.children):
+            if _text(_field(project, "_model")) != "project":
+                continue
+            devlog = pad.get(f"{project.path}/devlog")
+            devlog_entries = (
+                _query_records(devlog.children, "date", "source_commit_time")
+                if devlog is not None
+                else []
+            )
+            project_url = _canonical_url(project)
+            feeds.append(
+                Feed(
+                    title=f"Labyricorn - {_text(_field(project, 'title'), 'Project')}",
+                    description=_text(_field(project, "summary")),
+                    canonical_url=project_url,
+                    atom_url=urljoin(project_url, "feed.xml"),
+                    rss_url=urljoin(project_url, "rss.xml"),
+                    entries=tuple(
+                        [_entry(project), *(_entry(item) for item in devlog_entries)]
+                    ),
+                    author=_text(_field(project, "author")) or DEFAULT_AUTHOR,
+                )
+            )
+
+    return recent_records, tuple(feeds)
+
+
+def _clean_xml(value: str) -> str:
+    return INVALID_XML_CHARACTERS.sub("", value)
+
+
+def _atom_element(
+    parent: ET.Element,
+    name: str,
+    text: str | None = None,
+    **attrs: str,
+) -> ET.Element:
+    element = ET.SubElement(parent, f"{{{ATOM_NS}}}{name}", attrs)
+    if text is not None:
+        element.text = _clean_xml(text)
+    return element
+
+
+def atom_xml(feed: Feed) -> bytes:
+    ET.register_namespace("", ATOM_NS)
+    root = ET.Element(f"{{{ATOM_NS}}}feed")
+    _atom_element(root, "title", feed.title)
+    _atom_element(root, "id", feed.atom_url)
+    _atom_element(root, "updated", feed.updated.isoformat().replace("+00:00", "Z"))
+    _atom_element(
+        root,
+        "link",
+        rel="self",
+        href=feed.atom_url,
+        type="application/atom+xml",
+    )
+    _atom_element(
+        root,
+        "link",
+        rel="alternate",
+        href=feed.canonical_url,
+        type="text/html",
+    )
+    author = _atom_element(root, "author")
+    _atom_element(author, "name", feed.author)
+
+    for item in feed.entries:
+        entry = _atom_element(root, "entry")
+        _atom_element(entry, "title", item.title)
+        _atom_element(entry, "id", item.stable_id)
+        _atom_element(
+            entry,
+            "link",
+            rel="alternate",
+            href=item.canonical_url,
+            type="text/html",
+        )
+        _atom_element(entry, "published", item.published.isoformat().replace("+00:00", "Z"))
+        _atom_element(entry, "updated", item.updated.isoformat().replace("+00:00", "Z"))
+        if item.author:
+            entry_author = _atom_element(entry, "author")
+            _atom_element(entry_author, "name", item.author)
+        if item.summary:
+            _atom_element(entry, "summary", item.summary, type="text")
+        content = _atom_element(
+            entry,
+            "content",
+            item.content_html,
+            type="html",
+        )
+        content.set(f"{{{XML_NS}}}base", item.canonical_url)
+
+    ET.indent(root, space="  ")
+    return ET.tostring(
+        root,
+        encoding="utf-8",
+        xml_declaration=True,
+    ) + b"\n"
+
+
+def rss_xml(feed: Feed) -> bytes:
+    ET.register_namespace("atom", ATOM_NS)
+    ET.register_namespace("content", CONTENT_NS)
+    ET.register_namespace("dc", DC_NS)
+    root = ET.Element("rss", {"version": "2.0"})
+    channel = ET.SubElement(root, "channel")
+    ET.SubElement(channel, "title").text = _clean_xml(feed.title)
+    ET.SubElement(channel, "link").text = feed.canonical_url
+    ET.SubElement(channel, "description").text = _clean_xml(feed.description)
+    ET.SubElement(channel, "language").text = "en-us"
+    ET.SubElement(channel, "lastBuildDate").text = format_rfc2822_datetime(feed.updated)
+    ET.SubElement(
+        channel,
+        f"{{{ATOM_NS}}}link",
+        {"rel": "self", "href": feed.rss_url, "type": "application/rss+xml"},
+    )
+
+    for entry in feed.entries:
+        item = ET.SubElement(channel, "item")
+        ET.SubElement(item, "title").text = _clean_xml(entry.title)
+        ET.SubElement(item, "link").text = entry.canonical_url
+        ET.SubElement(item, "guid", {"isPermaLink": "true"}).text = entry.stable_id
+        ET.SubElement(item, "pubDate").text = format_rfc2822_datetime(entry.published)
+        ET.SubElement(item, f"{{{ATOM_NS}}}updated").text = entry.updated.isoformat().replace(
+            "+00:00", "Z"
+        )
+        if entry.author:
+            ET.SubElement(item, f"{{{DC_NS}}}creator").text = _clean_xml(entry.author)
+        ET.SubElement(item, "description").text = _clean_xml(entry.summary)
+        ET.SubElement(item, f"{{{CONTENT_NS}}}encoded").text = _clean_xml(
+            entry.content_html
+        )
+
+    ET.indent(root, space="  ")
+    return ET.tostring(root, encoding="utf-8", xml_declaration=True) + b"\n"
+
+
+def _asset_path(workspace: Path, public_url: str) -> Path:
+    relative = urlparse(public_url).path.lstrip("/")
+    return workspace / "assets" / Path(relative)
+
+
+def _write_recent_activity_paths(workspace: Path, records: Iterable[Any]) -> None:
+    root_record = workspace / "content" / "contents.lr"
+    text = root_record.read_text(encoding="utf-8").rstrip("\r\n")
+    marker = "\n---\nrecent_activity_paths:\n"
+    if marker in text:
+        text = text.split(marker, 1)[0].rstrip("\r\n")
+    paths = "\n".join(record.path for record in records)
+    root_record.write_text(
+        f"{text}{marker}\n{paths}\n",
+        encoding="utf-8",
+    )
+
+
+def prepare_feed_assets(project_file: Path, workspace: Path) -> tuple[Feed, ...]:
+    """Create feed assets and homepage activity data inside an isolated workspace."""
+    from lektor.db import Database
+    from lektor.environment import Environment
+    from lektor.project import Project
+
+    project = Project.from_file(str(project_file))
+    environment = Environment(project, load_plugins=False)
+    pad = Database(environment).new_pad()
+    recent_records, feeds = build_feed_catalog(pad)
+    _write_recent_activity_paths(workspace, recent_records)
+
+    for feed in feeds:
+        atom_path = _asset_path(workspace, feed.atom_url)
+        rss_path = _asset_path(workspace, feed.rss_url)
+        for path in (atom_path, rss_path):
+            path.parent.mkdir(parents=True, exist_ok=True)
+        atom_path.write_bytes(atom_xml(feed))
+        rss_path.write_bytes(rss_xml(feed))
+    return feeds