rose-ash/federation/sxc/pages/__init__.py

"""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)


def _social_full(ctx: dict, **kw: Any) -> str:
    from shared.sx.helpers import root_header_sx, header_child_sx
    from sx.sx_components import _social_header_sx

    actor = ctx.get("actor")
    root_hdr = root_header_sx(ctx)
    social_hdr = _social_header_sx(actor)
    child = header_child_sx(social_hdr)
    return "(<> " + root_hdr + " " + child + ")"


def _social_oob(ctx: dict, **kw: Any) -> str:
    from shared.sx.helpers import root_header_sx, sx_call, SxExpr
    from sx.sx_components import _social_header_sx

    actor = ctx.get("actor")
    social_hdr = _social_header_sx(actor)
    child_oob = sx_call("oob-header-sx",
                         parent_id="root-header-child",
                         row=SxExpr(social_hdr))
    root_hdr_oob = 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 _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
    actor = _require_actor()
    items = await services.federation.get_home_timeline(g.s, actor.id)
    from sx.sx_components import _timeline_content_sx
    return _timeline_content_sx(items, "home", actor)


async def _h_public_timeline_content(**kw):
    from quart import g
    from shared.services.registry import services
    actor = _get_actor()
    items = await services.federation.get_public_timeline(g.s)
    from sx.sx_components import _timeline_content_sx
    return _timeline_content_sx(items, "public", actor)


async def _h_compose_content(**kw):
    from quart import request
    actor = _require_actor()
    from sx.sx_components import _compose_content_sx
    reply_to = request.args.get("reply_to")
    return _compose_content_sx(actor, reply_to)


async def _h_search_content(**kw):
    from quart import g, request
    from shared.services.registry import services
    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}
    from sx.sx_components import _search_content_sx
    return _search_content_sx(query, actors_list, total, 1, followed_urls, actor)


async def _h_following_content(**kw):
    from quart import g
    from shared.services.registry import services
    actor = _require_actor()
    actors_list, total = await services.federation.get_following(
        g.s, actor.preferred_username,
    )
    from sx.sx_components import _following_content_sx
    return _following_content_sx(actors_list, total, actor)


async def _h_followers_content(**kw):
    from quart import g
    from shared.services.registry import services
    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}
    from sx.sx_components import _followers_content_sx
    return _followers_content_sx(actors_list, total, followed_urls, actor)


async def _h_actor_timeline_content(id=None, **kw):
    from quart import g, abort
    from shared.services.registry import services
    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
    from sx.sx_components import _actor_timeline_content_sx
    return _actor_timeline_content_sx(remote_dto, items, is_following, actor)


async def _h_notifications_content(**kw):
    from quart import g
    from shared.services.registry import services
    actor = _require_actor()
    items = await services.federation.get_notifications(g.s, actor.id)
    await services.federation.mark_notifications_read(g.s, actor.id)
    from sx.sx_components import _notifications_content_sx
    return _notifications_content_sx(items)