devplacepy/devplacepy/routers/admin/users.py

# retoor <retoor@molodetz.nl>

import logging
from typing import Annotated
from fastapi import Depends, APIRouter, Request
from fastapi.responses import HTMLResponse, JSONResponse
from devplacepy.models import AdminRoleForm, AdminPasswordForm, BanForm, SuspensionForm
from devplacepy.database import (
    get_table,
    build_pagination,
    get_post_counts_by_user_uids,
    invalidate_admins_cache,
    is_account_active,
)
from devplacepy.utils import (
    require_admin,
    hash_password_async,
    clear_user_cache,
)
from devplacepy.seo import base_seo_context, site_url, website_schema
from devplacepy.responses import respond, action_result
from devplacepy.schemas import AdminUsersOut, UserAiUsageOut
from devplacepy.services.audit import record as audit
from devplacepy.services.moderation import enforcement, queue
from devplacepy.services.manager import service_manager
from devplacepy.services.openai_gateway.analytics import build_user_usage
from devplacepy.services.openai_gateway.usage import pricing_from_cfg
from devplacepy.dependencies import json_or_form
from devplacepy.routers.admin._shared import deny_senior, is_senior_admin

logger = logging.getLogger(__name__)
router = APIRouter()


@router.get("/users/{uid}/ai-usage")
async def admin_user_ai_usage(request: Request, uid: str, hours: int = 24):
    require_admin(request)
    service = service_manager.get_service("openai")
    pricing = pricing_from_cfg(service.get_config()) if service is not None else None
    usage = build_user_usage(uid, hours=hours, pricing=pricing)
    return JSONResponse(UserAiUsageOut(**usage).model_dump())

@router.get("/users", response_class=HTMLResponse)
async def admin_users(request: Request, page: int = 1):
    admin = require_admin(request)
    users_table = get_table("users")
    total = users_table.count()
    pagination = build_pagination(page, total)
    offset = (pagination["page"] - 1) * pagination["per_page"]
    page_users = list(
        users_table.find(
            order_by=["-created_at"], _limit=pagination["per_page"], _offset=offset
        )
    )
    post_counts = get_post_counts_by_user_uids([u["uid"] for u in page_users])
    for u in page_users:
        u["posts_count"] = post_counts.get(u["uid"], 0)
    base = site_url(request)
    seo_ctx = base_seo_context(
        request,
        title="Users - Admin",
        description="Manage DevPlace users.",
        robots="noindex,nofollow",
        breadcrumbs=[
            {"name": "Home", "url": "/feed"},
            {"name": "Admin", "url": "/admin"},
            {"name": "Users", "url": "/admin/users"},
        ],
        schemas=[website_schema(base)],
    )
    return respond(
        request,
        "admin_users.html",
        {
            **seo_ctx,
            "request": request,
            "user": admin,
            "users": page_users,
            "pagination": pagination,
            "admin_section": "users",
        },
        model=AdminUsersOut,
    )

@router.post("/users/{uid}/role")
async def admin_user_role(
    request: Request, uid: str, data: Annotated[AdminRoleForm, Depends(json_or_form(AdminRoleForm))]
):
    admin = require_admin(request)
    role = data.role.capitalize()
    if uid == admin["uid"]:
        audit.record(
            request,
            "admin.user.role.change",
            user=admin,
            result="denied",
            target_type="user",
            target_uid=uid,
            target_label=admin.get("username"),
            summary=f"admin {admin['username']} cannot change their own role",
            links=[audit.target("user", uid, admin.get("username"))],
        )
        return action_result(request, "/admin/users")
    users = get_table("users")
    target_user = users.find_one(uid=uid)
    if is_senior_admin(admin, target_user):
        return deny_senior(request, admin, uid, target_user, "admin.user.role.change")
    old_role = target_user.get("role") if target_user else None
    users.update({"uid": uid, "role": role}, ["uid"])
    clear_user_cache(uid)
    invalidate_admins_cache()
    logger.info(f"Admin {admin['username']} set user {uid} role to {role}")
    audit.record(
        request,
        "admin.user.role.change",
        user=admin,
        target_type="user",
        target_uid=uid,
        target_label=target_user.get("username") if target_user else None,
        old_value=old_role,
        new_value=role,
        summary=f"admin {admin['username']} changed role of {target_user.get('username') if target_user else uid} to {role}",
        links=[
            audit.target(
                "user", uid, target_user.get("username") if target_user else None
            )
        ],
    )
    return action_result(request, "/admin/users")

@router.post("/users/{uid}/password")
async def admin_user_password(
    request: Request, uid: str, data: Annotated[AdminPasswordForm, Depends(json_or_form(AdminPasswordForm))]
):
    admin = require_admin(request)
    users = get_table("users")
    target_user = users.find_one(uid=uid)
    if is_senior_admin(admin, target_user):
        return deny_senior(request, admin, uid, target_user, "admin.user.password.reset")
    users.update({"uid": uid, "password_hash": await hash_password_async(data.password)}, ["uid"])
    logger.info(f"Admin {admin['username']} changed password for user {uid}")
    audit.record(
        request,
        "admin.user.password.reset",
        user=admin,
        target_type="user",
        target_uid=uid,
        target_label=target_user.get("username") if target_user else None,
        summary=f"admin {admin['username']} reset the password of {target_user.get('username') if target_user else uid}",
        links=[
            audit.target(
                "user", uid, target_user.get("username") if target_user else None
            )
        ],
    )
    return action_result(request, "/admin/users")

@router.post("/users/{uid}/toggle")
async def admin_user_toggle(request: Request, uid: str):
    admin = require_admin(request)
    if uid == admin["uid"]:
        audit.record(
            request,
            "admin.user.active.disable",
            user=admin,
            result="denied",
            target_type="user",
            target_uid=uid,
            target_label=admin.get("username"),
            summary=f"admin {admin['username']} cannot disable their own account",
            links=[audit.target("user", uid, admin.get("username"))],
        )
        return action_result(request, "/admin/users")
    users = get_table("users")
    user = users.find_one(uid=uid)
    if is_senior_admin(admin, user):
        return deny_senior(request, admin, uid, user, "admin.user.active.disable")
    if user:
        new_state = not is_account_active(user)
        users.update({"uid": uid, "is_active": new_state}, ["uid"])
        clear_user_cache(uid)
        logger.info(
            f"Admin {admin['username']} {'disabled' if not new_state else 'enabled'} user {uid}"
        )
        audit.record(
            request,
            "admin.user.active.enable" if new_state else "admin.user.active.disable",
            user=admin,
            target_type="user",
            target_uid=uid,
            target_label=user.get("username"),
            new_value=1 if new_state else 0,
            summary=f"admin {admin['username']} {'enabled' if new_state else 'disabled'} user {user.get('username')}",
            links=[audit.target("user", uid, user.get("username"))],
        )
    return action_result(request, "/admin/users")

def _enforcement_target(request: Request, admin: dict, uid: str, event_key: str):
    if uid == admin["uid"]:
        audit.record(
            request,
            event_key,
            user=admin,
            result="denied",
            target_type="user",
            target_uid=uid,
            target_label=admin.get("username"),
            summary=f"admin {admin['username']} cannot enforce against their own account",
            links=[audit.target("user", uid, admin.get("username"))],
        )
        return None, action_result(request, "/admin/users")
    target_user = get_table("users").find_one(uid=uid)
    if not target_user:
        return None, action_result(request, "/admin/users")
    if is_senior_admin(admin, target_user):
        return None, deny_senior(request, admin, uid, target_user, event_key)
    return target_user, None


def _record_enforcement(
    request: Request, admin: dict, target_user: dict, event_key: str, metadata: dict
):
    logger.info(
        f"Admin {admin['username']} applied {event_key} to {target_user['username']}"
    )
    audit.record(
        request,
        event_key,
        user=admin,
        target_type="user",
        target_uid=target_user["uid"],
        target_label=target_user.get("username"),
        metadata=metadata,
        summary=f"{admin['username']} applied {event_key} to {target_user['username']}",
        links=[audit.target("user", target_user["uid"], target_user.get("username"))],
    )
    queue.record_action(
        report_uid="",
        actor_uid=admin["uid"],
        action=event_key.rsplit(".", 1)[-1],
        target_type="user",
        target_uid=target_user["uid"],
        subject_uid=target_user["uid"],
        reason=metadata.get("reason", ""),
        expires_at=metadata.get("expires_at", ""),
    )


@router.post("/users/{uid}/suspend")
async def admin_user_suspend(
    request: Request,
    uid: str,
    data: Annotated[SuspensionForm, Depends(json_or_form(SuspensionForm))],
):
    admin = require_admin(request)
    target_user, refusal = _enforcement_target(request, admin, uid, "moderation.suspend")
    if refusal is not None:
        return refusal
    expires_at = enforcement.suspend_user(target_user, data.duration_hours, data.reason)
    _record_enforcement(
        request,
        admin,
        target_user,
        "moderation.suspend",
        {"reason": data.reason, "expires_at": expires_at, "hours": data.duration_hours},
    )
    enforcement.notify_subject(
        uid,
        f"Your account is suspended until {expires_at}. Reason: {data.reason or 'policy violation'}.",
    )
    return action_result(request, "/admin/users")


@router.post("/users/{uid}/lift")
async def admin_user_lift(request: Request, uid: str):
    admin = require_admin(request)
    target_user, refusal = _enforcement_target(request, admin, uid, "moderation.lift")
    if refusal is not None:
        return refusal
    enforcement.lift_suspension(target_user)
    enforcement.unban_user(target_user)
    _record_enforcement(request, admin, target_user, "moderation.lift", {})
    enforcement.notify_subject(uid, "Your account restriction has been lifted.")
    return action_result(request, "/admin/users")


@router.post("/users/{uid}/ban")
async def admin_user_ban(
    request: Request,
    uid: str,
    data: Annotated[BanForm, Depends(json_or_form(BanForm))],
):
    admin = require_admin(request)
    target_user, refusal = _enforcement_target(request, admin, uid, "moderation.ban")
    if refusal is not None:
        return refusal
    enforcement.ban_user(target_user, data.reason)
    _record_enforcement(
        request, admin, target_user, "moderation.ban", {"reason": data.reason}
    )
    enforcement.notify_subject(
        uid, f"Your account has been closed. Reason: {data.reason or 'policy violation'}."
    )
    return action_result(request, "/admin/users")


@router.post("/users/{uid}/reset-ai-quota")
async def admin_user_reset_ai_quota(request: Request, uid: str):
    admin = require_admin(request)
    target_user = get_table("users").find_one(uid=uid)
    if is_senior_admin(admin, target_user):
        return deny_senior(request, admin, uid, target_user, "admin.user.ai_quota.reset")
    devii = service_manager.get_service("devii")
    removed = devii.reset_quota("user", uid) if devii is not None else 0
    logger.info(
        f"Admin {admin['username']} reset AI quota for user {uid} ({removed} ledger rows)"
    )
    audit.record(
        request,
        "admin.user.ai_quota.reset",
        user=admin,
        target_type="user",
        target_uid=uid,
        target_label=target_user.get("username") if target_user else None,
        metadata={"rows_removed": removed},
        summary=f"admin {admin['username']} reset the AI quota of {target_user.get('username') if target_user else uid}",
        links=[
            audit.target(
                "user", uid, target_user.get("username") if target_user else None
            )
        ],
    )
    return action_result(request, "/admin/users")