Files
mono/federation/sx/sx_components.py
giles c1ad6fd8d4 Replace Python sx_call loops with data-driven SX defcomps using map
Move rendering logic from Python for-loops building sx_call strings into
SX defcomp components that use map/lambda over data dicts. Python now
serializes display data into plain dicts and passes them via a single
sx_call; the SX layer handles iteration and conditional rendering.

Covers orders (rows, items, calendar, tickets), federation (timeline,
search, actors, profile activities), and blog (cards, pages, filters,
snippets, menu items, tag groups, page search, nav OOB).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-05 16:03:29 +00:00

732 lines
27 KiB
Python

"""
Federation service s-expression page components.
Renders social timeline, compose, search, following/followers, notifications,
actor profiles, login, and username selection pages.
"""
from __future__ import annotations
import os
from typing import Any
from markupsafe import escape
from shared.sx.jinja_bridge import load_service_components
from shared.sx.parser import serialize
from shared.sx.helpers import (
sx_call, SxExpr,
root_header_sx, full_page_sx, header_child_sx,
)
# Load federation-specific .sx components + handlers at import time
load_service_components(os.path.dirname(os.path.dirname(__file__)),
service_name="federation")
# ---------------------------------------------------------------------------
# Social header nav
# ---------------------------------------------------------------------------
def _social_nav_sx(actor: Any) -> str:
"""Build the social header nav bar content."""
from quart import url_for, request
if not actor:
choose_url = url_for("identity.choose_username_form")
return sx_call("federation-nav-choose-username", url=choose_url)
links = [
("social.defpage_home_timeline", "Timeline"),
("social.defpage_public_timeline", "Public"),
("social.defpage_compose_form", "Compose"),
("social.defpage_following_list", "Following"),
("social.defpage_followers_list", "Followers"),
("social.defpage_search", "Search"),
]
parts = []
for endpoint, label in links:
href = url_for(endpoint)
bold = " font-bold" if request.path == href else ""
cls = f"px-2 py-1 rounded hover:bg-stone-200{bold}"
parts.append(f'(a :href {serialize(href)} :class {serialize(cls)} {serialize(label)})')
# Notifications with live badge
notif_url = url_for("social.defpage_notifications")
notif_count_url = url_for("social.notification_count")
notif_bold = " font-bold" if request.path == notif_url else ""
parts.append(sx_call(
"federation-nav-notification-link",
href=notif_url,
cls=f"px-2 py-1 rounded hover:bg-stone-200 relative{notif_bold}",
count_url=notif_count_url,
))
# Profile link
profile_url = url_for("activitypub.actor_profile", username=actor.preferred_username)
parts.append(f'(a :href {serialize(profile_url)} :class "px-2 py-1 rounded hover:bg-stone-200" {serialize("@" + actor.preferred_username)})')
items_sx = "(<> " + " ".join(parts) + ")"
return sx_call("federation-nav-bar", items=SxExpr(items_sx))
def _social_header_sx(actor: Any) -> str:
"""Build the social section header row."""
nav_sx = _social_nav_sx(actor)
return sx_call("federation-social-header", nav=SxExpr(nav_sx))
def _social_page(ctx: dict, actor: Any, *, content: str,
title: str = "Rose Ash", meta_html: str = "") -> str:
"""Render a social page with header and content."""
hdr = root_header_sx(ctx)
social_hdr = _social_header_sx(actor)
child = header_child_sx(social_hdr)
header_rows = "(<> " + hdr + " " + child + ")"
return full_page_sx(ctx, header_rows=header_rows, content=content,
meta_html=meta_html or f'<title>{escape(title)}</title>')
# ---------------------------------------------------------------------------
# Post card
# ---------------------------------------------------------------------------
def _interaction_buttons_sx(item: Any, actor: Any) -> str:
"""Render like/boost/reply buttons for a post."""
from shared.browser.app.csrf import generate_csrf_token
from quart import url_for
oid = getattr(item, "object_id", "") or ""
ainbox = getattr(item, "author_inbox", "") or ""
lcount = getattr(item, "like_count", 0) or 0
bcount = getattr(item, "boost_count", 0) or 0
liked = getattr(item, "liked_by_me", False)
boosted = getattr(item, "boosted_by_me", False)
csrf = generate_csrf_token()
safe_id = oid.replace("/", "_").replace(":", "_")
target = f"#interactions-{safe_id}"
if liked:
like_action = url_for("social.unlike")
like_cls = "text-red-500 hover:text-red-600"
like_icon = "\u2665"
else:
like_action = url_for("social.like")
like_cls = "hover:text-red-500"
like_icon = "\u2661"
if boosted:
boost_action = url_for("social.unboost")
boost_cls = "text-green-600 hover:text-green-700"
else:
boost_action = url_for("social.boost")
boost_cls = "hover:text-green-600"
reply_url = url_for("social.defpage_compose_form", reply_to=oid) if oid else ""
reply_sx = sx_call("federation-reply-link", url=reply_url) if reply_url else ""
like_form = sx_call(
"federation-like-form",
action=like_action, target=target, oid=oid, ainbox=ainbox,
csrf=csrf, cls=f"flex items-center gap-1 {like_cls}",
icon=like_icon, count=str(lcount),
)
boost_form = sx_call(
"federation-boost-form",
action=boost_action, target=target, oid=oid, ainbox=ainbox,
csrf=csrf, cls=f"flex items-center gap-1 {boost_cls}",
count=str(bcount),
)
return sx_call(
"federation-interaction-buttons",
like=SxExpr(like_form),
boost=SxExpr(boost_form),
reply=SxExpr(reply_sx) if reply_sx else None,
)
# ---------------------------------------------------------------------------
# Post card data serializer
# ---------------------------------------------------------------------------
def _post_card_data(item: Any, actor: Any) -> dict:
"""Serialize a timeline item to a display-data dict for sx rendering."""
from quart import url_for
boosted_by = getattr(item, "boosted_by", None)
actor_icon = getattr(item, "actor_icon", None)
actor_name = getattr(item, "actor_name", "?")
actor_username = getattr(item, "actor_username", "")
actor_domain = getattr(item, "actor_domain", "")
oid = getattr(item, "object_id", "") or ""
safe_id = oid.replace("/", "_").replace(":", "_")
initial = actor_name[0].upper() if (not actor_icon and actor_name) else "?"
url = getattr(item, "url", None)
post_type = getattr(item, "post_type", "")
published = getattr(item, "published", None)
summary = getattr(item, "summary", None)
d: dict[str, Any] = dict(
boosted_by=str(escape(boosted_by)) if boosted_by else None,
actor_icon=actor_icon or None,
actor_name=str(escape(actor_name)),
actor_username=str(escape(actor_username)),
domain=f"@{escape(actor_domain)}" if actor_domain else "",
time=published.strftime("%b %d, %H:%M") if published else "",
content=getattr(item, "content", ""),
summary=str(escape(summary)) if summary else None,
original_url=url if url and post_type == "remote" else None,
object_id=oid,
author_inbox=getattr(item, "author_inbox", "") or "",
safe_id=safe_id,
initial=initial,
like_count=str(getattr(item, "like_count", 0) or 0),
boost_count=str(getattr(item, "boost_count", 0) or 0),
liked_by_me=bool(getattr(item, "liked_by_me", False)),
boosted_by_me=bool(getattr(item, "boosted_by_me", False)),
)
if actor and oid:
d["reply_url"] = url_for("social.defpage_compose_form", reply_to=oid)
return d
# ---------------------------------------------------------------------------
# Timeline items (pagination fragment)
# ---------------------------------------------------------------------------
def _timeline_items_sx(items: list, timeline_type: str, actor: Any,
actor_id: int | None = None) -> str:
"""Render timeline items with infinite scroll sentinel."""
from quart import url_for
from shared.browser.app.csrf import generate_csrf_token
item_dicts = [_post_card_data(item, actor) for item in items]
next_url = None
if items:
last = items[-1]
before = last.published.isoformat() if last.published else ""
if timeline_type == "actor" and actor_id is not None:
next_url = url_for("social.actor_timeline_page", id=actor_id, before=before)
else:
next_url = url_for(f"social.{timeline_type}_timeline_page", before=before)
kwargs: dict[str, Any] = dict(items=item_dicts, next_url=next_url)
if actor:
kwargs["has_actor"] = True
kwargs["csrf"] = generate_csrf_token()
kwargs["like_url"] = url_for("social.like")
kwargs["unlike_url"] = url_for("social.unlike")
kwargs["boost_url"] = url_for("social.boost")
kwargs["unboost_url"] = url_for("social.unboost")
return sx_call("federation-timeline-items-from-data", **kwargs)
# ---------------------------------------------------------------------------
# Search results (pagination fragment)
# ---------------------------------------------------------------------------
def _actor_card_data(a: Any, followed_urls: set,
*, list_type: str = "search") -> dict:
"""Serialize an actor to a display-data dict for sx rendering."""
from quart import url_for
display_name = getattr(a, "display_name", None) or getattr(a, "preferred_username", "")
username = getattr(a, "preferred_username", "")
domain = getattr(a, "domain", "")
icon_url = getattr(a, "icon_url", None)
actor_url = getattr(a, "actor_url", "")
summary = getattr(a, "summary", None)
aid = getattr(a, "id", None)
safe_id = actor_url.replace("/", "_").replace(":", "_")
initial = (display_name or username)[0].upper() if (not icon_url and (display_name or username)) else "?"
is_followed = actor_url in (followed_urls or set())
d: dict[str, Any] = dict(
display_name=str(escape(display_name)),
username=str(escape(username)),
domain=str(escape(domain)),
icon_url=icon_url or None,
actor_url=actor_url,
summary=summary,
safe_id=safe_id,
initial=initial,
is_followed=is_followed,
)
if (list_type in ("following", "search")) and aid:
d["name_href"] = url_for("social.defpage_actor_timeline", id=aid)
else:
d["name_href"] = f"https://{domain}/@{username}"
d["external_link"] = True
return d
def _actor_card_sx(a: Any, actor: Any, followed_urls: set,
*, list_type: str = "search") -> str:
"""Render a single actor card (used by follow/unfollow re-render)."""
from shared.browser.app.csrf import generate_csrf_token
from quart import url_for
d = _actor_card_data(a, followed_urls, list_type=list_type)
return sx_call(
"federation-actor-card-from-data",
d=d,
has_actor=actor is not None,
csrf=generate_csrf_token() if actor else None,
follow_url=url_for("social.follow") if actor else None,
unfollow_url=url_for("social.unfollow") if actor else None,
list_type=list_type,
)
def _search_results_sx(actors: list, query: str, page: int,
followed_urls: set, actor: Any) -> str:
"""Render search results with pagination sentinel."""
from quart import url_for
from shared.browser.app.csrf import generate_csrf_token
actor_dicts = [_actor_card_data(a, followed_urls, list_type="search") for a in actors]
next_url = url_for("social.search_page", q=query, page=page + 1) if len(actors) >= 20 else None
return sx_call(
"federation-actor-list-from-data",
actors=actor_dicts, next_url=next_url, list_type="search",
has_actor=actor is not None,
csrf=generate_csrf_token() if actor else None,
follow_url=url_for("social.follow") if actor else None,
unfollow_url=url_for("social.unfollow") if actor else None,
)
def _actor_list_items_sx(actors: list, page: int, list_type: str,
followed_urls: set, actor: Any) -> str:
"""Render actor list items (following/followers) with pagination sentinel."""
from quart import url_for
from shared.browser.app.csrf import generate_csrf_token
actor_dicts = [_actor_card_data(a, followed_urls, list_type=list_type) for a in actors]
next_url = url_for(f"social.{list_type}_list_page", page=page + 1) if len(actors) >= 20 else None
return sx_call(
"federation-actor-list-from-data",
actors=actor_dicts, next_url=next_url, list_type=list_type,
has_actor=actor is not None,
csrf=generate_csrf_token() if actor else None,
follow_url=url_for("social.follow") if actor else None,
unfollow_url=url_for("social.unfollow") if actor else None,
)
# ---------------------------------------------------------------------------
# Notification card
# ---------------------------------------------------------------------------
def _notification_sx(notif: Any) -> str:
"""Render a single notification."""
from_name = getattr(notif, "from_actor_name", "?")
from_username = getattr(notif, "from_actor_username", "")
from_domain = getattr(notif, "from_actor_domain", "")
from_icon = getattr(notif, "from_actor_icon", None)
ntype = getattr(notif, "notification_type", "")
preview = getattr(notif, "target_content_preview", None)
created = getattr(notif, "created_at", None)
read = getattr(notif, "read", True)
app_domain = getattr(notif, "app_domain", "")
border = " border-l-4 border-l-stone-400" if not read else ""
initial = from_name[0].upper() if (not from_icon and from_name) else "?"
avatar = sx_call(
"avatar", src=from_icon or None,
cls="w-8 h-8 rounded-full" if from_icon else "w-8 h-8 rounded-full bg-stone-300 flex items-center justify-center text-stone-600 font-bold text-xs",
initial=None if from_icon else initial,
)
domain_str = f"@{escape(from_domain)}" if from_domain else ""
type_map = {
"follow": "followed you",
"like": "liked your post",
"boost": "boosted your post",
"mention": "mentioned you",
"reply": "replied to your post",
}
action = type_map.get(ntype, "")
if ntype == "follow" and app_domain and app_domain != "federation":
action += f" on {escape(app_domain)}"
preview_sx = sx_call(
"federation-notification-preview", preview=str(escape(preview)),
) if preview else ""
time_str = created.strftime("%b %d, %H:%M") if created else ""
return sx_call(
"federation-notification-card",
cls=f"bg-white rounded-lg shadow-sm border border-stone-200 p-4{border}",
avatar=SxExpr(avatar),
from_name=str(escape(from_name)),
from_username=str(escape(from_username)),
from_domain=domain_str, action_text=action,
preview=SxExpr(preview_sx) if preview_sx else None,
time=time_str,
)
# ---------------------------------------------------------------------------
# Public API: Home page
# ---------------------------------------------------------------------------
async def render_federation_home(ctx: dict) -> str:
"""Full page: federation home (minimal)."""
hdr = root_header_sx(ctx)
return full_page_sx(ctx, header_rows=hdr)
# ---------------------------------------------------------------------------
# Public API: Login
# ---------------------------------------------------------------------------
async def render_login_page(ctx: dict) -> str:
"""Full page: federation login form."""
from shared.browser.app.csrf import generate_csrf_token
from quart import url_for
error = ctx.get("error", "")
email = ctx.get("email", "")
action = url_for("auth.start_login")
csrf = generate_csrf_token()
error_sx = sx_call("auth-error-banner", error=error) if error else ""
content = sx_call(
"auth-login-form",
error=SxExpr(error_sx) if error_sx else None,
action=action, csrf_token=csrf,
email=str(escape(email)),
)
return _social_page(ctx, None, content=content,
title="Login \u2014 Rose Ash")
async def render_check_email_page(ctx: dict) -> str:
"""Full page: check email after magic link sent."""
email = ctx.get("email", "")
email_error = ctx.get("email_error")
error_sx = sx_call(
"auth-check-email-error", error=str(escape(email_error)),
) if email_error else ""
content = sx_call(
"auth-check-email",
email=str(escape(email)),
error=SxExpr(error_sx) if error_sx else None,
)
return _social_page(ctx, None, content=content,
title="Check your email \u2014 Rose Ash")
# ---------------------------------------------------------------------------
# Content builders (used by defpage before_request)
# ---------------------------------------------------------------------------
def _timeline_content_sx(items: list, timeline_type: str, actor: Any) -> str:
"""Build timeline content SX string."""
from quart import url_for
label = "Home" if timeline_type == "home" else "Public"
compose_sx = ""
if actor:
compose_url = url_for("social.defpage_compose_form")
compose_sx = sx_call("federation-compose-button", url=compose_url)
timeline_sx = _timeline_items_sx(items, timeline_type, actor)
return sx_call(
"federation-timeline-page",
label=label,
compose=SxExpr(compose_sx) if compose_sx else None,
timeline=SxExpr(timeline_sx) if timeline_sx else None,
)
async def render_timeline_items(items: list, timeline_type: str,
actor: Any, actor_id: int | None = None) -> str:
"""Pagination fragment: timeline items."""
return _timeline_items_sx(items, timeline_type, actor, actor_id)
def _compose_content_sx(actor: Any, reply_to: str | None) -> str:
"""Build compose form content SX string."""
from shared.browser.app.csrf import generate_csrf_token
from quart import url_for
csrf = generate_csrf_token()
action = url_for("social.compose_submit")
reply_sx = ""
if reply_to:
reply_sx = sx_call(
"federation-compose-reply",
reply_to=str(escape(reply_to)),
)
return sx_call(
"federation-compose-form",
action=action, csrf=csrf,
reply=SxExpr(reply_sx) if reply_sx else None,
)
def _search_content_sx(query: str, actors: list, total: int,
page: int, followed_urls: set, actor: Any) -> str:
"""Build search page content SX string."""
from quart import url_for
search_url = url_for("social.defpage_search")
search_page_url = url_for("social.search_page")
results_sx = _search_results_sx(actors, query, page, followed_urls, actor)
info_sx = ""
if query and total:
s = "s" if total != 1 else ""
info_sx = sx_call(
"federation-search-info",
cls="text-sm text-stone-500 mb-4",
text=f"{total} result{s} for <strong>{escape(query)}</strong>",
)
elif query:
info_sx = sx_call(
"federation-search-info",
cls="text-stone-500 mb-4",
text=f"No results found for <strong>{escape(query)}</strong>",
)
return sx_call(
"federation-search-page",
search_url=search_url, search_page_url=search_page_url,
query=str(escape(query)),
info=SxExpr(info_sx) if info_sx else None,
results=SxExpr(results_sx) if results_sx else None,
)
async def render_search_results(actors: list, query: str, page: int,
followed_urls: set, actor: Any) -> str:
"""Pagination fragment: search results."""
return _search_results_sx(actors, query, page, followed_urls, actor)
def _following_content_sx(actors: list, total: int, actor: Any) -> str:
"""Build following list content SX string."""
items_sx = _actor_list_items_sx(actors, 1, "following", set(), actor)
return sx_call(
"federation-actor-list-page",
title="Following", count_str=f"({total})",
items=SxExpr(items_sx) if items_sx else None,
)
async def render_following_items(actors: list, page: int, actor: Any) -> str:
"""Pagination fragment: following items."""
return _actor_list_items_sx(actors, page, "following", set(), actor)
def _followers_content_sx(actors: list, total: int,
followed_urls: set, actor: Any) -> str:
"""Build followers list content SX string."""
items_sx = _actor_list_items_sx(actors, 1, "followers", followed_urls, actor)
return sx_call(
"federation-actor-list-page",
title="Followers", count_str=f"({total})",
items=SxExpr(items_sx) if items_sx else None,
)
async def render_followers_items(actors: list, page: int,
followed_urls: set, actor: Any) -> str:
"""Pagination fragment: followers items."""
return _actor_list_items_sx(actors, page, "followers", followed_urls, actor)
def _actor_timeline_content_sx(remote_actor: Any, items: list,
is_following: bool, actor: Any) -> str:
"""Build actor timeline content SX string."""
from shared.browser.app.csrf import generate_csrf_token
from quart import url_for
csrf = generate_csrf_token()
display_name = remote_actor.display_name or remote_actor.preferred_username
icon_url = getattr(remote_actor, "icon_url", None)
summary = getattr(remote_actor, "summary", None)
actor_url = getattr(remote_actor, "actor_url", "")
initial = display_name[0].upper() if (not icon_url and display_name) else "?"
avatar = sx_call(
"avatar", src=icon_url or None,
cls="w-16 h-16 rounded-full" if icon_url else "w-16 h-16 rounded-full bg-stone-300 flex items-center justify-center text-stone-600 font-bold text-xl",
initial=None if icon_url else initial,
)
summary_sx = sx_call("federation-profile-summary", summary=summary) if summary else ""
follow_sx = ""
if actor:
if is_following:
follow_sx = sx_call(
"federation-follow-form",
action=url_for("social.unfollow"), csrf=csrf, actor_url=actor_url,
label="Unfollow",
cls="border border-stone-300 rounded px-4 py-2 hover:bg-stone-100",
)
else:
follow_sx = sx_call(
"federation-follow-form",
action=url_for("social.follow"), csrf=csrf, actor_url=actor_url,
label="Follow",
cls="bg-stone-800 text-white rounded px-4 py-2 hover:bg-stone-700",
)
timeline_sx = _timeline_items_sx(items, "actor", actor, remote_actor.id)
header_sx = sx_call(
"federation-actor-profile-header",
avatar=SxExpr(avatar),
display_name=str(escape(display_name)),
username=str(escape(remote_actor.preferred_username)),
domain=str(escape(remote_actor.domain)),
summary=SxExpr(summary_sx) if summary_sx else None,
follow=SxExpr(follow_sx) if follow_sx else None,
)
return sx_call(
"federation-actor-timeline-layout",
header=SxExpr(header_sx),
timeline=SxExpr(timeline_sx) if timeline_sx else None,
)
async def render_actor_timeline_items(items: list, actor_id: int,
actor: Any) -> str:
"""Pagination fragment: actor timeline items."""
return _timeline_items_sx(items, "actor", actor, actor_id)
def _notifications_content_sx(notifications: list) -> str:
"""Build notifications content SX string."""
if not notifications:
notif_sx = sx_call("empty-state", message="No notifications yet.",
cls="text-stone-500")
else:
items_sx = "(<> " + " ".join(_notification_sx(n) for n in notifications) + ")"
notif_sx = sx_call(
"federation-notifications-list",
items=SxExpr(items_sx),
)
return sx_call("federation-notifications-page", notifs=SxExpr(notif_sx))
# ---------------------------------------------------------------------------
# Public API: Choose username
# ---------------------------------------------------------------------------
async def render_choose_username_page(ctx: dict) -> str:
"""Full page: choose username form."""
from shared.browser.app.csrf import generate_csrf_token
from quart import url_for
from shared.config import config
csrf = generate_csrf_token()
error = ctx.get("error", "")
username = ctx.get("username", "")
ap_domain = config().get("ap_domain", "rose-ash.com")
check_url = url_for("identity.check_username")
actor = ctx.get("actor")
error_sx = sx_call("auth-error-banner", error=error) if error else ""
content = sx_call(
"federation-choose-username",
domain=str(escape(ap_domain)),
error=SxExpr(error_sx) if error_sx else None,
csrf=csrf, username=str(escape(username)),
check_url=check_url,
)
return _social_page(ctx, actor, content=content,
title="Choose Username \u2014 Rose Ash")
# ---------------------------------------------------------------------------
# Public API: Actor profile
# ---------------------------------------------------------------------------
async def render_profile_page(ctx: dict, actor: Any, activities: list,
total: int) -> str:
"""Full page: actor profile."""
from shared.config import config
ap_domain = config().get("ap_domain", "rose-ash.com")
display_name = actor.display_name or actor.preferred_username
summary_sx = sx_call(
"federation-profile-summary-text", text=str(escape(actor.summary)),
) if actor.summary else ""
activity_dicts = [
dict(
activity_type=a.activity_type,
published=a.published.strftime("%Y-%m-%d %H:%M") if a.published else "",
object_type=a.object_type,
)
for a in activities
]
activities_sx = sx_call("federation-activities-from-data", activities=activity_dicts)
content = sx_call(
"federation-profile-page",
display_name=str(escape(display_name)),
username=str(escape(actor.preferred_username)),
domain=str(escape(ap_domain)),
summary=SxExpr(summary_sx) if summary_sx else None,
activities_heading=f"Activities ({total})",
activities=SxExpr(activities_sx),
)
return _social_page(ctx, actor, content=content,
title=f"@{actor.preferred_username} \u2014 Rose Ash")
# ---------------------------------------------------------------------------
# Public API: POST handler fragment renderers
# ---------------------------------------------------------------------------
def render_interaction_buttons(object_id: str, author_inbox: str,
like_count: int, boost_count: int,
liked_by_me: bool, boosted_by_me: bool,
actor: Any) -> str:
"""Render interaction buttons fragment for HTMX POST response."""
from types import SimpleNamespace
item = SimpleNamespace(
object_id=object_id,
author_inbox=author_inbox,
like_count=like_count,
boost_count=boost_count,
liked_by_me=liked_by_me,
boosted_by_me=boosted_by_me,
)
return _interaction_buttons_sx(item, actor)
def render_actor_card(actor_dto: Any, actor: Any, followed_urls: set,
*, list_type: str = "following") -> str:
"""Render a single actor card fragment for HTMX POST response."""
return _actor_card_sx(actor_dto, actor, followed_urls, list_type=list_type)