feat(elections): design voting system

.env.example

@@ -10,3 +10,5 @@ MAIL_PASSWORD=your-gmail-app-password
 
 # Advent of Code session cookie for leaderboard API
 AOC_SESSION_COOKIE=your-aoc-session-cookie
+
+ADMIN_PASSWORD=codesoc-pass

.github/workflows/deploy.yml

@@ -48,6 +48,7 @@ jobs:
             echo "MAIL_PASSWORD=${{ secrets.MAIL_PASSWORD }}" >> .env
             echo "AOC_SESSION_COOKIE=${{ secrets.AOC_SESSION_COOKIE }}" >> .env
             echo "SU_KEY=${{ secrets.SU_KEY }}" >> .env
+            echo "ADMIN_PASSWORD=${{ secrets.ADMIN_PASSWORD }}" >> .env
 
       - name: Install dependencies
         if: success()

codesoc.py

@@ -1,14 +1,18 @@
-from flask import Flask, render_template, request, jsonify
+from flask import Flask, render_template, request, jsonify, redirect, url_for, abort, session
 from flask_migrate import Migrate
-from db_schema import db, ExecMember, BlogPost, Sponsor, SponsorNews
+from db_schema import db, ExecMember, BlogPost, Sponsor, SponsorNews, ElectionCandidate, ElectionVote, ElectionSettings
 from flask_mail import Mail, Message
 import click
 import os
+import json
+import xml.etree.ElementTree as ET
 from dotenv import load_dotenv
 from db_manager import DatabaseManager
 from schema_generator import generate_all_schemas, save_schemas_to_file
+import csv
 import requests
 from datetime import datetime, timedelta
+from sqlalchemy.exc import IntegrityError
 
 # Load environment variables from .env file
 # Get the directory where this script is located
@@ -18,6 +22,9 @@
 
 app = Flask(__name__)
 
+ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "")
+app.secret_key = ADMIN_PASSWORD
+
 app.config["TEMPLATES_AUTO_RELOAD"] = True
 app.config["MAIL_SERVER"] = "smtp.gmail.com"
 app.config["MAIL_PORT"] = 465
@@ -51,6 +58,49 @@
 # Cache for leaderboard data
 aoc_cache = {"data": None, "last_updated": None}
 
+# SU membership API configuration
+SU_KEY = os.environ.get("SU_KEY", "")
+SU_CACHE_DURATION = timedelta(hours=1)
+su_cache = {"members": None, "last_updated": None}
+
+
+def get_su_members():
+    """Fetch and cache SU membership list. Returns dict of {student_id: full_name}."""
+    now = datetime.now()
+    if (
+        su_cache["members"] is not None
+        and su_cache["last_updated"] is not None
+        and now - su_cache["last_updated"] < SU_CACHE_DURATION
+    ):
+        return su_cache["members"]
+
+    url = f"https://www.warwicksu.com/membershipapi/listMembers/{SU_KEY}/"
+    response = requests.get(url, timeout=15)
+    response.raise_for_status()
+
+    root = ET.fromstring(response.text)
+    members = {}
+    for member in root.findall("Member"):
+        uid_el = member.find("UniqueID")
+        first_el = member.find("FirstName")
+        last_el = member.find("LastName")
+        if uid_el is not None:
+            uid = int(uid_el.text)
+            first = (first_el.text or "").strip() if first_el is not None else ""
+            last = (last_el.text or "").strip() if last_el is not None else ""
+            members[uid] = f"{first} {last}".strip()
+
+    su_cache["members"] = members
+    su_cache["last_updated"] = now
+    return members
+
+
+def load_elections_json():
+    """Load elections.json and return the categories list."""
+    path = os.path.join(os.path.dirname(__file__), "data", "elections.json")
+    with open(path, "r", encoding="utf-8") as f:
+        return json.load(f)["categories"]
+
 
 @app.route("/")
 def home():
@@ -202,7 +252,283 @@ def aoc_leaderboard():
         return jsonify({"error": str(e)}), 500
 
 
+# ── Elections ─────────────────────────────────────────────────────────────────
+
+@app.context_processor
+def inject_elections_status():
+    try:
+        settings = ElectionSettings.query.get(1)
+        return {"elections_open": settings.isOpen if settings else False}
+    except Exception:
+        return {"elections_open": False}
+
+
+@app.route("/elections")
+def elections():
+    settings = ElectionSettings.query.get(1)
+    is_open = settings.isOpen if settings else False
+    categories = load_elections_json()
+    return render_template("elections.html", pageName="elections", is_open=is_open, categories=categories)
+
+
+@app.route("/elections/administrator/login", methods=["GET", "POST"])
+def elections_admin_login():
+    error = None
+    if request.method == "POST":
+        if request.form.get("password") == ADMIN_PASSWORD:
+            session["admin"] = True
+            return redirect(url_for("elections_admin"))
+        error = "Incorrect password."
+    return render_template("elections-admin-login.html", error=error)
+
+
+@app.route("/elections/administrator/logout")
+def elections_admin_logout():
+    session.pop("admin", None)
+    return redirect(url_for("elections_admin_login"))
+
+
+@app.route("/elections/administrator")
+def elections_admin():
+    if not session.get("admin"):
+        return redirect(url_for("elections_admin_login"))
+    settings = ElectionSettings.query.get(1)
+    if not settings:
+        settings = ElectionSettings()
+        settings.id = 1
+        db.session.add(settings)
+        db.session.commit()
+    return render_template("elections-admin.html", pageName="elections", is_open=settings.isOpen)
+
+
+@app.route("/elections/<slug>")
+def elections_category(slug):
+    settings = ElectionSettings.query.get(1)
+    if not settings or not settings.isOpen:
+        return redirect(url_for("elections"))
+
+    categories = load_elections_json()
+    category = next((c for c in categories if c["slug"] == slug), None)
+    if category is None:
+        abort(404)
+
+    position_order = [p["title"] for p in category["positions"]]
+    candidates = ElectionCandidate.query.filter_by(categorySlug=slug).all()
+
+    positions_dict = {title: [] for title in position_order}
+    for c in candidates:
+        if c.positionTitle in positions_dict:
+            positions_dict[c.positionTitle].append(c)
+
+    # Only include positions that have at least one candidate
+    positions = [(title, positions_dict[title]) for title in position_order if positions_dict[title]]
+
+    return render_template(
+        "elections-category.html",
+        pageName="elections",
+        category=category,
+        positions=positions,
+    )
+
+
+@app.route("/api/elections/settings", methods=["GET"])
+def elections_settings_get():
+    settings = ElectionSettings.query.get(1)
+    return jsonify({"isOpen": settings.isOpen if settings else False})
+
+
+@app.route("/api/elections/settings", methods=["POST"])
+def elections_settings_post():
+    data = request.get_json()
+    settings = ElectionSettings.query.get(1)
+    if not settings:
+        settings = ElectionSettings()
+        settings.id = 1
+        db.session.add(settings)
+    closing = settings.isOpen and not bool(data.get("isOpen", False))
+    settings.isOpen = bool(data.get("isOpen", False))
+    db.session.commit()
+
+    if closing:
+        _dump_votes_to_csv()
+
+    return jsonify({"isOpen": settings.isOpen})
+
+
+@app.route("/api/elections/verify", methods=["POST"])
+def elections_verify():
+    data = request.get_json()
+    category_slug = data.get("category_slug")
+
+    try:
+        raw_id = str(data.get("student_id", "")).strip().lstrip("Uu")
+        student_id = int(raw_id)
+    except (ValueError, TypeError):
+        return jsonify({"valid": False, "error": "Invalid student ID format"}), 400
+
+    try:
+        members = get_su_members()
+    except Exception:
+        return jsonify({"valid": False, "error": "Could not verify membership, please try again"}), 500
+
+    if student_id not in members:
+        return jsonify({"valid": False, "error": "Student ID not found in CodeSoc membership list"})
+
+    already_voted = False
+    if category_slug:
+        already_voted = ElectionVote.query.filter_by(
+            voterStudentID=student_id, categorySlug=category_slug
+        ).first() is not None
+
+    return jsonify({"valid": True, "name": members[student_id], "alreadyVoted": already_voted})
+
+
+@app.route("/api/elections/vote", methods=["POST"])
+def elections_vote():
+    settings = ElectionSettings.query.get(1)
+    if not settings or not settings.isOpen:
+        return jsonify({"error": "Elections are currently closed"}), 403
+
+    data = request.get_json()
+    category_slug = data.get("category_slug")
+    votes = data.get("votes", [])
+
+    try:
+        raw_id = str(data.get("student_id", "")).strip().lstrip("Uu")
+        student_id = int(raw_id)
+    except (ValueError, TypeError):
+        return jsonify({"error": "Invalid student ID"}), 400
+
+    if not category_slug:
+        return jsonify({"error": "Missing category"}), 400
+
+    # Re-verify membership server-side
+    try:
+        members = get_su_members()
+    except Exception:
+        return jsonify({"error": "Could not verify membership"}), 500
+
+    if student_id not in members:
+        return jsonify({"error": "Not a CodeSoc member"}), 403
+
+    # Check already voted
+    if ElectionVote.query.filter_by(voterStudentID=student_id, categorySlug=category_slug).first():
+        return jsonify({"error": "Already voted in this category"}), 409
+
+    POINTS = {1: 5, 2: 2, 3: 1}
+
+    try:
+        for v in votes:
+            rank = v.get("rank")
+            candidate_id = v.get("candidate_id")
+            if rank not in POINTS or not candidate_id:
+                continue
+            candidate = ElectionCandidate.query.get(candidate_id)
+            if not candidate or candidate.categorySlug != category_slug:
+                return jsonify({"error": "Invalid candidate"}), 400
+            db.session.add(ElectionVote(
+                voterStudentID=student_id,
+                candidateID=candidate_id,
+                positionTitle=candidate.positionTitle,
+                categorySlug=category_slug,
+                rank=rank,
+                points=POINTS[rank],
+            ))
+        db.session.commit()
+        return jsonify({"success": True})
+    except IntegrityError:
+        db.session.rollback()
+        return jsonify({"error": "Duplicate vote detected"}), 409
+    except Exception as e:
+        db.session.rollback()
+        return jsonify({"error": str(e)}), 500
+
+
+@app.route("/api/elections/results")
+def elections_results():
+    categories = load_elections_json()
+    results = []
+
+    for category in categories:
+        cat_data = {"name": category["name"], "slug": category["slug"], "positions": []}
+        for position in category["positions"]:
+            title = position["title"]
+            candidates = ElectionCandidate.query.filter_by(
+                categorySlug=category["slug"], positionTitle=title
+            ).all()
+
+            rows = []
+            for c in candidates:
+                r1 = ElectionVote.query.filter_by(candidateID=c.candidateID, rank=1).count()
+                r2 = ElectionVote.query.filter_by(candidateID=c.candidateID, rank=2).count()
+                r3 = ElectionVote.query.filter_by(candidateID=c.candidateID, rank=3).count()
+                rows.append({
+                    "name": c.name,
+                    "points": r1 * 5 + r2 * 2 + r3,
+                    "rank1": r1, "rank2": r2, "rank3": r3,
+                })
+            rows.sort(key=lambda x: x["points"], reverse=True)
+            if rows:
+                cat_data["positions"].append({"title": title, "candidates": rows})
+
+        if cat_data["positions"]:
+            results.append(cat_data)
+
+    return jsonify(results)
+
+
+def _dump_votes_to_csv():
+    """Write all current votes to a timestamped CSV file. Returns the filename."""
+    timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
+    filename = f"votes_{timestamp}.csv"
+
+    votes = ElectionVote.query.order_by(
+        ElectionVote.voterStudentID,
+        ElectionVote.categorySlug,
+        ElectionVote.positionTitle,
+        ElectionVote.rank,
+    ).all()
+
+    candidates = {c.candidateID: c.name for c in ElectionCandidate.query.all()}
+
+    with open(filename, "w", newline="", encoding="utf-8") as f:
+        writer = csv.writer(f)
+        writer.writerow([
+            "voter_student_id",
+            "category_slug",
+            "position_title",
+            "rank",
+            "points",
+            "candidate_id",
+            "candidate_name",
+            "submitted_at",
+        ])
+        for v in votes:
+            writer.writerow([
+                v.voterStudentID,
+                v.categorySlug,
+                v.positionTitle,
+                v.rank,
+                v.points,
+                v.candidateID,
+                candidates.get(v.candidateID, "Unknown"),
+                v.submittedAt.isoformat() if v.submittedAt else "",
+            ])
+
+    return filename, len(votes)
+
+
 # CLI Commands for database management
+@app.cli.command("db-dump-votes")
+def db_dump_votes():
+    """Dump all election votes to a timestamped CSV file."""
+    filename, count = _dump_votes_to_csv()
+    if count == 0:
+        click.echo("⚠ No votes to dump.")
+    else:
+        click.echo(f"✅ Dumped {count} votes to {filename}")
+
+
 @app.cli.command("db-reset")
 def db_reset():
     """Reset database with validated JSON data"""