200 lines
7.3 KiB
Python
200 lines
7.3 KiB
Python
"""
|
||
app/routes/public.py
|
||
--------------------
|
||
Login-free, token-addressed facility pages reached by scanning a facility's
|
||
QR code. The QR encodes /f/<public_token> (an unguessable token, so the pages
|
||
cannot be enumerated by facility id).
|
||
|
||
Routes
|
||
------
|
||
GET /f/<token> Occupant-friendly facility summary (no login).
|
||
POST /f/<token>/report Occupant "report a problem" → creates an open Issue.
|
||
|
||
Design notes
|
||
------------
|
||
- Occupant-friendly: shows a quality rating, last-inspected date, and open-issue
|
||
COUNT only — never issue descriptions, inspector names, or internal scores.
|
||
- Inactive facilities 404 (a decommissioned QR reveals nothing).
|
||
- The report form is rate-limited and honeypot-guarded against bots, and reuses
|
||
the normal issue-creation notification path so staff/customers are alerted.
|
||
"""
|
||
|
||
import logging
|
||
from datetime import timedelta
|
||
|
||
from flask import Blueprint, render_template, redirect, url_for, flash, request, abort
|
||
from sqlalchemy import func
|
||
|
||
from app import db, limiter
|
||
from app.models.facility import Facility, Area
|
||
from app.models.inspection import Inspection
|
||
from app.models.issue import Issue
|
||
from app.utils.forms import PublicIssueReportForm
|
||
from app.utils.time_utils import now_eastern
|
||
from app.utils.notifications import notify_by_matrix
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
bp = Blueprint('public', __name__, url_prefix='/f')
|
||
|
||
|
||
def _facility_by_token_or_404(token: str) -> Facility:
|
||
"""Resolve an ACTIVE facility from its public token, else 404."""
|
||
if not token:
|
||
abort(404)
|
||
facility = Facility.query.filter_by(public_token=token).first()
|
||
if facility is None or not facility.active:
|
||
abort(404)
|
||
return facility
|
||
|
||
|
||
def _rating_label(score):
|
||
"""Map a 0–100 score to an occupant-friendly label + Bootstrap colour."""
|
||
if score is None:
|
||
return ('Not yet rated', 'secondary')
|
||
if score >= 90:
|
||
return ('Excellent', 'success')
|
||
if score >= 80:
|
||
return ('Good', 'success')
|
||
if score >= 70:
|
||
return ('Fair', 'warning')
|
||
return ('Needs attention', 'danger')
|
||
|
||
|
||
def _build_summary(facility: Facility) -> dict:
|
||
"""Assemble the occupant-facing summary for a facility."""
|
||
fid = facility.id
|
||
now = now_eastern()
|
||
cutoff = now - timedelta(days=90)
|
||
|
||
# Most recent completed, scored inspection
|
||
last_insp = (
|
||
Inspection.query
|
||
.filter(Inspection.facility_id == fid,
|
||
Inspection.status == 'completed',
|
||
Inspection.overall_score.isnot(None))
|
||
.order_by(Inspection.inspection_date.desc())
|
||
.first()
|
||
)
|
||
|
||
# Average score over the last 90 days (fallback: all-time) for the rating
|
||
avg_90 = (
|
||
db.session.query(func.avg(Inspection.overall_score))
|
||
.filter(Inspection.facility_id == fid,
|
||
Inspection.status == 'completed',
|
||
Inspection.overall_score.isnot(None),
|
||
Inspection.inspection_date >= cutoff)
|
||
.scalar()
|
||
)
|
||
if avg_90 is None:
|
||
avg_90 = (
|
||
db.session.query(func.avg(Inspection.overall_score))
|
||
.filter(Inspection.facility_id == fid,
|
||
Inspection.status == 'completed',
|
||
Inspection.overall_score.isnot(None))
|
||
.scalar()
|
||
)
|
||
avg_score = round(float(avg_90), 1) if avg_90 is not None else None
|
||
|
||
# Open-issue COUNT (linked directly or via an area) — no details exposed
|
||
open_issue_count = (
|
||
Issue.query
|
||
.outerjoin(Area, Issue.area_id == Area.id)
|
||
.filter(Issue.status.in_(['open', 'in_progress']),
|
||
db.or_(Issue.facility_id == fid, Area.facility_id == fid))
|
||
.count()
|
||
)
|
||
|
||
label, colour = _rating_label(avg_score)
|
||
|
||
return {
|
||
'facility': facility,
|
||
'avg_score': avg_score,
|
||
'rating_label': label,
|
||
'rating_colour': colour,
|
||
'last_inspected': last_insp.inspection_date if last_insp else None,
|
||
'last_score': (round(float(last_insp.overall_score), 1)
|
||
if last_insp and last_insp.overall_score is not None else None),
|
||
'open_issue_count': open_issue_count,
|
||
}
|
||
|
||
|
||
@bp.route('/<token>', methods=['GET'])
|
||
def facility_summary(token):
|
||
facility = _facility_by_token_or_404(token)
|
||
summary = _build_summary(facility)
|
||
form = PublicIssueReportForm()
|
||
return render_template('public/facility.html',
|
||
form=form, token=token, **summary)
|
||
|
||
|
||
@bp.route('/<token>/report', methods=['POST'])
|
||
@limiter.limit('5 per hour; 20 per day')
|
||
def report_problem(token):
|
||
facility = _facility_by_token_or_404(token)
|
||
form = PublicIssueReportForm()
|
||
|
||
# Honeypot: silently accept-and-drop obvious bot submissions.
|
||
if form.website.data:
|
||
logger.info('PUBLIC REPORT | honeypot tripped | facility_id=%s | ip=%s',
|
||
facility.id, request.remote_addr)
|
||
flash('Thank you — your report has been received.', 'success')
|
||
return redirect(url_for('public.facility_summary', token=token))
|
||
|
||
if not form.validate_on_submit():
|
||
# Re-render the page with validation errors and the summary intact.
|
||
summary = _build_summary(facility)
|
||
return render_template('public/facility.html',
|
||
form=form, token=token, **summary), 400
|
||
|
||
# Save optional photo through the shared, magic-byte-validated saver.
|
||
from app.routes.inspections import _save_photo
|
||
photo_path = _save_photo(form.photo.data, subfolder='issue_photos')
|
||
|
||
# Fold optional reporter identity + location into the description; the
|
||
# public reporter is not a User, so reported_by stays NULL.
|
||
parts = ['[Reported via facility QR code]']
|
||
if form.area_label.data:
|
||
parts.append(f'Location: {form.area_label.data.strip()}')
|
||
reporter_bits = [b for b in (form.reporter_name.data, form.reporter_contact.data) if b]
|
||
if reporter_bits:
|
||
parts.append('Reporter: ' + ' — '.join(b.strip() for b in reporter_bits))
|
||
parts.append('')
|
||
parts.append(form.description.data.strip())
|
||
description = '\n'.join(parts)
|
||
|
||
issue = Issue(
|
||
facility_id = facility.id,
|
||
area_id = None,
|
||
severity = 'medium',
|
||
description = description,
|
||
photo_path = photo_path,
|
||
status = 'open',
|
||
reported_at = now_eastern(),
|
||
reported_by = None,
|
||
)
|
||
db.session.add(issue)
|
||
db.session.commit()
|
||
|
||
logger.info('PUBLIC REPORT | issue_id=%s | facility_id=%s | ip=%s | photo=%s',
|
||
issue.id, facility.id, request.remote_addr, bool(photo_path))
|
||
|
||
# Reuse the standard issue-created routing (staff + facility customers).
|
||
notify_by_matrix(
|
||
event_type = 'issue_created',
|
||
title = f'New Issue #{issue.id} at {facility.name} (QR report)',
|
||
body = (
|
||
f'A problem was reported at {facility.name} via the facility QR code. '
|
||
f'Description: {form.description.data.strip()[:120]}'
|
||
f'{"…" if len(form.description.data.strip()) > 120 else ""}'
|
||
),
|
||
link = url_for('issues.view', issue_id=issue.id),
|
||
issue_id = issue.id,
|
||
facility_id = facility.id,
|
||
)
|
||
db.session.commit()
|
||
|
||
flash('Thank you — your report has been received and the team has been notified.',
|
||
'success')
|
||
return redirect(url_for('public.facility_summary', token=token))
|