-
Tryouts{tryouts}
-
Items{items}
-
Sessions{sessions}
-
Completed Sessions{completed_sessions}
+
+
{greeting}, {escape(admin.username)}! 👋
+
Here's what's happening with your exam system today.
-
Open AI Playground
+
+ {alerts_html}
+
+
📊 System Overview
+
+
+
📋
+
+
{tryouts_count}
+
Exams
+
+
+
+
📝
+
+
{items_count}
+
Questions
+
+
+
+
👥
+
+
{completed_count}
+
Completed Tests
+
{completion_rate:.0f}% completion rate
+
+
+
+
🌐
+
+
{websites_count}
+
Websites
+
+
+
+
+
🚀 Quick Actions
+
+
+
📈 Recent Activity
+ {activity_html}
"""
+
return _render_admin_page(request, "IRT Bank Soal Admin", "Dashboard", body)
+def _format_relative_time(dt: datetime) -> str:
+ """Format datetime as relative time string."""
+ if dt is None:
+ return "Unknown"
+
+ now = datetime.now(timezone.utc)
+ if dt.tzinfo is None:
+ dt = dt.replace(tzinfo=timezone.utc)
+
+ diff = now - dt
+ seconds = diff.total_seconds()
+
+ if seconds < 60:
+ return "just now"
+ elif seconds < 3600:
+ minutes = int(seconds / 60)
+ return f"{minutes} minute{'s' if minutes > 1 else ''} ago"
+ elif seconds < 86400:
+ hours = int(seconds / 3600)
+ return f"{hours} hour{'s' if hours > 1 else ''} ago"
+ else:
+ days = int(seconds / 86400)
+ return f"{days} day{'s' if days > 1 else ''} ago"
+
+
+# ============================================================
+# NEW HUMAN-FRIENDLY ROUTES
+# ============================================================
+
+
+@router.get("/questions", include_in_schema=False)
+async def questions_view(
+ request: Request,
+ db: AsyncSession = Depends(get_db),
+ q: str = "",
+ difficulty: str = "",
+ status: str = "",
+ website_id: int | None = None,
+ tryout_id: str = "",
+ page: int = 1,
+):
+ """Questions bank - list all questions with working filters and pagination."""
+ admin = await _current_admin(request)
+ if not admin:
+ return _login_redirect()
+
+ # Build query with filters
+ query = select(Item)
+ count_query = select(func.count()).select_from(Item)
+
+ # Search filter (search in stem)
+ if q:
+ search_filter = or_(
+ Item.stem.ilike(f"%{q}%"),
+ Item.tryout_id.ilike(f"%{q}%"),
+ )
+ query = query.where(search_filter)
+ count_query = count_query.where(search_filter)
+
+ # Website filter
+ if website_id:
+ query = query.where(Item.website_id == website_id)
+ count_query = count_query.where(Item.website_id == website_id)
+
+ # Tryout filter
+ if tryout_id:
+ query = query.where(Item.tryout_id == tryout_id)
+ count_query = count_query.where(Item.tryout_id == tryout_id)
+
+ # Get total count before pagination
+ total_result = await db.execute(count_query)
+ total_items = total_result.scalar() or 0
+
+ # Calculate pagination
+ per_page = 25
+ total_pages = max(1, (total_items + per_page - 1) // per_page)
+ page = max(1, min(page, total_pages))
+ offset = (page - 1) * per_page
+
+ # Get paginated items
+ result = await db.execute(
+ query.order_by(Item.website_id.asc(), Item.tryout_id.asc(), Item.slot.asc())
+ .offset(offset)
+ .limit(per_page)
+ )
+ items = list(result.scalars().all())
+
+ # Get websites for filter dropdown
+ websites_result = await db.execute(select(Website).order_by(Website.site_name))
+ websites = list(websites_result.scalars().all())
+
+ # Build question rows
+ question_rows = []
+ for item in items:
+ # Calculate human-readable difficulty
+ p_value = item.ctt_p
+ if p_value is None:
+ difficulty_label = "Unknown"
+ difficulty_class = "difficulty-unknown"
+ elif p_value > 0.70:
+ difficulty_label = "Easy"
+ difficulty_class = "difficulty-easy"
+ elif p_value >= 0.30:
+ difficulty_label = "Medium"
+ difficulty_class = "difficulty-medium"
+ else:
+ difficulty_label = "Hard"
+ difficulty_class = "difficulty-hard"
+
+ # Truncate stem for preview
+ stem_preview = escape(_truncate(_html_to_text(item.stem or ""), 100))
+
+ question_rows.append(f"""
+
+ |
+ #{item.id} |
+
+ {stem_preview}
+
+ {difficulty_label}
+ |
+ Used {item.calibration_sample_size or 0}x
+ |
+ Slot {item.slot}
+
+ |
+ {escape(item.level or "-")} |
+
+
+ {"✅ Calibrated" if item.calibrated else "⏳ Needs Data"}
+
+ |
+
+ View
+ |
+
+ """)
+
+ # Build pagination HTML
+ pagination_html = ""
+ if total_pages > 1:
+ page_links = []
+ for p in range(max(1, page - 2), min(total_pages + 1, page + 3)):
+ active_class = "active" if p == page else ""
+ page_links.append(
+ f'
{p}'
+ )
+
+ pagination_html = f"""
+
+ """
+
+ # Filter selects
+ difficulty_selected = {
+ "easy": 'value="easy" selected',
+ "medium": 'value="medium" selected',
+ "hard": 'value="hard" selected',
+ }.get(difficulty.lower(), "")
+
+ status_selected = {
+ "calibrated": 'value="calibrated" selected',
+ "uncalibrated": 'value="uncalibrated" selected',
+ }.get(status.lower(), "")
+
+ # Build website options
+ website_options = ['
']
+ for site in websites:
+ selected = "selected" if website_id == site.id else ""
+ website_options.append(
+ f'
'
+ )
+
+ table_html = (
+ '
"
+ )
+
+ body = f"""
+
Manage your question bank. Click any question to see details and options.
+
+
+
+
+
{total_items} questions total
+
+
+
+ {table_html}
+ {pagination_html}
+
+
+ """
+
+ return _render_admin_page(request, "Questions", "📝 Question Bank", body)
+
+
+@router.get("/questions/{item_id}", include_in_schema=False)
+async def question_detail_view(
+ item_id: int,
+ request: Request,
+ db: AsyncSession = Depends(get_db),
+):
+ """Question detail view - shows full question with all options and statistics."""
+ admin = await _current_admin(request)
+ if not admin:
+ return _login_redirect()
+
+ # Get the item
+ item = await db.get(Item, item_id)
+ if not item:
+ body = """
+
+
⚠️
+
+
Question not found
+
The question you're looking for doesn't exist or has been deleted.
+
+
+
← Back to Questions
+ """
+ return _render_admin_page(
+ request, "Question Not Found", "Question Not Found", body
+ )
+
+ # Get tryout info
+ tryout_result = await db.execute(
+ select(Tryout).where(
+ Tryout.tryout_id == item.tryout_id,
+ Tryout.website_id == item.website_id,
+ )
+ )
+ tryout = tryout_result.scalar_one_or_none()
+
+ # Get website info
+ website_result = await db.execute(
+ select(Website).where(Website.id == item.website_id)
+ )
+ website = website_result.scalar_one_or_none()
+
+ # Calculate difficulty
+ p_value = item.ctt_p
+ if p_value is None:
+ difficulty_label = "Unknown"
+ difficulty_class = "difficulty-unknown"
+ difficulty_explanation = "Not enough data yet to determine difficulty."
+ elif p_value > 0.70:
+ difficulty_label = "Easy"
+ difficulty_class = "difficulty-easy"
+ difficulty_explanation = (
+ f"{p_value:.1%} of students answered correctly. This is an easy question."
+ )
+ elif p_value >= 0.30:
+ difficulty_label = "Medium"
+ difficulty_class = "difficulty-medium"
+ difficulty_explanation = f"{p_value:.1%} of students answered correctly. This is a medium difficulty question."
+ else:
+ difficulty_label = "Hard"
+ difficulty_class = "difficulty-hard"
+ difficulty_explanation = f"{p_value:.1%} of students answered correctly. This is a difficult question."
+
+ # Parse options from JSON
+ options = item.options or {}
+
+ # Build options HTML
+ options_html = ""
+ correct_key = item.correct_answer or ""
+ for key in sorted(options.keys()):
+ is_correct = key.upper() == correct_key.upper()
+ row_class = "correct-option" if is_correct else ""
+ check_mark = " ✅" if is_correct else ""
+ options_html += f'
| {key}{check_mark} | {str(options[key])} |
'
+
+ # Build stats cards
+ stats_html = f"""
+
+
+ Difficulty
+ {difficulty_label}
+ {p_value if p_value else "N/A"}
+
+
+ Calibration Status
+
+ {"✅ Calibrated" if item.calibrated else "⏳ Needs Data"}
+
+
+
+ Sample Size
+ {item.calibration_sample_size or 0}
+ responses
+
+
+ IRT Difficulty (b)
+ {f"{item.irt_b:.2f}" if item.irt_b else "N/A"}
+
+
+
+ """
+
+ # Context info
+ context_html = f"""
+
+
📍 Context
+
+
+ Website
+ {escape(website.site_name if website else f"ID: {item.website_id}")}
+
+
+ Exam
+ {escape(tryout.name if tryout else item.tryout_id)}
+
+
+ Slot
+ {item.slot}
+
+
+ Level
+ {escape(item.level or "Not specified")}
+
+
+ Item ID
+ #{item.id}
+
+
+ Created
+ {escape(str(item.created_at)[:10] if item.created_at else "Unknown")}
+
+
+
+ """
+
+ # Difficulty explanation
+ difficulty_info = f"""
+
+
💡
+
+
About Difficulty
+
{difficulty_explanation}
+
+
+ """
+
+ body = f"""
+
← Back to Questions
+
+
+
+ {difficulty_info}
+
+
📝 Question
+
{item.stem or "No question text"}
+
+
🔘 Answer Options
+
+
+ | Key | Answer Text |
|---|
+
+ {options_html if options_html else '| No options available |
'}
+
+
+
+
+
📊 Statistics
+ {stats_html}
+
+ {context_html}
+
+
+
ℹ️ What is Calibration?
+
A question becomes "calibrated" after many students (100+) have answered it. Once calibrated, the system can accurately measure student ability and provide adaptive testing.
+
The IRT parameters (difficulty, discrimination, guessing) are calculated from student response patterns.
+
+
+
+ """
+
+ return _render_admin_page(
+ request, f"Question #{item_id}", "📝 Question Details", body
+ )
+
+
+@router.get("/question-quality", include_in_schema=False)
+async def question_quality_view(request: Request, db: AsyncSession = Depends(get_db)):
+ """Question Quality - shows calibration status with human-friendly explanations."""
+ admin = await _current_admin(request)
+ if not admin:
+ return _login_redirect()
+
+ # Get calibration stats by tryout
+ result = await db.execute(
+ select(
+ Tryout.tryout_id,
+ Tryout.name,
+ func.count(Item.id).label("total_items"),
+ func.sum(func.cast(Item.calibrated, Integer)).label("calibrated_items"),
+ )
+ .join(
+ Item,
+ (Tryout.tryout_id == Item.tryout_id)
+ & (Tryout.website_id == Item.website_id),
+ )
+ .group_by(Tryout.tryout_id, Tryout.name)
+ .order_by(Tryout.name)
+ )
+ tryout_stats = list(result.all())
+
+ # Calculate totals
+ total_items = sum(s.total_items or 0 for s in tryout_stats)
+ total_calibrated = sum(s.calibrated_items or 0 for s in tryout_stats)
+ overall_percentage = (
+ (total_calibrated / total_items * 100) if total_items > 0 else 0
+ )
+
+ # Build tryout rows
+ tryout_rows = []
+ for stat in tryout_stats:
+ total = stat.total_items or 0
+ calibrated = stat.calibrated_items or 0
+ percentage = (calibrated / total * 100) if total > 0 else 0
+
+ if percentage >= 90:
+ status = '
✅ Ready'
+ elif percentage >= 50:
+ status = '
⚠️ Partial'
+ else:
+ status = '
❌ Needs Data'
+
+ # Calculate bar width
+ bar_width = min(100, percentage)
+
+ tryout_rows.append(f"""
+
+ | {escape(stat.name or stat.tryout_id)} |
+ {total} |
+ {calibrated} |
+
+
+ |
+ {status} |
+
+ """)
+
+ body = f"""
+
+
📖 What is Question Quality?
+
Questions become "calibrated" after many students answer them. Well-calibrated questions give accurate student scores.
+
How it works: When 100+ students answer a question, we can calculate its true difficulty (p-value) and use it for adaptive testing.
+
+
+
+
+
+
{total_calibrated} of {total_items} questions calibrated
+
+
+
📋 By Exam
+
+
+
+ | Exam Name |
+ Total Questions |
+ Calibrated |
+ Progress |
+ Status |
+
+
+
+ {"".join(tryout_rows) if tryout_rows else '| No exams with questions yet. |
'}
+
+
+
+
+
+
+ """
+
+ return _render_admin_page(request, "Question Quality", "📊 Question Quality", body)
+
+
+@router.get("/exams", include_in_schema=False)
+async def exams_view(request: Request, db: AsyncSession = Depends(get_db)):
+ """Exams overview - list all exams with human-friendly display and visual cards."""
+ admin = await _current_admin(request)
+ if not admin:
+ return _login_redirect()
+
+ # Get all tryouts with stats
+ result = await db.execute(
+ select(Tryout)
+ .options(selectinload(Tryout.stats))
+ .order_by(Tryout.created_at.desc())
+ )
+ tryouts = list(result.scalars().all())
+
+ # Get summary stats
+ total_tryouts = len(tryouts)
+ total_participants = sum(
+ s.stats.participant_count if s.stats else 0 for s in tryouts
+ )
+ total_items_result = await db.execute(select(func.count(Item.id)))
+ total_items = total_items_result.scalar() or 0
+
+ # Build exam cards
+ exam_cards = []
+ for tryout in tryouts:
+ stats = tryout.stats
+ participant_count = stats.participant_count if stats else 0
+ avg_nm = stats.rataan if stats else None
+ std_nm = stats.std if stats else None
+ min_nm = stats.minimum if stats else None
+ max_nm = stats.maximum if stats else None
+
+ # Get item count
+ items_result = await db.execute(
+ select(func.count(Item.id)).where(
+ Item.tryout_id == tryout.tryout_id, Item.website_id == tryout.website_id
+ )
+ )
+ item_count = items_result.scalar() or 0
+
+ # Get calibrated items count
+ calibrated_result = await db.execute(
+ select(func.count(Item.id)).where(
+ Item.tryout_id == tryout.tryout_id,
+ Item.website_id == tryout.website_id,
+ Item.calibrated == True,
+ )
+ )
+ calibrated_count = calibrated_result.scalar() or 0
+
+ # Scoring mode badge with colors
+ mode_colors = {
+ "ctt": ("CTT", "background: #dbeafe; color: #1e40af;", "📐"),
+ "irt": ("IRT", "background: #fce7f3; color: #9d174d;", "📈"),
+ "hybrid": ("Hybrid", "background: #fef3c7; color: #92400e;", "🔄"),
+ }
+ mode_info = mode_colors.get(
+ tryout.scoring_mode, (tryout.scoring_mode.upper(), "", "📋")
+ )
+ mode_badge = f'
{mode_info[2]} {mode_info[0]}'
+
+ # Calibration progress
+ calibration_pct = (calibrated_count / item_count * 100) if item_count > 0 else 0
+ calibration_color = (
+ "#10b981"
+ if calibration_pct >= 90
+ else "#f59e0b"
+ if calibration_pct >= 50
+ else "#ef4444"
+ )
+
+ exam_cards.append(f"""
+
+
+
ID: {escape(tryout.tryout_id)}
+
+
+
+
📝
+
+ {item_count}
+ Questions
+
+
+
+
👥
+
+ {participant_count}
+ Students
+
+
+
+
📊
+
+ {"N/A" if avg_nm is None else f"{avg_nm:.0f}"}
+ Avg Score
+
+
+
+
+
+
+
+
+ Score Range:
+ {"N/A" if min_nm is None else f"{min_nm:.0f}"} - {"N/A" if max_nm is None else f"{max_nm:.0f}"}
+
+
+ Std Dev:
+ {"N/A" if std_nm is None else f"{std_nm:.1f}"}
+
+
+
+
+
+ """)
+
+ # Summary cards
+ summary_html = f"""
+
+
+
📋
+
+ {total_tryouts}
+ Total Exams
+
+
+
+
👥
+
+ {total_participants}
+ Total Students
+
+
+
+
📝
+
+ {total_items}
+ Total Questions
+
+
+
+ """
+
+ body = f"""
+
View and manage your exams. Each exam shows student statistics and calibration progress.
+
+ {summary_html}
+
+
All Exams
+
+
+ {"".join(exam_cards) if exam_cards else '
'}
+
+
+
+ """
+
+ return _render_admin_page(request, "Exams", "📋 Exams", body)
+
+
+@router.get("/reports", include_in_schema=False)
+async def reports_view(request: Request, db: AsyncSession = Depends(get_db)):
+ """Reports dashboard - human-friendly report access with quick stats."""
+ admin = await _current_admin(request)
+ if not admin:
+ return _login_redirect()
+
+ # Get quick stats for the overview
+ items_result = await db.execute(select(func.count(Item.id)))
+ total_items = items_result.scalar() or 0
+
+ calibrated_result = await db.execute(
+ select(func.count(Item.id)).where(Item.calibrated == True)
+ )
+ calibrated_items = calibrated_result.scalar() or 0
+
+ sessions_result = await db.execute(select(func.count(Session.id)))
+ total_sessions = sessions_result.scalar() or 0
+
+ calibration_pct = (calibrated_items / total_items * 100) if total_items > 0 else 0
+
+ body = f"""
+
Access detailed analysis reports for your exams, questions, and students.
+
+
+
+
📝
+
+ {total_items}
+ Total Questions
+
+
+
+
✅
+
+ {calibrated_items}
+ Calibrated ({calibration_pct:.0f}%)
+
+
+
+
📋
+
+ {total_sessions}
+ Student Sessions
+
+
+
+
+
Analysis Reports
+
+
+
+
Quick Actions
+
+
+
+
+ """
+
+ return _render_admin_page(request, "Reports", "📈 Reports", body)
+
+
+@router.get("/settings", include_in_schema=False)
+async def settings_view(request: Request, db: AsyncSession = Depends(get_db)):
+ """Settings dashboard - access to configuration pages."""
+ admin = await _current_admin(request)
+ if not admin:
+ return _login_redirect()
+
+ body = f"""
+
+
+
Configuration
+
+
+
Account
+
+
+
System Information
+
+
+
+
🚀
+
+ Version
+ 1.0.0
+
+
+
+
⚡
+
+ Framework
+ FastAPI
+
+
+
+
💾
+
+ Database
+ PostgreSQL
+
+
+
+
🔄
+
+ Session Timeout
+ {settings.ADMIN_SESSION_EXPIRE_SECONDS}s
+
+
+
+
+
+
+ """
+
+ return _render_admin_page(request, "Settings", "⚙️ Settings", body)
+
+
+
+
+# ============================================================
+# LEGACY ROUTES (backward compatibility)
+# ============================================================
+
+
@router.get("/hierarchy", include_in_schema=False)
async def hierarchy_view(request: Request, db: AsyncSession = Depends(get_db)):
admin = await _current_admin(request)
@@ -2035,7 +3884,9 @@ async def snapshot_questions_view(
)
return _render_admin_page(request, "Tryout Import", "Tryout Import", body)
- questions, promoted_items_by_slot, _ = await _load_snapshot_question_context(snapshot, db)
+ questions, promoted_items_by_slot, _ = await _load_snapshot_question_context(
+ snapshot, db
+ )
body = _snapshot_questions_body(snapshot, questions, promoted_items_by_slot)
return _render_admin_page(request, "Snapshot Questions", "Snapshot Questions", body)
@@ -2063,14 +3914,18 @@ async def snapshot_question_promote_bulk(
return _render_admin_page(request, "Tryout Import", "Tryout Import", body)
if not snapshot_question_ids:
- questions, promoted_items_by_slot, _ = await _load_snapshot_question_context(snapshot, db)
+ questions, promoted_items_by_slot, _ = await _load_snapshot_question_context(
+ snapshot, db
+ )
body = _snapshot_questions_body(
snapshot,
questions,
promoted_items_by_slot,
error="Select at least one snapshot question to promote.",
)
- return _render_admin_page(request, "Snapshot Questions", "Snapshot Questions", body)
+ return _render_admin_page(
+ request, "Snapshot Questions", "Snapshot Questions", body
+ )
question_result = await db.execute(
select(TryoutSnapshotQuestion).where(
@@ -2097,21 +3952,27 @@ async def snapshot_question_promote_bulk(
await db.commit()
- questions, promoted_items_by_slot, _ = await _load_snapshot_question_context(snapshot, db)
+ questions, promoted_items_by_slot, _ = await _load_snapshot_question_context(
+ snapshot, db
+ )
success_parts = []
if created_items:
success_parts.append(f"created {len(created_items)} item(s)")
if existing_items:
success_parts.append(f"reused {len(existing_items)} existing item(s)")
if missing_option_count:
- success_parts.append(f"skipped {missing_option_count} question(s) with missing option text")
+ success_parts.append(
+ f"skipped {missing_option_count} question(s) with missing option text"
+ )
if mismatch_count:
success_parts.append(f"skipped {mismatch_count} mismatched question(s)")
success_message = "Bulk promote finished: " + ", ".join(success_parts) + "."
if created_items:
success_message += f" Latest basis item ID: {created_items[-1].id}."
- body = _snapshot_questions_body(snapshot, questions, promoted_items_by_slot, success=success_message)
+ body = _snapshot_questions_body(
+ snapshot, questions, promoted_items_by_slot, success=success_message
+ )
return _render_admin_page(request, "Snapshot Questions", "Snapshot Questions", body)
@@ -2121,7 +3982,9 @@ async def calibration_status_view(request: Request, db: AsyncSession = Depends(g
if not admin:
return _login_redirect()
- result = await db.execute(select(Tryout.tryout_id, Tryout.name, Tryout.website_id).order_by(Tryout.id))
+ result = await db.execute(
+ select(Tryout.tryout_id, Tryout.name, Tryout.website_id).order_by(Tryout.id)
+ )
tryouts = result.all()
rows = []
@@ -2133,13 +3996,20 @@ async def calibration_status_view(request: Request, db: AsyncSession = Depends(g
name,
status["total_items"],
status["calibrated_items"],
- f'{status["calibration_percentage"]:.2f}%',
+ f"{status['calibration_percentage']:.2f}%",
"Yes" if status["ready_for_irt"] else "No",
]
)
body = _table(
- ["Tryout ID", "Name", "Total Items", "Calibrated", "Calibration %", "Ready for IRT"],
+ [
+ "Tryout ID",
+ "Name",
+ "Total Items",
+ "Calibrated",
+ "Calibration %",
+ "Ready for IRT",
+ ],
rows,
)
return _render_admin_page(request, "Calibration Status", "Calibration Status", body)
@@ -2156,12 +4026,16 @@ async def item_statistics_view(request: Request, db: AsyncSession = Depends(get_
rows = []
for level in levels:
- item_result = await db.execute(select(Item).where(Item.level == level).order_by(Item.slot).limit(10))
+ item_result = await db.execute(
+ select(Item).where(Item.level == level).order_by(Item.slot).limit(10)
+ )
items = item_result.scalars().all()
total_responses = sum(item.calibration_sample_size or 0 for item in items)
calibrated_count = sum(1 for item in items if item.calibrated)
calibration_percentage = (calibrated_count / len(items) * 100) if items else 0
- avg_correctness = sum(item.ctt_p or 0 for item in items) / len(items) if items else 0
+ avg_correctness = (
+ sum(item.ctt_p or 0 for item in items) / len(items) if items else 0
+ )
rows.append(
[
level,
@@ -2174,7 +4048,14 @@ async def item_statistics_view(request: Request, db: AsyncSession = Depends(get_
)
body = _table(
- ["Level", "Total Items", "Calibrated", "Calibration %", "Responses", "Avg Correctness"],
+ [
+ "Level",
+ "Total Items",
+ "Calibrated",
+ "Calibration %",
+ "Responses",
+ "Avg Correctness",
+ ],
rows,
)
return _render_admin_page(request, "Item Statistics", "Item Statistics", body)
@@ -2186,7 +4067,9 @@ async def session_overview_view(request: Request, db: AsyncSession = Depends(get
if not admin:
return _login_redirect()
- result = await db.execute(select(Session).order_by(Session.created_at.desc()).limit(50))
+ result = await db.execute(
+ select(Session).order_by(Session.created_at.desc()).limit(50)
+ )
sessions = result.scalars().all()
rows = [
@@ -2204,7 +4087,17 @@ async def session_overview_view(request: Request, db: AsyncSession = Depends(get
for session in sessions
]
body = _table(
- ["Session ID", "WP User", "Tryout", "Completed", "Mode", "Benar", "NM", "NN", "Theta"],
+ [
+ "Session ID",
+ "WP User",
+ "Tryout",
+ "Completed",
+ "Mode",
+ "Benar",
+ "NM",
+ "NN",
+ "Theta",
+ ],
rows,
)
return _render_admin_page(request, "Session Overview", "Session Overview", body)
@@ -2249,7 +4142,11 @@ async def basis_item_workspace_view(
}
basis_item = await db.get(Item, basis_item_id)
- if basis_item is None or basis_item.generated_by == "ai" or basis_item.level != "sedang":
+ if (
+ basis_item is None
+ or basis_item.generated_by == "ai"
+ or basis_item.level != "sedang"
+ ):
result = await db.execute(
select(Item)
.where(Item.level == "sedang", Item.generated_by != "ai")
@@ -2267,7 +4164,8 @@ async def basis_item_workspace_view(
)
runs = list(run_result.scalars().all())
variant_result = await db.execute(
- select(Item).where(Item.generated_by == "ai", Item.basis_item_id == basis_item.id)
+ select(Item)
+ .where(Item.generated_by == "ai", Item.basis_item_id == basis_item.id)
.order_by(Item.created_at.desc(), Item.id.desc())
.limit(300)
)
@@ -2299,7 +4197,8 @@ async def basis_item_workspace_view(
family_stats,
filters,
)
- return _render_admin_page(request,
+ return _render_admin_page(
+ request,
f"Basis Item #{basis_item.id}",
f"Basis Item Workspace #{basis_item.id}",
body,
@@ -2324,8 +4223,14 @@ async def basis_item_generate_submit(
filters = {"status": "", "level": "", "run_id": "", "min_frequency": ""}
basis_item = await db.get(Item, basis_item_id)
- if basis_item is None or basis_item.generated_by == "ai" or basis_item.level != "sedang":
- return RedirectResponse(url="/admin/basis-items", status_code=HTTP_303_SEE_OTHER)
+ if (
+ basis_item is None
+ or basis_item.generated_by == "ai"
+ or basis_item.level != "sedang"
+ ):
+ return RedirectResponse(
+ url="/admin/basis-items", status_code=HTTP_303_SEE_OTHER
+ )
# Llama-only policy for production quality consistency.
ai_model = settings.OPENROUTER_MODEL_LLAMA
@@ -2347,7 +4252,9 @@ async def basis_item_generate_submit(
)
runs = list(run_result.scalars().all())
variants = list(variant_result.scalars().all())
- usage_metrics, family_stats = await _family_usage_stats(db, basis_item, variants)
+ usage_metrics, family_stats = await _family_usage_stats(
+ db, basis_item, variants
+ )
body = _basis_item_workspace_body(
basis_item,
runs,
@@ -2363,16 +4270,21 @@ async def basis_item_generate_submit(
include_note_for_admin=note_for_admin,
include_note_in_prompt=note_in_prompt,
)
- return _render_admin_page(request,
+ return _render_admin_page(
+ request,
f"Basis Item #{basis_item.id}",
f"Basis Item Workspace #{basis_item.id}",
body,
)
if target_level not in {"mudah", "sulit"}:
- return RedirectResponse(url=f"/admin/basis-items/{basis_item.id}", status_code=HTTP_303_SEE_OTHER)
+ return RedirectResponse(
+ url=f"/admin/basis-items/{basis_item.id}", status_code=HTTP_303_SEE_OTHER
+ )
if generation_count < 1 or generation_count > 50:
- return RedirectResponse(url=f"/admin/basis-items/{basis_item.id}", status_code=HTTP_303_SEE_OTHER)
+ return RedirectResponse(
+ url=f"/admin/basis-items/{basis_item.id}", status_code=HTTP_303_SEE_OTHER
+ )
run_id = await create_generation_run(
basis_item_id=basis_item.id,
@@ -2456,7 +4368,8 @@ async def basis_item_generate_submit(
include_note_for_admin=note_for_admin,
include_note_in_prompt=note_in_prompt,
)
- return _render_admin_page(request,
+ return _render_admin_page(
+ request,
f"Basis Item #{basis_item.id}",
f"Basis Item Workspace #{basis_item.id}",
body,
@@ -2478,7 +4391,9 @@ async def basis_item_review_bulk(
basis_item = await db.get(Item, basis_item_id)
if basis_item is None:
- return RedirectResponse(url="/admin/basis-items", status_code=HTTP_303_SEE_OTHER)
+ return RedirectResponse(
+ url="/admin/basis-items", status_code=HTTP_303_SEE_OTHER
+ )
valid_actions = {"approved", "rejected", "archived", "stale", "active"}
if action in valid_actions and item_ids:
@@ -2521,7 +4436,8 @@ async def basis_item_review_bulk(
filters,
success=f"Applied status '{action}' to selected variants.",
)
- return _render_admin_page(request,
+ return _render_admin_page(
+ request,
f"Basis Item #{basis_item.id}",
f"Basis Item Workspace #{basis_item.id}",
body,
@@ -2531,8 +4447,7 @@ async def basis_item_review_bulk(
AI_PLAYGROUND_TABS = (
("generate", "Generate"),
("review", "Review Queue"),
- ("runs", "Runs"),
- ("basis", "Basis Items"),
+ ("runs", "Batches"),
)
AI_VARIANT_STATUSES = ("draft", "approved", "active", "rejected", "archived", "stale")
AI_VARIANT_LEVELS = ("mudah", "sulit")
@@ -2542,21 +4457,23 @@ def _selected_option(value: str, selected_value: str) -> str:
return "selected" if value == selected_value else ""
-def _ai_tab_nav(active_tab: str) -> str:
+def _ai_tab_nav(item_id: int, active_tab: str) -> str:
links = []
for tab, label in AI_PLAYGROUND_TABS:
active_class = "active" if tab == active_tab else ""
aria = ' aria-current="page"' if tab == active_tab else ""
links.append(
- f'
{escape(label)}'
+ f'
{escape(label)}'
)
- return f'
'
+ return f'
'
def _status_pill(status: str | None) -> str:
value = status or "unknown"
css_value = re.sub(r"[^a-z0-9_-]+", "-", value.lower())
- return f'
{escape(value)}'
+ return (
+ f'
{escape(value)}'
+ )
def _ai_status_strip(
@@ -2577,7 +4494,7 @@ def _ai_status_strip(
OpenRouter{"Yes" if key_configured else "No"}
AI Items{stats.get("total_ai_items", 0)}
-
Latest Run{escape(latest_run)}
+
Latest Batch{escape(latest_run)}
Saved{escape(latest_saved)}
"""
@@ -2589,7 +4506,7 @@ def _ai_generation_summary(generation_summary: dict[str, Any] | None) -> str:
saved_item_ids = generation_summary.get("saved_item_ids") or []
return f"""
-
Run ID{generation_summary.get("run_id", "-")}
+
Batch ID{generation_summary.get("run_id", "-")}
Requested{generation_summary.get("requested_count", 0)}
Generated{generation_summary.get("generated_count", 0)}
Saved{len(saved_item_ids)}
@@ -2598,9 +4515,8 @@ def _ai_generation_summary(generation_summary: dict[str, Any] | None) -> str:
def _ai_generate_tab(
- basis_items: list[Item],
+ item: Item,
generation_summary: dict[str, Any] | None,
- basis_item_id: str,
target_level: str,
ai_model: str,
generation_count: str,
@@ -2608,100 +4524,60 @@ def _ai_generate_tab(
include_note_for_admin: bool,
include_note_in_prompt: bool,
) -> str:
- seed_callout = ""
- if not basis_items:
- seed_callout = """
-
No sedang basis items found yet.
-
- """
- selected_basis_id = str(basis_item_id or "")
- basis_options = ['
']
- for item in basis_items:
- item_id = str(item.id)
- selected = _selected_option(item_id, selected_basis_id)
- stem_preview = _truncate(_html_to_text(item.stem), 82)
- basis_options.append(
- f'
'
- )
+ full_stem = escape(_html_to_text(item.stem))
+ basis_selection_html = f"""
+
+
Basis Item Context
+
Tryout: {escape(str(item.tryout_id))} | Slot: {item.slot} | ID: #{item.id}
+
"{full_stem}"
+
+
+ """
return f"""
- {seed_callout}
{_ai_generation_summary(generation_summary)}
-