1017 lines
37 KiB
Python
1017 lines
37 KiB
Python
"""
|
|
FastAPI Admin configuration for IRT Bank Soal system.
|
|
|
|
Provides admin panel for managing tryouts, items, sessions, users, and tryout stats.
|
|
Includes custom actions for calibration, AI generation toggle, and normalization reset.
|
|
"""
|
|
|
|
import secrets
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from typing import Any, Dict, Optional
|
|
|
|
import aioredis
|
|
from fastapi import Depends, Form, HTTPException, Request
|
|
from fastapi_admin import constants
|
|
from fastapi_admin.app import app as admin_app
|
|
from fastapi_admin.depends import get_current_admin
|
|
from fastapi_admin.providers import Provider
|
|
from fastapi_admin.resources import (
|
|
Field,
|
|
Link,
|
|
Model,
|
|
)
|
|
from fastapi_admin.widgets import displays, inputs
|
|
from sqlalchemy import select
|
|
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
|
|
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, TryoutStats, User
|
|
|
|
settings = get_settings()
|
|
|
|
|
|
# =============================================================================
|
|
# Authentication Provider
|
|
# =============================================================================
|
|
|
|
@dataclass
|
|
class AdminPrincipal:
|
|
"""Minimal admin user object expected by fastapi-admin templates."""
|
|
|
|
pk: str
|
|
username: str
|
|
avatar: str = ""
|
|
|
|
|
|
class EnvCredentialProvider(Provider):
|
|
"""
|
|
FastAPI-Admin provider backed by env credentials and Redis session tokens.
|
|
|
|
Compatible with fastapi-admin 1.0.x provider API without requiring
|
|
Tortoise admin models.
|
|
"""
|
|
|
|
# fastapi-admin login templates expect `request.app.login_provider`
|
|
# to exist for resolving login/logout URLs.
|
|
name = "login_provider"
|
|
access_token = "access_token"
|
|
|
|
def __init__(
|
|
self,
|
|
username: str,
|
|
password: str,
|
|
login_path: str = "/login",
|
|
logout_path: str = "/logout",
|
|
login_title: str = "Admin Login",
|
|
login_logo_url: str | None = None,
|
|
expire_seconds: int = 3600,
|
|
) -> None:
|
|
self._username = username
|
|
self._password = password
|
|
self.login_path = login_path
|
|
self.logout_path = logout_path
|
|
self.login_title = login_title
|
|
self.login_logo_url = login_logo_url
|
|
self.expire_seconds = expire_seconds
|
|
|
|
async def register(self, app: "FastAPIAdmin") -> None:
|
|
await super().register(app)
|
|
# Keep explicit assignment for compatibility across fastapi-admin versions.
|
|
app.login_provider = self
|
|
app.get("/")(self.index_view)
|
|
app.get(self.login_path)(self.login_view)
|
|
app.post(self.login_path)(self.login)
|
|
app.get(self.logout_path)(self.logout)
|
|
app.get("/password")(self.password_view)
|
|
app.post("/password")(self.password)
|
|
app.add_middleware(BaseHTTPMiddleware, dispatch=self.authenticate)
|
|
|
|
def _render_auth_page(
|
|
self,
|
|
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>{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>{title}</h1>
|
|
<p>{subtitle}</p>
|
|
{body.replace("__REMEMBER_ME_CHECKED__", remember_me_checked)}
|
|
</main>
|
|
</body>
|
|
</html>"""
|
|
return HTMLResponse(html, status_code=status_code)
|
|
|
|
@staticmethod
|
|
def _admin_home(request: Request) -> str:
|
|
return request.app.admin_path.rstrip("/") + "/dashboard"
|
|
|
|
@staticmethod
|
|
def _login_url(request: Request) -> str:
|
|
return request.app.admin_path.rstrip("/") + "/login"
|
|
|
|
async def index_view(self, request: Request):
|
|
# fastapi-admin has no default "/" page in this setup.
|
|
if getattr(request.state, "admin", None):
|
|
return RedirectResponse(url=self._admin_home(request), status_code=HTTP_303_SEE_OTHER)
|
|
return RedirectResponse(url=self._login_url(request), status_code=HTTP_303_SEE_OTHER)
|
|
|
|
async def login_view(self, request: Request):
|
|
body = f"""
|
|
<form method="post" action="{request.app.admin_path}{self.login_path}" 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 self._render_auth_page(
|
|
request=request,
|
|
title=self.login_title,
|
|
subtitle="Use the configured admin credentials to access the dashboard.",
|
|
body=body,
|
|
)
|
|
|
|
async def login(
|
|
self,
|
|
request: Request,
|
|
username: str = Form(...),
|
|
password: str = Form(...),
|
|
remember_me: Optional[str] = Form(None),
|
|
):
|
|
if not (
|
|
secrets.compare_digest(username, self._username)
|
|
and secrets.compare_digest(password, self._password)
|
|
):
|
|
body = f"""
|
|
<div class="error">Invalid username or password.</div>
|
|
<form method="post" action="{request.app.admin_path}{self.login_path}" autocomplete="off">
|
|
<label for="username">Username</label>
|
|
<input id="username" name="username" type="text" autocomplete="username" value="{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 self._render_auth_page(
|
|
request=request,
|
|
title=self.login_title,
|
|
subtitle="Use the configured admin credentials to access the dashboard.",
|
|
body=body,
|
|
status_code=HTTP_401_UNAUTHORIZED,
|
|
)
|
|
|
|
response = RedirectResponse(url=self._admin_home(request), status_code=HTTP_303_SEE_OTHER)
|
|
expire = self.expire_seconds
|
|
if remember_me == "on":
|
|
expire = max(self.expire_seconds, 3600 * 24 * 30)
|
|
response.set_cookie("remember_me", "on")
|
|
else:
|
|
response.delete_cookie("remember_me")
|
|
|
|
token = uuid.uuid4().hex
|
|
response.set_cookie(
|
|
self.access_token,
|
|
token,
|
|
expires=expire,
|
|
path=request.app.admin_path,
|
|
httponly=True,
|
|
)
|
|
await request.app.redis.set(constants.LOGIN_USER.format(token=token), self._username, ex=expire)
|
|
return response
|
|
|
|
async def authenticate(self, request: Request, call_next: RequestResponseEndpoint):
|
|
token = request.cookies.get(self.access_token)
|
|
path = request.scope["path"]
|
|
admin = None
|
|
|
|
if token:
|
|
key = constants.LOGIN_USER.format(token=token)
|
|
username = await request.app.redis.get(key)
|
|
if username:
|
|
admin = AdminPrincipal(pk=str(username), username=str(username))
|
|
|
|
request.state.admin = admin
|
|
|
|
if path.endswith(self.login_path) and admin:
|
|
return RedirectResponse(url=self._admin_home(request), status_code=HTTP_303_SEE_OTHER)
|
|
|
|
return await call_next(request)
|
|
|
|
async def logout(self, request: Request):
|
|
response = RedirectResponse(
|
|
url=request.app.admin_path + self.login_path,
|
|
status_code=HTTP_303_SEE_OTHER,
|
|
)
|
|
token = request.cookies.get(self.access_token)
|
|
if token:
|
|
await request.app.redis.delete(constants.LOGIN_USER.format(token=token))
|
|
response.delete_cookie(self.access_token, path=request.app.admin_path)
|
|
return response
|
|
|
|
async def password_view(
|
|
self,
|
|
request: Request,
|
|
admin: AdminPrincipal = Depends(get_current_admin),
|
|
):
|
|
body = f"""
|
|
<p class="muted">Signed in as <strong>{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>{self.expire_seconds}</strong> seconds.</p>
|
|
<p><a href="{request.app.admin_path}/dashboard">Back to dashboard</a></p>
|
|
"""
|
|
return self._render_auth_page(
|
|
request=request,
|
|
title="Password Management",
|
|
subtitle="Runtime password rotation is intentionally disabled.",
|
|
body=body,
|
|
)
|
|
|
|
async def password(
|
|
self,
|
|
request: Request,
|
|
old_password: str = Form(...),
|
|
new_password: str = Form(...),
|
|
re_new_password: str = Form(...),
|
|
admin: AdminPrincipal = Depends(get_current_admin),
|
|
):
|
|
_ = (old_password, new_password, re_new_password, admin)
|
|
body = f"""
|
|
<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="{request.app.admin_path}/dashboard">Back to dashboard</a></p>
|
|
"""
|
|
return self._render_auth_page(
|
|
request=request,
|
|
title="Password Management",
|
|
subtitle="Runtime password rotation is intentionally disabled.",
|
|
body=body,
|
|
status_code=400,
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Admin Model Resources
|
|
# =============================================================================
|
|
|
|
class TryoutResource(Model):
|
|
"""
|
|
Admin resource for Tryout model.
|
|
|
|
Displays tryout configuration and provides calibration and AI generation actions.
|
|
"""
|
|
|
|
label = "Tryouts"
|
|
model = Tryout
|
|
page_size = 20
|
|
|
|
# Fields to display
|
|
fields = [
|
|
Field(name="id", label="ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(name="website_id", label="Website ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(name="tryout_id", label="Tryout ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(name="name", label="Name", input_=inputs.Input(), display=displays.Display()),
|
|
Field(
|
|
name="description",
|
|
label="Description",
|
|
input_=inputs.TextArea(),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="scoring_mode",
|
|
label="Scoring Mode",
|
|
input_=inputs.Select(default="ctt"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="selection_mode",
|
|
label="Selection Mode",
|
|
input_=inputs.Select(default="fixed"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="normalization_mode",
|
|
label="Normalization Mode",
|
|
input_=inputs.Select(default="static"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="min_sample_for_dynamic",
|
|
label="Min Sample for Dynamic",
|
|
input_=inputs.Input(type="number"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="static_rataan",
|
|
label="Static Mean (Rataan)",
|
|
input_=inputs.Input(type="number"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="static_sb",
|
|
label="Static Std Dev (SB)",
|
|
input_=inputs.Input(type="number"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="ai_generation_enabled",
|
|
label="Enable AI Generation",
|
|
input_=inputs.Switch(),
|
|
display=displays.Boolean(true_text="Enabled", false_text="Disabled"),
|
|
),
|
|
Field(
|
|
name="hybrid_transition_slot",
|
|
label="Hybrid Transition Slot",
|
|
input_=inputs.Input(type="number"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="min_calibration_sample",
|
|
label="Min Calibration Sample",
|
|
input_=inputs.Input(type="number"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="theta_estimation_method",
|
|
label="Theta Estimation Method",
|
|
input_=inputs.Select(default="mle"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="fallback_to_ctt_on_error",
|
|
label="Fallback to CTT on Error",
|
|
input_=inputs.Switch(),
|
|
display=displays.Boolean(true_text="Yes", false_text="No"),
|
|
),
|
|
Field(name="created_at", label="Created At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
|
|
Field(name="updated_at", label="Updated At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
|
|
]
|
|
|
|
|
|
class ItemResource(Model):
|
|
"""
|
|
Admin resource for Item model.
|
|
|
|
Displays items with CTT and IRT parameters, and calibration status.
|
|
"""
|
|
|
|
label = "Items"
|
|
model = Item
|
|
page_size = 50
|
|
|
|
# Fields to display
|
|
fields = [
|
|
Field(name="id", label="ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(name="tryout_id", label="Tryout ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(name="website_id", label="Website ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(name="slot", label="Slot", input_=inputs.Input(type="number"), display=displays.Display()),
|
|
Field(
|
|
name="level",
|
|
label="Difficulty Level",
|
|
input_=inputs.Select(default="sedang"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="stem",
|
|
label="Question Stem",
|
|
input_=inputs.TextArea(),
|
|
display=displays.Display(),
|
|
),
|
|
Field(name="options", label="Options", input_=inputs.Json(), display=displays.Json()),
|
|
Field(name="correct_answer", label="Correct Answer", input_=inputs.Input(), display=displays.Display()),
|
|
Field(
|
|
name="explanation",
|
|
label="Explanation",
|
|
input_=inputs.TextArea(),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="ctt_p",
|
|
label="CTT p-value",
|
|
input_=inputs.Input(type="number"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="ctt_bobot",
|
|
label="CTT Bobot",
|
|
input_=inputs.Input(type="number"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="ctt_category",
|
|
label="CTT Category",
|
|
input_=inputs.Select(),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="irt_b",
|
|
label="IRT b-parameter",
|
|
input_=inputs.Input(type="number"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="irt_se",
|
|
label="IRT SE",
|
|
input_=inputs.Input(type="number"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="calibrated",
|
|
label="Calibrated",
|
|
input_=inputs.Switch(),
|
|
display=displays.Boolean(true_text="Yes", false_text="No"),
|
|
),
|
|
Field(
|
|
name="calibration_sample_size",
|
|
label="Calibration Sample Size",
|
|
input_=inputs.Input(type="number"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="generated_by",
|
|
label="Generated By",
|
|
input_=inputs.Select(default="manual"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(name="ai_model", label="AI Model", input_=inputs.Input(), display=displays.Display()),
|
|
Field(
|
|
name="basis_item_id",
|
|
label="Basis Item ID",
|
|
input_=inputs.Input(type="number"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(name="created_at", label="Created At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
|
|
Field(name="updated_at", label="Updated At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
|
|
]
|
|
|
|
|
|
class UserResource(Model):
|
|
"""
|
|
Admin resource for User model.
|
|
|
|
Displays WordPress users and their tryout sessions.
|
|
"""
|
|
|
|
label = "Users"
|
|
model = User
|
|
page_size = 50
|
|
|
|
# Fields
|
|
fields = [
|
|
Field(name="id", label="ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(name="wp_user_id", label="WordPress User ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(name="website_id", label="Website ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(name="created_at", label="Created At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
|
|
Field(name="updated_at", label="Updated At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
|
|
]
|
|
|
|
|
|
class SessionResource(Model):
|
|
"""
|
|
Admin resource for Session model.
|
|
|
|
Displays tryout sessions with scoring results (NM, NN, theta).
|
|
"""
|
|
|
|
label = "Sessions"
|
|
model = Session
|
|
page_size = 50
|
|
|
|
# Fields
|
|
fields = [
|
|
Field(name="id", label="ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(name="session_id", label="Session ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(name="wp_user_id", label="WordPress User ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(name="website_id", label="Website ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(name="tryout_id", label="Tryout ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(name="start_time", label="Start Time", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
|
|
Field(name="end_time", label="End Time", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
|
|
Field(
|
|
name="is_completed",
|
|
label="Completed",
|
|
input_=inputs.Switch(),
|
|
display=displays.Boolean(true_text="Yes", false_text="No"),
|
|
),
|
|
Field(
|
|
name="scoring_mode_used",
|
|
label="Scoring Mode Used",
|
|
input_=inputs.Select(),
|
|
display=displays.Display(),
|
|
),
|
|
Field(name="total_benar", label="Total Benar", input_=inputs.Input(type="number"), display=displays.Display()),
|
|
Field(name="total_bobot_earned", label="Total Bobot Earned", input_=inputs.Input(type="number"), display=displays.Display()),
|
|
Field(name="NM", label="NM Score", input_=inputs.Input(type="number"), display=displays.Display()),
|
|
Field(name="NN", label="NN Score", input_=inputs.Input(type="number"), display=displays.Display()),
|
|
Field(name="theta", label="Theta", input_=inputs.Input(type="number"), display=displays.Display()),
|
|
Field(name="theta_se", label="Theta SE", input_=inputs.Input(type="number"), display=displays.Display()),
|
|
Field(name="rataan_used", label="Rataan Used", input_=inputs.Input(type="number"), display=displays.Display()),
|
|
Field(name="sb_used", label="SB Used", input_=inputs.Input(type="number"), display=displays.Display()),
|
|
Field(name="created_at", label="Created At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
|
|
Field(name="updated_at", label="Updated At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
|
|
]
|
|
|
|
|
|
class TryoutStatsResource(Model):
|
|
"""
|
|
Admin resource for TryoutStats model.
|
|
|
|
Displays tryout-level statistics and provides normalization reset action.
|
|
"""
|
|
|
|
label = "Tryout Stats"
|
|
model = TryoutStats
|
|
page_size = 20
|
|
|
|
# Fields
|
|
fields = [
|
|
Field(name="id", label="ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(name="website_id", label="Website ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(name="tryout_id", label="Tryout ID", input_=inputs.Input(), display=displays.Display()),
|
|
Field(
|
|
name="participant_count",
|
|
label="Participant Count",
|
|
input_=inputs.Input(type="number"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="total_nm_sum",
|
|
label="Total NM Sum",
|
|
input_=inputs.Input(type="number"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="total_nm_sq_sum",
|
|
label="Total NM Squared Sum",
|
|
input_=inputs.Input(type="number"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(name="rataan", label="Rataan", input_=inputs.Input(type="number"), display=displays.Display()),
|
|
Field(name="sb", label="SB", input_=inputs.Input(type="number"), display=displays.Display()),
|
|
Field(name="min_nm", label="Min NM", input_=inputs.Input(type="number"), display=displays.Display()),
|
|
Field(name="max_nm", label="Max NM", input_=inputs.Input(type="number"), display=displays.Display()),
|
|
Field(
|
|
name="last_calculated",
|
|
label="Last Calculated",
|
|
input_=inputs.DateTime(),
|
|
display=displays.DatetimeDisplay(),
|
|
),
|
|
Field(name="created_at", label="Created At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
|
|
Field(name="updated_at", label="Updated At", input_=inputs.DateTime(), display=displays.DatetimeDisplay()),
|
|
]
|
|
|
|
|
|
# =============================================================================
|
|
# Custom Dashboard Views
|
|
# =============================================================================
|
|
|
|
class CalibrationDashboardLink(Link):
|
|
"""
|
|
Link to calibration status dashboard.
|
|
|
|
Displays calibration percentage and items awaiting calibration.
|
|
"""
|
|
|
|
label = "Calibration Status"
|
|
icon = "fas fa-chart-line"
|
|
url = "/admin/calibration_status"
|
|
|
|
async def get(self, request: Request) -> Dict[str, Any]:
|
|
"""Get calibration status for all tryouts."""
|
|
# Get all tryouts
|
|
db_gen = get_db()
|
|
db = await db_gen.__anext__()
|
|
|
|
try:
|
|
result = await db.execute(
|
|
select(
|
|
Tryout.id,
|
|
Tryout.tryout_id,
|
|
Tryout.name,
|
|
)
|
|
)
|
|
tryouts = result.all()
|
|
|
|
calibration_data = []
|
|
for tryout_id, tryout_str, name in tryouts:
|
|
# Get calibration status
|
|
from app.services.irt_calibration import get_calibration_status
|
|
|
|
status = await get_calibration_status(tryout_str, 1, db)
|
|
calibration_data.append({
|
|
"tryout_id": tryout_str,
|
|
"name": name,
|
|
"total_items": status["total_items"],
|
|
"calibrated_items": status["calibrated_items"],
|
|
"calibration_percentage": status["calibration_percentage"],
|
|
"ready_for_irt": status["ready_for_irt"],
|
|
})
|
|
|
|
return {
|
|
"status": "success",
|
|
"data": calibration_data,
|
|
}
|
|
finally:
|
|
await db_gen.aclose()
|
|
|
|
|
|
class ItemStatisticsLink(Link):
|
|
"""
|
|
Link to item statistics view.
|
|
|
|
Displays items grouped by difficulty level with calibration status.
|
|
"""
|
|
|
|
label = "Item Statistics"
|
|
icon = "fas fa-chart-bar"
|
|
url = "/admin/item_statistics"
|
|
|
|
async def get(self, request: Request) -> Dict[str, Any]:
|
|
"""Get item statistics grouped by difficulty level."""
|
|
db_gen = get_db()
|
|
db = await db_gen.__anext__()
|
|
|
|
try:
|
|
# Get items grouped by level
|
|
result = await db.execute(
|
|
select(
|
|
Item.level,
|
|
)
|
|
.distinct()
|
|
)
|
|
levels = result.scalars().all()
|
|
|
|
stats = []
|
|
for level in levels:
|
|
# Get items for this level
|
|
item_result = await db.execute(
|
|
select(Item)
|
|
.where(Item.level == level)
|
|
.order_by(Item.slot)
|
|
.limit(10)
|
|
)
|
|
items = item_result.scalars().all()
|
|
|
|
# Calculate average correctness rate
|
|
total_responses = sum(item.calibration_sample_size for item in items)
|
|
calibrated_count = sum(1 for item in items if item.calibrated)
|
|
|
|
level_stats = {
|
|
"level": level,
|
|
"total_items": len(items),
|
|
"calibrated_items": calibrated_count,
|
|
"calibration_percentage": (calibrated_count / len(items) * 100) if len(items) > 0 else 0,
|
|
"total_responses": total_responses,
|
|
"avg_correctness": sum(item.ctt_p or 0 for item in items) / len(items) if len(items) > 0 else 0,
|
|
"items": [
|
|
{
|
|
"id": item.id,
|
|
"slot": item.slot,
|
|
"calibrated": item.calibrated,
|
|
"ctt_p": item.ctt_p,
|
|
"irt_b": item.irt_b,
|
|
"calibration_sample_size": item.calibration_sample_size,
|
|
}
|
|
for item in items
|
|
],
|
|
}
|
|
stats.append(level_stats)
|
|
|
|
return {
|
|
"status": "success",
|
|
"data": stats,
|
|
}
|
|
finally:
|
|
await db_gen.aclose()
|
|
|
|
|
|
class SessionOverviewLink(Link):
|
|
"""
|
|
Link to session overview view.
|
|
|
|
Displays sessions with scores (NM, NN, theta) and completion status.
|
|
"""
|
|
|
|
label = "Session Overview"
|
|
icon = "fas fa-users"
|
|
url = "/admin/session_overview"
|
|
|
|
async def get(self, request: Request) -> Dict[str, Any]:
|
|
"""Get session overview with filters."""
|
|
db_gen = get_db()
|
|
db = await db_gen.__anext__()
|
|
|
|
try:
|
|
# Get recent sessions
|
|
result = await db.execute(
|
|
select(Session)
|
|
.order_by(Session.created_at.desc())
|
|
.limit(50)
|
|
)
|
|
sessions = result.scalars().all()
|
|
|
|
session_data = [
|
|
{
|
|
"session_id": session.session_id,
|
|
"wp_user_id": session.wp_user_id,
|
|
"tryout_id": session.tryout_id,
|
|
"is_completed": session.is_completed,
|
|
"scoring_mode_used": session.scoring_mode_used,
|
|
"total_benar": session.total_benar,
|
|
"NM": session.NM,
|
|
"NN": session.NN,
|
|
"theta": session.theta,
|
|
"theta_se": session.theta_se,
|
|
"start_time": session.start_time.isoformat() if session.start_time else None,
|
|
"end_time": session.end_time.isoformat() if session.end_time else None,
|
|
}
|
|
for session in sessions
|
|
]
|
|
|
|
return {
|
|
"status": "success",
|
|
"data": session_data,
|
|
}
|
|
finally:
|
|
await db_gen.aclose()
|
|
|
|
|
|
# =============================================================================
|
|
# Initialize FastAPI Admin
|
|
# =============================================================================
|
|
|
|
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>{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); }}
|
|
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>{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>{header}</th>" for header in headers)
|
|
body_rows = []
|
|
for row in rows:
|
|
cols = "".join(f"<td>{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>"
|
|
|
|
|
|
def _prune_incompatible_admin_routes() -> None:
|
|
admin_app.router.routes[:] = [
|
|
route
|
|
for route in admin_app.router.routes
|
|
if not getattr(route, "path", "").startswith("/{resource}/")
|
|
]
|
|
|
|
|
|
async def dashboard_view(request: Request, admin: AdminPrincipal = Depends(get_current_admin)):
|
|
_ = admin
|
|
body = (
|
|
'<p class="muted">This admin runs in SQLAlchemy-safe mode. '
|
|
"The original fastapi-admin CRUD pages depend on Tortoise ORM and were removed.</p>"
|
|
"<p>Use the navigation links to inspect operational data.</p>"
|
|
)
|
|
return _render_admin_page("IRT Bank Soal Admin", "Dashboard", body)
|
|
|
|
|
|
async def calibration_status_view(
|
|
request: Request,
|
|
admin: AdminPrincipal = Depends(get_current_admin),
|
|
):
|
|
_ = admin
|
|
data = await CalibrationDashboardLink().get(request)
|
|
rows = [
|
|
[
|
|
item["tryout_id"],
|
|
item["name"],
|
|
item["total_items"],
|
|
item["calibrated_items"],
|
|
f'{item["calibration_percentage"]:.2f}%',
|
|
"Yes" if item["ready_for_irt"] else "No",
|
|
]
|
|
for item in data["data"]
|
|
]
|
|
body = _table(
|
|
["Tryout ID", "Name", "Total Items", "Calibrated", "Calibration %", "Ready for IRT"],
|
|
rows,
|
|
)
|
|
return _render_admin_page("Calibration Status", "Calibration Status", body)
|
|
|
|
|
|
async def item_statistics_view(
|
|
request: Request,
|
|
admin: AdminPrincipal = Depends(get_current_admin),
|
|
):
|
|
_ = admin
|
|
data = await ItemStatisticsLink().get(request)
|
|
rows = [
|
|
[
|
|
item["level"],
|
|
item["total_items"],
|
|
item["calibrated_items"],
|
|
f'{item["calibration_percentage"]:.2f}%',
|
|
item["total_responses"],
|
|
f'{item["avg_correctness"]:.4f}',
|
|
]
|
|
for item in data["data"]
|
|
]
|
|
body = _table(
|
|
["Level", "Total Items", "Calibrated", "Calibration %", "Responses", "Avg Correctness"],
|
|
rows,
|
|
)
|
|
return _render_admin_page("Item Statistics", "Item Statistics", body)
|
|
|
|
|
|
async def session_overview_view(
|
|
request: Request,
|
|
admin: AdminPrincipal = Depends(get_current_admin),
|
|
):
|
|
_ = admin
|
|
data = await SessionOverviewLink().get(request)
|
|
rows = [
|
|
[
|
|
item["session_id"],
|
|
item["wp_user_id"],
|
|
item["tryout_id"],
|
|
"Yes" if item["is_completed"] else "No",
|
|
item["scoring_mode_used"],
|
|
item["total_benar"],
|
|
item["NM"],
|
|
item["NN"],
|
|
item["theta"],
|
|
]
|
|
for item in data["data"]
|
|
]
|
|
body = _table(
|
|
["Session ID", "WP User", "Tryout", "Completed", "Mode", "Benar", "NM", "NN", "Theta"],
|
|
rows,
|
|
)
|
|
return _render_admin_page("Session Overview", "Session Overview", body)
|
|
|
|
def create_admin_app() -> Any:
|
|
"""
|
|
Create and configure FastAPI Admin application.
|
|
|
|
Returns:
|
|
FastAPI app with admin panel
|
|
"""
|
|
# Configure admin app
|
|
# admin_app.settings.logo_url = "/static/logo.png"
|
|
# admin_app.settings.site_title = "IRT Bank Soal Admin"
|
|
# admin_app.settings.site_description = "Admin Panel for Adaptive Question Bank System"
|
|
|
|
# Register authentication provider
|
|
# NOTE: fastapi-admin 1.0.4 requires provider registration via app.configure(...).
|
|
# Keep provider implementation here for future integration during startup configure.
|
|
|
|
# Reset singleton registries so stale state cannot survive restarts.
|
|
admin_app.resources = []
|
|
admin_app.model_resources = {}
|
|
_prune_incompatible_admin_routes()
|
|
|
|
admin_app.register(CalibrationDashboardLink)
|
|
admin_app.register(ItemStatisticsLink)
|
|
admin_app.register(SessionOverviewLink)
|
|
|
|
admin_app.get("/dashboard", dependencies=[Depends(get_current_admin)])(dashboard_view)
|
|
admin_app.get("/calibration_status", dependencies=[Depends(get_current_admin)])(calibration_status_view)
|
|
admin_app.get("/item_statistics", dependencies=[Depends(get_current_admin)])(item_statistics_view)
|
|
admin_app.get("/session_overview", dependencies=[Depends(get_current_admin)])(session_overview_view)
|
|
|
|
# Preserve previously exposed broken list URLs and redirect them to the safe dashboard.
|
|
for legacy_path in (
|
|
"/tryout/list",
|
|
"/item/list",
|
|
"/user/list",
|
|
"/session/list",
|
|
"/tryoutstats/list",
|
|
):
|
|
admin_app.get(legacy_path, dependencies=[Depends(get_current_admin)])(dashboard_view)
|
|
|
|
return admin_app
|
|
|
|
|
|
_admin_configured = False
|
|
_admin_redis = None
|
|
|
|
|
|
async def configure_admin_app() -> None:
|
|
"""Configure fastapi-admin runtime (redis + auth provider)."""
|
|
global _admin_configured, _admin_redis
|
|
|
|
if _admin_configured:
|
|
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,
|
|
)
|
|
|
|
provider = EnvCredentialProvider(
|
|
username=settings.ADMIN_USERNAME,
|
|
password=settings.ADMIN_PASSWORD,
|
|
login_title="IRT Bank Soal Admin",
|
|
expire_seconds=settings.ADMIN_SESSION_EXPIRE_SECONDS,
|
|
)
|
|
|
|
await admin_app.configure(
|
|
redis=_admin_redis,
|
|
admin_path="/admin",
|
|
providers=[provider],
|
|
)
|
|
_admin_configured = True
|
|
|
|
|
|
async def shutdown_admin_app() -> None:
|
|
"""Close admin redis client cleanly."""
|
|
global _admin_redis
|
|
|
|
if _admin_redis is None:
|
|
return
|
|
|
|
try:
|
|
await _admin_redis.close()
|
|
finally:
|
|
_admin_redis = None
|
|
|
|
|
|
# Export admin app for mounting in main.py
|
|
admin = create_admin_app()
|