#!/usr/bin/env python3
"""
multi_course_runner.py
----------------------
Grades both courses in one command and routes students to TAs automatically.
Courses:
Course 1 — AI for Analytics (TA1 + TA2)
Course 2 — Foundation of Business Statistics (TA3 + TA4)
What it produces for each course:
• all_results.json — structured grades
• dashboard.html — per-course visual dashboard
• gold_standard_template.csv — fill with human scores for evaluation
• ta_assignments/ — per-TA student subsets (JSON + HTML)
Plus a combined TA overview HTML at: output/combined_overview.html
Usage (interactive):
python multi_course_runner.py
Usage (command-line, both courses):
python multi_course_runner.py \\
--course1-instructions path/to/ai_analytics_instructions.pdf \\
--course1-rubric path/to/ai_analytics_rubric.pdf \\
--course1-submissions path/to/ai_analytics_submissions/ \\
--course2-instructions path/to/stats_instructions.pdf \\
--course2-rubric path/to/stats_rubric.pdf \\
--course2-submissions path/to/stats_submissions/ \\
--output path/to/output_folder/ \\
--offset 3.5
Usage (single course only):
python multi_course_runner.py \\
--course1-instructions path/to/instructions.pdf \\
--course1-rubric path/to/rubric.pdf \\
--course1-submissions path/to/submissions/ \\
--output path/to/output_folder/
"""
from pathlib import Path
import os
# Load .env file
env_path = Path(".env")
if env_path.exists():
with open(env_path) as f:
for line in f:
line = line.strip()
if line and not line.startswith("#") and "=" in line:
key, _, value = line.partition("=")
os.environ[key.strip()] = value.strip()
import argparse
import json
import time
import re
from datetime import datetime
from typing import List, Dict, Optional
import anthropic
from document_reader import read_document, load_student_submissions
from privacy_processor import anonymize
from rag_retriever import build_rag_evidence
from rubric_parser import parse_rubric, criteria_summary, build_rubric_prompt_section
from evaluator import create_gold_standard_template
from dashboard import build_dashboard
# ─── TA Configuration ──────────────────────────────────────────────────────────
# Adjust names and split sizes to match your actual TAs.
# Students are assigned sequentially — first N go to TA1, next N to TA2, etc.
COURSE_CONFIG = {
"AI for Analytics": {
"code": "CAI_3801",
"tas": ["TA1", "TA2"],
"split": "equal", # "equal" | integer (e.g. 100 = first 100 go to TA1)
},
"Foundation of Business Statistics": {
"code": "ISM_6145",
"tas": ["TA3", "TA4"],
"split": "equal",
},
}
# ─── Grading prompt templates (same as calibrated_grader.py) ──────────────────
_SYSTEM_TEMPLATE = """
You are an expert Teaching Assistant grading student submissions.
Your goal is to provide CONSISTENT rubric-calibrated grading that matches
how a fair human TA would grade — not stricter.
CALIBRATION GUIDANCE (based on observed human TA grading patterns):
- A student who attempts all sections and shows reasonable effort typically earns 80–95% of total points.
- Only award 0 for a criterion when that section is entirely absent from the submission.
- Award 50–75% of points when the student made a clear attempt but had gaps.
- Award 80–100% of points when the work is present and mostly correct.
- Do NOT deduct points for minor wording differences, informal phrasing, or imperfect formatting.
IMPORTANT RULES:
- Be lenient and constructive — match the generosity of a supportive human TA.
- Focus mainly on whether required sections are present and the student attempted the task.
- Award partial credit generously for reasonable attempts.
- Do NOT harshly penalize wording or writing style.
- Use the SAME grading interpretation for all students.
- Do not invent missing work.
- Grade ONLY the evidence provided — do not assume content not shown.
Assignment: {assignment_name}
Course: {course_name}
Rubric:
{rubric_section}
For EACH criterion:
1. State what was completed (based on the provided evidence)
2. State what is missing or weak
3. Give ONE improvement suggestion
Return ONLY valid JSON — no markdown fences, no extra text.
JSON FORMAT:
{{
"student_id": "",
"total_score": 0,
"max_score": {max_score},
"percentage": 0.0,
"criteria": [
{{
"name": "",
"max_points": 0,
"awarded_points": 0,
"completed": "",
"missing_or_weak": "",
"suggestion": ""
}}
],
"overall_feedback": "",
"consistency_notes": ""
}}
"""
_USER_TEMPLATE = """
ASSIGNMENT INSTRUCTIONS:
{instructions}
RUBRIC-EXTRACTED EVIDENCE:
(Only the most relevant portions of the student's anonymized submission
are shown below, organized by rubric criterion.)
{rag_evidence}
Evaluate ONLY the evidence shown above. If a criterion's evidence section
is empty or unclear, award 0 and explain what was expected.
Return ONLY JSON.
"""
# ─── Core grader (mirrors CalibratedGrader from calibrated_grader.py) ─────────
def _apply_calibration(result: dict, criteria: List[Dict], calibration_offset: float) -> dict:
"""Apply calibration offset proportionally across criteria."""
max_score = result.get("max_score", sum(c["max_points"] for c in criteria))
crit_list = result.get("criteria", [])
crit_total = sum(c.get("max_points", 0) for c in crit_list)
for c in crit_list:
cmax = c.get("max_points", 0)
prop = calibration_offset * (cmax / crit_total) if crit_total > 0 else 0
raw = c.get("awarded_points", 0)
c["awarded_points"] = round(min(cmax, raw + prop), 1)
c["awarded_points_raw"] = raw
raw_total = result.get("total_score", 0)
calibrated = round(min(max_score, raw_total + calibration_offset), 1)
result["total_score"] = calibrated
result["total_score_raw"] = raw_total
result["calibration_offset_applied"] = calibration_offset
result["percentage"] = round((calibrated / max_score) * 100, 1) if max_score else 0
pct = result["percentage"]
if pct >= 93: result["letter_grade"] = "A"
elif pct >= 90: result["letter_grade"] = "A-"
elif pct >= 87: result["letter_grade"] = "B+"
elif pct >= 83: result["letter_grade"] = "B"
elif pct >= 80: result["letter_grade"] = "B-"
elif pct >= 77: result["letter_grade"] = "C+"
elif pct >= 73: result["letter_grade"] = "C"
elif pct >= 70: result["letter_grade"] = "C-"
elif pct >= 60: result["letter_grade"] = "D"
else: result["letter_grade"] = "F"
return result
def grade_submission(
client: anthropic.Anthropic,
model: str,
criteria: List[Dict],
assignment_name: str,
course_name: str,
instructions_text: str,
submission: Dict,
student_id: str,
calibration_offset: float,
) -> Dict:
"""Grade a single submission: anonymize → RAG → Claude → calibrate."""
anon_text, anon_log = anonymize(submission["text"], known_name=submission["name"])
if anon_log:
print(f" [privacy] {student_id}: {', '.join(anon_log)}")
rag_evidence = build_rag_evidence(anon_text, criteria=criteria, top_n=3)
max_score = sum(c["max_points"] for c in criteria)
system = _SYSTEM_TEMPLATE.format(
assignment_name=assignment_name,
course_name=course_name,
rubric_section=build_rubric_prompt_section(criteria),
max_score=max_score,
)
user = _USER_TEMPLATE.format(
instructions=instructions_text[:3000],
rag_evidence=rag_evidence,
)
response = client.messages.create(
model=model,
max_tokens=2500,
system=system,
messages=[{"role": "user", "content": user}],
)
raw_text = response.content[0].text.strip()
cleaned = re.sub(r"^```(?:json)?\s*", "", raw_text, flags=re.MULTILINE)
cleaned = re.sub(r"\s*```$", "", cleaned, flags=re.MULTILINE)
m = re.search(r"\{.*\}", cleaned, re.DOTALL)
if m:
cleaned = m.group()
result = json.loads(cleaned)
result["student_id"] = student_id
result["student_name"] = submission["name"]
result["source_file"] = submission.get("file", "")
result["assignment_name"] = assignment_name
result["course_name"] = course_name
result["rubric_criteria"] = criteria
return _apply_calibration(result, criteria, calibration_offset)
# ─── TA assignment routing ────────────────────────────────────────────────────
def assign_tas(results: List[Dict], tas: List[str], split) -> Dict[str, List[Dict]]:
"""
Assign graded results to TAs.
split = "equal" → split evenly
split = int N → first N to TA1, rest to TA2 (for 2 TAs)
"""
n = len(results)
assignments: Dict[str, List[Dict]] = {ta: [] for ta in tas}
if split == "equal":
chunk = max(1, -(-n // len(tas))) # ceiling division
for i, r in enumerate(results):
ta = tas[min(i // chunk, len(tas) - 1)]
assignments[ta].append(r)
elif isinstance(split, int):
for i, r in enumerate(results):
ta = tas[0] if i < split else tas[1]
assignments[ta].append(r)
return assignments
# ─── Per-TA HTML mini-dashboard ───────────────────────────────────────────────
def build_ta_html(ta_name: str, course_name: str, results: List[Dict]) -> str:
"""Build a simple HTML page listing this TA's assigned students."""
rows = ""
for r in results:
pct = r.get("percentage", 0)
color = "#22c55e" if pct >= 80 else "#3b82f6" if pct >= 60 else "#ef4444"
rows += f"""
"""
scores = [r.get("percentage", 0) for r in results]
avg = round(sum(scores) / len(scores), 1) if scores else 0
pending = len(results)
return f"""
{ta_name} — {course_name}
📋 {ta_name} — Review Queue
{course_name} · Generated {datetime.now().strftime('%B %d, %Y at %I:%M %p')}
{len(results)}
Students Assigned
{avg}%
Avg Score
{pending}
Pending Review
Students — Review AI Grades Below
ID
Student
Score
%
Grade
AI Feedback Preview
{rows}
⚠️ These are AI-generated grades. Please review each student and override any grades you disagree with before finalizing.
"""
# ─── Combined overview dashboard ──────────────────────────────────────────────
def build_combined_overview(course_summaries: List[Dict]) -> str:
"""Build a single HTML page summarizing both courses for the lead TA / professor."""
course_cards = ""
for cs in course_summaries:
results = cs["results"]
scores = [r.get("percentage", 0) for r in results]
avg = round(sum(scores) / len(scores), 1) if scores else 0
passing = sum(1 for s in scores if s >= 60)
dist = {"A": 0, "B": 0, "C": 0, "D/F": 0}
for s in scores:
if s >= 90: dist["A"] += 1
elif s >= 80: dist["B"] += 1
elif s >= 70: dist["C"] += 1
else: dist["D/F"] += 1
ta_rows = ""
for ta, ta_results in cs["ta_assignments"].items():
ta_scores = [r.get("percentage", 0) for r in ta_results]
ta_avg = round(sum(ta_scores) / len(ta_scores), 1) if ta_scores else 0
ta_rows += f"
{ta}
{len(ta_results)} students
{ta_avg}%
"
color = "#22c55e" if avg >= 80 else "#3b82f6" if avg >= 60 else "#ef4444"
course_cards += f"""