""" app/enrollment/routes.py ------------------------ The JQC Enrollment Form. GET /enrollment public form (NO login) POST /enrollment submit → thank-you page GET /enrollment/admin admin: all submissions GET /enrollment/admin/ admin: one submission POST /enrollment/admin/ admin: office-use fields + status GET /enrollment/admin/.json admin: raw JSON download GET /enrollment/admin/export.csv admin: all submissions as CSV Separation ---------- This module imports NOTHING from app.models and writes NOTHING to the database (see app/enrollment/__init__.py). Its only couplings to the rest of the app are the ones it cannot sensibly reinvent: the app factory, CSRF, the rate limiter, and @admin_required for the admin views. The public page is login-free, so it follows the same hardening as the `public` blueprint (rule 74): CSRF-protected form, rate limited, honeypot-guarded, and its own standalone template with no authenticated nav. """ import csv import io import logging import re from datetime import datetime from flask import (Blueprint, render_template, request, redirect, url_for, flash, abort, Response, current_app) from flask_login import login_required from app import limiter from app.utils.decorators import admin_required from . import mailer, schema, storage logger = logging.getLogger(__name__) bp = Blueprint( 'enrollment', __name__, url_prefix='/enrollment', # Own template folder — enrollment templates never mix into app/templates. template_folder='templates', ) #: Bots find public forms fast. A human filling in a 6-person enrollment form #: does not need more than a few attempts an hour from one address. _SUBMIT_RATE_LIMIT = '5 per hour' _MAX_TEXT = 200 # per free-text field; anything longer is truncated _MAX_NOTES = 2000 def _clean(value, limit=_MAX_TEXT): """Trim and length-cap one submitted text field.""" return (value or '').strip()[:limit] #: Person rows are named person__. The client controls (rows can #: be added and removed in any order), so the server discovers the indexes that #: were actually posted rather than trusting a count field. _PERSON_FIELD_RE = re.compile(r'^person_(\d+)_role$') def _parse_people(form): """Return the submitted people as an ordered list of dicts. Each entry gets a stable `key` (p1, p2, …) assigned by POSITION, not by the client's index — so the matrix keys in a stored submission are always dense and predictable no matter which rows the customer deleted before sending. """ indexes = sorted( int(m.group(1)) for m in (_PERSON_FIELD_RE.match(k) for k in form.keys()) if m ) people = [] for idx in indexes: role = form.get(f'person_{idx}_role', '') if role not in schema.ROLE_KEYS: role = schema.DEFAULT_FIRST_ROLE name = _clean(form.get(f'person_{idx}_name')) job_title = _clean(form.get(f'person_{idx}_job_title')) email = _clean(form.get(f'person_{idx}_email')) # Drop rows the customer added but left completely blank. if not (name or job_title or email): continue people.append({ 'key': f'p{len(people) + 1}', 'form_index': idx, # so the matrix cells can be read back 'role': role, 'name': name, 'job_title': job_title, 'email': email, }) if len(people) >= schema.MAX_PEOPLE: logger.warning('ENROLLMENT | people capped at %d', schema.MAX_PEOPLE) break return people def _seed_people(people, matrix, mobile_app): """Shape the submitted people for the page to re-render after an error. Folds each person's ticked tasks into their own row, so the browser can rebuild the table from scratch with fresh row indexes and still restore every answer. """ seed = [] for person in people: seed.append({ 'role': person['role'], 'name': person['name'], 'job_title': person['job_title'], 'email': person['email'], 'tasks': [ref for ref, _l, _s in schema.TASKS if matrix.get(str(ref), {}).get(person['key'])], 'mobile': bool(mobile_app.get(person['key'])), }) return seed # ── Public form ────────────────────────────────────────────────────────────── @bp.route('', methods=['GET']) @bp.route('/', methods=['GET']) def form(): """Render the blank enrollment form. No login — the link is emailed out.""" return render_template('enrollment/form.html', schema=schema, seed_people=[]) @bp.route('', methods=['POST']) @bp.route('/', methods=['POST']) @limiter.limit(_SUBMIT_RATE_LIMIT) def submit(): """Parse, validate and store one enrollment submission.""" # Honeypot: a field hidden from humans via CSS. Anything that fills it in # is a bot. Answer 200 as though accepted so it learns nothing. if (request.form.get('website') or '').strip(): logger.info('ENROLLMENT | honeypot tripped | ip=%s', request.remote_addr) return render_template('enrollment/submitted.html', reference=None) project_name = _clean(request.form.get('project_name')) request_by = _clean(request.form.get('request_by')) requester_email = _clean(request.form.get('requester_email')) people = _parse_people(request.form) # ── Step 2 matrix + Step 3 mobile, keyed by person ─────────────────── # Only cells the person's role actually offers are read, so a crafted POST # cannot record an admin-only task against an inspector. matrix = {} for ref, _label, scope in schema.TASKS: row = {} for person in people: if schema.task_applies(scope, person['role']): row[person['key']] = bool( request.form.get(f'task_{ref}_person_{person["form_index"]}')) matrix[str(ref)] = row mobile_app = { p['key']: bool(request.form.get(f'mobile_person_{p["form_index"]}')) for p in people } # ── Validation ─────────────────────────────────────────────────────── # A person counts only with BOTH a name and an email — a half-filled row # cannot be set up, so it must not pass as one. named = [p for p in people if p['name'] and p['email']] errors = [] if not project_name: errors.append('Project Name is required.') if not request_by: errors.append('Request by is required.') if not requester_email: errors.append('Requester email is required — we send your confirmation ' 'there.') elif '@' not in requester_email: errors.append('The requester email address does not look valid.') if not named: errors.append('Please add at least one person with both a name and an ' 'email address.') for p in people: if p['email'] and '@' not in p['email']: errors.append(f'"{p["name"] or p["key"]}" has an email address that ' f'does not look valid.') seen = set() for p in named: low = p['email'].lower() if low in seen: errors.append(f'{p["email"]} is listed more than once — each person ' f'needs their own email address.') seen.add(low) prior = { 'project_name': project_name, 'request_by': request_by, 'requester_email': requester_email, 'date_requested': _clean(request.form.get('date_requested')), 'notes': _clean(request.form.get('notes'), _MAX_NOTES), 'people': people, 'matrix': matrix, 'mobile_app': mobile_app, } if errors: for e in errors: flash(e, 'danger') # Re-render with what they typed so nothing is retyped. return render_template( 'enrollment/form.html', schema=schema, submitted=prior, seed_people=_seed_people(people, matrix, mobile_app)), 400 now = datetime.now() record = dict(prior) record.update({ 'id': storage.new_id(now), 'submitted_at': now.isoformat(timespec='seconds'), # Filled in later by staff on the admin page. 'office': {k: '' for k, _ in schema.OFFICE_FIELDS}, 'status': 'new', 'meta': { 'ip': request.remote_addr, 'user_agent': (request.headers.get('User-Agent') or '')[:300], }, }) try: storage.save(record) except Exception: logger.exception('ENROLLMENT | save failed | project=%r', project_name) flash('Sorry — we could not save your form. Please try again, or ' 'email us directly.', 'danger') return render_template( 'enrollment/form.html', schema=schema, submitted=prior, seed_people=_seed_people(people, matrix, mobile_app)), 500 logger.info('ENROLLMENT | submitted | id=%s project=%r people=%d ip=%s', record['id'], project_name, len(named), request.remote_addr) # Both emails fire AFTER the save and are fully guarded — a mail problem # must never cost the customer their submission. mailer.send_confirmation(record, base_url=request.host_url) # requester mailer.send_admin_notification(record, base_url=request.host_url) # JQC admins return render_template('enrollment/submitted.html', reference=record['id'], email=requester_email) # ── Admin ──────────────────────────────────────────────────────────────────── @bp.route('/admin') @login_required @admin_required def admin_list(): records = storage.load_all() logger.info('ENROLLMENT | admin_list | count=%d', len(records)) return render_template('enrollment/admin_list.html', records=records, schema=schema) @bp.route('/admin/', methods=['GET', 'POST']) @login_required @admin_required def admin_detail(submission_id): record = storage.load(submission_id) if record is None: abort(404) if request.method == 'POST': office = {k: _clean(request.form.get(k)) for k, _ in schema.OFFICE_FIELDS} status = request.form.get('status', 'new') if status not in schema.STATUSES: status = record.get('status', 'new') record = storage.update_office(submission_id, office, status) if record is None: abort(404) flash('Enrollment record updated.', 'success') return redirect(url_for('enrollment.admin_detail', submission_id=submission_id)) return render_template('enrollment/admin_detail.html', record=record, schema=schema) @bp.route('/admin/.json') @login_required @admin_required def admin_download(submission_id): import json record = storage.load(submission_id) if record is None: abort(404) return Response( json.dumps(record, indent=2, ensure_ascii=False), mimetype='application/json', headers={'Content-Disposition': f'attachment; filename=enrollment-{submission_id}.json'}, ) @bp.route('/admin/export.csv') @login_required @admin_required def admin_export_csv(): """One row per PERSON (not per submission) — that is the unit of work when actually setting the accounts up. Reads through schema.people_of(), so submissions stored in the older fixed-seat format export identically.""" records = storage.load_all() buf = io.StringIO() w = csv.writer(buf) task_headers = [f'{ref}. {label}' for ref, label, _ in schema.TASKS] w.writerow(['Submission ID', 'Submitted At', 'Status', 'Project Name', 'Requested By', 'Date Requested', 'Role', 'Name', 'Job Title', 'Email', 'Mobile App'] + task_headers) for rec in records: for person in schema.people_of(rec): row = [ rec.get('id', ''), rec.get('submitted_at', ''), schema.STATUS_LABELS.get(rec.get('status'), rec.get('status', '')), rec.get('project_name', ''), rec.get('request_by', ''), rec.get('date_requested', ''), person['role_label'], person['name'], person['job_title'], person['email'], 'Yes' if schema.wants_mobile(rec, person['key']) else '', ] for ref, _label, scope in schema.TASKS: if not schema.task_applies(scope, person['role']): row.append('n/a') else: row.append('Yes' if schema.cell(rec, ref, person['key']) else '') w.writerow(row) logger.info('ENROLLMENT | csv export | submissions=%d', len(records)) stamp = datetime.now().strftime('%Y%m%d') return Response( buf.getvalue(), mimetype='text/csv', headers={'Content-Disposition': f'attachment; filename=jqc-enrollments-{stamp}.csv'}, )