feat(community): RSS 2.0 feed generator + ActivityPub JSON-LD scaffold

This commit is contained in:
pyr0ball 2026-04-12 17:58:14 -07:00
parent 33188123d0
commit 7001d48378
4 changed files with 185 additions and 0 deletions

View file

@ -0,0 +1,44 @@
# app/services/community/ap_compat.py
# MIT License — AP (ActivityPub) scaffold only (no actor, inbox, outbox)
from __future__ import annotations
from datetime import datetime, timezone
def post_to_ap_json_ld(post: dict, base_url: str) -> dict:
"""Serialize a community post dict to an ActivityPub-compatible JSON-LD Note.
This is a read-only scaffold. No AP actor, inbox, or outbox is implemented.
The slug URI is stable so a future full AP implementation can envelope posts
without a database migration.
"""
slug = post["slug"]
published = post.get("published")
if isinstance(published, datetime):
published_str = published.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
else:
published_str = str(published)
dietary_tags: list[str] = post.get("dietary_tags") or []
tags = [{"type": "Hashtag", "name": "#kiwi"}]
for tag in dietary_tags:
tags.append({"type": "Hashtag", "name": f"#{tag.replace('-', '').replace(' ', '')}"})
return {
"@context": "https://www.w3.org/ns/activitystreams",
"type": "Note",
"id": f"{base_url}/api/v1/community/posts/{slug}",
"attributedTo": post.get("pseudonym", "anonymous"),
"content": _build_content(post),
"published": published_str,
"tag": tags,
}
def _build_content(post: dict) -> str:
title = post.get("title") or "Untitled"
desc = post.get("description")
if desc:
return f"{title}{desc}"
return title

View file

@ -0,0 +1,43 @@
# app/services/community/feed.py
# MIT License
from __future__ import annotations
from datetime import datetime, timezone
from email.utils import format_datetime
from xml.etree.ElementTree import Element, SubElement, tostring
def posts_to_rss(posts: list[dict], base_url: str) -> str:
"""Generate an RSS 2.0 feed from a list of community post dicts.
base_url: the root URL of this Kiwi instance (no trailing slash).
Returns UTF-8 XML string.
"""
rss = Element("rss", version="2.0")
channel = SubElement(rss, "channel")
_sub(channel, "title", "Kiwi Community Feed")
_sub(channel, "link", f"{base_url}/community")
_sub(channel, "description", "Meal plans and recipe outcomes from the Kiwi community")
_sub(channel, "language", "en")
_sub(channel, "lastBuildDate", format_datetime(datetime.now(timezone.utc)))
for post in posts:
item = SubElement(channel, "item")
_sub(item, "title", post.get("title") or "Untitled")
_sub(item, "link", f"{base_url}/api/v1/community/posts/{post['slug']}")
_sub(item, "guid", f"{base_url}/api/v1/community/posts/{post['slug']}")
if post.get("description"):
_sub(item, "description", post["description"])
published = post.get("published")
if isinstance(published, datetime):
_sub(item, "pubDate", format_datetime(published))
return '<?xml version="1.0" encoding="UTF-8"?>\n' + tostring(rss, encoding="unicode")
def _sub(parent: Element, tag: str, text: str) -> Element:
el = SubElement(parent, tag)
el.text = text
return el

View file

@ -0,0 +1,47 @@
# tests/services/community/test_ap_compat.py
import pytest
from datetime import datetime, timezone
from app.services.community.ap_compat import post_to_ap_json_ld
POST = {
"slug": "kiwi-plan-test-pasta-week",
"title": "Pasta Week",
"description": "Seven days of carbs",
"published": datetime(2026, 4, 12, 12, 0, 0, tzinfo=timezone.utc),
"pseudonym": "PastaWitch",
"dietary_tags": ["vegetarian"],
}
def test_ap_json_ld_context():
doc = post_to_ap_json_ld(POST, base_url="https://menagerie.circuitforge.tech/kiwi")
assert doc["@context"] == "https://www.w3.org/ns/activitystreams"
def test_ap_json_ld_type():
doc = post_to_ap_json_ld(POST, base_url="https://menagerie.circuitforge.tech/kiwi")
assert doc["type"] == "Note"
def test_ap_json_ld_id_is_uri():
doc = post_to_ap_json_ld(POST, base_url="https://menagerie.circuitforge.tech/kiwi")
assert doc["id"].startswith("https://")
assert POST["slug"] in doc["id"]
def test_ap_json_ld_published_is_iso8601():
doc = post_to_ap_json_ld(POST, base_url="https://menagerie.circuitforge.tech/kiwi")
from datetime import datetime
datetime.fromisoformat(doc["published"].replace("Z", "+00:00"))
def test_ap_json_ld_attributed_to_pseudonym():
doc = post_to_ap_json_ld(POST, base_url="https://menagerie.circuitforge.tech/kiwi")
assert doc["attributedTo"] == "PastaWitch"
def test_ap_json_ld_tags_include_kiwi():
doc = post_to_ap_json_ld(POST, base_url="https://menagerie.circuitforge.tech/kiwi")
tag_names = [t["name"] for t in doc.get("tag", [])]
assert "#kiwi" in tag_names

View file

@ -0,0 +1,51 @@
# tests/services/community/test_feed.py
import pytest
from datetime import datetime, timezone
from app.services.community.feed import posts_to_rss
def make_post_dict(**kwargs):
defaults = dict(
slug="kiwi-plan-test-pasta-week",
title="Pasta Week",
description="Seven days of carbs",
published=datetime(2026, 4, 12, 12, 0, 0, tzinfo=timezone.utc),
post_type="plan",
pseudonym="PastaWitch",
)
defaults.update(kwargs)
return defaults
def test_rss_is_valid_xml():
import xml.etree.ElementTree as ET
rss = posts_to_rss([make_post_dict()], base_url="https://menagerie.circuitforge.tech/kiwi")
root = ET.fromstring(rss)
assert root.tag == "rss"
assert root.attrib.get("version") == "2.0"
def test_rss_contains_item():
import xml.etree.ElementTree as ET
rss = posts_to_rss([make_post_dict()], base_url="https://menagerie.circuitforge.tech/kiwi")
root = ET.fromstring(rss)
items = root.findall(".//item")
assert len(items) == 1
def test_rss_item_has_required_fields():
import xml.etree.ElementTree as ET
rss = posts_to_rss([make_post_dict()], base_url="https://menagerie.circuitforge.tech/kiwi")
root = ET.fromstring(rss)
item = root.find(".//item")
assert item.find("title") is not None
assert item.find("link") is not None
assert item.find("pubDate") is not None
def test_rss_empty_posts():
import xml.etree.ElementTree as ET
rss = posts_to_rss([], base_url="https://menagerie.circuitforge.tech/kiwi")
root = ET.fromstring(rss)
items = root.findall(".//item")
assert len(items) == 0