import logging
from typing import Annotated
from datetime import datetime
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, db, 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:
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