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}"'}, ) # ── Issues Aging ────────────────────────────────────────────────────────────── def _age_bucket(age_h): if age_h < 24: return '<24h' if age_h < 72: return '1–3 days' if age_h < 168: return '3–7 days' if age_h < 720: return '1–4 weeks' return '>4 weeks' AGING_BUCKETS = ['<24h', '1–3 days', '3–7 days', '1–4 weeks', '>4 weeks'] def _load_open_issues_scoped(customer_facility_ids, severity_filter, facility_id_filter): """Return open issues with area+facility+assigned_user loaded, filters applied.""" q = Issue.query.options( joinedload(Issue.area).joinedload(Area.facility), joinedload(Issue.assigned_user), ).filter(Issue.status != 'resolved') if customer_facility_ids is not None: if not customer_facility_ids: return [] q = 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)) ) if severity_filter: q = q.filter(Issue.severity == severity_filter) issues = q.order_by(Issue.reported_at).all() if facility_id_filter: issues = [i for i in issues if (i.facility_id == facility_id_filter) or (i.area and i.area.facility_id == facility_id_filter)] return issues @bp.route('/issues-aging') @login_required def issues_aging(): from app.utils.sla import sla_status now = now_eastern() customer_facility_ids = get_customer_scope(current_user) severity_filter = request.args.get('severity', '') facility_id_filter = request.args.get('facility_id', type=int) issues = _load_open_issues_scoped(customer_facility_ids, severity_filter, facility_id_filter) buckets = {b: [] for b in AGING_BUCKETS} for issue in issues: age_h = (now - issue.reported_at).total_seconds() / 3600 buckets[_age_bucket(age_h)].append({ 'issue': issue, 'age_h': round(age_h, 1), 'sla': sla_status(issue), }) sla_breached = sum(1 for i in issues if sla_status(i) == 'breached') sla_at_risk = sum(1 for i in issues if sla_status(i) == 'at_risk') if customer_facility_ids is not None: facilities = (Facility.query.filter(Facility.id.in_(customer_facility_ids), Facility.active == True) .order_by(Facility.name).all()) if customer_facility_ids else [] else: facilities = Facility.query.filter_by(active=True).order_by(Facility.name).all() return render_template('reports/issues_aging.html', now=now, buckets=buckets, bucket_labels=AGING_BUCKETS, total=len(issues), sla_breached=sla_breached, sla_at_risk=sla_at_risk, severity_filter=severity_filter, facility_id_filter=facility_id_filter, facilities=facilities, ) @bp.route('/export/issues-aging') @login_required def export_issues_aging(): 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) from app.utils.sla import sla_status now = now_eastern() customer_facility_ids = get_customer_scope(current_user) severity_filter = request.args.get('severity', '') facility_id_filter = request.args.get('facility_id', type=int) issues = _load_open_issues_scoped(customer_facility_ids, severity_filter, facility_id_filter) log_action(ACTION_EXPORT, 'Issue', None, 'Issues Aging Excel', f'severity={severity_filter or "all"} facility={facility_id_filter or "all"}') wb = Workbook() ws = wb.active ws.title = 'Issues Aging' ws.freeze_panes = 'A2' hdr_font = Font(bold=True, color='FFFFFF', size=11) hdr_fill = PatternFill('solid', fgColor='DC2626') good_fill = PatternFill('solid', fgColor='D1FAE5') warn_fill = PatternFill('solid', fgColor='FEF3C7') bad_fill = PatternFill('solid', fgColor='FEE2E2') sub_fill = PatternFill('solid', fgColor='FFF5F5') thin = Side(style='thin', color='D1D5DB') bdr = Border(left=thin, right=thin, top=thin, bottom=thin) ctr = Alignment(horizontal='center', vertical='center') lft = Alignment(horizontal='left', vertical='center') hdrs = ['#', 'Reported', 'Age (h)', 'Age Bucket', 'Facility', 'Area', 'Severity', 'Description', 'Status', 'SLA', 'Assigned To'] widths = [6, 18, 10, 14, 28, 20, 12, 45, 20, 14, 22] for ci, (h, w) in enumerate(zip(hdrs, widths), 1): cell = ws.cell(row=1, column=ci, value=h) cell.font = hdr_font; cell.fill = hdr_fill; cell.border = bdr cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True) ws.column_dimensions[get_column_letter(ci)].width = w ws.row_dimensions[1].height = 22 sev_fill = {'critical': bad_fill, 'high': bad_fill, 'medium': warn_fill, 'low': sub_fill} sla_fill = {'breached': bad_fill, 'at_risk': warn_fill, 'ok': good_fill} for ri, issue in enumerate(issues, 2): age_h = (now - issue.reported_at).total_seconds() / 3600 sla = sla_status(issue) fac = issue.resolved_facility stripe = sub_fill if ri % 2 == 0 else None row = [issue.id, issue.reported_at.strftime('%Y-%m-%d %H:%M'), round(age_h, 1), _age_bucket(age_h), fac.name if fac else '—', issue.area.name if issue.area else '—', issue.severity.title() if issue.severity else '—', issue.description, issue.status.replace('_', ' ').title(), (sla or '—').replace('_', ' ').title(), issue.assigned_user.display_name if issue.assigned_user else '—'] for ci, val in enumerate(row, 1): cell = ws.cell(row=ri, column=ci, value=val) cell.border = bdr cell.alignment = lft if ci in (5, 6, 8, 11) else ctr if ci not in (7, 10) and stripe: cell.fill = stripe sf = sev_fill.get(issue.severity) if sf: ws.cell(row=ri, column=7).fill = sf slaf = sla_fill.get(sla) if slaf: ws.cell(row=ri, column=10).fill = slaf ws.row_dimensions[ri].height = 16 buf = io.BytesIO(); wb.save(buf); buf.seek(0) fname = f'issues_aging_{now.strftime("%Y%m%d")}.xlsx' return Response(buf.read(), mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', headers={'Content-Disposition': f'attachment; filename="{fname}"'}) # ── SLA Compliance Summary ──────────────────────────────────────────────────── def _sla_within(issue): from app.utils.sla import SLA_HOURS if not issue.resolved_at or not issue.reported_at: return False elapsed_h = (issue.resolved_at - issue.reported_at).total_seconds() / 3600 return elapsed_h <= SLA_HOURS.get(issue.severity, 9999) @bp.route('/sla-compliance') @login_required def sla_compliance(): from app.utils.sla import SLA_HOURS start, end = _date_range() customer_facility_ids = get_customer_scope(current_user) facility_id_filter = request.args.get('facility_id', type=int) q = Issue.query.options( joinedload(Issue.area).joinedload(Area.facility), ).filter( Issue.status == 'resolved', Issue.reported_at >= start, Issue.reported_at <= end, ) if customer_facility_ids is not None: if not customer_facility_ids: q = q.filter(False) else: q = 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)) ) issues = q.all() if facility_id_filter: issues = [i for i in issues if (i.facility_id == facility_id_filter) or (i.area and i.area.facility_id == facility_id_filter)] total = len(issues) met = sum(1 for i in issues if _sla_within(i)) overall_pct = round(met / total * 100, 1) if total else None by_severity = {} for sev in ('critical', 'high', 'medium', 'low'): sub = [i for i in issues if i.severity == sev] n = len(sub) m = sum(1 for i in sub if _sla_within(i)) by_severity[sev] = { 'total': n, 'met': m, 'pct': round(m / n * 100, 1) if n else None, 'sla_hours': SLA_HOURS.get(sev, 0), } fac_map = {} for issue in issues: fac = issue.resolved_facility if not fac: continue fid = fac.id if fid not in fac_map: fac_map[fid] = {'name': fac.name, 'total': 0, 'met': 0} fac_map[fid]['total'] += 1 if _sla_within(issue): fac_map[fid]['met'] += 1 by_facility = sorted([ {'name': d['name'], 'total': d['total'], 'met': d['met'], 'pct': round(d['met'] / d['total'] * 100, 1) if d['total'] else None} for d in fac_map.values() ], key=lambda x: (x['pct'] is None, -(x['pct'] or 0))) if customer_facility_ids is not None: facilities = (Facility.query.filter(Facility.id.in_(customer_facility_ids), Facility.active == True) .order_by(Facility.name).all()) if customer_facility_ids else [] else: facilities = Facility.query.filter_by(active=True).order_by(Facility.name).all() return render_template('reports/sla_compliance.html', start=start, end=end, total=total, met=met, overall_pct=overall_pct, by_severity=by_severity, by_facility=by_facility, facility_id_filter=facility_id_filter, facilities=facilities, ) @bp.route('/export/sla-compliance') @login_required def export_sla_compliance(): 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) from app.utils.sla import SLA_HOURS start, end = _date_range() customer_facility_ids = get_customer_scope(current_user) facility_id_filter = request.args.get('facility_id', type=int) q = Issue.query.options( joinedload(Issue.area).joinedload(Area.facility), ).filter(Issue.status == 'resolved', Issue.reported_at >= start, Issue.reported_at <= end) if customer_facility_ids is not None: if not customer_facility_ids: q = q.filter(False) else: q = 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)) ) issues = q.all() if facility_id_filter: issues = [i for i in issues if (i.facility_id == facility_id_filter) or (i.area and i.area.facility_id == facility_id_filter)] log_action(ACTION_EXPORT, 'Issue', None, 'SLA Compliance Excel', f'range={start.strftime("%Y-%m-%d")} to {end.strftime("%Y-%m-%d")}') wb = Workbook() thin = Side(style='thin', color='D1D5DB') bdr = Border(left=thin, right=thin, top=thin, bottom=thin) hdr_font = Font(bold=True, color='FFFFFF', size=11) hdr_fill = PatternFill('solid', fgColor='1A56DB') good_fill = PatternFill('solid', fgColor='D1FAE5') warn_fill = PatternFill('solid', fgColor='FEF3C7') bad_fill = PatternFill('solid', fgColor='FEE2E2') sub_fill = PatternFill('solid', fgColor='EFF6FF') ctr = Alignment(horizontal='center', vertical='center') lft = Alignment(horizontal='left', vertical='center') def _pct_fill(pct): if pct is None: return None return good_fill if pct >= 90 else warn_fill if pct >= 70 else bad_fill def _hdr(ws, row, headers, widths, fill=None): fill = fill or hdr_fill for ci, (h, w) in enumerate(zip(headers, widths), 1): cell = ws.cell(row=row, column=ci, value=h) cell.font = hdr_font; cell.fill = fill; cell.border = bdr cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True) ws.column_dimensions[get_column_letter(ci)].width = w ws.row_dimensions[row].height = 22 # ── Sheet 1: Severity Summary ──────────────────────────────────────── ws1 = wb.active ws1.title = 'By Severity' ws1.freeze_panes = 'A2' _hdr(ws1, 1, ['Severity', 'SLA Window', 'Total Resolved', 'Met SLA', 'Compliance %'], [14, 14, 16, 12, 16]) for ri, sev in enumerate(['critical', 'high', 'medium', 'low'], 2): sub = [i for i in issues if i.severity == sev] n = len(sub) m = sum(1 for i in sub if _sla_within(i)) pct = round(m / n * 100, 1) if n else None row = [sev.title(), f'{SLA_HOURS.get(sev)}h', n, m, f'{pct}%' if pct is not None else '—'] for ci, val in enumerate(row, 1): cell = ws1.cell(row=ri, column=ci, value=val) cell.border = bdr; cell.alignment = lft if ci == 1 else ctr if ri % 2 == 0: cell.fill = sub_fill pf = _pct_fill(pct) if pf: ws1.cell(row=ri, column=5).fill = pf # Overall totals row tr = 6 total = len(issues); met = sum(1 for i in issues if _sla_within(i)) op = round(met / total * 100, 1) if total else None for ci, val in enumerate(['OVERALL', '—', total, met, f'{op}%' if op is not None else '—'], 1): cell = ws1.cell(row=tr, column=ci, value=val) cell.font = Font(bold=True); cell.border = bdr; cell.alignment = lft if ci == 1 else ctr cell.fill = PatternFill('solid', fgColor='DBEAFE') pf = _pct_fill(op) if pf: ws1.cell(row=tr, column=5).fill = pf # ── Sheet 2: Facility Breakdown ────────────────────────────────────── ws2 = wb.create_sheet('By Facility') ws2.freeze_panes = 'A2' _hdr(ws2, 1, ['Facility', 'Total Resolved', 'Met SLA', 'Compliance %'], [30, 16, 12, 16]) fac_map = {} for issue in issues: fac = issue.resolved_facility if not fac: continue fid = fac.id if fid not in fac_map: fac_map[fid] = {'name': fac.name, 'total': 0, 'met': 0} fac_map[fid]['total'] += 1 if _sla_within(issue): fac_map[fid]['met'] += 1 rows_sorted = sorted(fac_map.values(), key=lambda x: ( (round(x['met']/x['total']*100,1) if x['total'] else None) is None, -(round(x['met']/x['total']*100,1) if x['total'] else 0) )) for ri, d in enumerate(rows_sorted, 2): pct = round(d['met'] / d['total'] * 100, 1) if d['total'] else None row = [d['name'], d['total'], d['met'], f'{pct}%' if pct is not None else '—'] for ci, val in enumerate(row, 1): cell = ws2.cell(row=ri, column=ci, value=val) cell.border = bdr; cell.alignment = lft if ci == 1 else ctr if ri % 2 == 0: cell.fill = sub_fill pf = _pct_fill(pct) if pf: ws2.cell(row=ri, column=4).fill = pf buf = io.BytesIO(); wb.save(buf); buf.seek(0) fname = f'sla_compliance_{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="{fname}"'}) # ── Follow-up Closure Rate ──────────────────────────────────────────────────── @bp.route('/followup-closure') @login_required @supervisor_required def followup_closure(): start, end = _date_range() flagged = Inspection.query.options( joinedload(Inspection.facility), joinedload(Inspection.inspector), joinedload(Inspection.template), ).filter( Inspection.follow_up_required == True, Inspection.inspection_date >= start, Inspection.inspection_date <= end, ).order_by(Inspection.inspection_date.desc()).all() flagged_ids = [i.id for i in flagged] followed_up_ids = set() if flagged_ids: followed_up_ids = { row[0] for row in db.session.query(Inspection.parent_inspection_id) .filter(Inspection.parent_inspection_id.in_(flagged_ids)) .distinct() if row[0] is not None } for insp in flagged: insp._has_followup = insp.id in followed_up_ids total = len(flagged) closed = sum(1 for i in flagged if i._has_followup) rate = round(closed / total * 100, 1) if total else None fac_map = {} for insp in flagged: fac = insp.facility if not fac: continue fid = fac.id if fid not in fac_map: fac_map[fid] = {'name': fac.name, 'total': 0, 'closed': 0} fac_map[fid]['total'] += 1 if insp._has_followup: fac_map[fid]['closed'] += 1 by_facility = sorted([ {'name': d['name'], 'total': d['total'], 'closed': d['closed'], 'rate': round(d['closed'] / d['total'] * 100, 1) if d['total'] else 0} for d in fac_map.values() ], key=lambda x: -x['total']) return render_template('reports/followup_closure.html', start=start, end=end, flagged=flagged, total=total, closed=closed, rate=rate, by_facility=by_facility, followed_up_ids=followed_up_ids, ) @bp.route('/export/followup-closure') @login_required @supervisor_required def export_followup_closure(): 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() flagged = Inspection.query.options( joinedload(Inspection.facility), joinedload(Inspection.inspector), joinedload(Inspection.template), ).filter( Inspection.follow_up_required == True, Inspection.inspection_date >= start, Inspection.inspection_date <= end, ).order_by(Inspection.inspection_date.desc()).all() flagged_ids = [i.id for i in flagged] followed_up_ids = set() if flagged_ids: followed_up_ids = { row[0] for row in db.session.query(Inspection.parent_inspection_id) .filter(Inspection.parent_inspection_id.in_(flagged_ids)) .distinct() if row[0] is not None } for insp in flagged: insp._has_followup = insp.id in followed_up_ids log_action(ACTION_EXPORT, 'Inspection', None, 'Follow-up Closure Excel', f'range={start.strftime("%Y-%m-%d")} to {end.strftime("%Y-%m-%d")}') wb = Workbook() ws = wb.active ws.title = 'Follow-up Detail' ws.freeze_panes = 'A2' thin = Side(style='thin', color='D1D5DB') bdr = Border(left=thin, right=thin, top=thin, bottom=thin) hdr_font = Font(bold=True, color='FFFFFF', size=11) hdr_fill = PatternFill('solid', fgColor='7C3AED') good_fill = PatternFill('solid', fgColor='D1FAE5') bad_fill = PatternFill('solid', fgColor='FEE2E2') sub_fill = PatternFill('solid', fgColor='F5F3FF') ctr = Alignment(horizontal='center', vertical='center') lft = Alignment(horizontal='left', vertical='center') hdrs = ['Inspection ID', 'Date', 'Facility', 'Area', 'Template', 'Inspector', 'Score (%)', 'Follow-up Note', 'Followed Up?'] widths = [14, 18, 28, 20, 24, 22, 12, 40, 14] for ci, (h, w) in enumerate(zip(hdrs, widths), 1): cell = ws.cell(row=1, column=ci, value=h) cell.font = hdr_font; cell.fill = hdr_fill; cell.border = bdr cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True) ws.column_dimensions[get_column_letter(ci)].width = w ws.row_dimensions[1].height = 22 for ri, insp in enumerate(flagged, 2): stripe = sub_fill if ri % 2 == 0 else None insp_name = insp.inspector.display_name if insp.inspector else '—' row = [insp.id, insp.inspection_date.strftime('%Y-%m-%d') if insp.inspection_date else '', insp.facility.name if insp.facility else '—', insp.area.name if insp.area else '—', insp.template.name if insp.template else '—', insp_name, insp.overall_score, insp.follow_up_note or '', 'Yes' if insp._has_followup else 'No'] for ci, val in enumerate(row, 1): cell = ws.cell(row=ri, column=ci, value=val) cell.border = bdr cell.alignment = lft if ci in (3, 4, 5, 6, 8) else ctr if stripe and ci != 9: cell.fill = stripe # Colour the Follow-up column fu_cell = ws.cell(row=ri, column=9) fu_cell.fill = good_fill if insp._has_followup else bad_fill fu_cell.font = Font(bold=True) ws.row_dimensions[ri].height = 16 buf = io.BytesIO(); wb.save(buf); buf.seek(0) fname = f'followup_closure_{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="{fname}"'}) # ── Customer Facility Summary PDF ──────────────────────────────────────────── @bp.route('/facility//summary-pdf') @login_required def facility_summary_pdf(facility_id): """One-click customer-facing PDF summary for a facility.""" from app.utils.pdf_export import generate_facility_summary_pdf from app.utils.sla import SLA_HOURS facility = db.session.get(Facility, facility_id) if facility is None: abort(404) # Access control if current_user.role == 'customer': cids = get_customer_scope(current_user) or [] if facility_id not in cids: abort(403) elif current_user.role == 'inspector': has = Inspection.query.filter_by(facility_id=facility_id, inspector_id=current_user.id).first() if not has: abort(403) days = request.args.get('days', 90, type=int) if days not in (30, 60, 90, 180, 365): days = 90 now = now_eastern() start = now - timedelta(days=days) completed_insp = Inspection.query.filter( Inspection.facility_id == facility_id, Inspection.inspection_date >= start, Inspection.status == 'completed', ).order_by(Inspection.inspection_date.desc()).all() 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 = Issue.query.join(Area).filter( Area.facility_id == facility_id, Issue.status != 'resolved', ).options(joinedload(Issue.area)).order_by(Issue.severity, Issue.reported_at).all() resolved_count = Issue.query.join(Area).filter( Area.facility_id == facility_id, Issue.status == 'resolved', Issue.resolved_at >= start, ).count() scores = [float(i.overall_score) for i in completed_insp if i.overall_score is not None] avg_score = round(sum(scores) / len(scores), 1) if scores else None log_action(ACTION_EXPORT, 'Facility', facility_id, facility.name, f'Customer Summary PDF days={days}') pdf_bytes = generate_facility_summary_pdf( facility=facility, days=days, start=start, now=now, total_inspections=len(completed_insp), avg_score=avg_score, area_scores=area_scores, open_issues=open_issues, resolved_count=resolved_count, ) fname = f'{facility.name.replace(" ", "_")}_summary_{now.strftime("%Y%m%d")}.pdf' return Response(pdf_bytes, mimetype='application/pdf', headers={'Content-Disposition': f'attachment; filename="{fname}"'})