Coverage for app/maintenance/routes.py: 100%
553 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-31 10:13 +0000
1import hashlib
2import io
3import json
4import os
5import uuid
6from datetime import date as _date
7from datetime import timedelta
8from typing import Any
10import openpyxl # pyright: ignore[reportMissingImports]
11from flask import ( # pyright: ignore[reportMissingImports]
12 Blueprint,
13 Response,
14 abort,
15 current_app,
16 flash,
17 redirect,
18 render_template,
19 request,
20 session,
21 url_for,
22)
23from flask.typing import ResponseReturnValue # pyright: ignore[reportMissingImports]
24from flask_babel import gettext as _ # pyright: ignore[reportMissingImports]
25from flask_babel import ngettext # pyright: ignore[reportMissingImports]
26from models import (
27 Aircraft,
28 AmpBasis,
29 AmpCategory,
30 AmpCertifyingPartyKind,
31 AmpDeclaration,
32 AmpDeclarationType,
33 AmpRevision,
34 Component,
35 ComponentType,
36 HoursBasis,
37 MaintenanceImportBatch,
38 MaintenanceRecord,
39 MaintenanceTrigger,
40 Role,
41 Snag,
42 TenantUser,
43 TriggerType,
44 db,
45) # pyright: ignore[reportMissingImports]
46from services.authorization import (
47 AuthorizationService, # pyright: ignore[reportMissingImports]
48)
49from utils import (
50 accessible_aircraft,
51 activity,
52 compute_aircraft_statuses,
53 login_required,
54 require_maint_access,
55 require_role,
56 user_can_access_aircraft,
57) # pyright: ignore[reportMissingImports]
58from weasyprint import HTML # pyright: ignore[reportMissingImports]
59from werkzeug.utils import secure_filename # pyright: ignore[reportMissingImports]
61from maintenance.amp_import import ( # pyright: ignore[reportMissingImports]
62 compute_due_fields,
63 format_interval,
64 hours_basis_for_component,
65 parse_amp_rows,
66)
67from maintenance.form_parsing import ( # pyright: ignore[reportMissingImports]
68 parse_service_fields,
69 parse_trigger_fields,
70)
72maintenance_bp = Blueprint("maintenance", __name__)
74_MAINT_ROLES = (Role.ADMIN, Role.OWNER, Role.MAINTENANCE)
77def _tenant_id() -> int:
78 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
79 if not tu:
80 abort(403)
81 return int(tu.tenant_id)
84def _get_aircraft_or_404(aircraft_id: int) -> Aircraft:
85 ac = db.session.get(Aircraft, aircraft_id)
86 if (
87 not ac
88 or ac.tenant_id != _tenant_id()
89 or not user_can_access_aircraft(aircraft_id)
90 ):
91 abort(404)
92 return ac
95def _get_trigger_or_404(aircraft: Aircraft, trigger_id: int) -> MaintenanceTrigger:
96 t = db.session.get(MaintenanceTrigger, trigger_id)
97 if not t or t.aircraft_id != aircraft.id:
98 abort(404)
99 return t
102# ── Fleet maintenance overview ────────────────────────────────────────────────
105@maintenance_bp.route("/maintenance")
106@login_required
107@require_maint_access
108def fleet_overview() -> ResponseReturnValue:
109 aircraft = accessible_aircraft(_tenant_id()).all()
110 aircraft_ids = [ac.id for ac in aircraft]
111 ac_by_id = {ac.id: ac for ac in aircraft}
112 hobbs_by_id = Aircraft.engine_hours_by_id(aircraft_ids)
113 landings_by_id = Aircraft.landings_by_id(aircraft_ids)
114 flight_hours_by_id = Aircraft.flight_hours_by_id(aircraft_ids)
116 triggers = (
117 (
118 MaintenanceTrigger.query.filter(
119 MaintenanceTrigger.aircraft_id.in_(aircraft_ids)
120 ).all()
121 )
122 if aircraft_ids
123 else []
124 )
126 from datetime import date as _date_cls
127 from datetime import datetime as _datetime
129 # Annotate each trigger with its status
130 trigger_rows = [
131 (
132 t,
133 t.status(
134 current_engine_hours=hobbs_by_id.get(t.aircraft_id),
135 current_landings=landings_by_id.get(t.aircraft_id),
136 current_flight_hours=flight_hours_by_id.get(t.aircraft_id),
137 ),
138 ac_by_id[t.aircraft_id],
139 )
140 for t in triggers
141 ]
143 # Sort: overdue → due_soon → ok; within status: calendar triggers by due_date asc,
144 # hours-based triggers (no reliable date) after calendar ones.
145 _status_order = {"overdue": 0, "due_soon": 1, "ok": 2}
146 _far_future = _date_cls(9999, 12, 31)
148 def _trigger_sort_key(row: Any) -> Any:
149 t, status, _ac = row
150 due = (
151 t.due_date
152 if t.trigger_type == TriggerType.CALENDAR and t.due_date
153 else _far_future
154 )
155 return (_status_order[status], due)
157 trigger_rows.sort(key=_trigger_sort_key)
159 # Open grounding snags — oldest reported first (most overdue on top)
160 grounding_snags = (
161 (
162 Snag.query.filter(
163 Snag.aircraft_id.in_(aircraft_ids),
164 Snag.is_grounding.is_(True),
165 Snag.resolved_at.is_(None),
166 )
167 .order_by(Snag.reported_at.asc())
168 .all()
169 )
170 if aircraft_ids
171 else []
172 )
173 grounding_snag_rows = [(s, ac_by_id[s.aircraft_id]) for s in grounding_snags]
175 # Open non-grounding snags — oldest reported first
176 open_snags = (
177 (
178 Snag.query.filter(
179 Snag.aircraft_id.in_(aircraft_ids),
180 Snag.is_grounding.is_(False),
181 Snag.resolved_at.is_(None),
182 )
183 .order_by(Snag.reported_at.asc())
184 .all()
185 )
186 if aircraft_ids
187 else []
188 )
189 open_snag_rows = [(s, ac_by_id[s.aircraft_id]) for s in open_snags]
191 aircraft_status = compute_aircraft_statuses(
192 aircraft, triggers, hobbs_by_id, landings_by_id, flight_hours_by_id
193 )
195 # Chronological view: single list sorted by due/reported date asc.
196 # Hours-based triggers have no reliable date → sorted after all dated items.
197 # Tuple structure: (sort_date, kind_order, label, obj, ac, extra)
198 # kind_order: grounding=0, snag=1, maintenance=2 (tiebreak within same date)
199 _far_dt = _datetime(_far_future.year, _far_future.month, _far_future.day)
200 chron_items = []
201 for s, ac in grounding_snag_rows:
202 dt = _datetime.combine(
203 s.reported_at.date() if hasattr(s.reported_at, "date") else s.reported_at,
204 _datetime.min.time(),
205 )
206 chron_items.append(("grounding", dt, s, ac, None))
207 for s, ac in open_snag_rows:
208 dt = _datetime.combine(
209 s.reported_at.date() if hasattr(s.reported_at, "date") else s.reported_at,
210 _datetime.min.time(),
211 )
212 chron_items.append(("snag", dt, s, ac, None))
213 for t, status, ac in trigger_rows:
214 if status in ("overdue", "due_soon") or t.needs_review:
215 if t.due_date:
216 dt = _datetime(t.due_date.year, t.due_date.month, t.due_date.day)
217 else:
218 dt = _far_dt # no calendar due date: push to end
219 chron_items.append(("maintenance", dt, t, ac, status))
221 _kind_order = {"grounding": 0, "snag": 1, "maintenance": 2}
222 chron_items.sort(key=lambda x: (x[1], _kind_order[x[0]]))
224 view = request.args.get("view", "by-type")
226 # Component TBO / calendar life limits that need attention
227 from services.component_limits import (
228 aircraft_limit_infos, # pyright: ignore[reportMissingImports]
229 )
231 component_limit_rows = []
232 for ac in aircraft:
233 for info in aircraft_limit_infos(ac):
234 if info["status"] in ("overdue", "due_soon"):
235 component_limit_rows.append((info, ac))
236 component_limit_rows.sort(key=lambda row: 0 if row[0]["status"] == "overdue" else 1)
238 any_needs_review = any(t.needs_review for t, _status, _ac in trigger_rows)
240 return render_template(
241 "maintenance/fleet.html",
242 aircraft=aircraft,
243 aircraft_status=aircraft_status,
244 component_limit_rows=component_limit_rows,
245 trigger_rows=trigger_rows,
246 grounding_snag_rows=grounding_snag_rows,
247 open_snag_rows=open_snag_rows,
248 chron_items=chron_items,
249 hobbs_by_id=hobbs_by_id,
250 landings_by_id=landings_by_id,
251 flight_hours_by_id=flight_hours_by_id,
252 any_needs_review=any_needs_review,
253 view=view,
254 )
257# ── Trigger list ──────────────────────────────────────────────────────────────
260def _group_trigger_rows_by_component(
261 trigger_rows: list[tuple[MaintenanceTrigger, str]],
262) -> list[tuple[Component | None, list[tuple[MaintenanceTrigger, str]]]]:
263 """Group (trigger, status) rows by trigger.component — unscoped
264 ("Airframe / general") rows first, then installed components in the
265 same (type, position) order used for the components list on
266 aircraft/detail.html. A component only gets a section if it actually
267 has at least one trigger."""
268 general: list[tuple[MaintenanceTrigger, str]] = []
269 by_component: dict[int, tuple[Component, list[tuple[MaintenanceTrigger, str]]]] = {}
270 for t, status in trigger_rows:
271 if t.component is None:
272 general.append((t, status))
273 else:
274 entry = by_component.setdefault(
275 t.component.id,
276 (t.component, []), # type: ignore[arg-type]
277 )
278 entry[1].append((t, status))
280 groups: list[tuple[Component | None, list[tuple[MaintenanceTrigger, str]]]] = []
281 if general:
282 groups.append((None, general))
283 for comp, rows in sorted(
284 by_component.values(), key=lambda cr: (cr[0].type, cr[0].position or "")
285 ):
286 groups.append((comp, rows))
287 return groups
290@maintenance_bp.route("/aircraft/<aircraft_ref:aircraft_id>/maintenance")
291@login_required
292def list_triggers(aircraft_id: int) -> ResponseReturnValue:
293 ac = _get_aircraft_or_404(aircraft_id)
294 current_hobbs = ac.total_engine_hours
295 current_landings = ac.total_landings
296 current_flight_hours = ac.total_flight_hours
297 all_triggers = (
298 MaintenanceTrigger.query.filter_by(aircraft_id=ac.id)
299 .order_by(MaintenanceTrigger.name)
300 .all()
301 )
302 tid = _tenant_id()
303 uid = session["user_id"]
304 maint_view = AuthorizationService.maintenance_view_level(uid, aircraft_id, tid)
306 def _status(t: MaintenanceTrigger) -> str:
307 return t.status(
308 current_engine_hours=current_hobbs,
309 current_landings=current_landings,
310 current_flight_hours=current_flight_hours,
311 )
313 # Limited view: show only overdue and due-soon items
314 if maint_view == "limited":
315 triggers = [t for t in all_triggers if _status(t) in ("overdue", "due_soon")]
316 else:
317 triggers = all_triggers
318 trigger_rows = [(t, _status(t)) for t in triggers]
319 component_groups = _group_trigger_rows_by_component(trigger_rows)
320 return render_template(
321 "maintenance/list.html",
322 aircraft=ac,
323 trigger_rows=trigger_rows,
324 component_groups=component_groups,
325 current_hobbs=current_hobbs,
326 current_landings=current_landings,
327 current_flight_hours=current_flight_hours,
328 maint_view=maint_view,
329 )
332# ── Add trigger ───────────────────────────────────────────────────────────────
335@maintenance_bp.route(
336 "/aircraft/<aircraft_ref:aircraft_id>/maintenance/new", methods=["GET", "POST"]
337)
338@login_required
339@require_role(*_MAINT_ROLES)
340def new_trigger(aircraft_id: int) -> ResponseReturnValue:
341 ac = _get_aircraft_or_404(aircraft_id)
342 if request.method == "POST":
343 return _save_trigger(ac, None)
344 return render_template(
345 "maintenance/trigger_form.html",
346 aircraft=ac,
347 trigger=None,
348 trigger_types=TriggerType,
349 hours_basis=HoursBasis,
350 )
353# ── Edit trigger ──────────────────────────────────────────────────────────────
356@maintenance_bp.route(
357 "/aircraft/<aircraft_ref:aircraft_id>/maintenance/<int:trigger_id>/edit",
358 methods=["GET", "POST"],
359)
360@login_required
361@require_role(*_MAINT_ROLES)
362def edit_trigger(aircraft_id: int, trigger_id: int) -> ResponseReturnValue:
363 ac = _get_aircraft_or_404(aircraft_id)
364 t = _get_trigger_or_404(ac, trigger_id)
365 if request.method == "POST":
366 return _save_trigger(ac, t)
367 return render_template(
368 "maintenance/trigger_form.html",
369 aircraft=ac,
370 trigger=t,
371 trigger_types=TriggerType,
372 hours_basis=HoursBasis,
373 )
376def _save_trigger(ac: Aircraft, t: MaintenanceTrigger | None) -> ResponseReturnValue:
377 values, errors = parse_trigger_fields(request.form)
379 component_id = values.get("component_id")
380 if component_id is not None:
381 comp = db.session.get(Component, component_id)
382 if comp is None or comp.aircraft_id != ac.id:
383 errors.append(_("Component selection is invalid."))
384 component_id = None
386 if errors:
387 for msg in errors:
388 flash(msg, "danger")
389 return render_template(
390 "maintenance/trigger_form.html",
391 aircraft=ac,
392 trigger=t,
393 trigger_types=TriggerType,
394 hours_basis=HoursBasis,
395 )
397 if t is None:
398 t = MaintenanceTrigger(aircraft_id=ac.id)
399 db.session.add(t)
401 t.name = values["name"]
402 t.trigger_type = values["trigger_type"]
403 t.component_id = component_id
404 t.due_date = values["due_date"]
405 t.interval_days = values["interval_days"]
406 t.warn_days = values["warn_days"]
407 t.due_engine_hours = values["due_engine_hours"]
408 t.interval_hours = values["interval_hours"]
409 t.warn_hours = values["warn_hours"]
410 t.hours_basis = values["hours_basis"]
411 t.due_landings = values["due_landings"]
412 t.interval_landings = values["interval_landings"]
413 t.warn_landings = values["warn_landings"]
414 t.notes = values["notes"]
415 db.session.commit()
417 flash(_("Maintenance item '%(name)s' saved.", name=t.name), "success")
418 return redirect(url_for("maintenance.list_triggers", aircraft_id=ac.id))
421# ── Delete trigger ────────────────────────────────────────────────────────────
424@maintenance_bp.route(
425 "/aircraft/<aircraft_ref:aircraft_id>/maintenance/<int:trigger_id>/delete",
426 methods=["POST"],
427)
428@login_required
429@require_role(*_MAINT_ROLES)
430def delete_trigger(aircraft_id: int, trigger_id: int) -> ResponseReturnValue:
431 ac = _get_aircraft_or_404(aircraft_id)
432 t = _get_trigger_or_404(ac, trigger_id)
433 name = t.name
434 db.session.delete(t)
435 db.session.commit()
436 flash(_("'%(name)s' deleted.", name=name), "success")
437 return redirect(url_for("maintenance.list_triggers", aircraft_id=ac.id))
440# ── Mark as serviced ──────────────────────────────────────────────────────────
443@maintenance_bp.route(
444 "/aircraft/<aircraft_ref:aircraft_id>/maintenance/<int:trigger_id>/service",
445 methods=["GET", "POST"],
446)
447@login_required
448@require_role(*_MAINT_ROLES)
449def service_trigger(aircraft_id: int, trigger_id: int) -> ResponseReturnValue:
450 ac = _get_aircraft_or_404(aircraft_id)
451 t = _get_trigger_or_404(ac, trigger_id)
453 # Phase 40: which readings are required depends on which due-field
454 # groups are actually populated on this trigger, not on trigger_type — a
455 # combined-interval trigger can have more than one group populated.
456 requires_hobbs = t.due_engine_hours is not None
457 requires_landings = t.due_landings is not None
459 if request.method == "POST":
460 values, errors = parse_service_fields(
461 request.form, requires_hobbs, requires_landings
462 )
463 performed_at = values["performed_at"]
464 hobbs_at_service = values["hobbs_at_service"]
465 landings_at_service = values["landings_at_service"]
467 if errors:
468 for msg in errors:
469 flash(msg, "danger")
470 return render_template(
471 "maintenance/service_form.html",
472 aircraft=ac,
473 trigger=t,
474 current_hobbs=ac.total_engine_hours,
475 current_landings=ac.total_landings,
476 current_flight_hours=ac.total_flight_hours,
477 today=_date.today().isoformat(),
478 )
480 record = MaintenanceRecord(
481 trigger_id=t.id,
482 performed_at=performed_at,
483 hobbs_at_service=hobbs_at_service,
484 landings_at_service=landings_at_service,
485 notes=values["notes"],
486 )
487 db.session.add(record)
489 # Advance every populated due-field group independently — a
490 # combined-interval trigger (both calendar and hours set) advances
491 # both together from the same service record.
492 if t.due_date is not None and t.interval_days and performed_at:
493 t.due_date = performed_at + timedelta(days=t.interval_days)
494 if (
495 t.due_engine_hours is not None
496 and t.interval_hours
497 and hobbs_at_service is not None
498 ):
499 t.due_engine_hours = hobbs_at_service + float(t.interval_hours)
500 if (
501 t.due_landings is not None
502 and t.interval_landings
503 and landings_at_service is not None
504 ):
505 t.due_landings = landings_at_service + t.interval_landings
507 db.session.commit()
508 activity(
509 "maintenance.serviced",
510 trigger_id=t.id,
511 aircraft_id=aircraft_id,
512 trigger_name=t.name,
513 record_id=record.id,
514 )
515 flash(_("'%(name)s' marked as serviced.", name=t.name), "success")
516 return redirect(url_for("maintenance.list_triggers", aircraft_id=ac.id))
518 return render_template(
519 "maintenance/service_form.html",
520 aircraft=ac,
521 trigger=t,
522 current_hobbs=ac.total_engine_hours,
523 current_landings=ac.total_landings,
524 current_flight_hours=ac.total_flight_hours,
525 today=_date.today().isoformat(),
526 )
529# ── AMP declaration profile ─────────────────────────────────────────────────
532@maintenance_bp.route(
533 "/aircraft/<aircraft_ref:aircraft_id>/amp/edit", methods=["GET", "POST"]
534)
535@login_required
536@require_role(*_MAINT_ROLES)
537def edit_amp_declaration(aircraft_id: int) -> ResponseReturnValue:
538 ac = _get_aircraft_or_404(aircraft_id)
539 decl: AmpDeclaration | None = ac.amp_declaration # type: ignore[assignment]
540 if request.method == "POST":
541 return _save_amp_declaration(ac, decl)
542 return render_template(
543 "maintenance/amp_declaration_form.html",
544 aircraft=ac,
545 decl=decl,
546 revisions=ac.amp_revisions,
547 amp_basis=AmpBasis,
548 declaration_types=AmpDeclarationType,
549 certifying_party_kinds=AmpCertifyingPartyKind,
550 )
553def _save_amp_declaration(
554 ac: Aircraft, decl: AmpDeclaration | None
555) -> ResponseReturnValue:
556 def _text(key: str) -> str | None:
557 v = (request.form.get(key) or "").strip()
558 return v or None
560 basis = request.form.get("basis", "").strip() or AmpBasis.DAH_ICA
561 declaration_type = (
562 request.form.get("declaration_type", "").strip() or AmpDeclarationType.OWNER
563 )
564 certifying_party_kind = (
565 request.form.get("certifying_party_kind", "").strip()
566 or AmpCertifyingPartyKind.OWNER_LESSEE_OPERATOR
567 )
569 errors = []
570 if basis not in AmpBasis.ALL:
571 errors.append(_("Invalid programme basis selected."))
572 if declaration_type not in AmpDeclarationType.ALL:
573 errors.append(_("Invalid declaration type selected."))
574 if certifying_party_kind not in AmpCertifyingPartyKind.ALL:
575 errors.append(_("Invalid certifying party selected."))
577 if errors:
578 for msg in errors:
579 flash(msg, "danger")
580 return render_template(
581 "maintenance/amp_declaration_form.html",
582 aircraft=ac,
583 decl=decl,
584 revisions=ac.amp_revisions,
585 amp_basis=AmpBasis,
586 declaration_types=AmpDeclarationType,
587 certifying_party_kinds=AmpCertifyingPartyKind,
588 )
590 if decl is None:
591 decl = AmpDeclaration(aircraft_id=ac.id)
592 db.session.add(decl)
594 decl.owner_name = _text("owner_name")
595 decl.owner_address = _text("owner_address")
596 decl.basis = basis
597 decl.mip_details = _text("mip_details")
598 decl.dah_ica_airframe_ref = _text("dah_ica_airframe_ref")
599 decl.dah_ica_engine_ref = _text("dah_ica_engine_ref")
600 decl.dah_ica_propeller_ref = _text("dah_ica_propeller_ref")
601 decl.pilot_owner_maintenance = request.form.get("pilot_owner_maintenance") == "on"
602 decl.pilot_owner_name = _text("pilot_owner_name")
603 decl.pilot_owner_licence_number = _text("pilot_owner_licence_number")
604 decl.declaration_type = declaration_type
605 decl.camo_cao_approval_reference = _text("camo_cao_approval_reference")
606 decl.certifying_party_kind = certifying_party_kind
607 decl.certifying_party_name = _text("certifying_party_name")
608 decl.certifying_party_address = _text("certifying_party_address")
609 decl.certifying_party_phone = _text("certifying_party_phone")
610 decl.certifying_party_email = _text("certifying_party_email")
611 decl.appendix_d_notes = _text("appendix_d_notes")
613 db.session.commit()
614 flash(_("AMP declaration saved."), "success")
615 return redirect(url_for("maintenance.list_triggers", aircraft_id=ac.id))
618# ── AMP revision history (block 10) ─────────────────────────────────────────
621@maintenance_bp.route(
622 "/aircraft/<aircraft_ref:aircraft_id>/amp/revisions/add", methods=["POST"]
623)
624@login_required
625@require_role(*_MAINT_ROLES)
626def add_amp_revision(aircraft_id: int) -> ResponseReturnValue:
627 ac = _get_aircraft_or_404(aircraft_id)
629 revision_number = (request.form.get("revision_number") or "").strip()
630 revision_content = (request.form.get("revision_content") or "").strip() or None
631 revision_date_raw = (request.form.get("revision_date") or "").strip()
633 errors = []
634 if not revision_number:
635 errors.append(_("Revision number is required."))
636 revision_date = None
637 if revision_date_raw:
638 try:
639 revision_date = _date.fromisoformat(revision_date_raw)
640 except ValueError:
641 errors.append(_("Revision date must be a valid date (YYYY-MM-DD)."))
643 if errors:
644 for msg in errors:
645 flash(msg, "danger")
646 return redirect(url_for("maintenance.edit_amp_declaration", aircraft_id=ac.id))
648 rev = AmpRevision(
649 aircraft_id=ac.id,
650 revision_number=revision_number,
651 revision_content=revision_content,
652 revision_date=revision_date,
653 )
654 db.session.add(rev)
655 db.session.flush()
657 # Snapshot what the AMP looks like right now as this revision's
658 # fingerprint — a later export whose data no longer matches this is a
659 # draft, not this revision. Only possible once a declaration profile
660 # exists; a revision added before that has no exportable content yet
661 # to fingerprint (content_hash stays unset, matching "nothing to
662 # compare against" rather than a false non-match).
663 context = _amp_export_context(ac)
664 if context is not None:
665 rev.content_hash = _amp_content_signature(context)
666 # Generate the canonical PDF now rather than waiting for a first
667 # download — closes the gap where a revision superseded before
668 # anyone ever downloaded it has no saved file, and makes the
669 # eventual download instant instead of rendering on demand.
670 # Best-effort: a render failure shouldn't block recording the
671 # revision itself — export_amp_pdf's own cache-fill fallback
672 # covers it if this doesn't happen for whatever reason.
673 try:
674 _render_and_cache_amp_pdf(context, rev)
675 except Exception:
676 current_app.logger.exception(
677 "Failed to pre-generate PDF for AMP revision %s", rev.id
678 )
680 db.session.commit()
681 flash(_("Revision added."), "success")
682 return redirect(url_for("maintenance.edit_amp_declaration", aircraft_id=ac.id))
685@maintenance_bp.route(
686 "/aircraft/<aircraft_ref:aircraft_id>/amp/revisions/<int:revision_id>/delete",
687 methods=["POST"],
688)
689@login_required
690@require_role(*_MAINT_ROLES)
691def delete_amp_revision(aircraft_id: int, revision_id: int) -> ResponseReturnValue:
692 ac = _get_aircraft_or_404(aircraft_id)
693 rev = db.session.get(AmpRevision, revision_id)
694 if not rev or rev.aircraft_id != ac.id:
695 abort(404)
696 number = rev.revision_number
697 if rev.pdf_path:
698 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
699 try:
700 os.remove(os.path.join(folder, rev.pdf_path))
701 except OSError:
702 current_app.logger.debug(
703 "AMP revision PDF already absent, skipping: %s", rev.pdf_path
704 )
705 db.session.delete(rev)
706 db.session.commit()
707 flash(_("Revision '%(number)s' deleted.", number=number), "success")
708 return redirect(url_for("maintenance.edit_amp_declaration", aircraft_id=ac.id))
711# ── AMP spreadsheet import ──────────────────────────────────────────────────
713_AMP_IMPORT_SESSION_KEY = "amp_import"
714_ALLOWED_AMP_IMPORT_EXTS = {".xlsx"}
715_MAX_AMP_IMPORT_BYTES = 10 * 1024 * 1024 # 10 MB
718def _amp_import_tmp_dir() -> str:
719 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
720 d = os.path.join(folder, "amp_import_tmp")
721 os.makedirs(d, exist_ok=True)
722 return d
725def _cleanup_amp_import_tmp(uid: int) -> None:
726 meta = session.get(_AMP_IMPORT_SESSION_KEY)
727 if meta and meta.get("uid") == uid:
728 tmp = meta.get("tmp_path")
729 if tmp and os.path.isfile(tmp):
730 try:
731 os.remove(tmp)
732 except OSError as exc:
733 current_app.logger.debug("cleanup tmp AMP import file: %s", exc)
734 session.pop(_AMP_IMPORT_SESSION_KEY, None)
737def _amp_preview_rows(
738 ac: Aircraft, rows: list[Any], components: list[Component]
739) -> list[dict[str, Any]]:
740 current_hobbs = ac.total_engine_hours
741 current_flight_hours = ac.total_flight_hours
742 components_by_id = {c.id: c for c in components}
744 preview = []
745 for r in rows:
746 suggested = (
747 components_by_id.get(r.suggested_component_id)
748 if (r.suggested_component_id is not None)
749 else None
750 )
751 basis = hours_basis_for_component(suggested)
752 due_h, due_d = compute_due_fields(
753 r.interval_hours,
754 r.interval_days,
755 basis,
756 current_hobbs,
757 current_flight_hours,
758 )
759 preview.append(
760 {
761 "parsed": r,
762 "suggested_component": suggested,
763 "due_engine_hours": due_h,
764 "due_date": due_d,
765 }
766 )
767 return preview
770@maintenance_bp.route(
771 "/aircraft/<aircraft_ref:aircraft_id>/maintenance/import", methods=["GET", "POST"]
772)
773@login_required
774@require_role(*_MAINT_ROLES)
775def import_amp_upload(aircraft_id: int) -> ResponseReturnValue:
776 ac = _get_aircraft_or_404(aircraft_id)
777 uid = session["user_id"]
779 if request.method == "GET":
780 return render_template("maintenance/amp_import_upload.html", aircraft=ac)
782 uploaded = request.files.get("amp_file")
783 if not uploaded or not uploaded.filename:
784 flash(_("Please select a file to upload."), "danger")
785 return render_template("maintenance/amp_import_upload.html", aircraft=ac), 422
787 ext = os.path.splitext(uploaded.filename)[1].lower()
788 if ext not in _ALLOWED_AMP_IMPORT_EXTS:
789 flash(_("Unsupported format. Please upload a .xlsx file."), "danger")
790 return render_template("maintenance/amp_import_upload.html", aircraft=ac), 422
792 data = uploaded.read()
793 if len(data) > _MAX_AMP_IMPORT_BYTES:
794 flash(_("File too large (maximum 10 MB)."), "danger")
795 return render_template("maintenance/amp_import_upload.html", aircraft=ac), 422
797 try:
798 wb = openpyxl.load_workbook(io.BytesIO(data), data_only=True)
799 components = list(ac.components)
800 rows = parse_amp_rows(wb, components)
801 except ValueError as exc:
802 flash(str(exc), "danger")
803 return render_template("maintenance/amp_import_upload.html", aircraft=ac), 422
805 if not rows:
806 flash(_("No task rows found in the uploaded file."), "danger")
807 return render_template("maintenance/amp_import_upload.html", aircraft=ac), 422
809 _cleanup_amp_import_tmp(uid)
810 safe_base = secure_filename(uploaded.filename) or "upload"
811 tmp_name = f"amp_import_{uid}_{uuid.uuid4().hex}_{safe_base}"
812 tmp_path = os.path.join(_amp_import_tmp_dir(), tmp_name)
813 with open(tmp_path, "wb") as fh:
814 fh.write(data)
816 session[_AMP_IMPORT_SESSION_KEY] = {
817 "uid": uid,
818 "aircraft_id": ac.id,
819 "tmp_path": tmp_path,
820 "original_filename": uploaded.filename,
821 }
823 preview_rows = _amp_preview_rows(ac, rows, components)
824 needs_review_count = sum(1 for r in rows if r.needs_review)
826 return render_template(
827 "maintenance/amp_import_review.html",
828 aircraft=ac,
829 preview_rows=preview_rows,
830 components=components,
831 row_count=len(rows),
832 needs_review_count=needs_review_count,
833 filename=uploaded.filename,
834 )
837@maintenance_bp.route(
838 "/aircraft/<aircraft_ref:aircraft_id>/maintenance/import/commit", methods=["POST"]
839)
840@login_required
841@require_role(*_MAINT_ROLES)
842def import_amp_commit(aircraft_id: int) -> ResponseReturnValue:
843 ac = _get_aircraft_or_404(aircraft_id)
844 uid = session["user_id"]
845 meta = session.get(_AMP_IMPORT_SESSION_KEY)
847 if not meta or meta.get("uid") != uid or meta.get("aircraft_id") != ac.id:
848 flash(_("Import session expired. Please upload the file again."), "warning")
849 return redirect(url_for("maintenance.import_amp_upload", aircraft_id=ac.id))
851 tmp_path: str = meta["tmp_path"]
852 original_filename: str = meta["original_filename"]
853 if not os.path.isfile(tmp_path):
854 flash(_("Temporary file not found. Please upload the file again."), "warning")
855 session.pop(_AMP_IMPORT_SESSION_KEY, None)
856 return redirect(url_for("maintenance.import_amp_upload", aircraft_id=ac.id))
858 with open(tmp_path, "rb") as fh:
859 data = fh.read()
860 wb = openpyxl.load_workbook(io.BytesIO(data), data_only=True)
861 components = list(ac.components)
862 components_by_id = {c.id: c for c in components}
863 rows = parse_amp_rows(wb, components)
865 current_hobbs = ac.total_engine_hours
866 current_flight_hours = ac.total_flight_hours
868 batch = MaintenanceImportBatch(
869 aircraft_id=ac.id,
870 source_filename=original_filename,
871 row_count=len(rows),
872 needs_review_count=sum(1 for r in rows if r.needs_review),
873 )
874 db.session.add(batch)
875 db.session.flush()
877 for i, r in enumerate(rows):
878 override_raw = (request.form.get(f"component_id_{i}") or "").strip()
879 component = None
880 if override_raw:
881 try:
882 component = components_by_id.get(int(override_raw))
883 except ValueError:
884 component = None
885 elif r.suggested_component_id is not None:
886 component = components_by_id.get(r.suggested_component_id)
888 basis = hours_basis_for_component(component)
889 due_h, due_d = compute_due_fields(
890 r.interval_hours,
891 r.interval_days,
892 basis,
893 current_hobbs,
894 current_flight_hours,
895 )
897 db.session.add(
898 MaintenanceTrigger(
899 aircraft_id=ac.id,
900 component_id=component.id if component else None,
901 name=r.name,
902 trigger_type=(
903 TriggerType.CALENDAR if due_d is not None else TriggerType.HOURS
904 ),
905 due_date=due_d,
906 interval_days=r.interval_days,
907 due_engine_hours=due_h,
908 interval_hours=r.interval_hours,
909 hours_basis=basis,
910 category=r.category,
911 reference=r.reference,
912 action=r.action,
913 part_number=r.part_number,
914 serial_number=r.serial_number,
915 notes=r.notes,
916 needs_review=r.needs_review,
917 import_batch_id=batch.id,
918 )
919 )
921 db.session.commit()
922 _cleanup_amp_import_tmp(uid)
923 activity(
924 "maintenance.amp_import",
925 aircraft_id=ac.id,
926 batch_id=batch.id,
927 row_count=batch.row_count,
928 )
930 review_phrase = ngettext(
931 "one flagged for review",
932 "%(n)s flagged for review",
933 batch.needs_review_count,
934 n=batch.needs_review_count,
935 )
936 flash(
937 ngettext(
938 "Imported %(count)d maintenance item — %(review_phrase)s.",
939 "Imported %(count)d maintenance items — %(review_phrase)s.",
940 batch.row_count,
941 count=batch.row_count,
942 review_phrase=review_phrase,
943 ),
944 "success",
945 )
946 return redirect(url_for("maintenance.list_triggers", aircraft_id=ac.id))
949@maintenance_bp.route("/aircraft/<aircraft_ref:aircraft_id>/maintenance/import/history")
950@login_required
951@require_role(*_MAINT_ROLES)
952def import_amp_history(aircraft_id: int) -> ResponseReturnValue:
953 ac = _get_aircraft_or_404(aircraft_id)
954 batches = (
955 MaintenanceImportBatch.query.filter_by(aircraft_id=ac.id)
956 .order_by(MaintenanceImportBatch.imported_at.desc())
957 .all()
958 )
959 return render_template(
960 "maintenance/amp_import_history.html", aircraft=ac, batches=batches
961 )
964@maintenance_bp.route(
965 "/aircraft/<aircraft_ref:aircraft_id>/maintenance/import/<int:batch_id>/rollback",
966 methods=["POST"],
967)
968@login_required
969@require_role(*_MAINT_ROLES)
970def import_amp_rollback(aircraft_id: int, batch_id: int) -> ResponseReturnValue:
971 ac = _get_aircraft_or_404(aircraft_id)
972 batch = db.session.get(MaintenanceImportBatch, batch_id)
973 if not batch or batch.aircraft_id != ac.id:
974 abort(404)
976 row_count = batch.row_count
977 MaintenanceTrigger.query.filter_by(import_batch_id=batch.id).delete()
978 db.session.delete(batch)
979 db.session.commit()
981 flash(
982 ngettext(
983 "Import rolled back: %(count)d item removed.",
984 "Import rolled back: %(count)d items removed.",
985 row_count,
986 count=row_count,
987 ),
988 "success",
989 )
990 return redirect(url_for("maintenance.import_amp_history", aircraft_id=ac.id))
993# ── AMP document export ─────────────────────────────────────────────────────
996def _amp_export_context(ac: Aircraft) -> dict[str, Any] | None:
997 """Build the template context for the AMP export (HTML preview and PDF
998 download render the same template from the same data — see
999 docs/maintenance_import.md's round-trip design note). Returns None if
1000 the aircraft has no AMP declaration profile yet."""
1001 decl: AmpDeclaration | None = ac.amp_declaration # type: ignore[assignment]
1002 if decl is None:
1003 return None
1005 components = list(ac.components)
1007 def _installed(comp_type: str) -> Component | None:
1008 return next(
1009 (c for c in components if c.type == comp_type and c.removed_at is None),
1010 None,
1011 )
1013 triggers = (
1014 MaintenanceTrigger.query.filter_by(aircraft_id=ac.id)
1015 .order_by(MaintenanceTrigger.name)
1016 .all()
1017 )
1019 category_has_items = {
1020 cat: any(t.category == cat for t in triggers) for cat in AmpCategory.ALL
1021 }
1022 has_alternative_tasks = any(t.is_alternative_to_ica for t in triggers)
1023 appendix_b_groups = [
1024 (cat, [t for t in triggers if t.category == cat])
1025 for cat in AmpCategory.ALL
1026 if category_has_items[cat]
1027 ]
1028 appendix_c_rows = [t for t in triggers if t.is_alternative_to_ica]
1029 interval_text = {
1030 t.id: format_interval(t.interval_hours, t.interval_days) for t in triggers
1031 }
1032 # A needs_review trigger has no interval yet and may not have a category
1033 # either — appendix_b_groups above would silently drop it if uncategorised.
1034 # Listed here regardless of category so nothing pending goes unmentioned in
1035 # the exported document, with its notes (the actual open question, if the
1036 # import captured one) alongside it.
1037 pending_review_rows = [t for t in triggers if t.needs_review]
1039 return {
1040 "aircraft": ac,
1041 "decl": decl,
1042 "triggers": triggers,
1043 "airframe_component": _installed(ComponentType.AIRFRAME),
1044 "engine_component": _installed(ComponentType.ENGINE),
1045 "propeller_component": _installed(ComponentType.PROPELLER),
1046 "category_has_items": category_has_items,
1047 "has_alternative_tasks": has_alternative_tasks,
1048 "appendix_b_groups": appendix_b_groups,
1049 "appendix_c_rows": appendix_c_rows,
1050 "interval_text": interval_text,
1051 "pending_review_rows": pending_review_rows,
1052 "revisions": ac.amp_revisions,
1053 "amp_basis": AmpBasis,
1054 "declaration_types": AmpDeclarationType,
1055 "certifying_party_kinds": AmpCertifyingPartyKind,
1056 }
1059def _amp_content_signature(context: dict[str, Any]) -> str:
1060 """SHA-256 fingerprint of the exportable AMP content — used to detect
1061 whether anything has changed since the last declared AmpRevision.
1063 Deliberately a hash of the underlying *data*, not of the rendered
1064 HTML/PDF: hashing the render would make every export "drift" after any
1065 future template/CSS tweak or a locale switch, even with byte-identical
1066 underlying data — a false positive on every single aircraft at once.
1067 """
1068 ac = context["aircraft"]
1069 decl = context["decl"]
1071 def _component(c: Component | None) -> dict[str, Any] | None:
1072 return (
1073 None
1074 if c is None
1075 else {"make": c.make, "model": c.model, "serial_number": c.serial_number}
1076 )
1078 payload = {
1079 "aircraft": {
1080 "registration": ac.registration,
1081 "make": ac.make,
1082 "model": ac.model,
1083 },
1084 "airframe": _component(context["airframe_component"]),
1085 "engine": _component(context["engine_component"]),
1086 "propeller": _component(context["propeller_component"]),
1087 "declaration": {
1088 col.name: str(getattr(decl, col.name))
1089 for col in AmpDeclaration.__table__.columns
1090 if col.name not in ("aircraft_id", "updated_at")
1091 },
1092 "triggers": [
1093 {
1094 "name": t.name,
1095 "category": t.category,
1096 "is_alternative_to_ica": t.is_alternative_to_ica,
1097 "alternative_task_notes": t.alternative_task_notes,
1098 "reference": t.reference,
1099 "interval_hours": str(t.interval_hours),
1100 "interval_days": t.interval_days,
1101 "needs_review": t.needs_review,
1102 "notes": t.notes,
1103 }
1104 for t in sorted(context["triggers"], key=lambda t: t.id)
1105 ],
1106 "revisions": [
1107 {
1108 "revision_number": r.revision_number,
1109 "revision_content": r.revision_content,
1110 "revision_date": str(r.revision_date),
1111 }
1112 for r in context["revisions"]
1113 ],
1114 }
1115 return hashlib.sha256(
1116 json.dumps(payload, sort_keys=True, default=str).encode("utf-8")
1117 ).hexdigest()
1120def _amp_draft_status(
1121 context: dict[str, Any],
1122) -> tuple[bool, AmpRevision | None]:
1123 """Return (is_draft, latest_revision). is_draft is True when there's no
1124 declared revision yet, or the AMP's current data no longer matches the
1125 fingerprint recorded when the latest one was declared."""
1126 revisions = context["revisions"]
1127 latest_revision = revisions[-1] if revisions else None
1128 if latest_revision is None:
1129 return True, None
1130 current_hash = _amp_content_signature(context)
1131 return current_hash != latest_revision.content_hash, latest_revision
1134def _amp_pdf_filename(
1135 ac: Aircraft, is_draft: bool, latest_revision: AmpRevision | None
1136) -> str:
1137 date_str = (
1138 latest_revision.revision_date.isoformat()
1139 if not is_draft and latest_revision and latest_revision.revision_date
1140 else _date.today().isoformat()
1141 )
1142 revision_str = (
1143 "draft"
1144 if is_draft or latest_revision is None
1145 else latest_revision.revision_number
1146 )
1147 return secure_filename(f"{date_str}-AMP-{ac.registration}-{revision_str}.pdf")
1150def _amp_pdf_storage_path(revision_id: int) -> tuple[str, str]:
1151 """Return (folder, relative_filename) for a revision's cached canonical PDF."""
1152 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
1153 return folder, f"amp_revision_{revision_id}.pdf"
1156def _render_and_cache_amp_pdf(context: dict[str, Any], revision: AmpRevision) -> bytes:
1157 """Render the (non-draft) PDF for *revision* and save it as its
1158 canonical file, setting pdf_path. Caller commits. Does not check
1159 is_draft — only call this when the caller has already established the
1160 current data matches this revision."""
1161 html = render_template("maintenance/amp_export_pdf.html", is_draft=False, **context)
1162 pdf_bytes: bytes = HTML(string=html, base_url=request.url_root).write_pdf()
1163 folder, stored_name = _amp_pdf_storage_path(revision.id)
1164 os.makedirs(folder, exist_ok=True)
1165 with open(os.path.join(folder, stored_name), "wb") as f:
1166 f.write(pdf_bytes)
1167 revision.pdf_path = stored_name
1168 return pdf_bytes
1171@maintenance_bp.route("/aircraft/<aircraft_ref:aircraft_id>/maintenance/amp/export")
1172@login_required
1173@require_role(*_MAINT_ROLES)
1174def export_amp(aircraft_id: int) -> ResponseReturnValue:
1175 ac = _get_aircraft_or_404(aircraft_id)
1176 context = _amp_export_context(ac)
1177 if context is None:
1178 flash(
1179 _("Fill in the AMP declaration profile before exporting the AMP document."),
1180 "warning",
1181 )
1182 return redirect(url_for("maintenance.edit_amp_declaration", aircraft_id=ac.id))
1184 is_draft, _latest_revision = _amp_draft_status(context)
1185 return render_template("maintenance/amp_export.html", is_draft=is_draft, **context)
1188@maintenance_bp.route("/aircraft/<aircraft_ref:aircraft_id>/maintenance/amp/export/pdf")
1189@login_required
1190@require_role(*_MAINT_ROLES)
1191def export_amp_pdf(aircraft_id: int) -> ResponseReturnValue:
1192 ac = _get_aircraft_or_404(aircraft_id)
1193 context = _amp_export_context(ac)
1194 if context is None:
1195 flash(
1196 _("Fill in the AMP declaration profile before exporting the AMP document."),
1197 "warning",
1198 )
1199 return redirect(url_for("maintenance.edit_amp_declaration", aircraft_id=ac.id))
1201 is_draft, latest_revision = _amp_draft_status(context)
1202 filename = _amp_pdf_filename(ac, is_draft, latest_revision)
1204 # Not a draft and already cached: serve the exact bytes generated the
1205 # first time this revision was downloaded, rather than a fresh render.
1206 if not is_draft and latest_revision and latest_revision.pdf_path:
1207 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
1208 cached_path = os.path.join(folder, latest_revision.pdf_path)
1209 if os.path.exists(cached_path):
1210 with open(cached_path, "rb") as f:
1211 pdf_bytes = f.read()
1212 return Response(
1213 pdf_bytes,
1214 mimetype="application/pdf",
1215 headers={"Content-Disposition": f'attachment; filename="{filename}"'},
1216 )
1218 if not is_draft and latest_revision:
1219 # Not a draft and not cached yet — normally already generated when
1220 # the revision was added; this is the fallback for a revision
1221 # created outside that flow (a script, a failed eager render).
1222 pdf_bytes = _render_and_cache_amp_pdf(context, latest_revision)
1223 db.session.commit()
1224 else:
1225 html = render_template(
1226 "maintenance/amp_export_pdf.html", is_draft=is_draft, **context
1227 )
1228 pdf_bytes = HTML(string=html, base_url=request.url_root).write_pdf()
1230 return Response(
1231 pdf_bytes,
1232 mimetype="application/pdf",
1233 headers={"Content-Disposition": f'attachment; filename="{filename}"'},
1234 )
1237@maintenance_bp.route(
1238 "/aircraft/<aircraft_ref:aircraft_id>/amp/revisions/<int:revision_id>/pdf"
1239)
1240@login_required
1241@require_role(*_MAINT_ROLES)
1242def download_amp_revision_pdf(
1243 aircraft_id: int, revision_id: int
1244) -> ResponseReturnValue:
1245 """Re-download a specific past revision's canonical PDF, byte-for-byte
1246 as first generated — even after later edits have moved the live AMP
1247 data on. Only available for a revision that was actually downloaded at
1248 least once while it was still the current, undrafted state; there is
1249 no field-level history to reconstruct one that wasn't."""
1250 ac = _get_aircraft_or_404(aircraft_id)
1251 rev = db.session.get(AmpRevision, revision_id)
1252 if not rev or rev.aircraft_id != ac.id:
1253 abort(404)
1254 if not rev.pdf_path:
1255 flash(
1256 _(
1257 "No saved PDF for revision '%(number)s' — it was never "
1258 "downloaded while current.",
1259 number=rev.revision_number,
1260 ),
1261 "warning",
1262 )
1263 return redirect(url_for("maintenance.edit_amp_declaration", aircraft_id=ac.id))
1265 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
1266 path = os.path.join(folder, rev.pdf_path)
1267 if not os.path.exists(path):
1268 flash(
1269 _(
1270 "The saved PDF for revision '%(number)s' is missing on disk.",
1271 number=rev.revision_number,
1272 ),
1273 "danger",
1274 )
1275 return redirect(url_for("maintenance.edit_amp_declaration", aircraft_id=ac.id))
1277 with open(path, "rb") as f:
1278 pdf_bytes = f.read()
1279 date_str = rev.revision_date.isoformat() if rev.revision_date else "undated"
1280 filename = secure_filename(
1281 f"{date_str}-AMP-{ac.registration}-{rev.revision_number}.pdf"
1282 )
1283 return Response(
1284 pdf_bytes,
1285 mimetype="application/pdf",
1286 headers={"Content-Disposition": f'attachment; filename="{filename}"'},
1287 )