Coverage for app/maintenance/routes.py: 100%
578 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 15:29 +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 # Projected due date for hours-based triggers (see
130 # maintenance/due_date_projection.py) — computed per aircraft, since the
131 # utilization rate is aircraft-specific.
132 projected_by_trigger_id: dict[int, Any] = {}
133 triggers_by_aircraft: dict[int, list[MaintenanceTrigger]] = {}
134 for t in triggers:
135 triggers_by_aircraft.setdefault(t.aircraft_id, []).append(t)
136 for ac_id, ac_triggers in triggers_by_aircraft.items():
137 projected_by_trigger_id.update(
138 _projected_due_dates(
139 ac_id,
140 ac_triggers,
141 hobbs_by_id.get(ac_id),
142 flight_hours_by_id.get(ac_id),
143 )
144 )
146 # Annotate each trigger with its status and (for hours-based triggers)
147 # a projected due date, always an estimate — never a substitute for the
148 # real hours-based due figure.
149 trigger_rows = [
150 (
151 t,
152 t.status(
153 current_engine_hours=hobbs_by_id.get(t.aircraft_id),
154 current_landings=landings_by_id.get(t.aircraft_id),
155 current_flight_hours=flight_hours_by_id.get(t.aircraft_id),
156 ),
157 ac_by_id[t.aircraft_id],
158 projected_by_trigger_id.get(t.id),
159 )
160 for t in triggers
161 ]
163 # Sort: overdue → due_soon → ok; within status: calendar triggers by due_date
164 # asc, then hours-based triggers with a projected date (also asc), then
165 # any remaining undated items last.
166 _status_order = {"overdue": 0, "due_soon": 1, "ok": 2}
167 _far_future = _date_cls(9999, 12, 31)
169 def _trigger_sort_key(row: Any) -> Any:
170 t, status, _ac, projected_date = row
171 if t.trigger_type == TriggerType.CALENDAR and t.due_date:
172 due = t.due_date
173 elif projected_date:
174 due = projected_date
175 else:
176 due = _far_future
177 return (_status_order[status], due)
179 trigger_rows.sort(key=_trigger_sort_key)
181 # Open grounding snags — oldest reported first (most overdue on top)
182 grounding_snags = (
183 (
184 Snag.query.filter(
185 Snag.aircraft_id.in_(aircraft_ids),
186 Snag.is_grounding.is_(True),
187 Snag.resolved_at.is_(None),
188 )
189 .order_by(Snag.reported_at.asc())
190 .all()
191 )
192 if aircraft_ids
193 else []
194 )
195 grounding_snag_rows = [(s, ac_by_id[s.aircraft_id]) for s in grounding_snags]
197 # Open non-grounding snags — oldest reported first
198 open_snags = (
199 (
200 Snag.query.filter(
201 Snag.aircraft_id.in_(aircraft_ids),
202 Snag.is_grounding.is_(False),
203 Snag.resolved_at.is_(None),
204 )
205 .order_by(Snag.reported_at.asc())
206 .all()
207 )
208 if aircraft_ids
209 else []
210 )
211 open_snag_rows = [(s, ac_by_id[s.aircraft_id]) for s in open_snags]
213 aircraft_status = compute_aircraft_statuses(
214 aircraft, triggers, hobbs_by_id, landings_by_id, flight_hours_by_id
215 )
217 # Chronological view: single list sorted by due/reported date asc.
218 # Hours-based triggers have no reliable date → sorted after all dated items.
219 # Tuple structure: (kind, sort_dt, obj, ac, status, projected_date)
220 # kind_order: grounding=0, snag=1, maintenance=2 (tiebreak within same date)
221 _far_dt = _datetime(_far_future.year, _far_future.month, _far_future.day)
222 chron_items = []
223 for s, ac in grounding_snag_rows:
224 dt = _datetime.combine(
225 s.reported_at.date() if hasattr(s.reported_at, "date") else s.reported_at,
226 _datetime.min.time(),
227 )
228 chron_items.append(("grounding", dt, s, ac, None, None))
229 for s, ac in open_snag_rows:
230 dt = _datetime.combine(
231 s.reported_at.date() if hasattr(s.reported_at, "date") else s.reported_at,
232 _datetime.min.time(),
233 )
234 chron_items.append(("snag", dt, s, ac, None, None))
235 for t, status, ac, projected_date in trigger_rows:
236 if status in ("overdue", "due_soon") or t.needs_review:
237 if t.due_date:
238 dt = _datetime(t.due_date.year, t.due_date.month, t.due_date.day)
239 elif projected_date:
240 dt = _datetime(
241 projected_date.year, projected_date.month, projected_date.day
242 )
243 else:
244 dt = _far_dt # no calendar due date, no projection: push to end
245 chron_items.append(("maintenance", dt, t, ac, status, projected_date))
247 _kind_order = {"grounding": 0, "snag": 1, "maintenance": 2}
248 chron_items.sort(key=lambda x: (x[1], _kind_order[x[0]]))
250 view = request.args.get("view", "by-type")
252 # Component TBO / calendar life limits that need attention
253 from services.component_limits import (
254 aircraft_limit_infos, # pyright: ignore[reportMissingImports]
255 )
257 component_limit_rows = []
258 for ac in aircraft:
259 for info in aircraft_limit_infos(ac):
260 if info["status"] in ("overdue", "due_soon"):
261 component_limit_rows.append((info, ac))
262 component_limit_rows.sort(key=lambda row: 0 if row[0]["status"] == "overdue" else 1)
264 any_needs_review = any(
265 t.needs_review for t, _status, _ac, _projected_date in trigger_rows
266 )
268 return render_template(
269 "maintenance/fleet.html",
270 aircraft=aircraft,
271 aircraft_status=aircraft_status,
272 component_limit_rows=component_limit_rows,
273 trigger_rows=trigger_rows,
274 grounding_snag_rows=grounding_snag_rows,
275 open_snag_rows=open_snag_rows,
276 chron_items=chron_items,
277 hobbs_by_id=hobbs_by_id,
278 landings_by_id=landings_by_id,
279 flight_hours_by_id=flight_hours_by_id,
280 any_needs_review=any_needs_review,
281 view=view,
282 )
285# ── Trigger list ──────────────────────────────────────────────────────────────
288def _projected_due_dates(
289 aircraft_id: int,
290 triggers: list[MaintenanceTrigger],
291 current_hobbs: "float | None",
292 current_flight_hours: "float | None",
293) -> dict[int, "Any | None"]:
294 """{trigger.id: projected calendar date} for every hours-based trigger,
295 or None where there's not enough recent utilization history to trust a
296 projection (see maintenance/due_date_projection.py). Computes the
297 utilization rate at most once per hours_basis actually in use, rather
298 than once per trigger."""
299 from maintenance.due_date_projection import ( # pyright: ignore[reportMissingImports]
300 project_due_date,
301 weekly_utilization_rate,
302 )
304 rate_by_basis: dict[str, float | None] = {}
305 result: dict[int, Any | None] = {}
306 for t in triggers:
307 if t.due_engine_hours is None:
308 continue
309 if t.hours_basis not in rate_by_basis:
310 rate_by_basis[t.hours_basis] = weekly_utilization_rate(
311 aircraft_id, t.hours_basis
312 )
313 current = (
314 current_flight_hours
315 if t.hours_basis == HoursBasis.FLIGHT
316 else current_hobbs
317 )
318 result[t.id] = project_due_date(
319 current, float(t.due_engine_hours), rate_by_basis[t.hours_basis]
320 )
321 return result
324def _group_trigger_rows_by_component(
325 trigger_rows: list[tuple[MaintenanceTrigger, str]],
326) -> list[tuple[Component | None, list[tuple[MaintenanceTrigger, str]]]]:
327 """Group (trigger, status) rows by trigger.component — unscoped
328 ("Airframe / general") rows first, then installed components in the
329 same (type, position) order used for the components list on
330 aircraft/detail.html. A component only gets a section if it actually
331 has at least one trigger."""
332 general: list[tuple[MaintenanceTrigger, str]] = []
333 by_component: dict[int, tuple[Component, list[tuple[MaintenanceTrigger, str]]]] = {}
334 for t, status in trigger_rows:
335 if t.component is None:
336 general.append((t, status))
337 else:
338 entry = by_component.setdefault(
339 t.component.id,
340 (t.component, []), # type: ignore[arg-type]
341 )
342 entry[1].append((t, status))
344 groups: list[tuple[Component | None, list[tuple[MaintenanceTrigger, str]]]] = []
345 if general:
346 groups.append((None, general))
347 for comp, rows in sorted(
348 by_component.values(), key=lambda cr: (cr[0].type, cr[0].position or "")
349 ):
350 groups.append((comp, rows))
351 return groups
354@maintenance_bp.route("/aircraft/<aircraft_ref:aircraft_id>/maintenance")
355@login_required
356def list_triggers(aircraft_id: int) -> ResponseReturnValue:
357 ac = _get_aircraft_or_404(aircraft_id)
358 current_hobbs = ac.total_engine_hours
359 current_landings = ac.total_landings
360 current_flight_hours = ac.total_flight_hours
361 all_triggers = (
362 MaintenanceTrigger.query.filter_by(aircraft_id=ac.id)
363 .order_by(MaintenanceTrigger.name)
364 .all()
365 )
366 tid = _tenant_id()
367 uid = session["user_id"]
368 maint_view = AuthorizationService.maintenance_view_level(uid, aircraft_id, tid)
370 def _status(t: MaintenanceTrigger) -> str:
371 return t.status(
372 current_engine_hours=current_hobbs,
373 current_landings=current_landings,
374 current_flight_hours=current_flight_hours,
375 )
377 # Limited view: show only overdue and due-soon items
378 if maint_view == "limited":
379 triggers = [t for t in all_triggers if _status(t) in ("overdue", "due_soon")]
380 else:
381 triggers = all_triggers
382 trigger_rows = [(t, _status(t)) for t in triggers]
383 component_groups = _group_trigger_rows_by_component(trigger_rows)
384 projected_dates = _projected_due_dates(
385 ac.id, triggers, current_hobbs, current_flight_hours
386 )
387 return render_template(
388 "maintenance/list.html",
389 aircraft=ac,
390 trigger_rows=trigger_rows,
391 component_groups=component_groups,
392 current_hobbs=current_hobbs,
393 current_landings=current_landings,
394 current_flight_hours=current_flight_hours,
395 projected_dates=projected_dates,
396 maint_view=maint_view,
397 )
400# ── Add trigger ───────────────────────────────────────────────────────────────
403@maintenance_bp.route(
404 "/aircraft/<aircraft_ref:aircraft_id>/maintenance/new", methods=["GET", "POST"]
405)
406@login_required
407@require_role(*_MAINT_ROLES)
408def new_trigger(aircraft_id: int) -> ResponseReturnValue:
409 ac = _get_aircraft_or_404(aircraft_id)
410 if request.method == "POST":
411 return _save_trigger(ac, None)
412 return render_template(
413 "maintenance/trigger_form.html",
414 aircraft=ac,
415 trigger=None,
416 trigger_types=TriggerType,
417 hours_basis=HoursBasis,
418 )
421# ── Edit trigger ──────────────────────────────────────────────────────────────
424@maintenance_bp.route(
425 "/aircraft/<aircraft_ref:aircraft_id>/maintenance/<int:trigger_id>/edit",
426 methods=["GET", "POST"],
427)
428@login_required
429@require_role(*_MAINT_ROLES)
430def edit_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 if request.method == "POST":
434 return _save_trigger(ac, t)
435 return render_template(
436 "maintenance/trigger_form.html",
437 aircraft=ac,
438 trigger=t,
439 trigger_types=TriggerType,
440 hours_basis=HoursBasis,
441 )
444def _save_trigger(ac: Aircraft, t: MaintenanceTrigger | None) -> ResponseReturnValue:
445 values, errors = parse_trigger_fields(request.form)
447 component_id = values.get("component_id")
448 if component_id is not None:
449 comp = db.session.get(Component, component_id)
450 if comp is None or comp.aircraft_id != ac.id:
451 errors.append(_("Component selection is invalid."))
452 component_id = None
454 if errors:
455 for msg in errors:
456 flash(msg, "danger")
457 return render_template(
458 "maintenance/trigger_form.html",
459 aircraft=ac,
460 trigger=t,
461 trigger_types=TriggerType,
462 hours_basis=HoursBasis,
463 )
465 if t is None:
466 t = MaintenanceTrigger(aircraft_id=ac.id)
467 db.session.add(t)
469 t.name = values["name"]
470 t.trigger_type = values["trigger_type"]
471 t.component_id = component_id
472 t.due_date = values["due_date"]
473 t.interval_days = values["interval_days"]
474 t.warn_days = values["warn_days"]
475 t.due_engine_hours = values["due_engine_hours"]
476 t.interval_hours = values["interval_hours"]
477 t.warn_hours = values["warn_hours"]
478 t.hours_basis = values["hours_basis"]
479 t.due_landings = values["due_landings"]
480 t.interval_landings = values["interval_landings"]
481 t.warn_landings = values["warn_landings"]
482 t.notes = values["notes"]
483 db.session.commit()
485 flash(_("Maintenance item '%(name)s' saved.", name=t.name), "success")
486 return redirect(url_for("maintenance.list_triggers", aircraft_id=ac.id))
489# ── Delete trigger ────────────────────────────────────────────────────────────
492@maintenance_bp.route(
493 "/aircraft/<aircraft_ref:aircraft_id>/maintenance/<int:trigger_id>/delete",
494 methods=["POST"],
495)
496@login_required
497@require_role(*_MAINT_ROLES)
498def delete_trigger(aircraft_id: int, trigger_id: int) -> ResponseReturnValue:
499 ac = _get_aircraft_or_404(aircraft_id)
500 t = _get_trigger_or_404(ac, trigger_id)
501 name = t.name
502 db.session.delete(t)
503 db.session.commit()
504 flash(_("'%(name)s' deleted.", name=name), "success")
505 return redirect(url_for("maintenance.list_triggers", aircraft_id=ac.id))
508# ── Mark as serviced ──────────────────────────────────────────────────────────
511@maintenance_bp.route(
512 "/aircraft/<aircraft_ref:aircraft_id>/maintenance/<int:trigger_id>/service",
513 methods=["GET", "POST"],
514)
515@login_required
516@require_role(*_MAINT_ROLES)
517def service_trigger(aircraft_id: int, trigger_id: int) -> ResponseReturnValue:
518 ac = _get_aircraft_or_404(aircraft_id)
519 t = _get_trigger_or_404(ac, trigger_id)
521 # Phase 40: which readings are required depends on which due-field
522 # groups are actually populated on this trigger, not on trigger_type — a
523 # combined-interval trigger can have more than one group populated.
524 requires_hobbs = t.due_engine_hours is not None
525 requires_landings = t.due_landings is not None
527 if request.method == "POST":
528 values, errors = parse_service_fields(
529 request.form, requires_hobbs, requires_landings
530 )
531 performed_at = values["performed_at"]
532 hobbs_at_service = values["hobbs_at_service"]
533 landings_at_service = values["landings_at_service"]
535 if errors:
536 for msg in errors:
537 flash(msg, "danger")
538 return render_template(
539 "maintenance/service_form.html",
540 aircraft=ac,
541 trigger=t,
542 current_hobbs=ac.total_engine_hours,
543 current_landings=ac.total_landings,
544 current_flight_hours=ac.total_flight_hours,
545 today=_date.today().isoformat(),
546 )
548 record = MaintenanceRecord(
549 trigger_id=t.id,
550 performed_at=performed_at,
551 hobbs_at_service=hobbs_at_service,
552 landings_at_service=landings_at_service,
553 notes=values["notes"],
554 )
555 db.session.add(record)
557 # Advance every populated due-field group independently — a
558 # combined-interval trigger (both calendar and hours set) advances
559 # both together from the same service record.
560 if t.due_date is not None and t.interval_days and performed_at:
561 t.due_date = performed_at + timedelta(days=t.interval_days)
562 if (
563 t.due_engine_hours is not None
564 and t.interval_hours
565 and hobbs_at_service is not None
566 ):
567 t.due_engine_hours = hobbs_at_service + float(t.interval_hours)
568 if (
569 t.due_landings is not None
570 and t.interval_landings
571 and landings_at_service is not None
572 ):
573 t.due_landings = landings_at_service + t.interval_landings
575 db.session.commit()
576 activity(
577 "maintenance.serviced",
578 trigger_id=t.id,
579 aircraft_id=aircraft_id,
580 trigger_name=t.name,
581 record_id=record.id,
582 )
583 flash(_("'%(name)s' marked as serviced.", name=t.name), "success")
584 return redirect(url_for("maintenance.list_triggers", aircraft_id=ac.id))
586 return render_template(
587 "maintenance/service_form.html",
588 aircraft=ac,
589 trigger=t,
590 current_hobbs=ac.total_engine_hours,
591 current_landings=ac.total_landings,
592 current_flight_hours=ac.total_flight_hours,
593 today=_date.today().isoformat(),
594 )
597# ── AMP declaration profile ─────────────────────────────────────────────────
600@maintenance_bp.route(
601 "/aircraft/<aircraft_ref:aircraft_id>/amp/edit", methods=["GET", "POST"]
602)
603@login_required
604@require_role(*_MAINT_ROLES)
605def edit_amp_declaration(aircraft_id: int) -> ResponseReturnValue:
606 ac = _get_aircraft_or_404(aircraft_id)
607 decl: AmpDeclaration | None = ac.amp_declaration # type: ignore[assignment]
608 if request.method == "POST":
609 return _save_amp_declaration(ac, decl)
610 return render_template(
611 "maintenance/amp_declaration_form.html",
612 aircraft=ac,
613 decl=decl,
614 revisions=ac.amp_revisions,
615 amp_basis=AmpBasis,
616 declaration_types=AmpDeclarationType,
617 certifying_party_kinds=AmpCertifyingPartyKind,
618 )
621def _save_amp_declaration(
622 ac: Aircraft, decl: AmpDeclaration | None
623) -> ResponseReturnValue:
624 def _text(key: str) -> str | None:
625 v = (request.form.get(key) or "").strip()
626 return v or None
628 basis = request.form.get("basis", "").strip() or AmpBasis.DAH_ICA
629 declaration_type = (
630 request.form.get("declaration_type", "").strip() or AmpDeclarationType.OWNER
631 )
632 certifying_party_kind = (
633 request.form.get("certifying_party_kind", "").strip()
634 or AmpCertifyingPartyKind.OWNER_LESSEE_OPERATOR
635 )
637 errors = []
638 if basis not in AmpBasis.ALL:
639 errors.append(_("Invalid programme basis selected."))
640 if declaration_type not in AmpDeclarationType.ALL:
641 errors.append(_("Invalid declaration type selected."))
642 if certifying_party_kind not in AmpCertifyingPartyKind.ALL:
643 errors.append(_("Invalid certifying party selected."))
645 if errors:
646 for msg in errors:
647 flash(msg, "danger")
648 return render_template(
649 "maintenance/amp_declaration_form.html",
650 aircraft=ac,
651 decl=decl,
652 revisions=ac.amp_revisions,
653 amp_basis=AmpBasis,
654 declaration_types=AmpDeclarationType,
655 certifying_party_kinds=AmpCertifyingPartyKind,
656 )
658 if decl is None:
659 decl = AmpDeclaration(aircraft_id=ac.id)
660 db.session.add(decl)
662 decl.owner_name = _text("owner_name")
663 decl.owner_address = _text("owner_address")
664 decl.basis = basis
665 decl.mip_details = _text("mip_details")
666 decl.dah_ica_airframe_ref = _text("dah_ica_airframe_ref")
667 decl.dah_ica_engine_ref = _text("dah_ica_engine_ref")
668 decl.dah_ica_propeller_ref = _text("dah_ica_propeller_ref")
669 decl.pilot_owner_maintenance = request.form.get("pilot_owner_maintenance") == "on"
670 decl.pilot_owner_name = _text("pilot_owner_name")
671 decl.pilot_owner_licence_number = _text("pilot_owner_licence_number")
672 decl.declaration_type = declaration_type
673 decl.camo_cao_approval_reference = _text("camo_cao_approval_reference")
674 decl.certifying_party_kind = certifying_party_kind
675 decl.certifying_party_name = _text("certifying_party_name")
676 decl.certifying_party_address = _text("certifying_party_address")
677 decl.certifying_party_phone = _text("certifying_party_phone")
678 decl.certifying_party_email = _text("certifying_party_email")
679 decl.appendix_d_notes = _text("appendix_d_notes")
681 db.session.commit()
682 flash(_("AMP declaration saved."), "success")
683 return redirect(url_for("maintenance.list_triggers", aircraft_id=ac.id))
686# ── AMP revision history (block 10) ─────────────────────────────────────────
689@maintenance_bp.route(
690 "/aircraft/<aircraft_ref:aircraft_id>/amp/revisions/add", methods=["POST"]
691)
692@login_required
693@require_role(*_MAINT_ROLES)
694def add_amp_revision(aircraft_id: int) -> ResponseReturnValue:
695 ac = _get_aircraft_or_404(aircraft_id)
697 revision_number = (request.form.get("revision_number") or "").strip()
698 revision_content = (request.form.get("revision_content") or "").strip() or None
699 revision_date_raw = (request.form.get("revision_date") or "").strip()
701 errors = []
702 if not revision_number:
703 errors.append(_("Revision number is required."))
704 revision_date = None
705 if revision_date_raw:
706 try:
707 revision_date = _date.fromisoformat(revision_date_raw)
708 except ValueError:
709 errors.append(_("Revision date must be a valid date (YYYY-MM-DD)."))
711 if errors:
712 for msg in errors:
713 flash(msg, "danger")
714 return redirect(url_for("maintenance.edit_amp_declaration", aircraft_id=ac.id))
716 rev = AmpRevision(
717 aircraft_id=ac.id,
718 revision_number=revision_number,
719 revision_content=revision_content,
720 revision_date=revision_date,
721 )
722 db.session.add(rev)
723 db.session.flush()
725 # Snapshot what the AMP looks like right now as this revision's
726 # fingerprint — a later export whose data no longer matches this is a
727 # draft, not this revision. Only possible once a declaration profile
728 # exists; a revision added before that has no exportable content yet
729 # to fingerprint (content_hash stays unset, matching "nothing to
730 # compare against" rather than a false non-match).
731 context = _amp_export_context(ac)
732 if context is not None:
733 rev.content_hash = _amp_content_signature(context)
734 # Generate the canonical PDF now rather than waiting for a first
735 # download — closes the gap where a revision superseded before
736 # anyone ever downloaded it has no saved file, and makes the
737 # eventual download instant instead of rendering on demand.
738 # Best-effort: a render failure shouldn't block recording the
739 # revision itself — export_amp_pdf's own cache-fill fallback
740 # covers it if this doesn't happen for whatever reason.
741 try:
742 _render_and_cache_amp_pdf(context, rev)
743 except Exception:
744 current_app.logger.exception(
745 "Failed to pre-generate PDF for AMP revision %s", rev.id
746 )
748 db.session.commit()
749 flash(_("Revision added."), "success")
750 return redirect(url_for("maintenance.edit_amp_declaration", aircraft_id=ac.id))
753@maintenance_bp.route(
754 "/aircraft/<aircraft_ref:aircraft_id>/amp/revisions/<int:revision_id>/delete",
755 methods=["POST"],
756)
757@login_required
758@require_role(*_MAINT_ROLES)
759def delete_amp_revision(aircraft_id: int, revision_id: int) -> ResponseReturnValue:
760 ac = _get_aircraft_or_404(aircraft_id)
761 rev = db.session.get(AmpRevision, revision_id)
762 if not rev or rev.aircraft_id != ac.id:
763 abort(404)
764 number = rev.revision_number
765 if rev.pdf_path:
766 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
767 try:
768 os.remove(os.path.join(folder, rev.pdf_path))
769 except OSError:
770 current_app.logger.debug(
771 "AMP revision PDF already absent, skipping: %s", rev.pdf_path
772 )
773 db.session.delete(rev)
774 db.session.commit()
775 flash(_("Revision '%(number)s' deleted.", number=number), "success")
776 return redirect(url_for("maintenance.edit_amp_declaration", aircraft_id=ac.id))
779# ── AMP spreadsheet import ──────────────────────────────────────────────────
781_AMP_IMPORT_SESSION_KEY = "amp_import"
782_ALLOWED_AMP_IMPORT_EXTS = {".xlsx"}
783_MAX_AMP_IMPORT_BYTES = 10 * 1024 * 1024 # 10 MB
786def _amp_import_tmp_dir() -> str:
787 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
788 d = os.path.join(folder, "amp_import_tmp")
789 os.makedirs(d, exist_ok=True)
790 return d
793def _cleanup_amp_import_tmp(uid: int) -> None:
794 meta = session.get(_AMP_IMPORT_SESSION_KEY)
795 if meta and meta.get("uid") == uid:
796 tmp = meta.get("tmp_path")
797 if tmp and os.path.isfile(tmp):
798 try:
799 os.remove(tmp)
800 except OSError as exc:
801 current_app.logger.debug("cleanup tmp AMP import file: %s", exc)
802 session.pop(_AMP_IMPORT_SESSION_KEY, None)
805def _amp_preview_rows(
806 ac: Aircraft, rows: list[Any], components: list[Component]
807) -> list[dict[str, Any]]:
808 current_hobbs = ac.total_engine_hours
809 current_flight_hours = ac.total_flight_hours
810 components_by_id = {c.id: c for c in components}
812 preview = []
813 for r in rows:
814 suggested = (
815 components_by_id.get(r.suggested_component_id)
816 if (r.suggested_component_id is not None)
817 else None
818 )
819 basis = hours_basis_for_component(suggested)
820 due_h, due_d = compute_due_fields(
821 r.interval_hours,
822 r.interval_days,
823 basis,
824 current_hobbs,
825 current_flight_hours,
826 )
827 preview.append(
828 {
829 "parsed": r,
830 "suggested_component": suggested,
831 "due_engine_hours": due_h,
832 "due_date": due_d,
833 }
834 )
835 return preview
838@maintenance_bp.route(
839 "/aircraft/<aircraft_ref:aircraft_id>/maintenance/import", methods=["GET", "POST"]
840)
841@login_required
842@require_role(*_MAINT_ROLES)
843def import_amp_upload(aircraft_id: int) -> ResponseReturnValue:
844 ac = _get_aircraft_or_404(aircraft_id)
845 uid = session["user_id"]
847 if request.method == "GET":
848 return render_template("maintenance/amp_import_upload.html", aircraft=ac)
850 uploaded = request.files.get("amp_file")
851 if not uploaded or not uploaded.filename:
852 flash(_("Please select a file to upload."), "danger")
853 return render_template("maintenance/amp_import_upload.html", aircraft=ac), 422
855 ext = os.path.splitext(uploaded.filename)[1].lower()
856 if ext not in _ALLOWED_AMP_IMPORT_EXTS:
857 flash(_("Unsupported format. Please upload a .xlsx file."), "danger")
858 return render_template("maintenance/amp_import_upload.html", aircraft=ac), 422
860 data = uploaded.read()
861 if len(data) > _MAX_AMP_IMPORT_BYTES:
862 flash(_("File too large (maximum 10 MB)."), "danger")
863 return render_template("maintenance/amp_import_upload.html", aircraft=ac), 422
865 try:
866 wb = openpyxl.load_workbook(io.BytesIO(data), data_only=True)
867 components = list(ac.components)
868 rows = parse_amp_rows(wb, components)
869 except ValueError as exc:
870 flash(str(exc), "danger")
871 return render_template("maintenance/amp_import_upload.html", aircraft=ac), 422
873 if not rows:
874 flash(_("No task rows found in the uploaded file."), "danger")
875 return render_template("maintenance/amp_import_upload.html", aircraft=ac), 422
877 _cleanup_amp_import_tmp(uid)
878 safe_base = secure_filename(uploaded.filename) or "upload"
879 tmp_name = f"amp_import_{uid}_{uuid.uuid4().hex}_{safe_base}"
880 tmp_path = os.path.join(_amp_import_tmp_dir(), tmp_name)
881 with open(tmp_path, "wb") as fh:
882 fh.write(data)
884 session[_AMP_IMPORT_SESSION_KEY] = {
885 "uid": uid,
886 "aircraft_id": ac.id,
887 "tmp_path": tmp_path,
888 "original_filename": uploaded.filename,
889 }
891 preview_rows = _amp_preview_rows(ac, rows, components)
892 needs_review_count = sum(1 for r in rows if r.needs_review)
894 return render_template(
895 "maintenance/amp_import_review.html",
896 aircraft=ac,
897 preview_rows=preview_rows,
898 components=components,
899 row_count=len(rows),
900 needs_review_count=needs_review_count,
901 filename=uploaded.filename,
902 )
905@maintenance_bp.route(
906 "/aircraft/<aircraft_ref:aircraft_id>/maintenance/import/commit", methods=["POST"]
907)
908@login_required
909@require_role(*_MAINT_ROLES)
910def import_amp_commit(aircraft_id: int) -> ResponseReturnValue:
911 ac = _get_aircraft_or_404(aircraft_id)
912 uid = session["user_id"]
913 meta = session.get(_AMP_IMPORT_SESSION_KEY)
915 if not meta or meta.get("uid") != uid or meta.get("aircraft_id") != ac.id:
916 flash(_("Import session expired. Please upload the file again."), "warning")
917 return redirect(url_for("maintenance.import_amp_upload", aircraft_id=ac.id))
919 tmp_path: str = meta["tmp_path"]
920 original_filename: str = meta["original_filename"]
921 if not os.path.isfile(tmp_path):
922 flash(_("Temporary file not found. Please upload the file again."), "warning")
923 session.pop(_AMP_IMPORT_SESSION_KEY, None)
924 return redirect(url_for("maintenance.import_amp_upload", aircraft_id=ac.id))
926 with open(tmp_path, "rb") as fh:
927 data = fh.read()
928 wb = openpyxl.load_workbook(io.BytesIO(data), data_only=True)
929 components = list(ac.components)
930 components_by_id = {c.id: c for c in components}
931 rows = parse_amp_rows(wb, components)
933 current_hobbs = ac.total_engine_hours
934 current_flight_hours = ac.total_flight_hours
936 batch = MaintenanceImportBatch(
937 aircraft_id=ac.id,
938 source_filename=original_filename,
939 row_count=len(rows),
940 needs_review_count=sum(1 for r in rows if r.needs_review),
941 )
942 db.session.add(batch)
943 db.session.flush()
945 for i, r in enumerate(rows):
946 override_raw = (request.form.get(f"component_id_{i}") or "").strip()
947 component = None
948 if override_raw:
949 try:
950 component = components_by_id.get(int(override_raw))
951 except ValueError:
952 component = None
953 elif r.suggested_component_id is not None:
954 component = components_by_id.get(r.suggested_component_id)
956 basis = hours_basis_for_component(component)
957 due_h, due_d = compute_due_fields(
958 r.interval_hours,
959 r.interval_days,
960 basis,
961 current_hobbs,
962 current_flight_hours,
963 )
965 db.session.add(
966 MaintenanceTrigger(
967 aircraft_id=ac.id,
968 component_id=component.id if component else None,
969 name=r.name,
970 trigger_type=(
971 TriggerType.CALENDAR if due_d is not None else TriggerType.HOURS
972 ),
973 due_date=due_d,
974 interval_days=r.interval_days,
975 due_engine_hours=due_h,
976 interval_hours=r.interval_hours,
977 hours_basis=basis,
978 category=r.category,
979 reference=r.reference,
980 action=r.action,
981 part_number=r.part_number,
982 serial_number=r.serial_number,
983 notes=r.notes,
984 needs_review=r.needs_review,
985 import_batch_id=batch.id,
986 )
987 )
989 db.session.commit()
990 _cleanup_amp_import_tmp(uid)
991 activity(
992 "maintenance.amp_import",
993 aircraft_id=ac.id,
994 batch_id=batch.id,
995 row_count=batch.row_count,
996 )
998 review_phrase = ngettext(
999 "one flagged for review",
1000 "%(n)s flagged for review",
1001 batch.needs_review_count,
1002 n=batch.needs_review_count,
1003 )
1004 flash(
1005 ngettext(
1006 "Imported %(count)d maintenance item — %(review_phrase)s.",
1007 "Imported %(count)d maintenance items — %(review_phrase)s.",
1008 batch.row_count,
1009 count=batch.row_count,
1010 review_phrase=review_phrase,
1011 ),
1012 "success",
1013 )
1014 return redirect(url_for("maintenance.list_triggers", aircraft_id=ac.id))
1017@maintenance_bp.route("/aircraft/<aircraft_ref:aircraft_id>/maintenance/import/history")
1018@login_required
1019@require_role(*_MAINT_ROLES)
1020def import_amp_history(aircraft_id: int) -> ResponseReturnValue:
1021 ac = _get_aircraft_or_404(aircraft_id)
1022 batches = (
1023 MaintenanceImportBatch.query.filter_by(aircraft_id=ac.id)
1024 .order_by(MaintenanceImportBatch.imported_at.desc())
1025 .all()
1026 )
1027 return render_template(
1028 "maintenance/amp_import_history.html", aircraft=ac, batches=batches
1029 )
1032@maintenance_bp.route(
1033 "/aircraft/<aircraft_ref:aircraft_id>/maintenance/import/<int:batch_id>/rollback",
1034 methods=["POST"],
1035)
1036@login_required
1037@require_role(*_MAINT_ROLES)
1038def import_amp_rollback(aircraft_id: int, batch_id: int) -> ResponseReturnValue:
1039 ac = _get_aircraft_or_404(aircraft_id)
1040 batch = db.session.get(MaintenanceImportBatch, batch_id)
1041 if not batch or batch.aircraft_id != ac.id:
1042 abort(404)
1044 row_count = batch.row_count
1045 MaintenanceTrigger.query.filter_by(import_batch_id=batch.id).delete()
1046 db.session.delete(batch)
1047 db.session.commit()
1049 flash(
1050 ngettext(
1051 "Import rolled back: %(count)d item removed.",
1052 "Import rolled back: %(count)d items removed.",
1053 row_count,
1054 count=row_count,
1055 ),
1056 "success",
1057 )
1058 return redirect(url_for("maintenance.import_amp_history", aircraft_id=ac.id))
1061# ── AMP document export ─────────────────────────────────────────────────────
1064def _amp_export_context(ac: Aircraft) -> dict[str, Any] | None:
1065 """Build the template context for the AMP export (HTML preview and PDF
1066 download render the same template from the same data — see
1067 docs/maintenance_import.md's round-trip design note). Returns None if
1068 the aircraft has no AMP declaration profile yet."""
1069 decl: AmpDeclaration | None = ac.amp_declaration # type: ignore[assignment]
1070 if decl is None:
1071 return None
1073 components = list(ac.components)
1075 def _installed(comp_type: str) -> Component | None:
1076 return next(
1077 (c for c in components if c.type == comp_type and c.removed_at is None),
1078 None,
1079 )
1081 triggers = (
1082 MaintenanceTrigger.query.filter_by(aircraft_id=ac.id)
1083 .order_by(MaintenanceTrigger.name)
1084 .all()
1085 )
1087 category_has_items = {
1088 cat: any(t.category == cat for t in triggers) for cat in AmpCategory.ALL
1089 }
1090 has_alternative_tasks = any(t.is_alternative_to_ica for t in triggers)
1091 appendix_b_groups = [
1092 (cat, [t for t in triggers if t.category == cat])
1093 for cat in AmpCategory.ALL
1094 if category_has_items[cat]
1095 ]
1096 appendix_c_rows = [t for t in triggers if t.is_alternative_to_ica]
1097 interval_text = {
1098 t.id: format_interval(t.interval_hours, t.interval_days) for t in triggers
1099 }
1100 # A needs_review trigger has no interval yet and may not have a category
1101 # either — appendix_b_groups above would silently drop it if uncategorised.
1102 # Listed here regardless of category so nothing pending goes unmentioned in
1103 # the exported document, with its notes (the actual open question, if the
1104 # import captured one) alongside it.
1105 pending_review_rows = [t for t in triggers if t.needs_review]
1107 return {
1108 "aircraft": ac,
1109 "decl": decl,
1110 "triggers": triggers,
1111 "airframe_component": _installed(ComponentType.AIRFRAME),
1112 "engine_component": _installed(ComponentType.ENGINE),
1113 "propeller_component": _installed(ComponentType.PROPELLER),
1114 "category_has_items": category_has_items,
1115 "has_alternative_tasks": has_alternative_tasks,
1116 "appendix_b_groups": appendix_b_groups,
1117 "appendix_c_rows": appendix_c_rows,
1118 "interval_text": interval_text,
1119 "pending_review_rows": pending_review_rows,
1120 "revisions": ac.amp_revisions,
1121 "amp_basis": AmpBasis,
1122 "declaration_types": AmpDeclarationType,
1123 "certifying_party_kinds": AmpCertifyingPartyKind,
1124 }
1127def _amp_content_signature(context: dict[str, Any]) -> str:
1128 """SHA-256 fingerprint of the exportable AMP content — used to detect
1129 whether anything has changed since the last declared AmpRevision.
1131 Deliberately a hash of the underlying *data*, not of the rendered
1132 HTML/PDF: hashing the render would make every export "drift" after any
1133 future template/CSS tweak or a locale switch, even with byte-identical
1134 underlying data — a false positive on every single aircraft at once.
1135 """
1136 ac = context["aircraft"]
1137 decl = context["decl"]
1139 def _component(c: Component | None) -> dict[str, Any] | None:
1140 return (
1141 None
1142 if c is None
1143 else {"make": c.make, "model": c.model, "serial_number": c.serial_number}
1144 )
1146 payload = {
1147 "aircraft": {
1148 "registration": ac.registration,
1149 "make": ac.make,
1150 "model": ac.model,
1151 },
1152 "airframe": _component(context["airframe_component"]),
1153 "engine": _component(context["engine_component"]),
1154 "propeller": _component(context["propeller_component"]),
1155 "declaration": {
1156 col.name: str(getattr(decl, col.name))
1157 for col in AmpDeclaration.__table__.columns
1158 if col.name not in ("aircraft_id", "updated_at")
1159 },
1160 "triggers": [
1161 {
1162 "name": t.name,
1163 "category": t.category,
1164 "is_alternative_to_ica": t.is_alternative_to_ica,
1165 "alternative_task_notes": t.alternative_task_notes,
1166 "reference": t.reference,
1167 "interval_hours": str(t.interval_hours),
1168 "interval_days": t.interval_days,
1169 "needs_review": t.needs_review,
1170 "notes": t.notes,
1171 }
1172 for t in sorted(context["triggers"], key=lambda t: t.id)
1173 ],
1174 "revisions": [
1175 {
1176 "revision_number": r.revision_number,
1177 "revision_content": r.revision_content,
1178 "revision_date": str(r.revision_date),
1179 }
1180 for r in context["revisions"]
1181 ],
1182 }
1183 return hashlib.sha256(
1184 json.dumps(payload, sort_keys=True, default=str).encode("utf-8")
1185 ).hexdigest()
1188def _amp_draft_status(
1189 context: dict[str, Any],
1190) -> tuple[bool, AmpRevision | None]:
1191 """Return (is_draft, latest_revision). is_draft is True when there's no
1192 declared revision yet, or the AMP's current data no longer matches the
1193 fingerprint recorded when the latest one was declared."""
1194 revisions = context["revisions"]
1195 latest_revision = revisions[-1] if revisions else None
1196 if latest_revision is None:
1197 return True, None
1198 current_hash = _amp_content_signature(context)
1199 return current_hash != latest_revision.content_hash, latest_revision
1202def _amp_pdf_filename(
1203 ac: Aircraft, is_draft: bool, latest_revision: AmpRevision | None
1204) -> str:
1205 date_str = (
1206 latest_revision.revision_date.isoformat()
1207 if not is_draft and latest_revision and latest_revision.revision_date
1208 else _date.today().isoformat()
1209 )
1210 revision_str = (
1211 "draft"
1212 if is_draft or latest_revision is None
1213 else latest_revision.revision_number
1214 )
1215 return secure_filename(f"{date_str}-AMP-{ac.registration}-{revision_str}.pdf")
1218def _amp_pdf_storage_path(revision_id: int) -> tuple[str, str]:
1219 """Return (folder, relative_filename) for a revision's cached canonical PDF."""
1220 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
1221 return folder, f"amp_revision_{revision_id}.pdf"
1224def _render_and_cache_amp_pdf(context: dict[str, Any], revision: AmpRevision) -> bytes:
1225 """Render the (non-draft) PDF for *revision* and save it as its
1226 canonical file, setting pdf_path. Caller commits. Does not check
1227 is_draft — only call this when the caller has already established the
1228 current data matches this revision."""
1229 html = render_template("maintenance/amp_export_pdf.html", is_draft=False, **context)
1230 pdf_bytes: bytes = HTML(string=html, base_url=request.url_root).write_pdf()
1231 folder, stored_name = _amp_pdf_storage_path(revision.id)
1232 os.makedirs(folder, exist_ok=True)
1233 with open(os.path.join(folder, stored_name), "wb") as f:
1234 f.write(pdf_bytes)
1235 revision.pdf_path = stored_name
1236 return pdf_bytes
1239@maintenance_bp.route("/aircraft/<aircraft_ref:aircraft_id>/maintenance/amp/export")
1240@login_required
1241@require_role(*_MAINT_ROLES)
1242def export_amp(aircraft_id: int) -> ResponseReturnValue:
1243 ac = _get_aircraft_or_404(aircraft_id)
1244 context = _amp_export_context(ac)
1245 if context is None:
1246 flash(
1247 _("Fill in the AMP declaration profile before exporting the AMP document."),
1248 "warning",
1249 )
1250 return redirect(url_for("maintenance.edit_amp_declaration", aircraft_id=ac.id))
1252 is_draft, _latest_revision = _amp_draft_status(context)
1253 return render_template("maintenance/amp_export.html", is_draft=is_draft, **context)
1256@maintenance_bp.route("/aircraft/<aircraft_ref:aircraft_id>/maintenance/amp/export/pdf")
1257@login_required
1258@require_role(*_MAINT_ROLES)
1259def export_amp_pdf(aircraft_id: int) -> ResponseReturnValue:
1260 ac = _get_aircraft_or_404(aircraft_id)
1261 context = _amp_export_context(ac)
1262 if context is None:
1263 flash(
1264 _("Fill in the AMP declaration profile before exporting the AMP document."),
1265 "warning",
1266 )
1267 return redirect(url_for("maintenance.edit_amp_declaration", aircraft_id=ac.id))
1269 is_draft, latest_revision = _amp_draft_status(context)
1270 filename = _amp_pdf_filename(ac, is_draft, latest_revision)
1272 # Not a draft and already cached: serve the exact bytes generated the
1273 # first time this revision was downloaded, rather than a fresh render.
1274 if not is_draft and latest_revision and latest_revision.pdf_path:
1275 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
1276 cached_path = os.path.join(folder, latest_revision.pdf_path)
1277 if os.path.exists(cached_path):
1278 with open(cached_path, "rb") as f:
1279 pdf_bytes = f.read()
1280 return Response(
1281 pdf_bytes,
1282 mimetype="application/pdf",
1283 headers={"Content-Disposition": f'attachment; filename="{filename}"'},
1284 )
1286 if not is_draft and latest_revision:
1287 # Not a draft and not cached yet — normally already generated when
1288 # the revision was added; this is the fallback for a revision
1289 # created outside that flow (a script, a failed eager render).
1290 pdf_bytes = _render_and_cache_amp_pdf(context, latest_revision)
1291 db.session.commit()
1292 else:
1293 html = render_template(
1294 "maintenance/amp_export_pdf.html", is_draft=is_draft, **context
1295 )
1296 pdf_bytes = HTML(string=html, base_url=request.url_root).write_pdf()
1298 return Response(
1299 pdf_bytes,
1300 mimetype="application/pdf",
1301 headers={"Content-Disposition": f'attachment; filename="{filename}"'},
1302 )
1305@maintenance_bp.route(
1306 "/aircraft/<aircraft_ref:aircraft_id>/amp/revisions/<int:revision_id>/pdf"
1307)
1308@login_required
1309@require_role(*_MAINT_ROLES)
1310def download_amp_revision_pdf(
1311 aircraft_id: int, revision_id: int
1312) -> ResponseReturnValue:
1313 """Re-download a specific past revision's canonical PDF, byte-for-byte
1314 as first generated — even after later edits have moved the live AMP
1315 data on. Only available for a revision that was actually downloaded at
1316 least once while it was still the current, undrafted state; there is
1317 no field-level history to reconstruct one that wasn't."""
1318 ac = _get_aircraft_or_404(aircraft_id)
1319 rev = db.session.get(AmpRevision, revision_id)
1320 if not rev or rev.aircraft_id != ac.id:
1321 abort(404)
1322 if not rev.pdf_path:
1323 flash(
1324 _(
1325 "No saved PDF for revision '%(number)s' — it was never "
1326 "downloaded while current.",
1327 number=rev.revision_number,
1328 ),
1329 "warning",
1330 )
1331 return redirect(url_for("maintenance.edit_amp_declaration", aircraft_id=ac.id))
1333 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
1334 path = os.path.join(folder, rev.pdf_path)
1335 if not os.path.exists(path):
1336 flash(
1337 _(
1338 "The saved PDF for revision '%(number)s' is missing on disk.",
1339 number=rev.revision_number,
1340 ),
1341 "danger",
1342 )
1343 return redirect(url_for("maintenance.edit_amp_declaration", aircraft_id=ac.id))
1345 with open(path, "rb") as f:
1346 pdf_bytes = f.read()
1347 date_str = rev.revision_date.isoformat() if rev.revision_date else "undated"
1348 filename = secure_filename(
1349 f"{date_str}-AMP-{ac.registration}-{rev.revision_number}.pdf"
1350 )
1351 return Response(
1352 pdf_bytes,
1353 mimetype="application/pdf",
1354 headers={"Content-Disposition": f'attachment; filename="{filename}"'},
1355 )