1691 lines
69 KiB
Python
1691 lines
69 KiB
Python
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.models.project import Project
|
||
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',
|
||
'external_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.is_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) — includes project_id so the report
|
||
# can group/filter facilities by Contract client-side.
|
||
fac_score_q = db.session.query(
|
||
Facility.name,
|
||
Facility.project_id,
|
||
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, Facility.project_id)\
|
||
.order_by(func.avg(Inspection.overall_score).desc()).all()
|
||
|
||
# Resolve contract names for the facilities present.
|
||
_proj_ids = {r.project_id for r in facility_scores if r.project_id}
|
||
_proj_names = (
|
||
{p.id: p.name for p in Project.query.filter(Project.id.in_(_proj_ids)).all()}
|
||
if _proj_ids else {}
|
||
)
|
||
|
||
# Prior-period facility scores for period-over-period delta badges
|
||
period_len = end - start
|
||
prior_end = start
|
||
prior_start = start - period_len
|
||
prior_fac_q = db.session.query(
|
||
Facility.name,
|
||
func.avg(Inspection.overall_score).label('avg_score'),
|
||
).join(Inspection, Facility.id == Inspection.facility_id)\
|
||
.filter(
|
||
Inspection.inspection_date >= prior_start,
|
||
Inspection.inspection_date <= prior_end,
|
||
Inspection.status == 'completed',
|
||
Inspection.overall_score.isnot(None),
|
||
)
|
||
if inspector_filter:
|
||
prior_fac_q = prior_fac_q.filter(Inspection.inspector_id == inspector_filter)
|
||
if customer_facility_ids is not None:
|
||
prior_fac_q = prior_fac_q.filter(
|
||
Facility.id.in_(customer_facility_ids) if customer_facility_ids else False
|
||
)
|
||
prior_scores_raw = prior_fac_q.group_by(Facility.id, Facility.name).all()
|
||
prior_scores_map = {r.name: round(float(r.avg_score), 2) for r in prior_scores_raw}
|
||
# Build delta map keyed by facility name: positive = improved, negative = declined
|
||
facility_deltas = {}
|
||
for row in facility_scores:
|
||
prior = prior_scores_map.get(row.name)
|
||
if prior is not None:
|
||
facility_deltas[row.name] = round(float(row.avg_score) - prior, 1)
|
||
|
||
# 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(User.role.in_(User.INSPECTOR_ROLES),
|
||
User.active == True)\
|
||
.order_by(User.full_name, User.username).all()
|
||
|
||
facility_scores_list = [{
|
||
'name': r.name,
|
||
'avg_score': round(float(r.avg_score), 2),
|
||
'count': r.count,
|
||
'project_id': r.project_id or 0,
|
||
'contract': _proj_names.get(r.project_id, 'No Contract'),
|
||
} for r in facility_scores]
|
||
# Attach prior avg and delta to each facility score dict for the template table
|
||
for row in facility_scores_list:
|
||
row['prior_avg'] = prior_scores_map.get(row['name'])
|
||
row['delta'] = facility_deltas.get(row['name'])
|
||
|
||
# Distinct contracts present, for the "Avg Score by Facility" contract filter.
|
||
score_contracts = sorted(
|
||
{(r['project_id'], r['contract']) for r in facility_scores_list},
|
||
key=lambda t: (t[1] or '').lower(),
|
||
)
|
||
|
||
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=facility_scores_list,
|
||
score_contracts=score_contracts,
|
||
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/<int:facility_id>')
|
||
@login_required
|
||
def facility_report(facility_id):
|
||
if current_user.role not in ['admin', 'director', 'project_manager', 'inspector',
|
||
'external_inspector', 'customer']:
|
||
from flask import flash, redirect, url_for
|
||
flash('Access denied.', 'danger')
|
||
return redirect(url_for('dashboard.index'))
|
||
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.is_inspector:
|
||
# Inspectors may only view the facility report for facilities where
|
||
# they have personally conducted at least one inspection.
|
||
has_access = Inspection.query.filter_by(
|
||
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/<int:facility_id>/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',
|
||
'external_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.is_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.in_(User.INSPECTOR_ROLES))
|
||
.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,
|
||
# phase49 — customer / third-party inspectors appear in the same
|
||
# table as our own crew, badged so the numbers can be read in
|
||
# context. Consumed by the HTML table and the Excel export.
|
||
'external': u.is_external_inspector,
|
||
'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.is_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.in_(User.INSPECTOR_ROLES),
|
||
)
|
||
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 = [
|
||
# phase49 — external inspectors share this table with our own crew;
|
||
# suffixed rather than given a column so the index-based styling
|
||
# below (score = col 5, vs_avg = col 6, …) stays correct.
|
||
s['display_name'] + (' (External)' if s.get('external') else ''),
|
||
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/<int:facility_id>/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.is_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}"'}) |