860 lines
30 KiB
Python
860 lines
30 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, get_resources
|
|
from fastapi_admin.providers import Provider
|
|
from fastapi_admin.resources import (
|
|
Field,
|
|
Link,
|
|
Model,
|
|
)
|
|
from fastapi_admin.template import templates
|
|
from fastapi_admin.widgets import displays, inputs
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
|
|
from starlette.responses import 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,
|
|
template: str = "providers/login/login.html",
|
|
) -> 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
|
|
self.template = template
|
|
|
|
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 _template_response(
|
|
self,
|
|
request: Request,
|
|
name: str,
|
|
context: Dict[str, Any],
|
|
status_code: int = 200,
|
|
):
|
|
"""Build a template response compatible with old/new Starlette signatures."""
|
|
payload = {"request": request, **context}
|
|
try:
|
|
# Starlette >= 1.0
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name=name,
|
|
context=payload,
|
|
status_code=status_code,
|
|
)
|
|
except TypeError:
|
|
# Starlette < 1.0
|
|
return templates.TemplateResponse(
|
|
name,
|
|
context=payload,
|
|
status_code=status_code,
|
|
)
|
|
|
|
@staticmethod
|
|
def _admin_home(request: Request) -> str:
|
|
"""
|
|
Resolve a concrete admin page path.
|
|
|
|
This project uses SQLAlchemy models, while fastapi-admin's built-in
|
|
Model CRUD pages are Tortoise-oriented. Prefer custom Link resources
|
|
and known safe admin pages.
|
|
"""
|
|
admin_path = request.app.admin_path.rstrip("/")
|
|
for resource in getattr(request.app, "resources", []):
|
|
try:
|
|
if issubclass(resource, Model):
|
|
model_name = resource.model.__name__.lower()
|
|
return f"{admin_path}/{model_name}/list"
|
|
except TypeError:
|
|
continue
|
|
for resource in getattr(request.app, "resources", []):
|
|
try:
|
|
if issubclass(resource, Link):
|
|
url = getattr(resource, "url", "")
|
|
if isinstance(url, str) and url.startswith("/"):
|
|
return url
|
|
except TypeError:
|
|
continue
|
|
return f"{admin_path}/password"
|
|
|
|
@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):
|
|
return self._template_response(
|
|
request=request,
|
|
name=self.template,
|
|
context={
|
|
"login_logo_url": self.login_logo_url,
|
|
"login_title": self.login_title,
|
|
},
|
|
)
|
|
|
|
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)
|
|
):
|
|
return self._template_response(
|
|
request=request,
|
|
name=self.template,
|
|
status_code=HTTP_401_UNAUTHORIZED,
|
|
context={
|
|
"error": "Invalid username or password",
|
|
"login_logo_url": self.login_logo_url,
|
|
"login_title": self.login_title,
|
|
},
|
|
)
|
|
|
|
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, resources=Depends(get_resources)):
|
|
return self._template_response(
|
|
request=request,
|
|
name="providers/login/password.html",
|
|
context={"resources": resources},
|
|
)
|
|
|
|
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),
|
|
resources=Depends(get_resources),
|
|
):
|
|
_ = admin
|
|
if not secrets.compare_digest(old_password, self._password):
|
|
return self._template_response(
|
|
request=request,
|
|
name="providers/login/password.html",
|
|
context={
|
|
"resources": resources,
|
|
"error": "Old password is incorrect",
|
|
},
|
|
)
|
|
if new_password != re_new_password:
|
|
return self._template_response(
|
|
request=request,
|
|
name="providers/login/password.html",
|
|
context={
|
|
"resources": resources,
|
|
"error": "New passwords do not match",
|
|
},
|
|
)
|
|
|
|
# Password is env-configured and immutable at runtime.
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Password rotation via UI is disabled. Update ADMIN_PASSWORD in environment.",
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# 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 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.
|
|
|
|
# NOTE:
|
|
# fastapi-admin Model resources rely on Tortoise ORM query APIs.
|
|
# This codebase uses SQLAlchemy, so register only Link resources here.
|
|
# Keep Model resource classes in source for future migration work.
|
|
|
|
# Register dashboard links (safe for SQLAlchemy-backed custom views)
|
|
admin_app.register(CalibrationDashboardLink)
|
|
admin_app.register(ItemStatisticsLink)
|
|
admin_app.register(SessionOverviewLink)
|
|
|
|
calibration_link = CalibrationDashboardLink()
|
|
item_stats_link = ItemStatisticsLink()
|
|
session_overview_link = SessionOverviewLink()
|
|
|
|
admin_app.get("/calibration_status", dependencies=[Depends(get_current_admin)])(calibration_link.get)
|
|
admin_app.get("/item_statistics", dependencies=[Depends(get_current_admin)])(item_stats_link.get)
|
|
admin_app.get("/session_overview", dependencies=[Depends(get_current_admin)])(session_overview_link.get)
|
|
|
|
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()
|