spencermandClaude Opus 5.5
a82e056b6a
Admin: Departments (name and stations; add shows the three groups IT must create), Profiles (welder stamp, station limits, active) with a CSV import that previews and applies only when clean, Checklists (weld and plain), Security (session, lockout and rate-limit settings with bounds, alerts for the last 24 hours, the integrity result), Access log with filters and a guarded CSV, Who is in, Stamp check, Notifications, and the employeeID flags. Every Admin change is audited. Worker: ad-reconcile (nightly and on "refresh from AD now"), session-refresh (15 minutes), integrity (seals, asset hashes, missing and orphan files, the audit chain), backup-check, notify-send and notify-digest. Notifications route per event type to AD groups, users, mailboxes or the department Supervisor group, immediately or in a daily digest, over STARTTLS verified against the PESCO CA, with dry run on the test stack. Backups: scripts/backup/ps-backup.sh and ps-restore.sh with a systemd timer at 02:00 America/Denver. Restore rehearsed on the test stack 2026-10-21. CI: pip-audit and npm audit. pypdf 6.21.1 and certifi 2026.10.19. Migration 0010. Closes the app part of I6, M3, O1, O2 and S3. Co-Authored-By: Claude Opus 5.5 <[email protected]>
371 lines
16 KiB
Python
371 lines
16 KiB
Python
"""People's Packet Station profiles: welder stamp, profile active, station limits. Edited one at a time or by a CSV
|
||
import keyed on the employee ID.
|
||
|
||
Roles never come from here; AD groups decide them. An inactive profile drops out of the welder picker and the
|
||
worker filter. Employee IDs stay text (leading zeros kept).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import io
|
||
import re
|
||
from collections import Counter
|
||
from collections.abc import Mapping
|
||
from typing import Any
|
||
|
||
from .. import audit
|
||
from ..api.errors import ApiError
|
||
|
||
RX_STAMP = re.compile(r"^[A-Za-z0-9-]{1,12}$")
|
||
CSV_MAX_BYTES = 1_000_000
|
||
CSV_MAX_ROWS = 2000
|
||
COLUMNS = ("employee_id", "welder_stamp", "active", "stations")
|
||
CLEAR = "-"
|
||
TRUE_WORDS = ("yes", "true", "1")
|
||
FALSE_WORDS = ("no", "false", "0")
|
||
|
||
|
||
# ---------------------------------------------------------------- reading
|
||
|
||
|
||
def station_limits(conn: Any, user_ids: list[Any]) -> dict[Any, dict[str, list[int]]]:
|
||
"""user id -> {dept_code: [station ids]}."""
|
||
out: dict[Any, dict[str, list[int]]] = {}
|
||
if not user_ids:
|
||
return out
|
||
rows = conn.execute(
|
||
"SELECT l.user_id, s.dept_code, s.id FROM user_station_limits l JOIN stations s ON s.id = l.station_id"
|
||
" WHERE l.user_id = ANY(%s) ORDER BY s.dept_code, s.sort, s.id",
|
||
(user_ids,),
|
||
).fetchall()
|
||
for uid, dept, sid in rows:
|
||
out.setdefault(uid, {}).setdefault(dept, []).append(int(sid))
|
||
return out
|
||
|
||
|
||
def profile_of(conn: Any, user_id: Any) -> dict[str, Any] | None:
|
||
row = conn.execute("SELECT welder_stamp, profile_active FROM users WHERE id = %s", (user_id,)).fetchone()
|
||
if row is None:
|
||
return None
|
||
return {
|
||
"welder_stamp": row[0],
|
||
"profile_active": bool(row[1]),
|
||
"station_limits": station_limits(conn, [user_id]).get(user_id, {}),
|
||
}
|
||
|
||
|
||
def _stamp_owner(conn: Any, stamp: str, user_id: Any) -> str | None:
|
||
row = conn.execute(
|
||
"SELECT display_name FROM users WHERE lower(welder_stamp) = lower(%s) AND id <> %s LIMIT 1", (stamp, user_id)
|
||
).fetchone()
|
||
return str(row[0]) if row else None
|
||
|
||
|
||
# ---------------------------------------------------------------- one person
|
||
|
||
|
||
def _audit(conn: Any, who: Any, user_id: Any, before: dict[str, Any], after: dict[str, Any], source: str) -> None:
|
||
audit.append(
|
||
conn,
|
||
action="admin.profile_updated",
|
||
user_id=who.user_id,
|
||
session_id=who.session_id,
|
||
role_snapshot=who.role_snapshot,
|
||
subject_type="user",
|
||
subject_id=user_id,
|
||
detail={"before": before, "after": after, "source": source},
|
||
ip=who.ip,
|
||
)
|
||
|
||
|
||
def _write(conn: Any, user_id: Any, stamp: Any, active: Any, limits: Mapping[str, list[int]] | None) -> None:
|
||
"""Write the fields given; `stamp` and `active` use ... (Ellipsis) for unchanged. `limits` replaces the named
|
||
departments' limits; None leaves them."""
|
||
if stamp is not ...:
|
||
conn.execute("UPDATE users SET welder_stamp = %s WHERE id = %s", (stamp, user_id))
|
||
if active is not ...:
|
||
conn.execute("UPDATE users SET profile_active = %s WHERE id = %s", (bool(active), user_id))
|
||
for dept, ids in (limits or {}).items():
|
||
conn.execute(
|
||
"DELETE FROM user_station_limits l USING stations s WHERE s.id = l.station_id AND l.user_id = %s"
|
||
" AND s.dept_code = %s",
|
||
(user_id, dept),
|
||
)
|
||
for sid in ids:
|
||
conn.execute("INSERT INTO user_station_limits (user_id, station_id) VALUES (%s, %s)", (user_id, sid))
|
||
|
||
|
||
def update(conn: Any, who: Any, user_id: Any, fields: Mapping[str, Any]) -> dict[str, Any]:
|
||
"""PATCH a profile: welder_stamp (None clears), profile_active, station_limits {dept: [station id]} (a dept key
|
||
replaces that department's limits; an empty list clears them)."""
|
||
stamp: Any = ...
|
||
if "welder_stamp" in fields:
|
||
raw = fields["welder_stamp"]
|
||
stamp = None if raw is None or not str(raw).strip() else str(raw).strip()
|
||
if stamp is not None and not RX_STAMP.match(stamp):
|
||
raise ApiError(422, "bad_stamp", "A stamp is 1 to 12 letters, digits or dashes.")
|
||
active: Any = fields["profile_active"] if fields.get("profile_active") is not None else ...
|
||
limits: dict[str, list[int]] | None = None
|
||
if fields.get("station_limits") is not None:
|
||
limits = {}
|
||
for dept, ids in dict(fields["station_limits"]).items():
|
||
ok = {int(r[0]) for r in conn.execute("SELECT id FROM stations WHERE dept_code = %s", (dept,)).fetchall()}
|
||
if not conn.execute("SELECT 1 FROM departments WHERE code = %s", (dept,)).fetchone():
|
||
raise ApiError(422, "unknown_department", f"No department {dept}.")
|
||
want = sorted({int(x) for x in ids})
|
||
bad = [x for x in want if x not in ok]
|
||
if bad:
|
||
raise ApiError(422, "unknown_station", f"Station {bad[0]} is not in department {dept}.")
|
||
limits[str(dept)] = want
|
||
with conn.transaction():
|
||
if conn.execute("SELECT 1 FROM users WHERE id = %s FOR UPDATE", (user_id,)).fetchone() is None:
|
||
raise ApiError(404, "not_found", "No such user.")
|
||
if stamp is not ... and stamp is not None:
|
||
owner = _stamp_owner(conn, stamp, user_id)
|
||
if owner is not None:
|
||
raise ApiError(409, "stamp_taken", f"Stamp {stamp} belongs to {owner}.")
|
||
before = profile_of(conn, user_id)
|
||
assert before is not None
|
||
_write(conn, user_id, stamp, active, limits)
|
||
after = profile_of(conn, user_id)
|
||
assert after is not None
|
||
if before != after:
|
||
_audit(conn, who, user_id, before, after, "edit")
|
||
return after
|
||
|
||
|
||
# ---------------------------------------------------------------- CSV import
|
||
|
||
|
||
def _station_index(conn: Any) -> tuple[set[str], dict[tuple[str, str], int], dict[int, str]]:
|
||
depts = {r[0] for r in conn.execute("SELECT code FROM departments").fetchall()}
|
||
by_name: dict[tuple[str, str], int] = {}
|
||
label: dict[int, str] = {}
|
||
for sid, dept, name in conn.execute("SELECT id, dept_code, name FROM stations").fetchall():
|
||
by_name[(dept.lower(), " ".join(name.split()).lower())] = int(sid)
|
||
label[int(sid)] = f"{dept}:{name}"
|
||
return depts, by_name, label
|
||
|
||
|
||
def _bool(text: str) -> bool | None:
|
||
t = text.strip().lower()
|
||
if t in TRUE_WORDS:
|
||
return True
|
||
if t in FALSE_WORDS:
|
||
return False
|
||
return None
|
||
|
||
|
||
def _labels(ids: list[int], label: dict[int, str]) -> list[str]:
|
||
return sorted(label.get(i, str(i)) for i in ids)
|
||
|
||
|
||
def parse_csv(text: str) -> tuple[list[str], list[tuple[int, dict[str, str]]]]:
|
||
"""(header columns, [(line, {column: cell})]). Raises ApiError for a file that cannot be read at all."""
|
||
if len(text.encode("utf-8")) > CSV_MAX_BYTES:
|
||
raise ApiError(413, "too_large", "The file is larger than 1 MB.")
|
||
reader = csv.reader(io.StringIO(text.lstrip("")))
|
||
try:
|
||
header = next(reader, None)
|
||
if header is None:
|
||
raise ApiError(422, "bad_csv", "The file is empty.")
|
||
cols = [h.strip().lower() for h in header]
|
||
unknown = [c for c in cols if c and c not in COLUMNS]
|
||
if "employee_id" not in cols:
|
||
raise ApiError(422, "bad_csv", "The header row has no employee_id column.")
|
||
if unknown:
|
||
raise ApiError(422, "bad_csv", f"Unknown column: {', '.join(unknown)}.")
|
||
if len(set(c for c in cols if c)) != len([c for c in cols if c]):
|
||
raise ApiError(422, "bad_csv", "A column is named twice.")
|
||
rows: list[tuple[int, dict[str, str]]] = []
|
||
for cells in reader:
|
||
if not any(c.strip() for c in cells):
|
||
continue
|
||
if len(rows) >= CSV_MAX_ROWS:
|
||
raise ApiError(422, "bad_csv", f"The file has more than {CSV_MAX_ROWS} rows.")
|
||
row = {c: (cells[i].strip() if i < len(cells) else "") for i, c in enumerate(cols) if c}
|
||
rows.append((reader.line_num, row))
|
||
except csv.Error as e:
|
||
raise ApiError(422, "bad_csv", f"The file is not valid CSV: {e}.") from None
|
||
return cols, rows
|
||
|
||
|
||
def import_profiles(conn: Any, who: Any, text: str, *, apply: bool) -> dict[str, Any]:
|
||
"""Preview (apply false) or apply a profile CSV. With any error, apply is refused: 422 with the same body."""
|
||
_, parsed = parse_csv(text)
|
||
depts, by_name, label = _station_index(conn)
|
||
emps = [r.get("employee_id", "") for _, r in parsed]
|
||
users: dict[str, list[dict[str, Any]]] = {}
|
||
want = sorted({e for e in emps if e})
|
||
if want:
|
||
for uid, emp, name, stamp, active in conn.execute(
|
||
"SELECT id, emp_no, display_name, welder_stamp, profile_active FROM users"
|
||
" WHERE active AND emp_no = ANY(%s)",
|
||
(want,),
|
||
).fetchall():
|
||
users.setdefault(emp, []).append(
|
||
{"id": uid, "display_name": name, "welder_stamp": stamp, "profile_active": bool(active)}
|
||
)
|
||
limits = station_limits(conn, [u["id"] for us in users.values() for u in us])
|
||
seen: Counter[str] = Counter()
|
||
first_line: dict[str, int] = {}
|
||
out: list[dict[str, Any]] = []
|
||
plans: list[tuple[dict[str, Any], dict[str, Any]]] = [] # (row out, plan)
|
||
for line, r in parsed:
|
||
emp = r.get("employee_id", "")
|
||
errors: list[str] = []
|
||
row_out: dict[str, Any] = {
|
||
"line": line,
|
||
"employee_id": emp,
|
||
"user_id": None,
|
||
"display_name": None,
|
||
"changes": {},
|
||
"errors": errors,
|
||
}
|
||
out.append(row_out)
|
||
if not emp:
|
||
errors.append("Employee ID is missing.")
|
||
continue
|
||
seen[emp] += 1
|
||
if seen[emp] > 1:
|
||
errors.append(f"Employee ID {emp} is repeated in the file (first on line {first_line[emp]}).")
|
||
continue
|
||
first_line[emp] = line
|
||
found = users.get(emp, [])
|
||
if not found:
|
||
errors.append(f"Unknown employee ID {emp}.")
|
||
continue
|
||
if len(found) > 1:
|
||
errors.append(f"Employee ID {emp} is on more than one user.")
|
||
continue
|
||
u = found[0]
|
||
row_out["user_id"] = str(u["id"])
|
||
row_out["display_name"] = u["display_name"]
|
||
plan: dict[str, Any] = {"user": u}
|
||
stamp_cell = r.get("welder_stamp", "")
|
||
if stamp_cell:
|
||
new_stamp = None if stamp_cell == CLEAR else stamp_cell
|
||
if new_stamp is not None and not RX_STAMP.match(new_stamp):
|
||
errors.append(f"Bad stamp {stamp_cell}: 1 to 12 letters, digits or dashes.")
|
||
elif new_stamp != u["welder_stamp"]:
|
||
plan["welder_stamp"] = new_stamp
|
||
active_cell = r.get("active", "")
|
||
if active_cell:
|
||
val = _bool(active_cell)
|
||
if val is None:
|
||
errors.append(f"Bad active value {active_cell}: use yes or no.")
|
||
elif val != u["profile_active"]:
|
||
plan["profile_active"] = val
|
||
st_cell = r.get("stations", "")
|
||
if st_cell:
|
||
current = sorted(i for ids in limits.get(u["id"], {}).values() for i in ids)
|
||
ids: list[int] = []
|
||
if st_cell != CLEAR:
|
||
for part in (p.strip() for p in st_cell.split(";")):
|
||
if not part:
|
||
continue
|
||
dept, sep, name = part.partition(":")
|
||
dept, name = dept.strip(), " ".join(name.split())
|
||
if not sep or not dept or not name:
|
||
errors.append(f"Bad station {part}: write <department code>:<station name>.")
|
||
elif dept not in depts:
|
||
errors.append(f"Unknown department {dept}.")
|
||
elif (dept.lower(), name.lower()) not in by_name:
|
||
errors.append(f"Unknown station {name} in department {dept}.")
|
||
else:
|
||
sid = by_name[(dept.lower(), name.lower())]
|
||
if sid not in ids:
|
||
ids.append(sid)
|
||
ids.sort()
|
||
if ids != current:
|
||
plan["stations"] = (current, ids)
|
||
plans.append((row_out, plan))
|
||
|
||
# Stamps must stay unique after the whole file applies.
|
||
final: dict[Any, str | None] = {}
|
||
for _, plan in plans:
|
||
if "welder_stamp" in plan:
|
||
final[plan["user"]["id"]] = plan["welder_stamp"]
|
||
claimed = [s for s in final.values() if s]
|
||
if claimed:
|
||
others = {
|
||
str(r[1]).lower(): (r[0], r[2])
|
||
for r in conn.execute(
|
||
"SELECT id, welder_stamp, display_name FROM users WHERE welder_stamp IS NOT NULL"
|
||
" AND lower(welder_stamp) = ANY(%s)",
|
||
([s.lower() for s in claimed],),
|
||
).fetchall()
|
||
}
|
||
names_by_id = {plan["user"]["id"]: plan["user"]["display_name"] for _, plan in plans}
|
||
taken_in_file: dict[str, Any] = {}
|
||
for row_out, plan in plans:
|
||
s = plan.get("welder_stamp")
|
||
if not s:
|
||
continue
|
||
uid = plan["user"]["id"]
|
||
holder = others.get(s.lower())
|
||
# The holder keeps the stamp unless the same file gives them another (or clears it)
|
||
freed = holder is not None and holder[0] in final and str(final[holder[0]] or "").lower() != s.lower()
|
||
if holder is not None and holder[0] != uid and not freed:
|
||
row_out["errors"].append(f"Stamp {s} is taken by {holder[1]}.")
|
||
elif s.lower() in taken_in_file:
|
||
row_out["errors"].append(f"Stamp {s} is taken by {names_by_id[taken_in_file[s.lower()]]}.")
|
||
else:
|
||
taken_in_file[s.lower()] = uid
|
||
|
||
for row_out, plan in plans:
|
||
u = plan["user"]
|
||
if "welder_stamp" in plan:
|
||
row_out["changes"]["welder_stamp"] = {"from": u["welder_stamp"], "to": plan["welder_stamp"]}
|
||
if "profile_active" in plan:
|
||
row_out["changes"]["profile_active"] = {"from": u["profile_active"], "to": plan["profile_active"]}
|
||
if "stations" in plan:
|
||
cur, ids = plan["stations"]
|
||
row_out["changes"]["station_limits"] = {"from": _labels(cur, label), "to": _labels(ids, label)}
|
||
|
||
n_err = sum(1 for r in out if r["errors"])
|
||
body: dict[str, Any] = {"rows": out, "ok": len(out) - n_err, "errors": n_err, "applied": False}
|
||
if not apply:
|
||
return body
|
||
if n_err:
|
||
raise ApiError(
|
||
422,
|
||
"import_errors",
|
||
f"{n_err} row{'s' if n_err != 1 else ''} with errors. Nothing was changed.",
|
||
extra=body,
|
||
)
|
||
changed = 0
|
||
with conn.transaction():
|
||
for row_out, plan in plans:
|
||
if not row_out["changes"]:
|
||
continue
|
||
uid = plan["user"]["id"]
|
||
conn.execute("SELECT 1 FROM users WHERE id = %s FOR UPDATE", (uid,))
|
||
before = profile_of(conn, uid)
|
||
stamp = plan.get("welder_stamp", ...)
|
||
active = plan.get("profile_active", ...)
|
||
lim: dict[str, list[int]] | None = None
|
||
if "stations" in plan:
|
||
conn.execute("DELETE FROM user_station_limits WHERE user_id = %s", (uid,))
|
||
for sid in plan["stations"][1]:
|
||
conn.execute("INSERT INTO user_station_limits (user_id, station_id) VALUES (%s, %s)", (uid, sid))
|
||
_write(conn, uid, stamp, active, lim)
|
||
after = profile_of(conn, uid)
|
||
assert before is not None and after is not None
|
||
_audit(conn, who, uid, before, after, "import")
|
||
changed += 1
|
||
audit.append(
|
||
conn,
|
||
action="admin.profile_import",
|
||
user_id=who.user_id,
|
||
session_id=who.session_id,
|
||
role_snapshot=who.role_snapshot,
|
||
subject_type="users",
|
||
subject_id="import",
|
||
detail={"rows": len(out), "changed": changed, "unchanged": len(out) - changed},
|
||
ip=who.ip,
|
||
)
|
||
body["applied"] = True
|
||
return body
|