Coverage for app/flights/shared_flight.py: 100%
205 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 15:29 +0000
1"""Editing a flight that is in more than one pilot's logbook.
3One Flight row is shared by both crew slots. The EASA figures and flight
4details on it ("shared fields") describe the flight itself, so they have a
5single owner: the pilot who logged it (``Flight.created_by_user_id``) while
6they are still linked to a slot — plus tenant owners/admins for a managed
7aircraft, who keep the aircraft log correct. Each slot's own name, role,
8function hours and personal remark ("personal fields") belong to the pilot
9in that slot alone.
11- ``can_edit_shared`` decides who may use the full flight form / offline
12 sync; everyone else linked to the flight is sent to the "my part of this
13 flight" page (``flights.crew_entry``) to edit their personal fields and
14 *suggest* corrections to shared ones (``FlightCorrectionSuggestion``).
15- ``protect_other_pilots`` / ``restore_other_pilots`` wrap every shared-field
16 save so it can never overwrite another linked pilot's personal fields.
17- ``notify_shared_changes`` e-mails the other linked pilots what changed.
18"""
20import logging
21from datetime import UTC, datetime
22from decimal import Decimal
23from typing import Any
25from flask import url_for # pyright: ignore[reportMissingImports]
26from flask_babel import gettext as _ # pyright: ignore[reportMissingImports]
27from flask_babel import lazy_gettext as _l # pyright: ignore[reportMissingImports]
28from models import ( # pyright: ignore[reportMissingImports]
29 CorrectionStatus,
30 CrewRole,
31 CrewSlot,
32 Flight,
33 FlightCorrectionSuggestion,
34 NotificationType,
35 Role,
36 TenantUser,
37 User,
38 db,
39)
40from offline.serialize import ( # pyright: ignore[reportMissingImports]
41 _fmt_decimal,
42 _fmt_int,
43 _fmt_str,
44 _fmt_time,
45)
46from pilots.form_parsing import ( # pyright: ignore[reportMissingImports]
47 _parse_date,
48 _parse_decimal,
49 _parse_int,
50 _parse_time,
51)
52from sqlalchemy import or_ # pyright: ignore[reportMissingImports]
54log = logging.getLogger(__name__)
56# ── Who may edit what ──────────────────────────────────────────────────────────
59def shared_editor_ids(fe: Flight) -> set[int]:
60 """Linked pilots allowed to edit the shared fields: the logger while still
61 linked, otherwise (logger gone, or not tracked) every linked pilot."""
62 linked = {u for u in (fe.pic_user_id, fe.second_crew_user_id) if u is not None}
63 if fe.created_by_user_id in linked:
64 return {fe.created_by_user_id}
65 return linked
68def can_edit_shared(fe: Flight, user_id: int | None, role: Any) -> bool:
69 """Whether *user_id* may change *fe*'s shared fields. Users not linked to
70 the flight keep the normal (tenant/aircraft) access rules."""
71 if fe.slot_for(user_id) is None:
72 return True
73 if fe.aircraft_id is not None and role in (Role.ADMIN, Role.OWNER):
74 return True
75 return user_id in shared_editor_ids(fe)
78_SLOT_PERSONAL_FIELDS: dict[str, tuple[str, ...]] = {
79 CrewSlot.PIC: ("pic_name", "function_pic", "pic_remarks"),
80 CrewSlot.SECOND: (
81 "second_crew_name",
82 "second_crew_role",
83 "function_copilot",
84 "function_dual",
85 "function_instructor",
86 "second_crew_remarks",
87 ),
88}
89_FUNCTION_FIELDS = {
90 "function_pic",
91 "function_copilot",
92 "function_dual",
93 "function_instructor",
94}
95# Which function_* column holds the second-crew slot's hours, per role.
96SECOND_CREW_FUNCTION_FIELD: dict[str, str] = {
97 CrewRole.IP: "function_instructor",
98 CrewRole.COPILOT: "function_copilot",
99 CrewRole.STUDENT: "function_dual",
100}
103def _flight_total(fe: Flight) -> Decimal | None:
104 if fe.flight_time is not None:
105 return Decimal(str(fe.flight_time))
106 total = fe.total_flight_time
107 return Decimal(str(total)) if total is not None else None
110def protect_other_pilots(fe: Flight, user_id: int | None) -> dict[str, Any]:
111 """Snapshot the personal fields of every slot linked to someone other
112 than *user_id* (``None`` → every linked slot), before a shared save."""
113 slots: dict[str, dict[str, Any]] = {}
114 for slot in CrewSlot.ALL:
115 occupant = fe.pic_user_id if slot == CrewSlot.PIC else fe.second_crew_user_id
116 if occupant is not None and occupant != user_id:
117 slots[slot] = {f: getattr(fe, f) for f in _SLOT_PERSONAL_FIELDS[slot]}
118 return {"total": _flight_total(fe), "slots": slots}
121def restore_other_pilots(fe: Flight, snapshot: dict[str, Any]) -> None:
122 """Undo whatever a shared save did to the protected personal fields.
123 Function hours that simply matched the old flight time follow the new
124 one, so a corrected flight time doesn't leave stale hours behind."""
125 new_total = _flight_total(fe)
126 old_total = snapshot["total"]
127 for values in snapshot["slots"].values():
128 for field, value in values.items():
129 if (
130 field in _FUNCTION_FIELDS
131 and value is not None
132 and old_total is not None
133 and new_total is not None
134 and Decimal(str(value)) == old_total
135 ):
136 value = new_total
137 setattr(fe, field, value)
140# ── Shared field specs (canonical strings) ────────────────────────────────────
142# field → (kind, label). Kinds: date | time | decimal | int | text | reg.
143FIELD_SPECS: dict[str, tuple[str, Any]] = {
144 "date": ("date", _l("Date")),
145 "other_aircraft_registration": ("reg", _l("Registration")),
146 "other_aircraft_type": ("text", _l("Aircraft type")),
147 "departure_icao": ("text", _l("Departure")),
148 "arrival_icao": ("text", _l("Arrival")),
149 "departure_time": ("time", _l("Departure time")),
150 "arrival_time": ("time", _l("Arrival time")),
151 "takeoff_time": ("time", _l("Takeoff time")),
152 "landing_time": ("time", _l("Landing time")),
153 "flight_time": ("decimal", _l("Flight time")),
154 "single_pilot_se": ("decimal", _l("S/E time")),
155 "single_pilot_me": ("decimal", _l("M/E time")),
156 "landings_day": ("int", _l("Day landings")),
157 "landings_night": ("int", _l("Night landings")),
158 "night_time": ("decimal", _l("Night time")),
159 "instrument_time": ("decimal", _l("Instrument time")),
160 "multi_pilot": ("decimal", _l("Multi-pilot time")),
161 "nature_of_flight": ("text", _l("Nature of flight")),
162 "notes": ("text", _l("Remarks")),
163}
166def _canonical(kind: str, value: Any) -> str:
167 if kind == "date":
168 return value.isoformat() if value else ""
169 if kind == "time":
170 return _fmt_time(value)
171 if kind == "decimal":
172 return _fmt_decimal(value, 1)
173 if kind == "int":
174 return _fmt_int(value)
175 return _fmt_str(value)
178def _parse(kind: str, raw: str, label: str) -> tuple[Any, str | None]:
179 if kind == "date":
180 value, err = _parse_date(raw, label)
181 if err is None and value is None:
182 return None, _("Date is required.")
183 return value, err
184 if kind == "time":
185 return _parse_time(raw, label)
186 if kind == "decimal":
187 return _parse_decimal(raw, label)
188 if kind == "int":
189 return _parse_int(raw, label)
190 text = raw.strip()
191 if kind == "reg":
192 text = text.upper()
193 return text or None, None
196def shared_snapshot(fe: Flight) -> dict[str, str]:
197 """Canonical values of every shared field, for change notifications.
198 The aircraft is described by its display registration/type, so a
199 managed aircraft reads the same way as a standalone one."""
200 values = {
201 f: _canonical(kind, getattr(fe, f)) for f, (kind, _lbl) in FIELD_SPECS.items()
202 }
203 values["other_aircraft_registration"] = _fmt_str(fe.display_registration)
204 values["other_aircraft_type"] = _fmt_str(fe.display_aircraft_type)
205 return values
208def suggestable_fields(fe: Flight) -> list[str]:
209 """Shared fields a pilot may propose new values for. On a managed
210 aircraft, times and flight time derive from counters the logger keeps
211 consistent, so they're left to the logger."""
212 if fe.aircraft_id is not None:
213 return [
214 "date",
215 "departure_icao",
216 "arrival_icao",
217 "landings_day",
218 "landings_night",
219 "night_time",
220 "instrument_time",
221 "multi_pilot",
222 "nature_of_flight",
223 "notes",
224 ]
225 time_fields = (
226 ["flight_time"]
227 if fe.flight_time is not None
228 else ["single_pilot_se", "single_pilot_me"]
229 )
230 return [
231 "date",
232 "other_aircraft_registration",
233 "other_aircraft_type",
234 "departure_icao",
235 "arrival_icao",
236 "departure_time",
237 "arrival_time",
238 "takeoff_time",
239 "landing_time",
240 *time_fields,
241 "landings_day",
242 "landings_night",
243 "night_time",
244 "instrument_time",
245 "multi_pilot",
246 "notes",
247 ]
250def field_label(field: str) -> str:
251 return str(FIELD_SPECS[field][1]) if field in FIELD_SPECS else field
254def current_values(fe: Flight) -> dict[str, str]:
255 return {f: _canonical(FIELD_SPECS[f][0], getattr(fe, f)) for f in FIELD_SPECS}
258# ── Personal fields ("my part of this flight") ────────────────────────────────
261def apply_personal_fields(fe: Flight, slot: str, form: Any) -> list[str]:
262 """Validate and apply the slot occupant's own name, role (second crew),
263 function hours and remark. Returns validation errors (nothing applied)."""
264 errors: list[str] = []
265 name = (form.get("name") or "").strip()
266 if not name:
267 errors.append(_("Name is required."))
268 hours, err = _parse_decimal(form.get("function_hours") or "", _("Function time"))
269 if err:
270 errors.append(err)
271 role = (form.get("role") or "").strip()
272 if slot == CrewSlot.SECOND and role not in (
273 CrewRole.IP,
274 CrewRole.COPILOT,
275 CrewRole.STUDENT,
276 CrewRole.SP,
277 ):
278 errors.append(_("Please choose your role on this flight."))
279 if errors:
280 return errors
282 remark = (form.get("remark") or "").strip() or None
283 if slot == CrewSlot.PIC:
284 fe.pic_name = name
285 fe.function_pic = hours
286 fe.pic_remarks = remark
287 else:
288 fe.second_crew_name = name
289 fe.second_crew_role = role
290 fe.function_copilot = None
291 fe.function_dual = None
292 fe.function_instructor = None
293 field = SECOND_CREW_FUNCTION_FIELD.get(role)
294 if field:
295 setattr(fe, field, hours)
296 fe.second_crew_remarks = remark
297 return []
300def personal_function_hours(fe: Flight, slot: str) -> Any:
301 if slot == CrewSlot.PIC:
302 return fe.function_pic
303 field = SECOND_CREW_FUNCTION_FIELD.get(fe.second_crew_role or "")
304 return getattr(fe, field) if field else None
307# ── Correction suggestions ────────────────────────────────────────────────────
310def build_suggestion(fe: Flight, form: Any) -> tuple[dict[str, list[str]], list[str]]:
311 """Compare the submitted values of the suggestable fields with *fe*.
312 Returns ``(changes, errors)`` with changes as field → [old, new]."""
313 changes: dict[str, list[str]] = {}
314 errors: list[str] = []
315 for field in suggestable_fields(fe):
316 kind, label = FIELD_SPECS[field]
317 value, err = _parse(kind, form.get(field) or "", str(label))
318 if err:
319 errors.append(err)
320 continue
321 old = _canonical(kind, getattr(fe, field))
322 new = _canonical(kind, value)
323 if old != new:
324 changes[field] = [old, new]
325 return changes, errors
328def apply_suggestion(
329 suggestion: FlightCorrectionSuggestion, actor_id: int
330) -> dict[str, str]:
331 """Apply an accepted suggestion to its flight, keeping every linked
332 pilot's personal fields intact. Returns the shared snapshot from before
333 the change (for ``notify_shared_changes``)."""
334 fe = _suggestion_flight(suggestion)
335 before = shared_snapshot(fe)
336 protected = protect_other_pilots(fe, None)
337 allowed = set(suggestable_fields(fe))
338 changed = {f: new for f, (_old, new) in suggestion.changes.items() if f in allowed}
339 for field, new in changed.items():
340 kind, label = FIELD_SPECS[field]
341 value, _err = _parse(kind, new, str(label))
342 setattr(fe, field, value)
343 # Pilot figures follow flight_time only while there's no engine-side
344 # (block) duration — see flights.routes.pilot_time.
345 if "flight_time" in changed and fe.engine_time is None:
346 if fe.single_pilot_me is not None:
347 fe.single_pilot_me = fe.flight_time
348 else:
349 fe.single_pilot_se = fe.flight_time
350 if fe.aircraft_id is not None and (
351 "landings_day" in changed or "landings_night" in changed
352 ):
353 fe.landing_count = (fe.landings_day or 0) + (fe.landings_night or 0)
354 restore_other_pilots(fe, protected)
355 suggestion.status = CorrectionStatus.ACCEPTED
356 suggestion.responded_at = datetime.now(UTC)
357 return before
360def reject_suggestion(suggestion: FlightCorrectionSuggestion) -> None:
361 suggestion.status = CorrectionStatus.REJECTED
362 suggestion.responded_at = datetime.now(UTC)
365def _suggestion_flight(suggestion: FlightCorrectionSuggestion) -> Flight:
366 fe = db.session.get(Flight, suggestion.flight_id)
367 assert fe is not None, "flight_id is a non-null CASCADE foreign key"
368 return fe
371def pending_suggestions_to_review(user_id: int) -> list[FlightCorrectionSuggestion]:
372 """Pending suggestions on flights *user_id* owns the shared fields of."""
373 candidates: list[FlightCorrectionSuggestion] = (
374 FlightCorrectionSuggestion.query.join(
375 Flight, Flight.id == FlightCorrectionSuggestion.flight_id
376 )
377 .filter(
378 FlightCorrectionSuggestion.status == CorrectionStatus.PENDING,
379 FlightCorrectionSuggestion.suggested_by_user_id != user_id,
380 or_(Flight.pic_user_id == user_id, Flight.second_crew_user_id == user_id),
381 )
382 .order_by(FlightCorrectionSuggestion.created_at)
383 .all()
384 )
385 return [
386 s for s in candidates if user_id in shared_editor_ids(_suggestion_flight(s))
387 ]
390def pending_suggestions_by(
391 user_id: int, fe: Flight
392) -> list[FlightCorrectionSuggestion]:
393 suggestions: list[FlightCorrectionSuggestion] = (
394 FlightCorrectionSuggestion.query.filter_by(
395 flight_id=fe.id,
396 suggested_by_user_id=user_id,
397 status=CorrectionStatus.PENDING,
398 )
399 .order_by(FlightCorrectionSuggestion.created_at)
400 .all()
401 )
402 return suggestions
405# ── Notifications ─────────────────────────────────────────────────────────────
408def _tenant_of(user_id: int) -> int | None:
409 tu = TenantUser.query.filter_by(user_id=user_id).first()
410 return tu.tenant_id if tu else None
413def _display_name(user_id: int | None) -> str:
414 user = db.session.get(User, user_id) if user_id else None
415 return user.display_name if user else "—"
418def _route(fe: Flight) -> str:
419 return f"{fe.departure_icao or '?'} → {fe.arrival_icao or '?'}"
422def dispatch_to_user(
423 user_id: int, notification_type: str, context: dict[str, Any]
424) -> None:
425 """Send one notification to one user, in their own tenant. Failures are
426 logged, never raised — the change it reports is already saved."""
427 from services.notification_service import ( # pyright: ignore[reportMissingImports]
428 dispatch,
429 )
431 tenant_id = _tenant_of(user_id)
432 if tenant_id is None:
433 return
434 try:
435 dispatch(notification_type, tenant_id, context, target_user_ids=[user_id])
436 except Exception:
437 log.exception("Failed to dispatch %s notification", notification_type)
440def _change_details(changes: list[tuple[str, str, str]]) -> list[tuple[Any, str]]:
441 return [
442 (FIELD_SPECS[f][1], f"{old or '—'} → {new or '—'}") for f, old, new in changes
443 ]
446def notify_shared_changes(fe: Flight, actor_id: int, before: dict[str, str]) -> None:
447 """E-mail every other linked pilot which shared fields *actor_id* changed."""
448 after = shared_snapshot(fe)
449 changes = [(f, before[f], after[f]) for f in after if before.get(f) != after[f]]
450 if not changes:
451 return
452 actor = _display_name(actor_id)
453 for user_id in sorted(fe.other_linked_user_ids(actor_id)):
454 dispatch_to_user(
455 user_id,
456 NotificationType.SHARED_FLIGHT_CHANGED,
457 {
458 "subject_key": _l("Your flight on %(date)s was updated"),
459 "subject_args": {"date": fe.date.isoformat()},
460 "notification_title_key": _l("%(name)s updated a flight you're on"),
461 "notification_title_args": {"name": actor},
462 "notification_message_key": _l(
463 "These details of your flight %(route)s on %(date)s changed:"
464 ),
465 "notification_message_args": {
466 "route": _route(fe),
467 "date": fe.date.isoformat(),
468 },
469 "details": _change_details(changes),
470 "cta_url": url_for("pilots.view_entry", entry_id=fe.id, _external=True),
471 },
472 )
475def notify_suggestion_created(suggestion: FlightCorrectionSuggestion) -> None:
476 fe = _suggestion_flight(suggestion)
477 suggester = _display_name(suggestion.suggested_by_user_id)
478 changes = [(f, old, new) for f, (old, new) in suggestion.changes.items()]
479 for user_id in sorted(shared_editor_ids(fe) - {suggestion.suggested_by_user_id}):
480 dispatch_to_user(
481 user_id,
482 NotificationType.FLIGHT_CORRECTION,
483 {
484 "subject_key": _l("Correction suggested for your flight on %(date)s"),
485 "subject_args": {"date": fe.date.isoformat()},
486 "notification_title_key": _l("%(name)s suggests a correction"),
487 "notification_title_args": {"name": suggester},
488 "notification_message_key": _l(
489 "%(name)s suggests changing your flight %(route)s on %(date)s. "
490 "Accept to apply it for both of you, or reject it."
491 ),
492 "notification_message_args": {
493 "name": suggester,
494 "route": _route(fe),
495 "date": fe.date.isoformat(),
496 },
497 "details": _change_details(changes),
498 "cta_url": url_for("pilots.logbook", _external=True)
499 + "#flight-corrections",
500 "cta_label": _l("Review in OpenHangar"),
501 },
502 )
505def notify_suggestion_rejected(
506 suggestion: FlightCorrectionSuggestion, actor_id: int
507) -> None:
508 fe = _suggestion_flight(suggestion)
509 actor = _display_name(actor_id)
510 changes = [(f, old, new) for f, (old, new) in suggestion.changes.items()]
511 dispatch_to_user(
512 suggestion.suggested_by_user_id,
513 NotificationType.FLIGHT_CORRECTION,
514 {
515 "subject_key": _l("Correction not applied to your flight on %(date)s"),
516 "subject_args": {"date": fe.date.isoformat()},
517 "notification_title_key": _l("%(name)s rejected your correction"),
518 "notification_title_args": {"name": actor},
519 "notification_message_key": _l(
520 "Your suggested correction for the flight %(route)s on %(date)s "
521 "was not applied."
522 ),
523 "notification_message_args": {
524 "route": _route(fe),
525 "date": fe.date.isoformat(),
526 },
527 "details": _change_details(changes),
528 "cta_url": url_for("flights.crew_entry", flight_id=fe.id, _external=True),
529 },
530 )