Coverage for app/pilots/routes.py: 100%
767 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-06-28 23:33 +0000
1import contextlib
2import json
3import logging
4import os
5import uuid
6from typing import Any
7from datetime import (
8 date as _date,
9 datetime as _datetime,
10 time as _time,
11 timedelta as _td,
12 timezone as _tz,
13)
15from sqlalchemy import func # pyright: ignore[reportMissingImports]
16from flask import ( # pyright: ignore[reportMissingImports]
17 Blueprint,
18 abort,
19 current_app,
20 flash,
21 redirect,
22 render_template,
23 request,
24 session,
25 url_for,
26)
27from flask.typing import ResponseReturnValue # pyright: ignore[reportMissingImports]
29from flask_babel import gettext as _, ngettext # pyright: ignore[reportMissingImports]
30from werkzeug.utils import secure_filename # pyright: ignore[reportMissingImports]
32from models import ( # pyright: ignore[reportMissingImports]
33 Aircraft,
34 Document,
35 FlightCrew,
36 FlightEntry,
37 GpsTrack,
38 LogbookImportBatch,
39 LogbookImportMapping,
40 PilotLogbookEntry,
41 PilotProfile,
42 TenantUser,
43 db,
44)
45from utils import login_required, require_pilot_access # pyright: ignore[reportMissingImports]
46from pilots.logbook_import import ( # pyright: ignore[reportMissingImports]
47 TARGET_FIELDS,
48 _norm,
49 execute_import,
50 parse_duration_value,
51 parse_file,
52 preview_rows,
53 propose_mapping,
54 type_hints,
55)
57log = logging.getLogger(__name__)
59pilots_bp = Blueprint("pilots", __name__)
62def _current_user_id() -> int:
63 return int(session["user_id"])
66def _openaip_key() -> str | None:
67 from models import AppSetting # pyright: ignore[reportMissingImports]
69 s = db.session.get(AppSetting, "openaip_api_key")
70 return s.value if s and s.value else None
73def _get_or_create_profile(user_id: int) -> PilotProfile:
74 profile: PilotProfile | None = PilotProfile.query.filter_by(user_id=user_id).first()
75 if not profile:
76 profile = PilotProfile(user_id=user_id)
77 db.session.add(profile)
78 db.session.flush()
79 return profile
82def _check_logbook_milestone(entry: PilotLogbookEntry, uid: int) -> None:
83 """Set one-shot session flags when a logbook milestone is crossed."""
84 total = PilotLogbookEntry.query.filter_by(pilot_user_id=uid).count()
85 if total == 100:
86 session["logbook_milestone"] = "100flights"
87 flash(_("🎉 100th logbook entry — congratulations!"), "success")
88 return
90 night = float(entry.night_time or 0)
91 if night > 0:
92 prev_night = (
93 db.session.query(func.count(PilotLogbookEntry.id))
94 .filter(
95 PilotLogbookEntry.pilot_user_id == uid,
96 PilotLogbookEntry.id != entry.id,
97 PilotLogbookEntry.night_time > 0,
98 )
99 .scalar()
100 or 0
101 )
102 if prev_night == 0:
103 session["logbook_milestone"] = "first_night"
104 flash(_("🌙 First night flight logged — well done!"), "success")
105 return
107 dep = (entry.departure_place or "").strip().upper()
108 arr = (entry.arrival_place or "").strip().upper()
109 if dep and arr and dep != arr:
110 prev_xc = (
111 db.session.query(func.count(PilotLogbookEntry.id))
112 .filter(
113 PilotLogbookEntry.pilot_user_id == uid,
114 PilotLogbookEntry.id != entry.id,
115 PilotLogbookEntry.departure_place.isnot(None),
116 PilotLogbookEntry.arrival_place.isnot(None),
117 PilotLogbookEntry.departure_place != PilotLogbookEntry.arrival_place,
118 )
119 .scalar()
120 or 0
121 )
122 if prev_xc == 0:
123 session["logbook_milestone"] = "first_xc"
124 flash(
125 _("✈️ First cross-country flight logged — congratulations!"), "success"
126 )
129def _parse_time(val: str, field: str) -> tuple[_time | None, str | None]:
130 val = val.strip()
131 if not val:
132 return None, None
133 try:
134 h, m = val.split(":")
135 t = _time(int(h), int(m))
136 return t, None
137 except (ValueError, AttributeError):
138 return None, _("%(field)s: enter a valid HH:MM time.", field=field)
141def _parse_decimal(val: str, field: str) -> tuple[float | None, str | None]:
142 val = val.strip()
143 if not val:
144 return None, None
145 try:
146 n = float(val)
147 if n < 0:
148 return None, _("%(field)s: must be non-negative.", field=field)
149 return n, None
150 except ValueError:
151 return None, _("%(field)s: must be a number.", field=field)
154def _parse_int(val: str, field: str) -> tuple[int | None, str | None]:
155 val = val.strip()
156 if not val:
157 return None, None
158 try:
159 n = int(val)
160 if n < 0:
161 return None, _("%(field)s: must be non-negative.", field=field)
162 return n, None
163 except ValueError:
164 return None, _("%(field)s: must be a whole number.", field=field)
167def _parse_date(val: str, field: str) -> tuple[_date | None, str | None]:
168 val = val.strip()
169 if not val:
170 return None, None
171 try:
172 return _date.fromisoformat(val), None
173 except ValueError:
174 return None, _("%(field)s: enter a valid date (YYYY-MM-DD).", field=field)
177# ── Profile ───────────────────────────────────────────────────────────────────
180@pilots_bp.route("/pilot/profile", methods=["GET", "POST"])
181@login_required
182@require_pilot_access
183def profile() -> ResponseReturnValue:
184 uid = _current_user_id()
185 p = _get_or_create_profile(uid)
187 if request.method == "POST":
188 errors = []
190 p.license_number = request.form.get("license_number", "").strip() or None
192 medical_str = request.form.get("medical_expiry", "")
193 medical, err = _parse_date(medical_str, "Medical expiry")
194 if err:
195 errors.append(err)
196 else:
197 p.medical_expiry = medical
199 sep_str = request.form.get("sep_expiry", "")
200 sep, err = _parse_date(sep_str, "SEP expiry")
201 if err:
202 errors.append(err)
203 else:
204 p.sep_expiry = sep
206 solo_str = request.form.get("first_solo_date", "")
207 solo, err = _parse_date(solo_str, _("First solo date"))
208 if err:
209 errors.append(err)
210 else:
211 p.first_solo_date = solo
213 ppl_str = request.form.get("ppl_issue_date", "")
214 ppl, err = _parse_date(ppl_str, _("PPL issue date"))
215 if err:
216 errors.append(err)
217 else:
218 p.ppl_issue_date = ppl
220 if errors:
221 for e in errors:
222 flash(e, "danger")
223 return (
224 render_template(
225 "pilots/profile.html", profile=p, pilot_docs=[], currency=None
226 ),
227 422,
228 )
230 db.session.commit()
231 flash(_("Profile saved."), "success")
232 return redirect(url_for("pilots.profile"))
234 from pilots.currency import currency_summary as _currency_summary # pyright: ignore[reportMissingImports]
236 pilot_entries = PilotLogbookEntry.query.filter_by(pilot_user_id=uid).all()
237 currency = _currency_summary(p, pilot_entries)
239 pilot_docs = (
240 Document.query.filter_by(pilot_user_id=uid)
241 .order_by(Document.uploaded_at.desc())
242 .all()
243 )
244 return render_template(
245 "pilots/profile.html", profile=p, pilot_docs=pilot_docs, currency=currency
246 )
249_VALID_PER_PAGE = (10, 20, 50, 100)
250_DEFAULT_PER_PAGE = 20
253# ── GPS tracks map ────────────────────────────────────────────────────────────
256@pilots_bp.route("/pilot/tracks")
257@login_required
258@require_pilot_access
259def pilot_tracks() -> ResponseReturnValue:
260 from flask import url_for as _url_for
262 uid = _current_user_id()
263 entries = (
264 PilotLogbookEntry.query.filter_by(pilot_user_id=uid)
265 .filter(PilotLogbookEntry.gps_track_id.isnot(None))
266 .order_by(PilotLogbookEntry.date.asc())
267 .all()
268 )
269 track_rows = [
270 {
271 "date": str(e.date),
272 "dep": e.departure_place or "",
273 "arr": e.arrival_place or "",
274 "time_str": f"{e.total_flight_time} h"
275 if e.total_flight_time is not None
276 else "",
277 "view_url": _url_for(
278 "aircraft.flight_detail",
279 aircraft_id=e.flight.aircraft_id,
280 flight_id=e.flight_id,
281 )
282 if e.flight_id and e.flight
283 else _url_for("pilots.view_entry", entry_id=e.id),
284 "geojson": e.gps_track.geojson if e.gps_track else None,
285 }
286 for e in entries
287 ]
289 return render_template(
290 "pilots/flight_tracks.html",
291 track_rows=track_rows,
292 openaip_key=_openaip_key(),
293 )
296@pilots_bp.route("/pilot/tracks/animation.gif")
297@login_required
298@require_pilot_access
299def pilot_tracks_gif() -> ResponseReturnValue:
300 from utils import generate_tracks_gif, sort_tracks_oldest_first # pyright: ignore[reportMissingImports]
301 from flask import Response # pyright: ignore[reportMissingImports]
303 uid = _current_user_id()
304 entries = (
305 PilotLogbookEntry.query.filter_by(pilot_user_id=uid)
306 .filter(PilotLogbookEntry.gps_track_id.isnot(None))
307 .all()
308 )
309 track_rows = sort_tracks_oldest_first(
310 [
311 {
312 "date": str(e.date),
313 "dep": e.departure_place or "",
314 "arr": e.arrival_place or "",
315 "geojson": e.gps_track.geojson if e.gps_track else None,
316 }
317 for e in entries
318 ]
319 )
320 portrait = request.args.get("orientation") == "portrait"
321 hires = request.args.get("quality") == "hires"
322 base_w, base_h = (480, 800) if portrait else (800, 480)
323 mul = 2 if hires else 1
324 canvas_w, canvas_h = base_w * mul, base_h * mul
325 gif_bytes = generate_tracks_gif(
326 track_rows,
327 _openaip_key=_openaip_key(),
328 canvas_w=canvas_w,
329 canvas_h=canvas_h,
330 high_res=hires,
331 )
332 orient_sfx = "-portrait" if portrait else ""
333 qual_sfx = "-hires" if hires else ""
334 suffix = orient_sfx + qual_sfx
335 return Response(
336 gif_bytes,
337 mimetype="image/gif",
338 headers={
339 "Content-Disposition": f'attachment; filename="my_tracks{suffix}.gif"'
340 },
341 )
344# ── Logbook entry detail (read-only) ─────────────────────────────────────────
347@pilots_bp.route("/pilot/logbook/<int:entry_id>/view")
348@login_required
349@require_pilot_access
350def view_entry(entry_id: int) -> ResponseReturnValue:
351 uid = _current_user_id()
352 entry = db.session.get(PilotLogbookEntry, entry_id)
353 if not entry or entry.pilot_user_id != uid:
354 abort(404)
356 return render_template(
357 "pilots/entry_detail.html",
358 entry=entry,
359 openaip_key=_openaip_key(),
360 )
363# ── Logbook list ──────────────────────────────────────────────────────────────
366@pilots_bp.route("/pilot/logbook")
367@login_required
368@require_pilot_access
369def logbook() -> ResponseReturnValue:
370 uid = _current_user_id()
371 order = request.args.get("order", "desc")
372 page = request.args.get("page", 1, type=int)
373 pp_raw = request.args.get("per_page", str(_DEFAULT_PER_PAGE))
374 show_all = pp_raw == "all"
375 per_page = (
376 None
377 if show_all
378 else (
379 int(pp_raw)
380 if pp_raw.isdigit() and int(pp_raw) in _VALID_PER_PAGE
381 else _DEFAULT_PER_PAGE
382 )
383 )
385 q = PilotLogbookEntry.query.filter_by(pilot_user_id=uid)
386 if order == "asc":
387 q = q.order_by(PilotLogbookEntry.date.asc(), PilotLogbookEntry.id.asc())
388 else:
389 q = q.order_by(PilotLogbookEntry.date.desc(), PilotLogbookEntry.id.desc())
391 if show_all:
392 entries = q.all()
393 pagination = None
394 else:
395 pagination = q.paginate(page=page, per_page=per_page, error_out=False)
396 entries = pagination.items
398 totals = _compute_totals_sql(uid)
399 logbook_milestone = session.pop("logbook_milestone", None)
401 return render_template(
402 "pilots/logbook.html",
403 entries=entries,
404 pagination=pagination,
405 totals=totals,
406 order=order,
407 per_page=pp_raw,
408 valid_per_page=_VALID_PER_PAGE,
409 logbook_milestone=logbook_milestone,
410 )
413def _compute_totals_sql(pilot_user_id: int) -> dict[str, object]:
414 """Aggregate totals over ALL entries for the pilot via a single SQL query."""
415 row = (
416 db.session.query(
417 func.sum(PilotLogbookEntry.night_time),
418 func.sum(PilotLogbookEntry.instrument_time),
419 func.sum(PilotLogbookEntry.landings_day),
420 func.sum(PilotLogbookEntry.landings_night),
421 func.sum(PilotLogbookEntry.single_pilot_se),
422 func.sum(PilotLogbookEntry.single_pilot_me),
423 func.sum(PilotLogbookEntry.multi_pilot),
424 func.sum(PilotLogbookEntry.function_pic),
425 func.sum(PilotLogbookEntry.function_copilot),
426 func.sum(PilotLogbookEntry.function_dual),
427 func.sum(PilotLogbookEntry.function_instructor),
428 )
429 .filter(PilotLogbookEntry.pilot_user_id == pilot_user_id)
430 .one()
431 )
433 sp_se = round(float(row[4] or 0), 1)
434 sp_me = round(float(row[5] or 0), 1)
435 multi = round(float(row[6] or 0), 1)
437 return {
438 "night_time": round(float(row[0] or 0), 1),
439 "instrument_time": round(float(row[1] or 0), 1),
440 "landings_day": int(row[2] or 0),
441 "landings_night": int(row[3] or 0),
442 "single_pilot_se": sp_se,
443 "single_pilot_me": sp_me,
444 "multi_pilot": multi,
445 "total_flight_time": round(sp_se + sp_me + multi, 1),
446 "function_pic": round(float(row[7] or 0), 1),
447 "function_copilot": round(float(row[8] or 0), 1),
448 "function_dual": round(float(row[9] or 0), 1),
449 "function_instructor": round(float(row[10] or 0), 1),
450 }
453# ── New entry ────────────────────────────────────────────────────────────────
456@pilots_bp.route("/pilot/logbook/new", methods=["GET", "POST"])
457@login_required
458@require_pilot_access
459def new_entry() -> ResponseReturnValue:
460 uid = _current_user_id()
462 if request.method == "POST":
463 entry, errors = _entry_from_form(uid)
464 if errors:
465 for e in errors:
466 flash(e, "danger")
467 return render_template(
468 "pilots/entry_form.html",
469 entry=None,
470 form=request.form,
471 action="new",
472 openaip_key=_openaip_key(),
473 ), 422
474 db.session.add(entry)
475 db.session.flush()
476 _apply_gps_to_pilot_entry(entry)
477 db.session.commit()
478 _check_logbook_milestone(entry, uid)
479 flash(_("Logbook entry saved."), "success")
480 return redirect(url_for("pilots.logbook"))
482 return render_template(
483 "pilots/entry_form.html",
484 entry=None,
485 form={},
486 action="new",
487 openaip_key=_openaip_key(),
488 )
491# ── Edit entry ────────────────────────────────────────────────────────────────
494@pilots_bp.route("/pilot/logbook/<int:entry_id>/edit", methods=["GET", "POST"])
495@login_required
496@require_pilot_access
497def edit_entry(entry_id: int) -> ResponseReturnValue:
498 uid = _current_user_id()
499 entry = db.session.get(PilotLogbookEntry, entry_id)
500 if not entry or entry.pilot_user_id != uid:
501 abort(404)
503 if entry.flight_id:
504 return redirect(url_for("flights.edit_flight", flight_id=entry.flight_id))
506 if request.method == "POST":
507 updated, errors = _entry_from_form(uid)
508 if errors:
509 for e in errors:
510 flash(e, "danger")
511 return render_template(
512 "pilots/entry_form.html",
513 entry=entry,
514 form=request.form,
515 action="edit",
516 openaip_key=_openaip_key(),
517 ), 422
518 for col in PilotLogbookEntry.__table__.columns:
519 if col.name not in ("id", "pilot_user_id", "gps_track_id"):
520 setattr(entry, col.name, getattr(updated, col.name))
521 _apply_gps_to_pilot_entry(entry)
522 db.session.commit()
523 flash(_("Logbook entry updated."), "success")
524 return redirect(url_for("pilots.logbook"))
526 return render_template(
527 "pilots/entry_form.html",
528 entry=entry,
529 form={},
530 action="edit",
531 openaip_key=_openaip_key(),
532 )
535# ── Delete entry ──────────────────────────────────────────────────────────────
538@pilots_bp.route("/pilot/logbook/<int:entry_id>/delete", methods=["POST"])
539@login_required
540@require_pilot_access
541def delete_entry(entry_id: int) -> ResponseReturnValue:
542 uid = _current_user_id()
543 entry = db.session.get(PilotLogbookEntry, entry_id)
544 if not entry or entry.pilot_user_id != uid:
545 abort(404)
546 db.session.delete(entry)
547 db.session.commit()
548 flash(_("Logbook entry deleted."), "success")
549 return redirect(url_for("pilots.logbook"))
552# ── GPS track helper ──────────────────────────────────────────────────────────
555def _apply_gps_to_pilot_entry(entry: PilotLogbookEntry) -> None:
556 """Create or update the GpsTrack linked to a pilot logbook entry from form data."""
557 f = request.form
558 gps_geojson_raw = f.get("gps_geojson", "").strip()
559 gps_filename = f.get("gps_filename", "").strip() or None
560 if not gps_geojson_raw and not gps_filename:
561 return
563 geojson = None
564 if gps_geojson_raw:
565 with contextlib.suppress(
566 ValueError
567 ): # malformed hidden field — GPS track simply not applied
568 geojson = json.loads(gps_geojson_raw)
570 def _parse_dt(raw: str) -> "_datetime | None":
571 try:
572 return _datetime.fromisoformat(raw) if raw else None
573 except ValueError:
574 return None
576 block_off = _parse_dt(f.get("gps_block_off_utc", "").strip())
577 block_on = _parse_dt(f.get("gps_block_on_utc", "").strip())
578 dep = f.get("departure_place", "").strip().upper()[:4] or None
579 arr = f.get("arrival_place", "").strip().upper()[:4] or None
581 if entry.gps_track_id:
582 gt = db.session.get(GpsTrack, entry.gps_track_id)
583 if gt:
584 if geojson is not None:
585 gt.geojson = geojson
586 if gps_filename:
587 gt.source_filename = gps_filename
588 if block_off:
589 gt.block_off_utc = block_off
590 if block_on:
591 gt.block_on_utc = block_on
592 return
594 gt = GpsTrack(
595 source_filename=gps_filename,
596 block_off_utc=block_off,
597 block_on_utc=block_on,
598 departure_icao=dep,
599 arrival_icao=arr,
600 geojson=geojson,
601 )
602 db.session.add(gt)
603 db.session.flush()
604 entry.gps_track_id = gt.id
607# ── Form parsing ──────────────────────────────────────────────────────────────
610def _entry_from_form(pilot_user_id: int) -> tuple[PilotLogbookEntry, list[str]]:
611 f = request.form
612 errors = []
614 date_str = f.get("date", "")
615 date_val, err = _parse_date(date_str, "Date")
616 if err:
617 errors.append(err)
618 elif date_val is None:
619 errors.append(_("Date is required."))
621 dep_time, err = _parse_time(f.get("departure_time", ""), "Departure time")
622 if err:
623 errors.append(err)
624 arr_time, err = _parse_time(f.get("arrival_time", ""), "Arrival time")
625 if err:
626 errors.append(err)
628 night_time, err = _parse_decimal(f.get("night_time", ""), "Night time")
629 if err:
630 errors.append(err)
631 instrument_time, err = _parse_decimal(
632 f.get("instrument_time", ""), "Instrument time"
633 )
634 if err:
635 errors.append(err)
636 landings_day, err = _parse_int(f.get("landings_day", ""), "Day landings")
637 if err:
638 errors.append(err)
639 landings_night, err = _parse_int(f.get("landings_night", ""), "Night landings")
640 if err:
641 errors.append(err)
642 sp_se, err = _parse_decimal(f.get("single_pilot_se", ""), "S/E time")
643 if err:
644 errors.append(err)
645 sp_me, err = _parse_decimal(f.get("single_pilot_me", ""), "M/E time")
646 if err:
647 errors.append(err)
648 multi_pilot, err = _parse_decimal(f.get("multi_pilot", ""), "Multi-pilot time")
649 if err:
650 errors.append(err)
651 fn_pic, err = _parse_decimal(f.get("function_pic", ""), "PIC function")
652 if err:
653 errors.append(err)
654 fn_co, err = _parse_decimal(f.get("function_copilot", ""), "Co-pilot function")
655 if err:
656 errors.append(err)
657 fn_dual, err = _parse_decimal(f.get("function_dual", ""), "Dual function")
658 if err:
659 errors.append(err)
660 fn_inst, err = _parse_decimal(
661 f.get("function_instructor", ""), "Instructor function"
662 )
663 if err:
664 errors.append(err)
666 entry = PilotLogbookEntry(
667 pilot_user_id=pilot_user_id,
668 date=date_val,
669 aircraft_type=f.get("aircraft_type", "").strip() or None,
670 aircraft_type_icao=f.get("aircraft_type_icao", "").strip() or None,
671 aircraft_registration=f.get("aircraft_registration", "").strip() or None,
672 departure_place=f.get("departure_place", "").strip() or None,
673 departure_time=dep_time,
674 arrival_place=f.get("arrival_place", "").strip() or None,
675 arrival_time=arr_time,
676 pic_name=f.get("pic_name", "").strip() or None,
677 night_time=night_time,
678 instrument_time=instrument_time,
679 landings_day=landings_day,
680 landings_night=landings_night,
681 single_pilot_se=sp_se,
682 single_pilot_me=sp_me,
683 multi_pilot=multi_pilot,
684 function_pic=fn_pic,
685 function_copilot=fn_co,
686 function_dual=fn_dual,
687 function_instructor=fn_inst,
688 remarks=f.get("remarks", "").strip() or None,
689 )
690 return entry, errors
693# ── Logbook Import ────────────────────────────────────────────────────────────
695_IMPORT_SESSION_KEY = "logbook_import"
696_ALLOWED_IMPORT_EXTS = {".csv", ".xlsx", ".xls"}
697_MAX_IMPORT_BYTES = 10 * 1024 * 1024 # 10 MB
700def _import_tmp_dir() -> str:
701 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
702 d = os.path.join(folder, "import_tmp")
703 os.makedirs(d, exist_ok=True)
704 return d
707def _cleanup_previous_tmp(uid: int) -> None:
708 """Delete any leftover temp import file for this user."""
709 meta = session.get(_IMPORT_SESSION_KEY)
710 if meta and meta.get("uid") == uid:
711 tmp = meta.get("tmp_path")
712 if tmp and os.path.isfile(tmp):
713 try:
714 os.remove(tmp)
715 except OSError as exc:
716 current_app.logger.debug("cleanup tmp import file: %s", exc)
717 session.pop(_IMPORT_SESSION_KEY, None)
720@pilots_bp.route("/pilot/logbook/import", methods=["GET", "POST"])
721@login_required
722@require_pilot_access
723def import_upload() -> ResponseReturnValue:
724 uid = _current_user_id()
726 if request.method == "GET":
727 return render_template("pilots/import_upload.html")
729 # ── POST: receive file, parse, present mapping page ───────────────────────
730 uploaded = request.files.get("logbook_file")
731 if not uploaded or not uploaded.filename:
732 flash(_("Please select a file to upload."), "danger")
733 return render_template("pilots/import_upload.html"), 422
735 ext = os.path.splitext(uploaded.filename)[1].lower()
736 if ext not in _ALLOWED_IMPORT_EXTS:
737 flash(
738 _("Unsupported format. Please upload a .csv or .xlsx file."),
739 "danger",
740 )
741 return render_template("pilots/import_upload.html"), 422
743 data = uploaded.read()
744 if len(data) > _MAX_IMPORT_BYTES:
745 flash(_("File too large (maximum 10 MB)."), "danger")
746 return render_template("pilots/import_upload.html"), 422
748 try:
749 parsed = parse_file(data, uploaded.filename)
750 except ValueError as exc:
751 flash(str(exc), "danger")
752 return render_template("pilots/import_upload.html"), 422
754 # Save to a temp file so execute step can re-parse without re-upload
755 _cleanup_previous_tmp(uid)
756 safe_base = secure_filename(uploaded.filename) or "upload"
757 tmp_name = f"import_{uid}_{uuid.uuid4().hex}_{safe_base}"
758 tmp_path = os.path.join(_import_tmp_dir(), tmp_name)
759 with open(tmp_path, "wb") as fh:
760 fh.write(data)
762 session[_IMPORT_SESSION_KEY] = {
763 "uid": uid,
764 "tmp_path": tmp_path,
765 "original_filename": uploaded.filename,
766 "norm_cols": parsed.norm_cols,
767 "raw_cols": parsed.raw_cols,
768 "fingerprint": parsed.fingerprint,
769 }
771 # Look up saved mappings for this pilot
772 saved = LogbookImportMapping.query.filter_by(pilot_user_id=uid).all()
773 proposal = propose_mapping(parsed, saved)
775 preview = preview_rows(parsed, proposal.mapping, n=5)
777 return render_template(
778 "pilots/import_map.html",
779 norm_cols=parsed.norm_cols,
780 raw_cols=parsed.raw_cols,
781 base_norm_cols=[_norm(r) for r in parsed.raw_cols],
782 mapping=proposal.mapping,
783 match_type=proposal.match_type,
784 fuzzy_score=proposal.fuzzy_score,
785 target_fields=TARGET_FIELDS,
786 preview=preview,
787 filename=uploaded.filename,
788 type_hints=type_hints(parsed, proposal.mapping),
789 )
792@pilots_bp.route("/pilot/logbook/import/execute", methods=["POST"])
793@login_required
794@require_pilot_access
795def import_execute() -> ResponseReturnValue:
796 uid = _current_user_id()
797 meta = session.get(_IMPORT_SESSION_KEY)
799 if not meta or meta.get("uid") != uid:
800 flash(_("Import session expired. Please upload the file again."), "warning")
801 return redirect(url_for("pilots.import_upload"))
803 tmp_path: str = meta["tmp_path"]
804 original_filename: str = meta["original_filename"]
805 norm_cols: list[str] = meta["norm_cols"]
806 fingerprint: str = meta["fingerprint"]
808 if not os.path.isfile(tmp_path):
809 flash(_("Temporary file not found. Please upload the file again."), "warning")
810 session.pop(_IMPORT_SESSION_KEY, None)
811 return redirect(url_for("pilots.import_upload"))
813 # Reconstruct mapping from form
814 mapping: dict[str, str] = {}
815 for col in norm_cols:
816 val = request.form.get(f"mapping_{col}", "ignore").strip()
817 mapping[col] = val if val in TARGET_FIELDS else "ignore"
819 # Validate: at least 'date' must be mapped
820 if "date" not in mapping.values():
821 flash(_("You must map at least one column to 'Date'."), "danger")
822 # Re-render the mapping page
823 with open(tmp_path, "rb") as fh:
824 data = fh.read()
825 try:
826 parsed = parse_file(data, original_filename)
827 except ValueError:
828 session.pop(_IMPORT_SESSION_KEY, None)
829 return redirect(url_for("pilots.import_upload"))
830 preview = preview_rows(parsed, mapping, n=5)
831 return render_template(
832 "pilots/import_map.html",
833 norm_cols=parsed.norm_cols,
834 raw_cols=parsed.raw_cols,
835 base_norm_cols=[_norm(r) for r in parsed.raw_cols],
836 mapping=mapping,
837 match_type="alias",
838 fuzzy_score=0.0,
839 target_fields=TARGET_FIELDS,
840 preview=preview,
841 filename=original_filename,
842 type_hints=type_hints(parsed, mapping),
843 ), 422
845 # Parse opening balance
846 opening_balance: dict[str, float | None] = {}
847 ob_fields = [
848 "single_pilot_se",
849 "single_pilot_me",
850 "multi_pilot",
851 "night_time",
852 "instrument_time",
853 "function_pic",
854 "function_copilot",
855 "function_dual",
856 "function_instructor",
857 ]
858 for f_name in ob_fields:
859 raw = request.form.get(f"ob_{f_name}", "").strip()
860 opening_balance[f_name] = parse_duration_value(raw) if raw else None
862 # Re-parse the temp file
863 with open(tmp_path, "rb") as fh:
864 data = fh.read()
865 try:
866 parsed = parse_file(data, original_filename)
867 except ValueError as exc:
868 flash(str(exc), "danger")
869 session.pop(_IMPORT_SESSION_KEY, None)
870 return redirect(url_for("pilots.import_upload"))
872 # Create or reuse the mapping record
873 saved_mappings = LogbookImportMapping.query.filter_by(pilot_user_id=uid).all()
874 mapping_record: LogbookImportMapping | None = None
875 for m in saved_mappings:
876 if m.source_fingerprint == fingerprint:
877 # Update the saved mapping with the user's potentially-refined choices
878 m.column_mapping = json.dumps(mapping)
879 mapping_record = m
880 break
881 if mapping_record is None:
882 mapping_record = LogbookImportMapping(
883 pilot_user_id=uid,
884 source_fingerprint=fingerprint,
885 column_mapping=json.dumps(mapping),
886 source_columns=json.dumps(norm_cols),
887 created_at=_datetime.now(_tz.utc),
888 )
889 db.session.add(mapping_record)
890 db.session.flush() # get mapping_record.id
892 # Create the batch record (row counts filled in after execute)
893 batch = LogbookImportBatch(
894 pilot_user_id=uid,
895 mapping_id=mapping_record.id,
896 source_filename=original_filename,
897 imported_at=_datetime.now(_tz.utc),
898 )
899 db.session.add(batch)
900 db.session.flush() # get batch.id
902 result = execute_import(
903 parsed=parsed,
904 mapping=mapping,
905 pilot_user_id=uid,
906 batch_id=batch.id,
907 opening_balance=opening_balance if any(opening_balance.values()) else None,
908 )
910 batch.row_count = result.imported
911 batch.subtotal_count = result.subtotals
912 batch.skipped_count = len(result.skipped)
913 batch.has_opening_balance = result.has_opening_balance
915 db.session.commit()
917 # Clean up temp file and session
918 try:
919 os.remove(tmp_path)
920 except OSError as exc:
921 current_app.logger.debug("cleanup tmp import file: %s", exc)
922 session.pop(_IMPORT_SESSION_KEY, None)
924 flash(
925 _(
926 "Import complete: %(imported)d entries imported, %(subtotals)d subtotal rows "
927 "skipped, %(skipped)d rows could not be parsed.",
928 imported=result.imported,
929 subtotals=result.subtotals,
930 skipped=len(result.skipped),
931 ),
932 "success",
933 )
934 if result.skipped:
935 detail = "; ".join(f"row {r}: {reason}" for r, reason in result.skipped[:5])
936 if len(result.skipped) > 5:
937 detail += f" … and {len(result.skipped) - 5} more"
938 flash(_("Skipped rows: %(detail)s", detail=detail), "warning")
940 if result.parse_warnings:
941 n = len(result.parse_warnings)
942 examples = "; ".join(
943 f"row {r}, {target}: {raw}"
944 for r, _col, target, raw in result.parse_warnings[:3]
945 )
946 if n > 3:
947 examples += f" … +{n - 3}"
948 flash(
949 ngettext(
950 "One cell value could not be parsed and was imported as blank: %(examples)s",
951 "%(n)d cell values could not be parsed and were imported as blank: %(examples)s",
952 n,
953 n=n,
954 examples=examples,
955 ),
956 "warning",
957 )
959 if result.total_mismatch_warnings:
960 n = len(result.total_mismatch_warnings)
961 examples = "; ".join(
962 _(
963 "row %(row)d (source %(src).1f h, computed %(comp).1f h)",
964 row=r,
965 src=src,
966 comp=comp,
967 )
968 for r, src, comp in result.total_mismatch_warnings[:3]
969 )
970 if n > 3:
971 examples += f" … +{n - 3}"
972 flash(
973 ngettext(
974 "One row has a total flight time that doesn't match the sum of its components — please review: %(examples)s",
975 "%(n)d rows have a total flight time that doesn't match the sum of their components — please review: %(examples)s",
976 n,
977 n=n,
978 examples=examples,
979 ),
980 "warning",
981 )
983 return redirect(url_for("pilots.import_history"))
986@pilots_bp.route("/pilot/logbook/import/history")
987@login_required
988@require_pilot_access
989def import_history() -> ResponseReturnValue:
990 uid = _current_user_id()
991 batches = (
992 LogbookImportBatch.query.filter_by(pilot_user_id=uid)
993 .order_by(LogbookImportBatch.imported_at.desc())
994 .all()
995 )
996 return render_template("pilots/import_history.html", batches=batches)
999@pilots_bp.route("/pilot/logbook/import/<int:batch_id>/rollback", methods=["POST"])
1000@login_required
1001@require_pilot_access
1002def import_rollback(batch_id: int) -> ResponseReturnValue:
1003 uid = _current_user_id()
1004 batch = db.session.get(LogbookImportBatch, batch_id)
1005 if not batch or batch.pilot_user_id != uid:
1006 abort(404)
1008 # Delete all entries belonging to this batch
1009 PilotLogbookEntry.query.filter_by(import_batch_id=batch_id).delete()
1010 db.session.delete(batch)
1011 db.session.commit()
1013 flash(
1014 ngettext(
1015 "Import deleted: one entry removed.",
1016 "Import deleted: all %(count)d entries removed.",
1017 batch.row_count,
1018 count=batch.row_count,
1019 ),
1020 "success",
1021 )
1022 return redirect(url_for("pilots.import_history"))
1025# ── Pilot GPS import (airplane-agnostic batch upload) ────────────────────────
1027_GPS_ALLOWED_EXTS = {".gpx", ".kml", ".csv"}
1028_GPS_MAX_BYTES = 20 * 1024 * 1024
1029_BLOCK_TOLERANCE_PILOT = _td(minutes=15)
1032def _pilot_gps_tmp_dir() -> str:
1033 from aircraft.routes import _gps_tmp_dir # noqa: PLC0415
1035 return _gps_tmp_dir()
1038def _pilot_tenant_id(user_id: int) -> int | None:
1039 tu = TenantUser.query.filter_by(user_id=user_id).first()
1040 return tu.tenant_id if tu else None
1043def _pilot_match_segment(
1044 user_id: int, block_off: "_datetime", block_on: "_datetime"
1045) -> list["FlightEntry"]:
1046 """Find FlightEntry records the pilot is associated with that overlap block times."""
1047 tol = _BLOCK_TOLERANCE_PILOT
1049 crew_flight_ids = db.session.query(FlightCrew.flight_id).filter(
1050 FlightCrew.user_id == user_id, FlightCrew.flight_id.isnot(None)
1051 )
1052 logbook_flight_ids = db.session.query(PilotLogbookEntry.flight_id).filter(
1053 PilotLogbookEntry.pilot_user_id == user_id,
1054 PilotLogbookEntry.flight_id.isnot(None),
1055 )
1056 all_ids = crew_flight_ids.union(logbook_flight_ids).scalar_subquery()
1058 return FlightEntry.query.filter( # type: ignore[no-any-return]
1059 FlightEntry.id.in_(all_ids),
1060 FlightEntry.block_off_utc.isnot(None),
1061 FlightEntry.block_on_utc.isnot(None),
1062 FlightEntry.block_off_utc < block_on + tol,
1063 FlightEntry.block_on_utc > block_off - tol,
1064 ).all()
1067def _pilot_seg_match_dict(matches: list[Any]) -> dict[str, Any]:
1068 """Summarise match results into fields stored on the segment dict."""
1069 if not matches:
1070 return {
1071 "matched_flight_id": None,
1072 "matched_flight_str": None,
1073 "matched_has_existing_track": False,
1074 "matched_aircraft_id": None,
1075 "matched_aircraft_reg": None,
1076 "matched_ambiguous": False,
1077 "matched_candidates": [],
1078 }
1079 candidates = [
1080 {
1081 "id": fe.id,
1082 "str": f"#{fe.id} — {fe.date} {fe.departure_icao} → {fe.arrival_icao}",
1083 "aircraft_id": fe.aircraft_id,
1084 "aircraft_reg": fe.aircraft.registration if fe.aircraft else "?",
1085 "has_existing_track": fe.gps_track_id is not None,
1086 }
1087 for fe in matches
1088 ]
1089 primary = matches[0]
1090 return {
1091 "matched_flight_id": primary.id,
1092 "matched_flight_str": candidates[0]["str"],
1093 "matched_has_existing_track": primary.gps_track_id is not None,
1094 "matched_aircraft_id": primary.aircraft_id,
1095 "matched_aircraft_reg": candidates[0]["aircraft_reg"],
1096 "matched_ambiguous": len(matches) > 1,
1097 "matched_candidates": candidates,
1098 }
1101@pilots_bp.route("/pilot/gps-import", methods=["GET", "POST"])
1102@login_required
1103@require_pilot_access
1104def pilot_gps_import_upload() -> ResponseReturnValue:
1105 uid = _current_user_id()
1106 tenant_id = _pilot_tenant_id(uid)
1107 tenant_aircraft = (
1108 Aircraft.query.filter_by(tenant_id=tenant_id)
1109 .order_by(Aircraft.registration)
1110 .all()
1111 if tenant_id
1112 else []
1113 )
1115 if request.method == "GET":
1116 return render_template(
1117 "pilots/gps_import_upload.html",
1118 tenant_aircraft=tenant_aircraft,
1119 )
1121 # ── POST: process uploaded files ──────────────────────────────────────────
1122 mode = request.form.get("mode", "agnostic") # "one_aircraft" | "agnostic"
1123 files = request.files.getlist("gps_files")
1125 if not files or all(f.filename == "" for f in files):
1126 flash(_("Please select at least one GPS log file."), "warning")
1127 return render_template(
1128 "pilots/gps_import_upload.html", tenant_aircraft=tenant_aircraft
1129 )
1131 from aircraft.gps_import import parse_gps_file # noqa: PLC0415
1133 tmp_dir = _pilot_gps_tmp_dir()
1134 parsed_meta: list[dict[str, Any]] = []
1135 errors: list[str] = []
1136 skipped_empty = 0
1138 for f in files:
1139 if not f.filename:
1140 continue # pragma: no cover – Werkzeug never yields empty-filename FileStorage objects
1141 ext = os.path.splitext(f.filename.lower())[1]
1142 if ext not in _GPS_ALLOWED_EXTS:
1143 errors.append(
1144 _(
1145 "%(fn)s: unsupported file type (use .gpx, .kml, or .csv).",
1146 fn=f.filename,
1147 )
1148 )
1149 continue
1150 data = f.read(_GPS_MAX_BYTES + 1)
1151 if len(data) > _GPS_MAX_BYTES:
1152 errors.append(_("%(fn)s: file too large (20 MB limit).", fn=f.filename))
1153 continue
1154 try:
1155 parsed = parse_gps_file(data, f.filename)
1156 except ValueError as exc:
1157 errors.append(_("%(fn)s: %(err)s", fn=f.filename, err=str(exc)))
1158 continue
1159 if parsed.classification == "empty":
1160 skipped_empty += 1
1161 continue
1162 uid_hex = uuid.uuid4().hex
1163 safe_name = f"{uid_hex}_{secure_filename(f.filename)}"
1164 tmp_path = os.path.join(tmp_dir, safe_name)
1165 with open(tmp_path, "wb") as fh:
1166 fh.write(data)
1167 parsed_meta.append(
1168 {
1169 "tmp_path": tmp_path,
1170 "original_filename": f.filename,
1171 "format": parsed.format,
1172 "classification": parsed.classification,
1173 "trkpt_count": len(parsed.trackpoints),
1174 "hint_dep": parsed.hint_departure_icao,
1175 "hint_arr": parsed.hint_arrival_icao,
1176 "device_id": getattr(parsed, "device_id", None),
1177 }
1178 )
1180 for e in errors:
1181 flash(e, "danger")
1182 if skipped_empty:
1183 flash(
1184 ngettext(
1185 "%(n)s file skipped — no movement detected.",
1186 "%(n)s files skipped — no movement detected.",
1187 skipped_empty,
1188 n=skipped_empty,
1189 ),
1190 "info",
1191 )
1192 if not parsed_meta:
1193 flash(_("No valid GPS files to import."), "warning")
1194 return render_template(
1195 "pilots/gps_import_upload.html", tenant_aircraft=tenant_aircraft
1196 )
1198 if mode == "one_aircraft":
1199 aircraft_id = request.form.get("aircraft_id", type=int)
1200 if not aircraft_id:
1201 flash(_("Please select an aircraft."), "warning")
1202 return render_template(
1203 "pilots/gps_import_upload.html", tenant_aircraft=tenant_aircraft
1204 )
1205 session["gps_import"] = {
1206 "user_id": session["user_id"],
1207 "aircraft_id": aircraft_id,
1208 "files": parsed_meta,
1209 "skipped_empty": skipped_empty,
1210 "other_aircraft": False,
1211 "other_ac_make_model": "",
1212 "other_ac_reg": "",
1213 }
1214 session.modified = True
1215 return redirect(url_for("aircraft.gps_import_review", aircraft_id=aircraft_id))
1217 # agnostic mode
1218 session["pilot_gps_import"] = {
1219 "user_id": session["user_id"],
1220 "files": parsed_meta,
1221 "skipped_empty": skipped_empty,
1222 }
1223 session.modified = True
1224 return redirect(url_for("pilots.pilot_gps_import_review"))
1227@pilots_bp.route("/pilot/gps-import/review", methods=["GET"])
1228@login_required
1229@require_pilot_access
1230def pilot_gps_import_review() -> ResponseReturnValue:
1231 uid = _current_user_id()
1232 state = session.get("pilot_gps_import")
1233 if not state:
1234 flash(_("Session expired — please upload your GPS files again."), "warning")
1235 return redirect(url_for("pilots.pilot_gps_import_upload"))
1237 from aircraft.gps_import import ( # noqa: PLC0415
1238 detect_segments,
1239 merge_and_sort,
1240 parse_gps_file,
1241 )
1242 from aircraft.routes import ( # noqa: PLC0415
1243 _linked_pilot_entries,
1244 _segment_for_session,
1245 _segment_to_dict,
1246 _gps_tmp_dir,
1247 )
1249 file_metas = state["files"]
1250 all_parsed = []
1251 for meta in file_metas:
1252 try:
1253 with open(meta["tmp_path"], "rb") as fh:
1254 data = fh.read()
1255 parsed = parse_gps_file(data, meta["original_filename"])
1256 parsed.hint_departure_icao = meta.get("hint_dep")
1257 parsed.hint_arrival_icao = meta.get("hint_arr")
1258 all_parsed.append(parsed)
1259 except (OSError, ValueError):
1260 flash(
1261 _(
1262 "Could not read %(fn)s — please upload again.",
1263 fn=meta["original_filename"],
1264 ),
1265 "warning",
1266 )
1267 return redirect(url_for("pilots.pilot_gps_import_upload"))
1269 merged = merge_and_sort(all_parsed)
1270 hint_dep = next(
1271 (p.hint_departure_icao for p in all_parsed if p.hint_departure_icao), None
1272 )
1273 hint_arr = next(
1274 (p.hint_arrival_icao for p in all_parsed if p.hint_arrival_icao), None
1275 )
1276 segments = detect_segments(merged, hint_dep=hint_dep, hint_arr=hint_arr)
1277 full_segs = [_segment_to_dict(seg, i) for i, seg in enumerate(segments)]
1279 for seg in full_segs:
1280 block_off = _datetime.fromisoformat(seg["block_off_utc"])
1281 block_on = _datetime.fromisoformat(seg["block_on_utc"])
1282 matches = _pilot_match_segment(uid, block_off, block_on)
1283 seg.update(_pilot_seg_match_dict(matches))
1284 if seg.get("matched_flight_id") and not seg.get("matched_ambiguous"):
1285 seg["linked_pilot_entries"] = _linked_pilot_entries(
1286 seg["matched_flight_id"], uid
1287 )
1288 else:
1289 seg["linked_pilot_entries"] = []
1291 tmp_dir = _gps_tmp_dir()
1292 state["segments"] = [_segment_for_session(s, tmp_dir) for s in full_segs]
1293 session["pilot_gps_import"] = state
1294 session.modified = True
1296 tenant_id = _pilot_tenant_id(uid)
1297 tenant_aircraft = (
1298 Aircraft.query.filter_by(tenant_id=tenant_id)
1299 .order_by(Aircraft.registration)
1300 .all()
1301 if tenant_id
1302 else []
1303 )
1305 from models import AppSetting # noqa: PLC0415 # pyright: ignore[reportMissingImports]
1307 tile_setting = db.session.get(AppSetting, "openaip_api_key")
1308 openaip_key = tile_setting.value if tile_setting and tile_setting.value else None
1310 return render_template(
1311 "pilots/gps_import_review.html",
1312 segments=full_segs,
1313 skipped_empty=state.get("skipped_empty", 0),
1314 confirmed_segments=state.get("confirmed_segments", {}),
1315 tenant_aircraft=tenant_aircraft,
1316 openaip_key=openaip_key,
1317 )
1320@pilots_bp.route("/pilot/gps-import/confirm-one", methods=["POST"])
1321@login_required
1322@require_pilot_access
1323def pilot_gps_import_confirm_one() -> ResponseReturnValue:
1324 import decimal as _dec # noqa: PLC0415
1325 from aircraft.routes import _load_segment_geojson, _gps_cleanup # noqa: PLC0415
1326 from aircraft.gps_import import round_flight_time # noqa: PLC0415
1328 uid = _current_user_id()
1329 state = session.get("pilot_gps_import")
1330 if not state:
1331 flash(_("Session expired — please upload your GPS files again."), "warning")
1332 return redirect(url_for("pilots.pilot_gps_import_upload"))
1334 segments_data: list[dict[str, Any]] = state.get("segments", [])
1335 if not segments_data:
1336 flash(_("No segments to import."), "warning")
1337 return redirect(url_for("pilots.pilot_gps_import_upload"))
1339 try:
1340 seg_idx = int(request.form.get("seg_idx", ""))
1341 except (ValueError, TypeError):
1342 flash(_("Invalid segment index."), "danger")
1343 return redirect(url_for("pilots.pilot_gps_import_review"))
1345 if seg_idx < 0 or seg_idx >= len(segments_data):
1346 flash(_("Invalid segment index."), "danger")
1347 return redirect(url_for("pilots.pilot_gps_import_review"))
1349 confirmed = state.get("confirmed_segments", {})
1350 if str(seg_idx) in confirmed:
1351 flash(_("This segment has already been confirmed."), "info")
1352 return redirect(url_for("pilots.pilot_gps_import_review"))
1354 pilot_role = request.form.get("pilot_role", "pic")
1355 if pilot_role not in ("pic", "dual", "none"):
1356 pilot_role = "pic"
1358 # ── Skip ─────────────────────────────────────────────────────────────────
1359 if request.form.get("skip") == "1":
1360 confirmed[str(seg_idx)] = "skip"
1361 state["confirmed_segments"] = confirmed
1362 session["pilot_gps_import"] = state
1363 session.modified = True
1364 if len(confirmed) == len(segments_data):
1365 _gps_cleanup(state)
1366 session.pop("pilot_gps_import", None)
1367 imported = sum(1 for v in confirmed.values() if v != "skip")
1368 skipped_count = len(segments_data) - imported
1369 if imported > 0:
1370 flash(
1371 ngettext(
1372 "%(n)s flight imported successfully.",
1373 "%(n)s flights imported successfully.",
1374 imported,
1375 n=imported,
1376 ),
1377 "success",
1378 )
1379 flash(
1380 ngettext(
1381 "%(n)s segment skipped.",
1382 "%(n)s segments skipped.",
1383 skipped_count,
1384 n=skipped_count,
1385 ),
1386 "info",
1387 )
1388 if imported > 0 and pilot_role in ("pic", "dual"):
1389 return redirect(url_for("pilots.logbook"))
1390 return redirect(url_for("pilots.logbook"))
1391 flash(_("Segment skipped."), "info")
1392 return redirect(url_for("pilots.pilot_gps_import_review"))
1394 # ── Confirm ───────────────────────────────────────────────────────────────
1395 seg = segments_data[seg_idx]
1396 file_metas = state.get("files", [])
1397 block_off = _datetime.fromisoformat(seg["block_off_utc"])
1398 block_on = _datetime.fromisoformat(seg["block_on_utc"])
1400 dep_icao = (
1401 request.form.get("dep_icao") or seg.get("departure_icao") or ""
1402 ).strip().upper()[:4] or "????"
1403 arr_icao = (
1404 request.form.get("arr_icao") or seg.get("arrival_icao") or ""
1405 ).strip().upper()[:4] or "????"
1406 nature = (request.form.get("nature") or "").strip()[:100] or None
1407 remarks = (request.form.get("remarks") or "").strip() or None
1409 geojson = _load_segment_geojson(seg)
1410 source_filename = (
1411 file_metas[0]["original_filename"] if len(file_metas) == 1 else None
1412 )
1413 device_id = next(
1414 (m.get("device_id") for m in file_metas if m.get("device_id")), None
1415 )
1417 create_pilot_entry = pilot_role in ("pic", "dual")
1418 matched_flight_id = seg.get("matched_flight_id")
1419 entry: FlightEntry | None = None
1420 gps_track: GpsTrack | None = None
1421 ac: Aircraft | None = None
1423 if matched_flight_id:
1424 # Link GPS track to the existing matched FlightEntry
1425 existing = db.session.get(FlightEntry, matched_flight_id)
1426 if existing:
1427 gps_track = GpsTrack(
1428 source_filename=source_filename,
1429 device_id=device_id,
1430 block_off_utc=block_off,
1431 block_on_utc=block_on,
1432 departure_icao=dep_icao,
1433 arrival_icao=arr_icao,
1434 geojson=geojson,
1435 )
1436 db.session.add(gps_track)
1437 db.session.flush()
1438 existing.gps_track_id = gps_track.id
1439 existing.block_off_utc = block_off
1440 existing.block_on_utc = block_on
1441 # Link track to other users' pilot logbook entries for this flight
1442 # (only when they have no existing GPS track — preserve their own data).
1443 for ple in PilotLogbookEntry.query.filter(
1444 PilotLogbookEntry.flight_id == existing.id,
1445 PilotLogbookEntry.pilot_user_id != uid,
1446 PilotLogbookEntry.gps_track_id.is_(None),
1447 ).all():
1448 ple.gps_track_id = gps_track.id
1449 db.session.flush()
1450 entry = existing
1451 else:
1452 matched_flight_id = None # stale — fall through to new entry
1454 if not matched_flight_id:
1455 resolution = request.form.get("resolution", "other_aircraft")
1457 if resolution == "managed_aircraft":
1458 aircraft_id = request.form.get("aircraft_id", type=int)
1459 if aircraft_id:
1460 ac = db.session.get(Aircraft, aircraft_id)
1462 if ac:
1463 # Create a new FlightEntry for the managed aircraft
1464 gps_track = GpsTrack(
1465 source_filename=source_filename,
1466 device_id=device_id,
1467 block_off_utc=block_off,
1468 block_on_utc=block_on,
1469 departure_icao=dep_icao,
1470 arrival_icao=arr_icao,
1471 geojson=geojson,
1472 )
1473 db.session.add(gps_track)
1474 db.session.flush()
1475 flight_time_h = round_flight_time(
1476 seg.get("flight_time_raw_h", 0),
1477 getattr(ac, "logbook_time_precision", "tenth_hour"),
1478 )
1479 entry = FlightEntry(
1480 aircraft_id=ac.id,
1481 date=block_off.date(),
1482 departure_icao=dep_icao,
1483 arrival_icao=arr_icao,
1484 departure_time=block_off.time().replace(tzinfo=None),
1485 arrival_time=block_on.time().replace(tzinfo=None),
1486 flight_time=_dec.Decimal(str(flight_time_h)),
1487 landing_count=seg.get("landing_count") or 0,
1488 nature_of_flight=nature,
1489 source="gps_import",
1490 block_off_utc=block_off,
1491 block_on_utc=block_on,
1492 gps_track_id=gps_track.id,
1493 )
1494 db.session.add(entry)
1495 db.session.flush()
1496 else:
1497 # Other / external aircraft — pilot-only, no FlightEntry
1498 if geojson:
1499 gps_track = GpsTrack(
1500 source_filename=source_filename,
1501 device_id=device_id,
1502 block_off_utc=block_off,
1503 block_on_utc=block_on,
1504 departure_icao=dep_icao,
1505 arrival_icao=arr_icao,
1506 geojson=geojson,
1507 )
1508 db.session.add(gps_track)
1509 db.session.flush()
1510 create_pilot_entry = (
1511 True # always create logbook entry for external aircraft
1512 )
1514 if create_pilot_entry:
1515 ac_type: str | None = None
1516 ac_reg: str | None = None
1517 ac_cat: str = "SEP"
1518 if ac:
1519 ac_type = f"{ac.make} {ac.model}".strip()
1520 ac_reg = ac.registration
1521 ac_cat = getattr(ac, "category", "SEP")
1522 elif entry and entry.aircraft:
1523 _rel_ac = entry.aircraft
1524 ac_type = f"{_rel_ac.make} {_rel_ac.model}".strip()
1525 ac_reg = _rel_ac.registration
1526 ac_cat = getattr(_rel_ac, "category", "SEP")
1527 else:
1528 other_reg = (request.form.get("other_reg") or "").strip().upper()
1529 other_mm = (request.form.get("other_make_model") or "").strip()
1530 ac_type = other_mm or None
1531 ac_reg = other_reg or None
1532 ac_cat = "SEP"
1534 flight_time_h = round_flight_time(seg.get("flight_time_raw_h", 0), "tenth_hour")
1535 single_pilot_se = (
1536 _dec.Decimal(str(flight_time_h)) if ac_cat in ("SEP", "SET", "") else None
1537 )
1538 single_pilot_me = (
1539 _dec.Decimal(str(flight_time_h)) if ac_cat in ("MEP", "MET") else None
1540 )
1542 from models import User as _User # noqa: PLC0415 # pyright: ignore[reportMissingImports]
1544 _pilot_user = db.session.get(_User, uid)
1545 pilot_display_name = _pilot_user.display_name if _pilot_user else ""
1547 pentry = PilotLogbookEntry(
1548 pilot_user_id=uid,
1549 flight_id=entry.id if entry else None,
1550 date=block_off.date(),
1551 aircraft_type=ac_type,
1552 aircraft_registration=ac_reg,
1553 departure_place=dep_icao,
1554 departure_time=block_off.time().replace(tzinfo=None),
1555 arrival_place=arr_icao,
1556 arrival_time=block_on.time().replace(tzinfo=None),
1557 pic_name=pilot_display_name,
1558 single_pilot_se=single_pilot_se,
1559 single_pilot_me=single_pilot_me,
1560 function_pic=_dec.Decimal(str(flight_time_h))
1561 if pilot_role == "pic"
1562 else None,
1563 function_dual=_dec.Decimal(str(flight_time_h))
1564 if pilot_role == "dual"
1565 else None,
1566 landings_day=seg.get("landing_count") or 0,
1567 remarks=remarks,
1568 source="gps_import",
1569 gps_track_id=gps_track.id if gps_track else None,
1570 )
1571 db.session.add(pentry)
1573 db.session.commit()
1575 confirmed[str(seg_idx)] = entry.id if entry else 0
1576 state["confirmed_segments"] = confirmed
1577 session["pilot_gps_import"] = state
1578 session.modified = True
1580 all_handled = len(confirmed) == len(segments_data)
1581 if all_handled:
1582 _gps_cleanup(state)
1583 session.pop("pilot_gps_import", None)
1584 total = sum(1 for v in confirmed.values() if v != "skip")
1585 flash(
1586 ngettext(
1587 "%(n)s flight imported successfully.",
1588 "%(n)s flights imported successfully.",
1589 total,
1590 n=total,
1591 ),
1592 "success",
1593 )
1594 return redirect(url_for("pilots.logbook"))
1596 flash(_("Flight confirmed."), "success")
1597 return redirect(url_for("pilots.pilot_gps_import_review"))