import logging
from typing import Annotated
from fastapi import APIRouter, Request, Form
from devplacepy.models import AdminRoleForm, AdminPasswordForm, AdminSettingsForm
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,
)
from devplacepy.attachments import delete_attachment
from devplacepy.utils import (
require_admin,
hash_password,
generate_uid,
time_ago,
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,
AdminNewsOut,
AdminSettingsOut,
AdminMediaOut,
GatewayUsageOut,
)
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()
@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.",
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_data(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.",
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"]:
return action_result(request, "/admin/users")
users = get_table("users")
users.update({"uid": uid, "role": role}, ["uid"])
clear_user_cache(uid)
logger.info(f"Admin {admin['username']} set user {uid} role to {role}")
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")
users.update({"uid": uid, "password_hash": hash_password(data.password)}, ["uid"])
logger.info(f"Admin {admin['username']} changed password for user {uid}")
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"]:
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}"
)
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)"
)
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.",
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)
delete_attachment(uid)
logger.info(f"Admin {admin['username']} purged media {uid}")
return action_result(request, "/admin/media")
@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)"
)
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)"
)
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.",
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,
)
@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()
pagination = build_pagination(page, total)
offset = (pagination["page"] - 1) * pagination["per_page"]
page_articles = list(
news_table.find(
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.",
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)
news_table.update({"uid": uid, "featured": 0 if current else 1}, ["uid"])
logger.info(
f"Admin {admin['username']} toggled news {uid} featured={'off' if current else 'on'}"
)
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}")
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)
news_table.update({"uid": uid, "show_on_landing": 0 if current else 1}, ["uid"])
logger.info(
f"Admin {admin['username']} toggled news {uid} landing={'on' if not current else 'off'}"
)
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)
if article:
images_table = get_table("news_images")
for img in images_table.find(news_uid=uid):
images_table.delete(id=img["id"])
news_table.delete(id=article["id"])
logger.info(f"Admin {admin['username']} deleted news {uid}")
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")
for key, value in data.model_dump().items():
existing = settings.find_one(key=key)
if existing:
if value == "":
continue
settings.update({"id": existing["id"], "key": key, "value": value}, ["id"])
else:
settings.insert({"uid": generate_uid(), "key": key, "value": value})
clear_settings_cache()
logger.info(f"Admin {admin['username']} updated settings")
return action_result(request, "/admin/settings")
Reference in New Issue View Git Blame Copy Permalink