giles
c243d17eeb
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 3m41s
Replace Python GET page handlers with declarative defpage definitions in .sx files across all 8 apps (sx docs, orders, account, market, cart, federation, events, blog). Each app now has sxc/pages/ with setup functions, layout registrations, page helpers, and .sx defpage declarations. Core infrastructure: add g I/O primitive, PageDef support for auth/layout/ data/content/filter/aside/menu slots, post_author auth level, and custom layout registration. Remove ~1400 lines of render_*_page/render_*_oob boilerplate. Update all endpoint references in routes, sx_components, and templates to defpage_* naming. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
743 lines
28 KiB
Python
743 lines
28 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,
|
|
)
|
|
|
|
|
|
def _post_card_sx(item: Any, actor: Any) -> str:
|
|
"""Render a single timeline post card."""
|
|
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", "")
|
|
content = getattr(item, "content", "")
|
|
summary = getattr(item, "summary", None)
|
|
published = getattr(item, "published", None)
|
|
url = getattr(item, "url", None)
|
|
post_type = getattr(item, "post_type", "")
|
|
|
|
boost_sx = sx_call(
|
|
"federation-boost-label", name=str(escape(boosted_by)),
|
|
) if boosted_by else ""
|
|
|
|
initial = actor_name[0].upper() if (not actor_icon and actor_name) else "?"
|
|
avatar = sx_call(
|
|
"avatar", src=actor_icon or None,
|
|
cls="w-10 h-10 rounded-full" if actor_icon else "w-10 h-10 rounded-full bg-stone-300 flex items-center justify-center text-stone-600 font-bold text-sm",
|
|
initial=None if actor_icon else initial,
|
|
)
|
|
|
|
domain_str = f"@{escape(actor_domain)}" if actor_domain else ""
|
|
time_str = published.strftime("%b %d, %H:%M") if published else ""
|
|
|
|
if summary:
|
|
content_sx = sx_call(
|
|
"federation-content",
|
|
content=content, summary=str(escape(summary)),
|
|
)
|
|
else:
|
|
content_sx = sx_call("federation-content", content=content)
|
|
|
|
original_sx = ""
|
|
if url and post_type == "remote":
|
|
original_sx = sx_call("federation-original-link", url=url)
|
|
|
|
interactions_sx = ""
|
|
if actor:
|
|
oid = getattr(item, "object_id", "") or ""
|
|
safe_id = oid.replace("/", "_").replace(":", "_")
|
|
interactions_sx = f'(div :id {serialize(f"interactions-{safe_id}")} {_interaction_buttons_sx(item, actor)})'
|
|
|
|
return sx_call(
|
|
"federation-post-card",
|
|
boost=SxExpr(boost_sx) if boost_sx else None,
|
|
avatar=SxExpr(avatar),
|
|
actor_name=str(escape(actor_name)),
|
|
actor_username=str(escape(actor_username)),
|
|
domain=domain_str, time=time_str,
|
|
content=SxExpr(content_sx),
|
|
original=SxExpr(original_sx) if original_sx else None,
|
|
interactions=SxExpr(interactions_sx) if interactions_sx else None,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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
|
|
|
|
parts = [_post_card_sx(item, actor) for item in items]
|
|
|
|
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)
|
|
parts.append(sx_call("federation-scroll-sentinel", url=next_url))
|
|
|
|
return "(<> " + " ".join(parts) + ")" if parts else ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Search results (pagination fragment)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _actor_card_sx(a: Any, actor: Any, followed_urls: set,
|
|
*, list_type: str = "search") -> str:
|
|
"""Render a single actor card with follow/unfollow button."""
|
|
from shared.browser.app.csrf import generate_csrf_token
|
|
from quart import url_for
|
|
|
|
csrf = generate_csrf_token()
|
|
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 "?"
|
|
avatar = sx_call(
|
|
"avatar", src=icon_url or None,
|
|
cls="w-12 h-12 rounded-full" if icon_url else "w-12 h-12 rounded-full bg-stone-300 flex items-center justify-center text-stone-600 font-bold",
|
|
initial=None if icon_url else initial,
|
|
)
|
|
|
|
# Name link
|
|
if (list_type in ("following", "search")) and aid:
|
|
name_sx = sx_call(
|
|
"federation-actor-name-link",
|
|
href=url_for("social.defpage_actor_timeline", id=aid),
|
|
name=str(escape(display_name)),
|
|
)
|
|
else:
|
|
name_sx = sx_call(
|
|
"federation-actor-name-link-external",
|
|
href=f"https://{domain}/@{username}",
|
|
name=str(escape(display_name)),
|
|
)
|
|
|
|
summary_sx = sx_call("federation-actor-summary", summary=summary) if summary else ""
|
|
|
|
# Follow/unfollow button
|
|
button_sx = ""
|
|
if actor:
|
|
is_followed = actor_url in (followed_urls or set())
|
|
if list_type == "following" or is_followed:
|
|
button_sx = sx_call(
|
|
"federation-unfollow-button",
|
|
action=url_for("social.unfollow"), csrf=csrf, actor_url=actor_url,
|
|
)
|
|
else:
|
|
label = "Follow Back" if list_type == "followers" else "Follow"
|
|
button_sx = sx_call(
|
|
"federation-follow-button",
|
|
action=url_for("social.follow"), csrf=csrf, actor_url=actor_url, label=label,
|
|
)
|
|
|
|
return sx_call(
|
|
"federation-actor-card",
|
|
cls="bg-white rounded-lg shadow-sm border border-stone-200 p-4 mb-3 flex items-center gap-4",
|
|
id=f"actor-{safe_id}",
|
|
avatar=SxExpr(avatar),
|
|
name=SxExpr(name_sx),
|
|
username=str(escape(username)), domain=str(escape(domain)),
|
|
summary=SxExpr(summary_sx) if summary_sx else None,
|
|
button=SxExpr(button_sx) if button_sx else None,
|
|
)
|
|
|
|
|
|
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
|
|
|
|
parts = [_actor_card_sx(a, actor, followed_urls, list_type="search") for a in actors]
|
|
if len(actors) >= 20:
|
|
next_url = url_for("social.search_page", q=query, page=page + 1)
|
|
parts.append(sx_call("federation-scroll-sentinel", url=next_url))
|
|
return "(<> " + " ".join(parts) + ")" if parts else ""
|
|
|
|
|
|
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
|
|
|
|
parts = [_actor_card_sx(a, actor, followed_urls, list_type=list_type) for a in actors]
|
|
if len(actors) >= 20:
|
|
next_url = url_for(f"social.{list_type}_list_page", page=page + 1)
|
|
parts.append(sx_call("federation-scroll-sentinel", url=next_url))
|
|
return "(<> " + " ".join(parts) + ")" if parts else ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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 ""
|
|
|
|
activities_sx = ""
|
|
if activities:
|
|
parts = []
|
|
for a in activities:
|
|
published = a.published.strftime("%Y-%m-%d %H:%M") if a.published else ""
|
|
obj_type_sx = sx_call(
|
|
"federation-activity-obj-type", obj_type=a.object_type,
|
|
) if a.object_type else ""
|
|
parts.append(sx_call(
|
|
"federation-activity-card",
|
|
activity_type=a.activity_type, published=published,
|
|
obj_type=SxExpr(obj_type_sx) if obj_type_sx else None,
|
|
))
|
|
items_sx = "(<> " + " ".join(parts) + ")"
|
|
activities_sx = sx_call("federation-activities-list", items=SxExpr(items_sx))
|
|
else:
|
|
activities_sx = sx_call("federation-activities-empty")
|
|
|
|
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)
|