318 lines
12 KiB
Python
318 lines
12 KiB
Python
"""
|
||
app/api/inspections.py
|
||
----------------------
|
||
Mobile API endpoints for submitting inspections from the iPad app.
|
||
|
||
POST /api/v1/inspections
|
||
Creates a new inspection record (status: in_progress or completed).
|
||
Accepts a mobile_local_id for idempotency — if an inspection with the
|
||
same local_id already exists, returns the existing record without
|
||
creating a duplicate.
|
||
|
||
PATCH /api/v1/inspections/<inspection_id>
|
||
Updates an existing inspection (e.g. draft → completed).
|
||
"""
|
||
|
||
import logging
|
||
|
||
from flask import Blueprint, request, g, current_app
|
||
from app import db
|
||
from app.models.inspection import Inspection, InspectionTemplate
|
||
from app.models.facility import Facility, Area
|
||
from app.api.errors import api_ok, api_error
|
||
from app.api.decorators import jwt_required
|
||
from app.utils.audit import log_action, ACTION_CREATE, ACTION_UPDATE
|
||
from app.utils.notifications import notify_by_matrix
|
||
from app.utils.time_utils import now_eastern
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
bp = Blueprint('api_inspections', __name__)
|
||
|
||
_ALLOWED_ROLES = {'admin', 'director', 'inspector', 'project_manager'}
|
||
|
||
|
||
def _parse_datetime(value):
|
||
"""Parse an ISO 8601 datetime string; return None on failure."""
|
||
if not value:
|
||
return None
|
||
from datetime import datetime
|
||
for fmt in ('%Y-%m-%dT%H:%M:%S', '%Y-%m-%dT%H:%M:%S.%f', '%Y-%m-%d'):
|
||
try:
|
||
return datetime.strptime(value, fmt)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
return None
|
||
|
||
|
||
# ── Create Inspection ─────────────────────────────────────────────────────────
|
||
|
||
@bp.route('/inspections', methods=['POST'])
|
||
@jwt_required
|
||
def create_inspection():
|
||
"""
|
||
Create a new inspection submitted from the iPad app.
|
||
|
||
Idempotency: if mobile_local_id is provided and an inspection with that
|
||
ID already exists, the existing record is returned without duplication.
|
||
This protects against double-submission when the network fails after the
|
||
server commits but before the device receives the response.
|
||
|
||
Request JSON
|
||
------------
|
||
{
|
||
"template_id": 3,
|
||
"facility_id": 7,
|
||
"area_id": 12, // optional
|
||
"status": "completed", // "in_progress" | "completed"
|
||
"form_data": { ... }, // { field_id: value }
|
||
"notes": "...", // optional inspector notes
|
||
"overall_score": 87.5, // optional — computed server-side if omitted
|
||
"inspection_date": "2026-05-01T14:30:00",
|
||
"completed_at": "2026-05-01T15:00:00", // required when status=completed
|
||
"mobile_local_id": "uuid-string" // idempotency key
|
||
}
|
||
|
||
Response 200
|
||
------------
|
||
{ "ok": true, "data": { "inspection_id": 42, "duplicate": false } }
|
||
"""
|
||
user = g.api_user
|
||
|
||
if user.role not in _ALLOWED_ROLES:
|
||
return api_error('Access denied', 403)
|
||
|
||
data = request.get_json(silent=True) or {}
|
||
|
||
# ── Idempotency check ─────────────────────────────────────────────────
|
||
mobile_local_id = data.get('mobile_local_id')
|
||
if mobile_local_id:
|
||
existing = Inspection.query.filter_by(mobile_local_id=mobile_local_id).first()
|
||
if existing:
|
||
logger.info('API INSPECTIONS | duplicate | local_id=%s | inspection_id=%d | user=%s',
|
||
mobile_local_id, existing.id, user.username)
|
||
return api_ok({'inspection_id': existing.id, 'duplicate': True})
|
||
|
||
# ── Validate required fields ──────────────────────────────────────────
|
||
template_id = data.get('template_id')
|
||
facility_id = data.get('facility_id')
|
||
|
||
if not template_id or not facility_id:
|
||
return api_error('template_id and facility_id are required', 400)
|
||
|
||
template = db.session.get(InspectionTemplate, template_id)
|
||
if template is None:
|
||
return api_error('Template not found', 404)
|
||
|
||
facility = db.session.get(Facility, facility_id)
|
||
if facility is None or not facility.active:
|
||
return api_error('Facility not found', 404)
|
||
|
||
area_id = data.get('area_id')
|
||
if area_id:
|
||
area = db.session.get(Area, area_id)
|
||
if area is None or area.facility_id != facility_id:
|
||
return api_error('Area not found or does not belong to the facility', 400)
|
||
|
||
status = data.get('status', 'completed')
|
||
if status not in ('in_progress', 'completed'):
|
||
return api_error('status must be "in_progress" or "completed"', 400)
|
||
|
||
# ── Score calculation ─────────────────────────────────────────────────
|
||
overall_score = data.get('overall_score')
|
||
if overall_score is None and status == 'completed':
|
||
form_data = data.get('form_data') or {}
|
||
form_fields = template.get_form_schema()
|
||
overall_score = _compute_score(form_fields, form_data)
|
||
|
||
# ── Build inspection record ───────────────────────────────────────────
|
||
inspection_date = _parse_datetime(data.get('inspection_date')) or now_eastern()
|
||
completed_at = _parse_datetime(data.get('completed_at'))
|
||
|
||
import json
|
||
form_data = data.get('form_data') or {}
|
||
notes_payload = {}
|
||
if data.get('notes'):
|
||
notes_payload['_inspector_notes'] = data['notes']
|
||
notes_payload['_form_data'] = form_data
|
||
|
||
inspection = Inspection(
|
||
template_id = template_id,
|
||
facility_id = facility_id,
|
||
area_id = area_id,
|
||
inspector_id = user.id,
|
||
inspection_date = inspection_date,
|
||
overall_score = overall_score,
|
||
status = status,
|
||
notes = json.dumps(notes_payload),
|
||
completed_at = completed_at if status == 'completed' else None,
|
||
mobile_local_id = mobile_local_id,
|
||
)
|
||
|
||
db.session.add(inspection)
|
||
db.session.flush() # get inspection.id before commit
|
||
|
||
# ── Notifications (completed inspections only) ─────────────────────────
|
||
if status == 'completed':
|
||
score_display = f'{overall_score:.1f}%' if overall_score is not None else 'N/A'
|
||
try:
|
||
from flask import url_for
|
||
inspection_link = url_for('inspections.view',
|
||
inspection_id=inspection.id, _external=False)
|
||
except RuntimeError:
|
||
inspection_link = f'/inspections/{inspection.id}'
|
||
|
||
notify_by_matrix(
|
||
event_type = 'inspection_completed',
|
||
title = f'Inspection #{inspection.id} Completed (Mobile)',
|
||
body = (
|
||
f'{user.username} completed an inspection at '
|
||
f'{facility.name} using the "{template.name}" template. '
|
||
f'Overall score: {score_display}.'
|
||
),
|
||
link = inspection_link,
|
||
inspection_id = inspection.id,
|
||
facility_id = facility_id,
|
||
exclude_user_ids = {user.id},
|
||
)
|
||
|
||
db.session.commit()
|
||
|
||
log_action(ACTION_CREATE, 'Inspection', inspection.id,
|
||
f'{template.name} @ {facility.name}',
|
||
f'source=mobile; status={status}; score={overall_score}; '
|
||
f'local_id={mobile_local_id}')
|
||
|
||
logger.info('API INSPECTIONS | created | inspection_id=%d | facility=%s | '
|
||
'template=%s | status=%s | score=%s | user=%s',
|
||
inspection.id, facility.name, template.name,
|
||
status, overall_score, user.username)
|
||
|
||
return api_ok({'inspection_id': inspection.id, 'duplicate': False})
|
||
|
||
|
||
# ── Update Inspection ─────────────────────────────────────────────────────────
|
||
|
||
@bp.route('/inspections/<int:inspection_id>', methods=['PATCH'])
|
||
@jwt_required
|
||
def update_inspection(inspection_id):
|
||
"""
|
||
Update an existing inspection — e.g. promoting a draft to completed.
|
||
|
||
Request JSON (all fields optional — only provided fields are updated)
|
||
------------
|
||
{
|
||
"status": "completed",
|
||
"form_data": { ... },
|
||
"notes": "...",
|
||
"overall_score": 91.0,
|
||
"completed_at": "2026-05-01T15:30:00"
|
||
}
|
||
|
||
Response 200
|
||
------------
|
||
{ "ok": true, "data": { "inspection_id": 42 } }
|
||
"""
|
||
user = g.api_user
|
||
|
||
if user.role not in _ALLOWED_ROLES:
|
||
return api_error('Access denied', 403)
|
||
|
||
inspection = db.session.get(Inspection, inspection_id)
|
||
if inspection is None:
|
||
return api_error('Inspection not found', 404)
|
||
|
||
# Inspectors can only update their own inspections
|
||
if user.role == 'inspector' and inspection.inspector_id != user.id:
|
||
return api_error('Access denied', 403)
|
||
|
||
data = request.get_json(silent=True) or {}
|
||
|
||
import json
|
||
if 'form_data' in data or 'notes' in data:
|
||
existing_notes = {}
|
||
if inspection.notes:
|
||
try:
|
||
existing_notes = json.loads(inspection.notes)
|
||
except (json.JSONDecodeError, TypeError):
|
||
existing_notes = {}
|
||
if 'form_data' in data:
|
||
existing_notes['_form_data'] = data['form_data']
|
||
if 'notes' in data:
|
||
existing_notes['_inspector_notes'] = data['notes']
|
||
inspection.notes = json.dumps(existing_notes)
|
||
|
||
if 'status' in data:
|
||
inspection.status = data['status']
|
||
|
||
if 'overall_score' in data:
|
||
inspection.overall_score = data['overall_score']
|
||
elif data.get('status') == 'completed' and inspection.overall_score is None:
|
||
form_data = data.get('form_data') or {}
|
||
form_fields = inspection.template.get_form_schema()
|
||
inspection.overall_score = _compute_score(form_fields, form_data)
|
||
|
||
if 'completed_at' in data:
|
||
inspection.completed_at = _parse_datetime(data['completed_at']) or now_eastern()
|
||
elif data.get('status') == 'completed' and not inspection.completed_at:
|
||
inspection.completed_at = now_eastern()
|
||
|
||
db.session.commit()
|
||
|
||
log_action(ACTION_UPDATE, 'Inspection', inspection.id,
|
||
f'{inspection.template.name} @ {inspection.facility.name}',
|
||
f'source=mobile; fields_updated={list(data.keys())}')
|
||
|
||
logger.info('API INSPECTIONS | updated | inspection_id=%d | user=%s | fields=%s',
|
||
inspection_id, user.username, list(data.keys()))
|
||
|
||
return api_ok({'inspection_id': inspection_id})
|
||
|
||
|
||
# ── Score helper (mirrors Python _compute_score_from_form exactly) ────────────
|
||
|
||
def _compute_score(form_fields, responses):
|
||
"""
|
||
Mirror of routes/inspections.py::_compute_score_from_form().
|
||
Returns float 0–100 or None if no scoreable fields exist.
|
||
Rating value 0 means unanswered — excluded from calculation.
|
||
"""
|
||
scoreable = [f for f in form_fields
|
||
if f.get('type') in ('rating', 'checkbox', 'radio', 'pass_fail')]
|
||
if not scoreable:
|
||
return None
|
||
|
||
total, earned = 0, 0
|
||
for field in scoreable:
|
||
fid = field.get('id')
|
||
val = responses.get(str(fid), responses.get(fid, ''))
|
||
|
||
ftype = field.get('type')
|
||
if ftype == 'rating':
|
||
try:
|
||
v = int(val)
|
||
if v == 0:
|
||
continue
|
||
earned += v
|
||
total += 5
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
elif ftype == 'checkbox':
|
||
total += 1
|
||
if val == 'true':
|
||
earned += 1
|
||
|
||
elif ftype == 'radio':
|
||
total += 1
|
||
if str(val).lower() in ('pass', 'yes', 'ok', 'good', 'acceptable', 'compliant'):
|
||
earned += 1
|
||
|
||
elif ftype == 'pass_fail':
|
||
if not val:
|
||
continue
|
||
total += 1
|
||
if str(val).lower() in ('pass', 'yes', 'ok', 'good', 'acceptable', 'compliant'):
|
||
earned += 1
|
||
|
||
return round((earned / total) * 100, 2) if total else None |