"""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]