""" 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(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 # 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()