# retoor <retoor@molodetz.nl>
import logging
from typing import Annotated
from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse
from devplacepy.database import (
REPORT_STATUSES,
SYSTEM_ACTOR,
get_table,
get_users_by_uids,
)
from devplacepy.dependencies import json_or_form
from devplacepy.models import ModerationDecisionForm, ReportStatusForm
from devplacepy.responses import action_result, json_error, respond
from devplacepy.routers.admin._shared import deny_senior, is_senior_admin
from devplacepy.schemas import AdminModerationOut, AdminReportOut
from devplacepy.seo import base_seo_context, site_url, website_schema
from devplacepy.services.audit import record as audit
from devplacepy.services.moderation import enforcement, queue, sla
from devplacepy.utils import create_notification, not_found, require_admin
logger = logging.getLogger(__name__)
router = APIRouter()
QUEUE_URL = "/admin/moderation"
STATUS_TABS = [
{"key": "open", "label": "Open"},
{"key": "acknowledged", "label": "Acknowledged"},
{"key": "actioned", "label": "Actioned"},
{"key": "dismissed", "label": "Dismissed"},
]
SUBJECT_ACTIONS = ("warn", "suspend", "ban", "lift")
ENFORCEMENT_EVENTS = {
"remove_content": "moderation.remove",
"restore_content": "moderation.restore",
"warn": "moderation.warn",
"suspend": "moderation.suspend",
"ban": "moderation.ban",
"lift": "moderation.lift",
}
DECISION_MESSAGES = {
"remove_content": "Your {target} was removed after a moderation review.",
"restore_content": "Your {target} was restored after a moderation review.",
"warn": "A moderator issued a warning about your {target}.",
"suspend": "Your account is suspended following a moderation review.",
"ban": "Your account has been closed following a moderation review.",
"lift": "Your account restriction has been lifted.",
}
def _breadcrumbs(extra: list[dict] | None = None) -> list[dict]:
trail = [
{"name": "Home", "url": "/feed"},
{"name": "Admin", "url": "/admin"},
{"name": "Moderation", "url": QUEUE_URL},
]
return trail + (extra or [])
def _available_actions(target_type: str) -> list[str]:
actions = ["dismiss", "escalate"]
if enforcement.can_remove(target_type):
actions = ["remove_content", "restore_content"] + actions
return actions + list(SUBJECT_ACTIONS)
def _subject(report: dict) -> dict | None:
owner_uid = report.get("owner_uid") or ""
if not owner_uid:
return None
return get_users_by_uids([owner_uid]).get(owner_uid)
def _action_view(rows: list[dict]) -> list[dict]:
actors = get_users_by_uids([row.get("actor_uid") for row in rows if row.get("actor_uid")])
view = []
for row in rows:
actor = actors.get(row.get("actor_uid"))
view.append(
{
"uid": row["uid"],
"report_uid": row.get("report_uid", ""),
"action": row.get("action", ""),
"actor_name": actor["username"] if actor else row.get("actor_uid", ""),
"reason": row.get("reason", ""),
"notes": row.get("notes", ""),
"expires_at": row.get("expires_at", ""),
"created_at": row.get("created_at", ""),
}
)
return view
@router.get("/moderation", response_class=HTMLResponse)
async def admin_moderation(request: Request, status: str = "open", page: int = 1):
admin = require_admin(request)
if status not in REPORT_STATUSES:
status = "open"
reports, pagination = queue.list_reports(status=status, page=page)
counts = queue.status_counts()
base = site_url(request)
seo_ctx = base_seo_context(
request,
title="Moderation - Admin",
description="Triage reported content and apply moderation decisions.",
robots="noindex,nofollow",
breadcrumbs=_breadcrumbs(),
schemas=[website_schema(base)],
)
return respond(
request,
"admin_moderation.html",
{
**seo_ctx,
"request": request,
"user": admin,
"reports": reports,
"pagination": pagination,
"status": status,
"statuses": [
{**tab, "count": counts.get(tab["key"], 0), "active": tab["key"] == status}
for tab in STATUS_TABS
],
"counts": counts,
"sla": sla.snapshot(),
"admin_section": "moderation",
},
model=AdminModerationOut,
)
@router.get("/moderation/{uid}", response_class=HTMLResponse)
async def admin_report_detail(request: Request, uid: str):
admin = require_admin(request)
report = queue.get_report(uid)
if not report:
raise not_found("Report not found")
view = queue.enrich_reports([report])[0]
subject = _subject(report)
base = site_url(request)
seo_ctx = base_seo_context(
request,
title=f"Report {uid} - Admin",
description="One reported item and the decisions taken on it.",
robots="noindex,nofollow",
breadcrumbs=_breadcrumbs([{"name": "Report", "url": f"{QUEUE_URL}/{uid}"}]),
schemas=[website_schema(base)],
)
return respond(
request,
"admin_report.html",
{
**seo_ctx,
"request": request,
"user": admin,
"report": view,
"actions": _action_view(queue.actions_for_report(uid)),
"history": _action_view(
queue.actions_for_subject(report.get("owner_uid", ""))
),
"available_actions": _available_actions(report["target_type"]),
"can_remove": enforcement.can_remove(report["target_type"]),
"subject": subject,
"sla": sla.snapshot(),
"admin_section": "moderation",
},
model=AdminReportOut,
)
@router.post("/moderation/{uid}/status")
async def admin_report_status(
request: Request,
uid: str,
data: Annotated[ReportStatusForm, Depends(json_or_form(ReportStatusForm))],
):
admin = require_admin(request)
report = queue.get_report(uid)
if not report:
raise not_found("Report not found")
updated = queue.set_status(uid, data.status, admin["uid"])
if not updated:
return json_error(400, "Report status could not be changed")
logger.info(f"Admin {admin['username']} set report {uid} to {data.status}")
audit.record(
request,
"report.status",
user=admin,
target_type=report["target_type"],
target_uid=report["target_uid"],
old_value=report.get("status"),
new_value=data.status,
metadata={"report_uid": uid},
summary=f"{admin['username']} set report {uid} to {data.status}",
links=[audit.target(report["target_type"], report["target_uid"])],
)
return action_result(request, f"{QUEUE_URL}/{uid}")
@router.post("/moderation/{uid}/decide")
async def admin_report_decide(
request: Request,
uid: str,
data: Annotated[ModerationDecisionForm, Depends(json_or_form(ModerationDecisionForm))],
):
admin = require_admin(request)
report = queue.get_report(uid)
if not report:
raise not_found("Report not found")
action = data.action
if action not in _available_actions(report["target_type"]):
return json_error(400, "That action does not apply to this target")
subject = _subject(report)
if action in SUBJECT_ACTIONS:
if not subject:
return json_error(400, "This report has no account to act on")
if is_senior_admin(admin, subject):
return deny_senior(
request,
admin,
subject["uid"],
subject,
f"moderation.{action}",
redirect_url=f"{QUEUE_URL}/{uid}",
)
if action == "escalate":
queue.escalate(uid)
else:
outcome = "dismissed" if action == "dismiss" else "actioned"
if not queue.claim_open(uid, outcome, admin["uid"]):
return json_error(409, "This report was already resolved")
expires_at = ""
if action == "remove_content":
enforcement.remove_content(
request, admin, report["target_type"], report["target_uid"]
)
elif action == "restore_content":
enforcement.restore_content(report["target_type"], report["target_uid"])
elif action == "suspend":
expires_at = enforcement.suspend_user(subject, data.duration_hours, data.reason)
elif action == "ban":
enforcement.ban_user(subject, data.reason)
elif action == "lift":
enforcement.lift_suspension(subject)
enforcement.unban_user(subject)
queue.record_action(
report_uid=uid,
actor_uid=admin["uid"],
action=action,
target_type=report["target_type"],
target_uid=report["target_uid"],
subject_uid=subject["uid"] if subject else "",
reason=data.reason,
notes=data.notes,
expires_at=expires_at,
)
logger.info(f"Admin {admin['username']} applied {action} to report {uid}")
metadata = {
"report_uid": uid,
"action": action,
"reason": data.reason,
"subject_uid": subject["uid"] if subject else "",
}
links = [audit.target(report["target_type"], report["target_uid"])]
audit.record(
request,
"report.decide",
user=admin,
target_type=report["target_type"],
target_uid=report["target_uid"],
metadata=metadata,
summary=f"{admin['username']} applied {action} on report {uid}",
links=links,
)
enforcement_key = ENFORCEMENT_EVENTS.get(action)
if enforcement_key:
audit.record(
request,
enforcement_key,
user=admin,
target_type=report["target_type"],
target_uid=report["target_uid"],
metadata=metadata,
summary=f"{admin['username']} applied {action} from report {uid}",
links=links,
)
_notify_subject(subject, action, report, data.reason)
if action != "escalate":
_notify_reporter(report, action)
return action_result(request, f"{QUEUE_URL}/{uid}")
def _notify_subject(subject: dict | None, action: str, report: dict, reason: str) -> None:
template = DECISION_MESSAGES.get(action)
if not subject or not template:
return
message = template.format(target=report["target_type"])
if reason:
message = f"{message} Reason: {reason}."
enforcement.notify_subject(subject["uid"], message)
def _notify_reporter(report: dict, action: str) -> None:
reporter_uid = report.get("reporter_uid") or ""
if not reporter_uid or reporter_uid == SYSTEM_ACTOR:
return
if not get_table("users").find_one(uid=reporter_uid):
return
verb = "dismissed" if action == "dismiss" else "actioned"
create_notification(
reporter_uid,
"moderation",
f"Your report on a {report['target_type']} was {verb}.",
reporter_uid,
"/reports/mine",
)
View Git Blame Copy Permalink