giles
418ac9424f
All three services now fetch page data via (service ...) IO primitives in .sx defpages instead of Python middleman functions. - Account: newsletters-data → AccountPageService.newsletters_data - Federation: 8 page helpers → FederationPageService methods (timeline, compose, search, following, followers, notifications) - Cart: 4 page helpers → CartPageService methods (overview, page-cart, admin, payments) - Serializers moved to service modules, thin delegates kept for routes - ~520 lines of Python page helpers removed Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
206 lines
8.0 KiB
Python
206 lines
8.0 KiB
Python
"""Federation page data service — provides serialized dicts for .sx defpages."""
|
|
from __future__ import annotations
|
|
|
|
|
|
def _serialize_actor(actor) -> dict | None:
|
|
if not actor:
|
|
return None
|
|
return {
|
|
"id": actor.id,
|
|
"preferred_username": actor.preferred_username,
|
|
"display_name": getattr(actor, "display_name", None),
|
|
"icon_url": getattr(actor, "icon_url", None),
|
|
"summary": getattr(actor, "summary", None),
|
|
"actor_url": getattr(actor, "actor_url", ""),
|
|
"domain": getattr(actor, "domain", ""),
|
|
}
|
|
|
|
|
|
def _serialize_timeline_item(item) -> dict:
|
|
published = getattr(item, "published", None)
|
|
return {
|
|
"object_id": getattr(item, "object_id", "") or "",
|
|
"author_inbox": getattr(item, "author_inbox", "") or "",
|
|
"actor_icon": getattr(item, "actor_icon", None),
|
|
"actor_name": getattr(item, "actor_name", "?"),
|
|
"actor_username": getattr(item, "actor_username", ""),
|
|
"actor_domain": getattr(item, "actor_domain", ""),
|
|
"content": getattr(item, "content", ""),
|
|
"summary": getattr(item, "summary", None),
|
|
"published": published.strftime("%b %d, %H:%M") if published else "",
|
|
"before_cursor": published.isoformat() if published else "",
|
|
"url": getattr(item, "url", None),
|
|
"post_type": getattr(item, "post_type", ""),
|
|
"boosted_by": getattr(item, "boosted_by", None),
|
|
"like_count": getattr(item, "like_count", 0) or 0,
|
|
"boost_count": getattr(item, "boost_count", 0) or 0,
|
|
"liked_by_me": getattr(item, "liked_by_me", False),
|
|
"boosted_by_me": getattr(item, "boosted_by_me", False),
|
|
}
|
|
|
|
|
|
def _serialize_remote_actor(a) -> dict:
|
|
return {
|
|
"id": getattr(a, "id", None),
|
|
"display_name": getattr(a, "display_name", None) or getattr(a, "preferred_username", ""),
|
|
"preferred_username": getattr(a, "preferred_username", ""),
|
|
"domain": getattr(a, "domain", ""),
|
|
"icon_url": getattr(a, "icon_url", None),
|
|
"actor_url": getattr(a, "actor_url", ""),
|
|
"summary": getattr(a, "summary", None),
|
|
}
|
|
|
|
|
|
def _get_actor():
|
|
from quart import g
|
|
return getattr(g, "_social_actor", None)
|
|
|
|
|
|
def _require_actor():
|
|
from quart import abort
|
|
actor = _get_actor()
|
|
if not actor:
|
|
abort(403, "You need to choose a federation username first")
|
|
return actor
|
|
|
|
|
|
class FederationPageService:
|
|
"""Service for federation page data, callable via (service "federation-page" ...)."""
|
|
|
|
async def home_timeline_data(self, session, **kw):
|
|
actor = _require_actor()
|
|
from shared.services.registry import services
|
|
items = await services.federation.get_home_timeline(session, actor.id)
|
|
return {
|
|
"items": [_serialize_timeline_item(i) for i in items],
|
|
"timeline_type": "home",
|
|
"actor": _serialize_actor(actor),
|
|
}
|
|
|
|
async def public_timeline_data(self, session, **kw):
|
|
actor = _get_actor()
|
|
from shared.services.registry import services
|
|
items = await services.federation.get_public_timeline(session)
|
|
return {
|
|
"items": [_serialize_timeline_item(i) for i in items],
|
|
"timeline_type": "public",
|
|
"actor": _serialize_actor(actor),
|
|
}
|
|
|
|
async def compose_data(self, session, **kw):
|
|
from quart import request
|
|
_require_actor()
|
|
reply_to = request.args.get("reply_to")
|
|
return {"reply_to": reply_to or None}
|
|
|
|
async def search_data(self, session, **kw):
|
|
from quart import request
|
|
actor = _get_actor()
|
|
from shared.services.registry import services
|
|
query = request.args.get("q", "").strip()
|
|
actors_list = []
|
|
total = 0
|
|
followed_urls: list[str] = []
|
|
if query:
|
|
actors_list, total = await services.federation.search_actors(session, query)
|
|
if actor:
|
|
following, _ = await services.federation.get_following(
|
|
session, actor.preferred_username, page=1, per_page=1000,
|
|
)
|
|
followed_urls = [a.actor_url for a in following]
|
|
return {
|
|
"query": query,
|
|
"actors": [_serialize_remote_actor(a) for a in actors_list],
|
|
"total": total,
|
|
"followed_urls": followed_urls,
|
|
"actor": _serialize_actor(actor),
|
|
}
|
|
|
|
async def following_data(self, session, **kw):
|
|
actor = _require_actor()
|
|
from shared.services.registry import services
|
|
actors_list, total = await services.federation.get_following(
|
|
session, actor.preferred_username,
|
|
)
|
|
return {
|
|
"actors": [_serialize_remote_actor(a) for a in actors_list],
|
|
"total": total,
|
|
"actor": _serialize_actor(actor),
|
|
}
|
|
|
|
async def followers_data(self, session, **kw):
|
|
actor = _require_actor()
|
|
from shared.services.registry import services
|
|
actors_list, total = await services.federation.get_followers_paginated(
|
|
session, actor.preferred_username,
|
|
)
|
|
following, _ = await services.federation.get_following(
|
|
session, actor.preferred_username, page=1, per_page=1000,
|
|
)
|
|
followed_urls = [a.actor_url for a in following]
|
|
return {
|
|
"actors": [_serialize_remote_actor(a) for a in actors_list],
|
|
"total": total,
|
|
"followed_urls": followed_urls,
|
|
"actor": _serialize_actor(actor),
|
|
}
|
|
|
|
async def actor_timeline_data(self, session, *, id=None, **kw):
|
|
from quart import abort
|
|
from sqlalchemy import select as sa_select
|
|
from shared.models.federation import RemoteActor
|
|
from shared.services.registry import services
|
|
from shared.services.federation_impl import _remote_actor_to_dto
|
|
|
|
actor = _get_actor()
|
|
actor_id = id
|
|
remote = (
|
|
await session.execute(
|
|
sa_select(RemoteActor).where(RemoteActor.id == actor_id)
|
|
)
|
|
).scalar_one_or_none()
|
|
if not remote:
|
|
abort(404)
|
|
remote_dto = _remote_actor_to_dto(remote)
|
|
items = await services.federation.get_actor_timeline(session, actor_id)
|
|
is_following = False
|
|
if actor:
|
|
from shared.models.federation import APFollowing
|
|
existing = (
|
|
await session.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
|
|
return {
|
|
"remote_actor": _serialize_remote_actor(remote_dto),
|
|
"items": [_serialize_timeline_item(i) for i in items],
|
|
"is_following": is_following,
|
|
"actor": _serialize_actor(actor),
|
|
}
|
|
|
|
async def notifications_data(self, session, **kw):
|
|
actor = _require_actor()
|
|
from shared.services.registry import services
|
|
items = await services.federation.get_notifications(session, actor.id)
|
|
await services.federation.mark_notifications_read(session, actor.id)
|
|
|
|
notif_dicts = []
|
|
for n in items:
|
|
created = getattr(n, "created_at", None)
|
|
notif_dicts.append({
|
|
"from_actor_name": getattr(n, "from_actor_name", "?"),
|
|
"from_actor_username": getattr(n, "from_actor_username", ""),
|
|
"from_actor_domain": getattr(n, "from_actor_domain", ""),
|
|
"from_actor_icon": getattr(n, "from_actor_icon", None),
|
|
"notification_type": getattr(n, "notification_type", ""),
|
|
"target_content_preview": getattr(n, "target_content_preview", None),
|
|
"created_at_formatted": created.strftime("%b %d, %H:%M") if created else "",
|
|
"read": getattr(n, "read", True),
|
|
"app_domain": getattr(n, "app_domain", ""),
|
|
})
|
|
return {"notifications": notif_dicts}
|