kopia lustrzana https://github.com/Aonrud/ila-microblog.pub
156 wiersze
3.7 KiB
Python
156 wiersze
3.7 KiB
Python
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
from typing import Any
|
|
from typing import Callable
|
|
|
|
from fastapi import APIRouter
|
|
from fastapi import Depends
|
|
from fastapi import Request
|
|
from loguru import logger
|
|
from starlette.responses import JSONResponse
|
|
|
|
if TYPE_CHECKING:
|
|
from app.ap_object import RemoteObject
|
|
|
|
|
|
_DATA_DIR = Path().parent.resolve() / "data"
|
|
_Handler = Callable[..., Any]
|
|
|
|
|
|
class HTMLPage:
|
|
def __init__(
|
|
self,
|
|
title: str,
|
|
html_file: str,
|
|
show_in_navbar: bool,
|
|
) -> None:
|
|
self.title = title
|
|
self.html_file = _DATA_DIR / html_file
|
|
self.show_in_navbar = show_in_navbar
|
|
|
|
|
|
class RawHandler:
|
|
def __init__(
|
|
self,
|
|
title: str,
|
|
handler: Any,
|
|
show_in_navbar: bool,
|
|
) -> None:
|
|
self.title = title
|
|
self.handler = handler
|
|
self.show_in_navbar = show_in_navbar
|
|
|
|
|
|
_CUSTOM_ROUTES: dict[str, HTMLPage | RawHandler] = {}
|
|
|
|
|
|
def register_html_page(
|
|
path: str,
|
|
*,
|
|
title: str,
|
|
html_file: str,
|
|
show_in_navbar: bool = True,
|
|
) -> None:
|
|
if path in _CUSTOM_ROUTES:
|
|
raise ValueError(f"{path} is already registered")
|
|
|
|
_CUSTOM_ROUTES[path] = HTMLPage(title, html_file, show_in_navbar)
|
|
|
|
|
|
def register_raw_handler(
|
|
path: str,
|
|
*,
|
|
title: str,
|
|
handler: _Handler,
|
|
show_in_navbar: bool = True,
|
|
) -> None:
|
|
if path in _CUSTOM_ROUTES:
|
|
raise ValueError(f"{path} is already registered")
|
|
|
|
_CUSTOM_ROUTES[path] = RawHandler(title, handler, show_in_navbar)
|
|
|
|
|
|
class ActivityPubResponse(JSONResponse):
|
|
media_type = "application/activity+json"
|
|
|
|
|
|
def _custom_page_handler(path: str, html_page: HTMLPage) -> Any:
|
|
from app import templates
|
|
from app.actor import LOCAL_ACTOR
|
|
from app.config import is_activitypub_requested
|
|
from app.database import AsyncSession
|
|
from app.database import get_db_session
|
|
|
|
async def _handler(
|
|
request: Request,
|
|
db_session: AsyncSession = Depends(get_db_session),
|
|
) -> templates.TemplateResponse | ActivityPubResponse:
|
|
if path == "/" and is_activitypub_requested(request):
|
|
return ActivityPubResponse(LOCAL_ACTOR.ap_actor)
|
|
|
|
return await templates.render_template(
|
|
db_session,
|
|
request,
|
|
"custom_page.html",
|
|
{
|
|
"page_content": html_page.html_file.read_text(),
|
|
"title": html_page.title,
|
|
},
|
|
)
|
|
|
|
return _handler
|
|
|
|
|
|
def get_custom_router() -> APIRouter | None:
|
|
if not _CUSTOM_ROUTES:
|
|
return None
|
|
|
|
router = APIRouter()
|
|
|
|
for path, handler in _CUSTOM_ROUTES.items():
|
|
if isinstance(handler, HTMLPage):
|
|
router.add_api_route(
|
|
path, _custom_page_handler(path, handler), methods=["GET"]
|
|
)
|
|
else:
|
|
router.add_api_route(path, handler.handler)
|
|
|
|
return router
|
|
|
|
|
|
@dataclass
|
|
class ObjectInfo:
|
|
# Is it a reply?
|
|
is_reply: bool
|
|
|
|
# Is it a reply to an outbox object
|
|
is_local_reply: bool
|
|
|
|
# Is the object mentioning the local actor
|
|
is_mention: bool
|
|
|
|
# Is it from someone the local actor is following
|
|
is_from_following: bool
|
|
|
|
# List of hashtags, e.g. #microblogpub
|
|
hashtags: list[str]
|
|
|
|
# @dev@microblog.pub
|
|
actor_handle: str
|
|
|
|
remote_object: "RemoteObject"
|
|
|
|
|
|
_StreamVisibilityCallback = Callable[[ObjectInfo], bool]
|
|
|
|
|
|
def default_stream_visibility_callback(object_info: ObjectInfo) -> bool:
|
|
result = (
|
|
(not object_info.is_reply and object_info.is_from_following)
|
|
or object_info.is_mention
|
|
or object_info.is_local_reply
|
|
)
|
|
logger.info(f"{object_info=}/{result=}")
|
|
return result
|