import logging
from typing import Annotated
from fastapi import APIRouter, Request, Form
from devplacepy.models import AdminRoleForm, AdminPasswordForm, AdminSettingsForm
from fastapi.responses import HTMLResponse, RedirectResponse
from devplacepy.database import get_table, build_pagination, get_post_counts_by_user_uids, get_news_images_by_uids, clear_settings_cache
from devplacepy.templating import templates
from devplacepy.utils import require_admin, hash_password, generate_uid, time_ago, clear_user_cache
from devplacepy.seo import base_seo_context, site_url, website_schema
logger = logging.getLogger(__name__)
router = APIRouter()
@router.get("", response_class=HTMLResponse)
async def admin_index(request: Request):
require_admin(request)
return RedirectResponse(url="/admin/users", status_code=302)
@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 templates.TemplateResponse(request, "admin_users.html", {
**seo_ctx,
"request": request,
"user": admin,
"users": page_users,
"pagination": pagination,
"admin_section": "users",
})
@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 RedirectResponse(url="/admin/users", status_code=302)
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 RedirectResponse(url="/admin/users", status_code=302)
@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 RedirectResponse(url="/admin/users", status_code=302)
@router.post("/users/{uid}/toggle")
async def admin_user_toggle(request: Request, uid: str):
admin = require_admin(request)
if uid == admin["uid"]:
return RedirectResponse(url="/admin/users", status_code=302)
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 RedirectResponse(url="/admin/users", status_code=302)
@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 templates.TemplateResponse(request, "admin_settings.html", {
**seo_ctx,
"request": request,
"user": admin,
"settings": all_settings,
"admin_section": "settings",
})
@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 templates.TemplateResponse(request, "admin_news.html", {
**seo_ctx,
"request": request,
"user": admin,
"articles": enriched,
"pagination": pagination,
"admin_section": "news",
})
@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 RedirectResponse(url="/admin/news", status_code=302)
@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 RedirectResponse(url="/admin/news", status_code=302)
@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 RedirectResponse(url="/admin/news", status_code=302)
@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 RedirectResponse(url="/admin/news", status_code=302)
@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 RedirectResponse(url="/admin/settings", status_code=302)
Reference in New Issue View Git Blame Copy Permalink