nngo/LT_Janitorial_Quality_Control
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Aug 17 - Update customer roles management

Browse Source
This commit is contained in:
nngo
2026-08-17 14:18:10 -04:00
parent 3209a0c717
commit 5486145a24
24 changed files with 1119 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/routes/customers.py
+401 -69
View File
@@ -3,13 +3,26 @@ app/routes/customers.py
-----------------------
Customer Management — admin-only consolidated view.
Owns BOTH customer-side roles (phase51 — see User.CUSTOMER_ROLES):
Customer Director role='customer' portal access, read-mostly,
scoped by CustomerAssignment
Customer Inspector role='external_inspector' full inspector capabilities,
scoped by InspectorAssignment
They are two seats of the same customer organisation, so they are listed,
invited, edited, assigned, switched and disabled here rather than in User
Management — which now excludes both.
Provides a single screen to:
- List all customer-role users with their assignment summary
- Create a new customer account
- Edit an existing customer (username / email / password / active)
- Manage assignments for a customer (add / remove)
- Quick-disable / enable a customer account
- View a customer's scoped facility access at a glance
- List all customer-side users with their assignment summary
- Invite a new customer account in either role
- Edit an existing account (username / email / password / active)
- Manage assignments (contracts/facilities for a director, contracts for an
inspector)
- Switch an account between the two roles
- Quick-disable / enable an account
- View the account's scoped facility access at a glance
"""
import logging
@@ -18,6 +31,7 @@ from flask_login import login_required, current_user
from app import db
from app.models.user import User
from app.models.project import Project, CustomerAssignment
from app.models.inspector_assignment import InspectorAssignment
from app.models.facility import Facility
from app.utils.forms import CustomerUserForm, CustomerAssignmentForm, CustomerInviteForm, SetPasswordForm
from app.utils.decorators import admin_required, supervisor_required, safe_redirect_url
@@ -29,16 +43,46 @@ logger = logging.getLogger(__name__)
bp = Blueprint('customers', __name__, url_prefix='/customers')
def _get_customer_or_redirect(customer_id):
"""Load a customer-side account, or return a redirect response.
Returns (account, None) on success and (None, response) when the id is not
a customer-side account. Every route here used to test
`customer.role != 'customer'`, which would now reject the Customer
Inspectors this page owns — the check is CUSTOMER_ROLES, once, here.
"""
account = db.session.get(User, customer_id)
if account is None:
abort(404)
if not account.is_customer_account:
flash('This page is only for customer accounts.', 'warning')
return None, redirect(url_for('customers.index'))
return account, None
def _inspector_scope_ids(user, project_facilities_map):
"""Facility IDs a Customer Inspector reaches, from its contract rows.
Mirrors get_inspector_scope() but reuses the caller's already-loaded
project → facilities map so the list view stays free of N+1 queries
(rule 13).
"""
ids = set()
for a in InspectorAssignment.query.filter_by(user_id=user.id).all():
ids.update(project_facilities_map.get(a.project_id, []))
return ids
# ── List ──────────────────────────────────────────────────────────────────────
@bp.route('/')
@login_required
@supervisor_required
def index():
"""Consolidated customer management dashboard."""
"""Consolidated customer management dashboard — both customer roles."""
customers = (
User.query
.filter_by(role='customer')
.filter(User.role.in_(User.CUSTOMER_ROLES))
.order_by(User.username)
.all()
)
@@ -57,10 +101,27 @@ def index():
for a in all_assignments:
assignment_map[a.user_id].append(a)
# ── Bulk query for the inspector-side assignments ─────────────────────
# Customer Inspectors are scoped by InspectorAssignment, not
# CustomerAssignment — the two roles read different tables for the same
# question ("which facilities does this account see?").
all_inspector_assignments = (
InspectorAssignment.query
.filter(InspectorAssignment.user_id.in_(customer_ids))
.all()
) if customer_ids else []
inspector_assignment_map = {c.id: [] for c in customers}
for a in all_inspector_assignments:
inspector_assignment_map[a.user_id].append(a)
# ── Single bulk query for all active facilities in assigned projects ──
# Resolves facility scope for every customer without repeated DB round-trips.
from collections import defaultdict
assigned_project_ids = {a.project_id for a in all_assignments}
assigned_project_ids = (
{a.project_id for a in all_assignments}
| {a.project_id for a in all_inspector_assignments}
)
project_facilities_map = defaultdict(list) # project_id → [facility_id, ...]
if assigned_project_ids:
@@ -78,11 +139,17 @@ def index():
scope_map = {} # user_id → sorted list[int] facility IDs
for customer in customers:
ids = set()
for a in assignment_map[customer.id]:
if a.facility_id:
ids.add(a.facility_id)
else:
if customer.is_inspector:
# Customer Inspector — contract-level rows only, no facility-level
# narrowing exists for inspectors (rule 57: no rows = sees nothing).
for a in inspector_assignment_map[customer.id]:
ids.update(project_facilities_map.get(a.project_id, []))
else:
for a in assignment_map[customer.id]:
if a.facility_id:
ids.add(a.facility_id)
else:
ids.update(project_facilities_map.get(a.project_id, []))
scope_map[customer.id] = sorted(ids)
# All active projects for the assignment modal
@@ -101,11 +168,12 @@ def index():
return render_template(
'customers/index.html',
customers = customers,
assignment_map = assignment_map,
scope_map = scope_map,
projects = projects,
expired_invitations = expired_invitations,
customers = customers,
assignment_map = assignment_map,
inspector_assignment_map = inspector_assignment_map,
scope_map = scope_map,
projects = projects,
expired_invitations = expired_invitations,
)
@@ -115,12 +183,16 @@ def index():
@login_required
@supervisor_required
def create():
"""Create a customer account via email invitation.
"""Create a customer-side account via email invitation.
Admin enters Full Name and Email only. A temporary username is
auto-generated from the email address. A one-time set-password link
is emailed; the customer chooses their own username and password when
they click it. The account is activated on completion.
Admin enters Full Name, Email and the role (Customer Director or Customer
Inspector). A temporary username is auto-generated from the email address.
A one-time set-password link is emailed; the invitee chooses their own
username and password when they click it. The account is activated on
completion.
Both roles take this identical path — an account that belongs to the
customer is never given a password we chose.
"""
form = CustomerInviteForm()
@@ -129,6 +201,11 @@ def create():
full_name = form.full_name.data.strip()
email = form.email.data.strip().lower()
role = form.role.data
# Defence in depth: never let a crafted POST mint a staff role through
# the customer invitation form, which sets no password.
if role not in User.CUSTOMER_ROLES:
role = 'customer'
# Auto-generate a temporary username from the email local part.
# The customer replaces this with their preferred username when
@@ -146,7 +223,7 @@ def create():
username = username,
full_name = full_name,
email = email,
role = 'customer',
role = role,
active = True,
password_set = False,
)
@@ -157,15 +234,15 @@ def create():
token = user.generate_set_password_token(expires_hours=72)
db.session.commit()
logger.info('CUSTOMERS | invite | admin=%s new_customer=%s email=%s',
current_user.username, user.username, user.email)
logger.info('CUSTOMERS | invite | admin=%s new_customer=%s role=%s email=%s',
current_user.username, user.username, user.role, user.email)
log_action(ACTION_CREATE, 'User', user.id, user.username,
f'role=customer; email={user.email}; invite_sent=True')
f'role={user.role}; email={user.email}; invite_sent=True')
_send_invite_email(user, token, base_url=request.host_url)
flash(
f'Customer account created for {full_name}. '
f'{user.role_label} account created for {full_name}. '
f'An invitation email has been sent to {email} with a link to set their username and password.',
'success'
)
@@ -258,12 +335,9 @@ def _send_invite_email(user, token, base_url=None):
@supervisor_required
def resend_invite(customer_id):
"""Generate a fresh token and resend the set-password invitation email."""
customer = db.session.get(User, customer_id)
if customer is None:
abort(404)
if customer.role != 'customer':
flash('This action is only for customer accounts.', 'warning')
return redirect(url_for('customers.index'))
customer, moved = _get_customer_or_redirect(customer_id)
if moved:
return moved
token = customer.generate_set_password_token(expires_hours=72)
customer.password_set = False
@@ -320,12 +394,9 @@ def set_password(token):
@login_required
@supervisor_required
def edit(customer_id):
customer = db.session.get(User, customer_id)
if customer is None:
abort(404)
if customer.role != 'customer':
flash('This page is only for customer accounts.', 'warning')
return redirect(url_for('customers.index'))
customer, moved = _get_customer_or_redirect(customer_id)
if moved:
return moved
form = CustomerUserForm(user=customer, obj=customer)
@@ -353,36 +424,78 @@ def edit(customer_id):
@login_required
@supervisor_required
def manage(customer_id):
"""Single-customer detail page: profile + all assignments."""
customer = db.session.get(User, customer_id)
if customer is None:
abort(404)
if customer.role != 'customer':
flash('This page is only for customer accounts.', 'warning')
return redirect(url_for('customers.index'))
"""Single-account detail page: profile + assignments + notification matrix.
assignments = CustomerAssignment.query.filter_by(user_id=customer_id).all()
facility_ids = get_customer_scope(customer) or []
facilities = (
The assignment editor differs by role. A Customer Director gets the
contract/facility assignment list (CustomerAssignment); a Customer
Inspector gets the contract checkbox set (InspectorAssignment) that used to
live on /auth/users/<id>/assign-contracts.
"""
customer, moved = _get_customer_or_redirect(customer_id)
if moved:
return moved
# Resolve the account's facility scope through the SAME helper the app uses
# at request time, so this page can never disagree with what the account
# actually sees.
if customer.is_inspector:
from app.utils.scope import get_inspector_scope
facility_ids = get_inspector_scope(customer) or []
else:
facility_ids = get_customer_scope(customer) or []
facilities = (
Facility.query
.filter(Facility.id.in_(facility_ids), Facility.active == True)
.order_by(Facility.name)
.all()
) if facility_ids else []
# Assignment form (populated here so it can be rendered inline)
aform = CustomerAssignmentForm()
projects = Project.query.filter_by(active=True).order_by(Project.name).all()
assignments = []
assigned_pids = set()
if customer.is_inspector:
assigned_pids = {
a.project_id
for a in InspectorAssignment.query.filter_by(user_id=customer_id).all()
}
else:
assignments = CustomerAssignment.query.filter_by(user_id=customer_id).all()
# Assignment form (populated here so it can be rendered inline)
aform = CustomerAssignmentForm()
aform.user_id.choices = [(customer.id, customer.username)]
aform.facility_id.choices = [(0, '— All facilities in contract —')]
# ── Per-account notification matrix ───────────────────────────────────
# For each event: what the global matrix would do for this account's role,
# and whether the account overrides it. The template renders a tri-state
# (Inherit / On / Off) so "inherit" stays visibly distinct from "explicitly
# set to the same value the global happens to have today".
from app.models.notification_matrix import MATRIX_EVENTS, is_enabled
from app.models.user_notification_matrix import overrides_for_user
overrides = overrides_for_user(customer.id)
matrix_rows = [
{
'event': event_key,
'label': label,
'global': is_enabled(event_key, customer.role),
'override': overrides.get(event_key), # True / False / None
}
for event_key, label in MATRIX_EVENTS.items()
]
return render_template(
'customers/manage.html',
customer = customer,
assignments = assignments,
facilities = facilities,
aform = aform,
projects = projects,
customer = customer,
assignments = assignments,
assigned_pids = assigned_pids,
facilities = facilities,
aform = aform,
projects = projects,
matrix_rows = matrix_rows,
)
@@ -392,12 +505,16 @@ def manage(customer_id):
@login_required
@supervisor_required
def add_assignment(customer_id):
customer = db.session.get(User, customer_id)
if customer is None:
abort(404)
if customer.role != 'customer':
flash('Assignments are only for customer accounts.', 'warning')
return redirect(url_for('customers.index'))
customer, moved = _get_customer_or_redirect(customer_id)
if moved:
return moved
if customer.is_inspector:
# A Customer Inspector is scoped by InspectorAssignment — writing a
# CustomerAssignment row for them would grant nothing while looking
# like it had.
flash('Customer Inspectors are assigned whole contracts — use the '
'contract list on this page.', 'warning')
return redirect(url_for('customers.manage', customer_id=customer_id))
project_id = request.form.get('project_id', type=int)
facility_id = request.form.get('facility_id', type=int) or None
@@ -466,18 +583,233 @@ def remove_assignment(assignment_id):
return redirect(url_for('customers.manage', customer_id=customer_id))
# ── Contract assignments for a Customer Inspector ────────────────────────────
@bp.route('/<int:customer_id>/contracts', methods=['POST'])
@login_required
@supervisor_required
def assign_contracts(customer_id):
"""Replace a Customer Inspector's whole InspectorAssignment set.
Same replace-the-entire-set semantics as auth.assign_inspector_contracts
(rule 59) — the form posts the complete checked list, rows not in the POST
body are deleted. Callers must always send the full desired set, never a
diff.
"""
customer, moved = _get_customer_or_redirect(customer_id)
if moved:
return moved
if not customer.is_inspector:
flash('Contract assignment is for Customer Inspector accounts. '
'Customer Directors are assigned per contract or facility below.',
'warning')
return redirect(url_for('customers.manage', customer_id=customer_id))
from app.utils.time_utils import now_eastern
selected_ids = set(request.form.getlist('project_ids', type=int))
existing = InspectorAssignment.query.filter_by(user_id=customer_id).all()
existing_pids = {a.project_id for a in existing}
for a in existing:
if a.project_id not in selected_ids:
db.session.delete(a)
for pid in selected_ids:
if pid not in existing_pids:
db.session.add(InspectorAssignment(
user_id = customer_id,
project_id = pid,
created_at = now_eastern(),
))
db.session.commit()
logger.info('CUSTOMERS | assign_contracts | admin=%s customer=%s projects=%s',
current_user.username, customer.username, sorted(selected_ids))
log_action(ACTION_UPDATE, 'User', customer.id, customer.username,
f'inspector_assignments={sorted(selected_ids)}')
if not selected_ids:
# Rule 57 is strict, and silently is exactly how it bites.
flash(f'{customer.display_name} now has no contracts assigned and will '
f'see nothing until at least one is granted.', 'warning')
else:
flash(f'Contract assignments updated for {customer.display_name}.', 'success')
return redirect(url_for('customers.manage', customer_id=customer_id))
# ── Per-account notification matrix ──────────────────────────────────────────
@bp.route('/<int:customer_id>/notifications', methods=['POST'])
@login_required
@supervisor_required
def save_notifications(customer_id):
"""Save this account's per-event notification overrides.
Each event posts one of 'inherit' / 'on' / 'off'. 'inherit' DELETES the row
rather than storing the global column's current value — so an account that
never expressed an opinion keeps following the global matrix when it
changes later.
"""
customer, moved = _get_customer_or_redirect(customer_id)
if moved:
return moved
from app.models.notification_matrix import MATRIX_EVENTS
from app.models.user_notification_matrix import set_overrides
tri = {'inherit': None, 'on': True, 'off': False}
values = {}
for event_key in MATRIX_EVENTS:
# Only events this form actually posted; an unknown or missing value
# is treated as inherit rather than guessed at.
choice = request.form.get(f'event_{event_key}')
if choice is not None:
values[event_key] = tri.get(choice)
changed = set_overrides(customer.id, values)
db.session.commit()
logger.info('CUSTOMERS | notif_matrix | admin=%s customer=%s changed=%s',
current_user.username, customer.username, changed)
log_action(ACTION_UPDATE, 'User', customer.id, customer.username,
f'notification overrides updated ({changed} change(s))')
flash(f'Notification settings saved for {customer.display_name}.'
if changed else 'No notification changes to save.',
'success' if changed else 'info')
return redirect(url_for('customers.manage', customer_id=customer.id))
# ── Switch between the two customer roles ────────────────────────────────────
@bp.route('/<int:customer_id>/switch-role', methods=['POST'])
@login_required
@admin_required
def switch_role(customer_id):
"""Flip an account between Customer Director and Customer Inspector.
The two roles read DIFFERENT scoping tables, so flipping the column alone
would leave the account correctly labelled and seeing nothing (rule 57 is
strict for inspectors, and a director with no CustomerAssignment rows is
equally blind). The contracts are therefore mirrored across: every contract
the account could reach before, it can reach after.
Facility-level narrowing does NOT survive a switch to inspector — there is
no per-facility row for inspectors, so a director scoped to one building in
a contract becomes an inspector on that whole contract. The confirm dialog
says so; the flash repeats it. Switching BACK is lossless though: the
original facility-level rows were never deleted, and the reverse mirror
skips contracts the account can already reach, so it does not pile a
contract-wide grant on top of them.
API access changes in both directions ('external_inspector' has mobile API
access, 'customer' is 403 everywhere), so the account's refresh tokens and
device registrations are revoked — an issued JWT would otherwise keep
working until it expired, and a signed-in iPad would keep syncing.
"""
customer, moved = _get_customer_or_redirect(customer_id)
if moved:
return moved
from app.utils.time_utils import now_eastern
from app.models.api_token import RefreshToken, DeviceToken
old_role = customer.role
new_role = 'external_inspector' if old_role == 'customer' else 'customer'
widened = False
if new_role == 'external_inspector':
# Director → Inspector: CustomerAssignment (contract or facility) →
# InspectorAssignment (contract only).
existing_pids = {
a.project_id
for a in InspectorAssignment.query.filter_by(user_id=customer.id).all()
}
for a in CustomerAssignment.query.filter_by(user_id=customer.id).all():
if a.facility_id:
widened = True
if a.project_id not in existing_pids:
db.session.add(InspectorAssignment(
user_id = customer.id,
project_id = a.project_id,
created_at = now_eastern(),
))
existing_pids.add(a.project_id)
else:
# Inspector → Director: contract-level CustomerAssignment rows
# (facility_id NULL = all facilities in the contract).
#
# `existing_pids` counts ANY row for the contract, facility-level ones
# included — NOT just the contract-wide ones. That is what makes a
# round trip lossless: an account narrowed to one facility, switched to
# inspector (which can only hold whole contracts) and switched back
# would otherwise gain a contract-wide row on top of its original
# facility row and come back with the whole contract. Skipping
# contracts the account can already reach as a director leaves the
# original narrowing intact, while contracts granted during the
# inspector spell still carry over.
existing_pids = {
a.project_id
for a in CustomerAssignment.query.filter_by(user_id=customer.id).all()
}
for a in InspectorAssignment.query.filter_by(user_id=customer.id).all():
if a.project_id not in existing_pids:
db.session.add(CustomerAssignment(
user_id = customer.id,
project_id = a.project_id,
facility_id = None,
))
existing_pids.add(a.project_id)
# The stale rows for the role being left are kept on purpose: switching
# back restores the account's original scope, including any facility-level
# narrowing that the inspector side cannot express. They are inert while
# the other role is active — each scope helper reads only its own table.
customer.role = new_role
revoked = (
RefreshToken.query
.filter_by(user_id=customer.id, revoked=False)
.update({'revoked': True}, synchronize_session=False)
)
devices = (
DeviceToken.query
.filter_by(user_id=customer.id)
.delete(synchronize_session=False)
)
db.session.commit()
logger.info('CUSTOMERS | switch_role | admin=%s customer=%s %s -> %s '
'tokens_revoked=%s devices_cleared=%s',
current_user.username, customer.username, old_role, new_role,
revoked, devices)
log_action(ACTION_UPDATE, 'User', customer.id, customer.username,
f'role switched {old_role} -> {new_role}; '
f'refresh_tokens_revoked={revoked}; devices_cleared={devices}')
msg = (f'{customer.display_name} is now a {customer.role_label}. '
f'Their contracts were carried across; any signed-in device must log in again.')
if widened:
msg += (' Note: facility-level limits do not exist for inspectors, so '
'this account now covers every facility in those contracts — '
'review the contract list below.')
flash(msg, 'warning' if widened else 'success')
return redirect(url_for('customers.manage', customer_id=customer.id))
# ── Toggle active ─────────────────────────────────────────────────────────────
@bp.route('/<int:customer_id>/toggle-active', methods=['POST'])
@login_required
@supervisor_required
def toggle_active(customer_id):
customer = db.session.get(User, customer_id)
if customer is None:
abort(404)
if customer.role != 'customer':
flash('This action is only for customer accounts.', 'warning')
return redirect(url_for('customers.index'))
customer, moved = _get_customer_or_redirect(customer_id)
if moved:
return moved
customer.active = not customer.active
db.session.commit()