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

05/25 Update inspectors contract assignment

Browse Source
This commit is contained in:
nngo
2026-05-25 13:58:37 -04:00
parent e703675370
commit 688308fca2
13 changed files with 463 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/facilities.py
+27 -14
View File
@@ -23,7 +23,7 @@ from app.models.facility import Facility, Area
from app.models.project import Project
from app.api.errors import api_ok, api_error
from app.api.decorators import jwt_required
from app.utils.scope import get_customer_scope
from app.utils.scope import get_customer_scope, get_inspector_scope
logger = logging.getLogger(__name__)
@@ -92,27 +92,35 @@ def list_facilities():
"""
user = g.api_user
# Customer role: honour facility-level scoping
facility_ids = get_customer_scope(user)
customer_fids = get_customer_scope(user)
inspector_fids = get_inspector_scope(user)
if facility_ids is not None:
if customer_fids is not None:
# Customer — scope to assigned facilities only
if not facility_ids:
if not customer_fids:
logger.info('API FACILITIES | user=%s | role=customer | no_assignments',
user.username)
return api_ok({'facilities': [], 'count': 0})
facilities = (
Facility.query
.filter(
Facility.id.in_(facility_ids),
Facility.active == True, # noqa: E712
)
.filter(Facility.id.in_(customer_fids), Facility.active == True)
.order_by(Facility.name)
.all()
)
elif inspector_fids is not None:
# Inspector — scope to contracted facilities
if not inspector_fids:
logger.info('API FACILITIES | user=%s | role=inspector | no_assignments',
user.username)
return api_ok({'facilities': [], 'count': 0})
facilities = (
Facility.query
.filter(Facility.id.in_(inspector_fids), Facility.active == True)
.order_by(Facility.name)
.all()
)
else:
# Internal staff — all active facilities
# All other staff — all active facilities
facilities = (
Facility.query
.filter(Facility.active == True) # noqa: E712
@@ -159,9 +167,14 @@ def list_areas(facility_id):
if facility is None or not facility.active:
return api_error('Facility not found', 404)
# Customer scope validation — ensure the customer is assigned to this facility
facility_ids = get_customer_scope(user)
if facility_ids is not None and facility_id not in facility_ids:
# Scope validation — customers and inspectors may only access their facilities
customer_fids = get_customer_scope(user)
inspector_fids = get_inspector_scope(user)
if customer_fids is not None and facility_id not in customer_fids:
logger.warning('API FACILITIES/AREAS | access denied | user=%s | facility_id=%d',
user.username, facility_id)
return api_error('Access denied', 403)
if inspector_fids is not None and facility_id not in inspector_fids:
logger.warning('API FACILITIES/AREAS | access denied | user=%s | facility_id=%d',
user.username, facility_id)
return api_error('Access denied', 403)
app/api/issues.py
+27 -13
View File
@@ -35,6 +35,7 @@ from app.api.decorators import jwt_required
from app.utils.audit import log_action, ACTION_CREATE, ACTION_UPDATE
from app.utils.notifications import notify_by_matrix
from app.utils.time_utils import now_eastern
from app.utils.scope import get_inspector_scope
logger = logging.getLogger(__name__)
@@ -112,14 +113,13 @@ def list_issues():
query = Issue.query
if user.role == 'inspector':
# Inspectors see issues assigned to them OR issues they reported.
# The reported_by path covers issues created on the iPad that haven't
# been assigned yet (assigned_to is NULL until a director assigns them).
# Pre-phase18 rows with reported_by = NULL still surface via assigned_to.
query = query.filter(
fids = get_inspector_scope(user)
if not fids:
return api_ok({'issues': [], 'total': 0, 'limit': limit, 'offset': offset})
query = query.outerjoin(Area, Issue.area_id == Area.id).filter(
db.or_(
Issue.assigned_to == user.id,
Issue.reported_by == user.id,
Issue.facility_id.in_(fids),
db.and_(Issue.area_id.isnot(None), Area.facility_id.in_(fids)),
)
)
else:
@@ -204,6 +204,11 @@ def create_issue():
if facility is None:
return api_error('Facility not found', 404)
if user.role == 'inspector':
fids = get_inspector_scope(user)
if not fids or facility_id not in fids:
return api_error('Access denied — facility is not in your assigned contracts', 403)
inspection_id = data.get('inspection_id')
if inspection_id:
inspection = db.session.get(Inspection, inspection_id)
@@ -290,8 +295,11 @@ def get_issue(issue_id):
if issue is None:
return api_error('Issue not found', 404)
if user.role == 'inspector' and issue.assigned_to != user.id and issue.reported_by != user.id:
return api_error('Access denied', 403)
if user.role == 'inspector':
fids = get_inspector_scope(user)
facility = issue.resolved_facility
if not fids or not facility or facility.id not in fids:
return api_error('Access denied', 403)
return api_ok(_issue_payload(issue))
@@ -320,8 +328,11 @@ def update_issue_status(issue_id):
if issue is None:
return api_error('Issue not found', 404)
if user.role == 'inspector' and issue.assigned_to != user.id and issue.reported_by != user.id:
return api_error('Access denied — you can only update issues assigned to or reported by you', 403)
if user.role == 'inspector':
fids = get_inspector_scope(user)
facility = issue.resolved_facility
if not fids or not facility or facility.id not in fids:
return api_error('Access denied', 403)
data = request.get_json(silent=True) or {}
new_status = (data.get('status') or '').strip().lower()
@@ -382,8 +393,11 @@ def update_issue_photos(issue_id):
if issue is None:
return api_error('Issue not found', 404)
if user.role == 'inspector' and issue.assigned_to != user.id and issue.reported_by != user.id:
return api_error('Access denied', 403)
if user.role == 'inspector':
fids = get_inspector_scope(user)
facility = issue.resolved_facility
if not fids or not facility or facility.id not in fids:
return api_error('Access denied', 403)
data = request.get_json(silent=True) or {}
raw = data.get('result_photos')
app/models/inspector_assignment.py
+25
View File
@@ -0,0 +1,25 @@
from app import db
from app.utils.time_utils import now_eastern
class InspectorAssignment(db.Model):
__tablename__ = 'inspector_assignments'
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer,
db.ForeignKey('users.id', ondelete='CASCADE'),
nullable=False, index=True)
project_id = db.Column(db.Integer,
db.ForeignKey('projects.id', ondelete='CASCADE'),
nullable=False)
created_at = db.Column(db.DateTime, nullable=False, default=now_eastern)
inspector = db.relationship('User', backref='inspector_assignments')
project = db.relationship('Project', backref='inspector_assignments')
__table_args__ = (
db.UniqueConstraint('user_id', 'project_id', name='uq_inspector_project'),
)
def __repr__(self):
return f'<InspectorAssignment user={self.user_id} project={self.project_id}>'
app/routes/auth.py
+62 -1
View File
@@ -120,9 +120,24 @@ def list_users():
.order_by(User.created_at.desc())
.all()
)
# Build a map of inspector_id -> assignment count for the Contracts column
from app.models.inspector_assignment import InspectorAssignment
from sqlalchemy import func
rows = (
db.session.query(
InspectorAssignment.user_id,
func.count(InspectorAssignment.id).label('cnt'),
)
.group_by(InspectorAssignment.user_id)
.all()
)
inspector_contract_counts = {r.user_id: r.cnt for r in rows}
logger.info('AUTH | list_users | admin=%s | internal_users_count=%s',
current_user.username, len(users))
return render_template('auth/users.html', users=users)
return render_template('auth/users.html', users=users,
inspector_contract_counts=inspector_contract_counts)
@bp.route('/users/new', methods=['GET', 'POST'])
@@ -194,6 +209,52 @@ def edit_user(user_id):
title='Edit User', director_editing=director_editing)
@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':
abort(404)
from app.models.project import Project
from app.models.inspector_assignment import InspectorAssignment
from app.utils.time_utils import now_eastern
projects = Project.query.filter_by(active=True).order_by(Project.name).all()
if request.method == 'POST':
selected_ids = set(request.form.getlist('project_ids', type=int))
existing = InspectorAssignment.query.filter_by(user_id=user_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 = user_id,
project_id = pid,
created_at = now_eastern(),
))
db.session.commit()
log_action(ACTION_UPDATE, 'User', user.id, user.username,
f'inspector_assignments={sorted(selected_ids)}')
flash(f'Contract assignments updated for {user.display_name}.', 'success')
return redirect(url_for('auth.list_users'))
assigned_pids = {
a.project_id
for a in InspectorAssignment.query.filter_by(user_id=user_id).all()
}
return render_template('auth/inspector_assignments.html',
user=user,
projects=projects,
assigned_pids=assigned_pids)
@bp.route('/users/<int:user_id>/delete', methods=['POST'])
@login_required
@admin_required
app/routes/dashboard.py
+74 -19
View File
@@ -7,7 +7,7 @@ from app.models.facility import Facility
from app.models.issue import Issue
from app.models.user import User
from app.utils.sla import sla_status, SLA_HOURS
from app.utils.scope import get_customer_scope
from app.utils.scope import get_customer_scope, get_inspector_scope
from sqlalchemy import func
from datetime import datetime, timedelta
from app.utils.time_utils import now_eastern
@@ -31,13 +31,20 @@ def index():
is_customer = current_user.role == 'customer'
is_project_manager = current_user.role == 'project_manager'
# Resolve facility scope for customer users
customer_facility_ids = get_customer_scope(current_user) # None for non-customers
# Resolve facility scope
customer_facility_ids = get_customer_scope(current_user) # None for non-customers
inspector_facility_ids = get_inspector_scope(current_user) # None for non-inspectors
# ── Today's stats ─────────────────────────────────────────────────────
# ── Today's stats (inspector: own work within contracted facilities) ───
base_q = Inspection.query
if is_inspector:
base_q = base_q.filter(Inspection.inspector_id == current_user.id)
if not inspector_facility_ids:
base_q = base_q.filter(False)
else:
base_q = base_q.filter(
Inspection.facility_id.in_(inspector_facility_ids),
Inspection.inspector_id == current_user.id,
)
elif is_customer:
if not customer_facility_ids:
base_q = base_q.filter(False) # no access
@@ -55,12 +62,19 @@ def index():
Inspection.inspection_date < today_end,
).count()
# ── Open issues ────────────────────────────────────────────────────────
# ── Open issues (inspector: all issues in contracted facilities) ───────
open_issues_q = Issue.query.filter(Issue.status.in_(['open', 'in_progress']))
if is_inspector:
open_issues_q = open_issues_q.join(
Inspection, Issue.inspection_id == Inspection.id
).filter(Inspection.inspector_id == current_user.id)
if not inspector_facility_ids:
open_issues_q = open_issues_q.filter(False)
else:
from app.models.facility import Area
open_issues_q = open_issues_q.outerjoin(
Area, Issue.area_id == Area.id
).filter(db.or_(
Issue.facility_id.in_(inspector_facility_ids),
Area.facility_id.in_(inspector_facility_ids)
))
elif is_customer:
if not customer_facility_ids:
open_issues_q = open_issues_q.filter(False)
@@ -83,14 +97,20 @@ def index():
'low': sum(1 for i in open_issues_all if i.severity == 'low'),
}
# ── Average score (last 30 days) ───────────────────────────────────────
# ── Average score (last 30 days, inspector: own work in contracted facilities)
score_q = db.session.query(func.avg(Inspection.overall_score)).filter(
Inspection.status == 'completed',
Inspection.overall_score.isnot(None),
Inspection.inspection_date >= thirty_days_ago,
)
if is_inspector:
score_q = score_q.filter(Inspection.inspector_id == current_user.id)
if not inspector_facility_ids:
score_q = score_q.filter(False)
else:
score_q = score_q.filter(
Inspection.facility_id.in_(inspector_facility_ids),
Inspection.inspector_id == current_user.id,
)
elif is_customer:
if customer_facility_ids:
score_q = score_q.filter(Inspection.facility_id.in_(customer_facility_ids))
@@ -101,7 +121,13 @@ def index():
# ── Recent inspections ─────────────────────────────────────────────────
recent_q = Inspection.query.order_by(Inspection.inspection_date.desc())
if is_inspector:
recent_q = recent_q.filter(Inspection.inspector_id == current_user.id)
if not inspector_facility_ids:
recent_q = recent_q.filter(False)
else:
recent_q = recent_q.filter(
Inspection.facility_id.in_(inspector_facility_ids),
Inspection.inspector_id == current_user.id,
)
elif is_customer:
if customer_facility_ids:
recent_q = recent_q.filter(Inspection.facility_id.in_(customer_facility_ids))
@@ -119,7 +145,13 @@ def index():
follow_up_required=True, status='completed'
).filter(~Inspection.follow_ups.any())
if is_inspector:
followup_q = followup_q.filter(Inspection.inspector_id == current_user.id)
if not inspector_facility_ids:
followup_q = followup_q.filter(False)
else:
followup_q = followup_q.filter(
Inspection.facility_id.in_(inspector_facility_ids),
Inspection.inspector_id == current_user.id,
)
elif is_customer:
if customer_facility_ids:
followup_q = followup_q.filter(Inspection.facility_id.in_(customer_facility_ids))
@@ -140,9 +172,17 @@ def index():
Facility.active == True,
).order_by(Facility.name).all()
# ── SLA summary (open + in_progress issues only) ──────────────────────
# ── SLA summary (open + in_progress issues, scoped) ───────────────────
sla_q = Issue.query.filter(Issue.status.in_(['open', 'in_progress']))
if is_customer and customer_facility_ids:
if is_inspector and inspector_facility_ids:
from app.models.facility import Area
sla_q = sla_q.outerjoin(Area, Issue.area_id == Area.id).filter(
db.or_(
Issue.facility_id.in_(inspector_facility_ids),
Area.facility_id.in_(inspector_facility_ids)
)
)
elif is_customer and customer_facility_ids:
from app.models.facility import Area
sla_q = sla_q.outerjoin(Area, Issue.area_id == Area.id).filter(
db.or_(
@@ -150,9 +190,14 @@ def index():
Area.facility_id.in_(customer_facility_ids)
)
)
all_open_issues = sla_q.all() if not is_customer or customer_facility_ids else []
sla_breached = sum(1 for i in all_open_issues if sla_status(i) == 'breached')
sla_at_risk = sum(1 for i in all_open_issues if sla_status(i) == 'at_risk')
if is_inspector and not inspector_facility_ids:
all_open_issues = []
elif is_customer and not customer_facility_ids:
all_open_issues = []
else:
all_open_issues = sla_q.all()
sla_breached = sum(1 for i in all_open_issues if sla_status(i) == 'breached')
sla_at_risk = sum(1 for i in all_open_issues if sla_status(i) == 'at_risk')
# ── Score trend (last 30 days, grouped by day) ────────────────────────
trend_q = (
@@ -167,7 +212,13 @@ def index():
)
)
if is_inspector:
trend_q = trend_q.filter(Inspection.inspector_id == current_user.id)
if not inspector_facility_ids:
trend_q = trend_q.filter(False)
else:
trend_q = trend_q.filter(
Inspection.facility_id.in_(inspector_facility_ids),
Inspection.inspector_id == current_user.id,
)
elif is_customer:
if customer_facility_ids:
trend_q = trend_q.filter(Inspection.facility_id.in_(customer_facility_ids))
@@ -222,6 +273,10 @@ def index():
# ── Facilities list for the trend-by-facility chart selector ────────────
if is_privileged or is_project_manager:
all_facilities = Facility.query.filter_by(active=True).order_by(Facility.name).all()
elif is_inspector and inspector_facility_ids:
all_facilities = Facility.query.filter(
Facility.id.in_(inspector_facility_ids), Facility.active == True
).order_by(Facility.name).all()
elif is_customer and customer_facility_ids:
all_facilities = Facility.query.filter(
Facility.id.in_(customer_facility_ids), Facility.active == True
app/routes/facilities.py
+6 -1
View File
@@ -7,7 +7,7 @@ from app.models.project import Project
from app.utils.forms import FacilityForm, AreaForm
from app.utils.decorators import supervisor_required, admin_required
from app.utils.audit import log_action, ACTION_CREATE, ACTION_UPDATE, ACTION_DELETE
from app.utils.scope import get_customer_scope
from app.utils.scope import get_customer_scope, get_inspector_scope
bp = Blueprint('facilities', __name__, url_prefix='/facilities')
@@ -21,6 +21,11 @@ def list_facilities():
facilities = Facility.query.filter(
Facility.id.in_(cids), Facility.active == True
).order_by(Facility.name).all()
elif current_user.role == 'inspector':
fids = get_inspector_scope(current_user) or []
facilities = Facility.query.filter(
Facility.id.in_(fids), Facility.active == True
).order_by(Facility.name).all()
else:
facilities = Facility.query.order_by(Facility.name).all()
app/routes/inspections.py
+26 -3
View File
@@ -23,7 +23,7 @@ from app.models.notification import (
EVENT_CUSTOMER_INSPECTION_DONE, EVENT_CUSTOMER_ISSUE_UPDATED,
)
from app.utils.audit import log_action, ACTION_CREATE, ACTION_UPDATE, ACTION_DELETE, ACTION_EXPORT
from app.utils.scope import get_customer_scope
from app.utils.scope import get_customer_scope, get_inspector_scope
bp = Blueprint('inspections', __name__, url_prefix='/inspections')
@@ -201,7 +201,11 @@ def index():
q = Inspection.query.order_by(Inspection.inspection_date.desc())
if current_user.role == 'inspector':
q = q.filter(Inspection.inspector_id == current_user.id)
fids = get_inspector_scope(current_user)
if not fids:
q = q.filter(False)
else:
q = q.filter(Inspection.facility_id.in_(fids))
elif current_user.role == 'customer':
customer_facility_ids = get_customer_scope(current_user)
if not customer_facility_ids:
@@ -223,7 +227,10 @@ def index():
).filter(~Inspection.follow_ups.any())
inspections = q.paginate(page=page, per_page=20, error_out=False)
if current_user.role == 'customer':
if current_user.role == 'inspector':
fids = get_inspector_scope(current_user) or []
facilities = Facility.query.filter(Facility.id.in_(fids), Facility.active == True).order_by(Facility.name).all()
elif current_user.role == 'customer':
cids = get_customer_scope(current_user) or []
facilities = Facility.query.filter(Facility.id.in_(cids), Facility.active == True).order_by(Facility.name).all()
else:
@@ -248,6 +255,15 @@ def start():
templates = InspectionTemplate.query.order_by(InspectionTemplate.name).all()
projects = Project.query.filter_by(active=True).order_by(Project.name).all()
# Scope projects to inspector's assigned contracts
if current_user.role == 'inspector':
from app.models.inspector_assignment import InspectorAssignment
assigned_pids = {
a.project_id for a in
InspectorAssignment.query.filter_by(user_id=current_user.id).all()
}
projects = [p for p in projects if p.id in assigned_pids]
form.template_id.choices = [(t.id, t.name) for t in templates]
form.project_id.choices = [(p.id, p.name) for p in projects]
@@ -284,6 +300,13 @@ def start():
if template is None:
abort(404)
# Inspector facility scope check — prevent crafted POST from selecting
# a facility outside their assigned contracts.
if current_user.role == 'inspector':
fids = get_inspector_scope(current_user)
if not fids or form.facility_id.data not in fids:
abort(403)
if not template.get_form_schema():
flash('This template has no form fields yet. Please build the form in the template editor first.', 'warning')
return redirect(url_for('inspections.start'))
app/routes/issues.py
+22 -10
View File
@@ -17,7 +17,7 @@ from app.utils.forms import IssueForm, IssueUpdateForm
from app.utils.decorators import supervisor_required
from app.utils.notifications import notify, notify_customers_for_facility, notify_by_matrix
from app.utils.audit import log_action, ACTION_CREATE, ACTION_UPDATE, ACTION_DELETE
from app.utils.scope import get_customer_scope
from app.utils.scope import get_customer_scope, get_inspector_scope
from app.utils.sla import sla_status
bp = Blueprint('issues', __name__, url_prefix='/issues')
@@ -80,8 +80,14 @@ def index():
)
if current_user.role == 'inspector':
q = q.filter(db.or_(Issue.assigned_to == current_user.id,
Issue.reported_by == current_user.id))
fids = get_inspector_scope(current_user)
if not fids:
q = q.filter(False)
else:
q = q.filter(db.or_(
Issue.facility_id.in_(fids),
db.and_(Issue.area_id.isnot(None), Area.facility_id.in_(fids)),
))
elif current_user.role == 'customer':
customer_facility_ids = get_customer_scope(current_user)
if not customer_facility_ids:
@@ -138,8 +144,13 @@ def index():
for f in IssueFollower.query.filter_by(user_id=current_user.id).all()
}
# Facilities for the filter dropdown — scoped for customers, full list otherwise
if current_user.role == 'customer':
# Facilities for the filter dropdown — scoped for inspectors/customers
if current_user.role == 'inspector':
fids = get_inspector_scope(current_user) or []
facilities = Facility.query.filter(
Facility.id.in_(fids), Facility.active == True
).order_by(Facility.name).all()
elif current_user.role == 'customer':
facility_ids = get_customer_scope(current_user) or []
facilities = Facility.query.filter(
Facility.id.in_(facility_ids), Facility.active == True
@@ -172,11 +183,12 @@ def view(issue_id):
if issue is None:
abort(404)
if current_user.role == 'inspector' and \
issue.assigned_to != current_user.id and \
issue.reported_by != current_user.id:
flash('Access denied.', 'danger')
return redirect(url_for('issues.index'))
if current_user.role == 'inspector':
fids = get_inspector_scope(current_user)
facility = issue.resolved_facility
if not fids or not facility or facility.id not in fids:
flash('Access denied.', 'danger')
return redirect(url_for('issues.index'))
if current_user.role == 'customer':
cids = get_customer_scope(current_user) or []
facility = issue.resolved_facility
app/templates/auth/inspector_assignments.html
+62
View File
@@ -0,0 +1,62 @@
{% extends "base.html" %}
{% block title %}Contract Assignments — {{ user.display_name }}{% endblock %}
{% block content %}
<div class="d-flex justify-content-between align-items-center mb-4">
<div>
<h2><i class="bi bi-briefcase text-primary me-2"></i>Contract Assignments</h2>
<p class="text-muted mb-0">
Inspector <strong>{{ user.display_name }}</strong> can only access facilities
belonging to the contracts ticked below.
</p>
</div>
<a href="{{ url_for('auth.list_users') }}" class="btn btn-sm btn-outline-secondary">
<i class="bi bi-arrow-left"></i> Back to Users
</a>
</div>
<form method="POST">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<div class="card shadow-sm mb-4">
<div class="card-header bg-light d-flex justify-content-between align-items-center">
<span class="fw-semibold">Active Contracts</span>
<span class="badge bg-primary">{{ assigned_pids|length }} assigned</span>
</div>
{% if projects %}
<div class="list-group list-group-flush">
{% for project in projects %}
<label class="list-group-item list-group-item-action d-flex align-items-center gap-3 py-3">
<input class="form-check-input flex-shrink-0" type="checkbox"
name="project_ids" value="{{ project.id }}"
{{ 'checked' if project.id in assigned_pids }}>
<div>
<div class="fw-semibold">{{ project.name }}</div>
{% if project.description %}
<div class="text-muted small">{{ project.description }}</div>
{% endif %}
<div class="text-muted small">
{{ project.facilities.count() }} facilit{{ 'ies' if project.facilities.count() != 1 else 'y' }}
</div>
</div>
</label>
{% endfor %}
</div>
{% else %}
<div class="card-body text-muted">
No active contracts exist. Create a contract first.
</div>
{% endif %}
</div>
{% if projects %}
<div class="d-flex gap-2">
<button type="submit" class="btn btn-primary">
<i class="bi bi-floppy me-1"></i>Save Assignments
</button>
<a href="{{ url_for('auth.list_users') }}" class="btn btn-outline-secondary">Cancel</a>
</div>
{% endif %}
</form>
{% endblock %}
app/templates/auth/users.html
+20 -1
View File
@@ -24,9 +24,10 @@
<th>Full Name</th>
<th>Email</th>
<th>Role</th>
<th>Contracts</th>
<th>Created</th>
<th>Status</th>
<th width="180">Actions</th>
<th width="220">Actions</th>
</tr>
</thead>
<tbody>
@@ -40,6 +41,18 @@
{{ user.role.replace('_',' ')|title }}
</span>
</td>
<td>
{% if user.role == 'inspector' %}
{% set cnt = inspector_contract_counts.get(user.id, 0) %}
{% if cnt > 0 %}
<span class="badge bg-success">{{ cnt }} contract{{ 's' if cnt != 1 else '' }}</span>
{% else %}
<span class="badge bg-warning text-dark" title="No contracts assigned — inspector sees nothing">None</span>
{% endif %}
{% else %}
<span class="text-muted small">—</span>
{% endif %}
</td>
<td>{{ user.created_at.strftime('%Y-%m-%d') }}</td>
<td>
{% if user.active %}
@@ -52,6 +65,12 @@
<a href="{{ url_for('auth.edit_user', user_id=user.id) }}" class="btn btn-sm btn-outline-primary" title="Edit">
<i class="bi bi-pencil"></i>
</a>
{% if user.role == 'inspector' %}
<a href="{{ url_for('auth.assign_inspector_contracts', user_id=user.id) }}"
class="btn btn-sm btn-outline-secondary" title="Assign contracts">
<i class="bi bi-briefcase"></i>
</a>
{% endif %}
{% if user.id != current_user.id %}
<form method="POST" action="{{ url_for('auth.toggle_active', user_id=user.id) }}" class="d-inline">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
app/utils/scope.py
+53 -14
View File
@@ -1,23 +1,18 @@
"""
app/utils/scope.py
------------------
Customer-scoping utility for the Janitorial QC portal.
Facility-scoping utilities for the Janitorial QC portal.
Provides a single entry-point — get_customer_scope(user) — that returns the
set of facility IDs a customer is authorised to view, derived from their
CustomerAssignment rows.
get_customer_scope(user) -> list[int] | None
Facility IDs a customer may access via CustomerAssignment rows.
Usage (inside any route that serves customer users):
get_inspector_scope(user) -> list[int] | None
Facility IDs an inspector may access via InspectorAssignment rows.
Returns [] (empty list) when the inspector has no contract assignments,
meaning they see nothing (strict mode).
from app.utils.scope import get_customer_scope
facility_ids = get_customer_scope(current_user)
inspections = Inspection.query.filter(
Inspection.facility_id.in_(facility_ids)
).all()
For non-customer roles the function returns None, signalling that no
facility-level scoping is required (full access applies).
For non-customer / non-inspector roles both functions return None, signalling
that no facility-level scoping is required (full access applies).
"""
import logging
@@ -76,4 +71,48 @@ def get_customer_scope(user) -> list[int] | None:
user.id, user.username, sorted(facility_ids),
)
return sorted(facility_ids)
def get_inspector_scope(user) -> list[int] | None:
"""Return the list of facility IDs accessible to a contract-scoped inspector.
Parameters
----------
user : User
The currently authenticated user.
Returns
-------
list[int]
Facility IDs the inspector may access. An empty list means the
inspector has no contract assignments and should see nothing.
None
Returned for non-inspector roles, indicating unrestricted access.
"""
if user.role != 'inspector':
return None
from app.models.inspector_assignment import InspectorAssignment
project_ids = [
a.project_id
for a in InspectorAssignment.query.filter_by(user_id=user.id).all()
]
if not project_ids:
return [] # strict: no assignments = no access
facility_ids = [
f.id for f in Facility.query.filter(
Facility.project_id.in_(project_ids),
Facility.active == True,
).all()
]
logger.debug(
'SCOPE | inspector_scope | user_id=%s username=%s facility_ids=%s',
user.id, user.username, sorted(facility_ids),
)
return sorted(facility_ids)