# 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
from devplacepy.database import (
get_table,
build_pagination,
get_post_counts_by_user_uids,
)
from devplacepy.utils import (
require_admin,
hash_password_async,
clear_user_cache,
get_current_user,
is_admin,
)
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.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
logger = logging.getLogger(__name__)
router = APIRouter()
def _seniority_key(u: dict) -> tuple[str, int]:
return (u.get("created_at") or "", u.get("id") or 0)
def _is_senior_admin(actor: dict, target: dict | None) -> bool:
if not target or target.get("role") != "Admin":
return False
if target.get("uid") == actor.get("uid"):
return False
return _seniority_key(target) < _seniority_key(actor)
def _deny_senior(request: Request, admin: dict, uid: str, target: dict, event_key: str):
audit.record(
request,
event_key,
user=admin,
result="denied",
target_type="user",
target_uid=uid,
target_label=target.get("username"),
summary=f"admin {admin['username']} cannot manage senior admin {target.get('username')}",
links=[audit.target("user", uid, target.get("username"))],
)
return action_result(request, "/admin/users")
@router.get("/users/{uid}/ai-usage")
async def admin_user_ai_usage(request: Request, uid: str, hours: int = 24):
user = get_current_user(request)
if not user:
return JSONResponse({"error": "Authentication required"}, status_code=401)
if not is_admin(user):
return JSONResponse({"error": "Admin access required"}, status_code=403)
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)
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 user.get("is_active", True)
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")
@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")
Reference in New Issue View Git Blame Copy Permalink