PacketStation/server/packetstation/domain/profiles.py

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