626 lines
22 KiB
Python
626 lines
22 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.
|
|
"""
|
|
|
|
from typing import Any, Dict, Optional
|
|
|
|
from fastapi import Request
|
|
from fastapi_admin.app import app as admin_app
|
|
from fastapi_admin.resources import (
|
|
Field,
|
|
Link,
|
|
Model,
|
|
)
|
|
from fastapi_admin.widgets import displays, inputs
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
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
|
|
# =============================================================================
|
|
|
|
class AdminAuthProvider:
|
|
"""
|
|
Authentication provider for FastAPI Admin.
|
|
|
|
Supports two modes:
|
|
1. WordPress JWT token integration (production)
|
|
2. Basic auth for testing (development)
|
|
"""
|
|
|
|
async def login(
|
|
self,
|
|
username: str,
|
|
password: str,
|
|
) -> Optional[str]:
|
|
"""
|
|
Authenticate user and return token.
|
|
|
|
Args:
|
|
username: Username
|
|
password: Password
|
|
|
|
Returns:
|
|
Access token if authentication successful, None otherwise
|
|
"""
|
|
# Development mode: basic auth
|
|
if settings.ENVIRONMENT == "development":
|
|
# Allow admin/admin or admin/password for testing
|
|
if (username == "admin" and password in ["admin", "password"]):
|
|
return f"dev_token_{username}"
|
|
|
|
# Production mode: WordPress JWT token validation
|
|
# For now, return None - implement WordPress integration when needed
|
|
return None
|
|
|
|
async def logout(self, request: Request) -> bool:
|
|
"""
|
|
Logout user.
|
|
|
|
Args:
|
|
request: FastAPI request
|
|
|
|
Returns:
|
|
True if logout successful
|
|
"""
|
|
return True
|
|
|
|
async def get_current_user(self, request: Request) -> Optional[dict]:
|
|
"""
|
|
Get current authenticated user.
|
|
|
|
Args:
|
|
request: FastAPI request
|
|
|
|
Returns:
|
|
User data if authenticated, None otherwise
|
|
"""
|
|
token = request.cookies.get("admin_token") or request.headers.get("Authorization")
|
|
|
|
if not token:
|
|
return None
|
|
|
|
# Development mode: validate dev token
|
|
if settings.ENVIRONMENT == "development" and token.startswith("dev_token_"):
|
|
username = token.replace("dev_token_", "")
|
|
return {
|
|
"id": 1,
|
|
"username": username,
|
|
"is_superuser": True,
|
|
}
|
|
|
|
return None
|
|
|
|
|
|
# =============================================================================
|
|
# 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(options=["ctt", "irt", "hybrid"], default="ctt"),
|
|
display=displays.Select(choices=["ctt", "irt", "hybrid"]),
|
|
),
|
|
Field(
|
|
name="selection_mode",
|
|
label="Selection Mode",
|
|
input_=inputs.Select(options=["fixed", "adaptive", "hybrid"], default="fixed"),
|
|
display=displays.Select(choices=["fixed", "adaptive", "hybrid"]),
|
|
),
|
|
Field(
|
|
name="normalization_mode",
|
|
label="Normalization Mode",
|
|
input_=inputs.Select(options=["static", "dynamic", "hybrid"], default="static"),
|
|
display=displays.Select(choices=["static", "dynamic", "hybrid"]),
|
|
),
|
|
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(options=["mle", "map", "eap"], default="mle"),
|
|
display=displays.Select(choices=["mle", "map", "eap"]),
|
|
),
|
|
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.DateTime()),
|
|
Field(name="updated_at", label="Updated At", input_=inputs.DateTime(), display=displays.DateTime()),
|
|
]
|
|
|
|
|
|
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(options=["mudah", "sedang", "sulit"], default="sedang"),
|
|
display=displays.Display(),
|
|
),
|
|
Field(
|
|
name="stem",
|
|
label="Question Stem",
|
|
input_=inputs.TextArea(),
|
|
display=displays.Text(maxlen=100),
|
|
),
|
|
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.Text(maxlen=100),
|
|
),
|
|
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(options=["mudah", "sedang", "sulit"]),
|
|
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(options=["manual", "ai"], 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.DateTime()),
|
|
Field(name="updated_at", label="Updated At", input_=inputs.DateTime(), display=displays.DateTime()),
|
|
]
|
|
|
|
|
|
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.DateTime()),
|
|
Field(name="updated_at", label="Updated At", input_=inputs.DateTime(), display=displays.DateTime()),
|
|
]
|
|
|
|
|
|
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.DateTime()),
|
|
Field(name="end_time", label="End Time", input_=inputs.DateTime(), display=displays.DateTime()),
|
|
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(options=["ctt", "irt", "hybrid"]),
|
|
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.DateTime()),
|
|
Field(name="updated_at", label="Updated At", input_=inputs.DateTime(), display=displays.DateTime()),
|
|
]
|
|
|
|
|
|
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.DateTime(),
|
|
),
|
|
Field(name="created_at", label="Created At", input_=inputs.DateTime(), display=displays.DateTime()),
|
|
Field(name="updated_at", label="Updated At", input_=inputs.DateTime(), display=displays.DateTime()),
|
|
]
|
|
|
|
|
|
# =============================================================================
|
|
# 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
|
|
admin_app.settings.auth_provider = AdminAuthProvider()
|
|
|
|
# Register model resources
|
|
admin_app.register(TryoutResource)
|
|
admin_app.register(ItemResource)
|
|
admin_app.register(UserResource)
|
|
admin_app.register(SessionResource)
|
|
admin_app.register(TryoutStatsResource)
|
|
|
|
# Register dashboard links
|
|
admin_app.register(CalibrationDashboardLink)
|
|
admin_app.register(ItemStatisticsLink)
|
|
admin_app.register(SessionOverviewLink)
|
|
|
|
return admin_app
|
|
|
|
|
|
# Export admin app for mounting in main.py
|
|
admin = create_admin_app()
|