Add full fediverse social service
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 52s
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 52s
Social blueprint with timeline, compose, search, follow/unfollow, like/boost interactions, and notifications. Inbox handler extended for Create/Update/Delete/Accept/Like/Announce with notification creation. HTMX-powered infinite scroll and interaction buttons. Nav updated with Timeline, Public, Search, and Notifications links. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
giles
0
bp/social/__init__.py
Normal file
0
bp/social/__init__.py
Normal file
302
bp/social/routes.py
Normal file
302
bp/social/routes.py
Normal file
@@ -0,0 +1,302 @@
|
||||
"""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, render_template, Response
|
||||
|
||||
from shared.services.registry import services
|
||||
|
||||
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
|
||||
|
||||
# -- Timeline -------------------------------------------------------------
|
||||
|
||||
@bp.get("/")
|
||||
async def home_timeline():
|
||||
if not g.get("user"):
|
||||
return redirect(url_for("auth.login_form"))
|
||||
actor = _require_actor()
|
||||
items = await services.federation.get_home_timeline(g.s, actor.id)
|
||||
return await render_template(
|
||||
"federation/timeline.html",
|
||||
items=items,
|
||||
timeline_type="home",
|
||||
actor=actor,
|
||||
)
|
||||
|
||||
@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,
|
||||
)
|
||||
return await render_template(
|
||||
"federation/_timeline_items.html",
|
||||
items=items,
|
||||
timeline_type="home",
|
||||
actor=actor,
|
||||
)
|
||||
|
||||
@bp.get("/public")
|
||||
async def public_timeline():
|
||||
items = await services.federation.get_public_timeline(g.s)
|
||||
actor = getattr(g, "_social_actor", None)
|
||||
return await render_template(
|
||||
"federation/timeline.html",
|
||||
items=items,
|
||||
timeline_type="public",
|
||||
actor=actor,
|
||||
)
|
||||
|
||||
@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)
|
||||
return await render_template(
|
||||
"federation/_timeline_items.html",
|
||||
items=items,
|
||||
timeline_type="public",
|
||||
actor=actor,
|
||||
)
|
||||
|
||||
# -- Compose --------------------------------------------------------------
|
||||
|
||||
@bp.get("/compose")
|
||||
async def compose_form():
|
||||
actor = _require_actor()
|
||||
reply_to = request.args.get("reply_to")
|
||||
return await render_template(
|
||||
"federation/compose.html",
|
||||
actor=actor,
|
||||
reply_to=reply_to,
|
||||
)
|
||||
|
||||
@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.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.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.home_timeline"))
|
||||
|
||||
# -- Search + Follow ------------------------------------------------------
|
||||
|
||||
@bp.get("/search")
|
||||
async def search():
|
||||
actor = getattr(g, "_social_actor", None)
|
||||
query = request.args.get("q", "").strip()
|
||||
result = None
|
||||
if query:
|
||||
result = await services.federation.search_remote_actor(g.s, query)
|
||||
return await render_template(
|
||||
"federation/search.html",
|
||||
query=query,
|
||||
result=result,
|
||||
actor=actor,
|
||||
)
|
||||
|
||||
@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,
|
||||
)
|
||||
return redirect(url_for("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,
|
||||
)
|
||||
return redirect(url_for("social.search"))
|
||||
|
||||
# -- 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 updated buttons for HTMX
|
||||
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
|
||||
from shared.services.federation_impl import SqlFederationService
|
||||
|
||||
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())
|
||||
|
||||
return await render_template(
|
||||
"federation/_interaction_buttons.html",
|
||||
item_object_id=object_id,
|
||||
item_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,
|
||||
)
|
||||
|
||||
# -- Notifications --------------------------------------------------------
|
||||
|
||||
@bp.get("/notifications")
|
||||
async def notifications():
|
||||
actor = _require_actor()
|
||||
items = await services.federation.get_notifications(g.s, actor.id)
|
||||
await services.federation.mark_notifications_read(g.s, actor.id)
|
||||
return await render_template(
|
||||
"federation/notifications.html",
|
||||
notifications=items,
|
||||
actor=actor,
|
||||
)
|
||||
|
||||
@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:
|
||||
return Response(
|
||||
f'<span class="bg-red-500 text-white text-xs rounded-full px-1.5 py-0.5">{count}</span>',
|
||||
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.notifications"))
|
||||
|
||||
return bp
|
||||
Reference in New Issue
Block a user