microblog/app/admin.py

536 lines
15 KiB
Python
Raw Normal View History

2022-06-22 18:11:22 +00:00
from fastapi import APIRouter
from fastapi import Cookie
from fastapi import Depends
from fastapi import Form
from fastapi import Request
from fastapi import UploadFile
from fastapi.exceptions import HTTPException
from fastapi.responses import RedirectResponse
from sqlalchemy.orm import Session
from sqlalchemy.orm import joinedload
from app import activitypub as ap
from app import boxes
from app import models
from app import templates
from app.actor import LOCAL_ACTOR
2022-06-22 18:11:22 +00:00
from app.actor import get_actors_metadata
from app.boxes import get_inbox_object_by_ap_id
from app.boxes import get_outbox_object_by_ap_id
2022-06-22 18:11:22 +00:00
from app.boxes import send_follow
2022-06-27 18:55:44 +00:00
from app.config import EMOJIS
2022-06-22 18:11:22 +00:00
from app.config import generate_csrf_token
from app.config import session_serializer
from app.config import verify_csrf_token
from app.config import verify_password
from app.database import get_db
from app.lookup import lookup
2022-06-23 19:07:20 +00:00
from app.uploads import save_upload
2022-06-28 18:10:25 +00:00
from app.utils import pagination
2022-06-27 18:55:44 +00:00
from app.utils.emoji import EMOJIS_BY_NAME
2022-06-22 18:11:22 +00:00
def user_session_or_redirect(
request: Request,
session: str | None = Cookie(default=None),
) -> None:
_RedirectToLoginPage = HTTPException(
status_code=302,
headers={"Location": request.url_for("login")},
)
if not session:
raise _RedirectToLoginPage
try:
loaded_session = session_serializer.loads(session, max_age=3600 * 12)
except Exception:
raise _RedirectToLoginPage
if not loaded_session.get("is_logged_in"):
raise _RedirectToLoginPage
return None
router = APIRouter(
dependencies=[Depends(user_session_or_redirect)],
)
unauthenticated_router = APIRouter()
@router.get("/")
def admin_index(
request: Request,
db: Session = Depends(get_db),
) -> templates.TemplateResponse:
return templates.render_template(db, request, "index.html", {"request": request})
@router.get("/lookup")
def get_lookup(
request: Request,
query: str | None = None,
db: Session = Depends(get_db),
) -> templates.TemplateResponse:
ap_object = None
actors_metadata = {}
if query:
ap_object = lookup(db, query)
if ap_object.ap_type in ap.ACTOR_TYPES:
2022-06-22 19:15:07 +00:00
actors_metadata = get_actors_metadata(db, [ap_object]) # type: ignore
2022-06-22 18:11:22 +00:00
else:
2022-06-22 19:15:07 +00:00
actors_metadata = get_actors_metadata(db, [ap_object.actor]) # type: ignore
2022-06-22 18:11:22 +00:00
print(ap_object)
return templates.render_template(
db,
request,
"lookup.html",
{
"query": query,
"ap_object": ap_object,
"actors_metadata": actors_metadata,
},
)
@router.get("/new")
def admin_new(
request: Request,
query: str | None = None,
2022-06-24 09:33:05 +00:00
in_reply_to: str | None = None,
2022-06-22 18:11:22 +00:00
db: Session = Depends(get_db),
) -> templates.TemplateResponse:
content = ""
2022-06-24 09:33:05 +00:00
in_reply_to_object = None
if in_reply_to:
in_reply_to_object = boxes.get_anybox_object_by_ap_id(db, in_reply_to)
# Add mentions to the initial note content
2022-06-24 09:33:05 +00:00
if not in_reply_to_object:
raise ValueError(f"Unknown object {in_reply_to=}")
if in_reply_to_object.actor.ap_id != LOCAL_ACTOR.ap_id:
content += f"{in_reply_to_object.actor.handle} "
for tag in in_reply_to_object.tags:
if tag.get("type") == "Mention" and tag["name"] != LOCAL_ACTOR.handle:
content += f'{tag["name"]} '
2022-06-24 09:33:05 +00:00
2022-06-22 18:11:22 +00:00
return templates.render_template(
db,
request,
"admin_new.html",
{
"in_reply_to_object": in_reply_to_object,
"content": content,
"visibility_choices": [
(v.name, ap.VisibilityEnum.get_display_name(v))
for v in ap.VisibilityEnum
],
2022-06-27 18:55:44 +00:00
"emojis": EMOJIS.split(" "),
"custom_emojis": sorted(
[dat for name, dat in EMOJIS_BY_NAME.items()],
key=lambda obj: obj["name"],
),
2022-06-22 18:11:22 +00:00
},
)
2022-06-26 09:09:43 +00:00
@router.get("/bookmarks")
def admin_bookmarks(
request: Request,
db: Session = Depends(get_db),
) -> templates.TemplateResponse:
stream = (
db.query(models.InboxObject)
.filter(
models.InboxObject.ap_type.in_(["Note", "Article", "Video", "Announce"]),
models.InboxObject.is_hidden_from_stream.is_(False),
models.InboxObject.undone_by_inbox_object_id.is_(None),
models.InboxObject.is_bookmarked.is_(True),
)
.order_by(models.InboxObject.ap_published_at.desc())
.limit(20)
.all()
)
return templates.render_template(
db,
request,
"admin_stream.html",
{
"stream": stream,
},
)
2022-06-25 06:23:28 +00:00
@router.get("/inbox")
def admin_inbox(
request: Request,
db: Session = Depends(get_db),
2022-06-28 18:10:25 +00:00
filter_by: str | None = None,
cursor: str | None = None,
2022-06-25 06:23:28 +00:00
) -> templates.TemplateResponse:
2022-06-28 18:10:25 +00:00
q = db.query(models.InboxObject).filter(
models.InboxObject.ap_type.not_in(["Accept"])
)
if filter_by:
q = q.filter(models.InboxObject.ap_type == filter_by)
if cursor:
q = q.filter(
models.InboxObject.ap_published_at < pagination.decode_cursor(cursor)
)
page_size = 20
remaining_count = q.count()
2022-06-25 06:23:28 +00:00
inbox = (
2022-06-28 18:10:25 +00:00
q.options(
2022-06-25 06:23:28 +00:00
joinedload(models.InboxObject.relates_to_inbox_object),
joinedload(models.InboxObject.relates_to_outbox_object),
)
.order_by(models.InboxObject.ap_published_at.desc())
.limit(20)
.all()
)
2022-06-28 18:10:25 +00:00
next_cursor = (
pagination.encode_cursor(inbox[-1].ap_published_at)
if inbox and remaining_count > page_size
else None
)
2022-06-25 06:23:28 +00:00
return templates.render_template(
db,
request,
"admin_inbox.html",
{
"inbox": inbox,
2022-06-28 18:10:25 +00:00
"next_cursor": next_cursor,
2022-06-25 06:23:28 +00:00
},
)
@router.get("/outbox")
def admin_outbox(
2022-06-28 18:10:25 +00:00
request: Request,
db: Session = Depends(get_db),
filter_by: str | None = None,
cursor: str | None = None,
2022-06-25 06:23:28 +00:00
) -> templates.TemplateResponse:
q = db.query(models.OutboxObject).filter(
models.OutboxObject.ap_type.not_in(["Accept"])
)
if filter_by:
q = q.filter(models.OutboxObject.ap_type == filter_by)
2022-06-28 18:10:25 +00:00
if cursor:
q = q.filter(
models.OutboxObject.ap_published_at < pagination.decode_cursor(cursor)
)
page_size = 20
remaining_count = q.count()
2022-06-25 06:23:28 +00:00
outbox = (
q.options(
2022-06-25 06:23:28 +00:00
joinedload(models.OutboxObject.relates_to_inbox_object),
joinedload(models.OutboxObject.relates_to_outbox_object),
joinedload(models.OutboxObject.relates_to_actor),
2022-06-25 06:23:28 +00:00
)
.order_by(models.OutboxObject.ap_published_at.desc())
2022-06-28 18:10:25 +00:00
.limit(page_size)
2022-06-25 06:23:28 +00:00
.all()
)
2022-06-28 18:10:25 +00:00
next_cursor = (
pagination.encode_cursor(outbox[-1].ap_published_at)
if outbox and remaining_count > page_size
else None
)
actors_metadata = get_actors_metadata(
db,
[
outbox_object.relates_to_actor
for outbox_object in outbox
if outbox_object.relates_to_actor
],
)
2022-06-25 06:23:28 +00:00
return templates.render_template(
db,
request,
"admin_outbox.html",
{
"actors_metadata": actors_metadata,
2022-06-25 06:23:28 +00:00
"outbox": outbox,
2022-06-28 18:10:25 +00:00
"next_cursor": next_cursor,
2022-06-25 06:23:28 +00:00
},
)
2022-06-22 18:11:22 +00:00
@router.get("/notifications")
def get_notifications(
request: Request, db: Session = Depends(get_db)
) -> templates.TemplateResponse:
notifications = (
db.query(models.Notification)
.options(
joinedload(models.Notification.actor),
joinedload(models.Notification.inbox_object),
joinedload(models.Notification.outbox_object),
)
.order_by(models.Notification.created_at.desc())
.all()
)
actors_metadata = get_actors_metadata(
db, [notif.actor for notif in notifications if notif.actor]
)
for notif in notifications:
notif.is_new = False
db.commit()
return templates.render_template(
db,
request,
"notifications.html",
{
"notifications": notifications,
"actors_metadata": actors_metadata,
},
)
2022-06-25 10:19:12 +00:00
@router.get("/object")
def admin_object(
request: Request,
ap_id: str,
db: Session = Depends(get_db),
) -> templates.TemplateResponse:
2022-06-25 10:29:35 +00:00
requested_object = boxes.get_anybox_object_by_ap_id(db, ap_id)
if not requested_object:
raise HTTPException(status_code=404)
replies_tree = boxes.get_replies_tree(db, requested_object)
2022-06-25 10:19:12 +00:00
return templates.render_template(
db,
request,
2022-06-25 10:29:35 +00:00
"object.html",
{"replies_tree": replies_tree},
2022-06-25 10:19:12 +00:00
)
@router.get("/profile")
def admin_profile(
request: Request,
actor_id: str,
db: Session = Depends(get_db),
) -> templates.TemplateResponse:
actor = db.query(models.Actor).filter(models.Actor.ap_id == actor_id).one_or_none()
if not actor:
raise HTTPException(status_code=404)
actors_metadata = get_actors_metadata(db, [actor])
inbox_objects = (
db.query(models.InboxObject)
.filter(
models.InboxObject.actor_id == actor.id,
models.InboxObject.ap_type.in_(["Note", "Article", "Video"]),
)
.order_by(models.InboxObject.ap_published_at.desc())
2022-06-25 10:19:12 +00:00
.all()
)
return templates.render_template(
db,
request,
"admin_profile.html",
{
"actors_metadata": actors_metadata,
"actor": actor,
"inbox_objects": inbox_objects,
},
)
2022-06-22 18:11:22 +00:00
@router.post("/actions/follow")
def admin_actions_follow(
request: Request,
ap_actor_id: str = Form(),
redirect_url: str = Form(),
csrf_check: None = Depends(verify_csrf_token),
db: Session = Depends(get_db),
) -> RedirectResponse:
print(f"Following {ap_actor_id}")
send_follow(db, ap_actor_id)
return RedirectResponse(redirect_url, status_code=302)
@router.post("/actions/like")
def admin_actions_like(
request: Request,
ap_object_id: str = Form(),
redirect_url: str = Form(),
csrf_check: None = Depends(verify_csrf_token),
db: Session = Depends(get_db),
) -> RedirectResponse:
boxes.send_like(db, ap_object_id)
return RedirectResponse(redirect_url, status_code=302)
@router.post("/actions/undo")
def admin_actions_undo(
request: Request,
ap_object_id: str = Form(),
redirect_url: str = Form(),
csrf_check: None = Depends(verify_csrf_token),
db: Session = Depends(get_db),
) -> RedirectResponse:
boxes.send_undo(db, ap_object_id)
return RedirectResponse(redirect_url, status_code=302)
@router.post("/actions/announce")
def admin_actions_announce(
request: Request,
ap_object_id: str = Form(),
redirect_url: str = Form(),
csrf_check: None = Depends(verify_csrf_token),
db: Session = Depends(get_db),
) -> RedirectResponse:
boxes.send_announce(db, ap_object_id)
return RedirectResponse(redirect_url, status_code=302)
@router.post("/actions/bookmark")
def admin_actions_bookmark(
request: Request,
ap_object_id: str = Form(),
redirect_url: str = Form(),
csrf_check: None = Depends(verify_csrf_token),
db: Session = Depends(get_db),
) -> RedirectResponse:
inbox_object = get_inbox_object_by_ap_id(db, ap_object_id)
if not inbox_object:
raise ValueError("Should never happen")
inbox_object.is_bookmarked = True
db.commit()
return RedirectResponse(redirect_url, status_code=302)
2022-06-26 09:09:43 +00:00
@router.post("/actions/unbookmark")
def admin_actions_unbookmark(
request: Request,
ap_object_id: str = Form(),
redirect_url: str = Form(),
csrf_check: None = Depends(verify_csrf_token),
db: Session = Depends(get_db),
) -> RedirectResponse:
inbox_object = get_inbox_object_by_ap_id(db, ap_object_id)
if not inbox_object:
raise ValueError("Should never happen")
inbox_object.is_bookmarked = False
db.commit()
return RedirectResponse(redirect_url, status_code=302)
@router.post("/actions/pin")
def admin_actions_pin(
request: Request,
ap_object_id: str = Form(),
redirect_url: str = Form(),
csrf_check: None = Depends(verify_csrf_token),
db: Session = Depends(get_db),
) -> RedirectResponse:
outbox_object = get_outbox_object_by_ap_id(db, ap_object_id)
if not outbox_object:
raise ValueError("Should never happen")
outbox_object.is_pinned = True
db.commit()
return RedirectResponse(redirect_url, status_code=302)
@router.post("/actions/unpin")
def admin_actions_unpin(
request: Request,
ap_object_id: str = Form(),
redirect_url: str = Form(),
csrf_check: None = Depends(verify_csrf_token),
db: Session = Depends(get_db),
) -> RedirectResponse:
outbox_object = get_outbox_object_by_ap_id(db, ap_object_id)
if not outbox_object:
raise ValueError("Should never happen")
outbox_object.is_pinned = False
db.commit()
return RedirectResponse(redirect_url, status_code=302)
2022-06-22 18:11:22 +00:00
@router.post("/actions/new")
2022-06-23 19:07:20 +00:00
def admin_actions_new(
2022-06-22 18:11:22 +00:00
request: Request,
2022-06-27 18:55:44 +00:00
files: list[UploadFile] = [],
2022-06-22 18:11:22 +00:00
content: str = Form(),
redirect_url: str = Form(),
2022-06-24 20:41:43 +00:00
in_reply_to: str | None = Form(None),
visibility: str = Form(),
2022-06-22 18:11:22 +00:00
csrf_check: None = Depends(verify_csrf_token),
db: Session = Depends(get_db),
) -> RedirectResponse:
# XXX: for some reason, no files restuls in an empty single file
2022-06-23 19:07:20 +00:00
uploads = []
2022-06-22 18:11:22 +00:00
if len(files) >= 1 and files[0].filename:
2022-06-23 19:07:20 +00:00
for f in files:
upload = save_upload(db, f)
uploads.append((upload, f.filename))
2022-06-24 09:33:05 +00:00
public_id = boxes.send_create(
db,
source=content,
uploads=uploads,
in_reply_to=in_reply_to or None,
visibility=ap.VisibilityEnum[visibility],
2022-06-24 09:33:05 +00:00
)
2022-06-22 18:11:22 +00:00
return RedirectResponse(
request.url_for("outbox_by_public_id", public_id=public_id),
status_code=302,
)
@unauthenticated_router.get("/login")
def login(
request: Request,
db: Session = Depends(get_db),
) -> templates.TemplateResponse:
return templates.render_template(
db,
request,
"login.html",
{"csrf_token": generate_csrf_token()},
)
@unauthenticated_router.post("/login")
def login_validation(
request: Request,
password: str = Form(),
csrf_check: None = Depends(verify_csrf_token),
) -> RedirectResponse:
if not verify_password(password):
raise HTTPException(status_code=401)
2022-06-27 18:55:44 +00:00
resp = RedirectResponse("/admin/inbox", status_code=302)
2022-06-22 18:11:22 +00:00
resp.set_cookie("session", session_serializer.dumps({"is_logged_in": True})) # type: ignore # noqa: E501
return resp
@router.get("/logout")
def logout(
request: Request,
) -> RedirectResponse:
resp = RedirectResponse(request.url_for("index"), status_code=302)
resp.set_cookie("session", session_serializer.dumps({"is_logged_in": False})) # type: ignore # noqa: E501
return resp