giles
e085fe43b4
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 2m6s
Python no longer generates s-expression strings. All SX rendering now goes through render_to_sx() which builds AST from native Python values and evaluates via async_eval_to_sx() — no SX string literals in Python. - Add render_to_sx()/render_to_html() infrastructure in shared/sx/helpers.py - Add (abort status msg) IO primitive in shared/sx/primitives_io.py - Convert all 9 services: ~650 sx_call() invocations replaced - Convert shared helpers (root_header_sx, full_page_sx, etc.) to async - Fix likes service import bug (likes.models → models) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
311 lines
12 KiB
Python
311 lines
12 KiB
Python
"""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).
|
|
|
|
All rendering uses s-expressions (no Jinja templates).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
from quart import Blueprint, request, g, redirect, url_for, abort, 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")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Federation session — AP tables live in db_federation.
|
|
# ------------------------------------------------------------------
|
|
from shared.db.session import needs_federation_session, create_federation_session
|
|
|
|
@bp.before_request
|
|
async def _open_ap_session():
|
|
if needs_federation_session():
|
|
sess = create_federation_session()
|
|
g._ap_s = sess
|
|
g._ap_tx = await sess.begin()
|
|
g._ap_own = True
|
|
else:
|
|
g._ap_s = g.s
|
|
g._ap_own = False
|
|
|
|
@bp.after_request
|
|
async def _commit_ap_session(response):
|
|
if getattr(g, "_ap_own", False):
|
|
if 200 <= response.status_code < 400:
|
|
try:
|
|
await g._ap_tx.commit()
|
|
except Exception:
|
|
try:
|
|
await g._ap_tx.rollback()
|
|
except Exception:
|
|
pass
|
|
return response
|
|
|
|
@bp.teardown_request
|
|
async def _close_ap_session(exc):
|
|
if getattr(g, "_ap_own", False):
|
|
s = getattr(g, "_ap_s", None)
|
|
if s:
|
|
if exc is not None or s.in_transaction():
|
|
tx = getattr(g, "_ap_tx", None)
|
|
if tx and tx.is_active:
|
|
try:
|
|
await tx.rollback()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
await s.close()
|
|
except Exception:
|
|
pass
|
|
|
|
@bp.before_request
|
|
async def load_actor():
|
|
if g.get("user"):
|
|
actor = await services.federation.get_actor_by_user_id(g._ap_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
|
|
|
|
async def _render_social_page(content: str, actor=None, title: str = "Social"):
|
|
"""Render a full social page or OOB response depending on request type."""
|
|
from shared.browser.app.utils.htmx import is_htmx_request
|
|
from shared.sx.page import get_template_context
|
|
from shared.sx.helpers import full_page_sx, oob_page_sx, sx_response
|
|
from shared.infrastructure.ap_social_sx import (
|
|
_social_full_headers, _social_oob_headers,
|
|
)
|
|
|
|
tctx = await get_template_context()
|
|
kw = {"actor": actor}
|
|
|
|
if is_htmx_request():
|
|
oob_headers = await _social_oob_headers(tctx, **kw)
|
|
return sx_response(await oob_page_sx(
|
|
oobs=oob_headers,
|
|
content=content,
|
|
))
|
|
else:
|
|
header_rows = await _social_full_headers(tctx, **kw)
|
|
return await full_page_sx(tctx, header_rows=header_rows, content=content)
|
|
|
|
# -- Index ----------------------------------------------------------------
|
|
|
|
@bp.get("/")
|
|
async def index():
|
|
actor = getattr(g, "_social_actor", None)
|
|
from shared.infrastructure.ap_social_sx import social_index_content_sx
|
|
content = social_index_content_sx(actor)
|
|
return await _render_social_page(content, actor, title="Social")
|
|
|
|
# -- 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._ap_s, query)
|
|
if actor:
|
|
following, _ = await services.federation.get_following(
|
|
g._ap_s, actor.preferred_username, page=1, per_page=1000,
|
|
)
|
|
followed_urls = {a.actor_url for a in following}
|
|
from shared.infrastructure.ap_social_sx import social_search_content_sx
|
|
content = social_search_content_sx(query, actors, total, 1, followed_urls, actor)
|
|
return await _render_social_page(content, actor, title="Search")
|
|
|
|
@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._ap_s, query, page=page,
|
|
)
|
|
if actor:
|
|
following, _ = await services.federation.get_following(
|
|
g._ap_s, actor.preferred_username, page=1, per_page=1000,
|
|
)
|
|
followed_urls = {a.actor_url for a in following}
|
|
from shared.infrastructure.ap_social_sx import search_results_sx
|
|
from shared.sx.helpers import sx_response
|
|
content = search_results_sx(actors, total, page, query, followed_urls, actor)
|
|
return sx_response(content)
|
|
|
|
# -- 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._ap_s, actor.preferred_username, remote_actor_url,
|
|
)
|
|
if request.headers.get("SX-Request") or 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._ap_s, actor.preferred_username, remote_actor_url,
|
|
)
|
|
if request.headers.get("SX-Request") or 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."""
|
|
remote_dto = await services.federation.get_or_fetch_remote_actor(
|
|
g._ap_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"
|
|
from shared.infrastructure.ap_social_sx import actor_list_items_sx
|
|
from shared.sx.helpers import sx_response
|
|
content = actor_list_items_sx(
|
|
[remote_dto], 0, 1, list_type, followed_urls, actor,
|
|
)
|
|
return sx_response(content)
|
|
|
|
# -- Followers ------------------------------------------------------------
|
|
|
|
@bp.get("/followers")
|
|
async def followers_list():
|
|
actor = _require_actor()
|
|
actors, total = await services.federation.get_followers_paginated(
|
|
g._ap_s, actor.preferred_username, app_domain=app_name,
|
|
)
|
|
following, _ = await services.federation.get_following(
|
|
g._ap_s, actor.preferred_username, page=1, per_page=1000,
|
|
)
|
|
followed_urls = {a.actor_url for a in following}
|
|
from shared.infrastructure.ap_social_sx import social_followers_content_sx
|
|
content = social_followers_content_sx(actors, total, 1, followed_urls, actor)
|
|
return await _render_social_page(content, actor, title="Followers")
|
|
|
|
@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._ap_s, actor.preferred_username, page=page, app_domain=app_name,
|
|
)
|
|
following, _ = await services.federation.get_following(
|
|
g._ap_s, actor.preferred_username, page=1, per_page=1000,
|
|
)
|
|
followed_urls = {a.actor_url for a in following}
|
|
from shared.infrastructure.ap_social_sx import actor_list_items_sx
|
|
from shared.sx.helpers import sx_response
|
|
content = actor_list_items_sx(actors, total, page, "followers", followed_urls, actor)
|
|
return sx_response(content)
|
|
|
|
# -- Following ------------------------------------------------------------
|
|
|
|
@bp.get("/following")
|
|
async def following_list():
|
|
actor = _require_actor()
|
|
actors, total = await services.federation.get_following(
|
|
g._ap_s, actor.preferred_username,
|
|
)
|
|
from shared.infrastructure.ap_social_sx import social_following_content_sx
|
|
content = social_following_content_sx(actors, total, 1, actor)
|
|
return await _render_social_page(content, actor, title="Following")
|
|
|
|
@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._ap_s, actor.preferred_username, page=page,
|
|
)
|
|
from shared.infrastructure.ap_social_sx import actor_list_items_sx
|
|
from shared.sx.helpers import sx_response
|
|
content = actor_list_items_sx(actors, total, page, "following", set(), actor)
|
|
return sx_response(content)
|
|
|
|
# -- 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._ap_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._ap_s, id)
|
|
is_following = False
|
|
if actor:
|
|
from shared.models.federation import APFollowing
|
|
existing = (
|
|
await g._ap_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
|
|
from shared.infrastructure.ap_social_sx import social_actor_timeline_content_sx
|
|
content = social_actor_timeline_content_sx(remote_dto, items, is_following, actor)
|
|
return await _render_social_page(content, 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._ap_s, id, before=before,
|
|
)
|
|
from shared.infrastructure.ap_social_sx import timeline_items_sx
|
|
from shared.sx.helpers import sx_response
|
|
content = timeline_items_sx(items, "actor", id, actor)
|
|
return sx_response(content)
|
|
|
|
return bp
|