562 lines
22 KiB
Python
562 lines
22 KiB
Python
import logging
|
|
from flask import Blueprint, render_template, redirect, url_for, flash, request, abort
|
|
from flask_login import login_required, current_user
|
|
from app import db
|
|
from app.models.facility import Facility, Area
|
|
from app.models.project import Project
|
|
from app.utils.forms import FacilityForm, AreaForm
|
|
from app.utils.decorators import supervisor_required, admin_required
|
|
from app.utils.audit import log_action, ACTION_CREATE, ACTION_UPDATE, ACTION_DELETE, ACTION_EXPORT
|
|
from app.utils.scope import get_customer_scope, get_inspector_scope
|
|
|
|
bp = Blueprint('facilities', __name__, url_prefix='/facilities')
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
@bp.route('/')
|
|
@login_required
|
|
def list_facilities():
|
|
if current_user.role == 'customer':
|
|
cids = get_customer_scope(current_user) or []
|
|
facilities = Facility.query.filter(
|
|
Facility.id.in_(cids), Facility.active == True
|
|
).order_by(Facility.name).all()
|
|
elif current_user.role == 'inspector':
|
|
fids = get_inspector_scope(current_user) or []
|
|
facilities = Facility.query.filter(
|
|
Facility.id.in_(fids), Facility.active == True
|
|
).order_by(Facility.name).all()
|
|
else:
|
|
facilities = Facility.query.order_by(Facility.name).all()
|
|
|
|
# Group facilities by Contract (Project) for the collapsible list view.
|
|
# Facilities with no contract are collected under key '__none__' and
|
|
# rendered last as "No Contract Assigned".
|
|
from collections import OrderedDict
|
|
grouped = OrderedDict()
|
|
ungrouped = []
|
|
for f in facilities:
|
|
if f.project:
|
|
key = f.project.name
|
|
grouped.setdefault(key, {'project': f.project, 'facilities': []})
|
|
grouped[key]['facilities'].append(f)
|
|
else:
|
|
ungrouped.append(f)
|
|
|
|
grouped = OrderedDict(sorted(grouped.items()))
|
|
if ungrouped:
|
|
grouped['__none__'] = {'project': None, 'facilities': ungrouped}
|
|
|
|
logger.info('FACILITIES | list | user=%s | total=%s | groups=%s',
|
|
current_user.username, len(facilities), len(grouped))
|
|
return render_template('facilities/list.html', facilities=facilities, grouped=grouped)
|
|
|
|
@bp.route('/new', methods=['GET', 'POST'])
|
|
@login_required
|
|
@supervisor_required
|
|
def create_facility():
|
|
form = FacilityForm()
|
|
projects = Project.query.filter_by(active=True).order_by(Project.name).all()
|
|
form.project_id.choices = [(0, '— None —')] + [(p.id, p.name) for p in projects]
|
|
|
|
if form.validate_on_submit():
|
|
project_id = form.project_id.data if form.project_id.data else None
|
|
facility = Facility(
|
|
name=form.name.data,
|
|
address=form.address.data,
|
|
contact_person=form.contact_person.data,
|
|
contact_phone=form.contact_phone.data,
|
|
project_id=project_id if project_id else None,
|
|
active=form.active.data
|
|
)
|
|
|
|
facility.ensure_public_token() # QR landing-page token
|
|
db.session.add(facility)
|
|
db.session.commit()
|
|
logger.info('FACILITIES | create | user=%s | facility_id=%s name=%r',
|
|
current_user.username, facility.id, facility.name)
|
|
log_action(ACTION_CREATE, 'Facility', facility.id, facility.name,
|
|
f'contact={facility.contact_person or ""}; project_id={facility.project_id}; active={facility.active}')
|
|
flash(f'Facility "{facility.name}" created successfully.', 'success')
|
|
return redirect(url_for('facilities.view_facility', facility_id=facility.id))
|
|
|
|
return render_template('facilities/form.html', form=form, title='Create Facility')
|
|
|
|
@bp.route('/<int:facility_id>')
|
|
@login_required
|
|
def view_facility(facility_id):
|
|
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:
|
|
flash('Access denied.', 'danger')
|
|
return redirect(url_for('facilities.list_facilities'))
|
|
areas = facility.areas.order_by(Area.name).all()
|
|
return render_template('facilities/view.html', facility=facility, areas=areas)
|
|
|
|
|
|
# ── Public QR code ────────────────────────────────────────────────────────────
|
|
# Staff (admin/director/pm/inspector) may access QR for any facility; customers
|
|
# may access QR only for facilities in their assigned scope.
|
|
|
|
def _public_facility_url(facility):
|
|
"""Absolute URL the QR encodes — the login-free occupant summary page."""
|
|
facility.ensure_public_token()
|
|
if not facility.public_token:
|
|
return None
|
|
return url_for('public.facility_summary',
|
|
token=facility.public_token, _external=True)
|
|
|
|
|
|
def _facility_for_qr_or_403(facility_id):
|
|
"""Load a facility for a QR action, enforcing customer facility scope.
|
|
|
|
Customers may only touch QR codes for facilities they are assigned to; all
|
|
other (staff) roles have unrestricted QR access.
|
|
"""
|
|
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:
|
|
abort(403)
|
|
return facility
|
|
|
|
|
|
@bp.route('/<int:facility_id>/qr.png')
|
|
@login_required
|
|
def facility_qr_png(facility_id):
|
|
"""Return the facility's QR code as a PNG image."""
|
|
facility = _facility_for_qr_or_403(facility_id)
|
|
|
|
# Token may not exist for pre-phase34 rows viewed before any commit.
|
|
created = not facility.public_token
|
|
url = _public_facility_url(facility)
|
|
if created:
|
|
db.session.commit()
|
|
|
|
import io
|
|
import qrcode
|
|
img = qrcode.make(url, box_size=10, border=2)
|
|
buf = io.BytesIO()
|
|
img.save(buf, format='PNG')
|
|
buf.seek(0)
|
|
|
|
from flask import Response
|
|
return Response(buf.getvalue(), mimetype='image/png', headers={
|
|
'Cache-Control': 'private, max-age=3600',
|
|
})
|
|
|
|
|
|
@bp.route('/<int:facility_id>/qr')
|
|
@login_required
|
|
def facility_qr_page(facility_id):
|
|
"""Printable page: facility name + QR + public URL + posting instructions."""
|
|
facility = _facility_for_qr_or_403(facility_id)
|
|
public_url = _public_facility_url(facility)
|
|
db.session.commit() # persist token if it was just generated
|
|
return render_template('facilities/qr.html',
|
|
facility=facility, public_url=public_url)
|
|
|
|
|
|
@bp.route('/<int:facility_id>/qr/regenerate', methods=['POST'])
|
|
@login_required
|
|
def facility_qr_regenerate(facility_id):
|
|
"""Mint a NEW public token, invalidating any previously printed QR code.
|
|
|
|
Allowed for admin/director, and for customers on their own assigned
|
|
facilities. Project managers and inspectors cannot regenerate.
|
|
"""
|
|
facility = _facility_for_qr_or_403(facility_id)
|
|
if current_user.role not in ('admin', 'director', 'customer'):
|
|
abort(403)
|
|
|
|
facility.public_token = Facility.generate_public_token()
|
|
db.session.commit()
|
|
|
|
logger.info('FACILITIES | qr_regenerate | user=%s | facility_id=%s',
|
|
current_user.username, facility.id)
|
|
log_action(ACTION_UPDATE, 'Facility', facility.id, facility.name,
|
|
'regenerated public QR token (old code invalidated)')
|
|
flash('QR code regenerated. Any previously printed codes for this facility no '
|
|
'longer work — reprint and repost.', 'warning')
|
|
return redirect(url_for('facilities.facility_qr_page', facility_id=facility.id))
|
|
|
|
|
|
def _qr_png_bytes(url):
|
|
"""Return PNG bytes for a QR code encoding *url* (same params as qr.png)."""
|
|
import io
|
|
import qrcode
|
|
img = qrcode.make(url, box_size=10, border=2)
|
|
buf = io.BytesIO()
|
|
img.save(buf, format='PNG')
|
|
return buf.getvalue()
|
|
|
|
|
|
@bp.route('/qr/print-all')
|
|
@login_required
|
|
def facility_qr_print_all():
|
|
"""Printable / selectable sheet of QR codes the user can see.
|
|
|
|
Query params (all optional):
|
|
?contract_id=<id> — limit to one contract; narrows the facility dropdown
|
|
?facility_id=<id> — limit to a single facility
|
|
?include_areas=1 — also render each facility's per-area QR codes
|
|
|
|
Inspectors have no QR management (403); customers are scoped to their
|
|
assigned facilities; managers see all active facilities.
|
|
"""
|
|
# QR management is not an inspector task.
|
|
if current_user.role == 'inspector':
|
|
abort(403)
|
|
|
|
contract_id = request.args.get('contract_id', type=int)
|
|
facility_id = request.args.get('facility_id', type=int)
|
|
include_areas = request.args.get('include_areas') in ('1', 'true', 'on')
|
|
|
|
# Facilities in the viewer's scope.
|
|
if current_user.role == 'customer':
|
|
fids = get_customer_scope(current_user) or []
|
|
scoped = Facility.query.filter(Facility.id.in_(fids), Facility.active == True)
|
|
else:
|
|
scoped = Facility.query.filter(Facility.active == True)
|
|
scoped_facilities = scoped.order_by(Facility.name).all()
|
|
|
|
# Contract dropdown — only contracts present among the scoped facilities.
|
|
contract_ids = {f.project_id for f in scoped_facilities if f.project_id}
|
|
contracts = (Project.query
|
|
.filter(Project.id.in_(contract_ids))
|
|
.order_by(Project.name).all()) if contract_ids else []
|
|
|
|
# Facility dropdown — narrowed by the selected contract.
|
|
facility_options = [f for f in scoped_facilities
|
|
if not contract_id or f.project_id == contract_id]
|
|
|
|
# The rendered grid — apply the contract + facility filters.
|
|
grid_facilities = facility_options
|
|
if facility_id:
|
|
grid_facilities = [f for f in grid_facilities if f.id == facility_id]
|
|
|
|
# Ensure every rendered facility (and area, if requested) has a token so
|
|
# its qr.png renders; collect areas keyed by facility id.
|
|
changed = False
|
|
areas_by_facility = {}
|
|
for f in grid_facilities:
|
|
if not f.public_token:
|
|
f.ensure_public_token()
|
|
changed = True
|
|
if include_areas:
|
|
fa = f.areas.order_by(Area.name).all()
|
|
for a in fa:
|
|
if not a.public_token:
|
|
a.ensure_public_token()
|
|
changed = True
|
|
areas_by_facility[f.id] = fa
|
|
if changed:
|
|
db.session.commit()
|
|
|
|
selected_contract = db.session.get(Project, contract_id) if contract_id else None
|
|
return render_template('facilities/qr_print_all.html',
|
|
facilities=grid_facilities,
|
|
areas_by_facility=areas_by_facility,
|
|
include_areas=include_areas,
|
|
contracts=contracts,
|
|
facility_options=facility_options,
|
|
selected_contract=selected_contract,
|
|
selected_contract_id=contract_id,
|
|
selected_facility_id=facility_id)
|
|
|
|
|
|
@bp.route('/qr/export-pdf', methods=['POST'])
|
|
@login_required
|
|
def facility_qr_export_pdf():
|
|
"""Export the selected facility + area QR codes to a single PDF.
|
|
|
|
Selection arrives as repeated `facility_ids` / `area_ids` form fields.
|
|
Scope is enforced per-id via the same helpers as the QR pages, so a
|
|
customer can never export a code outside their assigned facilities.
|
|
"""
|
|
if current_user.role == 'inspector':
|
|
abort(403)
|
|
|
|
facility_ids = request.form.getlist('facility_ids', type=int)
|
|
area_ids = request.form.getlist('area_ids', type=int)
|
|
|
|
if not facility_ids and not area_ids:
|
|
flash('Select at least one QR code to export.', 'warning')
|
|
return redirect(request.referrer or url_for('facilities.facility_qr_print_all'))
|
|
|
|
items = []
|
|
for fid in facility_ids:
|
|
facility = _facility_for_qr_or_403(fid) # 403 if out of scope
|
|
url = _public_facility_url(facility)
|
|
items.append({
|
|
'title': facility.name,
|
|
'subtitle': facility.project.name if facility.project else None,
|
|
'caption': 'Facility · Report a problem & view recent quality',
|
|
'png': _qr_png_bytes(url),
|
|
'_sort': ((facility.name or '').lower(), 0, ''),
|
|
})
|
|
for aid in area_ids:
|
|
area = _area_for_qr_or_403(aid) # 403 if out of scope
|
|
url = _public_area_url(area)
|
|
fac_name = area.facility.name if area.facility else ''
|
|
items.append({
|
|
'title': area.name,
|
|
'subtitle': fac_name or None,
|
|
'caption': 'Area · Report a problem & view recent quality',
|
|
'png': _qr_png_bytes(url),
|
|
'_sort': (fac_name.lower(), 1, (area.name or '').lower()),
|
|
})
|
|
|
|
# Persist any tokens minted by ensure_public_token() above.
|
|
db.session.commit()
|
|
|
|
# Group each facility with its own areas: facility card first, then areas.
|
|
items.sort(key=lambda x: x['_sort'])
|
|
|
|
from app.utils.pdf_export import generate_qr_codes_pdf
|
|
summary = f'{len(facility_ids)} facilit' + ('y' if len(facility_ids) == 1 else 'ies')
|
|
summary += f', {len(area_ids)} area' + ('' if len(area_ids) == 1 else 's')
|
|
pdf_bytes = generate_qr_codes_pdf(items, filter_summary=summary)
|
|
|
|
logger.info('FACILITIES | qr_export_pdf | user=%s | facilities=%s | areas=%s',
|
|
current_user.username, len(facility_ids), len(area_ids))
|
|
log_action(ACTION_EXPORT, 'Facility', 0, 'QR Codes',
|
|
f'exported {len(facility_ids)} facility + {len(area_ids)} area QR codes to PDF')
|
|
|
|
from flask import Response
|
|
return Response(pdf_bytes, mimetype='application/pdf', headers={
|
|
'Content-Disposition': 'attachment; filename="qr_codes.pdf"',
|
|
})
|
|
|
|
# ── Public Area QR code ───────────────────────────────────────────────────────
|
|
# Mirrors the facility QR routes above, but scoped to a single area. Customer
|
|
# scope is enforced via the area's parent facility.
|
|
|
|
def _public_area_url(area):
|
|
"""Absolute URL the area QR encodes — the login-free area summary page."""
|
|
area.ensure_public_token()
|
|
if not area.public_token:
|
|
return None
|
|
return url_for('public.area_summary',
|
|
token=area.public_token, _external=True)
|
|
|
|
|
|
def _area_for_qr_or_403(area_id):
|
|
"""Load an area for a QR action, enforcing customer facility scope."""
|
|
area = db.session.get(Area, area_id)
|
|
if area is None:
|
|
abort(404)
|
|
if current_user.role == 'customer':
|
|
cids = get_customer_scope(current_user) or []
|
|
if area.facility_id not in cids:
|
|
abort(403)
|
|
return area
|
|
|
|
|
|
@bp.route('/areas/<int:area_id>/qr.png')
|
|
@login_required
|
|
def area_qr_png(area_id):
|
|
"""Return the area's QR code as a PNG image."""
|
|
area = _area_for_qr_or_403(area_id)
|
|
|
|
created = not area.public_token
|
|
url = _public_area_url(area)
|
|
if created:
|
|
db.session.commit()
|
|
|
|
import io
|
|
import qrcode
|
|
img = qrcode.make(url, box_size=10, border=2)
|
|
buf = io.BytesIO()
|
|
img.save(buf, format='PNG')
|
|
buf.seek(0)
|
|
|
|
from flask import Response
|
|
return Response(buf.getvalue(), mimetype='image/png', headers={
|
|
'Cache-Control': 'private, max-age=3600',
|
|
})
|
|
|
|
|
|
@bp.route('/areas/<int:area_id>/qr')
|
|
@login_required
|
|
def area_qr_page(area_id):
|
|
"""Printable page: area name + facility + QR + public URL + instructions."""
|
|
area = _area_for_qr_or_403(area_id)
|
|
public_url = _public_area_url(area)
|
|
db.session.commit() # persist token if it was just generated
|
|
return render_template('facilities/area_qr.html',
|
|
area=area, facility=area.facility,
|
|
public_url=public_url)
|
|
|
|
|
|
@bp.route('/areas/<int:area_id>/qr/regenerate', methods=['POST'])
|
|
@login_required
|
|
def area_qr_regenerate(area_id):
|
|
"""Mint a NEW token for this area, invalidating any printed QR code.
|
|
|
|
Allowed for admin/director, and for customers on their own assigned
|
|
facilities. Project managers and inspectors cannot regenerate.
|
|
"""
|
|
area = _area_for_qr_or_403(area_id)
|
|
if current_user.role not in ('admin', 'director', 'customer'):
|
|
abort(403)
|
|
|
|
area.public_token = Area.generate_public_token()
|
|
db.session.commit()
|
|
|
|
logger.info('FACILITIES | area_qr_regenerate | user=%s | area_id=%s',
|
|
current_user.username, area.id)
|
|
log_action(ACTION_UPDATE, 'Area', area.id, area.name,
|
|
'regenerated public QR token (old code invalidated)')
|
|
flash('QR code regenerated. Any previously printed codes for this area no '
|
|
'longer work — reprint and repost.', 'warning')
|
|
return redirect(url_for('facilities.area_qr_page', area_id=area.id))
|
|
|
|
|
|
@bp.route('/<int:facility_id>/edit', methods=['GET', 'POST'])
|
|
@login_required
|
|
@supervisor_required
|
|
def edit_facility(facility_id):
|
|
facility = db.session.get(Facility, facility_id)
|
|
if facility is None:
|
|
abort(404)
|
|
form = FacilityForm(obj=facility)
|
|
projects = Project.query.filter_by(active=True).order_by(Project.name).all()
|
|
form.project_id.choices = [(0, '— None —')] + [(p.id, p.name) for p in projects]
|
|
|
|
if form.validate_on_submit():
|
|
project_id = form.project_id.data if form.project_id.data else None
|
|
facility.name = form.name.data
|
|
facility.address = form.address.data
|
|
facility.contact_person = form.contact_person.data
|
|
facility.contact_phone = form.contact_phone.data
|
|
facility.project_id = project_id if project_id else None
|
|
facility.active = form.active.data
|
|
|
|
db.session.commit()
|
|
logger.info('FACILITIES | edit | user=%s | facility_id=%s name=%r',
|
|
current_user.username, facility.id, facility.name)
|
|
log_action(ACTION_UPDATE, 'Facility', facility.id, facility.name,
|
|
f'project_id={facility.project_id}; active={facility.active}')
|
|
flash(f'Facility "{facility.name}" updated successfully.', 'success')
|
|
return redirect(url_for('facilities.view_facility', facility_id=facility.id))
|
|
|
|
return render_template('facilities/form.html', form=form, facility=facility, title='Edit Facility')
|
|
|
|
@bp.route('/<int:facility_id>/delete', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def delete_facility(facility_id):
|
|
facility = db.session.get(Facility, facility_id)
|
|
if facility is None:
|
|
abort(404)
|
|
|
|
if facility.inspections.count() > 0:
|
|
flash(f'Cannot delete "{facility.name}" — it has existing inspection records.', 'danger')
|
|
return redirect(url_for('facilities.view_facility', facility_id=facility.id))
|
|
|
|
facility_name = facility.name
|
|
facility_id_snap = facility.id
|
|
db.session.delete(facility)
|
|
db.session.commit()
|
|
logger.info('FACILITIES | delete | user=%s | facility_id=%s name=%r',
|
|
current_user.username, facility_id_snap, facility_name)
|
|
log_action(ACTION_DELETE, 'Facility', facility_id_snap, facility_name)
|
|
flash(f'Facility "{facility_name}" has been permanently deleted.', 'success')
|
|
return redirect(url_for('facilities.list_facilities'))
|
|
|
|
# Area Management Routes
|
|
|
|
@bp.route('/<int:facility_id>/areas/new', methods=['GET', 'POST'])
|
|
@login_required
|
|
@supervisor_required
|
|
def create_area(facility_id):
|
|
facility = db.session.get(Facility, facility_id)
|
|
if facility is None:
|
|
abort(404)
|
|
form = AreaForm()
|
|
form.facility_id.choices = [(facility.id, facility.name)]
|
|
form.facility_id.data = facility.id
|
|
|
|
if form.validate_on_submit():
|
|
area = Area(
|
|
name=form.name.data,
|
|
area_type=form.area_type.data,
|
|
facility_id=facility.id
|
|
)
|
|
area.ensure_public_token() # QR landing-page token
|
|
|
|
db.session.add(area)
|
|
db.session.commit()
|
|
logger.info('FACILITIES | create_area | user=%s | area_id=%s name=%r facility=%r',
|
|
current_user.username, area.id, area.name, facility.name)
|
|
log_action(ACTION_CREATE, 'Area', area.id, area.name,
|
|
f'facility={facility.name}; type={area.area_type or ""}')
|
|
flash(f'Area "{area.name}" created successfully.', 'success')
|
|
return redirect(url_for('facilities.view_facility', facility_id=facility.id))
|
|
|
|
return render_template('facilities/area_form.html', form=form, facility=facility, title='Create Area')
|
|
|
|
@bp.route('/areas/<int:area_id>/edit', methods=['GET', 'POST'])
|
|
@login_required
|
|
@supervisor_required
|
|
def edit_area(area_id):
|
|
area = db.session.get(Area, area_id)
|
|
if area is None:
|
|
abort(404)
|
|
form = AreaForm(obj=area)
|
|
|
|
facilities = Facility.query.filter_by(active=True).order_by(Facility.name).all()
|
|
form.facility_id.choices = [(f.id, f.name) for f in facilities]
|
|
|
|
if form.validate_on_submit():
|
|
area.name = form.name.data
|
|
area.area_type = form.area_type.data
|
|
area.facility_id = form.facility_id.data
|
|
|
|
db.session.commit()
|
|
logger.info('FACILITIES | edit_area | user=%s | area_id=%s name=%r',
|
|
current_user.username, area.id, area.name)
|
|
log_action(ACTION_UPDATE, 'Area', area.id, area.name,
|
|
f'facility_id={area.facility_id}; type={area.area_type or ""}')
|
|
flash(f'Area "{area.name}" updated successfully.', 'success')
|
|
return redirect(url_for('facilities.view_facility', facility_id=area.facility_id))
|
|
|
|
return render_template('facilities/area_form.html', form=form, area=area, facility=area.facility, title='Edit Area')
|
|
|
|
@bp.route('/areas/<int:area_id>/delete', methods=['POST'])
|
|
@login_required
|
|
@supervisor_required
|
|
def delete_area(area_id):
|
|
area = db.session.get(Area, area_id)
|
|
if area is None:
|
|
abort(404)
|
|
facility_id = area.facility_id
|
|
|
|
if area.inspections.count() > 0:
|
|
flash('Cannot delete area with existing inspections.', 'danger')
|
|
return redirect(url_for('facilities.view_facility', facility_id=facility_id))
|
|
|
|
issue_count = area.issues.count()
|
|
if issue_count > 0:
|
|
flash(
|
|
f'Cannot delete area "{area.name}" — it has {issue_count} issue record(s) on file. '
|
|
f'Resolve or reassign those issues first.',
|
|
'danger'
|
|
)
|
|
return redirect(url_for('facilities.view_facility', facility_id=facility_id))
|
|
|
|
area_name = area.name
|
|
area_id_snap = area.id
|
|
db.session.delete(area)
|
|
db.session.commit()
|
|
logger.info('FACILITIES | delete_area | user=%s | area_id=%s name=%r',
|
|
current_user.username, area_id_snap, area_name)
|
|
log_action(ACTION_DELETE, 'Area', area_id_snap, area_name)
|
|
flash(f'Area "{area_name}" deleted successfully.', 'success')
|
|
return redirect(url_for('facilities.view_facility', facility_id=facility_id)) |