From 7001d483786f45d71fe15cb0c34e5a28ceb7a779 Mon Sep 17 00:00:00 2001 From: pyr0ball Date: Sun, 12 Apr 2026 17:58:14 -0700 Subject: [PATCH] feat(community): RSS 2.0 feed generator + ActivityPub JSON-LD scaffold --- app/services/community/ap_compat.py | 44 +++++++++++++++++++ app/services/community/feed.py | 43 ++++++++++++++++++ tests/services/community/test_ap_compat.py | 47 ++++++++++++++++++++ tests/services/community/test_feed.py | 51 ++++++++++++++++++++++ 4 files changed, 185 insertions(+) create mode 100644 app/services/community/ap_compat.py create mode 100644 app/services/community/feed.py create mode 100644 tests/services/community/test_ap_compat.py create mode 100644 tests/services/community/test_feed.py diff --git a/app/services/community/ap_compat.py b/app/services/community/ap_compat.py new file mode 100644 index 0000000..c3ac181 --- /dev/null +++ b/app/services/community/ap_compat.py @@ -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 diff --git a/app/services/community/feed.py b/app/services/community/feed.py new file mode 100644 index 0000000..36000b9 --- /dev/null +++ b/app/services/community/feed.py @@ -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 '\n' + tostring(rss, encoding="unicode") + + +def _sub(parent: Element, tag: str, text: str) -> Element: + el = SubElement(parent, tag) + el.text = text + return el diff --git a/tests/services/community/test_ap_compat.py b/tests/services/community/test_ap_compat.py new file mode 100644 index 0000000..524eb58 --- /dev/null +++ b/tests/services/community/test_ap_compat.py @@ -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 diff --git a/tests/services/community/test_feed.py b/tests/services/community/test_feed.py new file mode 100644 index 0000000..bc23f8c --- /dev/null +++ b/tests/services/community/test_feed.py @@ -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