"""
Plain FastAPI admin UI backed by SQLAlchemy and Redis sessions.
This replaces the previous fastapi-admin runtime path, which depended on
Tortoise-oriented internals that do not match this project.
"""
import secrets
import uuid
from dataclasses import dataclass
from html import escape
from typing import Any
import aioredis
from fastapi import APIRouter, Depends, Form, Request
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from starlette.responses import HTMLResponse, RedirectResponse
from starlette.status import HTTP_303_SEE_OTHER, HTTP_401_UNAUTHORIZED
from app.core.config import get_settings
from app.database import get_db
from app.models import Item, Session, Tryout
from app.services.ai_generation import (
SUPPORTED_MODELS,
generate_question,
get_ai_stats,
validate_ai_model,
)
from app.services.irt_calibration import get_calibration_status
settings = get_settings()
router = APIRouter(prefix="/admin", tags=["admin-web"])
SESSION_COOKIE = "access_token"
SESSION_PREFIX = "admin:session:"
_admin_redis = None
@dataclass
class AdminPrincipal:
username: str
async def configure_admin_web() -> None:
global _admin_redis
if _admin_redis is not None:
return
if not settings.ADMIN_USERNAME or not settings.ADMIN_PASSWORD:
raise RuntimeError("ENABLE_ADMIN=true requires ADMIN_USERNAME and ADMIN_PASSWORD to be set.")
_admin_redis = aioredis.from_url(
settings.REDIS_URL,
encoding="utf-8",
decode_responses=True,
)
async def shutdown_admin_web() -> None:
global _admin_redis
if _admin_redis is None:
return
try:
await _admin_redis.close()
finally:
_admin_redis = None
async def _current_admin(request: Request) -> AdminPrincipal | None:
if _admin_redis is None:
return None
token = request.cookies.get(SESSION_COOKIE)
if not token:
return None
username = await _admin_redis.get(f"{SESSION_PREFIX}{token}")
if not username:
return None
return AdminPrincipal(username=str(username))
def _login_redirect() -> RedirectResponse:
return RedirectResponse(url="/admin/login", status_code=HTTP_303_SEE_OTHER)
def _dashboard_redirect() -> RedirectResponse:
return RedirectResponse(url="/admin/dashboard", status_code=HTTP_303_SEE_OTHER)
def _render_auth_page(
request: Request,
title: str,
subtitle: str,
body: str,
status_code: int = 200,
) -> HTMLResponse:
remember_me_checked = "checked" if request.cookies.get("remember_me") == "on" else ""
html = f"""
{escape(title)}
{escape(title)}
{escape(subtitle)}
{body.replace("__REMEMBER_ME_CHECKED__", remember_me_checked)}
"""
return HTMLResponse(html, status_code=status_code)
def _render_admin_page(title: str, page_title: str, body: str) -> HTMLResponse:
html = f"""
{escape(title)}
{escape(page_title)}
{body}
"""
return HTMLResponse(html)
def _table(headers: list[str], rows: list[list[Any]]) -> str:
head = "".join(f"{escape(str(header))} | " for header in headers)
body_rows = []
for row in rows:
cols = "".join(f"{escape(str(value))} | " for value in row)
body_rows.append(f"{cols}
")
body = "".join(body_rows) or f"| No data |
"
return f""
@router.get("", include_in_schema=False)
@router.get("/", include_in_schema=False)
async def admin_root(request: Request):
admin = await _current_admin(request)
if admin:
return _dashboard_redirect()
return _login_redirect()
@router.get("/login", include_in_schema=False)
async def login_view(request: Request):
admin = await _current_admin(request)
if admin:
return _dashboard_redirect()
body = """
Direct environment-backed admin access.
"""
return _render_auth_page(
request,
"Admin Login",
"Use the configured admin credentials to access the dashboard.",
body,
)
@router.post("/login", include_in_schema=False)
async def login_submit(
request: Request,
username: str = Form(...),
password: str = Form(...),
remember_me: str | None = Form(None),
):
if not (
secrets.compare_digest(username, settings.ADMIN_USERNAME)
and secrets.compare_digest(password, settings.ADMIN_PASSWORD)
):
body = f"""
Invalid username or password.
"""
return _render_auth_page(
request,
"Admin Login",
"Use the configured admin credentials to access the dashboard.",
body,
status_code=HTTP_401_UNAUTHORIZED,
)
expire = settings.ADMIN_SESSION_EXPIRE_SECONDS
response = _dashboard_redirect()
if remember_me == "on":
expire = max(expire, 3600 * 24 * 30)
response.set_cookie("remember_me", "on", expires=expire, path="/admin")
else:
response.delete_cookie("remember_me", path="/admin")
token = uuid.uuid4().hex
response.set_cookie(
SESSION_COOKIE,
token,
expires=expire,
path="/admin",
httponly=True,
samesite="lax",
)
await _admin_redis.set(f"{SESSION_PREFIX}{token}", settings.ADMIN_USERNAME, ex=expire)
return response
@router.get("/logout", include_in_schema=False)
async def logout(request: Request):
token = request.cookies.get(SESSION_COOKIE)
if token and _admin_redis is not None:
await _admin_redis.delete(f"{SESSION_PREFIX}{token}")
response = _login_redirect()
response.delete_cookie(SESSION_COOKIE, path="/admin")
response.delete_cookie("remember_me", path="/admin")
return response
@router.get("/password", include_in_schema=False)
async def password_view(request: Request):
admin = await _current_admin(request)
if not admin:
return _login_redirect()
body = f"""
Signed in as {escape(admin.username)}.
Password changes are disabled in the UI for this deployment.
Update ADMIN_PASSWORD in the server environment, then restart the app.
Session expiry is currently set to {settings.ADMIN_SESSION_EXPIRE_SECONDS} seconds.
Back to dashboard
"""
return _render_auth_page(
request,
"Password Management",
"Runtime password rotation is intentionally disabled.",
body,
)
@router.post("/password", include_in_schema=False)
async def password_submit(
request: Request,
old_password: str = Form(...),
new_password: str = Form(...),
re_new_password: str = Form(...),
):
_ = (old_password, new_password, re_new_password)
admin = await _current_admin(request)
if not admin:
return _login_redirect()
body = """
Password rotation via UI is disabled.
Update ADMIN_PASSWORD in the server environment, then restart the app.
Back to dashboard
"""
return _render_auth_page(
request,
"Password Management",
"Runtime password rotation is intentionally disabled.",
body,
status_code=400,
)
@router.get("/dashboard", include_in_schema=False)
async def dashboard_view(request: Request, db: AsyncSession = Depends(get_db)):
admin = await _current_admin(request)
if not admin:
return _login_redirect()
tryouts = await db.scalar(select(func.count()).select_from(Tryout)) or 0
items = await db.scalar(select(func.count()).select_from(Item)) or 0
sessions = await db.scalar(select(func.count()).select_from(Session)) or 0
completed_sessions = (
await db.scalar(select(func.count()).select_from(Session).where(Session.is_completed.is_(True)))
or 0
)
body = f"""
Signed in as {escape(admin.username)}.
Tryouts{tryouts}
Items{items}
Sessions{sessions}
Completed Sessions{completed_sessions}
Open AI Playground
"""
return _render_admin_page("IRT Bank Soal Admin", "Dashboard", body)
@router.get("/calibration-status", include_in_schema=False)
async def calibration_status_view(request: Request, db: AsyncSession = Depends(get_db)):
admin = await _current_admin(request)
if not admin:
return _login_redirect()
result = await db.execute(select(Tryout.tryout_id, Tryout.name, Tryout.website_id).order_by(Tryout.id))
tryouts = result.all()
rows = []
for tryout_id, name, website_id in tryouts:
status = await get_calibration_status(tryout_id, website_id, db)
rows.append(
[
tryout_id,
name,
status["total_items"],
status["calibrated_items"],
f'{status["calibration_percentage"]:.2f}%',
"Yes" if status["ready_for_irt"] else "No",
]
)
body = _table(
["Tryout ID", "Name", "Total Items", "Calibrated", "Calibration %", "Ready for IRT"],
rows,
)
return _render_admin_page("Calibration Status", "Calibration Status", body)
@router.get("/item-statistics", include_in_schema=False)
async def item_statistics_view(request: Request, db: AsyncSession = Depends(get_db)):
admin = await _current_admin(request)
if not admin:
return _login_redirect()
result = await db.execute(select(Item.level).distinct())
levels = result.scalars().all()
rows = []
for level in levels:
item_result = await db.execute(select(Item).where(Item.level == level).order_by(Item.slot).limit(10))
items = item_result.scalars().all()
total_responses = sum(item.calibration_sample_size or 0 for item in items)
calibrated_count = sum(1 for item in items if item.calibrated)
calibration_percentage = (calibrated_count / len(items) * 100) if items else 0
avg_correctness = sum(item.ctt_p or 0 for item in items) / len(items) if items else 0
rows.append(
[
level,
len(items),
calibrated_count,
f"{calibration_percentage:.2f}%",
total_responses,
f"{avg_correctness:.4f}",
]
)
body = _table(
["Level", "Total Items", "Calibrated", "Calibration %", "Responses", "Avg Correctness"],
rows,
)
return _render_admin_page("Item Statistics", "Item Statistics", body)
@router.get("/session-overview", include_in_schema=False)
async def session_overview_view(request: Request, db: AsyncSession = Depends(get_db)):
admin = await _current_admin(request)
if not admin:
return _login_redirect()
result = await db.execute(select(Session).order_by(Session.created_at.desc()).limit(50))
sessions = result.scalars().all()
rows = [
[
session.session_id,
session.wp_user_id,
session.tryout_id,
"Yes" if session.is_completed else "No",
session.scoring_mode_used,
session.total_benar,
session.NM,
session.NN,
session.theta,
]
for session in sessions
]
body = _table(
["Session ID", "WP User", "Tryout", "Completed", "Mode", "Benar", "NM", "NN", "Theta"],
rows,
)
return _render_admin_page("Session Overview", "Session Overview", body)
def _ai_form_body(
key_configured: bool,
stats: dict[str, Any],
error: str | None = None,
result: dict[str, Any] | None = None,
basis_item_id: str = "",
target_level: str = "mudah",
ai_model: str = settings.OPENROUTER_MODEL_QWEN,
) -> str:
error_html = f'{escape(error)}
' if error else ""
options_html = "".join(
f''
for model, label in SUPPORTED_MODELS.items()
)
result_html = ""
if result:
options = result.get("options") or {}
result_html = f"""
Preview Result
Model: {escape(result.get("ai_model", ""))}
Stem:
{escape(result.get("stem", ""))}
Options:
{_table(["Key", "Text"], [[key, value] for key, value in options.items()])}
Correct: {escape(result.get("correct", ""))}
Explanation:
{escape(result.get("explanation", ""))}
"""
return f"""
OpenRouter key configured: {"Yes" if key_configured else "No"}
Total AI-generated items: {stats.get("total_ai_items", 0)}
{error_html}
This is a preview only. Save-to-database UI is not added yet.
{result_html}
"""
@router.get("/ai-playground", include_in_schema=False)
async def ai_playground_view(request: Request, db: AsyncSession = Depends(get_db)):
admin = await _current_admin(request)
if not admin:
return _login_redirect()
stats = await get_ai_stats(db)
body = _ai_form_body(bool(settings.OPENROUTER_API_KEY), stats)
return _render_admin_page("AI Playground", "AI Playground", body)
@router.post("/ai-playground", include_in_schema=False)
async def ai_playground_submit(
request: Request,
db: AsyncSession = Depends(get_db),
basis_item_id: int = Form(...),
target_level: str = Form(...),
ai_model: str = Form(...),
):
admin = await _current_admin(request)
if not admin:
return _login_redirect()
stats = await get_ai_stats(db)
if not settings.OPENROUTER_API_KEY:
body = _ai_form_body(
False,
stats,
error="OPENROUTER_API_KEY is not configured in the environment.",
basis_item_id=str(basis_item_id),
target_level=target_level,
ai_model=ai_model,
)
return _render_admin_page("AI Playground", "AI Playground", body)
if target_level not in {"mudah", "sulit"}:
body = _ai_form_body(
True,
stats,
error="Target level must be mudah or sulit.",
basis_item_id=str(basis_item_id),
target_level=target_level,
ai_model=ai_model,
)
return _render_admin_page("AI Playground", "AI Playground", body)
if not validate_ai_model(ai_model):
body = _ai_form_body(
True,
stats,
error="Unsupported AI model.",
basis_item_id=str(basis_item_id),
target_level=target_level,
ai_model=ai_model,
)
return _render_admin_page("AI Playground", "AI Playground", body)
result = await db.execute(select(Item).where(Item.id == basis_item_id))
basis_item = result.scalar_one_or_none()
if not basis_item:
body = _ai_form_body(
True,
stats,
error=f"Basis item not found: {basis_item_id}",
basis_item_id=str(basis_item_id),
target_level=target_level,
ai_model=ai_model,
)
return _render_admin_page("AI Playground", "AI Playground", body)
if basis_item.level != "sedang":
body = _ai_form_body(
True,
stats,
error=f"Basis item must be sedang level, got: {basis_item.level}",
basis_item_id=str(basis_item_id),
target_level=target_level,
ai_model=ai_model,
)
return _render_admin_page("AI Playground", "AI Playground", body)
generated = await generate_question(
basis_item=basis_item,
target_level=target_level,
ai_model=ai_model,
)
if not generated:
body = _ai_form_body(
True,
stats,
error="AI generation failed. Check OPENROUTER_API_KEY, model availability, and server logs.",
basis_item_id=str(basis_item_id),
target_level=target_level,
ai_model=ai_model,
)
return _render_admin_page("AI Playground", "AI Playground", body)
body = _ai_form_body(
True,
stats,
result={
"stem": generated.stem,
"options": generated.options,
"correct": generated.correct,
"explanation": generated.explanation or "",
"ai_model": ai_model,
},
basis_item_id=str(basis_item_id),
target_level=target_level,
ai_model=ai_model,
)
return _render_admin_page("AI Playground", "AI Playground", body)
@router.get("/tryout/list", include_in_schema=False)
@router.get("/item/list", include_in_schema=False)
@router.get("/user/list", include_in_schema=False)
@router.get("/session/list", include_in_schema=False)
@router.get("/tryoutstats/list", include_in_schema=False)
async def legacy_admin_paths(request: Request):
admin = await _current_admin(request)
if not admin:
return _login_redirect()
return _dashboard_redirect()