Spaces:
Running
Running
Multi-select areas, colour-coded area tags, Creighton students only, titles removed, Creighton blue
4f5fdec verified | """Application logic that does not depend on HTTP: validation, filtering, | |
| sorting, approval and comparison of submissions against listings.""" | |
| from __future__ import annotations | |
| import datetime as dt | |
| import logging | |
| import os | |
| import re | |
| from typing import Any, Mapping, Optional | |
| from urllib.parse import urlsplit | |
| from zoneinfo import ZoneInfo, ZoneInfoNotFoundError | |
| from .db import Db, DbError, check_uuid | |
| from .fields import (CONTENT_FIELDS, FIELD_KEYS, SUBMITTER_FIELDS, Field, split_list) | |
| from .normalize import clean_text, join_choices | |
| NETID_RE = re.compile(r"^[A-Za-z0-9]{2,64}$") | |
| EMAIL_RE = re.compile(r"^[^\s@]+@[^\s@]+\.[^\s@]+$") | |
| URL_SCHEME_RE = re.compile(r"^(https?://|mailto:)", re.IGNORECASE) | |
| BARE_DOMAIN_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9.-]*\.[A-Za-z]{2,}(?:[:/?#].*)?$") | |
| log = logging.getLogger("curas.services") | |
| _zone_cache: dict[str, dt.tzinfo] = {} | |
| def tz() -> dt.tzinfo: | |
| """Time zone from APP_TIMEZONE (default America/Chicago), read when first | |
| needed so a value from the .env file is honoured.""" | |
| name = os.environ.get("APP_TIMEZONE", "America/Chicago").strip() or "America/Chicago" | |
| if name not in _zone_cache: | |
| try: | |
| _zone_cache[name] = ZoneInfo(name) | |
| except (ZoneInfoNotFoundError, ValueError): | |
| log.warning("APP_TIMEZONE %r is not a known zone; using UTC", name) | |
| _zone_cache[name] = dt.timezone.utc | |
| return _zone_cache[name] | |
| def today() -> dt.date: | |
| """Calendar date where CURAS is (APP_TIMEZONE, default America/Chicago).""" | |
| return dt.datetime.now(tz()).date() | |
| # -------------------------------------------------------------------------- | |
| # Validation | |
| # -------------------------------------------------------------------------- | |
| def normalize_url(raw: Optional[str]) -> Optional[str]: | |
| """Return a safe http(s)/mailto URL, None if blank, or raise ValueError.""" | |
| s = (raw or "").strip() | |
| if not s: | |
| return None | |
| if any(ch.isspace() for ch in s): | |
| raise ValueError("must not contain spaces") | |
| if not URL_SCHEME_RE.match(s): | |
| if BARE_DOMAIN_RE.match(s): | |
| s = "https://" + s | |
| else: | |
| raise ValueError("must start with https://, http:// or mailto:") | |
| if s.lower().startswith("mailto:"): | |
| if not EMAIL_RE.match(s[7:]): | |
| raise ValueError("mailto: must be followed by an email address") | |
| return s | |
| host = urlsplit(s).hostname or "" | |
| if "." not in host or host.startswith(".") or host.endswith("."): | |
| raise ValueError("must include a web address such as https://www.creighton.edu/...") | |
| return s | |
| def safe_href(url: Optional[str]) -> Optional[str]: | |
| """Only ever render links with a safe scheme; otherwise show as text.""" | |
| if not url: | |
| return None | |
| return url if URL_SCHEME_RE.match(url.strip()) else None | |
| class FormLike: | |
| """Adapter so plain dicts (tests) and Starlette FormData share one API.""" | |
| def __init__(self, data: Any): | |
| self._d = data | |
| def get(self, key: str) -> Optional[str]: | |
| v = self._d.get(key) | |
| if isinstance(v, (list, tuple)): | |
| return v[0] if v else None | |
| return v | |
| def getlist(self, key: str) -> list[str]: | |
| if hasattr(self._d, "getlist"): | |
| return [str(x) for x in self._d.getlist(key)] | |
| v = self._d.get(key) | |
| if v is None: | |
| return [] | |
| return [str(x) for x in v] if isinstance(v, (list, tuple)) else [str(v)] | |
| def _field_value(form: FormLike, f: Field, required: bool) -> tuple[Optional[str], Optional[str]]: | |
| """Return (value, error) for one content field. On an error the value is | |
| the text the person typed, so the form can show it again.""" | |
| if f.kind == "choices": | |
| selected = [s for s in form.getlist(f.key) if s in f.choices] | |
| other = form.get(f.key + "_other") or "" | |
| if len(other) > f.maxlen: | |
| return join_choices(selected, other[:f.maxlen], f.choices), f"Other text must be at most {f.maxlen} characters." | |
| value = join_choices(selected, other, f.choices) | |
| if required and not value: | |
| return None, "Choose at least one option or describe it in the text box." | |
| return value, None | |
| raw = form.get(f.key) | |
| raw = "" if raw is None else str(raw) | |
| typed = raw.strip() | |
| if f.kind == "url": | |
| try: | |
| value = normalize_url(raw) | |
| except ValueError as exc: | |
| return typed, f"Link {exc}." | |
| else: | |
| value = clean_text(raw, keep_newlines=(f.kind == "textarea")) | |
| if value and len(value) > f.maxlen: | |
| return typed, f"Must be at most {f.maxlen} characters." | |
| if required and not value: | |
| return None, "This field is required." | |
| return value, None | |
| def validate_content(form_data: Any, existing: Optional[Mapping] = None) -> tuple[dict, dict]: | |
| """Validate listing content fields. Returns (values, errors). | |
| When `existing` (the stored listing being edited) is given, a required | |
| field that is empty on the stored row is not demanded, so listings | |
| imported without, say, a standing can still be corrected without inventing | |
| content. New listings must satisfy every required field. | |
| """ | |
| form = FormLike(form_data) | |
| values: dict[str, Optional[str]] = {} | |
| errors: dict[str, str] = {} | |
| for f in CONTENT_FIELDS: | |
| required = f.required and (existing is None or bool(existing.get(f.key))) | |
| value, err = _field_value(form, f, required) | |
| if err: | |
| errors[f.key] = err | |
| values[f.key] = value | |
| return values, errors | |
| def validate_submitter(form_data: Any) -> tuple[dict, dict]: | |
| form = FormLike(form_data) | |
| values: dict[str, Optional[str]] = {} | |
| errors: dict[str, str] = {} | |
| for f in SUBMITTER_FIELDS: | |
| raw = form.get(f.key) | |
| value = clean_text(raw, keep_newlines=False) | |
| if value and len(value) > f.maxlen: | |
| errors[f.key] = f"Must be at most {f.maxlen} characters." | |
| elif f.required and not value: | |
| errors[f.key] = "This field is required." | |
| elif f.key == "submitter_netid" and value and not NETID_RE.match(value): | |
| errors[f.key] = "NetID may contain letters and digits only." | |
| elif f.key == "submitter_email" and value and not EMAIL_RE.match(value): | |
| errors[f.key] = "Enter a valid email address or leave this blank." | |
| values[f.key] = value | |
| if values.get("submitter_netid"): | |
| values["submitter_netid"] = values["submitter_netid"].lower() | |
| return values, errors | |
| # -------------------------------------------------------------------------- | |
| # Public browsing | |
| # -------------------------------------------------------------------------- | |
| # Only fields that are still part of a listing are searched, so values left in | |
| # retired database columns (such as the old project titles) never match. | |
| SEARCH_KEYS = [k for k in ["faculty_name", "keywords", "description", "methods", "areas_of_study", | |
| "prerequisites", "mentorship", "previous_student_work", "faculty_note", | |
| "commitment"] if k in FIELD_KEYS] | |
| def _items_lower(value: Optional[str]) -> list[str]: | |
| return [p.lower() for p in split_list(value)] | |
| def _matches_item(value: Optional[str], wanted: str, vocab: dict) -> bool: | |
| """Exact item match; free-text items (not in the vocabulary) match on a | |
| whole-word basis so entries such as "Second Semester Freshman year" still | |
| appear under Freshman.""" | |
| wanted_l = wanted.lower() | |
| canonical = {k.lower() for k in vocab} | |
| for item in _items_lower(value): | |
| if item == wanted_l: | |
| return True | |
| if item not in canonical and re.search(r"\b" + re.escape(wanted_l) + r"\b", item): | |
| return True | |
| return False | |
| def _matches_area(row: dict, area: str) -> bool: | |
| """A listing matches an area if it lists it, or if it is open to all majors.""" | |
| from .fields import AREAS_OF_STUDY | |
| areas = row.get("areas_of_study") | |
| if _matches_item(areas, area, AREAS_OF_STUDY): | |
| return True | |
| return area.lower() != "all majors" and _matches_item(areas, "All majors", AREAS_OF_STUDY) | |
| def filter_listings(rows: list[dict], *, q: str = "", areas: Optional[list[str]] = None, type_: str = "", | |
| term: str = "", standing: str = "") -> list[dict]: | |
| """Filter listings. `areas` may hold several areas: a listing is kept when it | |
| matches any of them.""" | |
| from .fields import OPPORTUNITY_TYPES, STANDINGS, TERMS | |
| q_l = (q or "").strip().lower() | |
| wanted_areas = [a for a in (areas or []) if a] | |
| out = [] | |
| for r in rows: | |
| if q_l: | |
| hay = " \n ".join(str(r.get(k) or "") for k in SEARCH_KEYS).lower() | |
| if q_l not in hay: | |
| continue | |
| if wanted_areas and not any(_matches_area(r, a) for a in wanted_areas): | |
| continue | |
| if type_ and not _matches_item(r.get("opportunity_type"), type_, OPPORTUNITY_TYPES): | |
| continue | |
| if term and not _matches_item(r.get("term_availability"), term, TERMS): | |
| continue | |
| if standing and not _matches_item(r.get("standing"), standing, STANDINGS): | |
| continue | |
| out.append(r) | |
| return out | |
| def surname_key(name: Optional[str]) -> str: | |
| s = (name or "").strip() | |
| s = re.sub(r",.*$", "", s) # drop ", S.J." style suffixes | |
| s = re.sub(r"^(dr|prof|professor|mr|ms|mrs)\.?\s+", "", s, flags=re.I) | |
| parts = s.split() | |
| return (parts[-1] + " " + " ".join(parts[:-1])).lower() if parts else "" | |
| def sort_listings(rows: list[dict], sort: str) -> list[dict]: | |
| if sort == "faculty": | |
| return sorted(rows, key=lambda r: surname_key(r.get("faculty_name"))) | |
| # default: most recently updated first, then faculty name for stability | |
| return sorted(rows, key=lambda r: (-_date_ord(r.get("updated_on")), surname_key(r.get("faculty_name")))) | |
| def _date_ord(value: Optional[str]) -> int: | |
| try: | |
| return dt.date.fromisoformat(str(value)[:10]).toordinal() | |
| except (TypeError, ValueError): | |
| return 0 | |
| def filter_options(rows: list[dict]) -> dict[str, list[str]]: | |
| """Vocabulary items that occur in at least one listing, in vocabulary order | |
| (areas alphabetical with 'All majors' first).""" | |
| from .fields import AREAS_OF_STUDY, OPPORTUNITY_TYPES, STANDINGS, TERMS | |
| def present(key: str, vocab: dict) -> list[str]: | |
| seen = set() | |
| for r in rows: | |
| for item in split_list(r.get(key)): | |
| seen.add(item.lower()) | |
| return [v for v in vocab if v.lower() in seen] | |
| areas = present("areas_of_study", AREAS_OF_STUDY) | |
| areas = sorted(areas, key=lambda a: (a != "All majors", a.lower())) | |
| return { | |
| "areas": areas, | |
| "types": present("opportunity_type", OPPORTUNITY_TYPES), | |
| "terms": present("term_availability", TERMS), | |
| "standings": present("standing", STANDINGS), | |
| } | |
| # -------------------------------------------------------------------------- | |
| # Submissions and approval | |
| # -------------------------------------------------------------------------- | |
| def content_of(row: Mapping) -> dict: | |
| return {k: row.get(k) for k in FIELD_KEYS} | |
| def diff_fields(current: Optional[Mapping], proposed: Mapping) -> list[dict]: | |
| """Field-by-field comparison for the admin review page.""" | |
| out = [] | |
| for f in CONTENT_FIELDS: | |
| cur = (current or {}).get(f.key) if current else None | |
| new = proposed.get(f.key) | |
| out.append({"field": f, "current": cur, "proposed": new, | |
| "changed": (cur or "") != (new or "")}) | |
| return out | |
| class ApprovalError(Exception): | |
| pass | |
| def _get_reviewable(db: Db, submission_id: str) -> tuple[dict, bool]: | |
| """Return (submission, needs_claim). A submission can be acted on while it | |
| is pending, or while it is approved but its listing was never written | |
| (published_at is null), which happens only if the process died mid-way.""" | |
| sid = check_uuid(submission_id) | |
| sub = db.get_one("submissions", sid) | |
| if sub is None: | |
| raise LookupError("submission not found") | |
| if sub.get("status") == "pending": | |
| return sub, True | |
| if sub.get("status") == "approved" and not sub.get("published_at"): | |
| return sub, False | |
| raise ApprovalError("This submission is no longer pending.") | |
| def is_unfinished(sub: Mapping) -> bool: | |
| return sub.get("status") == "approved" and not sub.get("published_at") | |
| def approve_submission(db: Db, submission_id: str) -> dict: | |
| """Publish a pending submission. Returns the resulting listing row. | |
| The submission is first marked approved with a conditional update (status | |
| must still be pending) so two admins cannot both approve it. The listing | |
| created for a new submission records the submission id, so if a later step | |
| fails the submission goes back to the queue and approving it again reuses | |
| that listing instead of creating a duplicate. published_at is set only | |
| when everything has been written; an approved submission without it is | |
| shown to admins as unfinished and can be approved again. | |
| An update to a listing that has since been retired changes the listing's | |
| content but leaves it retired; the admin can restore it explicitly. | |
| """ | |
| sub, needs_claim = _get_reviewable(db, submission_id) | |
| sid = sub["id"] | |
| now = dt.datetime.now(dt.timezone.utc).isoformat() | |
| if needs_claim: | |
| claimed = db.update("submissions", {"id": f"eq.{sid}", "status": "eq.pending"}, | |
| {"status": "approved", "reviewed_at": now}) | |
| if not claimed: | |
| raise ApprovalError("This submission is no longer pending.") | |
| content = content_of(sub) | |
| patch = dict(content) | |
| patch["updated_on"] = today().isoformat() | |
| listing = None | |
| try: | |
| if sub.get("kind") == "update" and sub.get("listing_id"): | |
| rows = db.update("listings", {"id": f"eq.{check_uuid(sub['listing_id'])}"}, patch) | |
| listing = rows[0] if rows else None | |
| if listing is None: | |
| existing = db.select("listings", {"submission_id": f"eq.{sid}"}, limit=1) | |
| if existing: | |
| listing = db.update("listings", {"id": f"eq.{existing[0]['id']}"}, patch)[0] | |
| else: | |
| row = dict(patch) | |
| row.update({"status": "published", "retired_at": None, "source": "form", "submission_id": sid}) | |
| listing = db.insert("listings", row) | |
| db.update("submissions", {"id": f"eq.{sid}"}, | |
| {"listing_id": listing["id"], "published_at": dt.datetime.now(dt.timezone.utc).isoformat()}) | |
| return listing | |
| except (DbError, ValueError) as exc: | |
| log.error("approval of submission %s failed after claiming it: %s", sid, exc) | |
| db.update("submissions", {"id": f"eq.{sid}"}, {"status": "pending", "reviewed_at": None}) | |
| raise ApprovalError("The listing could not be published because the database did not respond. " | |
| "The submission is back in the queue; approving it again is safe.") from exc | |
| def reject_submission(db: Db, submission_id: str, note: Optional[str]) -> dict: | |
| """Reject a pending (or unfinished) submission. If an earlier approval | |
| attempt already created a listing for it, that listing is retired so | |
| rejected content never stays public.""" | |
| sub, _ = _get_reviewable(db, submission_id) | |
| sid = sub["id"] | |
| now = dt.datetime.now(dt.timezone.utc).isoformat() | |
| rows = db.update("submissions", {"id": f"eq.{sid}", "status": "in.(pending,approved)", "published_at": "is.null"}, | |
| {"status": "rejected", "reviewed_at": now, | |
| "admin_note": clean_text(note, keep_newlines=True)}) | |
| if not rows: | |
| raise ApprovalError("This submission is no longer pending.") | |
| orphan = db.select("listings", {"submission_id": f"eq.{sid}", "status": "eq.published"}, limit=1) | |
| if orphan: | |
| db.update("listings", {"id": f"eq.{orphan[0]['id']}"}, {"status": "retired", "retired_at": now}) | |
| rows = db.update("submissions", {"id": f"eq.{sid}"}, {"listing_id": orphan[0]["id"]}) or rows | |
| log.warning("rejected submission %s had a listing from an earlier approval attempt; retired %s", sid, orphan[0]["id"]) | |
| return rows[0] | |