# retoor <retoor@molodetz.nl>
import json
import logging
from typing import Annotated
from fastapi import APIRouter, Request, Form
from devplacepy.models import (
AdminRoleForm,
AdminPasswordForm,
AdminSettingsForm,
NotificationDefaultForm,
)
from fastapi.responses import HTMLResponse, JSONResponse
from devplacepy.database import (
get_table,
build_pagination,
get_post_counts_by_user_uids,
get_news_images_by_uids,
clear_settings_cache,
get_platform_analytics,
get_deleted_media,
soft_delete,
restore,
purge,
list_deleted,
count_deleted,
restore_event,
purge_event,
get_users_by_uids,
resolve_object_url,
SOFT_DELETE_TABLES,
_now_iso,
NOTIFICATION_TYPES,
NOTIFICATION_CHANNELS,
get_notification_default,
set_notification_default,
)
from devplacepy.attachments import delete_attachment, _unlink_attachment_files
from devplacepy.project_files import _unlink_blob
from devplacepy.utils import (
require_admin,
hash_password,
generate_uid,
time_ago,
clear_user_cache,
get_current_user,
is_admin,
not_found,
)
from devplacepy.seo import base_seo_context, site_url, website_schema
from devplacepy.responses import respond, action_result
from devplacepy.schemas import (
AdminUsersOut,
AdminNewsOut,
AdminSettingsOut,
AdminNotificationsOut,
AdminMediaOut,
AdminTrashOut,
GatewayUsageOut,
AuditLogOut,
AuditEventOut,
)
from devplacepy.utils import not_found
from devplacepy.database import get_users_by_uids
from devplacepy.services.audit import query as audit_query
from devplacepy.services.audit import record as audit
from urllib.parse import urlencode
from devplacepy.services.manager import service_manager
from devplacepy.services.openai_gateway.analytics import (
build_analytics,
build_user_usage,
)
from devplacepy.services.openai_gateway.usage import pricing_from_cfg
logger = logging.getLogger(__name__)
router = APIRouter()
def _parse_metadata(raw: str | dict | None) -> dict | None:
if not raw:
return None
if isinstance(raw, dict):
return raw
try:
return json.loads(raw)
except (ValueError, TypeError):
return {"raw": str(raw)}
@router.get("", response_class=HTMLResponse)
async def admin_index(request: Request):
require_admin(request)
return action_result(request, "/admin/users")
@router.get("/analytics")
async def admin_analytics(request: Request, top_n: int = 10):
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)
return JSONResponse(get_platform_analytics(top_n))
@router.get("/ai-usage", response_class=HTMLResponse)
async def admin_ai_usage(request: Request):
admin = require_admin(request)
base = site_url(request)
seo_ctx = base_seo_context(
request,
title="AI usage - Admin",
description="AI gateway token usage, cost, latency, and reliability metrics.",
robots="noindex,nofollow",
breadcrumbs=[
{"name": "Home", "url": "/feed"},
{"name": "Admin", "url": "/admin"},
{"name": "AI usage", "url": "/admin/ai-usage"},
],
schemas=[website_schema(base)],
)
return respond(
request,
"ai_usage.html",
{
**seo_ctx,
"request": request,
"user": admin,
"admin_section": "ai-usage",
},
model=GatewayUsageOut,
)
@router.get("/ai-usage/data")
async def admin_ai_usage_json(request: Request, hours: int = 48, top_n: int = 10):
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_analytics(hours, top_n=top_n, pricing=pricing))
@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")
@router.get("/media", response_class=HTMLResponse)
async def admin_media(request: Request, page: int = 1):
admin = require_admin(request)
media, pagination = get_deleted_media(page)
base = site_url(request)
seo_ctx = base_seo_context(
request,
title="Media - Admin",
description="Restore or permanently remove soft-deleted media.",
robots="noindex,nofollow",
breadcrumbs=[
{"name": "Home", "url": "/feed"},
{"name": "Admin", "url": "/admin"},
{"name": "Media", "url": "/admin/media"},
],
schemas=[website_schema(base)],
)
return respond(
request,
"admin_media.html",
{
**seo_ctx,
"request": request,
"user": admin,
"media": media,
"pagination": pagination,
"admin_section": "media",
},
model=AdminMediaOut,
)
@router.post("/media/{uid}/purge")
async def admin_media_purge(request: Request, uid: str):
admin = require_admin(request)
att = get_table("attachments").find_one(uid=uid)
delete_attachment(uid)
logger.info(f"Admin {admin['username']} purged media {uid}")
audit.record(
request,
"attachment.delete",
user=admin,
target_type="attachment",
target_uid=uid,
target_label=att.get("filename") if att else None,
metadata={"purge": True},
summary=f"admin {admin['username']} purged attachment {att.get('filename') if att else uid}",
links=[audit.attachment_link(uid, att.get("filename") if att else None)],
)
return action_result(request, "/admin/media")
TRASH_TABLES = [
{"key": "posts", "label": "Posts", "type": "post"},
{"key": "comments", "label": "Comments", "type": "comment"},
{"key": "gists", "label": "Gists", "type": "gist"},
{"key": "projects", "label": "Projects", "type": "project"},
{"key": "news", "label": "News", "type": "news"},
{"key": "project_files", "label": "Project files", "type": None},
{"key": "attachments", "label": "Attachments", "type": None},
]
_TRASH_KEYS = {entry["key"] for entry in TRASH_TABLES}
_TRASH_TYPE = {entry["key"]: entry["type"] for entry in TRASH_TABLES}
def _trash_label(table: str, row: dict) -> str:
for field in ("title", "name", "path", "original_filename", "question", "content"):
value = row.get(field)
if value:
return str(value)[:80]
return row.get("slug") or row.get("uid", "")
def _trash_target_url(table: str, row: dict) -> str | None:
target_type = _TRASH_TYPE.get(table)
if target_type:
return resolve_object_url(target_type, row["uid"])
return None
@router.get("/trash", response_class=HTMLResponse)
async def admin_trash(request: Request, table: str = "posts", page: int = 1):
admin = require_admin(request)
if table not in _TRASH_KEYS:
table = "posts"
rows, pagination = list_deleted(table, page)
actor_ids = [r.get("deleted_by") for r in rows if r.get("deleted_by")]
actors = get_users_by_uids([a for a in actor_ids if a]) if actor_ids else {}
items = []
for row in rows:
actor = actors.get(row.get("deleted_by"))
items.append(
{
"uid": row["uid"],
"table": table,
"label": _trash_label(table, row),
"deleted_at": row.get("deleted_at"),
"deleted_by": actor["username"] if actor else row.get("deleted_by"),
"target_url": _trash_target_url(table, row),
}
)
tabs = [
{**entry, "count": count_deleted(entry["key"]), "active": entry["key"] == table}
for entry in TRASH_TABLES
]
base = site_url(request)
seo_ctx = base_seo_context(
request,
title="Trash - Admin",
description="Restore or permanently remove soft-deleted content.",
breadcrumbs=[
{"name": "Home", "url": "/feed"},
{"name": "Admin", "url": "/admin"},
{"name": "Trash", "url": "/admin/trash"},
],
schemas=[website_schema(base)],
)
return respond(
request,
"admin_trash.html",
{
**seo_ctx,
"request": request,
"user": admin,
"items": items,
"table": table,
"tables": tabs,
"pagination": pagination,
"admin_section": "trash",
},
model=AdminTrashOut,
)
@router.post("/trash/{table}/{uid}/restore")
async def admin_trash_restore(request: Request, table: str, uid: str):
admin = require_admin(request)
if table not in _TRASH_KEYS:
raise not_found("Unknown trash table")
row = get_table(table).find_one(uid=uid)
if row and row.get("deleted_at"):
restored = restore_event(row["deleted_at"])
logger.info(
f"Admin {admin['username']} restored {table} {uid} ({restored} rows)"
)
return action_result(request, f"/admin/trash?table={table}")
@router.post("/trash/{table}/{uid}/purge")
async def admin_trash_purge(request: Request, table: str, uid: str):
admin = require_admin(request)
if table not in _TRASH_KEYS:
raise not_found("Unknown trash table")
row = get_table(table).find_one(uid=uid)
if row and row.get("deleted_at"):
purged = purge_event(row["deleted_at"])
for purged_table, purged_rows in purged:
if purged_table == "attachments":
for attachment in purged_rows:
_unlink_attachment_files(attachment)
elif purged_table == "project_files":
for node in purged_rows:
if node.get("is_binary"):
_unlink_blob(node)
logger.info(f"Admin {admin['username']} purged {table} {uid}")
return action_result(request, f"/admin/trash?table={table}")
@router.post("/ai-quota/reset-guests")
async def admin_reset_guest_ai_quota(request: Request):
admin = require_admin(request)
devii = service_manager.get_service("devii")
removed = devii.reset_guest_quotas() if devii is not None else 0
logger.info(
f"Admin {admin['username']} reset all guest AI quotas ({removed} ledger rows)"
)
audit.record(
request,
"admin.ai_quota.reset_guests",
user=admin,
metadata={"rows_removed": removed},
summary=f"admin {admin['username']} reset all guest AI quotas",
)
return action_result(request, "/admin/ai-usage")
@router.post("/ai-quota/reset-all")
async def admin_reset_all_ai_quota(request: Request):
admin = require_admin(request)
devii = service_manager.get_service("devii")
removed = devii.reset_all_quotas() if devii is not None else 0
logger.info(
f"Admin {admin['username']} reset ALL AI quotas ({removed} ledger rows)"
)
audit.record(
request,
"admin.ai_quota.reset_all",
user=admin,
metadata={"rows_removed": removed},
summary=f"admin {admin['username']} reset all AI quotas",
)
return action_result(request, "/admin/ai-usage")
@router.get("/settings", response_class=HTMLResponse)
async def admin_settings(request: Request):
admin = require_admin(request)
settings = get_table("site_settings")
all_settings = {s["key"]: s["value"] for s in settings.all()}
base = site_url(request)
seo_ctx = base_seo_context(
request,
title="Settings - Admin",
description="Manage DevPlace site settings.",
robots="noindex,nofollow",
breadcrumbs=[
{"name": "Home", "url": "/feed"},
{"name": "Admin", "url": "/admin"},
{"name": "Settings", "url": "/admin/settings"},
],
schemas=[website_schema(base)],
)
return respond(
request,
"admin_settings.html",
{
**seo_ctx,
"request": request,
"user": admin,
"settings": all_settings,
"admin_section": "settings",
},
model=AdminSettingsOut,
)
def _notification_defaults_view() -> list:
defaults = []
for entry in NOTIFICATION_TYPES:
defaults.append(
{
"key": entry["key"],
"label": entry["label"],
"description": entry["description"],
"in_app": get_notification_default(entry["key"], "in_app"),
"push": get_notification_default(entry["key"], "push"),
}
)
return defaults
@router.get("/notifications", response_class=HTMLResponse)
async def admin_notifications(request: Request):
admin = require_admin(request)
base = site_url(request)
seo_ctx = base_seo_context(
request,
title="Notifications - Admin",
description="Manage default notification settings for all users.",
robots="noindex,nofollow",
breadcrumbs=[
{"name": "Home", "url": "/feed"},
{"name": "Admin", "url": "/admin"},
{"name": "Notifications", "url": "/admin/notifications"},
],
schemas=[website_schema(base)],
)
return respond(
request,
"admin_notifications.html",
{
**seo_ctx,
"request": request,
"user": admin,
"notification_defaults": _notification_defaults_view(),
"admin_section": "notifications",
},
model=AdminNotificationsOut,
)
@router.post("/notifications")
async def admin_set_notification_default(
request: Request, data: Annotated[NotificationDefaultForm, Form()]
):
admin = require_admin(request)
valid_types = {entry["key"] for entry in NOTIFICATION_TYPES}
if data.notification_type not in valid_types or data.channel not in NOTIFICATION_CHANNELS:
from devplacepy.responses import json_error
return json_error(400, "Unknown notification type or channel")
set_notification_default(data.notification_type, data.channel, data.value)
logger.info(
f"Admin {admin['username']} set default {data.notification_type}/{data.channel} "
f"to {'on' if data.value else 'off'}"
)
audit.record(
request,
"admin.notification.default",
user=admin,
target_type="setting",
target_uid=f"notif_default_{data.notification_type}_{data.channel}",
old_value=0 if data.value else 1,
new_value=1 if data.value else 0,
summary=(
f"admin {admin['username']} set default {data.channel} "
f"{data.notification_type} notifications to {'on' if data.value else 'off'}"
),
metadata={"type": data.notification_type, "channel": data.channel},
)
return action_result(
request,
"/admin/notifications",
data={
"notification_type": data.notification_type,
"channel": data.channel,
"value": data.value,
},
)
@router.get("/audit-log", response_class=HTMLResponse)
async def admin_audit_log(
request: Request,
page: int = 1,
event_key: str = "",
category: str = "",
actor_role: str = "",
actor_uid: str = "",
origin: str = "",
result: str = "",
q: str = "",
date_from: str = "",
date_to: str = "",
):
admin = require_admin(request)
filters = {
"event_key": event_key,
"category": category,
"actor_role": actor_role,
"actor_uid": actor_uid,
"origin": origin,
"result": result,
"q": q,
"date_from": date_from,
"date_to": date_to,
}
active = {key: value for key, value in filters.items() if value}
rows, pagination = audit_query.list_events(filters, page)
user_uids = [
row["actor_uid"]
for row in rows
if row.get("actor_kind") == "user" and row.get("actor_uid")
]
users_by_uid = get_users_by_uids(user_uids)
entries = []
for row in rows:
entry = dict(row)
entry["actor_user"] = users_by_uid.get(row.get("actor_uid"))
entry["time_ago"] = time_ago(row["created_at"]) if row.get("created_at") else ""
entry["metadata"] = _parse_metadata(row.get("metadata"))
entries.append(entry)
filter_qs = urlencode(active)
pagination_query = (filter_qs + "&") if filter_qs else ""
base = site_url(request)
seo_ctx = base_seo_context(
request,
title="Audit Log - Admin",
description="Platform audit trail of every state-changing action.",
robots="noindex,nofollow",
breadcrumbs=[
{"name": "Home", "url": "/feed"},
{"name": "Admin", "url": "/admin"},
{"name": "Audit Log", "url": "/admin/audit-log"},
],
schemas=[website_schema(base)],
)
return respond(
request,
"admin_audit_log.html",
{
**seo_ctx,
"request": request,
"user": admin,
"entries": entries,
"pagination": pagination,
"pagination_query": pagination_query,
"filters": active,
"options": audit_query.filter_options(),
"admin_section": "audit-log",
},
model=AuditLogOut,
)
@router.get("/audit-log/{uid}", response_class=HTMLResponse)
async def admin_audit_event(request: Request, uid: str):
admin = require_admin(request)
data = audit_query.get_event_with_links(uid)
if data is None:
raise not_found("Audit event not found")
event = dict(data["event"])
if event.get("actor_kind") == "user" and event.get("actor_uid"):
event["actor_user"] = get_users_by_uids([event["actor_uid"]]).get(
event["actor_uid"]
)
event["metadata"] = _parse_metadata(event.get("metadata"))
event["metadata_pretty"] = (
json.dumps(event["metadata"], indent=2) if event["metadata"] else ""
)
event["time_ago"] = (
time_ago(event["created_at"]) if event.get("created_at") else ""
)
base = site_url(request)
seo_ctx = base_seo_context(
request,
title="Audit Event - Admin",
description="A single audit event with related objects.",
robots="noindex,nofollow",
breadcrumbs=[
{"name": "Home", "url": "/feed"},
{"name": "Admin", "url": "/admin"},
{"name": "Audit Log", "url": "/admin/audit-log"},
{"name": "Event", "url": f"/admin/audit-log/{uid}"},
],
schemas=[website_schema(base)],
)
return respond(
request,
"admin_audit_event.html",
{
**seo_ctx,
"request": request,
"user": admin,
"event": event,
"links": data["links"],
"admin_section": "audit-log",
},
model=AuditEventOut,
)
@router.get("/news", response_class=HTMLResponse)
async def admin_news(request: Request, page: int = 1):
admin = require_admin(request)
news_table = get_table("news")
total = news_table.count(deleted_at=None)
pagination = build_pagination(page, total)
offset = (pagination["page"] - 1) * pagination["per_page"]
page_articles = list(
news_table.find(
deleted_at=None,
order_by=["-synced_at"],
_limit=pagination["per_page"],
_offset=offset,
)
)
images_by_news = get_news_images_by_uids([a["uid"] for a in page_articles])
enriched = []
for a in page_articles:
enriched.append(
{
"article": a,
"time_ago": time_ago(a["synced_at"]),
"synced_at": a.get("synced_at", ""),
"grade": a.get("grade", 0),
"has_image": a["uid"] in images_by_news,
}
)
base = site_url(request)
seo_ctx = base_seo_context(
request,
title="News - Admin",
description="Manage DevPlace news articles.",
robots="noindex,nofollow",
breadcrumbs=[
{"name": "Home", "url": "/feed"},
{"name": "Admin", "url": "/admin"},
{"name": "News", "url": "/admin/news"},
],
schemas=[website_schema(base)],
)
return respond(
request,
"admin_news.html",
{
**seo_ctx,
"request": request,
"user": admin,
"articles": enriched,
"pagination": pagination,
"admin_section": "news",
},
model=AdminNewsOut,
)
@router.post("/news/{uid}/toggle")
async def admin_news_toggle(request: Request, uid: str):
admin = require_admin(request)
news_table = get_table("news")
article = news_table.find_one(uid=uid)
if article:
current = article.get("featured", 0)
new_value = 0 if current else 1
news_table.update({"uid": uid, "featured": new_value}, ["uid"])
logger.info(
f"Admin {admin['username']} toggled news {uid} featured={'off' if current else 'on'}"
)
audit.record(
request,
"news.featured.toggle",
user=admin,
target_type="news",
target_uid=uid,
target_label=article.get("title"),
old_value=current,
new_value=new_value,
summary=f"admin {admin['username']} toggled featured for news {article.get('title')}",
links=[audit.target("news", uid, article.get("title"))],
)
return action_result(request, "/admin/news")
@router.post("/news/{uid}/publish")
async def admin_news_publish(request: Request, uid: str):
admin = require_admin(request)
news_table = get_table("news")
article = news_table.find_one(uid=uid)
if article:
current = article.get("status", "draft")
new_status = "draft" if current == "published" else "published"
news_table.update({"uid": uid, "status": new_status}, ["uid"])
logger.info(f"Admin {admin['username']} set news {uid} status={new_status}")
audit.record(
request,
"news.publish.toggle",
user=admin,
target_type="news",
target_uid=uid,
target_label=article.get("title"),
old_value=current,
new_value=new_status,
summary=f"admin {admin['username']} set news {article.get('title')} to {new_status}",
links=[audit.target("news", uid, article.get("title"))],
)
return action_result(request, "/admin/news")
@router.post("/news/{uid}/landing")
async def admin_news_landing(request: Request, uid: str):
admin = require_admin(request)
news_table = get_table("news")
article = news_table.find_one(uid=uid)
if article:
current = article.get("show_on_landing", 0)
new_value = 0 if current else 1
news_table.update({"uid": uid, "show_on_landing": new_value}, ["uid"])
logger.info(
f"Admin {admin['username']} toggled news {uid} landing={'on' if not current else 'off'}"
)
audit.record(
request,
"news.landing.toggle",
user=admin,
target_type="news",
target_uid=uid,
target_label=article.get("title"),
old_value=current,
new_value=new_value,
summary=f"admin {admin['username']} toggled landing visibility for news {article.get('title')}",
links=[audit.target("news", uid, article.get("title"))],
)
return action_result(request, "/admin/news")
@router.post("/news/{uid}/delete")
async def admin_news_delete(request: Request, uid: str):
admin = require_admin(request)
news_table = get_table("news")
article = news_table.find_one(uid=uid, deleted_at=None)
if article:
stamp = _now_iso()
image_count = soft_delete("news_images", admin["uid"], stamp=stamp, news_uid=uid)
soft_delete("news", admin["uid"], stamp=stamp, uid=uid)
logger.info(f"Admin {admin['username']} soft-deleted news {uid}")
audit.record(
request,
"news.delete",
user=admin,
target_type="news",
target_uid=uid,
target_label=article.get("title"),
metadata={"image_count": image_count},
summary=f"admin {admin['username']} deleted news article {article.get('title')}",
links=[audit.target("news", uid, article.get("title"))],
)
return action_result(request, "/admin/news")
@router.post("/settings")
async def admin_settings_save(
request: Request, data: Annotated[AdminSettingsForm, Form()]
):
admin = require_admin(request)
settings = get_table("site_settings")
changes = []
for key, value in data.model_dump().items():
existing = settings.find_one(key=key)
if existing:
if value == "":
continue
old_value = existing.get("value")
if str(old_value) == str(value):
continue
settings.update({"id": existing["id"], "key": key, "value": value}, ["id"])
changes.append((key, old_value, value))
else:
settings.insert({"uid": generate_uid(), "key": key, "value": value})
changes.append((key, None, value))
clear_settings_cache()
logger.info(f"Admin {admin['username']} updated settings")
for key, old_value, new_value in changes:
audit.record(
request,
"admin.setting.update",
user=admin,
target_type="setting",
target_uid=key,
target_label=key,
old_value=old_value,
new_value=new_value,
summary=f"admin {admin['username']} updated setting {key}",
links=[audit.setting(key)],
)
return action_result(request, "/admin/settings")
Reference in New Issue View Git Blame Copy Permalink