""" 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. fastapi-admin 1.0.x does not expose a root "/" view by default; the first usable page is a model list route: /{resource}/list. """ 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 return f"{admin_path}{getattr(request.app.login_provider, 'login_path', '/login')}" @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. # 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 _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()