Coverage for app/documents/routes.py: 18%
634 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-13 03:30 +0000
1import contextlib
2import io
3import logging
4import mimetypes
5import os
6import re as _re
7import uuid
8import zipfile
9from datetime import UTC, datetime
10from datetime import date as _date
12from flask import ( # pyright: ignore[reportMissingImports]
13 Blueprint,
14 Response,
15 abort,
16 current_app,
17 flash,
18 jsonify,
19 redirect,
20 render_template,
21 request,
22 session,
23 url_for,
24)
25from flask.typing import ResponseReturnValue # pyright: ignore[reportMissingImports]
26from flask_babel import gettext as _ # pyright: ignore[reportMissingImports]
27from flask_babel import lazy_gettext as _l
28from flask_babel import ngettext
29from models import ( # pyright: ignore[reportMissingImports]
30 Aircraft,
31 Component,
32 DocCategory,
33 DocType,
34 Document,
35 PendingReconcile,
36 Role,
37 Tenant,
38 TenantUser,
39 db,
40)
41from services import document_lifecycle as lifecycle # pyright: ignore[reportMissingImports]
42from services.upload_layout import ( # pyright: ignore[reportMissingImports]
43 is_app_managed,
44)
45from sqlalchemy.orm import joinedload # pyright: ignore[reportMissingImports]
46from utils import ( # pyright: ignore[reportMissingImports]
47 activity,
48 login_required,
49 require_role,
50 user_can_access_aircraft,
51)
52from werkzeug.datastructures import FileStorage
53from werkzeug.utils import secure_filename
55log = logging.getLogger(__name__)
57documents_bp = Blueprint("documents", __name__)
59_OWNER_ROLES = (Role.ADMIN, Role.OWNER)
61_ALLOWED_EXTS = {
62 ".jpg",
63 ".jpeg",
64 ".png",
65 ".gif",
66 ".webp",
67 ".heic",
68 ".pdf",
69 ".doc",
70 ".docx",
71 ".xls",
72 ".xlsx",
73 ".txt",
74}
76_PILOT_DOC_TYPES = [
77 (DocType.LICENSE, _l("Licence")),
78 (DocType.MEDICAL, _l("Medical certificate")),
79]
81_AIRCRAFT_DOC_TYPES = [
82 (DocType.INSURANCE_CERT, _l("Insurance Certificate")),
83 (DocType.ARC, _l("Airworthiness Review Certificate (ARC)")),
84]
86# Sentinel accepted by the aircraft Documents list's type filter alongside the
87# real DocType values above, to isolate documents with no type set -- e.g. an
88# insurance/ARC certificate uploaded without picking "Document type", which
89# silently falls outside the expiry-tracking/supersession machinery those two
90# real types drive (see _EXPIRY_DRIVING_DOC_TYPES).
91_NO_TYPE_FILTER = "none"
92_NO_TYPE_LABEL = _l("No type set")
94# Aircraft documents list pagination -- same values/convention as the pilot
95# logbook (pilots/routes.py _VALID_PER_PAGE/_DEFAULT_PER_PAGE).
96_DOCS_VALID_PER_PAGE = (10, 20, 50, 100)
97_DOCS_DEFAULT_PER_PAGE = 20
99# doc_type -> Aircraft attribute it drives. Aircraft.insurance_expiry/arc_expiry
100# are synced caches, not directly user-editable — see the comment on those
101# columns in models.py. Kept in step by _recompute_expiry_field() below
102# whenever a Document of one of these types is uploaded, edited, or deleted.
103_EXPIRY_DRIVING_DOC_TYPES: dict[str, str] = lifecycle.AIRCRAFT_FIELDS
105# Human-readable labels for each DocCategory value
106_CATEGORY_LABELS: dict[str, str] = {
107 DocCategory.MAINTENANCE: "Maintenance",
108 DocCategory.INSURANCE: "Insurance",
109 DocCategory.POH: "POH / Flight Manual",
110 DocCategory.AIRWORTHINESS: "Airworthiness",
111 DocCategory.LOGBOOK: "Logbook",
112 DocCategory.INVOICE: "Invoice",
113 DocCategory.OTHER: "Other",
114 DocCategory.UNCATEGORISED: "Uncategorised",
115}
118# ── Helpers ───────────────────────────────────────────────────────────────────
121def _tenant_id() -> int:
122 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
123 if not tu:
124 abort(403)
125 return int(tu.tenant_id)
128def _get_tenant() -> Tenant:
129 tid = _tenant_id()
130 t = db.session.get(Tenant, tid)
131 if not t:
132 abort(403) # pragma: no cover
133 return t
136def _get_aircraft_or_404(aircraft_id: int) -> Aircraft:
137 ac = db.session.get(Aircraft, aircraft_id)
138 if (
139 not ac
140 or ac.tenant_id != _tenant_id()
141 or not user_can_access_aircraft(aircraft_id)
142 ):
143 abort(404)
144 return ac
147def active_document_for(
148 aircraft_id: int,
149 doc_type: str,
150 component_id: int | None = None,
151 as_of: _date | None = None,
152) -> Document | None:
153 """The aircraft's (or component's) document of that type in force on
154 *as_of* — see services.document_lifecycle.active for the rules."""
155 return lifecycle.active(
156 lifecycle.Owner(aircraft_id=aircraft_id, component_id=component_id),
157 doc_type,
158 as_of,
159 )
162def effective_coverage_until(
163 aircraft_id: int,
164 doc_type: str,
165 component_id: int | None = None,
166 as_of: _date | None = None,
167) -> _date | None:
168 """How far the aircraft's coverage of that type runs without a gap,
169 through renewals already on file — services.document_lifecycle."""
170 return lifecycle.coverage_until(
171 lifecycle.Owner(aircraft_id=aircraft_id, component_id=component_id),
172 doc_type,
173 as_of,
174 )
177def _recompute_expiry_field(ac: Aircraft, doc_type: str | None) -> None:
178 """After any create/edit/delete that could change which certificate is
179 in force (and daily): mark superseded ones, and sync
180 Aircraft.insurance_expiry/arc_expiry — services.document_lifecycle.
181 A no-op for a type without a lifecycle."""
182 lifecycle.refresh(lifecycle.Owner(aircraft_id=ac.id), doc_type)
185def _parse_date_field(raw: str) -> _date | None:
186 """Parse an optional YYYY-MM-DD form field; invalid/blank input is
187 treated as "not set" rather than rejected outright, matching how
188 valid_until has always been handled here."""
189 raw = raw.strip()
190 if not raw:
191 return None
192 try:
193 return _date.fromisoformat(raw)
194 except ValueError as exc:
195 log.debug("Invalid date field: %s", exc)
196 return None
199def _get_aircraft_document_or_404(aircraft: Aircraft, document_id: int) -> Document:
200 doc = db.session.get(Document, document_id)
201 if not doc or doc.aircraft_id != aircraft.id:
202 abort(404)
203 return doc
206def _delete_file(filename: str | None) -> None:
207 """Move file to _trash/ instead of hard-deleting.
209 Syncthing propagates the move to all peers; the file is recoverable.
210 If the file is not found, log and continue silently.
211 """
212 if not filename:
213 return
214 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
215 src = os.path.join(folder, filename)
216 if not os.path.exists(src):
217 current_app.logger.debug("File already absent, skipping trash: %s", filename)
218 return
219 try:
220 trash_dir = os.path.join(folder, "_trash")
221 os.makedirs(trash_dir, exist_ok=True)
222 dest_name = os.path.basename(filename)
223 dest = os.path.join(trash_dir, dest_name)
224 if os.path.exists(dest):
225 base, ext = os.path.splitext(dest_name)
226 dest = os.path.join(trash_dir, f"{base}_{uuid.uuid4().hex[:8]}{ext}")
227 os.rename(src, dest)
228 except OSError:
229 current_app.logger.debug("Could not move to trash: %s", filename)
232def _resolve_component(ac: Aircraft) -> Component | None:
233 raw = request.args.get("component_id") or request.form.get("component_id")
234 if not raw:
235 return None
236 try:
237 cid = int(raw)
238 except (ValueError, TypeError):
239 return None
240 comp = db.session.get(Component, cid)
241 return comp if (comp and comp.aircraft_id == ac.id) else None
244def _save_upload(file: FileStorage, label: str) -> tuple[str, str, int]:
245 """Save *file* flat to upload folder (legacy path, no canonical structure)."""
246 original = secure_filename(file.filename or "")
247 ext = os.path.splitext(original)[1].lower()
248 stored = f"doc_{label}_{uuid.uuid4().hex[:12]}{ext}"
249 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
250 os.makedirs(folder, exist_ok=True)
251 file.save(os.path.join(folder, stored))
252 mime = mimetypes.guess_type(original)[0] or "application/octet-stream"
253 size = os.path.getsize(os.path.join(folder, stored))
254 return stored, mime, size
257def _ensure_tenant_slug(tenant: Tenant) -> str:
258 """Return tenant.slug, generating one from the name if not yet set."""
259 if tenant.slug:
260 return str(tenant.slug)
261 base = _re.sub(r"[^a-z0-9]+", "-", tenant.name.lower()).strip("-")[:64]
262 slug = base
263 n = 1
264 while Tenant.query.filter(Tenant.slug == slug, Tenant.id != tenant.id).first():
265 slug = f"{base}-{n}"
266 n += 1
267 tenant.slug = slug
268 db.session.flush()
269 return slug
272def _safe_path_component(s: str) -> str:
273 """Strip characters that are unsafe in filesystem path segments."""
274 return _re.sub(r'[<>:"/\\|?*\x00-\x1f]', "", s).strip()
277def _safe_join(upload_folder: str, *parts: str) -> str:
278 """Join parts under upload_folder; abort(400) if the result would escape it."""
279 root = os.path.realpath(upload_folder)
280 joined = os.path.normpath(os.path.join(root, *parts))
281 if not (joined == root or joined.startswith(root + os.sep)):
282 abort(400)
283 return joined
286_RECONCILE_FILENAME_RE = _re.compile(r"^(\d{4}-\d{2}-\d{2}) - (.+?)(\.[^.]+)?$")
289def _parse_reconcile_filename(
290 cat_str: str, name_part: str
291) -> tuple[str | None, str | None, _date | None]:
292 """Parse a Syncthing-mounted path's category segment and its
293 "YYYY-MM-DD - title.ext" filename segment into (category, title_hint,
294 date_hint). Shared by scan_documents and rename_reconcile_folder's
295 inline rescan so a scan and a post-rename rescan can never disagree on
296 how to interpret the same on-disk layout."""
297 category = cat_str.lower() if cat_str.lower() in DocCategory.ALL else None
298 title_hint: str | None = None
299 date_hint: _date | None = None
300 m = _RECONCILE_FILENAME_RE.match(name_part)
301 if m:
302 with contextlib.suppress(
303 ValueError
304 ): # regex matched date-like string but it's invalid; treat as no date
305 date_hint = _date.fromisoformat(m.group(1))
306 title_hint = m.group(2)
307 else:
308 title_hint = os.path.splitext(name_part)[0]
309 return category, title_hint, date_hint
312def _save_upload_canonical(
313 file: FileStorage,
314 tenant: Tenant,
315 aircraft: Aircraft,
316 category: str,
317 title: str | None,
318) -> tuple[str, str, int]:
319 """Save *file* to the canonical Syncthing-compatible path structure.
321 Returns (relpath, mime_type, size_bytes) where relpath is relative to
322 UPLOAD_FOLDER and suitable for storage in Document.filename.
323 """
324 original = secure_filename(file.filename or "unnamed")
325 ext = os.path.splitext(original)[1].lower()
326 today = _date.today().isoformat()
327 safe_title = _safe_path_component(title or os.path.splitext(original)[0])[:100]
328 fname = f"{today} - {safe_title}{ext}"
330 slug = _ensure_tenant_slug(tenant)
331 safe_reg = aircraft.registration.replace("/", "-").replace(" ", "-").upper()
332 relpath = os.path.join(slug, safe_reg, category, fname)
334 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
335 full_dir = _safe_join(folder, slug, safe_reg, category)
336 os.makedirs(full_dir, exist_ok=True)
338 dest = _safe_join(folder, relpath)
339 # If a file with this name already exists (e.g. same title + date), add a short suffix
340 if os.path.exists(dest):
341 base, ext2 = os.path.splitext(fname)
342 relpath = os.path.join(
343 slug, safe_reg, category, f"{base}_{uuid.uuid4().hex[:6]}{ext2}"
344 )
345 dest = _safe_join(folder, relpath)
347 file.save(dest)
348 mime = mimetypes.guess_type(original)[0] or "application/octet-stream"
349 size = os.path.getsize(dest)
350 return relpath, mime, size
353def _current_role() -> Role | None:
354 tu = TenantUser.query.filter_by(user_id=session.get("user_id")).first()
355 return tu.role if tu else None
358def _doc_broken(doc: Document) -> bool:
359 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
360 return not os.path.exists(os.path.join(folder, doc.filename))
363# ── Title suggestions ─────────────────────────────────────────────────────────
366@documents_bp.route("/documents/title-suggestions")
367@login_required
368def title_suggestions() -> ResponseReturnValue:
369 q = request.args.get("q", "").strip()
370 owner_type = request.args.get("owner_type", "aircraft")
371 uid = int(session["user_id"])
373 if owner_type == "pilot":
374 base = Document.query.filter(
375 Document.pilot_user_id == uid,
376 Document.title.isnot(None),
377 )
378 else:
379 tid = _tenant_id()
380 aircraft_ids = [
381 row.id
382 for row in Aircraft.query.filter_by(tenant_id=tid)
383 .with_entities(Aircraft.id)
384 .all()
385 ]
386 base = Document.query.filter(
387 Document.aircraft_id.in_(aircraft_ids),
388 Document.title.isnot(None),
389 )
390 if owner_type == "component":
391 base = base.filter(Document.component_id.isnot(None))
392 else:
393 base = base.filter(
394 Document.component_id.is_(None),
395 Document.flight_entry_id.is_(None),
396 )
398 if q:
399 base = base.filter(Document.title.ilike(f"{q}%"))
401 rows = (
402 base.with_entities(Document.title)
403 .distinct()
404 .order_by(Document.title)
405 .limit(10)
406 .all()
407 )
408 return jsonify([r.title for r in rows])
411# ── Aircraft document list ────────────────────────────────────────────────────
414@documents_bp.route("/aircraft/<aircraft_ref:aircraft_id>/documents")
415@login_required
416def list_documents(aircraft_id: int) -> ResponseReturnValue:
417 ac = _get_aircraft_or_404(aircraft_id)
418 show_sensitive = request.args.get("sensitive") == "1"
419 doc_type_labels = dict(_AIRCRAFT_DOC_TYPES)
420 doc_type_labels[_NO_TYPE_FILTER] = _NO_TYPE_LABEL
421 filter_doc_type = request.args.get("doc_type") or None
422 if filter_doc_type not in doc_type_labels:
423 filter_doc_type = None
424 query = Document.query.filter_by(aircraft_id=ac.id).options(
425 joinedload(Document.component), # type: ignore[arg-type]
426 joinedload(Document.flight_entry), # type: ignore[arg-type]
427 joinedload(Document.expense), # type: ignore[arg-type]
428 )
429 if not show_sensitive:
430 query = query.filter_by(is_sensitive=False)
431 if filter_doc_type == _NO_TYPE_FILTER:
432 query = query.filter(Document.doc_type.is_(None))
433 elif filter_doc_type:
434 query = query.filter_by(doc_type=filter_doc_type)
435 query = query.order_by(Document.uploaded_at.desc())
437 page = request.args.get("page", 1, type=int)
438 pp_raw = request.args.get("per_page", str(_DOCS_DEFAULT_PER_PAGE))
439 show_all = pp_raw == "all"
440 per_page = (
441 None
442 if show_all
443 else (
444 int(pp_raw)
445 if pp_raw.isdigit() and int(pp_raw) in _DOCS_VALID_PER_PAGE
446 else _DOCS_DEFAULT_PER_PAGE
447 )
448 )
449 if show_all:
450 docs = query.all()
451 pagination = None
452 total_count = len(docs)
453 else:
454 pagination = query.paginate(page=page, per_page=per_page, error_out=False)
455 docs = pagination.items
456 total_count = pagination.total
458 sensitive_count = Document.query.filter_by(
459 aircraft_id=ac.id, is_sensitive=True
460 ).count()
461 role = _current_role()
462 is_owner = role in _OWNER_ROLES
463 broken_ids = {doc.id for doc in docs if _doc_broken(doc)}
464 # One active document per doc_type actually present in this list --
465 # cheap even unfiltered, since an aircraft only ever has the two
466 # expiry-driving types (ARC/insurance) to look up.
467 doc_types_shown = {
468 d.doc_type for d in docs if d.doc_type in _EXPIRY_DRIVING_DOC_TYPES
469 }
470 active_doc_ids = {
471 active.id
472 for dt in doc_types_shown
473 for active in [active_document_for(ac.id, dt, component_id=None)]
474 if active is not None
475 }
476 return render_template(
477 "documents/list.html",
478 aircraft=ac,
479 docs=docs,
480 pagination=pagination,
481 per_page=pp_raw,
482 valid_per_page=_DOCS_VALID_PER_PAGE,
483 total_count=total_count,
484 show_sensitive=show_sensitive,
485 sensitive_count=sensitive_count,
486 is_owner=is_owner,
487 broken_ids=broken_ids,
488 category_labels=_CATEGORY_LABELS,
489 filter_doc_type=filter_doc_type,
490 doc_type_filter_options=list(doc_type_labels.items()),
491 active_doc_ids=active_doc_ids,
492 expiry_driving_doc_types=_EXPIRY_DRIVING_DOC_TYPES,
493 )
496# ── Upload aircraft document ──────────────────────────────────────────────────
499@documents_bp.route(
500 "/aircraft/<aircraft_ref:aircraft_id>/documents/upload", methods=["GET", "POST"]
501)
502@login_required
503@require_role(*_OWNER_ROLES)
504def upload_document(aircraft_id: int) -> ResponseReturnValue:
505 ac = _get_aircraft_or_404(aircraft_id)
506 component = _resolve_component(ac)
507 # "Replace" on the documents list: the certificate being replaced, to
508 # preselect its type and category and say what happens to it.
509 replacing = None
510 replace_id = request.values.get("replace", type=int)
511 if replace_id:
512 candidate = db.session.get(Document, replace_id)
513 if candidate is not None and candidate.aircraft_id == ac.id:
514 replacing = candidate
515 prefill = {
516 "replacing": replacing,
517 "selected_type": request.values.get("doc_type")
518 or (replacing.doc_type if replacing else ""),
519 "selected_category": replacing.category if replacing else "",
520 }
522 if request.method == "POST":
523 file = request.files.get("file")
524 title = request.form.get("title", "").strip() or None
525 is_sensitive = bool(request.form.get("is_sensitive"))
526 doc_type = request.form.get("doc_type") or None
527 category = request.form.get("category") or None
528 if category and category not in DocCategory.ALL:
529 category = None
530 valid_from = _parse_date_field(request.form.get("valid_from", ""))
531 valid_until = _parse_date_field(request.form.get("valid_until", ""))
533 def _re_render(msg: str | None = None) -> str:
534 if msg:
535 flash(msg, "danger")
536 return render_template(
537 "documents/upload_form.html",
538 aircraft=ac,
539 component=component,
540 doc_types=_AIRCRAFT_DOC_TYPES,
541 categories=list(_CATEGORY_LABELS.items()),
542 **prefill,
543 )
545 if not file or not file.filename:
546 return _re_render(_("Please select a file to upload."))
548 original = secure_filename(file.filename)
549 ext = os.path.splitext(original)[1].lower()
550 if ext not in _ALLOWED_EXTS:
551 return _re_render(
552 _("File type '%(ext)s' is not allowed.", ext=ext or "unknown")
553 )
555 if valid_from and valid_until and valid_from > valid_until:
556 return _re_render(_("'Valid from' must not be after 'Valid until'."))
558 # Use canonical path when a category is set and this is an aircraft doc
559 if category and not component:
560 tenant = _get_tenant()
561 stored, mime, size = _save_upload_canonical(
562 file, tenant, ac, category, title
563 )
564 else:
565 label = f"comp{component.id}" if component else f"ac{ac.id}"
566 stored, mime, size = _save_upload(file, label)
568 doc = Document(
569 aircraft_id=ac.id,
570 component_id=component.id if component else None,
571 filename=stored,
572 original_filename=original,
573 mime_type=mime,
574 size_bytes=size,
575 title=title,
576 doc_type=doc_type,
577 category=category,
578 valid_from=valid_from,
579 valid_until=valid_until,
580 is_sensitive=is_sensitive,
581 )
582 db.session.add(doc)
584 # Insurance document without an explicit type: infer it from the
585 # category for backward compatibility with the category-only flow.
586 if doc.doc_type is None and category == DocCategory.INSURANCE and not component:
587 doc.doc_type = DocType.INSURANCE_CERT
589 # An insurance/ARC document: work out again which one is in force
590 # (it may supersede the current one now, or — starting later — only
591 # on its valid_from) and sync the aircraft's expiry date.
592 if not component:
593 _recompute_expiry_field(ac, doc.doc_type)
595 db.session.commit()
596 activity(
597 "document.uploaded",
598 document_id=doc.id,
599 aircraft_id=ac.id,
600 title=doc.title or "",
601 )
603 flash(_("Document uploaded."), "success")
604 return redirect(url_for("documents.list_documents", aircraft_id=ac.id))
606 return render_template(
607 "documents/upload_form.html",
608 aircraft=ac,
609 component=component,
610 doc_types=_AIRCRAFT_DOC_TYPES,
611 categories=list(_CATEGORY_LABELS.items()),
612 **prefill,
613 )
616# ── Edit aircraft document ────────────────────────────────────────────────────
619@documents_bp.route(
620 "/aircraft/<aircraft_ref:aircraft_id>/documents/<int:document_id>/edit",
621 methods=["GET", "POST"],
622)
623@login_required
624@require_role(*_OWNER_ROLES)
625def edit_document(aircraft_id: int, document_id: int) -> ResponseReturnValue:
626 ac = _get_aircraft_or_404(aircraft_id)
627 doc = _get_aircraft_document_or_404(ac, document_id)
629 if request.method == "POST":
630 doc.title = request.form.get("title", "").strip() or None
631 doc.is_sensitive = bool(request.form.get("is_sensitive"))
632 category = request.form.get("category") or None
633 if category and category not in DocCategory.ALL:
634 category = None
636 old_doc_type = doc.doc_type
637 if not doc.component_id:
638 new_doc_type = request.form.get("doc_type") or None
639 if new_doc_type in dict(_AIRCRAFT_DOC_TYPES) or new_doc_type is None:
640 doc.doc_type = new_doc_type
642 old_category = doc.category
643 doc.category = category
645 # If the category changed and the file lives in the canonical path, move it
646 if category and old_category and category != old_category and doc.filename:
647 parts = doc.filename.replace("\\", "/").split("/")
648 if len(parts) >= 4 and parts[2] == old_category:
649 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
650 old_full = _safe_join(folder, doc.filename)
651 if os.path.exists(old_full):
652 new_relpath = "/".join(parts[:2] + [category] + parts[3:])
653 new_full = _safe_join(folder, new_relpath)
654 os.makedirs(os.path.dirname(new_full), exist_ok=True)
655 try:
656 os.rename(old_full, new_full)
657 doc.filename = new_relpath
658 except OSError as exc:
659 log.warning("Could not move document file: %s", exc)
661 new_valid_from = _parse_date_field(request.form.get("valid_from", ""))
662 new_valid_until = _parse_date_field(request.form.get("valid_until", ""))
663 if new_valid_from and new_valid_until and new_valid_from > new_valid_until:
664 flash(_("'Valid from' must not be after 'Valid until'."), "danger")
665 return render_template(
666 "documents/edit_form.html",
667 aircraft=ac,
668 doc=doc,
669 doc_types=_AIRCRAFT_DOC_TYPES,
670 categories=list(_CATEGORY_LABELS.items()),
671 )
672 doc.valid_from = new_valid_from
673 doc.valid_until = new_valid_until
675 # Recompute whichever Aircraft field(s) this document could affect —
676 # both its old and new doc_type, in case the type itself changed.
677 _recompute_expiry_field(ac, old_doc_type)
678 _recompute_expiry_field(ac, doc.doc_type)
680 db.session.commit()
681 flash(_("Document updated."), "success")
682 return redirect(url_for("documents.list_documents", aircraft_id=ac.id))
684 return render_template(
685 "documents/edit_form.html",
686 aircraft=ac,
687 doc=doc,
688 doc_types=_AIRCRAFT_DOC_TYPES,
689 categories=list(_CATEGORY_LABELS.items()),
690 )
693# ── Delete aircraft document ──────────────────────────────────────────────────
696@documents_bp.route(
697 "/aircraft/<aircraft_ref:aircraft_id>/documents/<int:document_id>/delete",
698 methods=["POST"],
699)
700@login_required
701@require_role(*_OWNER_ROLES)
702def delete_document(aircraft_id: int, document_id: int) -> ResponseReturnValue:
703 ac = _get_aircraft_or_404(aircraft_id)
704 doc = _get_aircraft_document_or_404(ac, document_id)
705 doc_type = doc.doc_type
706 activity("document.deleted", document_id=document_id, aircraft_id=aircraft_id)
707 _delete_file(doc.filename)
708 db.session.delete(doc)
709 db.session.commit()
711 # Deleting the certificate in force puts the one it had superseded back
712 # in force (or clears the aircraft's date when nothing is left).
713 if doc_type in _EXPIRY_DRIVING_DOC_TYPES:
714 _recompute_expiry_field(ac, doc_type)
715 db.session.commit()
717 flash(_("Document deleted."), "success")
718 return redirect(url_for("documents.list_documents", aircraft_id=ac.id))
721# ── Download all aircraft documents as ZIP ────────────────────────────────────
724@documents_bp.route("/aircraft/<aircraft_ref:aircraft_id>/documents/download-all")
725@login_required
726def download_all_documents(aircraft_id: int) -> ResponseReturnValue:
727 ac = _get_aircraft_or_404(aircraft_id)
728 role = _current_role()
729 include_sensitive = role in _OWNER_ROLES
731 query = Document.query.filter_by(aircraft_id=ac.id)
732 if not include_sensitive:
733 query = query.filter_by(is_sensitive=False)
734 docs = query.order_by(Document.uploaded_at.asc()).all()
736 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
737 buf = io.BytesIO()
738 manifest_lines = ["filename\ttitle\tcategory\ttype\tuploaded\n"]
740 with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
741 for doc in docs:
742 path = os.path.join(folder, doc.filename)
743 arcname = doc.original_filename
744 if os.path.exists(path):
745 zf.write(path, arcname=arcname)
746 doc_type_label = doc.doc_type or ""
747 category_label = _CATEGORY_LABELS.get(
748 doc.category or "", doc.category or ""
749 )
750 uploaded = doc.uploaded_at.strftime("%Y-%m-%d") if doc.uploaded_at else ""
751 manifest_lines.append(
752 f"{arcname}\t{doc.title or ''}\t{category_label}\t{doc_type_label}\t{uploaded}\n"
753 )
754 zf.writestr("manifest.txt", "".join(manifest_lines))
756 buf.seek(0)
757 reg = ac.registration.replace("/", "-")
758 zip_name = f"aircraft-{reg}-documents.zip"
759 return Response(
760 buf.read(),
761 mimetype="application/zip",
762 headers={"Content-Disposition": f'attachment; filename="{zip_name}"'},
763 )
766# ── Insurance / ARC certificate quick upload ──────────────────────────────────
769def _upload_expiry_cert(
770 aircraft_id: int, doc_type: str, category: str, title: str
771) -> ResponseReturnValue:
772 """Shared handler for the small "Upload cert" forms on the aircraft
773 detail page's Insurance and ARC sections — a file plus its valid-until
774 date, which becomes the new active certificate for that doc_type."""
775 ac = _get_aircraft_or_404(aircraft_id)
776 file = request.files.get("file")
778 if not file or not file.filename:
779 flash(_("Please select a file to upload."), "danger")
780 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
782 original = secure_filename(file.filename)
783 ext = os.path.splitext(original)[1].lower()
784 if ext not in _ALLOWED_EXTS:
785 flash(_("File type '%(ext)s' is not allowed.", ext=ext or "unknown"), "danger")
786 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
788 valid_from = _parse_date_field(request.form.get("valid_from", ""))
789 valid_until = _parse_date_field(request.form.get("valid_until", ""))
790 if valid_from and valid_until and valid_from > valid_until:
791 flash(_("'Valid from' must not be after 'Valid until'."), "danger")
792 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
794 tenant = _get_tenant()
795 stored, mime, size = _save_upload_canonical(file, tenant, ac, category, title)
797 new_cert = Document(
798 aircraft_id=ac.id,
799 filename=stored,
800 original_filename=original,
801 mime_type=mime,
802 size_bytes=size,
803 title=title,
804 doc_type=doc_type,
805 category=category,
806 valid_from=valid_from,
807 valid_until=valid_until,
808 is_sensitive=True,
809 )
810 db.session.add(new_cert)
811 _recompute_expiry_field(ac, doc_type)
812 db.session.commit()
813 flash(_("Certificate uploaded."), "success")
814 return redirect(url_for("aircraft.detail", aircraft_id=ac.id))
817@documents_bp.route(
818 "/aircraft/<aircraft_ref:aircraft_id>/insurance-cert/upload", methods=["POST"]
819)
820@login_required
821@require_role(*_OWNER_ROLES)
822def upload_insurance_cert(aircraft_id: int) -> ResponseReturnValue:
823 return _upload_expiry_cert(
824 aircraft_id,
825 DocType.INSURANCE_CERT,
826 DocCategory.INSURANCE,
827 _("Insurance Certificate"),
828 )
831@documents_bp.route(
832 "/aircraft/<aircraft_ref:aircraft_id>/arc-cert/upload", methods=["POST"]
833)
834@login_required
835@require_role(*_OWNER_ROLES)
836def upload_arc_cert(aircraft_id: int) -> ResponseReturnValue:
837 return _upload_expiry_cert(
838 aircraft_id,
839 DocType.ARC,
840 DocCategory.AIRWORTHINESS,
841 _("Airworthiness Review Certificate"),
842 )
845# ── Pilot document upload ─────────────────────────────────────────────────────
848@documents_bp.route("/pilot/documents/upload", methods=["GET", "POST"])
849@login_required
850def upload_pilot_document() -> ResponseReturnValue:
851 uid = int(session["user_id"])
852 owner = lifecycle.Owner(pilot_user_id=uid)
853 # "Replace" on the profile's documents list: the document being
854 # replaced, to preselect its type and say what happens to it.
855 replacing = None
856 replace_id = request.values.get("replace", type=int)
857 if replace_id:
858 candidate = db.session.get(Document, replace_id)
859 if candidate is not None and candidate.pilot_user_id == uid:
860 replacing = candidate
862 def _form(status: int = 200) -> ResponseReturnValue:
863 return (
864 render_template(
865 "documents/pilot_upload_form.html",
866 doc_types=_PILOT_DOC_TYPES,
867 replacing=replacing,
868 selected_type=request.values.get("doc_type")
869 or (replacing.doc_type if replacing else ""),
870 ),
871 status,
872 )
874 if request.method == "POST":
875 file = request.files.get("file")
876 title = request.form.get("title", "").strip() or None
877 doc_type = request.form.get("doc_type") or None
878 if doc_type not in dict(_PILOT_DOC_TYPES):
879 doc_type = None
880 valid_from = _parse_date_field(request.form.get("valid_from", ""))
881 valid_until = _parse_date_field(request.form.get("valid_until", ""))
883 if not file or not file.filename:
884 flash(_("Please select a file to upload."), "danger")
885 return _form()
887 original = secure_filename(file.filename)
888 ext = os.path.splitext(original)[1].lower()
889 if ext not in _ALLOWED_EXTS:
890 flash(
891 _("File type '%(ext)s' is not allowed.", ext=ext or "unknown"), "danger"
892 )
893 return _form()
894 if valid_from and valid_until and valid_from > valid_until:
895 flash(_("'Valid from' must not be after 'Valid until'."), "danger")
896 return _form()
898 stored, mime, size = _save_upload(file, f"pilot{uid}")
900 doc = Document(
901 pilot_user_id=uid,
902 filename=stored,
903 original_filename=original,
904 mime_type=mime,
905 size_bytes=size,
906 title=title,
907 doc_type=doc_type,
908 valid_from=valid_from,
909 valid_until=valid_until,
910 is_sensitive=True,
911 )
912 db.session.add(doc)
913 # A medical/licence: work out again which one is in force (this one
914 # may supersede the current one now, or only on its valid_from) and
915 # sync the profile's medical expiry.
916 lifecycle.refresh(owner, doc_type)
917 db.session.commit()
919 flash(_("Document uploaded."), "success")
920 return redirect(url_for("pilots.profile"))
922 return _form()
925# ── Pilot document delete ─────────────────────────────────────────────────────
928@documents_bp.route("/pilot/documents/<int:document_id>/delete", methods=["POST"])
929@login_required
930def delete_pilot_document(document_id: int) -> ResponseReturnValue:
931 uid = int(session["user_id"])
932 doc = db.session.get(Document, document_id)
933 if not doc or doc.pilot_user_id != uid:
934 abort(404)
935 doc_type = doc.doc_type
936 _delete_file(doc.filename)
937 db.session.delete(doc)
938 # Deleting the one in force puts the one it had superseded back in force.
939 lifecycle.refresh(lifecycle.Owner(pilot_user_id=uid), doc_type)
940 db.session.commit()
941 flash(_("Document deleted."), "success")
942 return redirect(url_for("pilots.profile"))
945# ── Reconcile: scan + list + import + ignore ──────────────────────────────────
948@documents_bp.route("/documents/reconcile")
949@login_required
950@require_role(*_OWNER_ROLES)
951def list_reconcile() -> ResponseReturnValue:
952 import difflib
954 tenant = _get_tenant()
955 pending = (
956 PendingReconcile.query.filter_by(
957 tenant_id=tenant.id, reconciled_at=None, ignored=False
958 )
959 .order_by(PendingReconcile.detected_at.desc())
960 .all()
961 )
962 aircraft_list = (
963 Aircraft.query.filter_by(tenant_id=tenant.id)
964 .order_by(Aircraft.registration)
965 .all()
966 )
968 # For entries with an unrecognised category folder, suggest the closest match
969 category_suggestions: dict[
970 int, tuple[str, str]
971 ] = {} # pr.id → (raw_folder, suggestion)
972 for pr in pending:
973 parts = pr.filepath.replace("\\", "/").split("/")
974 if len(parts) >= 4:
975 raw = parts[2]
976 if raw.lower() not in DocCategory.ALL:
977 close = difflib.get_close_matches(
978 raw.lower(), DocCategory.ALL, n=1, cutoff=0.6
979 )
980 if close:
981 # folder path up to and including the bad category dir
982 bad_folder = "/".join(parts[:3])
983 category_suggestions[pr.id] = (bad_folder, close[0])
985 return render_template(
986 "documents/reconcile.html",
987 tenant=tenant,
988 pending=pending,
989 aircraft_list=aircraft_list,
990 categories=list(_CATEGORY_LABELS.items()),
991 category_suggestions=category_suggestions,
992 )
995@documents_bp.route("/documents/reconcile/scan", methods=["POST"])
996@login_required
997@require_role(*_OWNER_ROLES)
998def scan_documents() -> ResponseReturnValue:
999 tenant = _get_tenant()
1000 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
1002 if not tenant.slug:
1003 flash(_("Set a Hangar ID in Settings before scanning for files."), "warning")
1004 return redirect(url_for("documents.list_reconcile"))
1006 slug_dir = os.path.join(folder, tenant.slug)
1007 if not os.path.isdir(slug_dir):
1008 flash(
1009 _(
1010 "No files found in '%(slug)s/'. Mount your Syncthing folder and try again.",
1011 slug=tenant.slug,
1012 ),
1013 "info",
1014 )
1015 return redirect(url_for("documents.list_reconcile"))
1017 # Build set of filenames already tracked in the documents table
1018 tid = tenant.id
1019 known: set[str] = {
1020 doc.filename
1021 for doc in Document.query.filter(
1022 Document.aircraft_id.in_(
1023 Aircraft.query.filter_by(tenant_id=tid).with_entities(Aircraft.id)
1024 )
1025 ).all()
1026 }
1027 # Prune stale pending entries whose file no longer exists on disk
1028 stale_removed = 0
1029 for pr in PendingReconcile.query.filter_by(
1030 tenant_id=tid, reconciled_at=None, ignored=False
1031 ).all():
1032 if not os.path.exists(os.path.join(folder, pr.filepath)):
1033 db.session.delete(pr)
1034 stale_removed += 1
1035 if stale_removed:
1036 db.session.flush()
1038 existing_pending: set[str] = {
1039 pr.filepath for pr in PendingReconcile.query.filter_by(tenant_id=tid).all()
1040 }
1042 aircraft_by_reg: dict[str, Aircraft] = {
1043 ac.registration.upper().replace("-", "").replace(" ", ""): ac
1044 for ac in Aircraft.query.filter_by(tenant_id=tid).all()
1045 }
1047 new_count = 0
1048 for dirpath, _dirs, filenames in os.walk(slug_dir):
1049 for fname in filenames:
1050 if fname.startswith((".", "_")):
1051 continue
1052 full = os.path.join(dirpath, fname)
1053 relpath = os.path.relpath(full, folder)
1054 # Normalise to forward slashes for DB consistency
1055 relpath = relpath.replace("\\", "/")
1057 if relpath in known or relpath in existing_pending:
1058 continue
1060 # Parse canonical path: slug/reg/category/YYYY-MM-DD - title.ext
1061 parts = relpath.split("/")
1062 if is_app_managed(relpath):
1063 continue
1064 aircraft_obj: Aircraft | None = None
1065 category: str | None = None
1066 title_hint: str | None = None
1067 date_hint: _date | None = None
1069 if len(parts) >= 4:
1070 reg_raw = parts[1].upper().replace("-", "").replace(" ", "")
1071 aircraft_obj = aircraft_by_reg.get(reg_raw)
1072 category, title_hint, date_hint = _parse_reconcile_filename(
1073 parts[2], parts[3]
1074 )
1076 pr = PendingReconcile(
1077 tenant_id=tid,
1078 aircraft_id=aircraft_obj.id if aircraft_obj else None,
1079 filepath=relpath,
1080 category=category,
1081 title_hint=title_hint,
1082 date_hint=date_hint,
1083 )
1084 db.session.add(pr)
1085 new_count += 1
1087 db.session.commit()
1088 parts_msg = []
1089 if new_count:
1090 parts_msg.append(
1091 ngettext(
1092 "one new file queued for review",
1093 "%(n)s new files queued for review",
1094 new_count,
1095 n=new_count,
1096 )
1097 )
1098 if stale_removed:
1099 parts_msg.append(
1100 ngettext(
1101 "one missing file removed from queue",
1102 "%(n)s missing files removed from queue",
1103 stale_removed,
1104 n=stale_removed,
1105 )
1106 )
1107 if parts_msg:
1108 flash(
1109 _("Scan complete — %(details)s.", details=", ".join(parts_msg)), "success"
1110 )
1111 else:
1112 flash(_("Scan complete — no new files found."), "info")
1113 return redirect(url_for("documents.list_reconcile"))
1116@documents_bp.route("/documents/reconcile/rename-folder", methods=["POST"])
1117@login_required
1118@require_role(*_OWNER_ROLES)
1119def rename_reconcile_folder() -> ResponseReturnValue:
1120 """Rename a misnamed category folder on disk (e.g. 'Maintenance' → 'maintenance',
1121 or a typo like 'maintenence' → 'maintenance'), prune stale pending entries for
1122 the old path, then run a fresh scan so the corrected files are picked up immediately.
1123 """
1124 import shutil
1126 tenant = _get_tenant()
1127 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
1129 bad_folder = request.form.get("bad_folder", "").strip().replace("\\", "/")
1130 new_category = request.form.get("new_category", "").strip()
1132 if new_category not in DocCategory.ALL:
1133 flash(_("Invalid category."), "danger")
1134 return redirect(url_for("documents.list_reconcile"))
1136 if not tenant.slug or not bad_folder.startswith(tenant.slug + "/"):
1137 abort(403)
1139 old_dir = _safe_join(folder, bad_folder)
1140 parent_rel = "/".join(bad_folder.split("/")[:2]) # slug/reg
1141 new_rel = parent_rel + "/" + new_category
1142 new_dir = _safe_join(folder, new_rel)
1144 if os.path.isdir(old_dir):
1145 if os.path.isdir(new_dir):
1146 for dirpath, _dirs, filenames in os.walk(old_dir):
1147 rel = os.path.relpath(dirpath, old_dir)
1148 dest_d = os.path.join(new_dir, rel)
1149 os.makedirs(dest_d, exist_ok=True)
1150 for fname in filenames:
1151 shutil.move(
1152 os.path.join(dirpath, fname), os.path.join(dest_d, fname)
1153 )
1154 shutil.rmtree(old_dir, ignore_errors=True)
1155 else:
1156 os.rename(old_dir, new_dir)
1158 # Prune stale pending entries for the old folder path
1159 tid = tenant.id
1160 for pr in PendingReconcile.query.filter(
1161 PendingReconcile.tenant_id == tid,
1162 PendingReconcile.filepath.like(bad_folder + "/%"),
1163 ).all():
1164 db.session.delete(pr)
1165 db.session.flush()
1167 # Inline scan: pick up the files now in the correct folder
1168 known: set[str] = {
1169 doc.filename
1170 for doc in Document.query.filter(
1171 Document.aircraft_id.in_(
1172 Aircraft.query.filter_by(tenant_id=tid).with_entities(Aircraft.id)
1173 )
1174 ).all()
1175 }
1176 existing_pending: set[str] = {
1177 pr.filepath for pr in PendingReconcile.query.filter_by(tenant_id=tid).all()
1178 }
1179 aircraft_by_reg: dict[str, Aircraft] = {
1180 ac.registration.upper().replace("-", "").replace(" ", ""): ac
1181 for ac in Aircraft.query.filter_by(tenant_id=tid).all()
1182 }
1183 new_count = 0
1184 if os.path.isdir(new_dir):
1185 for dirpath, _dirs, filenames in os.walk(new_dir):
1186 for fname in filenames:
1187 if fname.startswith((".", "_")):
1188 continue
1189 relpath = os.path.relpath(os.path.join(dirpath, fname), folder).replace(
1190 "\\", "/"
1191 )
1192 full = _safe_join(folder, relpath)
1193 if relpath in known or relpath in existing_pending:
1194 continue
1195 parts = relpath.split("/")
1196 aircraft_obj: Aircraft | None = None
1197 category: str | None = None
1198 title_hint: str | None = None
1199 date_hint: _date | None = None
1200 if len(parts) >= 4:
1201 reg_raw = parts[1].upper().replace("-", "").replace(" ", "")
1202 aircraft_obj = aircraft_by_reg.get(reg_raw)
1203 category, title_hint, date_hint = _parse_reconcile_filename(
1204 parts[2], parts[3]
1205 )
1206 if aircraft_obj and category:
1207 mime = mimetypes.guess_type(fname)[0] or "application/octet-stream"
1208 size = os.path.getsize(full) if os.path.exists(full) else None
1209 doc = Document(
1210 aircraft_id=aircraft_obj.id,
1211 filename=relpath,
1212 original_filename=fname,
1213 mime_type=mime,
1214 size_bytes=size,
1215 title=title_hint,
1216 category=category,
1217 )
1218 db.session.add(doc)
1219 else:
1220 pr = PendingReconcile(
1221 tenant_id=tid,
1222 aircraft_id=aircraft_obj.id if aircraft_obj else None,
1223 filepath=relpath,
1224 category=category,
1225 title_hint=title_hint,
1226 date_hint=date_hint,
1227 )
1228 db.session.add(pr)
1229 new_count += 1
1231 db.session.commit()
1232 flash(
1233 ngettext(
1234 "Folder renamed to '%(cat)s' — one file processed.",
1235 "Folder renamed to '%(cat)s' — %(n)s files processed.",
1236 new_count,
1237 cat=new_category,
1238 n=new_count,
1239 ),
1240 "success",
1241 )
1242 return redirect(url_for("documents.list_reconcile"))
1245@documents_bp.route("/documents/reconcile/<int:pending_id>/import", methods=["POST"])
1246@login_required
1247@require_role(*_OWNER_ROLES)
1248def import_reconcile(pending_id: int) -> ResponseReturnValue:
1249 tenant = _get_tenant()
1250 pr = PendingReconcile.query.filter_by(
1251 id=pending_id, tenant_id=tenant.id
1252 ).first_or_404()
1254 aircraft_id_raw = request.form.get("aircraft_id")
1255 try:
1256 aircraft_id: int | None = int(aircraft_id_raw) if aircraft_id_raw else None
1257 except (ValueError, TypeError):
1258 aircraft_id = None
1259 if aircraft_id is not None:
1260 _get_aircraft_or_404(aircraft_id)
1262 title = request.form.get("title", "").strip() or pr.title_hint
1263 category = request.form.get("category") or pr.category
1264 if category and category not in DocCategory.ALL:
1265 category = None
1266 valid_until_str = request.form.get("valid_until", "").strip()
1267 valid_until: _date | None = None
1268 if valid_until_str:
1269 with contextlib.suppress(
1270 ValueError
1271 ): # malformed date submitted; valid_until stays None
1272 valid_until = _date.fromisoformat(valid_until_str)
1274 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
1275 full_path = os.path.join(folder, pr.filepath)
1276 mime = (
1277 mimetypes.guess_type(os.path.basename(pr.filepath))[0]
1278 or "application/octet-stream"
1279 )
1280 size = os.path.getsize(full_path) if os.path.exists(full_path) else None
1282 doc = Document(
1283 aircraft_id=aircraft_id,
1284 filename=pr.filepath,
1285 original_filename=os.path.basename(pr.filepath),
1286 mime_type=mime,
1287 size_bytes=size,
1288 title=title,
1289 category=category,
1290 valid_until=valid_until,
1291 is_sensitive=False,
1292 )
1293 db.session.add(doc)
1294 pr.reconciled_at = datetime.now(UTC)
1295 db.session.commit()
1297 flash(_("Document imported."), "success")
1298 return redirect(url_for("documents.list_reconcile"))
1301@documents_bp.route("/documents/reconcile/<int:pending_id>/ignore", methods=["POST"])
1302@login_required
1303@require_role(*_OWNER_ROLES)
1304def ignore_reconcile(pending_id: int) -> ResponseReturnValue:
1305 tenant = _get_tenant()
1306 pr = PendingReconcile.query.filter_by(
1307 id=pending_id, tenant_id=tenant.id
1308 ).first_or_404()
1309 pr.ignored = True
1310 db.session.commit()
1311 flash(_("File ignored."), "info")
1312 return redirect(url_for("documents.list_reconcile"))