import csv import io import logging from datetime import datetime, timedelta from app.utils.time_utils import now_eastern from flask import (Blueprint, render_template, request, Response, stream_with_context, abort) from flask_login import login_required, current_user from sqlalchemy import func, literal_column from sqlalchemy.orm import joinedload from app import db from app.models.inspection import Inspection, InspectionTemplate 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 from app.utils.audit import log_action, ACTION_EXPORT bp = Blueprint('reports', __name__, url_prefix='/reports') logger = logging.getLogger(__name__) def _fmt_minutes(mins): """Format an integer minute value as '45m' or '1h 30m'. Returns None if input is None.""" if mins is None: return None mins = int(round(float(mins))) if mins >= 60: return f'{mins // 60}h {mins % 60}m' return f'{mins}m' def _date_range(): """Parse ?start= and ?end= query params; default to last 30 days.""" end_default = now_eastern() start_default = end_default - timedelta(days=30) try: start = datetime.strptime(request.args.get('start', ''), '%Y-%m-%d') except ValueError: start = start_default try: end = datetime.strptime(request.args.get('end', ''), '%Y-%m-%d') end = end.replace(hour=23, minute=59, second=59) except ValueError: end = end_default return start, end # ── Overview dashboard ──────────────────────────────────────────────────────── @bp.route('/') @login_required 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']: from flask import flash, redirect, url_for flash('Access denied.', 'danger') return redirect(url_for('dashboard.index')) start, end = _date_range() # 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' # Inspector filter — admin / director / project_manager only inspector_filter = None if current_user.role in ('admin', 'director', 'project_manager'): inspector_filter = request.args.get('inspector_id', type=int) or None # Pre-compute inspection ID sets used by _scope_issue to avoid join conflicts. inspector_inspection_ids = [] # own inspections (inspector role) filter_inspection_ids = None # filtered inspector's inspections (admin/dir/PM) if is_inspector: inspector_inspection_ids = [ row[0] for row in db.session.query(Inspection.id) .filter(Inspection.inspector_id == current_user.id) .all() ] elif inspector_filter: filter_inspection_ids = [ row[0] for row in db.session.query(Inspection.id) .filter(Inspection.inspector_id == inspector_filter) .all() ] def _scope_insp(q): if is_inspector: return q.filter(Inspection.inspector_id == current_user.id) if inspector_filter: q = q.filter(Inspection.inspector_id == inspector_filter) 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 is_inspector: if not inspector_inspection_ids: return q.filter(False) return q.filter(Issue.inspection_id.in_(inspector_inspection_ids)) if filter_inspection_ids is not None: if not filter_inspection_ids: return q.filter(False) return q.filter(Issue.inspection_id.in_(filter_inspection_ids)) if customer_facility_ids is not None: if not customer_facility_ids: return q.filter(False) return q.outerjoin(Area, Issue.area_id == Area.id).filter( db.or_( Issue.facility_id.in_(customer_facility_ids), 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 = _scope_issue(Issue.query.filter( Issue.reported_at >= start, Issue.reported_at <= end, Issue.status != 'resolved', )).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), ) avg_score = _scope_insp(avg_score).scalar() # Scores by facility (for bar chart) fac_score_q = db.session.query( Facility.name, func.avg(Inspection.overall_score).label('avg_score'), func.count(Inspection.id).label('count'), ).join(Inspection, Facility.id == Inspection.facility_id)\ .filter( Inspection.inspection_date >= start, Inspection.inspection_date <= end, Inspection.status == 'completed', Inspection.overall_score.isnot(None), ) if inspector_filter: fac_score_q = fac_score_q.filter(Inspection.inspector_id == inspector_filter) 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_q = db.session.query( func.date(Inspection.inspection_date).label('day'), func.avg(Inspection.overall_score).label('avg'), func.count(Inspection.id).label('count'), ).filter( Inspection.inspection_date >= start, Inspection.inspection_date <= end, Inspection.status == 'completed', Inspection.overall_score.isnot(None), ) 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 = _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() # Issue status breakdown 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() # Top inspectors — scoped per role: # customer → inspectors who worked at the customer's assigned facilities # inspector → own row only # others → org-wide top 10 (filtered by inspector_filter if set) top_insp_q = db.session.query( User.full_name, 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', ) top_insp_q = _scope_insp(top_insp_q) top_inspectors = top_insp_q.group_by(User.id, User.full_name, User.username)\ .order_by(func.count(Inspection.id).desc()).limit(10).all() # 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() inspectors = [] if current_user.role in ('admin', 'director', 'project_manager'): inspectors = User.query.filter_by(role='inspector', active=True)\ .order_by(User.full_name, User.username).all() return render_template('reports/index.html', start=start, end=end, total_inspections=total_inspections, completed=completed, flagged=flagged, avg_score=round(float(avg_score), 2) if avg_score else None, facility_scores=[{'name': r.name, 'avg_score': round(float(r.avg_score), 2), 'count': r.count} for r in facility_scores], daily_scores=[{'day': str(r.day), 'avg': round(float(r.avg), 2), 'count': r.count} for r in daily_scores], issue_severity=[{'severity': r.severity, 'count': r.count} for r in issue_severity], issue_status=[{'status': r.status, 'count': r.count} for r in issue_status], top_inspectors=[{'display_name': (r.full_name.strip() if r.full_name and r.full_name.strip() else r.username), 'count': r.count, 'avg_score': round(float(r.avg_score), 2) if r.avg_score else None} for r in top_inspectors], critical_issues=critical_issues, is_inspector=is_inspector, inspectors=inspectors, inspector_filter=inspector_filter, ) # ── Facility detail report ──────────────────────────────────────────────────── @bp.route('/facility/') @login_required def facility_report(facility_id): if current_user.role not in ['admin', 'director', 'project_manager', 'inspector', 'customer']: from flask import flash, redirect, url_for flash('Access denied.', 'danger') return redirect(url_for('dashboard.index')) facility = db.session.get(Facility, facility_id) if facility is None: abort(404) 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')) if current_user.role == '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( facility_id=facility_id, inspector_id=current_user.id, ).first() if not has_access: from flask import flash, redirect, url_for flash('Access denied. You have not conducted inspections at this facility.', 'danger') return redirect(url_for('reports.index')) start, end = _date_range() inspections = Inspection.query.filter( Inspection.facility_id == facility_id, Inspection.inspection_date >= start, Inspection.inspection_date <= end, ).order_by(Inspection.inspection_date.desc()).all() area_scores = db.session.query( Area.name, func.avg(Inspection.overall_score).label('avg_score'), func.count(Inspection.id).label('count'), ).join(Inspection, Area.id == Inspection.area_id)\ .filter( Inspection.facility_id == facility_id, Inspection.inspection_date >= start, Inspection.inspection_date <= end, Inspection.status == 'completed', ).group_by(Area.id, Area.name).all() open_issues = Issue.query.join(Area)\ .filter(Area.facility_id == facility_id, Issue.status != 'resolved')\ .order_by(Issue.severity.desc()).all() return render_template('reports/facility.html', facility=facility, inspections=inspections, area_scores=area_scores, open_issues=open_issues, start=start, end=end) # ── Facility Scorecard ──────────────────────────────────────────────────────── @bp.route('/facility//scorecard') @login_required 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']: from flask import flash, redirect, url_for flash('Access denied.', 'danger') return redirect(url_for('dashboard.index')) facility = db.session.get(Facility, facility_id) if facility is None: abort(404) 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')) if current_user.role == 'inspector': has_access = Inspection.query.filter_by( facility_id=facility_id, inspector_id=current_user.id, ).first() if not has_access: from flask import flash, redirect, url_for flash('Access denied. You have not conducted inspections at this facility.', 'danger') return redirect(url_for('reports.index')) from app.utils.sla import sla_status, SLA_HOURS from datetime import timedelta now = now_eastern() days = request.args.get('days', 90, type=int) if days not in (30, 60, 90, 180, 365): days = 90 start = now - timedelta(days=days) # ── Score trend (daily) ─────────────────────────────────────────────── trend_rows = db.session.query( func.date(Inspection.inspection_date).label('day'), func.avg(Inspection.overall_score).label('avg'), func.count(Inspection.id).label('count'), ).filter( Inspection.facility_id == facility_id, Inspection.inspection_date >= start, Inspection.status == 'completed', Inspection.overall_score.isnot(None), ).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] # ── KPI summary ─────────────────────────────────────────────────────── all_insp = Inspection.query.filter( Inspection.facility_id == facility_id, Inspection.inspection_date >= start, ).all() completed_insp = [i for i in all_insp if i.status == 'completed'] avg_score = ( round(sum(float(i.overall_score) for i in completed_insp if i.overall_score is not None) / len([i for i in completed_insp if i.overall_score is not None]), 2) if any(i.overall_score for i in completed_insp) else None ) # ── Area scores ─────────────────────────────────────────────────────── area_scores = db.session.query( Area.name, func.avg(Inspection.overall_score).label('avg'), func.count(Inspection.id).label('count'), ).join(Inspection, Area.id == Inspection.area_id) .filter( Inspection.facility_id == facility_id, Inspection.inspection_date >= start, Inspection.status == 'completed', Inspection.overall_score.isnot(None), ).group_by(Area.id, Area.name) .order_by(func.avg(Inspection.overall_score).desc()).all() # ── Open issues ─────────────────────────────────────────────────────── open_issues = Issue.query.join(Area) .filter(Area.facility_id == facility_id, Issue.status != 'resolved') .order_by(Issue.reported_at.desc()).all() # SLA compliance for closed issues in window closed_issues = Issue.query.join(Area).filter( Area.facility_id == facility_id, Issue.status == 'resolved', Issue.reported_at >= start, ).all() sla_met = sum(1 for i in closed_issues if i.resolved_at and i.reported_at and (i.resolved_at - i.reported_at).total_seconds() / 3600 <= SLA_HOURS.get(i.severity, 9999)) sla_total = len(closed_issues) sla_pct = round(sla_met / sla_total * 100, 1) if sla_total else None # Issue severity breakdown sev_counts = {} for sev in ('critical', 'high', 'medium', 'low'): sev_counts[sev] = Issue.query.join(Area).filter( Area.facility_id == facility_id, Issue.severity == sev, Issue.status != 'resolved', ).count() # Pending verification count pending_verification = Issue.query.join(Area).filter( Area.facility_id == facility_id, Issue.status == 'pending_verification', ).count() # Follow-up required inspections followup_required = Inspection.query.filter( Inspection.facility_id == facility_id, Inspection.follow_up_required == True, ).order_by(Inspection.inspection_date.desc()).limit(10).all() return render_template('reports/scorecard.html', facility = facility, days = days, start = start, now = now, total_inspections = len(all_insp), completed_insp = len(completed_insp), avg_score = avg_score, trend_labels = trend_labels, trend_data = trend_data, area_scores = area_scores, open_issues = open_issues, sla_pct = sla_pct, sla_met = sla_met, sla_total = sla_total, sev_counts = sev_counts, pending_verification = pending_verification, followup_required = followup_required, ) # ── CSV export ──────────────────────────────────────────────────────────────── @bp.route('/export/inspections') @login_required @supervisor_required def export_inspections(): start, end = _date_range() logger.info( 'REPORTS | export_inspections | user=%s | range=%s to %s', current_user.username, start.strftime('%Y-%m-%d'), end.strftime('%Y-%m-%d'), ) log_action( ACTION_EXPORT, 'Inspection', None, 'CSV Export', f'range={start.strftime("%Y-%m-%d")} to {end.strftime("%Y-%m-%d")}', ) rows = db.session.query( Inspection.id, Inspection.inspection_date, Facility.name.label('facility'), Area.name.label('area'), User.username.label('inspector'), InspectionTemplate.name.label('template'), Inspection.overall_score, Inspection.status, Inspection.completed_at, Inspection.notes, ).join(Facility, Inspection.facility_id == Facility.id)\ .outerjoin(Area, Inspection.area_id == Area.id)\ .join(User, Inspection.inspector_id == User.id)\ .join(InspectionTemplate, Inspection.template_id == InspectionTemplate.id)\ .filter( Inspection.inspection_date >= start, Inspection.inspection_date <= end, ).order_by(Inspection.inspection_date.desc()).all() def generate(): buf = io.StringIO() writer = csv.writer(buf) writer.writerow(['ID','Date','Facility','Area','Inspector','Template', 'Score','Status','Completed At','Notes']) yield buf.getvalue(); buf.seek(0); buf.truncate() for r in rows: writer.writerow([ r.id, r.inspection_date.strftime('%Y-%m-%d %H:%M') if r.inspection_date else '', r.facility, r.area or '', r.inspector, r.template, r.overall_score or '', r.status, r.completed_at.strftime('%Y-%m-%d %H:%M') if r.completed_at else '', (r.notes or '').replace('\n', ' '), ]) yield buf.getvalue(); buf.seek(0); buf.truncate() filename = f"inspections_{start.strftime('%Y%m%d')}_{end.strftime('%Y%m%d')}.csv" return Response( stream_with_context(generate()), mimetype='text/csv', headers={'Content-Disposition': f'attachment; filename="{filename}"'} ) @bp.route('/export/issues') @login_required @supervisor_required def export_issues(): start, end = _date_range() logger.info( 'REPORTS | export_issues | user=%s | range=%s to %s', current_user.username, start.strftime('%Y-%m-%d'), end.strftime('%Y-%m-%d'), ) log_action( ACTION_EXPORT, 'Issue', None, 'CSV Export', f'range={start.strftime("%Y-%m-%d")} to {end.strftime("%Y-%m-%d")}', ) rows = db.session.query( Issue.id, Issue.reported_at, Facility.name.label('facility'), Area.name.label('area'), Issue.severity, Issue.description, Issue.status, Issue.resolved_at, User.username.label('assigned_to'), ).outerjoin(Area, Issue.area_id == Area.id)\ .outerjoin(Facility, db.or_( Facility.id == Area.facility_id, Facility.id == Issue.facility_id ))\ .outerjoin(User, Issue.assigned_to == User.id)\ .filter( Issue.reported_at >= start, Issue.reported_at <= end, ).order_by(Issue.reported_at.desc()).all() def generate(): buf = io.StringIO() writer = csv.writer(buf) writer.writerow(['ID','Reported At','Facility','Area','Severity', 'Description','Status','Resolved At','Assigned To']) yield buf.getvalue(); buf.seek(0); buf.truncate() for r in rows: writer.writerow([ r.id, r.reported_at.strftime('%Y-%m-%d %H:%M') if r.reported_at else '', r.facility, r.area, r.severity, r.description.replace('\n', ' '), r.status, r.resolved_at.strftime('%Y-%m-%d %H:%M') if r.resolved_at else '', r.assigned_to or '', ]) yield buf.getvalue(); buf.seek(0); buf.truncate() filename = f"issues_{start.strftime('%Y%m%d')}_{end.strftime('%Y%m%d')}.csv" return Response( stream_with_context(generate()), mimetype='text/csv', headers={'Content-Disposition': f'attachment; filename="{filename}"'} ) # ── Inspector Performance ───────────────────────────────────────────────────── def _build_inspector_stats(start, end): """Return (inspector_stats, team_avg_score) for the given date range. inspector_stats is a list of dicts sorted by avg_score desc. team_avg_score is the mean of all inspectors' avg_score values (or None). """ total_rows = db.session.query( Inspection.inspector_id, func.count(Inspection.id).label('total'), func.count(func.distinct(Inspection.facility_id)).label('facilities'), ).filter( Inspection.inspection_date >= start, Inspection.inspection_date <= end, ).group_by(Inspection.inspector_id).all() completed_rows = db.session.query( Inspection.inspector_id, func.count(Inspection.id).label('completed'), func.avg(Inspection.overall_score).label('avg_score'), func.avg( func.timestampdiff( literal_column('MINUTE'), Inspection.inspection_date, Inspection.completed_at, ) ).label('avg_minutes'), ).filter( Inspection.inspection_date >= start, Inspection.inspection_date <= end, Inspection.status == 'completed', Inspection.completed_at.isnot(None), ).group_by(Inspection.inspector_id).all() issue_rows = db.session.query( Inspection.inspector_id, func.count(Issue.id).label('count'), ).join(Issue, Issue.inspection_id == Inspection.id)\ .filter( Inspection.inspection_date >= start, Inspection.inspection_date <= end, ).group_by(Inspection.inspector_id).all() followup_rows = db.session.query( Inspection.inspector_id, func.count(Inspection.id).label('count'), ).filter( Inspection.inspection_date >= start, Inspection.inspection_date <= end, Inspection.follow_up_required == True, ).group_by(Inspection.inspector_id).all() total_map = {r.inspector_id: {'total': r.total, 'facilities': r.facilities} for r in total_rows} completed_map = { r.inspector_id: { 'completed': r.completed, 'avg_score': r.avg_score, 'avg_minutes': r.avg_minutes, } for r in completed_rows } issues_map = {r.inspector_id: r.count for r in issue_rows} followups_map = {r.inspector_id: r.count for r in followup_rows} all_ids = set(total_map.keys()) active_inspectors = ( User.query .filter(User.id.in_(all_ids), User.active == True, User.role == 'inspector') .order_by(User.full_name, User.username) .all() ) if all_ids else [] inspector_stats = [] for u in active_inspectors: t = total_map.get(u.id, {}) c = completed_map.get(u.id, {}) tot = t.get('total', 0) comp = c.get('completed', 0) avg = round(float(c['avg_score']), 1) if c.get('avg_score') else None avg_mins = round(float(c['avg_minutes'])) if c.get('avg_minutes') else None inspector_stats.append({ 'id': u.id, 'display_name': u.display_name, 'total': tot, 'completed': comp, 'completion_rate': round(comp / tot * 100) if tot else 0, 'avg_score': avg, 'avg_minutes': avg_mins, 'avg_time': _fmt_minutes(avg_mins), 'issues_flagged': issues_map.get(u.id, 0), 'follow_ups': followups_map.get(u.id, 0), 'facilities': t.get('facilities', 0), }) inspector_stats.sort(key=lambda x: (x['avg_score'] is None, -(x['avg_score'] or 0))) _scores = [s['avg_score'] for s in inspector_stats if s['avg_score'] is not None] team_avg_score = round(sum(_scores) / len(_scores), 1) if _scores else None for s in inspector_stats: if s['avg_score'] is not None and team_avg_score is not None: s['vs_avg'] = round(s['avg_score'] - team_avg_score, 1) else: s['vs_avg'] = None return inspector_stats, team_avg_score @bp.route('/inspector-performance') @login_required @supervisor_required def inspector_performance(): """KPI & performance dashboard for all inspectors — admin/director only.""" start, end = _date_range() selected_id = request.args.get('inspector_id', type=int) inspector_stats, team_avg_score = _build_inspector_stats(start, end) # ── Individual drill-down ───────────────────────────────────────────── selected_inspector = None selected_kpis = None trend_data = [] recent_inspections = [] if selected_id: selected_inspector = db.session.get(User, selected_id) if selected_inspector and selected_inspector.role == 'inspector': selected_kpis = next((s for s in inspector_stats if s['id'] == selected_id), None) trend_rows = db.session.query( func.date(Inspection.inspection_date).label('day'), func.avg(Inspection.overall_score).label('avg'), func.count(Inspection.id).label('count'), ).filter( Inspection.inspector_id == selected_id, Inspection.inspection_date >= start, 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() trend_data = [ {'day': str(r.day), 'avg': round(float(r.avg), 2), 'count': r.count} for r in trend_rows ] recent_inspections = ( Inspection.query .options( joinedload(Inspection.facility), joinedload(Inspection.area), joinedload(Inspection.template), ) .filter( Inspection.inspector_id == selected_id, Inspection.inspection_date >= start, Inspection.inspection_date <= end, ) .order_by(Inspection.inspection_date.desc()) .limit(20) .all() ) return render_template('reports/inspector_performance.html', start=start, end=end, inspector_stats=inspector_stats, team_avg_score=team_avg_score, selected_inspector=selected_inspector, selected_kpis=selected_kpis, trend_data=trend_data, recent_inspections=recent_inspections, selected_id=selected_id, ) @bp.route('/export/inspector-performance') @login_required @supervisor_required def export_inspector_performance(): """Download inspector performance summary as an Excel workbook (.xlsx).""" try: from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter except ImportError: abort(501) start, end = _date_range() selected_id = request.args.get('inspector_id', type=int) logger.info( 'REPORTS | export_inspector_performance | user=%s | range=%s to %s', current_user.username, start.strftime('%Y-%m-%d'), end.strftime('%Y-%m-%d'), ) log_action( ACTION_EXPORT, 'InspectorPerformance', None, 'Excel Export', f'range={start.strftime("%Y-%m-%d")} to {end.strftime("%Y-%m-%d")}', ) inspector_stats, team_avg_score = _build_inspector_stats(start, end) # ── Inspection detail rows ──────────────────────────────────────────── detail_q = db.session.query( Inspection.id, Inspection.inspection_date, Inspection.completed_at, Inspection.overall_score, Inspection.status, Inspection.follow_up_required, User.full_name.label('inspector_name'), User.username.label('inspector_username'), Facility.name.label('facility'), Area.name.label('area'), InspectionTemplate.name.label('template'), ).join(User, Inspection.inspector_id == User.id)\ .join(Facility, Inspection.facility_id == Facility.id)\ .outerjoin(Area, Inspection.area_id == Area.id)\ .join(InspectionTemplate, Inspection.template_id == InspectionTemplate.id)\ .filter( Inspection.inspection_date >= start, Inspection.inspection_date <= end, User.role == 'inspector', ) if selected_id: detail_q = detail_q.filter(Inspection.inspector_id == selected_id) detail_rows = detail_q.order_by(User.full_name, Inspection.inspection_date.desc()).all() # ── Build workbook ──────────────────────────────────────────────────── wb = Workbook() # ── Shared styles ───────────────────────────────────────────────────── hdr_font = Font(bold=True, color='FFFFFF', size=11) hdr_fill = PatternFill('solid', fgColor='1A56DB') # blue sub_fill = PatternFill('solid', fgColor='E8F0FE') # light blue stripe good_fill = PatternFill('solid', fgColor='D1FAE5') # green warn_fill = PatternFill('solid', fgColor='FEF3C7') # yellow bad_fill = PatternFill('solid', fgColor='FEE2E2') # red center = Alignment(horizontal='center', vertical='center', wrap_text=False) left = Alignment(horizontal='left', vertical='center') thin = Side(style='thin', color='D1D5DB') cell_border = Border(left=thin, right=thin, top=thin, bottom=thin) def _apply_header(ws, headers, col_widths): ws.row_dimensions[1].height = 22 for col_idx, (text, width) in enumerate(zip(headers, col_widths), start=1): cell = ws.cell(row=1, column=col_idx, value=text) cell.font = hdr_font cell.fill = hdr_fill cell.alignment = center cell.border = cell_border ws.column_dimensions[get_column_letter(col_idx)].width = width def _score_fill(score): if score is None: return None if score >= 90: return good_fill if score >= 70: return warn_fill return bad_fill # ── Sheet 1: Performance Summary ────────────────────────────────────── ws1 = wb.active ws1.title = 'Performance Summary' ws1.freeze_panes = 'A2' period_label = f'{start.strftime("%b %d, %Y")} — {end.strftime("%b %d, %Y")}' ws1['A1'] = f'Inspector Performance Summary | {period_label}' ws1.merge_cells('A1:K1') title_cell = ws1['A1'] title_cell.font = Font(bold=True, size=13, color='1A56DB') title_cell.alignment = left ws1.row_dimensions[1].height = 28 summary_headers = [ 'Inspector', 'Total\nInspections', 'Completed', 'Completion\nRate (%)', 'Avg Score\n(%)', 'vs. Team\nAvg', 'Avg Time', 'Issues\nFlagged', 'Follow-\nUps', 'Facilities', 'Team Avg\nScore (%)', ] summary_widths = [24, 13, 12, 15, 13, 13, 11, 13, 10, 12, 14] hdr_row = 2 ws1.row_dimensions[hdr_row].height = 34 for col_idx, (text, width) in enumerate(zip(summary_headers, summary_widths), start=1): cell = ws1.cell(row=hdr_row, column=col_idx, value=text) cell.font = hdr_font cell.fill = hdr_fill cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True) cell.border = cell_border ws1.column_dimensions[get_column_letter(col_idx)].width = width for row_idx, s in enumerate(inspector_stats, start=3): stripe = sub_fill if row_idx % 2 == 0 else None row_data = [ s['display_name'], s['total'], s['completed'], s['completion_rate'], s['avg_score'], s['vs_avg'], s['avg_time'] or '—', s['issues_flagged'], s['follow_ups'], s['facilities'], team_avg_score, ] for col_idx, value in enumerate(row_data, start=1): cell = ws1.cell(row=row_idx, column=col_idx, value=value) cell.border = cell_border cell.alignment = center if col_idx > 1 else left if stripe: cell.fill = stripe # Score colour bands score_cell = ws1.cell(row=row_idx, column=5) score_fill = _score_fill(s['avg_score']) if score_fill: score_cell.fill = score_fill score_cell.font = Font(bold=True) # vs. avg colour vs_cell = ws1.cell(row=row_idx, column=6) if s['vs_avg'] is not None: vs_cell.fill = good_fill if s['vs_avg'] >= 0 else bad_fill vs_cell.font = Font(bold=True) # Completion rate colour cr_cell = ws1.cell(row=row_idx, column=4) rate = s['completion_rate'] cr_cell.fill = (good_fill if rate >= 90 else warn_fill if rate >= 70 else bad_fill) ws1.row_dimensions[row_idx].height = 18 # Totals row if inspector_stats: tot_row = len(inspector_stats) + 3 ws1.row_dimensions[tot_row].height = 20 total_cell = ws1.cell(row=tot_row, column=1, value='TEAM TOTAL / AVG') total_cell.font = Font(bold=True) total_cell.border = cell_border total_cell.alignment = left total_inspections = sum(s['total'] for s in inspector_stats) total_completed = sum(s['completed'] for s in inspector_stats) total_issues = sum(s['issues_flagged'] for s in inspector_stats) total_followups = sum(s['follow_ups'] for s in inspector_stats) team_cr = round(total_completed / total_inspections * 100) if total_inspections else 0 for col_idx, value in enumerate([ total_inspections, total_completed, team_cr, team_avg_score, None, None, total_issues, total_followups, None, None, ], start=2): cell = ws1.cell(row=tot_row, column=col_idx, value=value) cell.font = Font(bold=True) cell.border = cell_border cell.alignment = center cell.fill = PatternFill('solid', fgColor='DBEAFE') # ── Sheet 2: Inspection Detail ──────────────────────────────────────── ws2 = wb.create_sheet('Inspection Detail') ws2.freeze_panes = 'A2' detail_headers = [ 'ID', 'Inspector', 'Date', 'Completed At', 'Facility', 'Area', 'Template', 'Score (%)', 'Status', 'Follow-up\nRequired', 'Duration\n(min)', ] detail_widths = [7, 22, 18, 18, 28, 20, 22, 12, 18, 14, 13] _apply_header(ws2, detail_headers, detail_widths) status_map = { 'completed': 'Submitted', 'in_progress': 'In Progress', 'flagged': 'Flagged', } for row_idx, r in enumerate(detail_rows, start=2): duration = None if r.completed_at and r.inspection_date: duration = round((r.completed_at - r.inspection_date).total_seconds() / 60) inspector_label = (r.inspector_name.strip() if r.inspector_name and r.inspector_name.strip() else r.inspector_username) row_data = [ r.id, inspector_label, r.inspection_date.strftime('%Y-%m-%d %H:%M') if r.inspection_date else '', r.completed_at.strftime('%Y-%m-%d %H:%M') if r.completed_at else '', r.facility, r.area or '', r.template, r.overall_score, status_map.get(r.status, r.status), 'Yes' if r.follow_up_required else 'No', duration, ] stripe = sub_fill if row_idx % 2 == 0 else None for col_idx, value in enumerate(row_data, start=1): cell = ws2.cell(row=row_idx, column=col_idx, value=value) cell.border = cell_border cell.alignment = center if col_idx != 2 else left if stripe: cell.fill = stripe score_cell = ws2.cell(row=row_idx, column=8) sf = _score_fill(r.overall_score) if sf: score_cell.fill = sf score_cell.font = Font(bold=True) ws2.row_dimensions[row_idx].height = 16 # ── Serialize & return ──────────────────────────────────────────────── buf = io.BytesIO() wb.save(buf) buf.seek(0) suffix = f'_{selected_id}' if selected_id else '' filename = ( f'inspector_performance{suffix}_' f'{start.strftime("%Y%m%d")}_{end.strftime("%Y%m%d")}.xlsx' ) return Response( buf.read(), mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', headers={'Content-Disposition': f'attachment; filename="{filename}"'}, )