giles
e085fe43b4
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 2m6s
Python no longer generates s-expression strings. All SX rendering now goes through render_to_sx() which builds AST from native Python values and evaluates via async_eval_to_sx() — no SX string literals in Python. - Add render_to_sx()/render_to_html() infrastructure in shared/sx/helpers.py - Add (abort status msg) IO primitive in shared/sx/primitives_io.py - Convert all 9 services: ~650 sx_call() invocations replaced - Convert shared helpers (root_header_sx, full_page_sx, etc.) to async - Fix likes service import bug (likes.models → models) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
294 lines
12 KiB
Python
294 lines
12 KiB
Python
"""
|
|
Federation service s-expression page components.
|
|
|
|
Page helpers now call assembled defcomps in .sx files. This file contains
|
|
only functions still called directly from route handlers: full-page renders
|
|
(login, choose-username, profile) and POST fragment renderers (interaction
|
|
buttons, actor cards, pagination items).
|
|
"""
|
|
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.helpers import (
|
|
render_to_sx,
|
|
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")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Serialization helpers (shared with pages/__init__.py)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _serialize_actor(actor) -> dict | None:
|
|
if not actor:
|
|
return None
|
|
return {
|
|
"id": actor.id,
|
|
"preferred_username": actor.preferred_username,
|
|
"display_name": getattr(actor, "display_name", None),
|
|
"icon_url": getattr(actor, "icon_url", None),
|
|
"summary": getattr(actor, "summary", None),
|
|
"actor_url": getattr(actor, "actor_url", ""),
|
|
"domain": getattr(actor, "domain", ""),
|
|
}
|
|
|
|
|
|
def _serialize_timeline_item(item) -> dict:
|
|
published = getattr(item, "published", None)
|
|
return {
|
|
"object_id": getattr(item, "object_id", "") or "",
|
|
"author_inbox": getattr(item, "author_inbox", "") or "",
|
|
"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", ""),
|
|
"content": getattr(item, "content", ""),
|
|
"summary": getattr(item, "summary", None),
|
|
"published": published.strftime("%b %d, %H:%M") if published else "",
|
|
"before_cursor": published.isoformat() if published else "",
|
|
"url": getattr(item, "url", None),
|
|
"post_type": getattr(item, "post_type", ""),
|
|
"boosted_by": getattr(item, "boosted_by", None),
|
|
"like_count": getattr(item, "like_count", 0) or 0,
|
|
"boost_count": getattr(item, "boost_count", 0) or 0,
|
|
"liked_by_me": getattr(item, "liked_by_me", False),
|
|
"boosted_by_me": getattr(item, "boosted_by_me", False),
|
|
}
|
|
|
|
|
|
def _serialize_remote_actor(a) -> dict:
|
|
return {
|
|
"id": getattr(a, "id", None),
|
|
"display_name": getattr(a, "display_name", None) or getattr(a, "preferred_username", ""),
|
|
"preferred_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),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Social page shell
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def _social_page(ctx: dict, actor: Any, *, content: str,
|
|
title: str = "Rose Ash", meta_html: str = "") -> str:
|
|
from shared.sx.parser import SxExpr
|
|
actor_data = _serialize_actor(actor)
|
|
nav = await render_to_sx("federation-social-nav", actor=actor_data)
|
|
social_hdr = await render_to_sx("federation-social-header", nav=SxExpr(nav))
|
|
hdr = await root_header_sx(ctx)
|
|
child = await header_child_sx(social_hdr)
|
|
header_rows = "(<> " + hdr + " " + child + ")"
|
|
return await full_page_sx(ctx, header_rows=header_rows, content=content,
|
|
meta_html=meta_html or f'<title>{escape(title)}</title>')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Public API: Full page renders
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def render_federation_home(ctx: dict) -> str:
|
|
hdr = await root_header_sx(ctx)
|
|
return await full_page_sx(ctx, header_rows=hdr)
|
|
|
|
|
|
async def render_login_page(ctx: dict) -> str:
|
|
error = ctx.get("error", "")
|
|
email = ctx.get("email", "")
|
|
content = await render_to_sx("account-login-content",
|
|
error=error or None, email=str(escape(email)))
|
|
return await _social_page(ctx, None, content=content, title="Login \u2014 Rose Ash")
|
|
|
|
|
|
async def render_check_email_page(ctx: dict) -> str:
|
|
email = ctx.get("email", "")
|
|
email_error = ctx.get("email_error")
|
|
content = await render_to_sx("account-check-email-content",
|
|
email=str(escape(email)), email_error=email_error)
|
|
return await _social_page(ctx, None, content=content,
|
|
title="Check your email \u2014 Rose Ash")
|
|
|
|
|
|
async def render_choose_username_page(ctx: dict) -> str:
|
|
from shared.browser.app.csrf import generate_csrf_token
|
|
from quart import url_for
|
|
from shared.config import config
|
|
from shared.sx.parser import SxExpr
|
|
|
|
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 = await render_to_sx("auth-error-banner", error=error) if error else ""
|
|
content = await render_to_sx(
|
|
"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 await _social_page(ctx, actor, content=content,
|
|
title="Choose Username \u2014 Rose Ash")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Public API: Pagination fragment renderers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def render_timeline_items(items: list, timeline_type: str,
|
|
actor: Any, actor_id: int | None = None) -> str:
|
|
from quart import url_for
|
|
item_dicts = [_serialize_timeline_item(i) for i in items]
|
|
actor_data = _serialize_actor(actor)
|
|
|
|
# Build next URL
|
|
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)
|
|
|
|
return await render_to_sx("federation-timeline-items",
|
|
items=item_dicts,
|
|
timeline_type=timeline_type,
|
|
actor=actor_data,
|
|
next_url=next_url)
|
|
|
|
|
|
async def render_search_results(actors: list, query: str, page: int,
|
|
followed_urls: set, actor: Any) -> str:
|
|
from quart import url_for
|
|
actor_dicts = [_serialize_remote_actor(a) for a in actors]
|
|
actor_data = _serialize_actor(actor)
|
|
parts = []
|
|
for ad in actor_dicts:
|
|
parts.append(await render_to_sx("federation-actor-card-from-data",
|
|
a=ad,
|
|
actor=actor_data,
|
|
followed_urls=list(followed_urls),
|
|
list_type="search"))
|
|
if len(actors) >= 20:
|
|
next_url = url_for("social.search_page", q=query, page=page + 1)
|
|
parts.append(await render_to_sx("federation-scroll-sentinel", url=next_url))
|
|
return "(<> " + " ".join(parts) + ")" if parts else ""
|
|
|
|
|
|
async def render_following_items(actors: list, page: int, actor: Any) -> str:
|
|
from quart import url_for
|
|
actor_dicts = [_serialize_remote_actor(a) for a in actors]
|
|
actor_data = _serialize_actor(actor)
|
|
parts = []
|
|
for ad in actor_dicts:
|
|
parts.append(await render_to_sx("federation-actor-card-from-data",
|
|
a=ad,
|
|
actor=actor_data,
|
|
followed_urls=[],
|
|
list_type="following"))
|
|
if len(actors) >= 20:
|
|
next_url = url_for("social.following_list_page", page=page + 1)
|
|
parts.append(await render_to_sx("federation-scroll-sentinel", url=next_url))
|
|
return "(<> " + " ".join(parts) + ")" if parts else ""
|
|
|
|
|
|
async def render_followers_items(actors: list, page: int,
|
|
followed_urls: set, actor: Any) -> str:
|
|
from quart import url_for
|
|
actor_dicts = [_serialize_remote_actor(a) for a in actors]
|
|
actor_data = _serialize_actor(actor)
|
|
parts = []
|
|
for ad in actor_dicts:
|
|
parts.append(await render_to_sx("federation-actor-card-from-data",
|
|
a=ad,
|
|
actor=actor_data,
|
|
followed_urls=list(followed_urls),
|
|
list_type="followers"))
|
|
if len(actors) >= 20:
|
|
next_url = url_for("social.followers_list_page", page=page + 1)
|
|
parts.append(await render_to_sx("federation-scroll-sentinel", url=next_url))
|
|
return "(<> " + " ".join(parts) + ")" if parts else ""
|
|
|
|
|
|
async def render_actor_timeline_items(items: list, actor_id: int,
|
|
actor: Any) -> str:
|
|
return await render_timeline_items(items, "actor", actor, actor_id)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Public API: POST handler fragment renderers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async 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 POST response."""
|
|
from shared.browser.app.csrf import generate_csrf_token
|
|
from quart import url_for
|
|
from shared.sx.parser import SxExpr
|
|
|
|
csrf = generate_csrf_token()
|
|
safe_id = object_id.replace("/", "_").replace(":", "_")
|
|
target = f"#interactions-{safe_id}"
|
|
|
|
if liked_by_me:
|
|
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_by_me:
|
|
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=object_id) if object_id else ""
|
|
reply_sx = await render_to_sx("federation-reply-link", url=reply_url) if reply_url else ""
|
|
|
|
like_form = await render_to_sx("federation-like-form",
|
|
action=like_action, target=target, oid=object_id, ainbox=author_inbox,
|
|
csrf=csrf, cls=f"flex items-center gap-1 {like_cls}",
|
|
icon=like_icon, count=str(like_count))
|
|
|
|
boost_form = await render_to_sx("federation-boost-form",
|
|
action=boost_action, target=target, oid=object_id, ainbox=author_inbox,
|
|
csrf=csrf, cls=f"flex items-center gap-1 {boost_cls}",
|
|
count=str(boost_count))
|
|
|
|
return await render_to_sx("federation-interaction-buttons",
|
|
like=SxExpr(like_form),
|
|
boost=SxExpr(boost_form),
|
|
reply=SxExpr(reply_sx) if reply_sx else None)
|
|
|
|
|
|
async def render_actor_card(actor_dto: Any, actor: Any, followed_urls: set,
|
|
*, list_type: str = "following") -> str:
|
|
"""Render a single actor card fragment for POST response."""
|
|
actor_data = _serialize_actor(actor)
|
|
ad = _serialize_remote_actor(actor_dto)
|
|
return await render_to_sx("federation-actor-card-from-data",
|
|
a=ad,
|
|
actor=actor_data,
|
|
followed_urls=list(followed_urls),
|
|
list_type=list_type)
|