Aug 7 - Update: add external inspector

This commit is contained in:
2026-08-07 16:08:10 -04:00
parent 97c1dec54d
commit 6ca30c0dea
28 changed files with 727 additions and 96 deletions
+88 -7
View File
@@ -431,20 +431,64 @@ def create_user():
if form.validate_on_submit():
role = 'inspector' if director_editing else form.role.data
# MT-15 — an external inspector works for the customer or a third
# party, so we never set a password on their behalf. They are invited
# exactly like a customer: created with password_set=False (which the
# login route refuses until they finish), given a one-time token, and
# emailed a link to choose their own password.
invite = (role == 'external_inspector')
# UserForm.password is Optional() because the same form is used for
# EDIT, where blank means "keep current". On CREATE a blank password
# would otherwise store the hash of an empty string, so require one
# unless the account is being invited to choose their own.
if not invite and not form.password.data:
flash('Please set a password, or choose the External Inspector role '
'to send an invitation instead.', 'danger')
return render_template('auth/user_form.html', form=form, user=None,
title='Create User',
director_editing=director_editing)
user = User(
username=form.username.data,
full_name=form.full_name.data.strip() or None,
email=form.email.data.strip().lower(),
role=role
role=role,
password_set=not invite,
)
user.set_password(form.password.data)
if invite:
# A random unguessable placeholder — password_set=False already
# blocks login, but never leave an account holding a known or
# empty-string hash.
import secrets
user.set_password(secrets.token_hex(32))
else:
user.set_password(form.password.data)
db.session.add(user)
db.session.flush() # need user.id before minting the token
token = user.generate_set_password_token(expires_hours=72) if invite else None
db.session.commit()
logger.info('AUTH | user_create | admin_id=%s admin=%s new_user=%s role=%s',
current_user.id, current_user.username, user.username, user.role)
logger.info('AUTH | user_create | admin_id=%s admin=%s new_user=%s role=%s invite=%s',
current_user.id, current_user.username, user.username,
user.role, invite)
log_action(ACTION_CREATE, 'User', user.id, user.username,
f'role={user.role}; email={user.email}')
flash(f'User {user.username} created successfully.', 'success')
f'role={user.role}; email={user.email}; invite_sent={invite}')
if invite:
# Reuses the customer invitation email — the copy ("an account has
# been created for you… set your password") is already correct for
# any invited account. Imported inside the function to keep the
# auth ↔ customers import graph acyclic.
from app.routes.customers import _send_invite_email
_send_invite_email(user, token, base_url=request.host_url)
flash(f'External inspector {user.display_name} created. An invitation '
f'email has been sent to {user.email} with a link to set their '
f'password.', 'success')
else:
flash(f'User {user.username} created successfully.', 'success')
return redirect(url_for('auth.list_users'))
return render_template('auth/user_form.html', form=form, title='Create User',
@@ -488,12 +532,49 @@ def edit_user(user_id):
title='Edit User', director_editing=director_editing)
@bp.route('/users/<int:user_id>/resend-invite', methods=['POST'])
@login_required
@admin_required
def resend_invite(user_id):
"""Re-send the set-password invitation for an account still awaiting setup.
Without this an invitation that bounces, is deleted or expires leaves the
account permanently unusable — password_set=False blocks login and only a
valid token can clear it. Mirrors customers.resend_invite for staff-side
accounts (currently only external inspectors are ever invited this way).
"""
user = db.session.get(User, user_id)
if user is None:
abort(404)
if user.password_set:
flash(f'{user.display_name} has already completed their account setup.',
'info')
return redirect(url_for('auth.list_users'))
# A fresh token invalidates the previous link.
token = user.generate_set_password_token(expires_hours=72)
db.session.commit()
logger.info('AUTH | resend_invite | admin=%s user=%s',
current_user.username, user.username)
log_action(ACTION_UPDATE, 'User', user.id, user.username,
'invitation email resent')
from app.routes.customers import _send_invite_email
_send_invite_email(user, token, base_url=request.host_url)
flash(f'Invitation resent to {user.email}.', 'success')
return redirect(url_for('auth.list_users'))
@bp.route('/users/<int:user_id>/assign-contracts', methods=['GET', 'POST'])
@login_required
@admin_required
def assign_inspector_contracts(user_id):
user = db.session.get(User, user_id)
if user is None or user.role != 'inspector':
# MT-15: external inspectors are scoped by the same InspectorAssignment
# rows, so this page must accept them too.
if user is None or not user.is_inspector:
abort(404)
from app.models.project import Project
+4 -2
View File
@@ -22,10 +22,12 @@ logger = logging.getLogger(__name__)
bp = Blueprint('broadcast', __name__, url_prefix='/admin/broadcast')
# All roles that can hold an active iOS session
BROADCAST_ROLES = ['inspector', 'project_manager', 'director', 'admin']
BROADCAST_ROLES = ['inspector', 'external_inspector', 'project_manager',
'director', 'admin']
ROLE_LABELS = {
'inspector': 'Inspectors',
'inspector': 'Inspectors',
'external_inspector': 'External Inspectors',
'project_manager': 'Project Managers',
'director': 'Directors',
'admin': 'Admins',
+2 -2
View File
@@ -25,7 +25,7 @@ def index():
today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
today_end = today_start + timedelta(days=1)
is_inspector = current_user.role == 'inspector'
is_inspector = current_user.is_inspector
is_privileged = current_user.role in ['admin', 'director']
is_customer = current_user.role == 'customer'
is_project_manager = current_user.role == 'project_manager'
@@ -305,7 +305,7 @@ def index():
if is_privileged or is_project_manager or is_auditor:
active_inspectors = (
User.query
.filter_by(role='inspector', active=True)
.filter(User.role.in_(User.INSPECTOR_ROLES), User.active == True)
.order_by(User.full_name, User.username)
.all()
)
+6 -6
View File
@@ -23,7 +23,7 @@ def list_facilities():
facilities = Facility.query.filter(
Facility.id.in_(cids), Facility.active == True
).order_by(Facility.name).all()
elif current_user.role == 'inspector':
elif current_user.is_inspector:
fids = get_inspector_scope(current_user) or []
facilities = Facility.query.filter(
Facility.id.in_(fids), Facility.active == True
@@ -262,7 +262,7 @@ def _facility_for_qr_or_403(facility_id):
if facility is None:
abort(404)
# QR management is not an inspector task (matches qr_print_all/qr_export_pdf).
if current_user.role == 'inspector':
if current_user.is_inspector:
abort(403)
if current_user.role == 'customer':
cids = get_customer_scope(current_user) or []
@@ -329,7 +329,7 @@ def qr_sheet():
"""Bulk print sheet — one labeled QR card per active facility."""
from app.utils.qr import qr_svg
if current_user.role == 'inspector':
if current_user.is_inspector:
abort(403)
if current_user.role == 'customer':
@@ -395,7 +395,7 @@ def _area_for_qr_or_403(area_id):
if area is None:
abort(404)
# QR management is not an inspector task (matches qr_print_all/qr_export_pdf).
if current_user.role == 'inspector':
if current_user.is_inspector:
abort(403)
if current_user.role == 'customer':
cids = get_customer_scope(current_user) or []
@@ -483,7 +483,7 @@ def qr_print_all():
Inspectors have no QR management (403); customers are scoped to their
assigned facilities; managers see all active facilities.
"""
if current_user.role == 'inspector':
if current_user.is_inspector:
abort(403)
contract_id = request.args.get('contract_id', type=int)
@@ -552,7 +552,7 @@ def qr_export_pdf():
Scope is enforced per-id via the same helpers as the QR pages, so a
customer can never export a code outside their assigned facilities.
"""
if current_user.role == 'inspector':
if current_user.is_inspector:
abort(403)
facility_ids = request.form.getlist('facility_ids', type=int)
+1 -1
View File
@@ -125,7 +125,7 @@ def _can_view_full(facility):
return False
if current_user.role in ('admin', 'director', 'project_manager'):
return True
if current_user.role == 'inspector':
if current_user.is_inspector:
return facility.id in (get_inspector_scope(current_user) or [])
if current_user.role == 'customer':
return facility.id in (get_customer_scope(current_user) or [])
+4 -3
View File
@@ -189,7 +189,8 @@ def _active_inspectors():
"""Users who can be assigned inspections (inspector-capable roles)."""
return User.query.filter(
User.active.is_(True),
User.role.in_(['inspector', 'project_manager', 'director', 'admin']),
User.role.in_(['inspector', 'external_inspector', 'project_manager',
'director', 'admin']),
).order_by(User.full_name, User.username).all()
@@ -356,7 +357,7 @@ def index():
base = InspectionSchedule.query
# Inspectors see only their own assignments; managers see everything.
if current_user.role == 'inspector':
if current_user.is_inspector:
base = base.filter(InspectionSchedule.inspector_id == current_user.id)
# Counts are computed on the same scoped query, so the badges match what the
@@ -645,7 +646,7 @@ def start(schedule_id):
abort(404)
if current_user.role == 'customer':
abort(403)
if current_user.role == 'inspector' and schedule.inspector_id != current_user.id:
if current_user.is_inspector and schedule.inspector_id != current_user.id:
abort(403)
if not schedule.active:
+27 -19
View File
@@ -216,7 +216,7 @@ def index():
joinedload(Inspection.area),
).order_by(Inspection.inspection_date.desc())
if current_user.role == 'inspector':
if current_user.is_inspector:
fids = get_inspector_scope(current_user)
if not fids:
q = q.filter(False)
@@ -285,12 +285,12 @@ def index():
q = q.filter(Inspection.overall_score <= float(score_max_filter))
except ValueError:
score_max_filter = ''
if inspector_filter.isdigit() and current_user.role != 'inspector':
if inspector_filter.isdigit() and not current_user.is_inspector:
q = q.filter(Inspection.inspector_id == int(inspector_filter))
inspections = q.paginate(page=page, per_page=20, error_out=False)
if current_user.role == 'inspector':
if current_user.is_inspector:
fids = get_inspector_scope(current_user) or []
_fq = Facility.query.filter(Facility.id.in_(fids), Facility.active == True)
elif current_user.role == 'customer':
@@ -317,9 +317,9 @@ def index():
projects = Project.query.filter_by(active=True).order_by(Project.name).all()
# Inspector dropdown — shown to all roles except inspector (they only see their own)
if current_user.role != 'inspector':
if not current_user.is_inspector:
inspectors = (User.query
.filter(User.role == 'inspector', User.active == True)
.filter(User.role.in_(User.INSPECTOR_ROLES), User.active == True)
.order_by(User.full_name, User.username).all())
else:
inspectors = []
@@ -354,7 +354,7 @@ def start():
projects = Project.query.filter_by(active=True).order_by(Project.name).all()
# Scope projects to inspector's assigned contracts
if current_user.role == 'inspector':
if current_user.is_inspector:
from app.models.inspector_assignment import InspectorAssignment
assigned_pids = {
a.project_id for a in
@@ -410,7 +410,7 @@ def start():
# Inspector facility scope check — prevent crafted POST from selecting
# a facility outside their assigned contracts.
if current_user.role == 'inspector':
if current_user.is_inspector:
fids = get_inspector_scope(current_user)
if not fids or form.facility_id.data not in fids:
abort(403)
@@ -474,7 +474,7 @@ def execute(inspection_id):
if inspection is None:
abort(404)
if current_user.role == 'inspector' and inspection.inspector_id != current_user.id:
if current_user.is_inspector and inspection.inspector_id != current_user.id:
flash('Access denied.', 'danger')
return redirect(url_for('inspections.index'))
@@ -610,7 +610,8 @@ def execute(inspection_id):
return redirect(url_for('inspections.execute', inspection_id=inspection_id))
staff_for_flag_issue = User.query.filter(
User.role.in_(['director', 'inspector', 'project_manager', 'auditor']),
User.role.in_(['director', 'inspector', 'external_inspector',
'project_manager', 'auditor']),
User.active == True,
).order_by(User.full_name, User.username).all()
@@ -650,7 +651,7 @@ def save_draft_ajax(inspection_id):
if inspection is None:
abort(404)
if current_user.role == 'inspector' and inspection.inspector_id != current_user.id:
if current_user.is_inspector and inspection.inspector_id != current_user.id:
return jsonify({'ok': False, 'error': 'Access denied'}), 403
if inspection.status == 'completed':
@@ -689,7 +690,7 @@ def upload_photo_ajax(inspection_id):
if inspection is None:
return jsonify({'ok': False, 'error': 'Not found'}), 404
if current_user.role == 'inspector' and inspection.inspector_id != current_user.id:
if current_user.is_inspector and inspection.inspector_id != current_user.id:
return jsonify({'ok': False, 'error': 'Access denied'}), 403
if inspection.status == 'completed':
@@ -716,7 +717,7 @@ def view(inspection_id):
if inspection is None:
abort(404)
if current_user.role == 'inspector' and inspection.inspector_id != current_user.id:
if current_user.is_inspector and inspection.inspector_id != current_user.id:
flash('Access denied.', 'danger')
return redirect(url_for('inspections.index'))
if current_user.role == 'customer':
@@ -930,15 +931,22 @@ def flag_issue(inspection_id):
if inspection is None:
abort(404)
if current_user.role == 'inspector' and inspection.inspector_id != current_user.id:
if current_user.is_inspector and inspection.inspector_id != current_user.id:
flash('Access denied.', 'danger')
return redirect(url_for('inspections.index'))
form = IssueForm()
staff = User.query.filter(User.role.in_(['director', 'inspector'])).order_by(User.username).all()
staff = User.query.filter(
User.role.in_(['director', 'inspector', 'external_inspector'])
).order_by(User.username).all()
form.facility_id.choices = [(inspection.facility_id, inspection.facility.name)]
form.assigned_to.choices = [(0, '— Unassigned —')] + [(u.id, u.username) for u in staff]
# MT-15: suffix external (customer / third-party) inspectors so whoever is
# triaging can see the work is going outside the company. Display only.
form.assigned_to.choices = [(0, '— Unassigned —')] + [
(u.id, u.username + (' (External)' if u.is_external_inspector else ''))
for u in staff
]
if form.validate_on_submit():
photo_path = _save_photo(form.photo.data, subfolder='issue_photos')
@@ -1020,7 +1028,7 @@ def export_list_pdf():
joinedload(Inspection.area),
).order_by(Inspection.inspection_date.desc())
if current_user.role == 'inspector':
if current_user.is_inspector:
fids = get_inspector_scope(current_user)
if not fids:
q = q.filter(False)
@@ -1083,7 +1091,7 @@ def export_list_pdf():
q = q.filter(Inspection.overall_score <= float(score_max_filter))
except ValueError:
pass
if inspector_filter.isdigit() and current_user.role != 'inspector':
if inspector_filter.isdigit() and not current_user.is_inspector:
q = q.filter(Inspection.inspector_id == int(inspector_filter))
inspections = q.all()
@@ -1113,7 +1121,7 @@ def export_list_pdf():
filter_parts.append(f'Min score: {score_min_filter}%')
if score_max_filter:
filter_parts.append(f'Max score: {score_max_filter}%')
if inspector_filter.isdigit() and current_user.role != 'inspector':
if inspector_filter.isdigit() and not current_user.is_inspector:
u = db.session.get(User, int(inspector_filter))
if u:
filter_parts.append(f'Inspector: {u.display_name}')
@@ -1143,7 +1151,7 @@ def export_pdf(inspection_id):
if inspection is None:
abort(404)
if current_user.role == 'inspector' and inspection.inspector_id != current_user.id:
if current_user.is_inspector and inspection.inspector_id != current_user.id:
flash('Access denied.', 'danger')
return redirect(url_for('inspections.index'))
if current_user.role == 'customer':
+27 -10
View File
@@ -71,6 +71,18 @@ class _SLAFilteredPage:
return iter([1])
def _assignee_label(user):
"""Dropdown label for an assignee.
MT-15 external (customer / third-party) inspectors are assignable just
like the tenant's own crew, but are suffixed so whoever is triaging can see
at a glance that the work is going outside the company. Display only; the
stored value is still the user id.
"""
return (f'{user.display_name} (External)'
if user.is_external_inspector else user.display_name)
@bp.route('/export-list-pdf')
@login_required
def export_list_pdf():
@@ -86,7 +98,7 @@ def export_list_pdf():
.order_by(Issue.reported_at.desc())
)
if current_user.role == 'inspector':
if current_user.is_inspector:
fids = get_inspector_scope(current_user)
if not fids:
q = q.filter(False)
@@ -211,7 +223,7 @@ def index():
.order_by(Issue.reported_at.desc())
)
if current_user.role == 'inspector':
if current_user.is_inspector:
fids = get_inspector_scope(current_user)
if not fids:
q = q.filter(False)
@@ -319,7 +331,7 @@ def index():
# Facilities for the filter dropdown — scoped for inspectors/customers,
# then narrowed to the selected contract when contract_filter is active.
if current_user.role == 'inspector':
if current_user.is_inspector:
fids = get_inspector_scope(current_user) or []
_fq = Facility.query.filter(Facility.id.in_(fids), Facility.active == True)
elif current_user.role == 'customer':
@@ -347,7 +359,8 @@ def index():
# Staff for quick-assign dropdown — same roles as the full issue form
staff = User.query.filter(
User.role.in_(['director', 'inspector', 'auditor']), User.active == True
User.role.in_(['director', 'inspector', 'external_inspector', 'auditor']),
User.active == True
).order_by(User.username).all()
# Reporters dropdown — users who have actually filed at least one issue
@@ -385,7 +398,7 @@ def view(issue_id):
if issue is None:
abort(404)
if current_user.role == 'inspector':
if current_user.is_inspector:
fids = get_inspector_scope(current_user)
facility = issue.resolved_facility
if not fids or not facility or facility.id not in fids:
@@ -422,7 +435,7 @@ def view(issue_id):
return redirect(url_for('issues.view', issue_id=issue_id))
form = IssueUpdateForm(obj=issue)
staff = User.query.filter(User.role.in_(['director', 'inspector', 'auditor'])).order_by(User.username).all()
staff = User.query.filter(User.role.in_(['director', 'inspector', 'external_inspector', 'auditor'])).order_by(User.username).all()
# Preserve any pre-existing assignee who is no longer in the assignable set
# (e.g. an admin assigned before admins were removed from the dropdown) so
# saving the form doesn't silently unassign them.
@@ -430,7 +443,9 @@ def view(issue_id):
current_assignee = db.session.get(User, issue.assigned_to)
if current_assignee:
staff.append(current_assignee)
form.assigned_to.choices = [(0, '— Unassigned —')] + [(u.id, u.display_name) for u in staff]
form.assigned_to.choices = [(0, '— Unassigned —')] + [
(u.id, _assignee_label(u)) for u in staff
]
form.status.data = form.status.data or issue.status
if form.validate_on_submit():
@@ -743,10 +758,12 @@ def create():
else:
facilities = Facility.query.filter_by(active=True).order_by(Facility.name).all()
projects = Project.query.filter_by(active=True).order_by(Project.name).all()
staff = User.query.filter(User.role.in_(['director', 'inspector', 'auditor'])).order_by(User.username).all()
staff = User.query.filter(User.role.in_(['director', 'inspector', 'external_inspector', 'auditor'])).order_by(User.username).all()
form.facility_id.choices = [(f.id, f.name) for f in facilities]
form.assigned_to.choices = [(0, '— Unassigned —')] + [(u.id, u.display_name) for u in staff]
form.assigned_to.choices = [(0, '— Unassigned —')] + [
(u.id, _assignee_label(u)) for u in staff
]
# On POST validation error: identify which contract the submitted facility
# belongs to so the contract selector can be restored on re-render.
@@ -1129,7 +1146,7 @@ def export_pdf(issue_id):
if issue is None:
abort(404)
if current_user.role == 'inspector':
if current_user.is_inspector:
fids = get_inspector_scope(current_user)
facility = issue.resolved_facility
if not fids or not facility or facility.id not in fids:
+20 -11
View File
@@ -57,7 +57,8 @@ def index():
# Inspectors get a scoped view of their own inspections and related issues.
# Customers get a facility-scoped view.
# Internal management roles (director+) get the full unscoped view.
if current_user.role not in ['admin', 'director', 'project_manager', 'inspector', 'customer']:
if current_user.role not in ['admin', 'director', 'project_manager', 'inspector',
'external_inspector', 'customer']:
from flask import flash, redirect, url_for
flash('Access denied.', 'danger')
return redirect(url_for('dashboard.index'))
@@ -66,7 +67,7 @@ def index():
# Resolve scoping for customers (facility list) and inspectors (inspector_id)
customer_facility_ids = get_customer_scope(current_user) # None = unrestricted
is_inspector = current_user.role == 'inspector'
is_inspector = current_user.is_inspector
# Inspector filter — admin / director / project_manager only
inspector_filter = None
@@ -262,7 +263,8 @@ def index():
inspectors = []
if current_user.role in ('admin', 'director', 'project_manager'):
inspectors = User.query.filter_by(role='inspector', active=True)\
inspectors = User.query.filter(User.role.in_(User.INSPECTOR_ROLES),
User.active == True)\
.order_by(User.full_name, User.username).all()
facility_scores_list = [{
@@ -307,7 +309,8 @@ def index():
@bp.route('/facility/<int:facility_id>')
@login_required
def facility_report(facility_id):
if current_user.role not in ['admin', 'director', 'project_manager', 'inspector', 'customer']:
if current_user.role not in ['admin', 'director', 'project_manager', 'inspector',
'external_inspector', 'customer']:
from flask import flash, redirect, url_for
flash('Access denied.', 'danger')
return redirect(url_for('dashboard.index'))
@@ -320,7 +323,7 @@ def facility_report(facility_id):
from flask import flash, redirect, url_for
flash('Access denied.', 'danger')
return redirect(url_for('reports.index'))
if current_user.role == 'inspector':
if current_user.is_inspector:
# Inspectors may only view the facility report for facilities where
# they have personally conducted at least one inspection.
has_access = Inspection.query.filter_by(
@@ -369,7 +372,8 @@ def facility_report(facility_id):
def facility_scorecard(facility_id):
"""Comprehensive per-facility scorecard: score trend, SLA compliance,
issue breakdown by severity, inspection frequency."""
if current_user.role not in ['admin', 'director', 'project_manager', 'inspector', 'customer']:
if current_user.role not in ['admin', 'director', 'project_manager', 'inspector',
'external_inspector', 'customer']:
from flask import flash, redirect, url_for
flash('Access denied.', 'danger')
return redirect(url_for('dashboard.index'))
@@ -385,7 +389,7 @@ def facility_scorecard(facility_id):
flash('Access denied.', 'danger')
return redirect(url_for('reports.index'))
if current_user.role == 'inspector':
if current_user.is_inspector:
has_access = Inspection.query.filter_by(
facility_id=facility_id,
inspector_id=current_user.id,
@@ -701,7 +705,8 @@ def _build_inspector_stats(start, end):
all_ids = set(total_map.keys())
active_inspectors = (
User.query
.filter(User.id.in_(all_ids), User.active == True, User.role == 'inspector')
.filter(User.id.in_(all_ids), User.active == True,
User.role.in_(User.INSPECTOR_ROLES))
.order_by(User.full_name, User.username)
.all()
) if all_ids else []
@@ -717,6 +722,10 @@ def _build_inspector_stats(start, end):
inspector_stats.append({
'id': u.id,
'display_name': u.display_name,
# MT-15 — customer / third-party inspectors appear in the same
# table as the tenant's own crew, badged so the numbers can be read
# in context. Consumed by the HTML table and the export.
'external': u.is_external_inspector,
'total': tot,
'completed': comp,
'completion_rate': round(comp / tot * 100) if tot else 0,
@@ -759,7 +768,7 @@ def inspector_performance():
if selected_id:
selected_inspector = db.session.get(User, selected_id)
if selected_inspector and selected_inspector.role == 'inspector':
if selected_inspector and selected_inspector.is_inspector:
selected_kpis = next((s for s in inspector_stats if s['id'] == selected_id), None)
trend_rows = db.session.query(
@@ -857,7 +866,7 @@ def export_inspector_performance():
.filter(
Inspection.inspection_date >= start,
Inspection.inspection_date <= end,
User.role == 'inspector',
User.role.in_(User.INSPECTOR_ROLES),
)
if selected_id:
detail_q = detail_q.filter(Inspection.inspector_id == selected_id)
@@ -1628,7 +1637,7 @@ def facility_summary_pdf(facility_id):
cids = get_customer_scope(current_user) or []
if facility_id not in cids:
abort(403)
elif current_user.role == 'inspector':
elif current_user.is_inspector:
has = Inspection.query.filter_by(facility_id=facility_id,
inspector_id=current_user.id).first()
if not has: