rose-ash/shared/infrastructure/ap_social.py

"""Per-app AP social blueprint: search, follow/unfollow, followers, following, actor timeline.

Lightweight social UI for blog/market/events. Federation keeps the full
social hub (timeline, compose, notifications, interactions).
"""
from __future__ import annotations

import logging
from datetime import datetime

from quart import Blueprint, request, g, redirect, url_for, abort, render_template, Response

from shared.services.registry import services

log = logging.getLogger(__name__)


def create_ap_social_blueprint(app_name: str) -> Blueprint:
    """Create a per-app social blueprint scoped to *app_name*."""
    bp = Blueprint("ap_social", __name__, url_prefix="/social")

    @bp.before_request
    async def load_actor():
        if g.get("user"):
            actor = await services.federation.get_actor_by_user_id(g.s, g.user.id)
            g._social_actor = actor

    def _require_actor():
        actor = getattr(g, "_social_actor", None)
        if not actor:
            abort(403, "You need to choose a federation username first")
        return actor

    # -- Index ----------------------------------------------------------------

    @bp.get("/")
    async def index():
        actor = getattr(g, "_social_actor", None)
        return await render_template(
            "social/index.html",
            actor=actor,
        )

    # -- Search ---------------------------------------------------------------

    @bp.get("/search")
    async def search():
        actor = getattr(g, "_social_actor", None)
        query = request.args.get("q", "").strip()
        actors = []
        total = 0
        followed_urls: set[str] = set()
        if query:
            actors, 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_template(
            "social/search.html",
            query=query,
            actors=actors,
            total=total,
            page=1,
            followed_urls=followed_urls,
            actor=actor,
        )

    @bp.get("/search/page")
    async def search_page():
        actor = getattr(g, "_social_actor", None)
        query = request.args.get("q", "").strip()
        page = request.args.get("page", 1, type=int)
        actors = []
        total = 0
        followed_urls: set[str] = set()
        if query:
            actors, total = await services.federation.search_actors(
                g.s, query, page=page,
            )
            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_template(
            "social/_search_results.html",
            actors=actors,
            total=total,
            page=page,
            query=query,
            followed_urls=followed_urls,
            actor=actor,
        )

    # -- Follow / Unfollow ----------------------------------------------------

    @bp.post("/follow")
    async def follow():
        actor = _require_actor()
        form = await request.form
        remote_actor_url = form.get("actor_url", "")
        if remote_actor_url:
            await services.federation.send_follow(
                g.s, actor.preferred_username, remote_actor_url,
            )
        if request.headers.get("HX-Request"):
            return await _actor_card_response(actor, remote_actor_url, is_followed=True)
        return redirect(request.referrer or url_for("ap_social.search"))

    @bp.post("/unfollow")
    async def unfollow():
        actor = _require_actor()
        form = await request.form
        remote_actor_url = form.get("actor_url", "")
        if remote_actor_url:
            await services.federation.unfollow(
                g.s, actor.preferred_username, remote_actor_url,
            )
        if request.headers.get("HX-Request"):
            return await _actor_card_response(actor, remote_actor_url, is_followed=False)
        return redirect(request.referrer or url_for("ap_social.search"))

    async def _actor_card_response(actor, remote_actor_url, is_followed):
        """Re-render a single actor card after follow/unfollow via HTMX."""
        remote_dto = await services.federation.get_or_fetch_remote_actor(
            g.s, remote_actor_url,
        )
        if not remote_dto:
            return Response("", status=200)
        followed_urls = {remote_actor_url} if is_followed else set()
        referer = request.referrer or ""
        if "/followers" in referer:
            list_type = "followers"
        else:
            list_type = "following"
        return await render_template(
            "social/_actor_list_items.html",
            actors=[remote_dto],
            total=0,
            page=1,
            list_type=list_type,
            followed_urls=followed_urls,
            actor=actor,
        )

    # -- Followers ------------------------------------------------------------

    @bp.get("/followers")
    async def followers_list():
        actor = _require_actor()
        actors, total = await services.federation.get_followers_paginated(
            g.s, actor.preferred_username, app_domain=app_name,
        )
        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_template(
            "social/followers.html",
            actors=actors,
            total=total,
            page=1,
            followed_urls=followed_urls,
            actor=actor,
        )

    @bp.get("/followers/page")
    async def followers_list_page():
        actor = _require_actor()
        page = request.args.get("page", 1, type=int)
        actors, total = await services.federation.get_followers_paginated(
            g.s, actor.preferred_username, page=page, app_domain=app_name,
        )
        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_template(
            "social/_actor_list_items.html",
            actors=actors,
            total=total,
            page=page,
            list_type="followers",
            followed_urls=followed_urls,
            actor=actor,
        )

    # -- Following ------------------------------------------------------------

    @bp.get("/following")
    async def following_list():
        actor = _require_actor()
        actors, total = await services.federation.get_following(
            g.s, actor.preferred_username,
        )
        return await render_template(
            "social/following.html",
            actors=actors,
            total=total,
            page=1,
            actor=actor,
        )

    @bp.get("/following/page")
    async def following_list_page():
        actor = _require_actor()
        page = request.args.get("page", 1, type=int)
        actors, total = await services.federation.get_following(
            g.s, actor.preferred_username, page=page,
        )
        return await render_template(
            "social/_actor_list_items.html",
            actors=actors,
            total=total,
            page=page,
            list_type="following",
            followed_urls=set(),
            actor=actor,
        )

    # -- Actor timeline -------------------------------------------------------

    @bp.get("/actor/<int:id>")
    async def actor_timeline(id: int):
        actor = getattr(g, "_social_actor", None)
        from shared.models.federation import RemoteActor
        from sqlalchemy import select as sa_select
        remote = (
            await g.s.execute(
                sa_select(RemoteActor).where(RemoteActor.id == 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, 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 == id,
                    )
                )
            ).scalar_one_or_none()
            is_following = existing is not None
        return await render_template(
            "social/actor_timeline.html",
            remote_actor=remote_dto,
            items=items,
            is_following=is_following,
            actor=actor,
        )

    @bp.get("/actor/<int:id>/timeline")
    async def actor_timeline_page(id: int):
        actor = getattr(g, "_social_actor", None)
        before_str = request.args.get("before")
        before = None
        if before_str:
            try:
                before = datetime.fromisoformat(before_str)
            except ValueError:
                pass
        items = await services.federation.get_actor_timeline(
            g.s, id, before=before,
        )
        return await render_template(
            "social/_timeline_items.html",
            items=items,
            timeline_type="actor",
            actor_id=id,
            actor=actor,
        )

    return bp