chore: remove stale agent reports and add PYTHONDONTWRITEBYTECODE export to Makefile
Delete four stale SEO and style agent report JSON/MD files from agents/reports/ that had zero findings or were superseded. Export PYTHONDONTWRITEBYTECODE=1 in the Makefile so all make targets suppress bytecode generation, and add a `tree` target that lists the repository file tree via git ls-tree. Update AGENTS.md and CLAUDE.md documentation to reflect that routers now form a directory tree mirroring URL paths and that bytecode writing is disabled project-wide.
This commit is contained in:
@@ -0,0 +1,220 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
import logging
|
||||
from typing import Annotated
|
||||
from fastapi import APIRouter, Request, Form
|
||||
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,
|
||||
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
|
||||
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
|
||||
|
||||
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):
|
||||
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
|
||||
return JSONResponse(build_user_usage(uid, hours=hours, pricing=pricing))
|
||||
|
||||
|
||||
@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, Form()]
|
||||
):
|
||||
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)
|
||||
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, Form()]
|
||||
):
|
||||
admin = require_admin(request)
|
||||
users = get_table("users")
|
||||
target_user = users.find_one(uid=uid)
|
||||
users.update({"uid": uid, "password_hash": hash_password(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 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)
|
||||
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)"
|
||||
)
|
||||
target_user = get_table("users").find_one(uid=uid)
|
||||
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
Block a user