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>
420 lines
17 KiB
Python
420 lines
17 KiB
Python
"""Social fediverse routes: timeline, compose, search, follow, interactions, notifications."""
|
|
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
|
|
from shared.sx.helpers import sx_response
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def _require_actor():
|
|
"""Return actor context or abort 403."""
|
|
actor = g.get("ctx", {}).get("actor") if hasattr(g, "ctx") else None
|
|
if not actor:
|
|
actor = getattr(g, "_social_actor", None)
|
|
if not actor:
|
|
abort(403, "You need to choose a federation username first")
|
|
return actor
|
|
|
|
|
|
def register(url_prefix="/social"):
|
|
bp = Blueprint("social", __name__, url_prefix=url_prefix)
|
|
|
|
@bp.before_request
|
|
async def load_actor():
|
|
"""Load actor profile for authenticated users."""
|
|
if g.get("user"):
|
|
actor = await services.federation.get_actor_by_user_id(g.s, g.user.id)
|
|
g._social_actor = actor
|
|
|
|
@bp.before_request
|
|
async def _prepare_page_data():
|
|
"""Pre-render content for defpage routes."""
|
|
endpoint = request.endpoint or ""
|
|
|
|
if endpoint.endswith("defpage_home_timeline"):
|
|
actor = _require_actor()
|
|
items = await services.federation.get_home_timeline(g.s, actor.id)
|
|
from sx.sx_components import _timeline_content_sx
|
|
g.home_timeline_content = _timeline_content_sx(items, "home", actor)
|
|
|
|
elif endpoint.endswith("defpage_public_timeline"):
|
|
actor = getattr(g, "_social_actor", None)
|
|
items = await services.federation.get_public_timeline(g.s)
|
|
from sx.sx_components import _timeline_content_sx
|
|
g.public_timeline_content = _timeline_content_sx(items, "public", actor)
|
|
|
|
elif endpoint.endswith("defpage_compose_form"):
|
|
actor = _require_actor()
|
|
from sx.sx_components import _compose_content_sx
|
|
reply_to = request.args.get("reply_to")
|
|
g.compose_content = _compose_content_sx(actor, reply_to)
|
|
|
|
elif endpoint.endswith("defpage_search"):
|
|
actor = getattr(g, "_social_actor", None)
|
|
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
|
|
g.search_content = _search_content_sx(query, actors_list, total, 1, followed_urls, actor)
|
|
|
|
elif endpoint.endswith("defpage_following_list"):
|
|
actor = _require_actor()
|
|
actors_list, total = await services.federation.get_following(
|
|
g.s, actor.preferred_username,
|
|
)
|
|
from sx.sx_components import _following_content_sx
|
|
g.following_content = _following_content_sx(actors_list, total, actor)
|
|
|
|
elif endpoint.endswith("defpage_followers_list"):
|
|
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
|
|
g.followers_content = _followers_content_sx(actors_list, total, followed_urls, actor)
|
|
|
|
elif endpoint.endswith("defpage_actor_timeline"):
|
|
actor = getattr(g, "_social_actor", None)
|
|
actor_id = request.view_args.get("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
|
|
g.actor_timeline_content = _actor_timeline_content_sx(remote_dto, items, is_following, actor)
|
|
|
|
elif endpoint.endswith("defpage_notifications"):
|
|
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
|
|
g.notifications_content = _notifications_content_sx(items)
|
|
|
|
# -- Timeline pagination ---------------------------------------------------
|
|
|
|
@bp.get("/timeline")
|
|
async def home_timeline_page():
|
|
actor = _require_actor()
|
|
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_home_timeline(
|
|
g.s, actor.id, before=before,
|
|
)
|
|
from sx.sx_components import render_timeline_items
|
|
sx_src = await render_timeline_items(items, "home", actor)
|
|
return sx_response(sx_src)
|
|
|
|
@bp.get("/public/timeline")
|
|
async def public_timeline_page():
|
|
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_public_timeline(g.s, before=before)
|
|
actor = getattr(g, "_social_actor", None)
|
|
from sx.sx_components import render_timeline_items
|
|
sx_src = await render_timeline_items(items, "public", actor)
|
|
return sx_response(sx_src)
|
|
|
|
# -- Compose ---------------------------------------------------------------
|
|
|
|
@bp.post("/compose")
|
|
async def compose_submit():
|
|
actor = _require_actor()
|
|
form = await request.form
|
|
content = form.get("content", "").strip()
|
|
if not content:
|
|
return redirect(url_for("social.defpage_compose_form"))
|
|
|
|
visibility = form.get("visibility", "public")
|
|
in_reply_to = form.get("in_reply_to") or None
|
|
|
|
await services.federation.create_local_post(
|
|
g.s, actor.id,
|
|
content=content,
|
|
visibility=visibility,
|
|
in_reply_to=in_reply_to,
|
|
)
|
|
return redirect(url_for("social.defpage_home_timeline"))
|
|
|
|
@bp.post("/delete/<int:post_id>")
|
|
async def delete_post(post_id: int):
|
|
actor = _require_actor()
|
|
await services.federation.delete_local_post(g.s, actor.id, post_id)
|
|
return redirect(url_for("social.defpage_home_timeline"))
|
|
|
|
# -- Search + Follow -------------------------------------------------------
|
|
|
|
@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_list = []
|
|
total = 0
|
|
followed_urls: set[str] = set()
|
|
if query:
|
|
actors_list, 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}
|
|
from sx.sx_components import render_search_results
|
|
sx_src = await render_search_results(actors_list, query, page, followed_urls, actor)
|
|
return sx_response(sx_src)
|
|
|
|
@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("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("social.defpage_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("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("social.defpage_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"
|
|
from sx.sx_components import render_actor_card
|
|
return sx_response(render_actor_card(remote_dto, actor, followed_urls, list_type=list_type))
|
|
|
|
# -- Interactions ----------------------------------------------------------
|
|
|
|
@bp.post("/like")
|
|
async def like():
|
|
actor = _require_actor()
|
|
form = await request.form
|
|
object_id = form.get("object_id", "")
|
|
author_inbox = form.get("author_inbox", "")
|
|
await services.federation.like_post(g.s, actor.id, object_id, author_inbox)
|
|
return await _interaction_buttons_response(actor, object_id, author_inbox)
|
|
|
|
@bp.post("/unlike")
|
|
async def unlike():
|
|
actor = _require_actor()
|
|
form = await request.form
|
|
object_id = form.get("object_id", "")
|
|
author_inbox = form.get("author_inbox", "")
|
|
await services.federation.unlike_post(g.s, actor.id, object_id, author_inbox)
|
|
return await _interaction_buttons_response(actor, object_id, author_inbox)
|
|
|
|
@bp.post("/boost")
|
|
async def boost():
|
|
actor = _require_actor()
|
|
form = await request.form
|
|
object_id = form.get("object_id", "")
|
|
author_inbox = form.get("author_inbox", "")
|
|
await services.federation.boost_post(g.s, actor.id, object_id, author_inbox)
|
|
return await _interaction_buttons_response(actor, object_id, author_inbox)
|
|
|
|
@bp.post("/unboost")
|
|
async def unboost():
|
|
actor = _require_actor()
|
|
form = await request.form
|
|
object_id = form.get("object_id", "")
|
|
author_inbox = form.get("author_inbox", "")
|
|
await services.federation.unboost_post(g.s, actor.id, object_id, author_inbox)
|
|
return await _interaction_buttons_response(actor, object_id, author_inbox)
|
|
|
|
async def _interaction_buttons_response(actor, object_id, author_inbox):
|
|
"""Re-render interaction buttons after a like/boost action."""
|
|
from shared.models.federation import APInteraction, APRemotePost, APActivity
|
|
from sqlalchemy import select
|
|
|
|
svc = services.federation
|
|
post_type, post_id = await svc._resolve_post(g.s, object_id)
|
|
|
|
like_count = 0
|
|
boost_count = 0
|
|
liked_by_me = False
|
|
boosted_by_me = False
|
|
|
|
if post_type:
|
|
from sqlalchemy import func as sa_func
|
|
like_count = (await g.s.execute(
|
|
select(sa_func.count(APInteraction.id)).where(
|
|
APInteraction.post_type == post_type,
|
|
APInteraction.post_id == post_id,
|
|
APInteraction.interaction_type == "like",
|
|
)
|
|
)).scalar() or 0
|
|
boost_count = (await g.s.execute(
|
|
select(sa_func.count(APInteraction.id)).where(
|
|
APInteraction.post_type == post_type,
|
|
APInteraction.post_id == post_id,
|
|
APInteraction.interaction_type == "boost",
|
|
)
|
|
)).scalar() or 0
|
|
liked_by_me = bool((await g.s.execute(
|
|
select(APInteraction.id).where(
|
|
APInteraction.actor_profile_id == actor.id,
|
|
APInteraction.post_type == post_type,
|
|
APInteraction.post_id == post_id,
|
|
APInteraction.interaction_type == "like",
|
|
).limit(1)
|
|
)).scalar())
|
|
boosted_by_me = bool((await g.s.execute(
|
|
select(APInteraction.id).where(
|
|
APInteraction.actor_profile_id == actor.id,
|
|
APInteraction.post_type == post_type,
|
|
APInteraction.post_id == post_id,
|
|
APInteraction.interaction_type == "boost",
|
|
).limit(1)
|
|
)).scalar())
|
|
|
|
from sx.sx_components import render_interaction_buttons
|
|
return sx_response(render_interaction_buttons(
|
|
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,
|
|
actor=actor,
|
|
))
|
|
|
|
# -- Following / Followers pagination --------------------------------------
|
|
|
|
@bp.get("/following/page")
|
|
async def following_list_page():
|
|
actor = _require_actor()
|
|
page = request.args.get("page", 1, type=int)
|
|
actors_list, total = await services.federation.get_following(
|
|
g.s, actor.preferred_username, page=page,
|
|
)
|
|
from sx.sx_components import render_following_items
|
|
sx_src = await render_following_items(actors_list, page, actor)
|
|
return sx_response(sx_src)
|
|
|
|
@bp.get("/followers/page")
|
|
async def followers_list_page():
|
|
actor = _require_actor()
|
|
page = request.args.get("page", 1, type=int)
|
|
actors_list, total = await services.federation.get_followers_paginated(
|
|
g.s, actor.preferred_username, page=page,
|
|
)
|
|
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 render_followers_items
|
|
sx_src = await render_followers_items(actors_list, page, followed_urls, actor)
|
|
return sx_response(sx_src)
|
|
|
|
@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,
|
|
)
|
|
from sx.sx_components import render_actor_timeline_items
|
|
sx_src = await render_actor_timeline_items(items, id, actor)
|
|
return sx_response(sx_src)
|
|
|
|
# -- Notifications ---------------------------------------------------------
|
|
|
|
@bp.get("/notifications/count")
|
|
async def notification_count():
|
|
actor = getattr(g, "_social_actor", None)
|
|
if not actor:
|
|
return Response("0", content_type="text/plain")
|
|
count = await services.federation.unread_notification_count(g.s, actor.id)
|
|
if count > 0:
|
|
from shared.sx.jinja_bridge import render as render_comp
|
|
return Response(
|
|
render_comp("notification-badge", count=str(count)),
|
|
content_type="text/html",
|
|
)
|
|
return Response("", content_type="text/html")
|
|
|
|
@bp.post("/notifications/read")
|
|
async def mark_read():
|
|
actor = _require_actor()
|
|
await services.federation.mark_notifications_read(g.s, actor.id)
|
|
return redirect(url_for("social.defpage_notifications"))
|
|
|
|
return bp
|