Coverage for app/pilots/logbook_import.py: 100%
621 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-31 10:13 +0000
1"""Pilot logbook import — parsing, normalisation, mapping, and execution."""
3from __future__ import annotations
5import csv
6import hashlib
7import io
8import json
9import math
10import re
11from collections.abc import Callable
12from dataclasses import dataclass, field
13from datetime import date, datetime, time, timedelta
14from typing import Any
16import openpyxl # pyright: ignore[reportMissingImports]
17from flask_babel import gettext as _ # pyright: ignore[reportMissingImports]
18from flask_babel import pgettext # pyright: ignore[reportMissingImports]
20# ── Constants ─────────────────────────────────────────────────────────────────
22TARGET_FIELDS: list[str] = [
23 "date",
24 "aircraft_type",
25 "aircraft_registration",
26 "departure_place",
27 "departure_time",
28 "arrival_place",
29 "arrival_time",
30 "pic_name",
31 "night_time",
32 "instrument_time",
33 "cross_country",
34 "landings_day",
35 "landings_night",
36 "single_pilot_se",
37 "single_pilot_me",
38 "multi_pilot",
39 "function_pic",
40 "function_copilot",
41 "function_dual",
42 "function_instructor",
43 "remarks",
44 # virtual target — used for import validation only, not stored
45 "total_flight_time_check",
46]
48# Normalised source column name → target field.
49# Positional disambiguation suffixes (_2, _3 …) are applied before lookup,
50# so "time" is the first TIME column (departure) and "time_2" is the second (arrival).
51_ALIASES: dict[str, str] = {
52 # EASA standard logbook column names (normalised)
53 "date dd/mm/yy": "date",
54 "date": "date",
55 "aircraft type": "aircraft_type",
56 "type": "aircraft_type",
57 "aircraft registration number": "aircraft_registration",
58 "registration": "aircraft_registration",
59 "reg": "aircraft_registration",
60 "from": "departure_place",
61 "departure": "departure_place",
62 "dep": "departure_place",
63 "time": "departure_time", # first TIME → departure
64 "time_2": "arrival_time", # second TIME → arrival
65 "departure time": "departure_time",
66 "arrival time": "arrival_time",
67 "to": "arrival_place",
68 "arrival": "arrival_place",
69 "arr": "arrival_place",
70 "pic name": "pic_name",
71 "pic name (if not student pilot)": "pic_name",
72 "captain": "pic_name",
73 # Landings group — first DAY / NIGHT pair
74 "day": "landings_day",
75 "night": "landings_night",
76 # Operational conditions group — second DAY / NIGHT pair (positional suffix)
77 "day_2": "ignore", # cross-country day — not a logbook field
78 "night_2": "night_time", # night flying time
79 # Aircraft category
80 "se": "single_pilot_se",
81 "single engine": "single_pilot_se",
82 "single pilot se": "single_pilot_se",
83 "me": "single_pilot_me",
84 "multi engine": "single_pilot_me",
85 "single pilot me": "single_pilot_me",
86 "multi pilot": "multi_pilot",
87 "mp": "multi_pilot",
88 # Pilot function
89 "pic": "function_pic",
90 "p1": "function_pic",
91 "co-pic": "function_copilot",
92 "co-pilot": "function_copilot",
93 "copilot": "function_copilot",
94 "p2": "function_copilot",
95 "dual received": "function_dual",
96 "dual": "function_dual",
97 "student": "function_dual",
98 "instructor": "function_instructor",
99 "fi": "function_instructor",
100 # Cross-country time
101 "cross-country": "cross_country",
102 "cross country": "cross_country",
103 "x-country": "cross_country",
104 "xc": "cross_country",
105 # Total flight time — imported only to validate against computed sum
106 "total flight time": "total_flight_time_check",
107 # Catch-all
108 "total": "ignore",
109 "no. istr. appr.": "ignore",
110 "ifr approaches": "ignore",
111 "page": "ignore",
112 "line": "ignore",
113 "remarks": "remarks",
114 "notes": "remarks",
115 "comments": "remarks",
116 "instrument time": "instrument_time",
117 "ifr": "instrument_time",
118 "instrument": "instrument_time",
119 "night time": "night_time",
120 # Group-prefixed column names (e.g. Belgian EASA logbook with a span header row)
121 "departure & arrival from": "departure_place",
122 "departure & arrival time": "departure_time",
123 "departure & arrival time_2": "arrival_time",
124 "departure & arrival to": "arrival_place",
125 "aircraft category se": "single_pilot_se",
126 "aircraft category me": "single_pilot_me",
127 "operational conditions cross-country": "cross_country",
128 "operational conditions day": "ignore",
129 "operational conditions night": "night_time",
130 "pilot function pic": "function_pic",
131 "pilot function co-pic": "function_copilot",
132 "pilot function dual received": "function_dual",
133 "page subtotals total flight time": "ignore",
134 "page subtotals total flight time_2": "ignore",
135 "page subtotals pic": "ignore",
136 "page subtotals dual": "ignore",
137 "page subtotals night": "ignore",
138 "page subtotals landings – day": "ignore",
139 "page subtotals landings – night": "ignore",
140 "page subtotals formated date": "ignore",
141 "page subtotals formated type": "ignore",
142 "page subtotals days since flight": "ignore",
143 "page subtotals daytime duration": "ignore",
144 "page subtotals non pic or dual": "ignore",
145 "formated date": "ignore",
146 "formated type": "ignore",
147 "days since flight": "ignore",
148 "daytime duration": "ignore",
149 "non pic or dual": "ignore",
150 # Landings group (group-prefixed)
151 "landings day": "landings_day",
152 "landings night": "landings_night",
153 # Aircraft type (multiline cell name normalised to single space)
154 "aircraft type name, model, variant": "aircraft_type",
155}
157# ── Data structures ───────────────────────────────────────────────────────────
160@dataclass
161class ParsedFile:
162 """Result of parsing an uploaded logbook file."""
164 norm_cols: list[str] # normalised + disambiguated column keys
165 raw_cols: list[str] # original column labels (for display)
166 header_row_index: int # 0-based row index of the detected header
167 data_rows: list[list[Any]] # all rows after the header (including subtotals)
168 fingerprint: str # SHA-256 of norm_cols
171@dataclass
172class MappingProposal:
173 """Proposed column mapping and how it was derived."""
175 mapping: dict[str, str] # norm_col_key → target_field or "ignore"
176 match_type: str # "exact", "fuzzy", "alias"
177 fuzzy_score: float = 0.0 # 0.0–1.0, only meaningful for "fuzzy"
178 matched_mapping_id: int | None = None
181@dataclass
182class ImportResult:
183 """Summary returned after executing an import."""
185 imported: int = 0
186 subtotals: int = 0
187 skipped: list[tuple[int, str]] = field(default_factory=list) # (row_num, reason)
188 # (row_num, reason) — rows that matched an entry already in the pilot's logbook
189 duplicates: list[tuple[int, str]] = field(default_factory=list)
190 # (row_num, source_col, target_field, repr(raw)) — non-empty cells that couldn't parse
191 parse_warnings: list[tuple[int, str, str, str]] = field(default_factory=list)
192 # (row_num, source_total, computed_total) — rows where total ≠ sum of components
193 total_mismatch_warnings: list[tuple[int, float, float]] = field(
194 default_factory=list
195 )
196 has_opening_balance: bool = False
199# ── Normalisation ─────────────────────────────────────────────────────────────
202def _norm(text: str) -> str:
203 """Strip, collapse whitespace, lower-case."""
204 return re.sub(r"\s+", " ", str(text).strip().lower())
207def _disambiguate(names: list[str]) -> list[str]:
208 """Append _2, _3 … to duplicate names to make them unique."""
209 seen: dict[str, int] = {}
210 result: list[str] = []
211 for n in names:
212 if n in seen:
213 seen[n] += 1
214 result.append(f"{n}_{seen[n]}")
215 else:
216 seen[n] = 1
217 result.append(n)
218 return result
221def _fingerprint(norm_cols: list[str]) -> str:
222 payload = json.dumps(norm_cols, ensure_ascii=False, sort_keys=False)
223 return hashlib.sha256(payload.encode()).hexdigest()
226# ── Header detection ──────────────────────────────────────────────────────────
229def _is_header_row(row: list[Any]) -> bool:
230 """True if ≥ 50 % of non-empty cells are non-numeric strings and ≥ 4 non-empty."""
231 non_empty = [c for c in row if c is not None and str(c).strip()]
232 if len(non_empty) < 4:
233 return False
234 string_like = [
235 c for c in non_empty if isinstance(c, str) and not _is_numeric_str(str(c))
236 ]
237 return len(string_like) / len(non_empty) >= 0.5
240def _is_numeric_str(s: str) -> bool:
241 try:
242 float(s)
243 return True
244 except ValueError:
245 return False
248def _header_alias_score(row: list[Any]) -> int:
249 """Count how many cells in *row* match a known alias."""
250 return sum(1 for c in row if c is not None and _norm(str(c)) in _ALIASES)
253def _find_header_row(rows: list[list[Any]], max_scan: int = 20) -> int | None:
254 """Return 0-based index of the best header row within the first max_scan rows.
256 Many logbook templates have a group-label row (e.g. "DEPARTURE & ARRIVAL",
257 "LANDINGS") above the actual column-header row. Both pass _is_header_row,
258 but the actual header row has far more alias matches. We score every
259 candidate and return the one with the highest score; ties go to the earlier
260 row. If no alias matches are found we fall back to the first text-like row.
261 """
262 best_idx: int | None = None
263 best_score = -1
264 first_text_row: int | None = None
266 for i, row in enumerate(rows[:max_scan]):
267 if not _is_header_row(row):
268 continue
269 if first_text_row is None:
270 first_text_row = i
271 score = _header_alias_score(row)
272 if score > best_score:
273 best_score = score
274 best_idx = i
276 return best_idx if (best_idx is not None and best_score > 0) else first_text_row
279def _trim_trailing_empty_cols(
280 all_rows: list[list[Any]], max_scan: int = 50
281) -> list[list[Any]]:
282 """Trim columns beyond the rightmost non-empty value in the first max_scan rows.
284 Excel templates often declare thousands of formatted-but-empty columns.
285 Without trimming, every empty cell becomes a separate mapping entry.
286 """
287 max_col = 0
288 for row in all_rows[:max_scan]:
289 for j in range(len(row) - 1, -1, -1):
290 if row[j] is not None and str(row[j]).strip():
291 max_col = max(max_col, j + 1)
292 break
293 if max_col == 0:
294 return (
295 all_rows # pragma: no cover — blank file rejected by header detection first
296 )
297 return [row[:max_col] for row in all_rows]
300# ── Group-header detection ────────────────────────────────────────────────────
303def _merge_label_map(ws: Any) -> dict[tuple[int, int], str]:
304 """Return {(0-based row, 0-based col): group_label} for every merged region.
306 The label is the value of the top-left cell of each merged range.
307 Every cell in the range gets the same label so downstream code can do a
308 simple lookup without forward-fill.
309 """
310 result: dict[tuple[int, int], str] = {}
311 for mc in ws.merged_cells.ranges:
312 val = ws.cell(mc.min_row, mc.min_col).value
313 if val is None or not str(val).strip():
314 continue
315 label = str(val).strip()
316 for r in range(mc.min_row - 1, mc.max_row):
317 for c in range(mc.min_col - 1, mc.max_col):
318 result[(r, c)] = label
319 return result
322def _group_labels_from_map(
323 merge_map: dict[tuple[int, int], str], row_idx: int, width: int
324) -> list[str]:
325 """Return per-column group labels for *row_idx* using exact merge metadata."""
326 return [merge_map.get((row_idx, c), "") for c in range(width)]
329def _group_labels_heuristic(row: list[Any], width: int) -> list[str] | None:
330 """Fallback for CSV: forward-fill a sparse row to infer group labels.
332 Requires at least two non-empty values with a gap between them (so a single
333 spanning title cell does not fire). Returns None if the pattern is absent.
334 """
335 padded: list[Any] = list(row[:width]) + [None] * max(0, width - len(row))
337 prev_nonempty_idx: int | None = None
338 has_span = False
339 for i, val in enumerate(padded):
340 if val is not None and str(val).strip():
341 if prev_nonempty_idx is not None and i > prev_nonempty_idx + 1:
342 has_span = True
343 break
344 prev_nonempty_idx = i
345 if not has_span:
346 return None
348 result: list[str] = []
349 current = ""
350 for val in padded:
351 if val is not None and str(val).strip():
352 current = str(val).strip()
353 result.append(current)
354 return result
357def _apply_group_labels(group_labels: list[str], raw_header: list[str]) -> list[str]:
358 """Prepend each non-empty group label to the corresponding column header."""
359 result = []
360 for i, col in enumerate(raw_header):
361 label = group_labels[i] if i < len(group_labels) else ""
362 if label and col:
363 result.append(f"{label} {col}")
364 elif label:
365 result.append(label)
366 else:
367 result.append(col)
368 return result
371# ── File parsing ──────────────────────────────────────────────────────────────
374def parse_file(data: bytes, filename: str) -> ParsedFile:
375 """Parse bytes from an uploaded file into a ParsedFile.
377 Raises ValueError with a user-friendly message on format errors.
378 """
379 ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
380 if ext in ("xlsx", "xls"):
381 return _parse_excel(data, filename)
382 if ext == "csv":
383 return _parse_csv(data, filename)
384 raise ValueError(
385 f"Unsupported file format: .{ext} — please upload a .csv or .xlsx file."
386 )
389# English-only Excel tab names used to identify the logbook sheet.
390# Locale-specific equivalents live in the translation catalogue, keyed under
391# the "excel tab name" pgettext context (see _preferred_sheet_names below).
392_PREFERRED_SHEET_NAMES_EN: frozenset[str] = frozenset(
393 {
394 "logbook",
395 "log",
396 "flights",
397 "flight log",
398 "flights log",
399 "journal",
400 }
401)
404def _preferred_sheet_names() -> frozenset[str]:
405 """Return preferred sheet names merged with their active-locale translations.
407 Uses pgettext with a dedicated context rather than plain _() — these
408 keywords are matched against user-supplied Excel tab names, not
409 displayed anywhere in the UI, and several of them (logbook/flight log,
410 log/journal) intentionally translate to the same word in French. A
411 context keeps that expected collision scoped to this lookup, so it can
412 never silently mask a real duplicate-translation issue if the same
413 English word is ever used as an actual UI label elsewhere.
415 The context string is repeated as a literal at each call site (not a
416 shared constant) because pybabel's extractor is a static source scanner —
417 it can only resolve literal string arguments, not a variable reference.
418 """
419 t_logbook = pgettext("excel tab name", "logbook")
420 t_log = pgettext("excel tab name", "log")
421 t_flights = pgettext("excel tab name", "flights")
422 t_flight_log = pgettext("excel tab name", "flight log")
423 t_flights_log = pgettext("excel tab name", "flights log")
424 t_journal = pgettext("excel tab name", "journal")
425 translated = frozenset(
426 {t_logbook, t_log, t_flights, t_flight_log, t_flights_log, t_journal}
427 )
428 return _PREFERRED_SHEET_NAMES_EN | frozenset(s.strip().lower() for s in translated)
431def _pick_best_excel_sheet(wb: Any) -> tuple[str, list[list[Any]]]:
432 """Return (sheet_name, all_rows) for the most likely logbook sheet.
434 Prefers sheets whose name matches a known logbook keyword (fast path, reads
435 only the chosen sheet). When no name matches, reads every sheet and scores
436 by alias matches in the detected header, returning the highest-scoring sheet
437 together with its rows. Falls back to the first sheet when scoring is
438 inconclusive.
440 Rows are returned here because openpyxl read-only worksheets use a one-shot
441 generator — calling iter_rows() twice on the same worksheet yields nothing on
442 the second call.
443 """
444 preferred = _preferred_sheet_names()
445 for ws in wb.worksheets:
446 if ws.title.strip().lower() in preferred:
447 return ws.title, [list(row) for row in ws.iter_rows(values_only=True)]
449 best_name = wb.worksheets[0].title if wb.worksheets else ""
450 best_score = -1
451 best_rows: list[list[Any]] = []
452 for ws in wb.worksheets:
453 rows = [list(row) for row in ws.iter_rows(values_only=True)]
454 trimmed = _trim_trailing_empty_cols(list(rows))
455 hi = _find_header_row(trimmed)
456 score = _header_alias_score(trimmed[hi]) if hi is not None else 0
457 if score > best_score:
458 best_score = score
459 best_name = ws.title
460 best_rows = rows
461 return best_name, best_rows
464def _parse_excel(data: bytes, filename: str) -> ParsedFile:
465 try:
466 wb_ro = openpyxl.load_workbook(io.BytesIO(data), read_only=True, data_only=True)
467 except Exception as exc:
468 raise ValueError(f"Could not open Excel file: {exc}") from exc
469 sheet_name, all_rows = _pick_best_excel_sheet(wb_ro)
470 wb_ro.close()
472 # Second load (non-read-only) to access merged cell ranges for exact group labels.
473 excel_merge_map: dict[tuple[int, int], str] | None = None
474 try:
475 wb_full = openpyxl.load_workbook(io.BytesIO(data), data_only=True)
476 excel_merge_map = _merge_label_map(wb_full[sheet_name])
477 wb_full.close()
478 except Exception: # noqa: S110, BLE001 # fall back to heuristic in _build_parsed_file
479 pass
481 return _build_parsed_file(all_rows, filename, excel_merge_map=excel_merge_map)
484def _parse_csv(data: bytes, filename: str) -> ParsedFile:
485 try:
486 text = data.decode("utf-8-sig")
487 except UnicodeDecodeError:
488 text = data.decode("latin-1")
490 sample = text[:4096]
491 try:
492 dialect = csv.Sniffer().sniff(sample)
493 except csv.Error:
494 dialect = csv.excel
496 reader = csv.reader(io.StringIO(text), dialect)
497 try:
498 all_rows: list[list[Any]] = list(reader)
499 except csv.Error as exc:
500 raise ValueError(f"Could not parse CSV file: {exc}") from exc
501 return _build_parsed_file(all_rows, filename)
504def _build_parsed_file(
505 all_rows: list[list[Any]],
506 filename: str,
507 excel_merge_map: dict[tuple[int, int], str] | None = None,
508) -> ParsedFile:
509 all_rows = _trim_trailing_empty_cols(all_rows)
510 header_idx = _find_header_row(all_rows)
511 if header_idx is None:
512 raise ValueError(
513 "Could not detect a header row in this file. "
514 "Make sure the file contains column names."
515 )
517 raw_header = [str(c).strip() if c is not None else "" for c in all_rows[header_idx]]
519 # Prepend group labels from the row above the header, if one exists
520 if header_idx > 0:
521 width = len(raw_header)
522 if excel_merge_map is not None:
523 # Excel: exact boundaries from merged cell metadata
524 _gl = _group_labels_from_map(excel_merge_map, header_idx - 1, width)
525 group_labels: list[str] | None = _gl if any(_gl) else None
526 else:
527 # CSV: heuristic forward-fill (merged cell info unavailable)
528 group_labels = _group_labels_heuristic(all_rows[header_idx - 1], width)
529 if group_labels is not None:
530 raw_header = _apply_group_labels(group_labels, raw_header)
532 norm_raw = [_norm(c) for c in raw_header]
533 norm_cols = _disambiguate(norm_raw)
534 data_rows = all_rows[header_idx + 1 :]
536 return ParsedFile(
537 norm_cols=norm_cols,
538 raw_cols=raw_header,
539 header_row_index=header_idx,
540 data_rows=data_rows,
541 fingerprint=_fingerprint(norm_cols),
542 )
545# ── Mapping proposal ──────────────────────────────────────────────────────────
548def propose_mapping(
549 parsed: ParsedFile,
550 existing_mappings: list[Any], # list[LogbookImportMapping]
551) -> MappingProposal:
552 """Return the best mapping proposal for *parsed*, checking saved mappings first."""
553 # 1. Exact fingerprint match
554 for m in existing_mappings:
555 if m.source_fingerprint == parsed.fingerprint:
556 return MappingProposal(
557 mapping=json.loads(m.column_mapping),
558 match_type="exact",
559 matched_mapping_id=m.id,
560 )
562 # 2. Fuzzy match — best overlap among saved mappings
563 best_score = 0.0
564 best_m = None
565 new_set = set(parsed.norm_cols)
566 for m in existing_mappings:
567 saved_cols: list[str] = json.loads(m.source_columns)
568 saved_set = set(saved_cols)
569 if not saved_set:
570 continue
571 overlap = len(new_set & saved_set)
572 score = overlap / max(len(new_set), len(saved_set))
573 if score > best_score:
574 best_score = score
575 best_m = m
577 if best_m is not None and best_score >= 0.6:
578 saved_map: dict[str, str] = json.loads(best_m.column_mapping)
579 # Build a mapping for the new columns, falling back to alias for unmatched ones
580 merged = _alias_mapping(parsed.norm_cols)
581 for col in parsed.norm_cols:
582 if col in saved_map:
583 merged[col] = saved_map[col]
584 return MappingProposal(
585 mapping=merged,
586 match_type="fuzzy",
587 fuzzy_score=best_score,
588 matched_mapping_id=best_m.id,
589 )
591 # 3. Alias-only auto-mapping
592 return MappingProposal(
593 mapping=_alias_mapping(parsed.norm_cols),
594 match_type="alias",
595 )
598def _alias_mapping(norm_cols: list[str]) -> dict[str, str]:
599 """Apply the built-in alias table; unknown columns default to 'ignore'."""
600 result: dict[str, str] = {}
601 for col in norm_cols:
602 result[col] = _ALIASES.get(col, "ignore")
603 return result
606# ── Subtotal detection ────────────────────────────────────────────────────────
609def _is_subtotal_row(row: list[Any], date_col_idx: int | None) -> bool:
610 if date_col_idx is None:
611 return False
612 if date_col_idx >= len(row):
613 return True
614 val = row[date_col_idx]
615 if isinstance(val, timedelta):
616 return True
617 if val is None or (isinstance(val, str) and not val.strip()):
618 return True
619 return bool(isinstance(val, str) and "total" in val.lower())
622# ── Value parsing ─────────────────────────────────────────────────────────────
625def parse_date_value(val: Any) -> date | None:
626 if isinstance(val, datetime):
627 return val.date()
628 if isinstance(val, date):
629 return val
630 if isinstance(val, (int, float)):
631 # Excel serial date number — openpyxl returns datetime; guard anyway
632 return None
633 if not isinstance(val, str):
634 return None
635 s = val.strip()
636 if not s:
637 return None
638 for fmt in ("%d/%m/%y", "%d/%m/%Y", "%Y-%m-%d", "%m/%d/%Y", "%d-%m-%Y"):
639 try:
640 return datetime.strptime(s, fmt).date()
641 except ValueError:
642 pass
643 return None
646def parse_time_value(val: Any) -> time | None:
647 """Parse a time-of-day value (HH:MM or Python time)."""
648 if isinstance(val, time):
649 return val
650 if isinstance(val, datetime):
651 return val.time()
652 if not isinstance(val, str):
653 return None
654 s = val.strip()
655 if not s:
656 return None
657 m = re.match(r"^(\d{1,2}):(\d{2})$", s)
658 if m:
659 try:
660 return time(int(m.group(1)), int(m.group(2)))
661 except ValueError:
662 return None
663 return None
666def parse_duration_value(val: Any) -> float | None:
667 """Parse a duration into decimal hours (e.g. time(0,42) → 0.7, '1:24' → 1.4)."""
668 if isinstance(val, timedelta):
669 hours = val.total_seconds() / 3600
670 return round(hours, 1) if hours >= 0 else None
671 if isinstance(val, time):
672 return round(val.hour + val.minute / 60, 1)
673 if isinstance(val, datetime):
674 # Excel sometimes returns a dummy date + the time-of-day
675 t = val.time()
676 return round(t.hour + t.minute / 60, 1)
677 if isinstance(val, (int, float)):
678 if not math.isfinite(val) or val < 0:
679 return None
680 return round(float(val), 1)
681 if isinstance(val, str):
682 s = val.strip()
683 if not s:
684 return None
685 m = re.match(r"^(\d+):(\d{2})$", s)
686 if m:
687 return round(int(m.group(1)) + int(m.group(2)) / 60, 1)
688 try:
689 v = float(s)
690 return round(v, 1) if math.isfinite(v) and v >= 0 else None
691 except ValueError:
692 return None
693 return None
696def parse_int_value(val: Any) -> int | None:
697 if isinstance(val, int):
698 return val if val >= 0 else None
699 if isinstance(val, float):
700 try:
701 return int(val) if val >= 0 else None
702 except (ValueError, OverflowError):
703 return None
704 if isinstance(val, str):
705 s = val.strip()
706 if not s:
707 return None
708 try:
709 n = int(float(s))
710 except (ValueError, OverflowError):
711 return None
712 return n if n >= 0 else None
713 return None
716def _is_nonempty(val: Any) -> bool:
717 """Return True when val carries a real value (not None, not blank string)."""
718 if val is None:
719 return False
720 if isinstance(val, str):
721 return bool(val.strip())
722 return True
725# Map target field → (human-readable type name, parser function)
726_FIELD_TYPE: dict[str, tuple[str, Callable[[Any], Any]]] = {
727 "date": ("date", parse_date_value),
728 "departure_time": ("time", parse_time_value),
729 "arrival_time": ("time", parse_time_value),
730 "night_time": ("duration", parse_duration_value),
731 "instrument_time": ("duration", parse_duration_value),
732 "cross_country": ("duration", parse_duration_value),
733 "total_flight_time_check": ("duration", parse_duration_value),
734 "single_pilot_se": ("duration", parse_duration_value),
735 "single_pilot_me": ("duration", parse_duration_value),
736 "multi_pilot": ("duration", parse_duration_value),
737 "function_pic": ("duration", parse_duration_value),
738 "function_copilot": ("duration", parse_duration_value),
739 "function_dual": ("duration", parse_duration_value),
740 "function_instructor": ("duration", parse_duration_value),
741 "landings_day": ("integer", parse_int_value),
742 "landings_night": ("integer", parse_int_value),
743}
745_HINT_SAMPLE_ROWS = 5
748def type_hints(parsed: ParsedFile, mapping: dict[str, str]) -> dict[str, str]:
749 """Return {col: hint_text} for columns where sample data doesn't match the proposed type.
751 Samples up to _HINT_SAMPLE_ROWS rows. Returns a hint when non-empty values are
752 present but any of them fail to parse — indicating a likely mapping mismatch.
753 """
754 col_index = {col: i for i, col in enumerate(parsed.norm_cols)}
755 hints: dict[str, str] = {}
756 for col, target in mapping.items():
757 if target not in _FIELD_TYPE:
758 continue
759 type_name, parser = _FIELD_TYPE[target]
760 idx = col_index.get(col)
761 if idx is None:
762 continue
763 sample = [
764 row[idx]
765 for row in parsed.data_rows[:_HINT_SAMPLE_ROWS]
766 if idx < len(row) and _is_nonempty(row[idx])
767 ]
768 if not sample:
769 continue
770 failed = [v for v in sample if parser(v) is None]
771 if failed:
772 example = str(failed[0])[:30]
773 hints[col] = (
774 f"Sample data doesn't look like a {type_name} (e.g. {example!r})"
775 )
776 return hints
779# ── Preview rows ──────────────────────────────────────────────────────────────
782def preview_rows(
783 parsed: ParsedFile,
784 mapping: dict[str, str],
785 n: int = 5,
786) -> list[dict[str, Any]]:
787 """Return up to *n* non-subtotal data rows mapped to target field names."""
788 date_idx = _date_col_index(parsed.norm_cols, mapping)
789 result: list[dict[str, Any]] = []
790 for row in parsed.data_rows:
791 if _is_subtotal_row(row, date_idx):
792 continue
793 mapped: dict[str, Any] = {}
794 for i, col in enumerate(parsed.norm_cols):
795 target = mapping.get(col, "ignore")
796 if target == "ignore":
797 continue
798 mapped[target] = row[i] if i < len(row) else None
799 result.append(mapped)
800 if len(result) >= n:
801 break
802 return result
805def _date_col_index(norm_cols: list[str], mapping: dict[str, str]) -> int | None:
806 for i, col in enumerate(norm_cols):
807 if mapping.get(col) == "date":
808 return i
809 return None
812# ── Import execution ──────────────────────────────────────────────────────────
815def _row_get(row: list[Any], col_index: dict[str, int], col: str) -> Any:
816 i = col_index.get(col)
817 return row[i] if i is not None and i < len(row) else None
820def _parse_row_date(
821 row: list[Any], mapping: dict[str, str], col_index: dict[str, int]
822) -> date | None:
823 for col, target in mapping.items():
824 if target == "date":
825 return parse_date_value(_row_get(row, col_index, col))
826 return None
829def _build_entry_kwargs(
830 row: list[Any],
831 mapping: dict[str, str],
832 col_index: dict[str, int],
833 date_val: date,
834) -> tuple[dict[str, Any], float | None, list[tuple[str, str, str]]]:
835 """Build Flight (standalone, aircraft_id NULL) field kwargs for one data
836 row (identity/batch fields like pic_user_id are the caller's
837 responsibility to add).
839 Returns (kwargs, source_total_flight_time_check, parse_warnings), where
840 parse_warnings is a list of (col, target, raw_repr) for non-empty cells
841 that couldn't be parsed as their target field's type. Shared by the
842 normal import pass and the near-match conflict finder below, so the two
843 can never compute a row's fields differently.
844 """
845 kwargs: dict[str, Any] = {"date": date_val}
846 source_total: float | None = None
847 parse_warnings: list[tuple[str, str, str]] = []
849 for col, target in mapping.items():
850 if target in ("ignore", "date"):
851 continue
852 raw = _row_get(row, col_index, col)
853 if target == "total_flight_time_check":
854 if _is_nonempty(raw):
855 source_total = parse_duration_value(raw)
856 continue
857 if target in _FIELD_TYPE:
858 # NB: unpack into _type_name, not _ — callers of this module also
859 # call gettext (imported as `_`) in the same function scope;
860 # reassigning `_` would shadow it for the rest of that function
861 # (Python's function-scoping, not block-scoping).
862 _type_name, parser = _FIELD_TYPE[target]
863 parsed_val = parser(raw)
864 if parsed_val is None and _is_nonempty(raw):
865 parse_warnings.append((col, target, repr(str(raw)[:40])))
866 kwargs[target] = parsed_val
867 elif target == "aircraft_type":
868 val = str(raw).strip() if raw is not None else None
869 kwargs["other_aircraft_type"] = val
870 if val:
871 from utils import (
872 resolve_aircraft_type_icao, # pyright: ignore[reportMissingImports]
873 )
875 kwargs["other_aircraft_type_icao"] = resolve_aircraft_type_icao(val)
876 elif target == "aircraft_registration":
877 kwargs["other_aircraft_registration"] = (
878 str(raw).strip() if raw is not None else None
879 )
880 elif target == "departure_place":
881 kwargs["departure_icao"] = str(raw).strip() if raw is not None else None
882 elif target == "arrival_place":
883 kwargs["arrival_icao"] = str(raw).strip() if raw is not None else None
884 elif target == "pic_name":
885 kwargs["pic_name"] = str(raw).strip() if raw is not None else None
886 elif target == "remarks":
887 kwargs["notes"] = str(raw).strip() if raw is not None else None
889 return kwargs, source_total, parse_warnings
892def _num(v: Any) -> float | None:
893 # Numeric columns come back as decimal.Decimal; the freshly parsed side
894 # is always plain float (parse_duration_value). Decimal('0.7') != 0.7
895 # (float) directly — comparing Decimals converted from float literals
896 # hits binary-rounding mismatches — so normalise both sides to float
897 # before building/comparing keys.
898 return None if v is None else float(v)
901def _dup_key(kwargs: dict[str, Any]) -> tuple[Any, ...]:
902 """Exact-duplicate key: date + aircraft + duration + landings. Shared by
903 execute_import (silently skips exact matches) and find_conflicting_rows
904 (must not also flag them as a near-match conflict needing review)."""
905 return (
906 kwargs["date"],
907 kwargs.get("other_aircraft_registration"),
908 kwargs.get("single_pilot_se"),
909 kwargs.get("single_pilot_me"),
910 kwargs.get("multi_pilot"),
911 kwargs.get("landings_day"),
912 kwargs.get("landings_night"),
913 )
916def _fetch_existing_dedup_keys(pilot_user_id: int) -> set[tuple[Any, ...]]:
917 from models import Aircraft, Flight, db # pyright: ignore[reportMissingImports]
918 from sqlalchemy import or_ # pyright: ignore[reportMissingImports]
920 # Registration is Aircraft.registration for a managed-aircraft row, or
921 # the free-text other_aircraft_registration for a standalone one.
922 rows = (
923 db.session.query(
924 Flight.date,
925 Flight.other_aircraft_registration,
926 Aircraft.registration,
927 Flight.single_pilot_se,
928 Flight.single_pilot_me,
929 Flight.multi_pilot,
930 Flight.landings_day,
931 Flight.landings_night,
932 )
933 .outerjoin(Aircraft, Aircraft.id == Flight.aircraft_id)
934 .filter(
935 or_(
936 Flight.pic_user_id == pilot_user_id,
937 Flight.second_crew_user_id == pilot_user_id,
938 )
939 )
940 )
941 return {
942 (
943 row[0],
944 row[2] or row[1],
945 _num(row[3]),
946 _num(row[4]),
947 _num(row[5]),
948 row[6],
949 row[7],
950 )
951 for row in rows
952 }
955def _assign_pilot_identity(kwargs: dict[str, Any], pilot_user_id: int) -> None:
956 """Put *pilot_user_id* into the correct crew slot for one row (mutates
957 *kwargs* in place), based on its function_* breakdown.
959 A personal-logbook row's "PIC name" column (already in kwargs["pic_name"]
960 by the time this runs) records the actual PIC's name as free text,
961 independent of which slot the importing pilot occupies on that row. When
962 the row shows dual-received or copilot time and no PIC time, the
963 importing pilot was not the PIC — they belong in the second_crew_* slot
964 (as student or copilot) instead, leaving pic_user_id unset so pic_name's
965 free-text name remains the row's only PIC identity. Any other case (PIC
966 time logged, instructor time logged, or no function breakdown at all —
967 e.g. a source file with no "Pilot function" column) keeps pic_user_id as
968 the importing pilot, matching this app's original single-slot behaviour.
969 """
970 from models import CrewRole # pyright: ignore[reportMissingImports]
972 function_pic = kwargs.get("function_pic")
973 function_dual = kwargs.get("function_dual")
974 function_copilot = kwargs.get("function_copilot")
976 if not function_pic and function_dual:
977 kwargs["second_crew_user_id"] = pilot_user_id
978 kwargs["second_crew_role"] = CrewRole.STUDENT
979 elif not function_pic and function_copilot:
980 kwargs["second_crew_user_id"] = pilot_user_id
981 kwargs["second_crew_role"] = CrewRole.COPILOT
982 else:
983 kwargs["pic_user_id"] = pilot_user_id
986def execute_import(
987 parsed: ParsedFile,
988 mapping: dict[str, str],
989 pilot_user_id: int,
990 batch_id: int,
991 opening_balance: dict[str, Any] | None = None,
992 skip_row_nums: set[int] | None = None,
993) -> ImportResult:
994 """Create standalone Flight rows (aircraft_id NULL) from *parsed* using
995 *mapping*.
997 Returns an ImportResult describing what happened. Entries are added to
998 db.session but NOT committed — the caller commits after also saving the
999 batch/mapping records. *skip_row_nums* (1-based, matching the row
1000 numbering used throughout this module) lets the caller carve out rows
1001 it's handling separately — e.g. near-match conflicts routed to the
1002 interactive review step in app/pilots/routes.py — so they're excluded
1003 entirely from this pass (not counted as imported, duplicate, or skipped).
1004 """
1005 from models import Flight, db # pyright: ignore[reportMissingImports]
1007 result = ImportResult()
1008 date_idx = _date_col_index(parsed.norm_cols, mapping)
1009 col_index = {col: i for i, col in enumerate(parsed.norm_cols)}
1010 skip_row_nums = skip_row_nums or set()
1012 entries_to_add: list[Flight] = []
1014 # Duplicate detection: re-importing the same file — either by mistake, or
1015 # deliberately after appending new rows to the source spreadsheet, which
1016 # is the expected workflow for keeping an import-based logbook current —
1017 # must only add genuinely new rows, not double up everything already
1018 # imported. Match on date + aircraft + the duration/landings figures
1019 # rather than departure/arrival place & time: those are the fields that
1020 # actually drive currency and compliance totals, and they stay populated
1021 # even for entries whose route/time genuinely isn't captured (e.g. a
1022 # hand-entered or backfilled logbook row) — a key that depended on
1023 # route/time would silently fail to match those.
1024 existing_keys = _fetch_existing_dedup_keys(pilot_user_id)
1026 for row_num, row in enumerate(parsed.data_rows, start=1):
1027 if row_num in skip_row_nums:
1028 continue
1029 if _is_subtotal_row(row, date_idx):
1030 result.subtotals += 1
1031 continue
1033 date_val = _parse_row_date(row, mapping, col_index)
1034 if date_val is None:
1035 raw_date = (
1036 row[date_idx] if date_idx is not None and date_idx < len(row) else None
1037 )
1038 result.skipped.append((row_num, f"unparseable date: {raw_date!r}"))
1039 continue
1041 kwargs, source_total, parse_warnings = _build_entry_kwargs(
1042 row, mapping, col_index, date_val
1043 )
1044 _assign_pilot_identity(kwargs, pilot_user_id)
1045 kwargs["import_batch_id"] = batch_id
1046 kwargs["source"] = "import"
1047 for col, target, raw_repr in parse_warnings:
1048 result.parse_warnings.append((row_num, col, target, raw_repr))
1050 dup_key = _dup_key(kwargs)
1051 if dup_key in existing_keys:
1052 result.duplicates.append(
1053 (
1054 row_num,
1055 _(
1056 "matches an entry already in your logbook "
1057 "(same date, aircraft, duration and landings)"
1058 ),
1059 )
1060 )
1061 continue
1062 existing_keys.add(dup_key)
1064 if source_total is not None:
1065 computed = round(
1066 sum(
1067 float(kwargs.get(f) or 0)
1068 for f in ("single_pilot_se", "single_pilot_me", "multi_pilot")
1069 ),
1070 1,
1071 )
1072 if abs(source_total - computed) >= 0.15:
1073 result.total_mismatch_warnings.append((row_num, source_total, computed))
1075 entries_to_add.append(Flight(**kwargs))
1077 result.imported = len(entries_to_add)
1078 for e in entries_to_add:
1079 db.session.add(e)
1081 # Opening balance — synthetic entry dated one day before earliest imported.
1082 # Only one opening-balance entry ever makes sense per pilot — re-importing
1083 # with opening-balance values filled in again must not create a second one.
1084 if opening_balance and any(v for v in opening_balance.values() if v):
1085 ob_exists = (
1086 db.session.query(Flight.id)
1087 .filter_by(pic_user_id=pilot_user_id, notes="Opening balance (imported)")
1088 .first()
1089 is not None
1090 )
1091 if ob_exists:
1092 result.duplicates.append(
1093 (
1094 0,
1095 _(
1096 "opening balance: an opening balance entry already "
1097 "exists for this pilot — not created again"
1098 ),
1099 )
1100 )
1101 else:
1102 earliest = (
1103 min(e.date for e in entries_to_add) if entries_to_add else date.today()
1104 )
1105 from datetime import timedelta as _td
1107 balance_date = earliest - _td(days=1)
1108 balance_entry = Flight(
1109 pic_user_id=pilot_user_id,
1110 import_batch_id=batch_id,
1111 source="import",
1112 date=balance_date,
1113 notes="Opening balance (imported)",
1114 night_time=opening_balance.get("night_time"),
1115 instrument_time=opening_balance.get("instrument_time"),
1116 single_pilot_se=opening_balance.get("single_pilot_se"),
1117 single_pilot_me=opening_balance.get("single_pilot_me"),
1118 multi_pilot=opening_balance.get("multi_pilot"),
1119 function_pic=opening_balance.get("function_pic"),
1120 function_copilot=opening_balance.get("function_copilot"),
1121 function_dual=opening_balance.get("function_dual"),
1122 function_instructor=opening_balance.get("function_instructor"),
1123 )
1124 db.session.add(balance_entry)
1125 result.has_opening_balance = True
1127 return result
1130# ── Near-match conflict detection (possible corrections) ───────────────────────
1132_CANDIDATE_MIN_SCORE = 3
1133_CANDIDATE_DURATION_TOLERANCE = 1.0
1136@dataclass
1137class ConflictRow:
1138 """A parsed row that isn't an exact duplicate but scores highly enough
1139 against one or more existing entries to plausibly be an edited version
1140 of one of them — needs a human decision, not a guess."""
1142 row_num: int
1143 kwargs: dict[str, Any]
1144 candidates: list[tuple[float, int]] # (score, existing_entry_id), best first
1147def _kwargs_duration(kwargs: dict[str, Any]) -> float | None:
1148 parts = [
1149 kwargs.get(f) for f in ("single_pilot_se", "single_pilot_me", "multi_pilot")
1150 ]
1151 vals = [float(p) for p in parts if p is not None]
1152 return round(sum(vals), 1) if vals else None
1155def _score_candidate(kwargs: dict[str, Any], existing: Any) -> float:
1156 """Score how likely *existing* (a Flight row) is the same real-world
1157 flight as *kwargs* (a freshly parsed row), across 7 points: registration,
1158 departure, arrival, departure time, arrival time, total duration,
1159 landings. A point only counts toward the score if both sides have data
1160 for it — missing data on either side is neutral, never a mismatch, so a
1161 row whose route/time isn't captured on one side can still be recognised
1162 via duration + landings alone. The two time points can be fractional —
1163 see services.time_band_matching.
1164 """
1165 from services.time_band_matching import ( # pyright: ignore[reportMissingImports]
1166 DEFAULT_OFFSET_HOURS,
1167 offset_ring_step_minutes,
1168 shift_time,
1169 time_band_score,
1170 )
1172 score: float = 0.0
1174 reg_new = (kwargs.get("other_aircraft_registration") or "").strip().upper()
1175 reg_old = (existing.display_registration or "").strip().upper()
1176 if reg_new and reg_old and reg_new == reg_old:
1177 score += 1
1179 dep_new = (kwargs.get("departure_icao") or "").strip().upper()
1180 dep_old = (existing.departure_icao or "").strip().upper()
1181 if dep_new and dep_old and dep_new == dep_old:
1182 score += 1
1184 arr_new = (kwargs.get("arrival_icao") or "").strip().upper()
1185 arr_old = (existing.arrival_icao or "").strip().upper()
1186 if arr_new and arr_old and arr_new == arr_old:
1187 score += 1
1189 # Same-pair (existing.departure_time/arrival_time set) is scored tight,
1190 # expecting a near-exact repeat of the same real instant, with a ring
1191 # width of 1/3 of the managed aircraft's flight_counter_offset — pilots
1192 # log to a fixed precision (0.1h/6min at the 0.3h default), so a flat
1193 # few-minutes tolerance would put an ordinary rounding difference in the
1194 # wrong ring. Falls back to the model's own 0.3h default when this row
1195 # has no managed aircraft to read a real offset from. Cross-pair
1196 # (existing only has takeoff_time/landing_time — an airframe-import
1197 # placeholder, e.g. a hand-entered row from before this pilot ever ran
1198 # a logbook import) is scored against that time shifted by this same
1199 # offset, the closest estimate available of the block-to-airborne time
1200 # gap — this needs a *real* managed-aircraft offset, so a standalone
1201 # existing row (no aircraft_id, e.g. a rental logged nowhere else) has
1202 # this comparison skipped entirely rather than guessing one.
1203 offset_hours = (
1204 float(existing.aircraft.flight_counter_offset) if existing.aircraft else None
1205 )
1206 same_pair_step = offset_ring_step_minutes(
1207 offset_hours if offset_hours is not None else DEFAULT_OFFSET_HOURS
1208 )
1210 dep_time_new = kwargs.get("departure_time")
1211 if existing.departure_time is not None:
1212 score += time_band_score(
1213 dep_time_new, (existing.departure_time,), same_pair_step
1214 )
1215 elif existing.takeoff_time is not None and offset_hours is not None:
1216 offset_minutes = round(offset_hours * 60)
1217 center = shift_time(existing.takeoff_time, -offset_minutes)
1218 score += time_band_score(
1219 dep_time_new, (center,), offset_ring_step_minutes(offset_hours)
1220 )
1222 arr_time_new = kwargs.get("arrival_time")
1223 if existing.arrival_time is not None:
1224 score += time_band_score(arr_time_new, (existing.arrival_time,), same_pair_step)
1225 elif existing.landing_time is not None and offset_hours is not None:
1226 offset_minutes = round(offset_hours * 60)
1227 center = shift_time(existing.landing_time, offset_minutes)
1228 score += time_band_score(
1229 arr_time_new, (center,), offset_ring_step_minutes(offset_hours)
1230 )
1232 dur_new = _kwargs_duration(kwargs)
1233 dur_old = existing.total_flight_time
1234 if (
1235 dur_new is not None
1236 and dur_old is not None
1237 and abs(dur_new - dur_old) <= _CANDIDATE_DURATION_TOLERANCE
1238 ):
1239 score += 1
1241 land_day = kwargs.get("landings_day")
1242 land_night = kwargs.get("landings_night")
1243 if (
1244 land_day is not None
1245 and land_night is not None
1246 and existing.landings_day is not None
1247 and existing.landings_night is not None
1248 and land_day == existing.landings_day
1249 and land_night == existing.landings_night
1250 ):
1251 score += 1
1253 return score
1256def find_conflicting_rows(
1257 parsed: ParsedFile,
1258 mapping: dict[str, str],
1259 pilot_user_id: int,
1260 exclude_row_nums: set[int] | None = None,
1261) -> list[ConflictRow]:
1262 """Find rows that aren't an exact duplicate but plausibly match an
1263 existing entry closely enough (score >= _CANDIDATE_MIN_SCORE) to need a
1264 human decision: keep the existing entry, overwrite it with the new data,
1265 or import as a genuinely separate new entry.
1267 Skips subtotal rows, rows with an unparseable date, and exact duplicates
1268 (same as execute_import's own dedup — an unmodified re-upload scores the
1269 maximum on every point, so without this check it would incorrectly be
1270 routed to review instead of being silently skipped), plus any row_num in
1271 *exclude_row_nums* — typically rows already resolved in an earlier pass
1272 of this same review.
1274 Candidates also include *unclaimed* rows (pic_user_id AND
1275 second_crew_user_id both NULL) on a managed aircraft belonging to one of
1276 the pilot's own tenants — typically a hand-entered airframe-log row from
1277 before this pilot ever ran a logbook import. Without this, those rows
1278 are invisible to both this check and execute_import's own dedup (both
1279 only look at rows already linked to *pilot_user_id*), so re-importing a
1280 personal logbook can never discover that the real-world flight is
1281 already logged from the airframe side — it silently creates a second,
1282 duplicate Flight row instead of surfacing a conflict to resolve.
1283 """
1284 from models import ( # pyright: ignore[reportMissingImports]
1285 Aircraft,
1286 Flight,
1287 TenantUser,
1288 )
1289 from sqlalchemy import and_, or_ # pyright: ignore[reportMissingImports]
1291 exclude_row_nums = exclude_row_nums or set()
1292 date_idx = _date_col_index(parsed.norm_cols, mapping)
1293 col_index = {col: i for i, col in enumerate(parsed.norm_cols)}
1294 existing_keys = _fetch_existing_dedup_keys(pilot_user_id)
1295 conflicts: list[ConflictRow] = []
1297 tenant_ids = [
1298 row.tenant_id for row in TenantUser.query.filter_by(user_id=pilot_user_id).all()
1299 ]
1301 for row_num, row in enumerate(parsed.data_rows, start=1):
1302 if row_num in exclude_row_nums:
1303 continue
1304 if _is_subtotal_row(row, date_idx):
1305 continue
1306 date_val = _parse_row_date(row, mapping, col_index)
1307 if date_val is None:
1308 continue
1310 kwargs, _source_total, _parse_warnings = _build_entry_kwargs(
1311 row, mapping, col_index, date_val
1312 )
1313 if _dup_key(kwargs) in existing_keys:
1314 continue # exact duplicate — execute_import's own dedup handles this
1316 same_day_filters = [
1317 Flight.pic_user_id == pilot_user_id,
1318 Flight.second_crew_user_id == pilot_user_id,
1319 ]
1320 if tenant_ids:
1321 same_day_filters.append(
1322 and_(
1323 Flight.pic_user_id.is_(None),
1324 Flight.second_crew_user_id.is_(None),
1325 Flight.aircraft_id.in_(
1326 Aircraft.query.with_entities(Aircraft.id).filter(
1327 Aircraft.tenant_id.in_(tenant_ids)
1328 )
1329 ),
1330 )
1331 )
1332 same_day = Flight.query.filter(
1333 or_(*same_day_filters),
1334 Flight.date == date_val,
1335 ).all()
1336 scored = [
1337 (score, existing.id)
1338 for existing in same_day
1339 if (score := _score_candidate(kwargs, existing)) >= _CANDIDATE_MIN_SCORE
1340 ]
1341 if scored:
1342 scored.sort(key=lambda t: -t[0])
1343 conflicts.append(
1344 ConflictRow(row_num=row_num, kwargs=kwargs, candidates=scored)
1345 )
1347 return conflicts
1350def link_entries_to_aircraft(entries: list[Any]) -> int:
1351 """Promote each standalone Flight row (aircraft_id NULL,
1352 other_aircraft_registration set) in *entries* to a managed aircraft, for
1353 any whose registration matches an Aircraft belonging to one of the row's
1354 pic/second-crew pilot's own tenants. Returns the count promoted. Caller
1355 must commit.
1357 Unified-model note: this used to create a brand-new FlightEntry +
1358 FlightCrew and link the pilot's existing PilotLogbookEntry to it via
1359 flight_id. There's only one row now, so promotion is just updating
1360 aircraft_id (+ clearing the free-text other_aircraft_* fields) on the
1361 row that already exists — no second row, no separate crew row (the
1362 row's pic_user_id/pic_name are already whatever the CSV import set).
1363 """
1364 from models import ( # pyright: ignore[reportMissingImports]
1365 Aircraft,
1366 TenantUser,
1367 User,
1368 db,
1369 )
1371 def _norm_reg(reg: str) -> str:
1372 return reg.upper().replace("-", "").replace(" ", "")
1374 ac_by_tenant: dict[int, dict[str, Any]] = {}
1375 for ac in Aircraft.query.all():
1376 ac_by_tenant.setdefault(ac.tenant_id, {})[_norm_reg(ac.registration)] = ac
1378 pilot_tenant_ids: dict[int, set[int]] = {}
1380 def _tenants_for_pilot(pilot_user_id: int) -> set[int]:
1381 if pilot_user_id not in pilot_tenant_ids:
1382 pilot_tenant_ids[pilot_user_id] = {
1383 row.tenant_id
1384 for row in TenantUser.query.filter_by(user_id=pilot_user_id).all()
1385 }
1386 return pilot_tenant_ids[pilot_user_id]
1388 def _place_icao(place: str | None) -> str:
1389 if not place:
1390 return "ZZZZ"
1391 clean = re.sub(r"[^A-Z0-9]", "", place.upper())[:4]
1392 return clean if len(clean) == 4 else "ZZZZ"
1394 promoted = 0
1395 for entry in entries:
1396 if entry.aircraft_id is not None or not entry.other_aircraft_registration:
1397 continue
1398 pilot_user_id = entry.pic_user_id or entry.second_crew_user_id
1399 if pilot_user_id is None:
1400 continue
1401 norm_reg = _norm_reg(entry.other_aircraft_registration)
1402 ac = None
1403 for tid in _tenants_for_pilot(pilot_user_id):
1404 ac = ac_by_tenant.get(tid, {}).get(norm_reg)
1405 if ac is not None:
1406 break
1407 if ac is None:
1408 continue
1410 entry.aircraft_id = ac.id
1411 entry.departure_icao = _place_icao(entry.departure_icao)
1412 entry.arrival_icao = _place_icao(entry.arrival_icao)
1413 entry.other_aircraft_type = None
1414 entry.other_aircraft_type_icao = None
1415 entry.other_aircraft_registration = None
1416 entry.source = "logbook_import"
1418 # Step 2 (docs/backlog.md "reconcile imports from either side"): the
1419 # true airframe-side fields (flight_time_counter_*/flight_time,
1420 # engine_time_counter_*/engine_time — the maintenance-relevant hour
1421 # figures) aren't known from a personal logbook and stay NULL, same
1422 # as everywhere else in this app "if a field is not filled, it's not
1423 # filled" — no supposition, interpolation, or guessed value gets
1424 # invented here. That's also what keeps the flight list's
1425 # needs-attention icon (source == "logbook_import" and flight_time
1426 # is None) lit until a human fills these in from a real airframe
1427 # source. (A previous version of this function seeded a rough
1428 # engine_time_counter estimate here, on the assumption that nothing
1429 # read it for maintenance tracking — services/component_limits.py
1430 # now sums engine_time_counter for engine/propeller TBO, so that
1431 # assumption no longer holds and the guess was removed.)
1433 if not entry.pic_name:
1434 pilot_user = db.session.get(User, pilot_user_id)
1435 if pilot_user:
1436 entry.pic_name = pilot_user.name or pilot_user.email
1438 promoted += 1
1440 return promoted