489 lines
16 KiB
Python
489 lines
16 KiB
Python
#!/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
|