giles
dacb61b0ae
Phase 2 (Orders): - Checkout error/return renders moved directly into route handlers - Removed orphaned test_sx_helpers.py Phase 3 (Federation): - Auth pages use _render_social_auth_page() helper in routes - Choose-username render inlined into identity routes - Timeline/search/follow/interaction renders inlined into social routes using serializers imported from sxc.pages - Added _social_page() to sxc/pages/__init__.py for shared use - Home page renders inline in app.py Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
311 lines
12 KiB
Python
311 lines
12 KiB
Python
"""Federation defpage setup — registers layouts, page helpers, and loads .sx pages."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
|
|
def setup_federation_pages() -> None:
|
|
"""Register federation-specific layouts, page helpers, and load page definitions."""
|
|
_register_federation_layouts()
|
|
_register_federation_helpers()
|
|
_load_federation_page_files()
|
|
|
|
|
|
def _load_federation_page_files() -> None:
|
|
import os
|
|
from shared.sx.pages import load_page_dir
|
|
load_page_dir(os.path.dirname(__file__), "federation")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Layouts
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _register_federation_layouts() -> None:
|
|
from shared.sx.layouts import register_custom_layout
|
|
register_custom_layout("social", _social_full, _social_oob)
|
|
|
|
|
|
async def _social_full(ctx: dict, **kw: Any) -> str:
|
|
from shared.sx.helpers import root_header_sx, header_child_sx, render_to_sx
|
|
from shared.sx.parser import SxExpr
|
|
|
|
actor = ctx.get("actor")
|
|
actor_data = _serialize_actor(actor) if actor else None
|
|
nav = await render_to_sx("federation-social-nav", actor=actor_data)
|
|
social_hdr = await render_to_sx("federation-social-header", nav=SxExpr(nav))
|
|
root_hdr = await root_header_sx(ctx)
|
|
child = await header_child_sx(social_hdr)
|
|
return "(<> " + root_hdr + " " + child + ")"
|
|
|
|
|
|
async def _social_oob(ctx: dict, **kw: Any) -> str:
|
|
from shared.sx.helpers import root_header_sx, render_to_sx
|
|
from shared.sx.parser import SxExpr
|
|
|
|
actor = ctx.get("actor")
|
|
actor_data = _serialize_actor(actor) if actor else None
|
|
nav = await render_to_sx("federation-social-nav", actor=actor_data)
|
|
social_hdr = await render_to_sx("federation-social-header", nav=SxExpr(nav))
|
|
child_oob = await render_to_sx("oob-header-sx",
|
|
parent_id="root-header-child",
|
|
row=SxExpr(social_hdr))
|
|
root_hdr_oob = await root_header_sx(ctx, oob=True)
|
|
return "(<> " + child_oob + " " + root_hdr_oob + ")"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Page helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _register_federation_helpers() -> None:
|
|
from shared.sx.pages import register_page_helpers
|
|
|
|
register_page_helpers("federation", {
|
|
"home-timeline-content": _h_home_timeline_content,
|
|
"public-timeline-content": _h_public_timeline_content,
|
|
"compose-content": _h_compose_content,
|
|
"search-content": _h_search_content,
|
|
"following-content": _h_following_content,
|
|
"followers-content": _h_followers_content,
|
|
"actor-timeline-content": _h_actor_timeline_content,
|
|
"notifications-content": _h_notifications_content,
|
|
})
|
|
|
|
|
|
def _serialize_actor(actor) -> dict | None:
|
|
"""Serialize an actor profile to a dict for sx defcomps."""
|
|
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:
|
|
"""Serialize a timeline item DTO to a dict for sx defcomps."""
|
|
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:
|
|
"""Serialize a remote actor DTO to a dict for sx defcomps."""
|
|
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),
|
|
}
|
|
|
|
|
|
async def _social_page(ctx: dict, actor, *, content: str,
|
|
title: str = "Rose Ash", meta_html: str = "") -> str:
|
|
"""Build a full social page with social header."""
|
|
from shared.sx.helpers import render_to_sx, root_header_sx, header_child_sx, full_page_sx
|
|
from shared.sx.parser import SxExpr
|
|
from markupsafe import escape
|
|
|
|
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>')
|
|
|
|
|
|
def _get_actor():
|
|
"""Return current user's actor or None."""
|
|
from quart import g
|
|
return getattr(g, "_social_actor", None)
|
|
|
|
|
|
def _require_actor():
|
|
"""Return current user's actor or abort 403."""
|
|
from quart import abort
|
|
actor = _get_actor()
|
|
if not actor:
|
|
abort(403, "You need to choose a federation username first")
|
|
return actor
|
|
|
|
|
|
async def _h_home_timeline_content(**kw):
|
|
from quart import g
|
|
from shared.services.registry import services
|
|
from shared.sx.helpers import render_to_sx
|
|
actor = _require_actor()
|
|
items = await services.federation.get_home_timeline(g.s, actor.id)
|
|
return await render_to_sx("federation-timeline-content",
|
|
items=[_serialize_timeline_item(i) for i in items],
|
|
timeline_type="home",
|
|
actor=_serialize_actor(actor))
|
|
|
|
|
|
async def _h_public_timeline_content(**kw):
|
|
from quart import g
|
|
from shared.services.registry import services
|
|
from shared.sx.helpers import render_to_sx
|
|
actor = _get_actor()
|
|
items = await services.federation.get_public_timeline(g.s)
|
|
return await render_to_sx("federation-timeline-content",
|
|
items=[_serialize_timeline_item(i) for i in items],
|
|
timeline_type="public",
|
|
actor=_serialize_actor(actor))
|
|
|
|
|
|
async def _h_compose_content(**kw):
|
|
from quart import request
|
|
from shared.sx.helpers import render_to_sx
|
|
_require_actor()
|
|
reply_to = request.args.get("reply_to")
|
|
return await render_to_sx("federation-compose-content",
|
|
reply_to=reply_to or None)
|
|
|
|
|
|
async def _h_search_content(**kw):
|
|
from quart import g, request
|
|
from shared.services.registry import services
|
|
from shared.sx.helpers import render_to_sx
|
|
actor = _get_actor()
|
|
query = request.args.get("q", "").strip()
|
|
actors_list = []
|
|
total = 0
|
|
followed_urls: set[str] = set()
|
|
if query:
|
|
actors_list, total = await services.federation.search_actors(g.s, query)
|
|
if actor:
|
|
following, _ = await services.federation.get_following(
|
|
g.s, actor.preferred_username, page=1, per_page=1000,
|
|
)
|
|
followed_urls = {a.actor_url for a in following}
|
|
return await render_to_sx("federation-search-content",
|
|
query=query,
|
|
actors=[_serialize_remote_actor(a) for a in actors_list],
|
|
total=total,
|
|
followed_urls=list(followed_urls),
|
|
actor=_serialize_actor(actor))
|
|
|
|
|
|
async def _h_following_content(**kw):
|
|
from quart import g
|
|
from shared.services.registry import services
|
|
from shared.sx.helpers import render_to_sx
|
|
actor = _require_actor()
|
|
actors_list, total = await services.federation.get_following(
|
|
g.s, actor.preferred_username,
|
|
)
|
|
return await render_to_sx("federation-following-content",
|
|
actors=[_serialize_remote_actor(a) for a in actors_list],
|
|
total=total,
|
|
actor=_serialize_actor(actor))
|
|
|
|
|
|
async def _h_followers_content(**kw):
|
|
from quart import g
|
|
from shared.services.registry import services
|
|
from shared.sx.helpers import render_to_sx
|
|
actor = _require_actor()
|
|
actors_list, total = await services.federation.get_followers_paginated(
|
|
g.s, actor.preferred_username,
|
|
)
|
|
following, _ = await services.federation.get_following(
|
|
g.s, actor.preferred_username, page=1, per_page=1000,
|
|
)
|
|
followed_urls = {a.actor_url for a in following}
|
|
return await render_to_sx("federation-followers-content",
|
|
actors=[_serialize_remote_actor(a) for a in actors_list],
|
|
total=total,
|
|
followed_urls=list(followed_urls),
|
|
actor=_serialize_actor(actor))
|
|
|
|
|
|
async def _h_actor_timeline_content(id=None, **kw):
|
|
from quart import g, abort
|
|
from shared.services.registry import services
|
|
from shared.sx.helpers import render_to_sx
|
|
actor = _get_actor()
|
|
actor_id = id
|
|
from shared.models.federation import RemoteActor
|
|
from sqlalchemy import select as sa_select
|
|
remote = (
|
|
await g.s.execute(
|
|
sa_select(RemoteActor).where(RemoteActor.id == actor_id)
|
|
)
|
|
).scalar_one_or_none()
|
|
if not remote:
|
|
abort(404)
|
|
from shared.services.federation_impl import _remote_actor_to_dto
|
|
remote_dto = _remote_actor_to_dto(remote)
|
|
items = await services.federation.get_actor_timeline(g.s, actor_id)
|
|
is_following = False
|
|
if actor:
|
|
from shared.models.federation import APFollowing
|
|
existing = (
|
|
await g.s.execute(
|
|
sa_select(APFollowing).where(
|
|
APFollowing.actor_profile_id == actor.id,
|
|
APFollowing.remote_actor_id == actor_id,
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
is_following = existing is not None
|
|
return await render_to_sx("federation-actor-timeline-content",
|
|
remote_actor=_serialize_remote_actor(remote_dto),
|
|
items=[_serialize_timeline_item(i) for i in items],
|
|
is_following=is_following,
|
|
actor=_serialize_actor(actor))
|
|
|
|
|
|
async def _h_notifications_content(**kw):
|
|
from quart import g
|
|
from shared.services.registry import services
|
|
from shared.sx.helpers import render_to_sx
|
|
actor = _require_actor()
|
|
items = await services.federation.get_notifications(g.s, actor.id)
|
|
await services.federation.mark_notifications_read(g.s, actor.id)
|
|
|
|
notif_dicts = []
|
|
for n in items:
|
|
created = getattr(n, "created_at", None)
|
|
notif_dicts.append({
|
|
"from_actor_name": getattr(n, "from_actor_name", "?"),
|
|
"from_actor_username": getattr(n, "from_actor_username", ""),
|
|
"from_actor_domain": getattr(n, "from_actor_domain", ""),
|
|
"from_actor_icon": getattr(n, "from_actor_icon", None),
|
|
"notification_type": getattr(n, "notification_type", ""),
|
|
"target_content_preview": getattr(n, "target_content_preview", None),
|
|
"created_at_formatted": created.strftime("%b %d, %H:%M") if created else "",
|
|
"read": getattr(n, "read", True),
|
|
"app_domain": getattr(n, "app_domain", ""),
|
|
})
|
|
return await render_to_sx("federation-notifications-content",
|
|
notifications=notif_dicts)
|