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

Mar 04 2026: Implement customer's view functionalities - Phase 3

Browse Source
This commit is contained in:
nngo
2026-03-04 13:17:30 -05:00
parent a58a0c8ded
commit 0a7259ae72
8 changed files with 242 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/routes/dashboard.py
+58 -8
View File
@@ -6,6 +6,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 sqlalchemy import func
from datetime import datetime, timedelta
from app.utils.time_utils import now_eastern
@@ -22,13 +23,23 @@ def index():
today_end = today_start + timedelta(days=1)
thirty_days_ago = now - timedelta(days=30)
is_inspector = current_user.role == 'inspector'
is_privileged = current_user.role in ['admin', 'supervisor']
is_inspector = current_user.role == 'inspector'
is_privileged = current_user.role in ['admin', 'supervisor']
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
# ── Today's stats ─────────────────────────────────────────────────────
base_q = Inspection.query
if is_inspector:
base_q = base_q.filter(Inspection.inspector_id == current_user.id)
elif is_customer:
if not customer_facility_ids:
base_q = base_q.filter(False) # no access
else:
base_q = base_q.filter(Inspection.facility_id.in_(customer_facility_ids))
today_inspections = base_q.filter(
Inspection.inspection_date >= today_start,
@@ -47,6 +58,14 @@ def index():
open_issues_q = open_issues_q.join(
Inspection, Issue.inspection_id == Inspection.id
).filter(Inspection.inspector_id == current_user.id)
elif is_customer:
if not customer_facility_ids:
open_issues_q = open_issues_q.filter(False)
else:
from app.models.facility import Area
open_issues_q = open_issues_q.join(
Area, Issue.area_id == Area.id
).filter(Area.facility_id.in_(customer_facility_ids))
open_issues = open_issues_q.count()
# ── Average score (last 30 days) ───────────────────────────────────────
@@ -57,12 +76,22 @@ def index():
)
if is_inspector:
score_q = score_q.filter(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))
else:
score_q = score_q.filter(False)
avg_score = score_q.scalar()
# ── 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)
elif is_customer:
if customer_facility_ids:
recent_q = recent_q.filter(Inspection.facility_id.in_(customer_facility_ids))
else:
recent_q = recent_q.filter(False)
recent_inspections = recent_q.limit(5).all()
# ── System stats (admin/supervisor) ───────────────────────────────────
@@ -70,13 +99,27 @@ def index():
total_templates = InspectionTemplate.query.count() if is_privileged else 0
total_users = User.query.count() if current_user.role == 'admin' else 0
# ── Customer: scoped facilities summary ───────────────────────────────
customer_facilities = []
if is_customer and customer_facility_ids:
customer_facilities = Facility.query.filter(
Facility.id.in_(customer_facility_ids),
Facility.active == True,
).order_by(Facility.name).all()
# ── SLA summary (open + in_progress issues only) ──────────────────────
all_open_issues = Issue.query.filter(Issue.status.in_(['open', 'in_progress'])).all()
sla_q = Issue.query.filter(Issue.status.in_(['open', 'in_progress']))
if is_customer and customer_facility_ids:
from app.models.facility import Area
sla_q = sla_q.join(Area, Issue.area_id == Area.id).filter(
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')
# ── Score trend (last 30 days, grouped by day) ────────────────────────
trend_rows = (
trend_q = (
db.session.query(
func.date(Inspection.inspection_date).label('day'),
func.avg(Inspection.overall_score).label('avg'),
@@ -86,16 +129,22 @@ def index():
Inspection.overall_score.isnot(None),
Inspection.inspection_date >= thirty_days_ago,
)
.group_by(func.date(Inspection.inspection_date))
.order_by(func.date(Inspection.inspection_date))
.all()
)
if is_inspector:
trend_q = trend_q.filter(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))
else:
trend_q = trend_q.filter(False)
trend_rows = trend_q.group_by(func.date(Inspection.inspection_date))\
.order_by(func.date(Inspection.inspection_date)).all()
trend_labels = [str(r.day) for r in trend_rows]
trend_data = [round(float(r.avg), 2) for r in trend_rows]
# ── Facility performance (last 30 days, privileged users only) ─────────
facility_perf = []
if is_privileged:
if is_privileged or is_project_manager:
perf_rows = (
db.session.query(
Facility.name,
@@ -133,4 +182,5 @@ def index():
trend_labels = trend_labels,
trend_data = trend_data,
facility_perf = facility_perf,
customer_facilities = customer_facilities,
)
app/routes/facilities.py
+13 -1
View File
@@ -6,13 +6,20 @@ 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
bp = Blueprint('facilities', __name__, url_prefix='/facilities')
@bp.route('/')
@login_required
def list_facilities():
facilities = Facility.query.order_by(Facility.name).all()
if 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:
facilities = Facility.query.order_by(Facility.name).all()
return render_template('facilities/list.html', facilities=facilities)
@bp.route('/new', methods=['GET', 'POST'])
@@ -47,6 +54,11 @@ def create_facility():
@login_required
def view_facility(facility_id):
facility = Facility.query.get_or_404(facility_id)
if current_user.role == 'customer':
cids = get_customer_scope(current_user) or []
if facility_id not in cids:
flash('Access denied.', 'danger')
return redirect(url_for('facilities.list_facilities'))
areas = facility.areas.order_by(Area.name).all()
return render_template('facilities/view.html', facility=facility, areas=areas)
app/routes/inspections.py
+22 -1
View File
@@ -18,6 +18,7 @@ from app.utils.pdf_export import generate_inspection_pdf
from app.utils.notifications import notify
from app.models.notification import EVENT_INSPECTION_DONE, EVENT_ISSUE_ASSIGNED
from app.utils.audit import log_action, ACTION_CREATE, ACTION_UPDATE, ACTION_DELETE, ACTION_EXPORT
from app.utils.scope import get_customer_scope
bp = Blueprint('inspections', __name__, url_prefix='/inspections')
@@ -180,6 +181,12 @@ def index():
if current_user.role == 'inspector':
q = q.filter(Inspection.inspector_id == current_user.id)
elif current_user.role == 'customer':
customer_facility_ids = get_customer_scope(current_user)
if not customer_facility_ids:
q = q.filter(False)
else:
q = q.filter(Inspection.facility_id.in_(customer_facility_ids))
status_filter = request.args.get('status', '')
facility_filter = request.args.get('facility_id', '')
@@ -189,7 +196,11 @@ def index():
q = q.filter(Inspection.facility_id == int(facility_filter))
inspections = q.paginate(page=page, per_page=20, error_out=False)
facilities = Facility.query.filter_by(active=True).order_by(Facility.name).all()
if 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:
facilities = Facility.query.filter_by(active=True).order_by(Facility.name).all()
return render_template('inspections/list.html',
inspections=inspections,
@@ -428,6 +439,11 @@ def view(inspection_id):
if current_user.role == 'inspector' and inspection.inspector_id != current_user.id:
flash('Access denied.', 'danger')
return redirect(url_for('inspections.index'))
if current_user.role == 'customer':
cids = get_customer_scope(current_user) or []
if inspection.facility_id not in cids:
flash('Access denied.', 'danger')
return redirect(url_for('inspections.index'))
template = inspection.template
form_fields = sorted(template.get_form_schema(),
@@ -533,6 +549,11 @@ def export_pdf(inspection_id):
if current_user.role == 'inspector' and inspection.inspector_id != current_user.id:
flash('Access denied.', 'danger')
return redirect(url_for('inspections.index'))
if current_user.role == 'customer':
cids = get_customer_scope(current_user) or []
if inspection.facility_id not in cids:
flash('Access denied.', 'danger')
return redirect(url_for('inspections.index'))
template = inspection.template
form_fields = sorted(template.get_form_schema(),
app/routes/issues.py
+17
View File
@@ -14,6 +14,7 @@ from app.utils.forms import IssueForm, IssueUpdateForm
from app.utils.decorators import supervisor_required
from app.utils.notifications import notify
from app.utils.audit import log_action, ACTION_CREATE, ACTION_UPDATE, ACTION_DELETE
from app.utils.scope import get_customer_scope
from app.utils.sla import sla_status
bp = Blueprint('issues', __name__, url_prefix='/issues')
@@ -49,6 +50,15 @@ def index():
if current_user.role == 'inspector':
q = q.filter(Issue.assigned_to == current_user.id)
elif current_user.role == 'customer':
from app.models.facility import Area
customer_facility_ids = get_customer_scope(current_user)
if not customer_facility_ids:
q = q.filter(False)
else:
q = q.join(Area, Issue.area_id == Area.id).filter(
Area.facility_id.in_(customer_facility_ids)
)
severity_filter = request.args.get('severity', '')
status_filter = request.args.get('status', '')
@@ -93,6 +103,13 @@ def view(issue_id):
if current_user.role == 'inspector' and issue.assigned_to != current_user.id:
flash('Access denied. You can only view issues assigned to you.', 'danger')
return redirect(url_for('issues.index'))
if current_user.role == 'customer':
from app.models.facility import Area
cids = get_customer_scope(current_user) or []
area = Area.query.get(issue.area_id)
if not area or area.facility_id not in cids:
flash('Access denied.', 'danger')
return redirect(url_for('issues.index'))
form = IssueUpdateForm(obj=issue)
staff = User.query.filter(User.role.in_(['supervisor','inspector'])).order_by(User.username).all()
app/routes/reports.py
+75 -34
View File
@@ -12,6 +12,7 @@ from app.models.facility import Facility, Area
from app.models.issue import Issue
from app.models.user import User
from app.utils.decorators import supervisor_required
from app.utils.scope import get_customer_scope
bp = Blueprint('reports', __name__, url_prefix='/reports')
@@ -36,33 +37,56 @@ def _date_range():
@bp.route('/')
@login_required
@supervisor_required
def index():
# Customers get a scoped view; internal staff need supervisor+ access
if current_user.role not in ['admin', 'supervisor', 'project_manager', 'customer']:
from flask import flash, redirect, url_for
flash('Access denied.', 'danger')
return redirect(url_for('dashboard.index'))
start, end = _date_range()
base = Inspection.query.filter(
# Resolve facility scope for customers
customer_facility_ids = get_customer_scope(current_user) # None = unrestricted
def _scope_insp(q):
if customer_facility_ids is not None:
if not customer_facility_ids:
return q.filter(False)
return q.filter(Inspection.facility_id.in_(customer_facility_ids))
return q
def _scope_issue(q):
if customer_facility_ids is not None:
if not customer_facility_ids:
return q.filter(False)
return q.join(Area, Issue.area_id == Area.id).filter(
Area.facility_id.in_(customer_facility_ids)
)
return q
base = _scope_insp(Inspection.query.filter(
Inspection.inspection_date >= start,
Inspection.inspection_date <= end,
)
))
total_inspections = base.count()
completed = base.filter(Inspection.status == 'completed').count()
# "Flagged" = open or in-progress issues logged within the date range,
# not inspections with status='flagged' (those get completed on submit).
flagged = Issue.query.filter(
flagged = _scope_issue(Issue.query.filter(
Issue.reported_at >= start,
Issue.reported_at <= end,
Issue.status != 'resolved',
).count()
)).count()
avg_score = db.session.query(func.avg(Inspection.overall_score)).filter(
Inspection.inspection_date >= start,
Inspection.inspection_date <= end,
Inspection.status == 'completed',
Inspection.overall_score.isnot(None),
).scalar()
)
avg_score = _scope_insp(avg_score).scalar()
# Scores by facility (for bar chart)
facility_scores = db.session.query(
fac_score_q = db.session.query(
Facility.name,
func.avg(Inspection.overall_score).label('avg_score'),
func.count(Inspection.id).label('count'),
@@ -72,11 +96,16 @@ def index():
Inspection.inspection_date <= end,
Inspection.status == 'completed',
Inspection.overall_score.isnot(None),
).group_by(Facility.id, Facility.name)\
.order_by(func.avg(Inspection.overall_score).desc()).all()
)
if customer_facility_ids is not None:
fac_score_q = fac_score_q.filter(
Facility.id.in_(customer_facility_ids) if customer_facility_ids else False
)
facility_scores = fac_score_q.group_by(Facility.id, Facility.name)\
.order_by(func.avg(Inspection.overall_score).desc()).all()
# Score trend — daily averages (line chart)
daily_scores = db.session.query(
daily_q = db.session.query(
func.date(Inspection.inspection_date).label('day'),
func.avg(Inspection.overall_score).label('avg'),
func.count(Inspection.id).label('count'),
@@ -85,47 +114,50 @@ def index():
Inspection.inspection_date <= end,
Inspection.status == 'completed',
Inspection.overall_score.isnot(None),
).group_by(func.date(Inspection.inspection_date))\
.order_by(func.date(Inspection.inspection_date)).all()
)
daily_scores = _scope_insp(daily_q).group_by(func.date(Inspection.inspection_date))\
.order_by(func.date(Inspection.inspection_date)).all()
# Issue breakdown by severity
issue_severity = db.session.query(
issue_severity = _scope_issue(db.session.query(
Issue.severity,
func.count(Issue.id).label('count'),
).filter(
Issue.reported_at >= start,
Issue.reported_at <= end,
).group_by(Issue.severity).all()
)).group_by(Issue.severity).all()
# Issue status breakdown
issue_status = db.session.query(
issue_status = _scope_issue(db.session.query(
Issue.status,
func.count(Issue.id).label('count'),
).filter(
Issue.reported_at >= start,
Issue.reported_at <= end,
).group_by(Issue.status).all()
)).group_by(Issue.status).all()
# Top inspectors by inspection count
top_inspectors = db.session.query(
User.username,
func.count(Inspection.id).label('count'),
func.avg(Inspection.overall_score).label('avg_score'),
).join(Inspection, User.id == Inspection.inspector_id)\
.filter(
Inspection.inspection_date >= start,
Inspection.inspection_date <= end,
Inspection.status == 'completed',
).group_by(User.id, User.username)\
.order_by(func.count(Inspection.id).desc()).limit(10).all()
# Top inspectors by inspection count (hidden for customer role)
top_inspectors = []
if current_user.role != 'customer':
top_inspectors = db.session.query(
User.username,
func.count(Inspection.id).label('count'),
func.avg(Inspection.overall_score).label('avg_score'),
).join(Inspection, User.id == Inspection.inspector_id)\
.filter(
Inspection.inspection_date >= start,
Inspection.inspection_date <= end,
Inspection.status == 'completed',
).group_by(User.id, User.username)\
.order_by(func.count(Inspection.id).desc()).limit(10).all()
# Recent issues (critical/high)
critical_issues = Issue.query.filter(
# Recent issues (critical/high) — scoped for customers
critical_issues = _scope_issue(Issue.query.filter(
Issue.severity.in_(['critical', 'high']),
Issue.status != 'resolved',
Issue.reported_at >= start,
Issue.reported_at <= end,
).order_by(Issue.reported_at.desc()).limit(10).all()
)).order_by(Issue.reported_at.desc()).limit(10).all()
return render_template('reports/index.html',
start=start, end=end,
@@ -146,9 +178,18 @@ def index():
@bp.route('/facility/<int:facility_id>')
@login_required
@supervisor_required
def facility_report(facility_id):
if current_user.role not in ['admin', 'supervisor', 'project_manager', 'customer']:
from flask import flash, redirect, url_for
flash('Access denied.', 'danger')
return redirect(url_for('dashboard.index'))
facility = Facility.query.get_or_404(facility_id)
if current_user.role == 'customer':
cids = get_customer_scope(current_user) or []
if facility_id not in cids:
from flask import flash, redirect, url_for
flash('Access denied.', 'danger')
return redirect(url_for('reports.index'))
start, end = _date_range()
inspections = Inspection.query.filter(