Benjamin Mako Hill
3814229f49
Publish review_attendance.py, the email and failed-message templates, and the example config used to track CSSS 590 seminar attendance and send warning DMs through the Canvas Conversations API. README walks through the weekly workflow, the Roll Call CSV quirks worth knowing about, and what must stay out of git. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
370 lines
14 KiB
Python
Executable File
370 lines
14 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Review CSSS 590 attendance and optionally send warnings to at-risk students.
|
||
|
||
Reads the most recent Roll Call CSV in this directory (matching
|
||
``attendance_reports_*.csv``), pulls the active enrollment list from Canvas,
|
||
and prints an attendance grid plus an X/N totals list. With
|
||
``--send-warning`` it additionally shows which students at or past the
|
||
``absences_threshold`` would be warned, and the rendered message for each.
|
||
With ``--send-warning -f`` it actually POSTs the warnings via the Canvas
|
||
Conversations API and appends each one to ``students_contacted.tsv``.
|
||
|
||
Usage:
|
||
review_attendance.py # grid + totals only
|
||
review_attendance.py --send-warning # also preview warning messages
|
||
review_attendance.py --send-warning -f # actually send the warnings
|
||
review_attendance.py --csv FILE # override which CSV to read
|
||
|
||
The template is evaluated as a Python f-string, so any expression that
|
||
references the variables below is valid. Example for "must attend N (or
|
||
all) of the remaining sessions":
|
||
|
||
{"all" if sessions_required - attended >= sessions_remaining
|
||
else sessions_required - attended}
|
||
|
||
Template variables:
|
||
{name} full student name
|
||
{first_name} first whitespace-separated token of the name
|
||
{attended} number of sessions attended so far
|
||
{sessions_held} total sessions held to date
|
||
{sessions_remaining} sessions still to come this quarter
|
||
{sessions_required} sessions required per quarter (from config)
|
||
{sessions_expected} sessions expected per quarter (from config)
|
||
"""
|
||
|
||
import argparse
|
||
import csv
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import tomllib
|
||
from datetime import date
|
||
from pathlib import Path
|
||
|
||
import requests
|
||
|
||
HERE = Path(__file__).parent
|
||
CONFIG_PATH = HERE / "config.toml"
|
||
|
||
# Canvas Roll Call CSVs have a trailing empty field on data rows but not the
|
||
# header, so DictReader gets confused. Read with these explicit field names.
|
||
CSV_FIELDS = [
|
||
"Course ID", "SIS Course ID", "Course Code", "Course Name",
|
||
"Section Name", "Section ID", "SIS Section ID", "Teacher ID",
|
||
"Teacher Name", "Student ID", "Student Name", "Class Date",
|
||
"Attendance", "Timestamp", "Extra",
|
||
]
|
||
|
||
|
||
def load_config():
|
||
with open(CONFIG_PATH, "rb") as f:
|
||
return tomllib.load(f)
|
||
|
||
|
||
def newest_attendance_csv():
|
||
files = sorted(HERE.glob("attendance_reports_*.csv"))
|
||
if not files:
|
||
sys.exit("No attendance_reports_*.csv found in this directory.")
|
||
return files[-1]
|
||
|
||
|
||
def parse_attendance(path):
|
||
"""Return students, sessions, attended, present_set parsed from the CSV.
|
||
|
||
students: dict[student_id -> name]
|
||
sessions: sorted list of class-date strings
|
||
attended: dict[student_id -> int]
|
||
present_set: set of (student_id, class_date) for present entries
|
||
|
||
Absences are derived as ``len(sessions) - attended[sid]``; the CSV
|
||
omits rows entirely for sessions where Roll Call wasn't marked for a
|
||
student, so counting explicit "absent" rows undercounts.
|
||
"""
|
||
students = {}
|
||
sessions = set()
|
||
attended = {}
|
||
present_set = set()
|
||
with open(path, newline="") as f:
|
||
reader = csv.reader(f)
|
||
next(reader) # skip header — we use our own field list
|
||
for row in reader:
|
||
r = dict(zip(CSV_FIELDS, row))
|
||
sid = r["Student ID"].strip()
|
||
name = r["Student Name"].strip()
|
||
students[sid] = name
|
||
sessions.add(r["Class Date"])
|
||
if r["Attendance"] == "present":
|
||
attended[sid] = attended.get(sid, 0) + 1
|
||
present_set.add((sid, r["Class Date"]))
|
||
return students, sorted(sessions), attended, present_set
|
||
|
||
|
||
def read_contacted(tsv_path):
|
||
"""Return dict[student_id -> emailed_date] for previously-warned students."""
|
||
contacted = {}
|
||
if not tsv_path.exists():
|
||
return contacted
|
||
with open(tsv_path, newline="") as f:
|
||
reader = csv.DictReader(f, delimiter="\t")
|
||
for row in reader:
|
||
sid = (row.get("Student ID") or "").strip()
|
||
if sid:
|
||
contacted[sid] = (row.get("Emailed Date") or "").strip()
|
||
return contacted
|
||
|
||
|
||
def append_contacted(tsv_path, student_id, name, when):
|
||
new_file = not tsv_path.exists() or tsv_path.stat().st_size == 0
|
||
with open(tsv_path, "a", newline="") as f:
|
||
w = csv.writer(f, delimiter="\t")
|
||
if new_file:
|
||
w.writerow(["Student ID", "Student Name", "Emailed Date"])
|
||
w.writerow([student_id, name, when])
|
||
|
||
|
||
def get_canvas_token(cfg):
|
||
"""Return the Canvas API token.
|
||
|
||
Resolution order:
|
||
1. CANVAS_TOKEN env var (raw token).
|
||
2. ``token`` (raw token) in [canvas] from config.toml.
|
||
3. ``token_command`` (argv list) in [canvas] from config.toml, run
|
||
as a subprocess; stdout (stripped) is the token.
|
||
"""
|
||
if (env := os.environ.get("CANVAS_TOKEN")):
|
||
return env.strip()
|
||
raw = cfg["canvas"].get("token")
|
||
if raw:
|
||
return raw.strip()
|
||
cmd = cfg["canvas"].get("token_command")
|
||
if not cmd:
|
||
sys.exit("Canvas token not configured: set CANVAS_TOKEN, or add "
|
||
"[canvas].token or [canvas].token_command to config.toml.")
|
||
r = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
||
return r.stdout.strip()
|
||
|
||
|
||
def get_enrolled_students(base_url, token, course_id):
|
||
"""Return dict[str(user_id) -> name] for active StudentEnrollments."""
|
||
url = f"{base_url}/api/v1/courses/{course_id}/enrollments"
|
||
params = {"type[]": "StudentEnrollment", "state[]": "active", "per_page": 100}
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
enrolled = {}
|
||
while url:
|
||
r = requests.get(url, params=params, headers=headers, timeout=30)
|
||
r.raise_for_status()
|
||
for e in r.json():
|
||
enrolled[str(e["user_id"])] = e["user"]["name"]
|
||
params = None # subsequent paginated URLs already carry the query
|
||
url = None
|
||
link = r.headers.get("Link", "")
|
||
for part in link.split(","):
|
||
if 'rel="next"' in part:
|
||
url = part.split(";", 1)[0].strip().strip("<>")
|
||
break
|
||
return enrolled
|
||
|
||
|
||
def render_template(template, env):
|
||
"""Render the template as a Python f-string with env in scope.
|
||
|
||
Allows the template author to write inline expressions like
|
||
``{"all" if must_attend >= sessions_remaining else must_attend}``.
|
||
Mako writes the template himself, so eval is acceptable.
|
||
"""
|
||
if '"""' in template:
|
||
raise ValueError("Template cannot contain triple-quotes (the renderer "
|
||
"wraps the file in a triple-quoted f-string).")
|
||
return eval(f'f"""{template}"""', {"__builtins__": {}}, env)
|
||
|
||
|
||
def send_canvas_message(base_url, token, course_id, user_id, subject, body):
|
||
resp = requests.post(
|
||
f"{base_url}/api/v1/conversations",
|
||
headers={"Authorization": f"Bearer {token}"},
|
||
data={
|
||
"recipients[]": str(user_id),
|
||
"context_code": f"course_{course_id}",
|
||
"subject": subject,
|
||
"body": body,
|
||
"force_new": "true",
|
||
},
|
||
timeout=30,
|
||
)
|
||
resp.raise_for_status()
|
||
return resp.json()
|
||
|
||
|
||
def print_grid(students, sessions, present_set):
|
||
"""Print a students × sessions grid sorted by last name."""
|
||
by_last = sorted(students.items(), key=lambda kv: (kv[1].rsplit(" ", 1)[-1], kv[1]))
|
||
width = max(len(n) for _, n in by_last)
|
||
header = " " * (width + 2) + " ".join(d[5:] for d in sessions)
|
||
print(header)
|
||
for sid, name in by_last:
|
||
cells = ["P" if (sid, d) in present_set else "-" for d in sessions]
|
||
print(f"{name:<{width}} " + " ".join(cells))
|
||
print()
|
||
|
||
|
||
def print_totals(students, attended, sessions_held, contacted):
|
||
"""Print an X/N attendance list sorted by attendance desc, then name."""
|
||
rows = sorted(
|
||
((sid, students[sid], attended.get(sid, 0)) for sid in students),
|
||
key=lambda r: (-r[2], r[1]),
|
||
)
|
||
width = max(len(n) for _, n, _ in rows)
|
||
print("Totals (attended / sessions held):\n")
|
||
for sid, name, n in rows:
|
||
warned = f" [warned {contacted[sid]}]" if sid in contacted else ""
|
||
print(f" {name:<{width}} {n}/{sessions_held}{warned}")
|
||
print()
|
||
|
||
|
||
def print_final_report(students, attended, sessions_held, sessions_required, contacted):
|
||
"""Print PASS/FAIL groupings for the end-of-quarter report."""
|
||
rows = sorted(
|
||
((sid, students[sid], attended.get(sid, 0)) for sid in students),
|
||
key=lambda r: (-r[2], r[1]),
|
||
)
|
||
passing = [(sid, name, n) for sid, name, n in rows if n >= sessions_required]
|
||
failing = [(sid, name, n) for sid, name, n in rows if n < sessions_required]
|
||
width = max(len(n) for _, n, _ in rows)
|
||
|
||
print(f"Final report (required: {sessions_required}/{sessions_held}):\n")
|
||
print(f" PASS ({len(passing)}):")
|
||
for sid, name, n in passing:
|
||
warned = f" [warned {contacted[sid]}]" if sid in contacted else ""
|
||
print(f" {name:<{width}} {n}/{sessions_held}{warned}")
|
||
print(f"\n FAIL ({len(failing)}):")
|
||
for sid, name, n in failing:
|
||
warned = f" [warned {contacted[sid]}]" if sid in contacted else ""
|
||
print(f" {name:<{width}} {n}/{sessions_held}{warned}")
|
||
print()
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description=__doc__)
|
||
ap.add_argument("--send-warning", action="store_true",
|
||
help="show warning-message preview (and with -f, send them)")
|
||
ap.add_argument("-f", "--force", action="store_true",
|
||
help="with --send-warning, actually POST messages to Canvas")
|
||
ap.add_argument("--final-report", action="store_true",
|
||
help="print end-of-quarter PASS/FAIL summary")
|
||
ap.add_argument("--csv", type=Path,
|
||
help="path to attendance CSV (default: newest in this dir)")
|
||
args = ap.parse_args()
|
||
|
||
if args.force and not args.send_warning:
|
||
ap.error("-f/--force requires --send-warning")
|
||
if args.send_warning and args.final_report:
|
||
ap.error("--send-warning and --final-report are mutually exclusive")
|
||
|
||
cfg = load_config()
|
||
sessions_required = cfg["attendance"]["sessions_required"]
|
||
sessions_expected = cfg["attendance"]["sessions_expected"]
|
||
threshold = sessions_expected - sessions_required
|
||
contacted_log = HERE / cfg["contact"]["contacted_log"]
|
||
template_path = HERE / cfg["contact"]["template_file"]
|
||
subject = cfg["contact"]["subject"]
|
||
|
||
csv_path = args.csv or newest_attendance_csv()
|
||
print(f"Reading: {csv_path.name}")
|
||
|
||
token = get_canvas_token(cfg)
|
||
enrolled = get_enrolled_students(
|
||
cfg["canvas"]["base_url"], token, cfg["course"]["canvas_id"]
|
||
)
|
||
print(f"Canvas enrollment: {len(enrolled)} active student(s)\n")
|
||
|
||
_, sessions, attended, present_set = parse_attendance(csv_path)
|
||
sessions_held = len(sessions)
|
||
|
||
# Restrict to currently-enrolled students — dropped students still show
|
||
# up in the CSV from before they left.
|
||
students = dict(enrolled)
|
||
absences = {sid: sessions_held - attended.get(sid, 0) for sid in students}
|
||
|
||
print(f"Sessions held: {sessions_held} ({', '.join(sessions)})\n")
|
||
|
||
contacted = read_contacted(contacted_log)
|
||
|
||
print_grid(students, sessions, present_set)
|
||
print_totals(students, attended, sessions_held, contacted)
|
||
|
||
if args.final_report:
|
||
print_final_report(students, attended, sessions_held,
|
||
sessions_required, contacted)
|
||
return
|
||
|
||
if not args.send_warning:
|
||
return
|
||
|
||
print(f"Threshold: {threshold}+ absences "
|
||
f"(required {sessions_required}/{sessions_expected})\n")
|
||
targets = sorted(
|
||
[(sid, students[sid], n) for sid, n in absences.items()
|
||
if n >= threshold and sid not in contacted],
|
||
key=lambda x: (-x[2], x[1]),
|
||
)
|
||
|
||
if not targets:
|
||
print("No students to warn.")
|
||
already = [(sid, students.get(sid, "?")) for sid in contacted
|
||
if absences.get(sid, 0) >= threshold]
|
||
if already:
|
||
print("(Already warned: " +
|
||
", ".join(n for _, n in already) + ")")
|
||
return
|
||
|
||
template = template_path.read_text() if template_path.exists() else ""
|
||
template_ready = bool(template.strip())
|
||
if not template_ready:
|
||
print(f"[!] {template_path.name} is empty — preview only; -f will refuse.\n")
|
||
|
||
print(f"Will warn {len(targets)} student(s):\n")
|
||
for sid, name, n in targets:
|
||
print(f" - {name} (id={sid}, {attended.get(sid, 0)}/{sessions_held}, "
|
||
f"{n} absence{'s' if n != 1 else ''})")
|
||
print()
|
||
|
||
if args.force and not template_ready:
|
||
sys.exit(f"Refusing to send: {template_path.name} is empty.")
|
||
|
||
today = date.today().isoformat()
|
||
|
||
for sid, name, n in targets:
|
||
if template_ready:
|
||
env = {
|
||
"name": name,
|
||
"first_name": name.split()[0],
|
||
"attended": attended.get(sid, 0),
|
||
"sessions_held": sessions_held,
|
||
"sessions_remaining": sessions_expected - sessions_held,
|
||
"sessions_required": sessions_required,
|
||
"sessions_expected": sessions_expected,
|
||
}
|
||
body = render_template(template, env)
|
||
else:
|
||
body = "[TEMPLATE EMPTY]"
|
||
|
||
print(f"--- {name} (id={sid}) ---")
|
||
print(f"Subject: {subject}")
|
||
print(body)
|
||
print()
|
||
|
||
if args.force:
|
||
send_canvas_message(
|
||
cfg["canvas"]["base_url"], token,
|
||
cfg["course"]["canvas_id"], sid, subject, body,
|
||
)
|
||
append_contacted(contacted_log, sid, name, today)
|
||
print(f" -> sent, logged to {contacted_log.name}\n")
|
||
|
||
if not args.force:
|
||
print("Preview only. Re-run with --send-warning -f to actually send.")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|