n.siegfried
2b597e68d8
ROOT CAUSE, exactly (the done-when asks for it):
Hold state was stored, twice, and derived nowhere.
1) Client: submitHold() wrote prevStatus='Issue', destroying the status the
hold interrupted at the moment it was placed - there was never anything to
return to. Clearing the last constraint then fell into the "Mark it as
Issued now?" confirm, because STATUS_ORDER.indexOf('Issue') is -1 and -1
reads as "before Issued". Decline it and the package stayed on hold with
zero open constraints, forever - the exact state reproduced live in front
of the Micron team.
2) Server: server/app.py's STATUS_ORDER put "Issue" at index 4, so
_released('Issue') was true and every transition OUT of hold skipped
enforce_release_gates() as "already released". POST /api/wps/{id}/status
could walk a held package to Issued past its open constraint. The comment
claimed the ladder was "mirrored in the front end"; the front end's ladder
has no 'Issue' in it at all.
What changed:
- setConstraint() recalculates hold state on EVERY constraint change: clearing
the last open constraint on a held package releases it immediately - no
refresh, no dialog - back to the status recorded on the hold entry (`from`),
which now rides on data.holds and survives save/reload.
- Every hold and release is history: pkgHolds entries carry ts, by, from/to,
reason; the exported Hold Log gained a By column; the server writes
hold_logged / hold_released audit rows (with the reason from data.holds) on
both the upsert and the /status endpoint.
- _released() no longer counts the hold: 'Issue' is a branch, not a rung.
Leaving hold to a field state re-runs the gates; entering hold never did and
still does not. The critical-reopen email keeps its old reach ("has been in
the field" includes on-hold).
- A1 preserved by name and by test: confirmEarlyRelease() still the one place
a gate override is written (comment-stripped grep asserts exactly one
pkgGateOverride assignment), still reason-first, still logged server-side.
D4 - what Urgent does (amended Aug 18): surface the audited path, add no new
one. confirmEarlyRelease() now also covers open constraints, but only for an
Urgent package, and the override must NAME every constraint it crosses - the
server refuses coverage by an old reason. The release banner gives an Urgent
package the override as its primary action (a real <button>); Normal and High
see nothing new and keep the same hard refusal, asserted per priority.
Banner button styled from tokens only; the banner now wraps at narrow widths.
Product question raised, not decided (per CLAUDE.md "asking versus assuming"):
Issue (hold) remains selectable from Draft and Scheduled, as it was before.
The done-when names no state list, so nothing was restricted. If a pre-release
hold is meaningless, closing it off is a one-line follow-up - needs Nick.
Verification (each probe run alone): NEW tests/hold_check.py 50/50, including
the clear-last-constraint regression specifically, the D4 priority matrix
against the server (six 409/200 cases), hold_logged/hold_released audit rows,
and an AST sweep proving every wp.status assignment in server/app.py sits in
a function that runs enforce_release_gates. Regressions: frame_check 39/39,
aggregates_check 16/16.
Items: CR-015, A1, D4 (X2 correction already recorded Aug 18)
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2814 lines
132 KiB
Python
2814 lines
132 KiB
Python
"""Work Package Suite API.
|
|
|
|
A small FastAPI service that stores project SOPs, Work Packages, and comments
|
|
in SQL (PostgreSQL in production; SQLite for local dev). NGINX serves the static
|
|
site and proxies /api/ here.
|
|
|
|
Run (dev): uvicorn server.app:app --reload --port 8000
|
|
Run (prod): gunicorn -k uvicorn.workers.UvicornWorker -b 127.0.0.1:8000 server.app:app
|
|
Interactive docs: http://<host>/api/docs
|
|
"""
|
|
import os
|
|
import re
|
|
import uuid
|
|
from datetime import timedelta, timezone
|
|
from time import monotonic
|
|
from typing import Any, Optional
|
|
from urllib.parse import urlparse
|
|
|
|
from fastapi import FastAPI, Depends, HTTPException, Query, Request, Response, BackgroundTasks
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
from sqlalchemy import select, delete, func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .db import Base, engine, get_db
|
|
from . import models, auth, notify
|
|
|
|
# Schema management:
|
|
# • Local dev (SQLite) auto-creates tables for a zero-config run.
|
|
# • Production (Postgres) owns its schema through Alembic migrations, which run
|
|
# at container start (`alembic upgrade head`, see Dockerfile / DEPLOYMENT.md).
|
|
# We must NOT create_all there, or it would race/collide with the migration.
|
|
if engine.dialect.name == "sqlite":
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
# Interactive docs are handy in dev but hand an attacker the full API map in prod,
|
|
# so enable them only on the SQLite dev fallback (production runs on Postgres).
|
|
_docs_enabled = engine.dialect.name == "sqlite"
|
|
app = FastAPI(
|
|
title="Work Package Suite API",
|
|
docs_url="/api/docs" if _docs_enabled else None,
|
|
redoc_url=None,
|
|
openapi_url="/api/openapi.json" if _docs_enabled else None,
|
|
)
|
|
|
|
# Same-origin in production (NGINX), so CORS is normally unnecessary. For
|
|
# cross-origin local dev, set CORS_ORIGINS="http://localhost:5500,..."
|
|
# allow_credentials is required so the browser sends the session cookie.
|
|
_origins = [o for o in os.getenv("CORS_ORIGINS", "").split(",") if o]
|
|
if _origins:
|
|
app.add_middleware(
|
|
CORSMiddleware, allow_origins=_origins, allow_credentials=True,
|
|
allow_methods=["*"], allow_headers=["*"], expose_headers=["X-Total-Count"],
|
|
)
|
|
|
|
|
|
# ── Authentication gate ────────────────────────────────────────────────────────
|
|
# Every /api/ data route requires a valid session cookie. Login, health, and the
|
|
# docs are exempt (see auth._needs_auth). This is the real security boundary —
|
|
# the static pages are only client-side guarded for UX. OPTIONS (CORS preflight)
|
|
# is always allowed so the browser can negotiate before sending credentials.
|
|
_UNSAFE_METHODS = {"POST", "PUT", "PATCH", "DELETE"}
|
|
|
|
|
|
def _csrf_ok(request: Request) -> bool:
|
|
"""CSRF defense-in-depth behind SameSite=Lax: when the browser sends an Origin
|
|
on a state-changing request, it must be same-origin (or an allowed CORS origin).
|
|
Non-browser clients (no Origin header) are unaffected."""
|
|
origin = request.headers.get("origin")
|
|
if not origin:
|
|
return True
|
|
if _origins and origin in _origins:
|
|
return True
|
|
try:
|
|
return urlparse(origin).netloc == request.headers.get("host", "")
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
@app.middleware("http")
|
|
async def auth_gate(request: Request, call_next):
|
|
path = request.url.path
|
|
method = request.method
|
|
if method != "OPTIONS" and auth._needs_auth(path):
|
|
if not auth.is_request_authenticated(request):
|
|
return JSONResponse(status_code=401, content={"detail": "Not authenticated"})
|
|
if method in _UNSAFE_METHODS and path.startswith("/api/") and not _csrf_ok(request):
|
|
return JSONResponse(status_code=403, content={"detail": "Cross-origin request rejected"})
|
|
return await call_next(request)
|
|
|
|
|
|
def gen_id(prefix: str) -> str:
|
|
return f"{prefix}_{uuid.uuid4().hex[:12]}"
|
|
|
|
|
|
# Clients may supply their own resource ids (offline-first). Constrain them to a
|
|
# safe charset so an id can never carry HTML/JS that a UI might place in markup.
|
|
_ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,40}$")
|
|
|
|
|
|
def check_id(v: Optional[str]) -> None:
|
|
if v and not _ID_RE.match(v):
|
|
raise HTTPException(status_code=400, detail="Invalid id format")
|
|
|
|
|
|
# ── Per-project access control ─────────────────────────────────────────────────
|
|
# A non-admin user may only touch projects they're a member of (project_members).
|
|
# Admins bypass all of this. Resources with no project_id (legacy/orphan) are not
|
|
# gated. List endpoints are scoped to accessible projects; single-resource and
|
|
# mutating endpoints raise 403 on no access.
|
|
def accessible_project_ids(db: Session, user: "models.User"):
|
|
"""Return the set of project ids the user may access, or None for 'all' (admin)."""
|
|
if auth.is_admin(user):
|
|
return None
|
|
rows = db.scalars(
|
|
select(models.ProjectMember.project_id).where(models.ProjectMember.user_id == user.id)
|
|
).all()
|
|
return set(rows)
|
|
|
|
|
|
def require_project_access(db: Session, user: "models.User", project_id: Optional[str]) -> None:
|
|
if auth.is_admin(user):
|
|
return
|
|
if not project_id:
|
|
# Non-admins may not read/mutate resources with no project assignment
|
|
# (orphan/legacy rows); only admins can touch project-less data.
|
|
raise HTTPException(status_code=403, detail="This resource is not assigned to a project you can access")
|
|
ok = db.scalar(
|
|
select(models.ProjectMember.id).where(
|
|
(models.ProjectMember.user_id == user.id)
|
|
& (models.ProjectMember.project_id == project_id)
|
|
)
|
|
)
|
|
if not ok:
|
|
raise HTTPException(status_code=403, detail="You don't have access to this project")
|
|
|
|
|
|
def effective_role(db: Session, user: "models.User", project_id: Optional[str]) -> str:
|
|
"""The user's permissions role ON THIS PROJECT.
|
|
|
|
An app admin is admin everywhere. Otherwise a membership row may carry its own
|
|
role — so a PM on one job can be a plain Project User on another — and an empty
|
|
membership role falls back to the account's own role."""
|
|
if auth.is_admin(user):
|
|
return auth.ROLE_ADMIN
|
|
if project_id:
|
|
row = db.scalars(
|
|
select(models.ProjectMember).where(
|
|
(models.ProjectMember.user_id == user.id)
|
|
& (models.ProjectMember.project_id == project_id)
|
|
)
|
|
).first()
|
|
if row and (row.role or "").strip():
|
|
return auth.normalize_role(row.role)
|
|
return auth.normalize_role(user.role)
|
|
|
|
|
|
def require_project_admin(db: Session, user: "models.User", project_id: Optional[str],
|
|
what: str = "this action") -> None:
|
|
"""Destructive / baseline-changing operations: deleting a work package or a
|
|
project, and editing a SOP that has already been completed. Requires project
|
|
access AND Project Admin *on that project*."""
|
|
require_project_access(db, user, project_id)
|
|
if effective_role(db, user, project_id) not in (
|
|
auth.ROLE_ADMIN, auth.ROLE_PROJECT_SUPER, auth.ROLE_PROJECT_ADMIN,
|
|
):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail=f"{what} requires the Project Admin role on this project",
|
|
)
|
|
|
|
|
|
def require_project_writable(db: Session, user_or_none, project_id: Optional[str],
|
|
what: str = "This change") -> None:
|
|
"""An archived project is frozen: everything on it stays readable, nothing on it
|
|
may be written. Every mutating path that lands on a project goes through here —
|
|
saving a SOP or a package, issuing, status changes, deletes, comments.
|
|
|
|
It is 409 and not 403 on purpose. Nobody lacks a permission here: the state of
|
|
the project is the objection, and even an admin gets refused until they unarchive
|
|
it — hence the caller identity is accepted for symmetry with the other guards but
|
|
deliberately unused. 409 also matters offline: the browser outbox in
|
|
html/project-data.js retires any 4xx op instead of retrying it forever, so an edit
|
|
queued before the archive dies quietly rather than looping against a frozen job."""
|
|
if not project_id:
|
|
return
|
|
proj = db.get(models.Project, project_id)
|
|
if proj is None:
|
|
# The project this write targets is gone — usually an outbox op queued before
|
|
# someone deleted the job. Refusing here is what keeps it a clean 409 instead
|
|
# of a foreign-key violation surfacing as a 500: the row could never be
|
|
# inserted anyway now that both engines enforce their FKs (see db.py). 409
|
|
# also matters because project-data.js retires a 4xx op and would retry a 5xx
|
|
# forever, so this is the difference between one quiet failure and a loop.
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=f"{what} — this project no longer exists. It was deleted, so there is "
|
|
f"nothing to save it against.",
|
|
)
|
|
if proj.archived_at is not None:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=(f"{what} — this project is archived (read-only). An administrator "
|
|
f"can unarchive it from the admin console."),
|
|
)
|
|
|
|
|
|
def scope_to_access(stmt, column, db: Session, user: "models.User"):
|
|
"""Restrict a SELECT to the user's accessible projects (no-op for admins)."""
|
|
ids = accessible_project_ids(db, user)
|
|
if ids is None:
|
|
return stmt
|
|
return stmt.where(column.in_(ids))
|
|
|
|
|
|
def grant_project_access(db: Session, user_id: str, project_id: str, role: str = "") -> bool:
|
|
"""Add a (user, project) membership if it isn't already there, carrying an
|
|
optional per-project role override ('' = inherit the account's own). Returns True
|
|
only when a row was actually added, so a caller can report what it did. Does NOT
|
|
commit — the caller owns the transaction."""
|
|
exists = db.scalar(
|
|
select(models.ProjectMember.id).where(
|
|
(models.ProjectMember.user_id == user_id)
|
|
& (models.ProjectMember.project_id == project_id)
|
|
)
|
|
)
|
|
if exists:
|
|
return False
|
|
db.add(models.ProjectMember(id=gen_id("pm"), user_id=user_id, project_id=project_id, role=role))
|
|
return True
|
|
|
|
|
|
def add_default_members(db: Session, project_id: str, actor) -> list[str]:
|
|
"""Put the standing default members onto a brand-new project.
|
|
|
|
Some people belong on every job the moment it exists — the PM who runs them all,
|
|
the QC lead — and flagging their account (users.auto_add_projects) is how an admin
|
|
says that once instead of remembering it at every project creation. App admins are
|
|
skipped because they already reach every project, an inactive account is not
|
|
revived by a new job, and an existing membership is never duplicated or
|
|
overwritten. Returns the usernames added; does NOT commit, matching
|
|
grant_project_access."""
|
|
rows = db.scalars(
|
|
select(models.User).where(
|
|
(models.User.auto_add_projects.is_(True)) & (models.User.is_active.is_(True))
|
|
).order_by(models.User.username)
|
|
).all()
|
|
added = []
|
|
for u in rows:
|
|
if auth.is_admin(u):
|
|
continue
|
|
if grant_project_access(db, u.id, project_id, (u.auto_add_role or "").strip()):
|
|
added.append(u.username)
|
|
if added:
|
|
# One line for the batch, not one per person — this is a single automatic act.
|
|
log_event(db, actor, "project_access_granted", "project", project_id,
|
|
project_id=project_id, summary=f"{len(added)} default member(s) added",
|
|
detail={"users": added, "reason": "auto_add_projects"})
|
|
return added
|
|
|
|
|
|
# ── User-administration scope ──────────────────────────────────────────────────
|
|
# User administration used to be one thing: an app admin did all of it. It is now
|
|
# two, because a project admin has to be able to staff their own job without an app
|
|
# admin on the phone. An app admin still manages every account; a PROJECT SUPER USER
|
|
# manages the accounts on the projects they hold that role on.
|
|
#
|
|
# Three questions, deliberately separate, because they have different answers:
|
|
# managed_project_ids which projects do I administer the users of?
|
|
# visible_user_ids whose entry may I SEE in the directory?
|
|
# manage_user_problem may I change this account? (much narrower than seeing it)
|
|
def managed_project_ids(db: Session, caller: "models.User") -> Optional[set[str]]:
|
|
"""Projects where `caller` may administer user accounts. None means every project
|
|
(an app admin). Read per-membership so the per-project override decides: a super
|
|
user demoted to plain member on one job does not administer its users, and an
|
|
ordinary account made super user on one job does administer that one."""
|
|
if auth.is_admin(caller):
|
|
return None
|
|
rows = db.scalars(
|
|
select(models.ProjectMember).where(models.ProjectMember.user_id == caller.id)
|
|
).all()
|
|
account_role = auth.normalize_role(caller.role)
|
|
out = set()
|
|
for r in rows:
|
|
role = auth.normalize_role(r.role) if (r.role or "").strip() else account_role
|
|
if role == auth.ROLE_PROJECT_SUPER:
|
|
out.add(r.project_id)
|
|
return out
|
|
|
|
|
|
def is_user_manager(db: Session, user: "models.User") -> bool:
|
|
"""May this account administer users at all? THE one definition — derived from the
|
|
managed set, never from the account role alone, because the super-user role can be
|
|
held on a single project (ProjectMember.role) by an otherwise ordinary account.
|
|
|
|
Falls out of it that a super user with no project memberships manages nobody,
|
|
which is right: the authority comes from the jobs, not the job title."""
|
|
managed = managed_project_ids(db, user)
|
|
return managed is None or bool(managed)
|
|
|
|
|
|
def require_user_manager(user: "models.User" = Depends(auth.get_current_user),
|
|
db: Session = Depends(get_db)) -> "models.User":
|
|
"""First gate on the user-administration routes: does this caller administer the
|
|
users of ANY project? Which accounts they may then touch is a second, narrower
|
|
check per target — require_manage_user."""
|
|
if not is_user_manager(db, user):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="Managing user accounts requires the Administrator role, or Project "
|
|
"Super User on a project",
|
|
)
|
|
return user
|
|
|
|
|
|
def member_project_ids(db: Session, user_id: str) -> set[str]:
|
|
"""Every project this user has a membership row for (no admin shortcut — this is
|
|
the raw set, which is exactly what the scope checks need to reason about)."""
|
|
return set(db.scalars(
|
|
select(models.ProjectMember.project_id).where(models.ProjectMember.user_id == user_id)
|
|
).all())
|
|
|
|
|
|
def visible_user_ids(db: Session, caller: "models.User") -> Optional[set[str]]:
|
|
"""Whose directory entry `caller` may read. None means everyone (an app admin).
|
|
|
|
Anyone signed in may look up the people they actually work with — their own
|
|
projects' members — plus the app admins, who are on every project implicitly and
|
|
are who you go to when something needs unblocking. Nobody else: the directory
|
|
must not become a company-wide address book for a single-project subcontractor."""
|
|
if auth.is_admin(caller):
|
|
return None
|
|
ids = {caller.id}
|
|
mine = accessible_project_ids(db, caller) or set()
|
|
if mine:
|
|
ids |= set(db.scalars(
|
|
select(models.ProjectMember.user_id).where(models.ProjectMember.project_id.in_(mine))
|
|
).all())
|
|
ids |= set(db.scalars(
|
|
select(models.User.id).where(models.User.role == auth.ROLE_ADMIN)
|
|
).all())
|
|
return ids
|
|
|
|
|
|
def manage_user_problem(db: Session, caller: "models.User", target: "models.User",
|
|
cache: Optional[dict] = None) -> Optional[str]:
|
|
"""None if `caller` may make ACCOUNT-level changes to `target` (password, name,
|
|
permissions role, enable/disable, delete); otherwise the reason they may not, in
|
|
words the console can show verbatim.
|
|
|
|
An app admin may always. A super user may only when the account sits ENTIRELY
|
|
inside the projects they administer, and is not itself an admin or super user.
|
|
Both limits matter:
|
|
• Exclusive scope, because these changes are global. Resetting a password or
|
|
disabling an account reaches every project that person is on, so a super user
|
|
must not be able to reach into a job they don't run by way of a shared member.
|
|
• No admin/super targets, because otherwise the role could be used to take over
|
|
a peer's account and inherit their scope.
|
|
Project-scoped changes (adding someone to MY project, their role THERE) are not
|
|
account-level and are checked against `managed_project_ids` instead.
|
|
|
|
`cache` lets a caller judging a whole page of users hand in the two lookups this
|
|
needs ('managed', and 'members' as {user_id: {project_id}}) so the verdict for
|
|
thirty rows costs two queries instead of sixty. The rule itself lives only here."""
|
|
if auth.is_admin(caller):
|
|
return None
|
|
cache = cache if cache is not None else {}
|
|
managed = cache.get("managed")
|
|
if managed is None:
|
|
managed = cache["managed"] = managed_project_ids(db, caller) or set()
|
|
if not managed:
|
|
return ("You don't administer the users of any project — that needs the Project "
|
|
"Super User role on the project")
|
|
if auth.normalize_role(target.role) in (auth.ROLE_ADMIN, auth.ROLE_PROJECT_SUPER):
|
|
return "Only an application administrator can change an Administrator or Project Super User account"
|
|
members = cache.get("members")
|
|
theirs = members.get(target.id, set()) if members is not None else member_project_ids(db, target.id)
|
|
if not theirs:
|
|
return ("This account isn't on any project, so only an application administrator "
|
|
"can change it")
|
|
outside = theirs - managed
|
|
if outside:
|
|
return (f"{target.username} is also on {len(outside)} project(s) you don't administer — "
|
|
"account changes there have to come from an application administrator. "
|
|
"You can still change their access and role on your own projects.")
|
|
return None
|
|
|
|
|
|
def require_manage_user(db: Session, caller: "models.User", target: "models.User") -> None:
|
|
problem = manage_user_problem(db, caller, target)
|
|
if problem:
|
|
raise HTTPException(status_code=403, detail=problem)
|
|
|
|
|
|
def require_see_user(db: Session, caller: "models.User", target: "models.User") -> None:
|
|
"""404, not 403: whether an account exists outside your projects is itself not
|
|
yours to learn, and a 403 would confirm the username."""
|
|
visible = visible_user_ids(db, caller)
|
|
if visible is not None and target.id not in visible:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
|
|
def grantable_roles(caller: "models.User") -> tuple:
|
|
"""Permissions roles `caller` may hand out. A super user may staff their job with
|
|
project admins and project users — never another admin or super user, which is the
|
|
line that keeps the role from being a route to app-wide control."""
|
|
if auth.is_admin(caller):
|
|
return auth.ROLES
|
|
return (auth.ROLE_PROJECT_ADMIN, auth.ROLE_PROJECT_USER)
|
|
|
|
|
|
def load_target_user(db: Session, user_id: str) -> "models.User":
|
|
u = db.get(models.User, user_id)
|
|
if not u:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return u
|
|
|
|
|
|
# ── Audit trail ────────────────────────────────────────────────────────────────
|
|
def log_event(db: Session, actor, action: str, entity_type: str, entity_id: str,
|
|
project_id: Optional[str] = None, summary: str = "", detail: Optional[dict] = None) -> None:
|
|
"""Append an audit-trail row in the CURRENT transaction so it commits
|
|
atomically with the change it describes. `actor` may be a User or a username."""
|
|
who = actor.username if isinstance(actor, models.User) else (actor or "")
|
|
db.add(models.AuditLog(
|
|
id=gen_id("ev"), actor=who, action=action, entity_type=entity_type,
|
|
entity_id=entity_id or "", project_id=project_id, summary=(summary or "")[:400], detail=detail or {},
|
|
))
|
|
|
|
|
|
# ── Assignment ─────────────────────────────────────────────────────────────────
|
|
def require_assignable(db: Session, user_id: str, project_id: Optional[str]) -> None:
|
|
"""A WP can only be assigned to an active user who can access its project."""
|
|
u = db.get(models.User, user_id)
|
|
if not u or not u.is_active:
|
|
raise HTTPException(status_code=400, detail="Assignee is not a valid user")
|
|
if auth.is_admin(u):
|
|
return
|
|
ok = db.scalar(
|
|
select(models.ProjectMember.id).where(
|
|
(models.ProjectMember.user_id == user_id) & (models.ProjectMember.project_id == project_id)
|
|
)
|
|
)
|
|
if not ok:
|
|
raise HTTPException(status_code=400, detail="Assignee is not a member of this project")
|
|
|
|
|
|
def wp_link(db: Session, wp: "models.WorkPackage") -> str:
|
|
base = (notify.get_settings(db).get("app_base_url") or "").rstrip("/")
|
|
path = f"/work-package-suite.html?tab=wp&project={wp.project_id or ''}"
|
|
return (base + path) if base else path
|
|
|
|
|
|
def assign_body(assignee: "models.User", wp: "models.WorkPackage", actor: "models.User", link: str) -> str:
|
|
# Deliberately minimal — a WP number + a link, NOT the package contents (keeps
|
|
# customer IP inside the app, behind login).
|
|
who = actor.full_name or actor.username
|
|
name = assignee.full_name or assignee.username
|
|
return (
|
|
f"Hi {name},\n\n"
|
|
f"{who} assigned you a work package: {wp.number or '(no number)'}.\n\n"
|
|
f"Open the Work Package Suite to view and action it:\n{link}\n\n"
|
|
f"— This is an automated message from the Work Package Suite."
|
|
)
|
|
|
|
|
|
# ── Request bodies ───────────────────────────────────────────────────────────
|
|
class ProjectIn(BaseModel):
|
|
id: Optional[str] = None
|
|
name: str = ""
|
|
number: str = ""
|
|
client: str = ""
|
|
division: str = ""
|
|
site: str = ""
|
|
sample: bool = False
|
|
created_by: str = ""
|
|
data: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class SopIn(BaseModel):
|
|
id: Optional[str] = None
|
|
project_id: Optional[str] = None
|
|
name: str = ""
|
|
number: str = ""
|
|
complete: bool = False
|
|
created_by: str = ""
|
|
data: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class WpIn(BaseModel):
|
|
id: Optional[str] = None
|
|
project_id: Optional[str] = None
|
|
sop_id: Optional[str] = None
|
|
parent_id: Optional[str] = None
|
|
number: str = ""
|
|
subject: str = ""
|
|
type: str = ""
|
|
status: str = "Draft"
|
|
assignee_id: Optional[str] = None
|
|
created_by: str = ""
|
|
data: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class SettingsIn(BaseModel):
|
|
model_config = ConfigDict(extra="ignore")
|
|
email_enabled: Optional[bool] = None
|
|
smtp_host: Optional[str] = None
|
|
smtp_port: Optional[int] = None
|
|
smtp_use_tls: Optional[bool] = None
|
|
smtp_username: Optional[str] = None
|
|
from_addr: Optional[str] = None
|
|
from_name: Optional[str] = None
|
|
app_base_url: Optional[str] = None
|
|
bim_enabled: Optional[bool] = None
|
|
default_locale: Optional[str] = None
|
|
default_timezone: Optional[str] = None
|
|
|
|
|
|
class TestEmailIn(BaseModel):
|
|
to: Optional[str] = None
|
|
|
|
|
|
class StatusIn(BaseModel):
|
|
status: str
|
|
|
|
|
|
class ArchiveIn(BaseModel):
|
|
archived: bool = True
|
|
|
|
|
|
class CommentIn(BaseModel):
|
|
# Tolerate any extra keys the feedback payload includes (timestamp, app, …).
|
|
model_config = ConfigDict(extra="allow")
|
|
source: Optional[str] = None
|
|
type: Optional[str] = None # client sends 'type'; treated as source
|
|
sop_id: Optional[str] = None
|
|
wp_id: Optional[str] = None
|
|
step: Optional[int] = None
|
|
author: Optional[str] = None
|
|
name: Optional[str] = None # home/SOP forms send 'name'
|
|
text: Optional[str] = None
|
|
page: Optional[str] = ""
|
|
|
|
|
|
# ── Health ───────────────────────────────────────────────────────────────────
|
|
@app.get("/api/health")
|
|
def health():
|
|
return {"ok": True}
|
|
|
|
|
|
# ── Authentication ─────────────────────────────────────────────────────────────
|
|
class LoginIn(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
|
|
class NewUserIn(BaseModel):
|
|
username: str
|
|
password: str
|
|
full_name: str = ""
|
|
email: str = ""
|
|
role: str = auth.ROLE_PROJECT_USER # permissions role — see auth.ROLES
|
|
project_role: str = "" # job function on the project (no permissions)
|
|
# Projects to put the new account on straight away. Optional for an app admin
|
|
# (who can assign later); REQUIRED for a super user, whose authority over an
|
|
# account comes from the projects it is on — see create_user.
|
|
project_ids: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class ProjectRoleIn(BaseModel):
|
|
project_role: str = ""
|
|
|
|
|
|
class PreferencesIn(BaseModel):
|
|
# Empty string clears the preference (fall back to the app default, then the
|
|
# browser). None means "leave this one alone".
|
|
locale: Optional[str] = None
|
|
timezone: Optional[str] = None
|
|
|
|
|
|
class ForgotPasswordIn(BaseModel):
|
|
username: str = "" # username or email
|
|
|
|
|
|
class ResetPasswordIn(BaseModel):
|
|
token: str
|
|
new_password: str
|
|
|
|
|
|
class PasswordChangeIn(BaseModel):
|
|
current_password: str
|
|
new_password: str
|
|
|
|
|
|
class AdminPasswordIn(BaseModel):
|
|
new_password: str
|
|
|
|
|
|
class ActiveIn(BaseModel):
|
|
is_active: bool
|
|
|
|
|
|
class RoleIn(BaseModel):
|
|
role: str # permissions role — see auth.ROLES
|
|
|
|
|
|
class ProjectAssignIn(BaseModel):
|
|
project_ids: list[str] = Field(default_factory=list)
|
|
# Optional per-project permissions role, {project_id: role}. Omit or use '' to
|
|
# inherit the account's own role on that project.
|
|
roles: dict[str, str] = Field(default_factory=dict)
|
|
|
|
|
|
class AutoAddIn(BaseModel):
|
|
auto_add: bool
|
|
# Role to give this user on the projects they're auto-added to; '' inherits the
|
|
# account's own role, same value space as ProjectMember.role.
|
|
role: str = ""
|
|
|
|
|
|
LOGIN_MAX_ATTEMPTS = int(os.getenv("AUTH_MAX_ATTEMPTS", "5"))
|
|
LOGIN_LOCKOUT_MINUTES = int(os.getenv("AUTH_LOCKOUT_MINUTES", "15"))
|
|
|
|
|
|
@app.post("/api/auth/login")
|
|
def login(body: LoginIn, request: Request, response: Response, db: Session = Depends(get_db)):
|
|
"""Verify credentials and, on success, set the HttpOnly session cookie.
|
|
Throttles online password guessing: after LOGIN_MAX_ATTEMPTS consecutive
|
|
failures an account is locked for LOGIN_LOCKOUT_MINUTES."""
|
|
user = auth.find_user(db, body.username)
|
|
now = models.utcnow()
|
|
# Always run the hash comparison first — even for missing or locked accounts —
|
|
# so response timing doesn't leak which usernames exist. verify_password
|
|
# tolerates an empty hash.
|
|
valid = auth.verify_password(body.password, user.password_hash if user else "")
|
|
locked = user.locked_until if user else None
|
|
if locked is not None and locked.tzinfo is None:
|
|
locked = locked.replace(tzinfo=timezone.utc) # SQLite returns naive datetimes; normalize to UTC
|
|
if locked is not None and locked > now:
|
|
raise HTTPException(status_code=429, detail="Too many failed attempts. Try again later.")
|
|
if not user or not valid:
|
|
if user:
|
|
user.failed_attempts = (user.failed_attempts or 0) + 1
|
|
if user.failed_attempts >= LOGIN_MAX_ATTEMPTS:
|
|
user.locked_until = now + timedelta(minutes=LOGIN_LOCKOUT_MINUTES)
|
|
user.failed_attempts = 0
|
|
log_event(db, user.username, "login_locked", "user", user.id, summary=user.username,
|
|
detail={"minutes": LOGIN_LOCKOUT_MINUTES})
|
|
db.commit()
|
|
raise HTTPException(status_code=401, detail="Invalid username or password")
|
|
if not user.is_active:
|
|
raise HTTPException(status_code=403, detail="Account is disabled")
|
|
user.failed_attempts = 0
|
|
user.locked_until = None
|
|
user.last_login_at = now
|
|
db.commit()
|
|
token = auth.create_token(user)
|
|
auth.set_session_cookie(response, request, token)
|
|
return {"user": user.to_dict()}
|
|
|
|
|
|
@app.post("/api/auth/logout")
|
|
def logout(response: Response):
|
|
auth.clear_session_cookie(response)
|
|
return {"ok": True}
|
|
|
|
|
|
# ── Self-service password reset (needs email switched on) ──────────────────────
|
|
RESET_COOLDOWN_SECONDS = int(os.getenv("AUTH_RESET_COOLDOWN_SECONDS", "120"))
|
|
# In-process throttle: one reset mail per (account, client) per cooldown. Enough to
|
|
# stop someone using the form to spam a colleague's inbox. Per-worker and lost on
|
|
# restart — deliberately simple; the token expiry is the real control.
|
|
_reset_last: dict[str, float] = {}
|
|
|
|
|
|
def _reset_throttled(request: Request, username: str) -> bool:
|
|
now = monotonic()
|
|
key = f"{(username or '').strip().lower()}|{request.client.host if request.client else ''}"
|
|
prev = _reset_last.get(key)
|
|
if prev is not None and (now - prev) < RESET_COOLDOWN_SECONDS:
|
|
return True
|
|
_reset_last[key] = now
|
|
if len(_reset_last) > 5000: # bound the dict on a long-lived worker
|
|
cutoff = now - RESET_COOLDOWN_SECONDS
|
|
for k in [k for k, t in _reset_last.items() if t < cutoff]:
|
|
_reset_last.pop(k, None)
|
|
return False
|
|
|
|
|
|
def reset_body(user: "models.User", link: str, minutes: int) -> str:
|
|
# No account detail beyond the username, and no customer data — same rule as
|
|
# the assignment mail. The link is the only sensitive thing in here.
|
|
who = user.full_name or user.username
|
|
return (
|
|
f"Hi {who},\n\n"
|
|
f"A password reset was requested for your Work Package Suite account "
|
|
f"({user.username}).\n\n"
|
|
f"Set a new password:\n{link}\n\n"
|
|
f"The link expires in {minutes} minutes and can only be used once. "
|
|
f"If you didn't request this, you can ignore this email — your current "
|
|
f"password still works.\n"
|
|
)
|
|
|
|
|
|
@app.get("/api/auth/reset-available")
|
|
def reset_available(db: Session = Depends(get_db)):
|
|
"""Whether the login page should offer 'Forgot password'. Self-service reset
|
|
depends entirely on outbound email, so it's off unless email is enabled AND
|
|
SMTP is configured — otherwise the only route is an admin reset."""
|
|
s = notify.get_settings(db)
|
|
return {"enabled": bool(s.get("email_enabled")) and notify.smtp_ready(s)}
|
|
|
|
|
|
@app.post("/api/auth/forgot-password")
|
|
def forgot_password(body: ForgotPasswordIn, request: Request, db: Session = Depends(get_db)):
|
|
"""Email a reset link. Always returns the same 200 response whether or not the
|
|
account exists — this endpoint is unauthenticated, so it must not become a
|
|
username/email oracle. Failures are recorded in the audit log instead."""
|
|
s = notify.get_settings(db)
|
|
if not (s.get("email_enabled") and notify.smtp_ready(s)):
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="Password reset by email isn't available. Ask an administrator to reset it for you.",
|
|
)
|
|
if _reset_throttled(request, body.username):
|
|
# Same shape as the success response — no oracle, no mail bomb.
|
|
return {"ok": True, "message": "If that account exists, a reset link is on its way."}
|
|
user = auth.find_user(db, body.username)
|
|
if user and user.is_active and user.email:
|
|
base = (s.get("app_base_url") or "").rstrip("/")
|
|
token = auth.create_reset_token(user)
|
|
link = f"{base}/login.html?reset={token}" if base else f"/login.html?reset={token}"
|
|
sent = notify.send_now(
|
|
db, user.email,
|
|
"Work Package Suite — reset your password",
|
|
reset_body(user, link, auth.RESET_MINUTES),
|
|
)
|
|
log_event(db, user.username, "password_reset_requested", "user", user.id,
|
|
summary=user.username, detail={"emailed": bool(sent)})
|
|
db.commit()
|
|
else:
|
|
# Log the miss for the admin's benefit; the caller can't tell the difference.
|
|
log_event(db, "(anonymous)", "password_reset_miss", "user", "",
|
|
summary=(body.username or "")[:200],
|
|
detail={"reason": "no account, inactive, or no email on file"})
|
|
db.commit()
|
|
return {"ok": True, "message": "If that account exists, a reset link is on its way."}
|
|
|
|
|
|
@app.post("/api/auth/reset-password")
|
|
def reset_password(body: ResetPasswordIn, db: Session = Depends(get_db)):
|
|
"""Complete a reset using the emailed token. The token carries the user's
|
|
token_version, and finishing a reset bumps it — so the link is single-use and
|
|
every existing session for that account is signed out."""
|
|
claims = auth.decode_reset_token(body.token or "")
|
|
if not claims:
|
|
raise HTTPException(status_code=400, detail="This reset link is invalid or has expired. Request a new one.")
|
|
user = db.get(models.User, claims.get("sub"))
|
|
if not user or not user.is_active:
|
|
raise HTTPException(status_code=400, detail="This reset link is no longer valid.")
|
|
if (claims.get("ver", 0) or 0) != (user.token_version or 0):
|
|
raise HTTPException(status_code=400, detail="This reset link has already been used. Request a new one.")
|
|
problem = auth.password_problem(body.new_password, user.username, user.email)
|
|
if problem:
|
|
raise HTTPException(status_code=400, detail=problem)
|
|
user.password_hash = auth.hash_password(body.new_password)
|
|
user.token_version = (user.token_version or 0) + 1 # burns the link + all sessions
|
|
# A completed reset also clears any login lockout — the person has proven
|
|
# control of the mailbox, so there's nothing left to throttle.
|
|
user.failed_attempts = 0
|
|
user.locked_until = None
|
|
log_event(db, user.username, "password_reset", "user", user.id, summary=user.username)
|
|
db.commit()
|
|
return {"ok": True}
|
|
|
|
|
|
@app.get("/api/auth/me")
|
|
def whoami(user: models.User = Depends(auth.get_current_user)):
|
|
"""Who is logged in. The frontend guard calls this on every page load.
|
|
|
|
`role` is normalized here so no page has to know that a pre-roles account stores
|
|
'user' where it now means 'project_user'."""
|
|
return {"user": {**user.to_dict(), "role": auth.normalize_role(user.role)}}
|
|
|
|
|
|
# ── Display preferences (self-service) ─────────────────────────────────────────
|
|
_LOCALE_RE = re.compile(r"^[A-Za-z]{2,3}(-[A-Za-z0-9]{2,8}){0,3}$")
|
|
|
|
|
|
def valid_timezone(tz: str) -> bool:
|
|
"""True if this is a real IANA zone name on this machine."""
|
|
try:
|
|
from zoneinfo import ZoneInfo
|
|
ZoneInfo(tz)
|
|
return True
|
|
except Exception: # noqa: BLE001 — unknown key, missing tzdata, bad type
|
|
return False
|
|
|
|
|
|
@app.get("/api/timezones")
|
|
def list_timezones(_user: models.User = Depends(auth.get_current_user)):
|
|
"""IANA zone names for the preferences picker, so the list matches what the
|
|
server will actually accept."""
|
|
try:
|
|
from zoneinfo import available_timezones
|
|
return sorted(available_timezones())
|
|
except Exception: # noqa: BLE001 — no tzdata: let the client fall back
|
|
return []
|
|
|
|
|
|
@app.post("/api/auth/preferences")
|
|
def set_preferences(body: PreferencesIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
"""A user's own locale / timezone. Empty string clears the preference; the app
|
|
default (admin console) applies next, then the browser's own settings."""
|
|
if body.locale is not None:
|
|
loc = body.locale.strip()
|
|
if loc and not _LOCALE_RE.match(loc):
|
|
raise HTTPException(status_code=400, detail="Locale must be a language tag like 'en-US' or 'es'.")
|
|
user.locale = loc[:20]
|
|
if body.timezone is not None:
|
|
tz = body.timezone.strip()
|
|
if tz and not valid_timezone(tz):
|
|
raise HTTPException(status_code=400, detail="Unknown time zone. Pick one from the list.")
|
|
user.timezone = tz[:60]
|
|
db.commit()
|
|
db.refresh(user)
|
|
return {"user": user.to_dict()}
|
|
|
|
|
|
@app.post("/api/auth/password")
|
|
def change_password(body: PasswordChangeIn, request: Request, response: Response, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
if not auth.verify_password(body.current_password, user.password_hash):
|
|
raise HTTPException(status_code=400, detail="Current password is incorrect")
|
|
problem = auth.password_problem(body.new_password, user.username, user.email)
|
|
if problem:
|
|
raise HTTPException(status_code=400, detail=problem)
|
|
user.password_hash = auth.hash_password(body.new_password)
|
|
user.token_version = (user.token_version or 0) + 1 # invalidate all OTHER existing sessions
|
|
db.commit()
|
|
db.refresh(user)
|
|
# Keep this session logged in by re-issuing a cookie carrying the new version.
|
|
auth.set_session_cookie(response, request, auth.create_token(user))
|
|
return {"ok": True}
|
|
|
|
|
|
# ── User administration ─────────────────────────────────────────────────────────
|
|
# Two kinds of caller reach these routes: an app admin, who manages every account,
|
|
# and a Project Super User, who manages the accounts on the projects they administer.
|
|
# Every route therefore asks TWO questions — does this role carry user administration
|
|
# (require_user_manager), and may it touch THIS account (require_manage_user) —
|
|
# and the read route asks a third, wider one (visible_user_ids) because looking a
|
|
# colleague up is not the same as being able to change them.
|
|
def directory_entry(db: Session, u: "models.User", caller: "models.User",
|
|
counts: Optional[dict] = None, cache: Optional[dict] = None) -> dict:
|
|
"""One row of the user directory, cut to what `caller` is entitled to see.
|
|
|
|
A manager gets the administrative record (last login, the auto-add flags, and a
|
|
`manageable` verdict with the reason when it's no). Everyone else gets the contact
|
|
card only — a project user has no business reading their colleagues' login history
|
|
out of a page whose job is "who is on this project and how do I reach them"."""
|
|
if not is_user_manager(db, caller):
|
|
return {
|
|
"id": u.id, "username": u.username, "full_name": u.full_name, "email": u.email,
|
|
"role": auth.normalize_role(u.role), "project_role": u.project_role or "",
|
|
"is_active": u.is_active, "manageable": False,
|
|
}
|
|
problem = manage_user_problem(db, caller, u, cache)
|
|
n = None if counts is None else counts.get(u.id, 0)
|
|
return {
|
|
**u.to_dict(),
|
|
"role": auth.normalize_role(u.role),
|
|
"manageable": problem is None,
|
|
"manage_blocked_reason": problem or "",
|
|
"project_count": n,
|
|
}
|
|
|
|
|
|
@app.get("/api/auth/users")
|
|
def list_users(user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
"""The user directory, scoped to the caller. An app admin sees every account; a
|
|
project member sees the people on their own projects (plus the app admins)."""
|
|
visible = visible_user_ids(db, user)
|
|
stmt = select(models.User).order_by(models.User.username)
|
|
if visible is not None:
|
|
stmt = stmt.where(models.User.id.in_(visible))
|
|
rows = db.scalars(stmt).all()
|
|
# One pass over the membership table serves both the project-access count and the
|
|
# per-row "may I manage this account" verdict. The old console fetched the counts
|
|
# with one HTTP request per user.
|
|
counts, cache = None, None
|
|
if is_user_manager(db, user):
|
|
members: dict[str, set] = {}
|
|
for uid, pid in db.execute(
|
|
select(models.ProjectMember.user_id, models.ProjectMember.project_id)
|
|
).all():
|
|
members.setdefault(uid, set()).add(pid)
|
|
counts = {uid: len(pids) for uid, pids in members.items()}
|
|
cache = {"members": members}
|
|
return [directory_entry(db, u, user, counts, cache) for u in rows]
|
|
|
|
|
|
@app.get("/api/auth/user-scope")
|
|
def user_scope(user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
"""What the signed-in account may do on the user directory page, so the page can
|
|
render the right controls instead of guessing at the rules and drawing buttons
|
|
that 403. Advisory only — every route re-checks server-side."""
|
|
managed = managed_project_ids(db, user)
|
|
if managed is None:
|
|
rows = db.scalars(select(models.Project).order_by(models.Project.name)).all()
|
|
else:
|
|
rows = db.scalars(
|
|
select(models.Project).where(models.Project.id.in_(managed)).order_by(models.Project.name)
|
|
).all() if managed else []
|
|
manager = managed is None or bool(managed)
|
|
return {
|
|
"can_manage_users": manager,
|
|
"scope": "all" if managed is None else "projects",
|
|
"role": auth.normalize_role(user.role),
|
|
"grantable_roles": list(grantable_roles(user)) if manager else [],
|
|
"grantable_project_roles": list(
|
|
auth.PROJECT_SCOPED_ROLES if auth.is_admin(user)
|
|
else (auth.ROLE_PROJECT_ADMIN, auth.ROLE_PROJECT_USER)
|
|
) if manager else [],
|
|
"role_labels": auth.ROLE_LABELS,
|
|
"project_roles": list(auth.PROJECT_ROLES),
|
|
"managed_projects": [{"id": p.id, "name": p.name, "number": p.number,
|
|
"archived": p.archived_at is not None} for p in rows],
|
|
}
|
|
|
|
|
|
@app.post("/api/auth/users")
|
|
def create_user(body: NewUserIn, actor: models.User = Depends(require_user_manager), db: Session = Depends(get_db)):
|
|
problem = auth.password_problem(body.password, body.username, body.email)
|
|
if problem:
|
|
raise HTTPException(status_code=400, detail=problem)
|
|
allowed = grantable_roles(actor)
|
|
if body.role not in allowed:
|
|
raise HTTPException(status_code=400, detail=f"role must be one of {', '.join(allowed)}")
|
|
if auth.find_user(db, body.username):
|
|
raise HTTPException(status_code=409, detail="A user with that username already exists")
|
|
managed = managed_project_ids(db, actor)
|
|
requested = [p for p in dict.fromkeys(body.project_ids) if p]
|
|
for pid in requested:
|
|
check_id(pid)
|
|
if managed is not None:
|
|
# A super user's authority over an account is derived from the projects that
|
|
# account is on. Creating one with no project — or on a job they don't run —
|
|
# would either produce an account they instantly cannot manage, or reach into
|
|
# someone else's job. Both are refused rather than silently narrowed.
|
|
if not requested:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Choose at least one project for the new account — you administer users per project",
|
|
)
|
|
outside = [p for p in requested if p not in managed]
|
|
if outside:
|
|
raise HTTPException(status_code=403, detail="You don't administer the users of one of those projects")
|
|
valid = set(db.scalars(select(models.Project.id).where(models.Project.id.in_(requested))).all()) if requested else set()
|
|
missing = [p for p in requested if p not in valid]
|
|
if missing:
|
|
raise HTTPException(status_code=400, detail="One of those projects no longer exists")
|
|
u = models.User(
|
|
id=gen_id("user"),
|
|
username=body.username.strip(),
|
|
email=body.email.strip(),
|
|
full_name=body.full_name.strip(),
|
|
password_hash=auth.hash_password(body.password),
|
|
role=body.role,
|
|
project_role=body.project_role.strip()[:120],
|
|
)
|
|
db.add(u)
|
|
# Flush the account before adding memberships that point at it. The ORM decides
|
|
# flush order from relationship() declarations, and models.py deliberately has
|
|
# none (plain columns + ForeignKey), so it will happily emit the project_members
|
|
# INSERT before the users one — which the database then rejects. Without this the
|
|
# whole call fails with a foreign-key violation on any engine that actually
|
|
# enforces them, which is every engine we run: Postgres always, and SQLite since
|
|
# db.py started setting `PRAGMA foreign_keys=ON`.
|
|
db.flush()
|
|
log_event(db, actor, "user_created", "user", u.id, summary=u.username,
|
|
detail={"role": u.role, "project_role": u.project_role,
|
|
"projects": len(valid)})
|
|
for pid in requested:
|
|
if pid in valid:
|
|
grant_project_access(db, u.id, pid)
|
|
if valid:
|
|
log_event(db, actor, "project_access_changed", "user", u.id, summary=u.username,
|
|
detail={"projects": len(valid), "reason": "created_with_access"})
|
|
db.commit()
|
|
db.refresh(u)
|
|
return directory_entry(db, u, actor)
|
|
|
|
|
|
@app.post("/api/auth/users/{user_id}/password")
|
|
def admin_reset_password(user_id: str, body: AdminPasswordIn, actor: models.User = Depends(require_user_manager), db: Session = Depends(get_db)):
|
|
u = load_target_user(db, user_id)
|
|
require_see_user(db, actor, u)
|
|
require_manage_user(db, actor, u)
|
|
problem = auth.password_problem(body.new_password, u.username, u.email)
|
|
if problem:
|
|
raise HTTPException(status_code=400, detail=problem)
|
|
u.password_hash = auth.hash_password(body.new_password)
|
|
u.token_version = (u.token_version or 0) + 1 # revoke the user's existing sessions
|
|
# An administrative password reset was the one user-account change that left no
|
|
# trace; it is the most impersonation-adjacent thing on this page, so it logs.
|
|
log_event(db, actor, "password_reset", "user", u.id, summary=u.username,
|
|
detail={"by": "administrator"})
|
|
db.commit()
|
|
return {"ok": True}
|
|
|
|
|
|
@app.post("/api/auth/users/{user_id}/active")
|
|
def set_user_active(user_id: str, body: ActiveIn, actor: models.User = Depends(require_user_manager), db: Session = Depends(get_db)):
|
|
u = load_target_user(db, user_id)
|
|
require_see_user(db, actor, u)
|
|
require_manage_user(db, actor, u)
|
|
if u.id == actor.id and not body.is_active:
|
|
raise HTTPException(status_code=400, detail="You cannot disable your own account")
|
|
u.is_active = body.is_active
|
|
log_event(db, actor, "user_enabled" if body.is_active else "user_disabled", "user", u.id,
|
|
summary=u.username, detail={"is_active": bool(body.is_active)})
|
|
db.commit()
|
|
return directory_entry(db, u, actor)
|
|
|
|
|
|
@app.post("/api/auth/users/{user_id}/role")
|
|
def set_user_role(user_id: str, body: RoleIn, actor: models.User = Depends(require_user_manager), db: Session = Depends(get_db)):
|
|
"""Change a user's PERMISSIONS role. Their job function on the project is separate
|
|
— see set_user_project_role.
|
|
|
|
Guards: you can't change your own role (avoids self-lockout), the last remaining
|
|
admin can't be demoted (keeps the app manageable), and a super user may only hand
|
|
out the roles in `grantable_roles` — never admin or another super user."""
|
|
allowed = grantable_roles(actor)
|
|
if body.role not in allowed:
|
|
raise HTTPException(status_code=400, detail=f"role must be one of {', '.join(allowed)}")
|
|
u = load_target_user(db, user_id)
|
|
require_see_user(db, actor, u)
|
|
require_manage_user(db, actor, u)
|
|
if u.id == actor.id:
|
|
raise HTTPException(status_code=400, detail="You cannot change your own role")
|
|
if auth.is_admin(u) and body.role != auth.ROLE_ADMIN:
|
|
other_admins = db.scalars(
|
|
select(models.User.id).where(
|
|
(models.User.role == auth.ROLE_ADMIN)
|
|
& (models.User.id != u.id)
|
|
& (models.User.is_active.is_(True))
|
|
)
|
|
).all()
|
|
if not other_admins:
|
|
raise HTTPException(status_code=400, detail="Can't remove the last admin account")
|
|
old_role = u.role
|
|
u.role = body.role
|
|
detail = {"from": old_role, "to": body.role}
|
|
# Promoting someone to admin retires their default-member flag: an admin already
|
|
# reaches every project, so the flag would do nothing except sit there invisibly
|
|
# (the console shows admins no controls) and come back to life the day they are
|
|
# demoted. Same reasoning as clearing the role in set_user_auto_add.
|
|
if body.role == auth.ROLE_ADMIN and (u.auto_add_projects or u.auto_add_role):
|
|
u.auto_add_projects = False
|
|
u.auto_add_role = ""
|
|
detail["auto_add_cleared"] = True
|
|
log_event(db, actor, "role_changed", "user", u.id, summary=u.username, detail=detail)
|
|
db.commit()
|
|
db.refresh(u)
|
|
return directory_entry(db, u, actor)
|
|
|
|
|
|
@app.post("/api/auth/users/{user_id}/project-role")
|
|
def set_user_project_role(user_id: str, body: ProjectRoleIn, actor: models.User = Depends(require_user_manager), db: Session = Depends(get_db)):
|
|
"""Set a user's job function on the project (Project Manager, Superintendent,
|
|
…). Purely descriptive — it grants nothing. This is what the SOP team pickers
|
|
and notification routing read, so it's worth keeping accurate."""
|
|
u = load_target_user(db, user_id)
|
|
require_see_user(db, actor, u)
|
|
require_manage_user(db, actor, u)
|
|
old = u.project_role or ""
|
|
u.project_role = (body.project_role or "").strip()[:120]
|
|
log_event(db, actor, "project_role_changed", "user", u.id, summary=u.username,
|
|
detail={"from": old, "to": u.project_role})
|
|
db.commit()
|
|
db.refresh(u)
|
|
return directory_entry(db, u, actor)
|
|
|
|
|
|
@app.post("/api/auth/users/{user_id}/auto-add")
|
|
def set_user_auto_add(user_id: str, body: AutoAddIn, admin: models.User = Depends(auth.require_admin), db: Session = Depends(get_db)):
|
|
"""Flag a user as a default member of every project created from here on, with
|
|
an optional role on those projects. It only touches NEW projects — existing
|
|
assignments stay under the admin's hand (see set_user_projects), because
|
|
back-filling everyone onto historical jobs is never what this flag means.
|
|
|
|
App-admin only, unlike the rest of user administration: this is a standing rule
|
|
about every project that will ever exist, including the ones a super user has no
|
|
part in."""
|
|
allowed = ("",) + auth.PROJECT_SCOPED_ROLES
|
|
role = (body.role or "").strip()
|
|
if role not in allowed:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"role must be '' (inherit) or one of {', '.join(auth.PROJECT_SCOPED_ROLES)}",
|
|
)
|
|
u = db.get(models.User, user_id)
|
|
if not u:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
u.auto_add_projects = bool(body.auto_add)
|
|
# A role left behind on a switched-off flag is a trap: it would quietly take
|
|
# effect the day someone switches the flag back on.
|
|
u.auto_add_role = role if u.auto_add_projects else ""
|
|
log_event(db, admin, "auto_add_changed", "user", u.id, summary=u.username,
|
|
detail={"auto_add": bool(u.auto_add_projects), "role": u.auto_add_role})
|
|
db.commit()
|
|
db.refresh(u)
|
|
return u.to_dict()
|
|
|
|
|
|
@app.delete("/api/auth/users/{user_id}")
|
|
def delete_user(user_id: str, actor: models.User = Depends(require_user_manager), db: Session = Depends(get_db)):
|
|
u = load_target_user(db, user_id)
|
|
require_see_user(db, actor, u)
|
|
require_manage_user(db, actor, u)
|
|
if u.id == actor.id:
|
|
raise HTTPException(status_code=400, detail="You cannot delete your own account")
|
|
log_event(db, actor, "user_deleted", "user", u.id, summary=u.username, detail={"role": u.role})
|
|
db.delete(u)
|
|
db.commit()
|
|
return {"deleted": user_id}
|
|
|
|
|
|
@app.get("/api/auth/users/{user_id}/projects")
|
|
def get_user_projects(user_id: str, actor: models.User = Depends(require_user_manager), db: Session = Depends(get_db)):
|
|
"""Which projects a user is assigned to, plus the project list to choose from.
|
|
(Admins implicitly access every project regardless of what's ticked here.)
|
|
|
|
A super user is shown ONLY the projects they administer, and `other_projects` says
|
|
how many more the person is on — enough for the dialog to be honest that it is
|
|
editing a slice of this account's access, without naming jobs that aren't theirs."""
|
|
u = load_target_user(db, user_id)
|
|
require_see_user(db, actor, u)
|
|
require_manage_user(db, actor, u)
|
|
rows = db.scalars(select(models.ProjectMember).where(models.ProjectMember.user_id == user_id)).all()
|
|
managed = managed_project_ids(db, actor)
|
|
if managed is None:
|
|
projects = db.scalars(select(models.Project).order_by(models.Project.name)).all()
|
|
in_scope = rows
|
|
else:
|
|
projects = db.scalars(
|
|
select(models.Project).where(models.Project.id.in_(managed)).order_by(models.Project.name)
|
|
).all() if managed else []
|
|
in_scope = [r for r in rows if r.project_id in managed]
|
|
return {
|
|
"user": directory_entry(db, u, actor),
|
|
"assigned": [r.project_id for r in in_scope],
|
|
# Per-project role overrides, keyed by project id ('' = inherit the account's).
|
|
"roles": {r.project_id: (r.role or "") for r in in_scope},
|
|
# Archived projects stay on this list on purpose — an existing assignment has
|
|
# to remain visible and removable — but they're flagged so the dialog can say
|
|
# so, rather than offering a finished job as though it were live work.
|
|
"projects": [{"id": p.id, "name": p.name, "number": p.number,
|
|
"archived": p.archived_at is not None} for p in projects],
|
|
"other_projects": len(rows) - len(in_scope),
|
|
"grantable_project_roles": list(
|
|
auth.PROJECT_SCOPED_ROLES if auth.is_admin(actor)
|
|
else (auth.ROLE_PROJECT_ADMIN, auth.ROLE_PROJECT_USER)
|
|
),
|
|
}
|
|
|
|
|
|
@app.put("/api/auth/users/{user_id}/projects")
|
|
def set_user_projects(user_id: str, body: ProjectAssignIn, actor: models.User = Depends(require_user_manager), db: Session = Depends(get_db)):
|
|
"""Replace a user's project assignments with the given set.
|
|
|
|
For an app admin the given set IS the whole answer. For a super user it replaces
|
|
only their own slice: memberships on projects they don't administer are left
|
|
exactly as they were, because a payload that simply omits them would otherwise cut
|
|
someone off from a job the caller can't even see."""
|
|
u = load_target_user(db, user_id)
|
|
require_see_user(db, actor, u)
|
|
require_manage_user(db, actor, u)
|
|
requested = [p for p in dict.fromkeys(body.project_ids) if p]
|
|
for pid in requested:
|
|
check_id(pid)
|
|
managed = managed_project_ids(db, actor)
|
|
if managed is not None:
|
|
outside = [p for p in requested if p not in managed]
|
|
if outside:
|
|
raise HTTPException(status_code=403, detail="You don't administer the users of one of those projects")
|
|
valid = set(db.scalars(select(models.Project.id).where(models.Project.id.in_(requested))).all()) if requested else set()
|
|
# A super user may hand out the project-scoped roles below their own; only an app
|
|
# admin can make someone a super user on a project. Anything unrecognised falls
|
|
# back to inheriting the account's own role.
|
|
allowed = auth.PROJECT_SCOPED_ROLES if auth.is_admin(actor) else (auth.ROLE_PROJECT_ADMIN, auth.ROLE_PROJECT_USER)
|
|
roles = {pid: r for pid, r in (body.roles or {}).items() if r in allowed}
|
|
# Rebuild only the rows this caller owns. Scoping the DELETE is the whole of the
|
|
# "leave other jobs alone" guarantee — get it wrong and a super user's save
|
|
# silently revokes access everywhere else.
|
|
doomed = delete(models.ProjectMember).where(models.ProjectMember.user_id == user_id)
|
|
if managed is not None:
|
|
# in_() on an empty set is a valid always-false predicate, so a caller who
|
|
# administers nothing deletes nothing (require_manage_user already refused them).
|
|
doomed = doomed.where(models.ProjectMember.project_id.in_(managed))
|
|
db.execute(doomed)
|
|
for pid in valid:
|
|
db.add(models.ProjectMember(id=gen_id("pm"), user_id=user_id, project_id=pid,
|
|
role=roles.get(pid, "")))
|
|
log_event(db, actor, "project_access_changed", "user", u.id, summary=u.username,
|
|
detail={"projects": len(valid),
|
|
"scope": "all" if managed is None else "managed",
|
|
"overrides": {p: r for p, r in roles.items() if p in valid}})
|
|
db.commit()
|
|
return {"assigned": sorted(valid), "roles": {p: roles.get(p, "") for p in sorted(valid)}}
|
|
|
|
|
|
# ── Projects ─────────────────────────────────────────────────────────────────
|
|
@app.post("/api/projects")
|
|
def upsert_project(body: ProjectIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
check_id(body.id)
|
|
proj = db.get(models.Project, body.id) if body.id else None
|
|
is_new = proj is None
|
|
if not is_new:
|
|
require_project_access(db, user, proj.id)
|
|
require_project_writable(db, user, proj.id, "Editing a project")
|
|
if proj is None:
|
|
proj = models.Project(id=body.id or gen_id("proj"))
|
|
db.add(proj)
|
|
proj.name = body.name
|
|
proj.number = body.number
|
|
proj.client = body.client
|
|
proj.division = body.division
|
|
proj.site = body.site
|
|
proj.sample = body.sample
|
|
proj.created_by = body.created_by or proj.created_by
|
|
proj.data = body.data
|
|
log_event(db, user, "created" if is_new else "updated", "project", proj.id,
|
|
project_id=proj.id, summary=(proj.name or proj.number or proj.id))
|
|
db.commit()
|
|
# A project created by a non-admin auto-grants its creator access. If that
|
|
# creator is ALSO a standing default member, this is the row that sticks —
|
|
# add_default_members below never overwrites an existing membership — so it has
|
|
# to carry the role they'd have been given, or someone whose flag says
|
|
# "Project Admin on every job" would silently land as a plain member on the one
|
|
# job they started themselves.
|
|
if is_new and not auth.is_admin(user):
|
|
creator_role = (user.auto_add_role or "").strip() if user.auto_add_projects else ""
|
|
grant_project_access(db, user.id, proj.id, creator_role)
|
|
db.commit()
|
|
# …and anyone flagged as a default member joins at creation time too.
|
|
if is_new:
|
|
add_default_members(db, proj.id, user)
|
|
db.commit()
|
|
db.refresh(proj)
|
|
return proj.to_dict()
|
|
|
|
|
|
@app.get("/api/projects")
|
|
def list_projects(
|
|
archived: str = Query("exclude", description="exclude (default) | only | all"),
|
|
user: models.User = Depends(auth.get_current_user),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
stmt = scope_to_access(select(models.Project), models.Project.id, db, user)
|
|
if archived == "only":
|
|
stmt = stmt.where(models.Project.archived_at.is_not(None))
|
|
elif archived != "all":
|
|
stmt = stmt.where(models.Project.archived_at.is_(None)) # default: hide archived
|
|
rows = db.scalars(stmt.order_by(models.Project.updated_at.desc())).all()
|
|
return [p.summary() for p in rows]
|
|
|
|
|
|
@app.get("/api/projects/{project_id}")
|
|
def get_project(project_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
proj = db.get(models.Project, project_id)
|
|
if not proj:
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
require_project_access(db, user, proj.id)
|
|
return proj.to_dict()
|
|
|
|
|
|
@app.delete("/api/projects/{project_id}")
|
|
def delete_project(project_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
proj = db.get(models.Project, project_id)
|
|
if not proj:
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
# Cascades to every SOP and work package on the project — Project Admin only.
|
|
require_project_admin(db, user, proj.id, "Deleting a project")
|
|
log_event(db, user, "deleted", "project", proj.id, project_id=proj.id,
|
|
summary=(proj.name or proj.number or proj.id))
|
|
db.delete(proj)
|
|
db.commit()
|
|
return {"deleted": project_id}
|
|
|
|
|
|
@app.post("/api/projects/{project_id}/archive")
|
|
def archive_project(project_id: str, body: ArchiveIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
"""Archive (or unarchive) a whole project — it drops out of every picker,
|
|
switcher and search, and freezes read-only, without losing a thing. This is how
|
|
a finished job gets out of everyone's way while staying on the record; delete is
|
|
still there for a job that should never have existed."""
|
|
proj = db.get(models.Project, project_id)
|
|
if not proj:
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
# Takes the whole job out of circulation for everyone on it, so it sits at the
|
|
# same bar as deleting it. Note the deliberate absence of require_project_writable
|
|
# — unarchiving is the one write an archived project must still accept.
|
|
require_project_admin(db, user, proj.id, "Archiving a project")
|
|
was_archived = proj.archived_at is not None
|
|
if body.archived and not was_archived:
|
|
proj.archived_at = models.utcnow()
|
|
log_event(db, user, "archived", "project", proj.id, project_id=proj.id,
|
|
summary=(proj.name or proj.number or proj.id))
|
|
elif not body.archived and was_archived:
|
|
proj.archived_at = None
|
|
log_event(db, user, "unarchived", "project", proj.id, project_id=proj.id,
|
|
summary=(proj.name or proj.number or proj.id))
|
|
db.commit()
|
|
db.refresh(proj)
|
|
return proj.to_dict()
|
|
|
|
|
|
# ── SOPs ─────────────────────────────────────────────────────────────────────
|
|
@app.post("/api/sops")
|
|
def upsert_sop(body: SopIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
check_id(body.id)
|
|
require_project_access(db, user, body.project_id)
|
|
# Both ends are checked: the project the SOP is being written INTO here, and the
|
|
# one it currently sits on below — so an archived job can be neither edited nor
|
|
# used as a source to move a SOP out of.
|
|
require_project_writable(db, user, body.project_id, "Saving a SOP")
|
|
sop = db.get(models.Sop, body.id) if body.id else None
|
|
if sop is not None:
|
|
require_project_access(db, user, sop.project_id)
|
|
require_project_writable(db, user, sop.project_id, "Saving a SOP")
|
|
# The SOP is the project's baseline: once it's been completed, changing it
|
|
# is a Project Admin action. Authoring and revising a draft is open to any
|
|
# project member, including marking it complete the first time.
|
|
if sop.complete:
|
|
require_project_admin(db, user, sop.project_id, "Changing a completed SOP")
|
|
is_new = sop is None
|
|
if sop is None:
|
|
sop = models.Sop(id=body.id or gen_id("sop"))
|
|
db.add(sop)
|
|
sop.project_id = body.project_id
|
|
sop.name = body.name
|
|
sop.number = body.number
|
|
sop.complete = body.complete
|
|
sop.created_by = body.created_by or sop.created_by
|
|
sop.data = body.data
|
|
log_event(db, user, "completed" if body.complete else ("created" if is_new else "updated"),
|
|
"sop", sop.id, project_id=sop.project_id, summary=(sop.name or sop.number or sop.id),
|
|
detail={"complete": bool(body.complete)})
|
|
db.commit()
|
|
db.refresh(sop)
|
|
return sop.to_dict()
|
|
|
|
|
|
@app.get("/api/sops")
|
|
def list_sops(project_id: Optional[str] = Query(None), full: bool = Query(False), user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
stmt = select(models.Sop)
|
|
if project_id:
|
|
stmt = stmt.where(models.Sop.project_id == project_id)
|
|
stmt = scope_to_access(stmt, models.Sop.project_id, db, user)
|
|
rows = db.scalars(stmt.order_by(models.Sop.updated_at.desc())).all()
|
|
# full=true includes the data JSON (the whole SOP document) for hydration;
|
|
# the default summary view stays lean for listing.
|
|
return [(s.to_dict() if full else s.summary()) for s in rows]
|
|
|
|
|
|
@app.get("/api/sops/latest")
|
|
def latest_sop(complete: Optional[bool] = None, project_id: Optional[str] = Query(None), user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
stmt = select(models.Sop)
|
|
if complete is not None:
|
|
stmt = stmt.where(models.Sop.complete == complete)
|
|
if project_id:
|
|
stmt = stmt.where(models.Sop.project_id == project_id)
|
|
stmt = scope_to_access(stmt, models.Sop.project_id, db, user)
|
|
sop = db.scalars(stmt.order_by(models.Sop.updated_at.desc()).limit(1)).first()
|
|
if not sop:
|
|
raise HTTPException(status_code=404, detail="No SOP found")
|
|
return sop.to_dict()
|
|
|
|
|
|
@app.get("/api/sops/{sop_id}")
|
|
def get_sop(sop_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
sop = db.get(models.Sop, sop_id)
|
|
if not sop:
|
|
raise HTTPException(status_code=404, detail="SOP not found")
|
|
require_project_access(db, user, sop.project_id)
|
|
return sop.to_dict()
|
|
|
|
|
|
@app.delete("/api/sops/{sop_id}")
|
|
def delete_sop(sop_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
sop = db.get(models.Sop, sop_id)
|
|
if not sop:
|
|
raise HTTPException(status_code=404, detail="SOP not found")
|
|
require_project_admin(db, user, sop.project_id, "Deleting a SOP")
|
|
require_project_writable(db, user, sop.project_id, "Deleting a SOP")
|
|
log_event(db, user, "deleted", "sop", sop.id, project_id=sop.project_id,
|
|
summary=(sop.name or sop.number or sop.id))
|
|
db.delete(sop)
|
|
db.commit()
|
|
return {"deleted": sop_id}
|
|
|
|
|
|
# ── Release gates (constraints + predecessors) ─────────────────────────────────
|
|
# Status ladder. The front end's STATUS_ORDER (wp-creation-app.js) has no "Issue"
|
|
# in it at all - the hold is a BRANCH off the released states, not a rung. It sits
|
|
# in this list so unknown statuses can still be told apart from known ones, but
|
|
# _released() must never count it: counting it is what let every transition out of
|
|
# hold skip the release gates as "already released", which made /status a side
|
|
# door past an open constraint (CR-015 / T7.3).
|
|
STATUS_ORDER = ["Draft", "Scheduled", "Issued", "In Progress", "Issue", "QC", "Closed"]
|
|
ISSUED_IDX = STATUS_ORDER.index("Issued")
|
|
DONE_STATUS = "Closed"
|
|
HOLD_STATUS = "Issue"
|
|
|
|
|
|
def _released(status: Optional[str]) -> bool:
|
|
"""Is this status a field state (Issued or beyond)? The hold is NOT one: a
|
|
package leaves hold through the gates, and enters it without them."""
|
|
if status == HOLD_STATUS:
|
|
return False
|
|
try:
|
|
return STATUS_ORDER.index(status or "") >= ISSUED_IDX
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def predecessor_ids(data: Optional[dict]) -> list[str]:
|
|
"""Work-package ids this package waits on. Ignores anything malformed rather
|
|
than failing a save — a bad id simply can't block."""
|
|
raw = (data or {}).get("predecessors") or []
|
|
if not isinstance(raw, list):
|
|
return []
|
|
return [p for p in raw if isinstance(p, str) and _ID_RE.match(p)]
|
|
|
|
|
|
def gate_override(data: Optional[dict]) -> Optional[dict]:
|
|
"""A deliberate, reasoned override of the predecessor gate. Planners legitimately
|
|
need to stage packages ahead of the work finishing, so the gate is refusable —
|
|
but only explicitly, and it lands in the audit log."""
|
|
ov = (data or {}).get("gateOverride")
|
|
if isinstance(ov, dict) and str(ov.get("reason") or "").strip():
|
|
return ov
|
|
return None
|
|
|
|
|
|
def _urgent_constraint_override(data: Optional[dict], open_names: list) -> bool:
|
|
"""D4: only an URGENT package may release past an open constraint, and only
|
|
through the audited override - the same gateOverride record the predecessor
|
|
gate uses, never a separate path. The override must NAME every constraint it
|
|
covers: a constraint opened after the reason was written cannot ride through
|
|
on it. Normal and High are unchanged - a hard refusal."""
|
|
if str((data or {}).get("priority") or "") != "Urgent":
|
|
return False
|
|
ov = gate_override(data)
|
|
if not ov:
|
|
return False
|
|
covered = ov.get("constraints")
|
|
if not isinstance(covered, list):
|
|
return False
|
|
return {str(n) for n in open_names} <= {str(c) for c in covered}
|
|
|
|
|
|
def blocking_predecessors(db: Session, wp_id: Optional[str], data: Optional[dict]) -> list[dict]:
|
|
"""Predecessors that are not Closed yet. A predecessor that no longer exists is
|
|
NOT blocking — a deleted package must not freeze everything downstream."""
|
|
ids = [p for p in predecessor_ids(data) if p != wp_id]
|
|
if not ids:
|
|
return []
|
|
rows = db.scalars(select(models.WorkPackage).where(models.WorkPackage.id.in_(ids))).all()
|
|
return [
|
|
{"id": r.id, "number": r.number, "subject": r.subject, "status": r.status}
|
|
for r in rows if r.status != DONE_STATUS
|
|
]
|
|
|
|
|
|
def check_predecessor_cycle(db: Session, wp_id: str, data: Optional[dict]) -> None:
|
|
"""Refuse a predecessor set that would make A wait on itself (directly or
|
|
through a chain). Walks the graph from the proposed predecessors; the visited
|
|
set also bounds the walk, so a cycle that already exists elsewhere in the data
|
|
can't spin here."""
|
|
start = [p for p in predecessor_ids(data)]
|
|
if wp_id in start:
|
|
raise HTTPException(status_code=400, detail="A work package cannot be its own predecessor.")
|
|
seen, stack = set(), list(start)
|
|
while stack:
|
|
cur = stack.pop()
|
|
if cur in seen:
|
|
continue
|
|
seen.add(cur)
|
|
row = db.get(models.WorkPackage, cur)
|
|
if row is None:
|
|
continue
|
|
nxt = predecessor_ids(row.data)
|
|
if wp_id in nxt:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"That would create a circular dependency ({row.number or row.id} already waits on this package).",
|
|
)
|
|
stack.extend(n for n in nxt if n not in seen)
|
|
|
|
|
|
def enforce_release_gates(db: Session, wp_id: Optional[str], data: Optional[dict],
|
|
new_status: str, old_status: Optional[str]) -> None:
|
|
"""Refuse a move to Issued (or beyond) while a release gate is unmet. Applied to
|
|
every path that can set a status — the plain upsert included, since that's how
|
|
the browser and the offline outbox save."""
|
|
if not _released(new_status) or _released(old_status):
|
|
return # not a release transition
|
|
constraints = (data or {}).get("constraints") or []
|
|
open_names = [c.get("name") for c in constraints if isinstance(c, dict) and c.get("status") == "open"]
|
|
if open_names and not _urgent_constraint_override(data, open_names):
|
|
raise HTTPException(status_code=409, detail={
|
|
"message": "Open constraints block release", "open": open_names,
|
|
})
|
|
blockers = blocking_predecessors(db, wp_id, data)
|
|
if blockers and not gate_override(data):
|
|
raise HTTPException(status_code=409, detail={
|
|
"message": "Predecessors are not closed yet",
|
|
"blocking": [f"{b['number'] or b['id']} ({b['status']})" for b in blockers],
|
|
})
|
|
|
|
|
|
# ── Critical-constraint notification ───────────────────────────────────────────
|
|
# A constraint flagged CRITICAL on the SOP, reopened after the package was
|
|
# released, is announced by email. We detect it by comparing the incoming
|
|
# constraints against the stored ones during the normal upsert rather than adding a
|
|
# separate endpoint: the browser saves through the sync outbox, which only replays
|
|
# POST /api/wps, so anything hung off another route would be lost offline.
|
|
def reopened_critical(old_data: Optional[dict], new_data: Optional[dict]) -> list[str]:
|
|
old = {}
|
|
for c in (old_data or {}).get("constraints") or []:
|
|
if isinstance(c, dict) and c.get("name"):
|
|
old[c["name"]] = c.get("status")
|
|
out = []
|
|
for c in (new_data or {}).get("constraints") or []:
|
|
if not isinstance(c, dict) or not c.get("critical"):
|
|
continue
|
|
name = c.get("name")
|
|
if not name or c.get("status") != "open":
|
|
continue
|
|
if old.get(name) not in (None, "open"): # was cleared/na, now open
|
|
out.append(name)
|
|
return out
|
|
|
|
|
|
def project_sop_team(db: Session, project_id: Optional[str]) -> list[str]:
|
|
"""PM + CM user ids from the project's most recent complete SOP."""
|
|
if not project_id:
|
|
return []
|
|
sop = db.scalars(
|
|
select(models.Sop)
|
|
.where((models.Sop.project_id == project_id) & (models.Sop.complete.is_(True)))
|
|
.order_by(models.Sop.updated_at.desc())
|
|
.limit(1)
|
|
).first()
|
|
if not sop:
|
|
return []
|
|
proj = (sop.data or {}).get("project") or {}
|
|
return [i for i in (proj.get("pmId"), proj.get("cmId")) if i]
|
|
|
|
|
|
def hold_body(user: "models.User", wp: "models.WorkPackage", names: list[str],
|
|
actor: "models.User", link: str) -> str:
|
|
# Constraint names and a WP number only — no package contents, same rule as the
|
|
# assignment mail.
|
|
who = user.full_name or user.username
|
|
by = actor.full_name or actor.username
|
|
which = ", ".join(names)
|
|
return (
|
|
f"Hi {who},\n\n"
|
|
f"A critical constraint was reopened on {wp.number or 'a work package'} "
|
|
f"after it was released to the field, so the package is on hold.\n\n"
|
|
f"Constraint: {which}\n"
|
|
f"Reopened by: {by}\n\n"
|
|
f"Open the package:\n{link}\n"
|
|
)
|
|
|
|
|
|
def notify_critical_reopen(db: Session, wp: "models.WorkPackage", names: list[str],
|
|
actor: "models.User") -> list["models.Notification"]:
|
|
"""Owner + PM + CM + everyone on the package's distribution list, minus whoever
|
|
did it (they already know) and minus duplicates."""
|
|
ids = []
|
|
if wp.assignee_id:
|
|
ids.append(wp.assignee_id)
|
|
ids.extend(project_sop_team(db, wp.project_id))
|
|
ids.extend([i for i in ((wp.data or {}).get("distributionIds") or []) if isinstance(i, str)])
|
|
seen, out = set(), []
|
|
link = wp_link(db, wp)
|
|
for uid in ids:
|
|
if uid in seen or uid == actor.id:
|
|
continue
|
|
seen.add(uid)
|
|
u = db.get(models.User, uid)
|
|
if not u or not u.is_active:
|
|
continue
|
|
out.append(notify.enqueue(
|
|
db, user=u, kind="wp_constraint_reopened",
|
|
subject=f"On hold: {wp.number or 'work package'} — critical constraint reopened",
|
|
body=hold_body(u, wp, names, actor, link),
|
|
link=link, wp_id=wp.id, project_id=wp.project_id,
|
|
))
|
|
return out
|
|
|
|
|
|
# ── Work Packages ────────────────────────────────────────────────────────────
|
|
@app.post("/api/wps")
|
|
def upsert_wp(body: WpIn, background_tasks: BackgroundTasks, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
check_id(body.id)
|
|
check_id(body.parent_id)
|
|
check_id(body.assignee_id)
|
|
require_project_access(db, user, body.project_id)
|
|
# Destination and origin both have to be live — see upsert_sop.
|
|
require_project_writable(db, user, body.project_id, "Saving a work package")
|
|
wp = db.get(models.WorkPackage, body.id) if body.id else None
|
|
if wp is not None:
|
|
require_project_access(db, user, wp.project_id)
|
|
require_project_writable(db, user, wp.project_id, "Saving a work package")
|
|
is_new = wp is None
|
|
old_status = None if is_new else wp.status
|
|
old_assignee = None if is_new else wp.assignee_id
|
|
old_data = None if is_new else (wp.data or {})
|
|
if wp is None:
|
|
wp = models.WorkPackage(id=body.id or gen_id("wp"))
|
|
db.add(wp)
|
|
# Predecessors: reject a cycle, and refuse a release while a gate is unmet.
|
|
# Both run before anything is written so a rejected save changes nothing.
|
|
wp_id_for_checks = body.id or wp.id
|
|
check_predecessor_cycle(db, wp_id_for_checks, body.data)
|
|
enforce_release_gates(db, wp_id_for_checks, body.data, body.status, old_status)
|
|
wp.project_id = body.project_id
|
|
wp.sop_id = body.sop_id
|
|
wp.parent_id = body.parent_id
|
|
wp.number = body.number
|
|
wp.subject = body.subject
|
|
wp.type = body.type
|
|
wp.status = body.status
|
|
new_assignee = body.assignee_id or None
|
|
if new_assignee:
|
|
require_assignable(db, new_assignee, body.project_id)
|
|
wp.assignee_id = new_assignee
|
|
wp.created_by = body.created_by or wp.created_by
|
|
wp.data = body.data
|
|
if is_new:
|
|
_act, _detail = "created", {"status": wp.status}
|
|
elif old_status != wp.status:
|
|
_act, _detail = "status_changed", {"from": old_status, "to": wp.status}
|
|
else:
|
|
_act, _detail = "updated", {"status": wp.status}
|
|
log_event(db, user, _act, "wp", wp.id, project_id=wp.project_id,
|
|
summary=(wp.number or wp.subject or wp.id), detail=_detail)
|
|
notifs = []
|
|
# A reasoned override of the predecessor gate is worth its own audit line —
|
|
# "released early, and here's why" is exactly what a reviewer looks for later.
|
|
ov = gate_override(body.data)
|
|
if ov and _released(wp.status) and not _released(old_status):
|
|
blockers = blocking_predecessors(db, wp.id, body.data)
|
|
_ov_constraints = [str(c) for c in ov.get("constraints") or [] if c]
|
|
log_event(db, user, "gate_overridden", "wp", wp.id, project_id=wp.project_id,
|
|
summary=(wp.number or wp.subject or wp.id),
|
|
detail={"reason": str(ov.get("reason"))[:300],
|
|
"blocking": [b["number"] or b["id"] for b in blockers],
|
|
"constraints": _ov_constraints})
|
|
# CR-015: holds and releases are history, not just state. Entering or leaving
|
|
# the hold branch gets its own audit line with the actor and the reason the
|
|
# browser recorded in data.holds - `status_changed` alone says from/to but
|
|
# not why, and the WHY is what a notice of delay is built from.
|
|
if not is_new and old_status != wp.status and "Issue" in (old_status, wp.status):
|
|
_hlds = [h for h in ((wp.data or {}).get("holds") or []) if isinstance(h, dict)]
|
|
if wp.status == "Issue":
|
|
_h = next((h for h in reversed(_hlds) if not h.get("released")), None)
|
|
log_event(db, user, "hold_logged", "wp", wp.id, project_id=wp.project_id,
|
|
summary=(wp.number or wp.subject or wp.id),
|
|
detail={"from": old_status,
|
|
"constraint": str((_h or {}).get("constraint") or "")[:200],
|
|
"reason": str((_h or {}).get("details") or "")[:300]})
|
|
else:
|
|
_h = next((h for h in reversed(_hlds) if h.get("released")), None)
|
|
log_event(db, user, "hold_released", "wp", wp.id, project_id=wp.project_id,
|
|
summary=(wp.number or wp.subject or wp.id),
|
|
detail={"to": wp.status,
|
|
"constraint": str((_h or {}).get("constraint") or "")[:200],
|
|
"reason": str((_h or {}).get("details") or "")[:300]})
|
|
# A critical constraint reopened after release: log it and tell the people who
|
|
# need to know (owner, PM, CM, distribution).
|
|
if not is_new and (_released(old_status) or old_status == HOLD_STATUS):
|
|
reopened = reopened_critical(old_data, wp.data)
|
|
if reopened:
|
|
log_event(db, user, "constraint_reopened", "wp", wp.id, project_id=wp.project_id,
|
|
summary=(wp.number or wp.subject or wp.id),
|
|
detail={"critical": reopened, "status": wp.status})
|
|
notifs.extend(notify_critical_reopen(db, wp, reopened, user))
|
|
# Notify a newly-assigned owner (skip self-assignment).
|
|
notif = None
|
|
if new_assignee and new_assignee != old_assignee and new_assignee != user.id:
|
|
assignee = db.get(models.User, new_assignee)
|
|
if assignee:
|
|
link = wp_link(db, wp)
|
|
notif = notify.enqueue(
|
|
db, user=assignee, kind="wp_assigned",
|
|
subject=f"You were assigned {wp.number or 'a work package'}",
|
|
body=assign_body(assignee, wp, user, link),
|
|
link=link, wp_id=wp.id, project_id=wp.project_id,
|
|
)
|
|
log_event(db, user, "assigned", "wp", wp.id, project_id=wp.project_id,
|
|
summary=(wp.number or wp.subject or wp.id), detail={"to": assignee.username})
|
|
db.commit()
|
|
db.refresh(wp)
|
|
if notif is not None:
|
|
notifs.append(notif)
|
|
for n in notifs:
|
|
background_tasks.add_task(notify.deliver, n.id)
|
|
return wp.to_dict()
|
|
|
|
|
|
@app.get("/api/wps")
|
|
def list_wps(
|
|
response: Response,
|
|
project_id: Optional[str] = Query(None),
|
|
sop_id: Optional[str] = Query(None),
|
|
parent_id: Optional[str] = Query(None),
|
|
status: Optional[str] = Query(None),
|
|
q: Optional[str] = Query(None, description="search number / subject / type"),
|
|
archived: str = Query("exclude", description="exclude (default) | only | all"),
|
|
limit: Optional[int] = Query(None, ge=1, le=1000),
|
|
offset: int = Query(0, ge=0),
|
|
full: bool = Query(False),
|
|
user: models.User = Depends(auth.get_current_user),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
stmt = select(models.WorkPackage)
|
|
if project_id:
|
|
stmt = stmt.where(models.WorkPackage.project_id == project_id)
|
|
if sop_id:
|
|
stmt = stmt.where(models.WorkPackage.sop_id == sop_id)
|
|
if parent_id:
|
|
stmt = stmt.where(models.WorkPackage.parent_id == parent_id)
|
|
if status:
|
|
stmt = stmt.where(models.WorkPackage.status == status)
|
|
if archived == "only":
|
|
stmt = stmt.where(models.WorkPackage.archived_at.is_not(None))
|
|
elif archived != "all":
|
|
stmt = stmt.where(models.WorkPackage.archived_at.is_(None)) # default: hide archived
|
|
if q and q.strip():
|
|
like = f"%{q.strip()}%"
|
|
stmt = stmt.where(
|
|
models.WorkPackage.number.ilike(like)
|
|
| models.WorkPackage.subject.ilike(like)
|
|
| models.WorkPackage.type.ilike(like)
|
|
)
|
|
stmt = scope_to_access(stmt, models.WorkPackage.project_id, db, user)
|
|
# Report the pre-pagination total so the client can build a pager.
|
|
total = db.scalar(select(func.count()).select_from(stmt.subquery()))
|
|
response.headers["X-Total-Count"] = str(total or 0)
|
|
stmt = stmt.order_by(models.WorkPackage.updated_at.desc()).offset(offset)
|
|
if limit is not None:
|
|
stmt = stmt.limit(limit)
|
|
rows = db.scalars(stmt).all()
|
|
# full=true includes the data JSON (full package document) so the creator can
|
|
# rehydrate everything in one request; default stays lean for listing.
|
|
return [(w.to_dict() if full else w.summary()) for w in rows]
|
|
|
|
|
|
# Weighted completion by status, 0..1, so progress reads as a slope rather than
|
|
# done/not-done. Mirrors PROGRESS_W in wp-creation-app.js — the dashboard used to
|
|
# compute this in the browser from localStorage, which is exactly what B4 removes.
|
|
PROGRESS_WEIGHT = {
|
|
"Draft": 0.0, "Scheduled": 0.25, "Issue": 0.4, "Issued": 0.5,
|
|
"In Progress": 0.75, "QC": 0.9, "Closed": 1.0,
|
|
}
|
|
|
|
# The dimensions a location rollup is grouped by. Today a work package carries one
|
|
# free-text `location` ("building / level / sector / room"), so that is the single
|
|
# dimension. CR-004 replaces it with structured building / floor / sector in wave 6;
|
|
# when it does, only this tuple and the keys inside each group change — the response
|
|
# shape does not, which is what T4.1 means by "can be grouped by location without a
|
|
# schema change". CR-018's rollup consumes `by_location.groups` either way.
|
|
# CR-004 landed at T6.3, so this is now what it was designed to become. T4.1's
|
|
# note said "only this tuple and the keys inside each group change — the response
|
|
# shape does not", and that held: CR-018's rollup consumes `by_location.groups`
|
|
# exactly as it did when there was one free-text dimension.
|
|
LOCATION_DIMENSIONS = ("building", "floor", "sector")
|
|
|
|
# A package with no location is not dropped from the rollup. `CR-018`'s own
|
|
# done-when says so, and the reason is arithmetic: a group set that silently
|
|
# omits the unlocated packages does not add up to the project total, and a rollup
|
|
# that does not reconcile is worse than no rollup.
|
|
LOCATION_UNSET = "(unassigned)"
|
|
|
|
|
|
def _location_key(data: dict) -> dict:
|
|
"""The location dimensions of one package, as a dict keyed by dimension name.
|
|
|
|
Structured fields win; the free-text `location` a package captured before
|
|
CR-004 is kept as its own dimension value so those packages group together
|
|
under what they actually said rather than all collapsing into one bucket."""
|
|
structured = {d: (data.get(d) or "").strip() for d in LOCATION_DIMENSIONS}
|
|
if any(structured.values()):
|
|
return {k: v or LOCATION_UNSET for k, v in structured.items()}
|
|
legacy = (data.get("location") or "").strip()
|
|
if legacy:
|
|
# One dimension deep, deliberately: free text is not a hierarchy and
|
|
# pretending it is would put "FAB / LVL 1" under a building called
|
|
# "FAB / LVL 1".
|
|
return {"building": legacy, "floor": LOCATION_UNSET, "sector": LOCATION_UNSET}
|
|
return {d: LOCATION_UNSET for d in LOCATION_DIMENSIONS}
|
|
|
|
|
|
def _location_levels(loc_groups: dict) -> dict:
|
|
"""Totals at each level of the hierarchy, not only at the leaf.
|
|
|
|
`by_location.groups` is one row per distinct (building, floor, sector). The
|
|
question CR-018 asks — "what is on floor 2" — is a level above that, and
|
|
every level has to reconcile against the project total or the rollup is
|
|
decoration. Each level therefore sums EVERY package, including the ones whose
|
|
value at that level is unassigned."""
|
|
out: dict[str, list] = {}
|
|
for dim in LOCATION_DIMENSIONS:
|
|
buckets: dict[str, dict] = {}
|
|
for g in loc_groups.values():
|
|
# Grouped by the value AT THIS LEVEL alone, not by the tuple of levels
|
|
# above it. Each stored value is already a full path — a floor is
|
|
# `B-ONE/L1`, not `L1` — so it carries its own ancestry and is unique
|
|
# across buildings without being re-qualified. Only the unassigned
|
|
# bucket is shared, which is what it should be: "these have no floor
|
|
# recorded" is one answer, not one answer per building.
|
|
path = g["key"].get(dim, LOCATION_UNSET) or LOCATION_UNSET
|
|
slot = buckets.setdefault(path, {
|
|
"dimension": dim,
|
|
"path": path,
|
|
"total": 0, "release_ready": 0, "on_hold": 0, "overdue": 0,
|
|
"est_hours": 0.0, "actual_hours": 0.0,
|
|
})
|
|
for f in ("total", "release_ready", "on_hold", "overdue", "est_hours", "actual_hours"):
|
|
slot[f] += g[f]
|
|
out[dim] = [
|
|
{**b, "est_hours": round(b["est_hours"]), "actual_hours": round(b["actual_hours"])}
|
|
for b in sorted(buckets.values(), key=lambda b: b["path"])
|
|
]
|
|
return out
|
|
|
|
|
|
@app.get("/api/wps/metrics")
|
|
def wp_metrics(project_id: Optional[str] = Query(None), sop_id: Optional[str] = Query(None), user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
"""Every count and rollup the creator's dashboard shows, computed from the
|
|
database rather than from the caller's own browser (B4).
|
|
|
|
Masters (data.split == true) are excluded so a split package's hours are not
|
|
double-counted with its instances. Archived packages are excluded throughout.
|
|
|
|
Two people on the same project get the same numbers from this endpoint. They
|
|
did not when each browser derived them from its own localStorage, and neither
|
|
was told — which is the failure B4 exists to remove."""
|
|
stmt = select(models.WorkPackage).where(models.WorkPackage.archived_at.is_(None))
|
|
if project_id:
|
|
stmt = stmt.where(models.WorkPackage.project_id == project_id)
|
|
if sop_id:
|
|
stmt = stmt.where(models.WorkPackage.sop_id == sop_id)
|
|
stmt = scope_to_access(stmt, models.WorkPackage.project_id, db, user)
|
|
rows = db.scalars(stmt).all()
|
|
|
|
today = models.utcnow().date().isoformat()
|
|
|
|
by_status: dict[str, int] = {}
|
|
by_discipline: dict[str, int] = {}
|
|
loc_groups: dict[tuple, dict] = {}
|
|
prog_by_disc: dict[str, dict] = {}
|
|
gating: list[dict] = []
|
|
total = ready = on_hold = overdue = mine = 0
|
|
est_hours = actual_hours = 0.0
|
|
progress_sum = 0.0
|
|
|
|
for w in rows:
|
|
data = w.data or {}
|
|
if data.get("split"):
|
|
continue
|
|
total += 1
|
|
by_status[w.status] = by_status.get(w.status, 0) + 1
|
|
if w.status == "Issue":
|
|
on_hold += 1
|
|
if w.assignee_id and w.assignee_id == user.id:
|
|
mine += 1
|
|
|
|
due = (data.get("due") or "").strip()
|
|
is_overdue = bool(due and w.status != "Closed" and due < today)
|
|
if is_overdue:
|
|
overdue += 1
|
|
|
|
constraints = data.get("constraints") or []
|
|
open_constraints = [c for c in constraints if c.get("status") == "open"]
|
|
# `waitingOn` are predecessor packages not yet closed; the browser counted
|
|
# them as blocking too, so the server has to, or "release-ready" changes
|
|
# meaning the moment the dashboard stops computing it locally.
|
|
waiting_on = [x for x in (data.get("waitingOn") or []) if x]
|
|
blocked = bool(open_constraints or waiting_on)
|
|
if not blocked and w.status not in ("Closed", "Issue"):
|
|
ready += 1
|
|
if open_constraints:
|
|
gating.append({
|
|
"id": w.id, "number": w.number, "subject": w.subject,
|
|
"blocked_by": [
|
|
{"name": c.get("name") or "", "comment": c.get("comment") or ""}
|
|
for c in open_constraints
|
|
],
|
|
})
|
|
|
|
try:
|
|
est_hours += float(data.get("hours") or 0)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
try:
|
|
actual_hours += float(data.get("actualHrs") or 0)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
|
|
weight = PROGRESS_WEIGHT.get(w.status, 0.0)
|
|
progress_sum += weight
|
|
|
|
disciplines = data.get("disciplines") or ["(none)"]
|
|
for d in disciplines:
|
|
by_discipline[d] = by_discipline.get(d, 0) + 1
|
|
slot = prog_by_disc.setdefault(d, {"total": 0, "done": 0, "weight": 0.0})
|
|
slot["total"] += 1
|
|
slot["weight"] += weight
|
|
if w.status == "Closed":
|
|
slot["done"] += 1
|
|
|
|
key = _location_key(data)
|
|
kt = tuple(key.get(d, "(unset)") for d in LOCATION_DIMENSIONS) if len(key) == len(LOCATION_DIMENSIONS) else tuple(sorted(key.items()))
|
|
slot = loc_groups.setdefault(kt, {"key": key, "total": 0, "release_ready": 0,
|
|
"on_hold": 0, "overdue": 0, "by_status": {},
|
|
"est_hours": 0.0, "actual_hours": 0.0})
|
|
slot["total"] += 1
|
|
slot["by_status"][w.status] = slot["by_status"].get(w.status, 0) + 1
|
|
if not blocked and w.status not in ("Closed", "Issue"):
|
|
slot["release_ready"] += 1
|
|
if w.status == "Issue":
|
|
slot["on_hold"] += 1
|
|
if is_overdue:
|
|
slot["overdue"] += 1
|
|
# CR-018: hours roll up along the same dimensions. Actual Hours is the one
|
|
# CR-017 retained and is the reason that decision mattered — it is what
|
|
# makes a floor's real cost visible.
|
|
try:
|
|
slot["est_hours"] += float(data.get("hours") or 0)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
try:
|
|
slot["actual_hours"] += float(data.get("actualHrs") or 0)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
|
|
dimensions = list(LOCATION_DIMENSIONS)
|
|
if loc_groups:
|
|
first = next(iter(loc_groups.values()))["key"]
|
|
dimensions = list(first.keys())
|
|
|
|
return {
|
|
"total": total, "mine": mine, "release_ready": ready, "on_hold": on_hold,
|
|
"overdue": overdue,
|
|
"est_hours": round(est_hours), "actual_hours": round(actual_hours),
|
|
"by_status": by_status, "by_discipline": by_discipline,
|
|
"progress": {
|
|
"overall_pct": round(progress_sum / total * 100) if total else 0,
|
|
"by_discipline": [
|
|
{"name": d, "pct": round(v["weight"] / v["total"] * 100) if v["total"] else 0,
|
|
"done": v["done"], "total": v["total"]}
|
|
for d, v in sorted(prog_by_disc.items())
|
|
],
|
|
},
|
|
"gating": sorted(gating, key=lambda g: (g["number"] or "", g["id"])),
|
|
"by_location": {
|
|
"dimensions": dimensions,
|
|
"unassigned_key": LOCATION_UNSET,
|
|
"groups": [
|
|
{**g, "est_hours": round(g["est_hours"]), "actual_hours": round(g["actual_hours"])}
|
|
for g in sorted(loc_groups.values(),
|
|
key=lambda g: tuple(str(v) for v in g["key"].values()))
|
|
],
|
|
# Rolled up one level at a time as well as by the full triple, because
|
|
# "how many on floor 2" is the question CR-018 is actually about and
|
|
# summing the leaf groups in the browser would be the same per-browser
|
|
# arithmetic B4 removed.
|
|
"levels": _location_levels(loc_groups),
|
|
},
|
|
"generated_at": models.utcnow().isoformat(),
|
|
}
|
|
|
|
|
|
# ── Location taxonomy — CR-005 ────────────────────────────────────────────────
|
|
# Building / Floor / Sector, configured once per project rather than hard-coded.
|
|
# See models.LocationNode for why this stores codes as well as names, and why
|
|
# nothing here deletes.
|
|
|
|
LOCATION_LEVELS = ("building", "floor", "sector")
|
|
_SLUG_STRIP = re.compile(r"[^A-Z0-9]+")
|
|
|
|
|
|
def location_slug(name: str) -> str:
|
|
"""A stable code from a display name: upper-cased, non-alphanumerics collapsed
|
|
to a single dash, trimmed. `Level 2` -> `LEVEL-2`, `1P (chase)` -> `1P-CHASE`.
|
|
|
|
Derived ONCE, at import, and never recomputed — see LocationNode. Returns ''
|
|
when there is nothing to make a code from, which the caller turns into a
|
|
rejected row with a reason rather than a silently-skipped one."""
|
|
return _SLUG_STRIP.sub("-", (name or "").strip().upper()).strip("-")[:60]
|
|
|
|
|
|
def parse_location_rows(text: str) -> tuple[list[tuple[int, list[str]]], list[dict]]:
|
|
"""Split pasted or uploaded text into (rows, rejected).
|
|
|
|
Rows come back paired with their SOURCE line number, not their index among the
|
|
accepted ones. "Duplicate on row 12" has to mean row 12 of the file somebody is
|
|
looking at, or the report sends them to the wrong line.
|
|
|
|
Accepts comma, tab or semicolon separators — a paste out of Excel is
|
|
tab-separated and a saved CSV is not, and asking which one somebody has is a
|
|
question the machine can answer. A header line naming the levels is skipped.
|
|
|
|
Every rejection carries the line number and a reason. A silent skip is the
|
|
failure mode this endpoint exists to avoid: an import that says "42 rows" over
|
|
a file with 50 in it has lost eight and told nobody."""
|
|
rows: list[tuple[int, list[str]]] = []
|
|
rejected: list[dict] = []
|
|
for i, raw in enumerate((text or "").splitlines(), start=1):
|
|
line = raw.strip()
|
|
if not line:
|
|
continue
|
|
if "\t" in raw:
|
|
parts = [c.strip() for c in raw.split("\t")]
|
|
elif ";" in line and "," not in line:
|
|
parts = [c.strip() for c in line.split(";")]
|
|
else:
|
|
parts = [c.strip() for c in line.split(",")]
|
|
parts = [p.strip().strip('"').strip() for p in parts]
|
|
while parts and not parts[-1]:
|
|
parts.pop()
|
|
if not parts:
|
|
continue
|
|
low = [p.lower() for p in parts]
|
|
if i == 1 and low[:1] in (["building"], ["bldg"]):
|
|
continue # header row
|
|
if len(parts) > len(LOCATION_LEVELS):
|
|
rejected.append({"line": i, "text": line,
|
|
"reason": "more than 3 columns — expected building, floor, sector"})
|
|
continue
|
|
if not parts[0]:
|
|
rejected.append({"line": i, "text": line,
|
|
"reason": "no building — a floor or sector needs one above it"})
|
|
continue
|
|
# A gap in the middle ("B100,,1P") would attach a sector to nothing.
|
|
gap = next((n for n, p in enumerate(parts) if not p), None)
|
|
if gap is not None and any(parts[gap + 1:]):
|
|
rejected.append({"line": i, "text": line,
|
|
"reason": "a %s is named with no %s above it"
|
|
% (LOCATION_LEVELS[len(parts) - 1], LOCATION_LEVELS[gap])})
|
|
continue
|
|
parts = [p for p in parts if p]
|
|
if any(not location_slug(p) for p in parts):
|
|
rejected.append({"line": i, "text": line,
|
|
"reason": "no letters or digits to make a code from"})
|
|
continue
|
|
rows.append((i, parts))
|
|
return rows, rejected
|
|
|
|
|
|
def _location_tree(db: Session, project_id: str, include_inactive: bool):
|
|
stmt = select(models.LocationNode).where(models.LocationNode.project_id == project_id)
|
|
if not include_inactive:
|
|
stmt = stmt.where(models.LocationNode.active.is_(True))
|
|
nodes = db.scalars(stmt.order_by(models.LocationNode.path)).all()
|
|
return nodes
|
|
|
|
|
|
@app.get("/api/projects/{project_id}/locations")
|
|
def list_locations(project_id: str,
|
|
include_inactive: bool = Query(False),
|
|
user: models.User = Depends(auth.get_current_user),
|
|
db: Session = Depends(get_db)):
|
|
"""The project's taxonomy, flat, ordered by path so a caller can rebuild the
|
|
tree without a second query.
|
|
|
|
`include_inactive` defaults to FALSE, which is what makes a deactivated value
|
|
disappear from new work packages. It is available as true because a package
|
|
that already references a deactivated value still has to render its label —
|
|
deactivating hides a choice, it does not rewrite history."""
|
|
if not db.get(models.Project, project_id):
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
require_project_access(db, user, project_id)
|
|
nodes = _location_tree(db, project_id, include_inactive)
|
|
return {
|
|
"project_id": project_id,
|
|
"levels": list(LOCATION_LEVELS),
|
|
"nodes": [n.to_dict() for n in nodes],
|
|
"counts": {lvl: sum(1 for n in nodes if n.level == lvl) for lvl in LOCATION_LEVELS},
|
|
"generated_at": models.utcnow().isoformat(),
|
|
}
|
|
|
|
|
|
class LocationImportIn(BaseModel):
|
|
text: str = ""
|
|
dry_run: bool = False
|
|
|
|
|
|
@app.post("/api/projects/{project_id}/locations/import")
|
|
def import_locations(project_id: str, body: LocationImportIn,
|
|
user: models.User = Depends(auth.get_current_user),
|
|
db: Session = Depends(get_db)):
|
|
"""Bulk import from CSV or a paste. Same code path for both: a file is read in
|
|
the browser and posted as text, because two parsers would be two sets of rules
|
|
about what a blank column means.
|
|
|
|
Reports rather than merges. A row naming a building/floor/sector combination
|
|
that already exists — in this file or in the project — comes back in
|
|
`duplicates` with its line number. Reusing a PARENT is not a duplicate:
|
|
`B100,L2,1P` and `B100,L2,2P` share a building and a floor by design, and it
|
|
is only the full path repeating that is a duplicate.
|
|
|
|
`dry_run` parses and reports without writing, which is what lets the wizard
|
|
show what an import will do before it does it."""
|
|
if not db.get(models.Project, project_id):
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
# Any project member, matching how the SOP baseline itself is authored — the
|
|
# Project Admin gate is on CHANGING a completed SOP, not on writing one.
|
|
require_project_access(db, user, project_id)
|
|
require_project_writable(db, user, project_id, "The location list cannot be changed")
|
|
|
|
rows, rejected = parse_location_rows(body.text)
|
|
|
|
existing = {n.path: n for n in db.scalars(
|
|
select(models.LocationNode).where(models.LocationNode.project_id == project_id)
|
|
).all()}
|
|
before = set(existing)
|
|
created: list[dict] = []
|
|
duplicates: list[dict] = []
|
|
reactivated: list[dict] = []
|
|
seen_in_file: dict[str, int] = {}
|
|
next_sort = max((n.sort for n in existing.values()), default=0)
|
|
|
|
for line_no, parts in rows:
|
|
segs = [location_slug(p) for p in parts]
|
|
full = "/".join(segs)
|
|
if full in seen_in_file:
|
|
duplicates.append({"line": line_no, "path": full, "names": parts,
|
|
"reason": "already on line %d of this import" % seen_in_file[full]})
|
|
continue
|
|
seen_in_file[full] = line_no
|
|
if full in before:
|
|
node = existing[full]
|
|
if not node.active:
|
|
# Re-importing a value somebody deactivated is a request to bring it
|
|
# back, not a duplicate — and it must reuse the SAME row, or every
|
|
# work package pointing at the old path is orphaned.
|
|
if not body.dry_run:
|
|
node.active = True
|
|
reactivated.append({"path": full, "names": parts})
|
|
else:
|
|
duplicates.append({"line": line_no, "path": full, "names": parts,
|
|
"reason": "already in this project"})
|
|
continue
|
|
# Create any missing ancestors, then the leaf. Sharing a parent is the
|
|
# normal case, not a collision.
|
|
parent_id = None
|
|
for depth, seg in enumerate(segs):
|
|
sub = "/".join(segs[:depth + 1])
|
|
node = existing.get(sub)
|
|
if node is None:
|
|
next_sort += 1
|
|
node = models.LocationNode(
|
|
id=gen_id("loc"), project_id=project_id, parent_id=parent_id,
|
|
level=LOCATION_LEVELS[depth], code=seg, path=sub,
|
|
name=parts[depth], active=True, sort=next_sort,
|
|
created_by=user.username,
|
|
)
|
|
existing[sub] = node
|
|
if not body.dry_run:
|
|
db.add(node)
|
|
if sub not in before:
|
|
created.append({"path": sub, "level": LOCATION_LEVELS[depth],
|
|
"code": seg, "name": parts[depth]})
|
|
parent_id = node.id
|
|
|
|
result = {
|
|
"project_id": project_id, "dry_run": bool(body.dry_run),
|
|
"read": len(rows) + len(rejected),
|
|
"created": created, "duplicates": duplicates,
|
|
"reactivated": reactivated, "rejected": rejected,
|
|
}
|
|
if body.dry_run:
|
|
db.rollback()
|
|
return result
|
|
log_event(db, user, "locations_imported", "project", project_id, project_id,
|
|
summary="%d location value(s) added" % len(created),
|
|
detail={"created": len(created), "duplicates": len(duplicates),
|
|
"rejected": len(rejected), "reactivated": len(reactivated)})
|
|
db.commit()
|
|
return result
|
|
|
|
|
|
class LocationIn(BaseModel):
|
|
level: str = "building"
|
|
parent_id: Optional[str] = None
|
|
name: str = ""
|
|
|
|
|
|
@app.post("/api/projects/{project_id}/locations")
|
|
def add_location(project_id: str, body: LocationIn,
|
|
user: models.User = Depends(auth.get_current_user),
|
|
db: Session = Depends(get_db)):
|
|
"""Add one value by hand. Same rules as the import — an import you cannot
|
|
correct afterwards is an import nobody trusts enough to run."""
|
|
if not db.get(models.Project, project_id):
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
# Any project member, matching how the SOP baseline itself is authored — the
|
|
# Project Admin gate is on CHANGING a completed SOP, not on writing one.
|
|
require_project_access(db, user, project_id)
|
|
require_project_writable(db, user, project_id, "The location list cannot be changed")
|
|
|
|
name = (body.name or "").strip()
|
|
code = location_slug(name)
|
|
if not code:
|
|
raise HTTPException(status_code=400,
|
|
detail="That name has no letters or digits to make a code from")
|
|
level = (body.level or "").strip().lower()
|
|
if level not in LOCATION_LEVELS:
|
|
raise HTTPException(status_code=400,
|
|
detail="Level must be one of: " + ", ".join(LOCATION_LEVELS))
|
|
depth = LOCATION_LEVELS.index(level)
|
|
parent = None
|
|
if body.parent_id:
|
|
parent = db.get(models.LocationNode, body.parent_id)
|
|
if not parent or parent.project_id != project_id:
|
|
raise HTTPException(status_code=400, detail="Parent is not on this project")
|
|
if depth == 0 and parent is not None:
|
|
raise HTTPException(status_code=400, detail="A building has nothing above it")
|
|
if depth > 0 and parent is None:
|
|
raise HTTPException(status_code=400,
|
|
detail="A %s needs a %s above it" % (level, LOCATION_LEVELS[depth - 1]))
|
|
if parent is not None and LOCATION_LEVELS.index(parent.level) != depth - 1:
|
|
raise HTTPException(status_code=400,
|
|
detail="A %s cannot sit under a %s" % (level, parent.level))
|
|
|
|
path = (parent.path + "/" + code) if parent else code
|
|
clash = db.scalars(select(models.LocationNode).where(
|
|
(models.LocationNode.project_id == project_id) & (models.LocationNode.path == path)
|
|
)).first()
|
|
if clash:
|
|
if not clash.active:
|
|
clash.active = True
|
|
log_event(db, user, "location_reactivated", "project", project_id, project_id,
|
|
summary=path)
|
|
db.commit()
|
|
return clash.to_dict()
|
|
raise HTTPException(status_code=409, detail="“%s” is already on this project" % name)
|
|
|
|
top = db.scalar(select(func.max(models.LocationNode.sort)).where(
|
|
models.LocationNode.project_id == project_id)) or 0
|
|
node = models.LocationNode(
|
|
id=gen_id("loc"), project_id=project_id, parent_id=(parent.id if parent else None),
|
|
level=level, code=code, path=path, name=name, active=True, sort=top + 1,
|
|
created_by=user.username,
|
|
)
|
|
db.add(node)
|
|
log_event(db, user, "location_added", "project", project_id, project_id, summary=path)
|
|
db.commit()
|
|
return node.to_dict()
|
|
|
|
|
|
class LocationPatch(BaseModel):
|
|
name: Optional[str] = None
|
|
active: Optional[bool] = None
|
|
|
|
|
|
@app.patch("/api/projects/{project_id}/locations/{node_id}")
|
|
def update_location(project_id: str, node_id: str, body: LocationPatch,
|
|
user: models.User = Depends(auth.get_current_user),
|
|
db: Session = Depends(get_db)):
|
|
"""Rename or deactivate. There is no DELETE, and that is the design:
|
|
|
|
rename changes `name` only. `code` and `path` are untouched, so every
|
|
work package pointing at this value keeps pointing at it. That
|
|
is the whole reason the two are separate columns.
|
|
deactivate hides the value from new work packages. Cascades DOWN — a floor
|
|
nobody can pick makes its sectors unpickable too, and leaving
|
|
them offered would be offering a path to nowhere. Reactivating
|
|
a child reactivates its ancestors for the same reason.
|
|
"""
|
|
if not db.get(models.Project, project_id):
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
# Any project member, matching how the SOP baseline itself is authored — the
|
|
# Project Admin gate is on CHANGING a completed SOP, not on writing one.
|
|
require_project_access(db, user, project_id)
|
|
require_project_writable(db, user, project_id, "The location list cannot be changed")
|
|
node = db.get(models.LocationNode, node_id)
|
|
if not node or node.project_id != project_id:
|
|
raise HTTPException(status_code=404, detail="Location not found on this project")
|
|
|
|
changed = {}
|
|
if body.name is not None:
|
|
name = body.name.strip()
|
|
if not name:
|
|
raise HTTPException(status_code=400, detail="A name cannot be empty")
|
|
if name != node.name:
|
|
changed["name"] = {"from": node.name, "to": name}
|
|
node.name = name
|
|
if body.active is not None and bool(body.active) != bool(node.active):
|
|
changed["active"] = {"from": bool(node.active), "to": bool(body.active)}
|
|
node.active = bool(body.active)
|
|
if not node.active:
|
|
for child in db.scalars(select(models.LocationNode).where(
|
|
(models.LocationNode.project_id == project_id)
|
|
& (models.LocationNode.path.like(node.path + "/%"))
|
|
)).all():
|
|
child.active = False
|
|
else:
|
|
# Walk up by path rather than by parent_id: one query, and it cannot
|
|
# loop on a malformed chain.
|
|
segs = node.path.split("/")
|
|
ancestors = ["/".join(segs[:n]) for n in range(1, len(segs))]
|
|
if ancestors:
|
|
for anc in db.scalars(select(models.LocationNode).where(
|
|
(models.LocationNode.project_id == project_id)
|
|
& (models.LocationNode.path.in_(ancestors))
|
|
)).all():
|
|
anc.active = True
|
|
|
|
if changed:
|
|
log_event(db, user, "location_updated", "project", project_id, project_id,
|
|
summary=node.path, detail=changed)
|
|
db.commit()
|
|
return node.to_dict()
|
|
|
|
|
|
@app.get("/api/projects/{project_id}/summary")
|
|
def project_summary(project_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
"""What the launcher needs to describe a project without asking the browser
|
|
what it remembers (B4).
|
|
|
|
The launcher used to read `wp_suite_sop_complete` out of localStorage, which is
|
|
a per-browser mirror: a colleague completing the SOP on their machine left your
|
|
card saying "Complete SOP first" with nothing to indicate the answer was stale."""
|
|
proj = db.get(models.Project, project_id)
|
|
if not proj:
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
require_project_access(db, user, project_id)
|
|
|
|
sop = db.scalars(
|
|
select(models.Sop)
|
|
.where(models.Sop.project_id == project_id, models.Sop.complete.is_(True))
|
|
.order_by(models.Sop.updated_at.desc())
|
|
.limit(1)
|
|
).first()
|
|
|
|
wp_total = db.scalar(
|
|
select(func.count()).select_from(models.WorkPackage).where(
|
|
models.WorkPackage.project_id == project_id,
|
|
models.WorkPackage.archived_at.is_(None),
|
|
)
|
|
) or 0
|
|
|
|
return {
|
|
"project_id": project_id,
|
|
"project_name": proj.name,
|
|
"sop_complete": sop is not None,
|
|
"sop_id": sop.id if sop else None,
|
|
"sop_name": (sop.name if sop else "") or "",
|
|
"sop_updated_at": models._iso(sop.updated_at) if sop else None,
|
|
"wp_total": wp_total,
|
|
"generated_at": models.utcnow().isoformat(),
|
|
}
|
|
|
|
|
|
@app.get("/api/wps/{wp_id}")
|
|
def get_wp(wp_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
wp = db.get(models.WorkPackage, wp_id)
|
|
if not wp:
|
|
raise HTTPException(status_code=404, detail="Work Package not found")
|
|
require_project_access(db, user, wp.project_id)
|
|
return wp.to_dict()
|
|
|
|
|
|
@app.delete("/api/wps/{wp_id}")
|
|
def delete_wp(wp_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
wp = db.get(models.WorkPackage, wp_id)
|
|
if not wp:
|
|
raise HTTPException(status_code=404, detail="Work Package not found")
|
|
# Deleting a work package is irreversible — Project Admin only. A project_user
|
|
# who wants one out of the way can archive it instead (reversible).
|
|
require_project_admin(db, user, wp.project_id, "Deleting a work package")
|
|
require_project_writable(db, user, wp.project_id, "Deleting a work package")
|
|
log_event(db, user, "deleted", "wp", wp.id, project_id=wp.project_id,
|
|
summary=(wp.number or wp.subject or wp.id))
|
|
db.delete(wp)
|
|
db.commit()
|
|
return {"deleted": wp_id}
|
|
|
|
|
|
@app.post("/api/wps/{wp_id}/issue")
|
|
def issue_wp(wp_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
"""Release a Work Package to the field. Refuses while a release gate is unmet:
|
|
any open constraint, or a predecessor package that isn't Closed."""
|
|
wp = db.get(models.WorkPackage, wp_id)
|
|
if not wp:
|
|
raise HTTPException(status_code=404, detail="Work Package not found")
|
|
require_project_access(db, user, wp.project_id)
|
|
require_project_writable(db, user, wp.project_id, "Issuing a work package")
|
|
enforce_release_gates(db, wp.id, wp.data, "Issued", wp.status)
|
|
wp.status = "Issued"
|
|
wp.issued_at = models.utcnow()
|
|
log_event(db, user, "issued", "wp", wp.id, project_id=wp.project_id,
|
|
summary=(wp.number or wp.subject or wp.id), detail={"to": "Issued"})
|
|
db.commit()
|
|
db.refresh(wp)
|
|
return wp.to_dict()
|
|
|
|
|
|
@app.post("/api/wps/{wp_id}/status")
|
|
def set_wp_status(wp_id: str, body: StatusIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
wp = db.get(models.WorkPackage, wp_id)
|
|
if not wp:
|
|
raise HTTPException(status_code=404, detail="Work Package not found")
|
|
require_project_access(db, user, wp.project_id)
|
|
require_project_writable(db, user, wp.project_id, "Changing a work package's status")
|
|
old_status = wp.status
|
|
# Same gates as /issue — this route must not be a way around them.
|
|
enforce_release_gates(db, wp.id, wp.data, body.status, old_status)
|
|
wp.status = body.status
|
|
if body.status == "Issued" and wp.issued_at is None:
|
|
wp.issued_at = models.utcnow()
|
|
if old_status != body.status:
|
|
log_event(db, user, "status_changed", "wp", wp.id, project_id=wp.project_id,
|
|
summary=(wp.number or wp.subject or wp.id), detail={"from": old_status, "to": body.status})
|
|
if "Issue" in (old_status, body.status):
|
|
_hlds = [h for h in ((wp.data or {}).get("holds") or []) if isinstance(h, dict)]
|
|
if body.status == "Issue":
|
|
_h = next((h for h in reversed(_hlds) if not h.get("released")), None)
|
|
log_event(db, user, "hold_logged", "wp", wp.id, project_id=wp.project_id,
|
|
summary=(wp.number or wp.subject or wp.id),
|
|
detail={"from": old_status,
|
|
"constraint": str((_h or {}).get("constraint") or "")[:200],
|
|
"reason": str((_h or {}).get("details") or "")[:300]})
|
|
else:
|
|
_h = next((h for h in reversed(_hlds) if h.get("released")), None)
|
|
log_event(db, user, "hold_released", "wp", wp.id, project_id=wp.project_id,
|
|
summary=(wp.number or wp.subject or wp.id),
|
|
detail={"to": body.status,
|
|
"constraint": str((_h or {}).get("constraint") or "")[:200],
|
|
"reason": str((_h or {}).get("details") or "")[:300]})
|
|
db.commit()
|
|
db.refresh(wp)
|
|
return wp.to_dict()
|
|
|
|
|
|
@app.post("/api/wps/{wp_id}/archive")
|
|
def archive_wp(wp_id: str, body: ArchiveIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
"""Archive (or unarchive) a Work Package — hides it from the default lists and
|
|
the dashboard without deleting it. Kept for the record on long-running jobs."""
|
|
wp = db.get(models.WorkPackage, wp_id)
|
|
if not wp:
|
|
raise HTTPException(status_code=404, detail="Work Package not found")
|
|
require_project_access(db, user, wp.project_id)
|
|
# An archived project is frozen whole: its packages keep the archive state they
|
|
# had, and tidying inside it waits until the job is unarchived.
|
|
require_project_writable(db, user, wp.project_id, "Archiving a work package")
|
|
was_archived = wp.archived_at is not None
|
|
if body.archived and not was_archived:
|
|
wp.archived_at = models.utcnow()
|
|
log_event(db, user, "archived", "wp", wp.id, project_id=wp.project_id,
|
|
summary=(wp.number or wp.subject or wp.id))
|
|
elif not body.archived and was_archived:
|
|
wp.archived_at = None
|
|
log_event(db, user, "unarchived", "wp", wp.id, project_id=wp.project_id,
|
|
summary=(wp.number or wp.subject or wp.id))
|
|
db.commit()
|
|
db.refresh(wp)
|
|
return wp.to_dict()
|
|
|
|
|
|
# ── Audit trail (history) ──────────────────────────────────────────────────────
|
|
@app.get("/api/audit")
|
|
def list_audit(
|
|
entity_type: Optional[str] = Query(None),
|
|
entity_id: Optional[str] = Query(None),
|
|
project_id: Optional[str] = Query(None),
|
|
action: Optional[str] = Query(None),
|
|
limit: int = Query(100, ge=1, le=500),
|
|
offset: int = Query(0, ge=0),
|
|
user: models.User = Depends(auth.get_current_user),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""History / audit trail. Pass entity_type+entity_id for one item's history,
|
|
or project_id for a project activity feed. Scoped to the caller's project
|
|
access; admins additionally see project-less events (user management)."""
|
|
stmt = select(models.AuditLog)
|
|
if entity_type:
|
|
stmt = stmt.where(models.AuditLog.entity_type == entity_type)
|
|
if entity_id:
|
|
stmt = stmt.where(models.AuditLog.entity_id == entity_id)
|
|
if action:
|
|
stmt = stmt.where(models.AuditLog.action == action)
|
|
if project_id:
|
|
require_project_access(db, user, project_id)
|
|
stmt = stmt.where(models.AuditLog.project_id == project_id)
|
|
# Non-admins only ever see events tied to a project they can access.
|
|
ids = accessible_project_ids(db, user)
|
|
if ids is not None:
|
|
stmt = stmt.where(models.AuditLog.project_id.in_(ids))
|
|
rows = db.scalars(stmt.order_by(models.AuditLog.at.desc()).limit(limit).offset(offset)).all()
|
|
return [e.to_dict() for e in rows]
|
|
|
|
|
|
# ── Global search ──────────────────────────────────────────────────────────────
|
|
def _like_term(q: str) -> str:
|
|
"""Escape LIKE wildcards so a user searching for '100%' or 'a_b' gets what they
|
|
typed rather than a pattern. Paired with escape='\\' on the comparison."""
|
|
return "%" + q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_").lower() + "%"
|
|
|
|
|
|
@app.get("/api/search")
|
|
def global_search(
|
|
q: str = Query("", min_length=0, max_length=200),
|
|
limit: int = Query(8, ge=1, le=25),
|
|
user: models.User = Depends(auth.get_current_user),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Type-ahead across projects, work packages and SOPs, scoped to the projects
|
|
the caller may access. Matches work-package number, subject, type and status,
|
|
project name/number/client/site, and SOP name/number."""
|
|
term = (q or "").strip()
|
|
if len(term) < 2:
|
|
return {"query": term, "projects": [], "wps": [], "sops": []}
|
|
pat = _like_term(term)
|
|
esc = "\\"
|
|
|
|
# An archived project is meant to be gone from view, and search is the one place
|
|
# it would otherwise come back — as the project itself, or as one of its packages
|
|
# or SOPs. So the archived ids are read once here and filtered out of all three
|
|
# result sets. Rows with no project at all (orphan/legacy, admin-only) survive
|
|
# explicitly: SQL's NOT IN drops NULLs, and they aren't on an archived job.
|
|
archived_pids = set(db.scalars(
|
|
select(models.Project.id).where(models.Project.archived_at.is_not(None))
|
|
).all())
|
|
|
|
proj_stmt = select(models.Project).where(
|
|
func.lower(models.Project.name).like(pat, escape=esc)
|
|
| func.lower(models.Project.number).like(pat, escape=esc)
|
|
| func.lower(models.Project.client).like(pat, escape=esc)
|
|
| func.lower(models.Project.site).like(pat, escape=esc)
|
|
)
|
|
if archived_pids:
|
|
proj_stmt = proj_stmt.where(models.Project.id.not_in(archived_pids))
|
|
proj_stmt = scope_to_access(proj_stmt, models.Project.id, db, user)
|
|
projects = db.scalars(proj_stmt.order_by(models.Project.updated_at.desc()).limit(limit)).all()
|
|
|
|
wp_stmt = select(models.WorkPackage).where(
|
|
(models.WorkPackage.archived_at.is_(None))
|
|
& (
|
|
func.lower(models.WorkPackage.number).like(pat, escape=esc)
|
|
| func.lower(models.WorkPackage.subject).like(pat, escape=esc)
|
|
| func.lower(models.WorkPackage.type).like(pat, escape=esc)
|
|
| func.lower(models.WorkPackage.status).like(pat, escape=esc)
|
|
)
|
|
)
|
|
if archived_pids:
|
|
wp_stmt = wp_stmt.where(
|
|
models.WorkPackage.project_id.is_(None)
|
|
| models.WorkPackage.project_id.not_in(archived_pids)
|
|
)
|
|
wp_stmt = scope_to_access(wp_stmt, models.WorkPackage.project_id, db, user)
|
|
wps = db.scalars(wp_stmt.order_by(models.WorkPackage.updated_at.desc()).limit(limit)).all()
|
|
|
|
sop_stmt = select(models.Sop).where(
|
|
func.lower(models.Sop.name).like(pat, escape=esc)
|
|
| func.lower(models.Sop.number).like(pat, escape=esc)
|
|
)
|
|
if archived_pids:
|
|
sop_stmt = sop_stmt.where(
|
|
models.Sop.project_id.is_(None) | models.Sop.project_id.not_in(archived_pids)
|
|
)
|
|
sop_stmt = scope_to_access(sop_stmt, models.Sop.project_id, db, user)
|
|
sops = db.scalars(sop_stmt.order_by(models.Sop.updated_at.desc()).limit(limit)).all()
|
|
|
|
# Project names for the WP/SOP rows, so a result reads unambiguously when the
|
|
# same WP number exists on two jobs.
|
|
pids = {w.project_id for w in wps} | {s.project_id for s in sops}
|
|
pids.discard(None)
|
|
names = {}
|
|
if pids:
|
|
for p in db.scalars(select(models.Project).where(models.Project.id.in_(pids))).all():
|
|
names[p.id] = p.name or p.number or p.id
|
|
|
|
return {
|
|
"query": term,
|
|
"projects": [{"id": p.id, "name": p.name, "number": p.number, "client": p.client} for p in projects],
|
|
"wps": [{
|
|
"id": w.id, "number": w.number, "subject": w.subject, "type": w.type,
|
|
"status": w.status, "project_id": w.project_id,
|
|
"project_name": names.get(w.project_id, ""),
|
|
} for w in wps],
|
|
"sops": [{
|
|
"id": s.id, "name": s.name, "number": s.number, "complete": s.complete,
|
|
"project_id": s.project_id, "project_name": names.get(s.project_id, ""),
|
|
} for s in sops],
|
|
}
|
|
|
|
|
|
# ── Settings (admin) ────────────────────────────────────────────────────────────
|
|
@app.get("/api/settings")
|
|
def get_app_settings(_admin: models.User = Depends(auth.require_admin), db: Session = Depends(get_db)):
|
|
return notify.public_settings(db)
|
|
|
|
|
|
@app.get("/api/app-flags")
|
|
def get_app_flags(_user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
"""Feature flags every signed-in page reads (e.g. whether the BIM/VDC tooling
|
|
is switched on). No secrets — safe for any authenticated user."""
|
|
return notify.app_flags(db)
|
|
|
|
|
|
@app.put("/api/settings")
|
|
def put_app_settings(body: SettingsIn, admin: models.User = Depends(auth.require_admin), db: Session = Depends(get_db)):
|
|
patch = {k: v for k, v in body.model_dump().items() if v is not None}
|
|
# Localization defaults are validated the same way a user's own preference is,
|
|
# so a typo can't leave every page formatting dates against a bogus zone.
|
|
loc = (patch.get("default_locale") or "").strip()
|
|
if loc and not _LOCALE_RE.match(loc):
|
|
raise HTTPException(status_code=400, detail="Default locale must be a language tag like 'en-US'.")
|
|
tz = (patch.get("default_timezone") or "").strip()
|
|
if tz and not valid_timezone(tz):
|
|
raise HTTPException(status_code=400, detail="Unknown default time zone.")
|
|
if "default_locale" in patch:
|
|
patch["default_locale"] = loc
|
|
if "default_timezone" in patch:
|
|
patch["default_timezone"] = tz
|
|
saved = notify.save_settings(db, patch)
|
|
log_event(db, admin, "settings_updated", "settings", "notifications",
|
|
summary="notifications", detail={"email_enabled": bool(saved.get("email_enabled"))})
|
|
db.commit()
|
|
return notify.public_settings(db)
|
|
|
|
|
|
@app.post("/api/settings/test-email")
|
|
def send_test_email(body: TestEmailIn, admin: models.User = Depends(auth.require_admin), db: Session = Depends(get_db)):
|
|
s = notify.get_settings(db)
|
|
if not notify.smtp_ready(s):
|
|
raise HTTPException(status_code=400, detail="Set the SMTP host and From address first.")
|
|
to = (body.to or admin.email or "").strip()
|
|
if not to:
|
|
raise HTTPException(status_code=400, detail="No recipient — add an email to your account or pass 'to'.")
|
|
try:
|
|
notify.send_email(s, to, "Work Package Suite — test email",
|
|
"This is a test from the Work Package Suite. If you got this, SMTP is working.")
|
|
except Exception as e: # noqa: BLE001
|
|
raise HTTPException(status_code=400, detail=f"Send failed: {e}")
|
|
return {"ok": True, "to": to}
|
|
|
|
|
|
# ── Notifications + project members ─────────────────────────────────────────────
|
|
@app.get("/api/notifications")
|
|
def list_notifications(all: bool = Query(False), limit: int = Query(100, ge=1, le=500),
|
|
user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
stmt = select(models.Notification)
|
|
if not (all and auth.is_admin(user)):
|
|
stmt = stmt.where(models.Notification.user_id == user.id)
|
|
rows = db.scalars(stmt.order_by(models.Notification.created_at.desc()).limit(limit)).all()
|
|
return [n.to_dict() for n in rows]
|
|
|
|
|
|
@app.get("/api/projects/{project_id}/members")
|
|
def project_members(project_id: str, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
"""Users who can be assigned WPs on this project — its members plus admins."""
|
|
require_project_access(db, user, project_id)
|
|
member_ids = set(db.scalars(select(models.ProjectMember.user_id).where(models.ProjectMember.project_id == project_id)).all())
|
|
members = db.scalars(select(models.User).where(models.User.id.in_(member_ids))).all() if member_ids else []
|
|
admins = db.scalars(select(models.User).where(models.User.role == auth.ROLE_ADMIN)).all()
|
|
out, seen = [], set()
|
|
for u in list(members) + list(admins):
|
|
if u.id in seen or not u.is_active:
|
|
continue
|
|
seen.add(u.id)
|
|
out.append({"id": u.id, "username": u.username, "full_name": u.full_name,
|
|
"email": u.email, "project_role": u.project_role or "",
|
|
"role": effective_role(db, u, project_id)})
|
|
out.sort(key=lambda x: (x["full_name"] or x["username"] or "").lower())
|
|
return out
|
|
|
|
|
|
# ── Comments / feedback ──────────────────────────────────────────────────────
|
|
def _save_comment(body: CommentIn, db: Session, user: "models.User") -> dict:
|
|
# A comment tied to a WP/SOP requires access to that resource's project, so
|
|
# a user can't write into another project's review thread. A review thread on an
|
|
# archived project is frozen with the rest of it; general app feedback isn't tied
|
|
# to a project at all and keeps working regardless.
|
|
#
|
|
# Both ids are checked independently — NOT if/elif. The row stores whichever ids
|
|
# the payload carried, so a body naming a WP you may touch AND a SOP you may not
|
|
# would, under an elif, be authorised on the WP alone and still land in the other
|
|
# project's SOP thread.
|
|
check_id(body.wp_id)
|
|
check_id(body.sop_id)
|
|
if body.wp_id:
|
|
wp = db.get(models.WorkPackage, body.wp_id)
|
|
if not wp:
|
|
raise HTTPException(status_code=404, detail="Work Package not found")
|
|
require_project_access(db, user, wp.project_id)
|
|
require_project_writable(db, user, wp.project_id, "Commenting on a work package")
|
|
if body.sop_id:
|
|
sop = db.get(models.Sop, body.sop_id)
|
|
if not sop:
|
|
raise HTTPException(status_code=404, detail="SOP not found")
|
|
require_project_access(db, user, sop.project_id)
|
|
require_project_writable(db, user, sop.project_id, "Commenting on a SOP")
|
|
extra = body.model_extra or {}
|
|
c = models.Comment(
|
|
id=gen_id("c"),
|
|
source=body.source or body.type or "",
|
|
sop_id=body.sop_id,
|
|
wp_id=body.wp_id,
|
|
step=body.step,
|
|
# Attribution comes from the authenticated session, NEVER the client
|
|
# payload — otherwise comments could be forged as another user.
|
|
author=(user.full_name or user.username),
|
|
text=body.text or "",
|
|
page=body.page or "",
|
|
extra=extra,
|
|
)
|
|
db.add(c)
|
|
db.commit()
|
|
db.refresh(c)
|
|
return c.to_dict()
|
|
|
|
|
|
@app.post("/api/comments")
|
|
def create_comment(body: CommentIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
return _save_comment(body, db, user)
|
|
|
|
|
|
# Alias so the existing client (which posts to /api/feedback) keeps working.
|
|
@app.post("/api/feedback")
|
|
def create_feedback(body: CommentIn, user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
|
|
return _save_comment(body, db, user)
|
|
|
|
|
|
@app.get("/api/comments")
|
|
def list_comments(
|
|
source: Optional[str] = Query(None),
|
|
sop_id: Optional[str] = Query(None),
|
|
wp_id: Optional[str] = Query(None),
|
|
step: Optional[int] = Query(None),
|
|
user: models.User = Depends(auth.get_current_user),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
stmt = select(models.Comment)
|
|
if source:
|
|
stmt = stmt.where(models.Comment.source == source)
|
|
if sop_id:
|
|
stmt = stmt.where(models.Comment.sop_id == sop_id)
|
|
if wp_id:
|
|
stmt = stmt.where(models.Comment.wp_id == wp_id)
|
|
if step is not None:
|
|
stmt = stmt.where(models.Comment.step == step)
|
|
# Non-admins see general app feedback plus comments on WPs/SOPs in their own
|
|
# projects only — never another project's review threads.
|
|
ids = accessible_project_ids(db, user)
|
|
if ids is not None:
|
|
acc_wp = select(models.WorkPackage.id).where(models.WorkPackage.project_id.in_(ids))
|
|
acc_sop = select(models.Sop.id).where(models.Sop.project_id.in_(ids))
|
|
# The "general feedback" branch is ONLY for comments not tied to any
|
|
# WP/SOP — otherwise a project-scoped comment tagged source=home_feedback
|
|
# by the client would leak across projects. Project-tied comments are
|
|
# visible strictly by project membership.
|
|
stmt = stmt.where(
|
|
((models.Comment.source == "home_feedback")
|
|
& models.Comment.wp_id.is_(None) & models.Comment.sop_id.is_(None))
|
|
| (models.Comment.wp_id.in_(acc_wp))
|
|
| (models.Comment.sop_id.in_(acc_sop))
|
|
)
|
|
rows = db.scalars(stmt.order_by(models.Comment.created_at.desc())).all()
|
|
return [c.to_dict() for c in rows]
|
|
|
|
|
|
# ── Local dev convenience: serve the static site from this app ──────────────────
|
|
# In production NGINX serves html/ and only proxies /api/ here, so this app never
|
|
# receives "/" requests, and the api Docker image doesn't even include html/ — so
|
|
# this mount stays inactive there. Locally (plain uvicorn, no NGINX) it lets you
|
|
# open the whole suite at http://localhost:8000/ with the API on the SAME origin,
|
|
# so the session cookie just works (no CORS, no Secure-cookie headache).
|
|
#
|
|
# Mounted LAST so the /api/* routes above always match first.
|
|
_html_dir = os.path.join(os.path.dirname(__file__), "..", "html")
|
|
if os.path.isdir(_html_dir):
|
|
|
|
class _NoCacheCode(StaticFiles):
|
|
"""Serve code assets with Cache-Control: no-cache.
|
|
|
|
Production runs behind NGINX (which now sets this itself), but the dev server
|
|
is what people actually click around in — and with no header at all the
|
|
browser applies HEURISTIC freshness per file (~10% of the file's age), so the
|
|
least recently changed file gets the longest lifetime and HTML/CSS/JS drift
|
|
apart between reloads. ETag/Last-Modified still make revalidation a cheap 304.
|
|
"""
|
|
|
|
async def get_response(self, path, scope):
|
|
res = await super().get_response(path, scope)
|
|
if path.endswith((".html", ".css", ".js", ".webmanifest")) or path in ("", "/", "."):
|
|
res.headers["Cache-Control"] = "no-cache"
|
|
return res
|
|
|
|
app.mount("/", _NoCacheCode(directory=_html_dir, html=True), name="site")
|