464 lines
16 KiB
Python
464 lines
16 KiB
Python
"""
|
|
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, 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.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"""<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
|
<title>{escape(title)}</title>
|
|
<style>
|
|
body {{ margin: 0; min-height: 100vh; display: grid; place-items: center; background: linear-gradient(135deg, #f8fafc, #e2e8f0); font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif; color: #0f172a; }}
|
|
.panel {{ width: min(420px, calc(100vw - 32px)); background: rgba(255,255,255,0.96); border-radius: 18px; box-shadow: 0 18px 60px rgba(15, 23, 42, 0.14); padding: 28px; }}
|
|
h1 {{ margin: 0 0 8px; font-size: 28px; }}
|
|
p {{ margin: 0 0 20px; color: #475569; }}
|
|
label {{ display: block; font-size: 14px; font-weight: 600; margin: 14px 0 8px; }}
|
|
input {{ width: 100%; box-sizing: border-box; border: 1px solid #cbd5e1; border-radius: 10px; padding: 12px 14px; font-size: 15px; }}
|
|
.row {{ display: flex; align-items: center; gap: 10px; margin-top: 16px; color: #334155; font-size: 14px; }}
|
|
.row input {{ width: auto; }}
|
|
button {{ width: 100%; margin-top: 18px; border: 0; border-radius: 10px; padding: 12px 14px; background: #0f172a; color: #fff; font-size: 15px; font-weight: 600; cursor: pointer; }}
|
|
.error {{ margin: 0 0 16px; padding: 12px 14px; border-radius: 10px; background: #fef2f2; color: #991b1b; border: 1px solid #fecaca; }}
|
|
.muted {{ color: #64748b; font-size: 13px; margin-top: 14px; }}
|
|
a {{ color: #0f172a; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<main class="panel">
|
|
<h1>{escape(title)}</h1>
|
|
<p>{escape(subtitle)}</p>
|
|
{body.replace("__REMEMBER_ME_CHECKED__", remember_me_checked)}
|
|
</main>
|
|
</body>
|
|
</html>"""
|
|
return HTMLResponse(html, status_code=status_code)
|
|
|
|
|
|
def _render_admin_page(title: str, page_title: str, body: str) -> HTMLResponse:
|
|
html = f"""<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
|
<title>{escape(title)}</title>
|
|
<style>
|
|
body {{ margin: 0; font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif; background: #f5f7fb; color: #162033; }}
|
|
.layout {{ display: grid; grid-template-columns: 240px 1fr; min-height: 100vh; }}
|
|
.sidebar {{ background: #0f172a; color: #e2e8f0; padding: 24px 16px; }}
|
|
.sidebar h1 {{ font-size: 18px; margin: 0 0 24px; }}
|
|
.sidebar a {{ display: block; color: #cbd5e1; text-decoration: none; padding: 10px 12px; border-radius: 8px; margin-bottom: 8px; }}
|
|
.sidebar a:hover {{ background: #1e293b; color: #fff; }}
|
|
.content {{ padding: 32px; }}
|
|
.card {{ background: #fff; border-radius: 14px; padding: 24px; box-shadow: 0 8px 30px rgba(15, 23, 42, 0.08); }}
|
|
.grid {{ display: grid; grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); gap: 16px; margin-top: 20px; }}
|
|
.stat {{ padding: 18px; border: 1px solid #e2e8f0; border-radius: 12px; background: #f8fafc; }}
|
|
.stat strong {{ display: block; font-size: 26px; margin-top: 6px; }}
|
|
table {{ width: 100%; border-collapse: collapse; margin-top: 16px; background: #fff; border-radius: 12px; overflow: hidden; }}
|
|
th, td {{ padding: 12px 14px; border-bottom: 1px solid #e5e7eb; text-align: left; vertical-align: top; font-size: 14px; }}
|
|
th {{ background: #f8fafc; font-weight: 600; }}
|
|
.muted {{ color: #64748b; font-size: 14px; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<div class="layout">
|
|
<aside class="sidebar">
|
|
<h1>IRT Bank Soal Admin</h1>
|
|
<a href="/admin/dashboard">Dashboard</a>
|
|
<a href="/admin/calibration-status">Calibration Status</a>
|
|
<a href="/admin/item-statistics">Item Statistics</a>
|
|
<a href="/admin/session-overview">Session Overview</a>
|
|
<a href="/admin/password">Password Info</a>
|
|
<a href="/admin/logout">Logout</a>
|
|
</aside>
|
|
<main class="content">
|
|
<div class="card">
|
|
<h2>{escape(page_title)}</h2>
|
|
{body}
|
|
</div>
|
|
</main>
|
|
</div>
|
|
</body>
|
|
</html>"""
|
|
return HTMLResponse(html)
|
|
|
|
|
|
def _table(headers: list[str], rows: list[list[Any]]) -> str:
|
|
head = "".join(f"<th>{escape(str(header))}</th>" for header in headers)
|
|
body_rows = []
|
|
for row in rows:
|
|
cols = "".join(f"<td>{escape(str(value))}</td>" for value in row)
|
|
body_rows.append(f"<tr>{cols}</tr>")
|
|
body = "".join(body_rows) or f"<tr><td colspan=\"{len(headers)}\">No data</td></tr>"
|
|
return f"<table><thead><tr>{head}</tr></thead><tbody>{body}</tbody></table>"
|
|
|
|
|
|
@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 = """
|
|
<form method="post" action="/admin/login" autocomplete="off">
|
|
<label for="username">Username</label>
|
|
<input id="username" name="username" type="text" autocomplete="username">
|
|
<label for="password">Password</label>
|
|
<input id="password" name="password" type="password" autocomplete="current-password">
|
|
<label class="row"><input type="checkbox" name="remember_me" __REMEMBER_ME_CHECKED__> Remember me</label>
|
|
<button type="submit">Sign in</button>
|
|
</form>
|
|
<p class="muted">Direct environment-backed admin access.</p>
|
|
"""
|
|
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"""
|
|
<div class="error">Invalid username or password.</div>
|
|
<form method="post" action="/admin/login" autocomplete="off">
|
|
<label for="username">Username</label>
|
|
<input id="username" name="username" type="text" autocomplete="username" value="{escape(username)}">
|
|
<label for="password">Password</label>
|
|
<input id="password" name="password" type="password" autocomplete="current-password">
|
|
<label class="row"><input type="checkbox" name="remember_me" __REMEMBER_ME_CHECKED__> Remember me</label>
|
|
<button type="submit">Sign in</button>
|
|
</form>
|
|
"""
|
|
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"""
|
|
<p class="muted">Signed in as <strong>{escape(admin.username)}</strong>.</p>
|
|
<p>Password changes are disabled in the UI for this deployment.</p>
|
|
<p>Update <code>ADMIN_PASSWORD</code> in the server environment, then restart the app.</p>
|
|
<p>Session expiry is currently set to <strong>{settings.ADMIN_SESSION_EXPIRE_SECONDS}</strong> seconds.</p>
|
|
<p><a href="/admin/dashboard">Back to dashboard</a></p>
|
|
"""
|
|
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 = """
|
|
<div class="error">Password rotation via UI is disabled.</div>
|
|
<p>Update <code>ADMIN_PASSWORD</code> in the server environment, then restart the app.</p>
|
|
<p><a href="/admin/dashboard">Back to dashboard</a></p>
|
|
"""
|
|
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"""
|
|
<p class="muted">Signed in as <strong>{escape(admin.username)}</strong>.</p>
|
|
<div class="grid">
|
|
<div class="stat">Tryouts<strong>{tryouts}</strong></div>
|
|
<div class="stat">Items<strong>{items}</strong></div>
|
|
<div class="stat">Sessions<strong>{sessions}</strong></div>
|
|
<div class="stat">Completed Sessions<strong>{completed_sessions}</strong></div>
|
|
</div>
|
|
"""
|
|
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)
|
|
|
|
|
|
@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()
|