""" 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'{escape(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)