Coverage for app/pilots/logbook_import.py: 100%
702 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 15:29 +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 :]
535 # Spreadsheets often carry a tail of formatted-but-empty rows; they'd
536 # otherwise each be reported as an ignored "subtotal" row. Only the
537 # trailing run is dropped, so row numbers stay those of the file.
538 while data_rows and not any(_is_nonempty(v) for v in data_rows[-1]):
539 data_rows.pop()
541 return ParsedFile(
542 norm_cols=norm_cols,
543 raw_cols=raw_header,
544 header_row_index=header_idx,
545 data_rows=data_rows,
546 fingerprint=_fingerprint(norm_cols),
547 )
550# ── Mapping proposal ──────────────────────────────────────────────────────────
553def propose_mapping(
554 parsed: ParsedFile,
555 existing_mappings: list[Any], # list[LogbookImportMapping]
556) -> MappingProposal:
557 """Return the best mapping proposal for *parsed*, checking saved mappings first."""
558 # 1. Exact fingerprint match
559 for m in existing_mappings:
560 if m.source_fingerprint == parsed.fingerprint:
561 return MappingProposal(
562 mapping=json.loads(m.column_mapping),
563 match_type="exact",
564 matched_mapping_id=m.id,
565 )
567 # 2. Fuzzy match — best overlap among saved mappings
568 best_score = 0.0
569 best_m = None
570 new_set = set(parsed.norm_cols)
571 for m in existing_mappings:
572 saved_cols: list[str] = json.loads(m.source_columns)
573 saved_set = set(saved_cols)
574 if not saved_set:
575 continue
576 overlap = len(new_set & saved_set)
577 score = overlap / max(len(new_set), len(saved_set))
578 if score > best_score:
579 best_score = score
580 best_m = m
582 if best_m is not None and best_score >= 0.6:
583 saved_map: dict[str, str] = json.loads(best_m.column_mapping)
584 # Build a mapping for the new columns, falling back to alias for unmatched ones
585 merged = _alias_mapping(parsed.norm_cols)
586 for col in parsed.norm_cols:
587 if col in saved_map:
588 merged[col] = saved_map[col]
589 return MappingProposal(
590 mapping=merged,
591 match_type="fuzzy",
592 fuzzy_score=best_score,
593 matched_mapping_id=best_m.id,
594 )
596 # 3. Alias-only auto-mapping
597 return MappingProposal(
598 mapping=_alias_mapping(parsed.norm_cols),
599 match_type="alias",
600 )
603def _alias_mapping(norm_cols: list[str]) -> dict[str, str]:
604 """Apply the built-in alias table; unknown columns default to 'ignore'."""
605 result: dict[str, str] = {}
606 for col in norm_cols:
607 result[col] = _ALIASES.get(col, "ignore")
608 return result
611# ── Subtotal detection ────────────────────────────────────────────────────────
614def _is_subtotal_row(row: list[Any], date_col_idx: int | None) -> bool:
615 if date_col_idx is None:
616 return False
617 if date_col_idx >= len(row):
618 return True
619 val = row[date_col_idx]
620 if isinstance(val, timedelta):
621 return True
622 if val is None or (isinstance(val, str) and not val.strip()):
623 return True
624 return bool(isinstance(val, str) and "total" in val.lower())
627# ── Value parsing ─────────────────────────────────────────────────────────────
630def parse_date_value(val: Any) -> date | None:
631 if isinstance(val, datetime):
632 return val.date()
633 if isinstance(val, date):
634 return val
635 if isinstance(val, (int, float)):
636 # Excel serial date number — openpyxl returns datetime; guard anyway
637 return None
638 if not isinstance(val, str):
639 return None
640 s = val.strip()
641 if not s:
642 return None
643 for fmt in ("%d/%m/%y", "%d/%m/%Y", "%Y-%m-%d", "%m/%d/%Y", "%d-%m-%Y"):
644 try:
645 return datetime.strptime(s, fmt).date()
646 except ValueError:
647 pass
648 return None
651def parse_time_value(val: Any) -> time | None:
652 """Parse a time-of-day value (HH:MM or Python time)."""
653 if isinstance(val, time):
654 return val
655 if isinstance(val, datetime):
656 return val.time()
657 if not isinstance(val, str):
658 return None
659 s = val.strip()
660 if not s:
661 return None
662 m = re.match(r"^(\d{1,2}):(\d{2})$", s)
663 if m:
664 try:
665 return time(int(m.group(1)), int(m.group(2)))
666 except ValueError:
667 return None
668 return None
671def parse_duration_value(val: Any) -> float | None:
672 """Parse a duration into decimal hours (e.g. time(0,42) → 0.7, '1:24' → 1.4)."""
673 if isinstance(val, timedelta):
674 hours = val.total_seconds() / 3600
675 return round(hours, 1) if hours >= 0 else None
676 if isinstance(val, time):
677 return round(val.hour + val.minute / 60, 1)
678 if isinstance(val, datetime):
679 # Excel sometimes returns a dummy date + the time-of-day
680 t = val.time()
681 return round(t.hour + t.minute / 60, 1)
682 if isinstance(val, (int, float)):
683 if not math.isfinite(val) or val < 0:
684 return None
685 return round(float(val), 1)
686 if isinstance(val, str):
687 s = val.strip()
688 if not s:
689 return None
690 m = re.match(r"^(\d+):(\d{2})$", s)
691 if m:
692 return round(int(m.group(1)) + int(m.group(2)) / 60, 1)
693 try:
694 v = float(s)
695 return round(v, 1) if math.isfinite(v) and v >= 0 else None
696 except ValueError:
697 return None
698 return None
701def parse_int_value(val: Any) -> int | None:
702 if isinstance(val, int):
703 return val if val >= 0 else None
704 if isinstance(val, float):
705 try:
706 return int(val) if val >= 0 else None
707 except (ValueError, OverflowError):
708 return None
709 if isinstance(val, str):
710 s = val.strip()
711 if not s:
712 return None
713 try:
714 n = int(float(s))
715 except (ValueError, OverflowError):
716 return None
717 return n if n >= 0 else None
718 return None
721def _is_nonempty(val: Any) -> bool:
722 """Return True when val carries a real value (not None, not blank string)."""
723 if val is None:
724 return False
725 if isinstance(val, str):
726 return bool(val.strip())
727 return True
730# Map target field → (human-readable type name, parser function)
731_FIELD_TYPE: dict[str, tuple[str, Callable[[Any], Any]]] = {
732 "date": ("date", parse_date_value),
733 "departure_time": ("time", parse_time_value),
734 "arrival_time": ("time", parse_time_value),
735 "night_time": ("duration", parse_duration_value),
736 "instrument_time": ("duration", parse_duration_value),
737 "cross_country": ("duration", parse_duration_value),
738 "total_flight_time_check": ("duration", parse_duration_value),
739 "single_pilot_se": ("duration", parse_duration_value),
740 "single_pilot_me": ("duration", parse_duration_value),
741 "multi_pilot": ("duration", parse_duration_value),
742 "function_pic": ("duration", parse_duration_value),
743 "function_copilot": ("duration", parse_duration_value),
744 "function_dual": ("duration", parse_duration_value),
745 "function_instructor": ("duration", parse_duration_value),
746 "landings_day": ("integer", parse_int_value),
747 "landings_night": ("integer", parse_int_value),
748}
750_HINT_SAMPLE_ROWS = 5
753def type_hints(parsed: ParsedFile, mapping: dict[str, str]) -> dict[str, str]:
754 """Return {col: hint_text} for columns where sample data doesn't match the proposed type.
756 Samples up to _HINT_SAMPLE_ROWS rows. Returns a hint when non-empty values are
757 present but any of them fail to parse — indicating a likely mapping mismatch.
758 """
759 col_index = {col: i for i, col in enumerate(parsed.norm_cols)}
760 hints: dict[str, str] = {}
761 for col, target in mapping.items():
762 if target not in _FIELD_TYPE:
763 continue
764 type_name, parser = _FIELD_TYPE[target]
765 idx = col_index.get(col)
766 if idx is None:
767 continue
768 sample = [
769 row[idx]
770 for row in parsed.data_rows[:_HINT_SAMPLE_ROWS]
771 if idx < len(row) and _is_nonempty(row[idx])
772 ]
773 if not sample:
774 continue
775 failed = [v for v in sample if parser(v) is None]
776 if failed:
777 example = str(failed[0])[:30]
778 hints[col] = (
779 f"Sample data doesn't look like a {type_name} (e.g. {example!r})"
780 )
781 return hints
784# ── Preview rows ──────────────────────────────────────────────────────────────
787def preview_rows(
788 parsed: ParsedFile,
789 mapping: dict[str, str],
790 n: int = 5,
791) -> list[dict[str, Any]]:
792 """Return up to *n* non-subtotal data rows mapped to target field names."""
793 date_idx = _date_col_index(parsed.norm_cols, mapping)
794 result: list[dict[str, Any]] = []
795 for row in parsed.data_rows:
796 if _is_subtotal_row(row, date_idx):
797 continue
798 mapped: dict[str, Any] = {}
799 for i, col in enumerate(parsed.norm_cols):
800 target = mapping.get(col, "ignore")
801 if target == "ignore":
802 continue
803 mapped[target] = row[i] if i < len(row) else None
804 result.append(mapped)
805 if len(result) >= n:
806 break
807 return result
810def _date_col_index(norm_cols: list[str], mapping: dict[str, str]) -> int | None:
811 for i, col in enumerate(norm_cols):
812 if mapping.get(col) == "date":
813 return i
814 return None
817# ── Import execution ──────────────────────────────────────────────────────────
820def _row_get(row: list[Any], col_index: dict[str, int], col: str) -> Any:
821 i = col_index.get(col)
822 return row[i] if i is not None and i < len(row) else None
825def _parse_row_date(
826 row: list[Any], mapping: dict[str, str], col_index: dict[str, int]
827) -> date | None:
828 for col, target in mapping.items():
829 if target == "date":
830 return parse_date_value(_row_get(row, col_index, col))
831 return None
834def _build_entry_kwargs(
835 row: list[Any],
836 mapping: dict[str, str],
837 col_index: dict[str, int],
838 date_val: date,
839) -> tuple[dict[str, Any], float | None, list[tuple[str, str, str]]]:
840 """Build Flight (standalone, aircraft_id NULL) field kwargs for one data
841 row (identity/batch fields like pic_user_id are the caller's
842 responsibility to add).
844 Returns (kwargs, source_total_flight_time_check, parse_warnings), where
845 parse_warnings is a list of (col, target, raw_repr) for non-empty cells
846 that couldn't be parsed as their target field's type. Shared by the
847 normal import pass and the near-match conflict finder below, so the two
848 can never compute a row's fields differently.
849 """
850 kwargs: dict[str, Any] = {"date": date_val}
851 source_total: float | None = None
852 parse_warnings: list[tuple[str, str, str]] = []
854 for col, target in mapping.items():
855 if target in ("ignore", "date"):
856 continue
857 raw = _row_get(row, col_index, col)
858 if target == "total_flight_time_check":
859 if _is_nonempty(raw):
860 source_total = parse_duration_value(raw)
861 continue
862 if target in _FIELD_TYPE:
863 # NB: unpack into _type_name, not _ — callers of this module also
864 # call gettext (imported as `_`) in the same function scope;
865 # reassigning `_` would shadow it for the rest of that function
866 # (Python's function-scoping, not block-scoping).
867 _type_name, parser = _FIELD_TYPE[target]
868 parsed_val = parser(raw)
869 if parsed_val is None and _is_nonempty(raw):
870 parse_warnings.append((col, target, repr(str(raw)[:40])))
871 kwargs[target] = parsed_val
872 elif target == "aircraft_type":
873 val = str(raw).strip() if raw is not None else None
874 kwargs["other_aircraft_type"] = val
875 if val:
876 from utils import (
877 resolve_aircraft_type_icao, # pyright: ignore[reportMissingImports]
878 )
880 kwargs["other_aircraft_type_icao"] = resolve_aircraft_type_icao(val)
881 elif target == "aircraft_registration":
882 kwargs["other_aircraft_registration"] = (
883 str(raw).strip() if raw is not None else None
884 )
885 elif target == "departure_place":
886 kwargs["departure_icao"] = str(raw).strip() if raw is not None else None
887 elif target == "arrival_place":
888 kwargs["arrival_icao"] = str(raw).strip() if raw is not None else None
889 elif target == "pic_name":
890 kwargs["pic_name"] = str(raw).strip() if raw is not None else None
891 elif target == "remarks":
892 kwargs["notes"] = str(raw).strip() if raw is not None else None
894 return kwargs, source_total, parse_warnings
897def _num(v: Any) -> float | None:
898 # Numeric columns come back as decimal.Decimal; the freshly parsed side
899 # is always plain float (parse_duration_value). Decimal('0.7') != 0.7
900 # (float) directly — comparing Decimals converted from float literals
901 # hits binary-rounding mismatches — so normalise both sides to float
902 # before building/comparing keys.
903 return None if v is None else float(v)
906def _dup_key(kwargs: dict[str, Any]) -> tuple[Any, ...]:
907 """Coarse exact-duplicate key: date + aircraft + duration + landings.
908 Necessary but not sufficient for a duplicate — see DedupIndex, which
909 additionally requires the route and departure time to be compatible.
910 Shared by execute_import (silently skips exact matches) and
911 find_conflicting_rows (must not also flag them as a near-match conflict
912 needing review)."""
913 return (
914 kwargs["date"],
915 kwargs.get("other_aircraft_registration"),
916 kwargs.get("single_pilot_se"),
917 kwargs.get("single_pilot_me"),
918 kwargs.get("multi_pilot"),
919 kwargs.get("landings_day"),
920 kwargs.get("landings_night"),
921 )
924def _place_key(place: str | None) -> str | None:
925 """Normalise a departure/arrival place for duplicate matching — the same
926 4-char cleanup link_entries_to_aircraft applies when promoting a row to
927 a managed aircraft, so a promoted row still matches its source row.
928 None (unknown) for anything that doesn't reduce to a 4-char code."""
929 if not place:
930 return None
931 clean = re.sub(r"[^A-Z0-9]", "", place.upper())[:4]
932 return clean if len(clean) == 4 and clean != "ZZZZ" else None
935def _dup_detail(
936 departure: str | None, arrival: str | None, departure_time: time | None
937) -> tuple[str | None, str | None, time | None]:
938 return (_place_key(departure), _place_key(arrival), departure_time)
941def _details_compatible(a: tuple[Any, ...], b: tuple[Any, ...]) -> bool:
942 """Equal wherever both sides know the value — a missing route or time on
943 either side is neutral, never a mismatch."""
944 return all(x is None or y is None or x == y for x, y in zip(a, b, strict=True))
947class DedupIndex:
948 """Exact-duplicate lookup shared by the pilot and airframe logbook
949 imports: a row is a duplicate of an existing entry when their coarse
950 key matches AND their details are compatible (see _details_compatible)
951 — for the pilot import, departure, arrival and departure time; for the
952 airframe import, takeoff/landing time and counter readings.
954 The coarse key alone is not enough: two genuinely different flights on
955 the same day in the same aircraft routinely share duration and landings
956 (an out-and-back pair of 0.5 h legs, or a day of identical training
957 circuits), and keying on it alone silently dropped them as
958 "duplicates" — or matched a row to the wrong flight of that day. The
959 details can't simply be added to the key either, since entries where
960 they genuinely aren't captured (a hand-entered or backfilled row) would
961 then never match at all.
962 """
964 def __init__(self) -> None:
965 self._by_key: dict[tuple[Any, ...], list[tuple[Any, ...]]] = {}
967 def add(self, key: tuple[Any, ...], detail: tuple[Any, ...]) -> None:
968 self._by_key.setdefault(key, []).append(detail)
970 def contains(self, key: tuple[Any, ...], detail: tuple[Any, ...]) -> bool:
971 return any(
972 _details_compatible(detail, other) for other in self._by_key.get(key, [])
973 )
976def _add_kwargs(index: DedupIndex, kwargs: dict[str, Any]) -> None:
977 index.add(_dup_key(kwargs), _kwargs_dup_detail(kwargs))
980def _contains_kwargs(index: DedupIndex, kwargs: dict[str, Any]) -> bool:
981 return index.contains(_dup_key(kwargs), _kwargs_dup_detail(kwargs))
984def _kwargs_dup_detail(kwargs: dict[str, Any]) -> tuple[Any, ...]:
985 return _dup_detail(
986 kwargs.get("departure_icao"),
987 kwargs.get("arrival_icao"),
988 kwargs.get("departure_time"),
989 )
992def _fetch_existing_dedup_index(pilot_user_id: int) -> DedupIndex:
993 from models import Aircraft, Flight, db # pyright: ignore[reportMissingImports]
994 from sqlalchemy import or_ # pyright: ignore[reportMissingImports]
996 # Registration is Aircraft.registration for a managed-aircraft row, or
997 # the free-text other_aircraft_registration for a standalone one.
998 rows = (
999 db.session.query(
1000 Flight.date,
1001 Flight.other_aircraft_registration,
1002 Aircraft.registration,
1003 Flight.single_pilot_se,
1004 Flight.single_pilot_me,
1005 Flight.multi_pilot,
1006 Flight.landings_day,
1007 Flight.landings_night,
1008 Flight.departure_icao,
1009 Flight.arrival_icao,
1010 Flight.departure_time,
1011 )
1012 .outerjoin(Aircraft, Aircraft.id == Flight.aircraft_id)
1013 .filter(
1014 or_(
1015 Flight.pic_user_id == pilot_user_id,
1016 Flight.second_crew_user_id == pilot_user_id,
1017 )
1018 )
1019 )
1020 index = DedupIndex()
1021 for row in rows:
1022 index.add(
1023 (
1024 row[0],
1025 row[2] or row[1],
1026 _num(row[3]),
1027 _num(row[4]),
1028 _num(row[5]),
1029 row[6],
1030 row[7],
1031 ),
1032 _dup_detail(row[8], row[9], row[10]),
1033 )
1034 return index
1037def _assign_pilot_identity(kwargs: dict[str, Any], pilot_user_id: int) -> None:
1038 """Put *pilot_user_id* into the correct crew slot for one row (mutates
1039 *kwargs* in place), based on its function_* breakdown.
1041 A personal-logbook row's "PIC name" column (already in kwargs["pic_name"]
1042 by the time this runs) records the actual PIC's name as free text,
1043 independent of which slot the importing pilot occupies on that row. When
1044 the row shows dual-received or copilot time and no PIC time, the
1045 importing pilot was not the PIC — they belong in the second_crew_* slot
1046 (as student or copilot) instead, leaving pic_user_id unset so pic_name's
1047 free-text name remains the row's only PIC identity. Any other case (PIC
1048 time logged, instructor time logged, or no function breakdown at all —
1049 e.g. a source file with no "Pilot function" column) keeps pic_user_id as
1050 the importing pilot, matching this app's original single-slot behaviour.
1051 """
1052 from models import CrewRole # pyright: ignore[reportMissingImports]
1054 function_pic = kwargs.get("function_pic")
1055 function_dual = kwargs.get("function_dual")
1056 function_copilot = kwargs.get("function_copilot")
1058 if not function_pic and function_dual:
1059 kwargs["second_crew_user_id"] = pilot_user_id
1060 kwargs["second_crew_role"] = CrewRole.STUDENT
1061 elif not function_pic and function_copilot:
1062 kwargs["second_crew_user_id"] = pilot_user_id
1063 kwargs["second_crew_role"] = CrewRole.COPILOT
1064 else:
1065 kwargs["pic_user_id"] = pilot_user_id
1068def execute_import(
1069 parsed: ParsedFile,
1070 mapping: dict[str, str],
1071 pilot_user_id: int,
1072 batch_id: int,
1073 opening_balance: dict[str, Any] | None = None,
1074 skip_row_nums: set[int] | None = None,
1075) -> ImportResult:
1076 """Create standalone Flight rows (aircraft_id NULL) from *parsed* using
1077 *mapping*.
1079 Returns an ImportResult describing what happened. Entries are added to
1080 db.session but NOT committed — the caller commits after also saving the
1081 batch/mapping records. *skip_row_nums* (1-based, matching the row
1082 numbering used throughout this module) lets the caller carve out rows
1083 it's handling separately — e.g. near-match conflicts routed to the
1084 interactive review step in app/pilots/routes.py — so they're excluded
1085 entirely from this pass (not counted as imported, duplicate, or skipped).
1086 """
1087 from models import Flight, db # pyright: ignore[reportMissingImports]
1089 result = ImportResult()
1090 date_idx = _date_col_index(parsed.norm_cols, mapping)
1091 col_index = {col: i for i, col in enumerate(parsed.norm_cols)}
1092 skip_row_nums = skip_row_nums or set()
1094 entries_to_add: list[Flight] = []
1096 # Duplicate detection: re-importing the same file — either by mistake, or
1097 # deliberately after appending new rows to the source spreadsheet, which
1098 # is the expected workflow for keeping an import-based logbook current —
1099 # must only add genuinely new rows, not double up everything already
1100 # imported. Match on date + aircraft + the duration/landings figures,
1101 # with departure/arrival place & time only required to agree where both
1102 # sides actually record them — see DedupIndex.
1103 existing = _fetch_existing_dedup_index(pilot_user_id)
1105 for row_num, row in enumerate(parsed.data_rows, start=1):
1106 if row_num in skip_row_nums:
1107 continue
1108 if _is_subtotal_row(row, date_idx):
1109 result.subtotals += 1
1110 continue
1112 date_val = _parse_row_date(row, mapping, col_index)
1113 if date_val is None:
1114 raw_date = (
1115 row[date_idx] if date_idx is not None and date_idx < len(row) else None
1116 )
1117 result.skipped.append((row_num, f"unparseable date: {raw_date!r}"))
1118 continue
1120 kwargs, source_total, parse_warnings = _build_entry_kwargs(
1121 row, mapping, col_index, date_val
1122 )
1123 _assign_pilot_identity(kwargs, pilot_user_id)
1124 kwargs["import_batch_id"] = batch_id
1125 kwargs["source"] = "import"
1126 for col, target, raw_repr in parse_warnings:
1127 result.parse_warnings.append((row_num, col, target, raw_repr))
1129 if _contains_kwargs(existing, kwargs):
1130 result.duplicates.append(
1131 (
1132 row_num,
1133 _(
1134 "matches an entry already in your logbook "
1135 "(same date, aircraft, duration and landings)"
1136 ),
1137 )
1138 )
1139 continue
1140 _add_kwargs(existing, kwargs)
1142 if source_total is not None:
1143 computed = round(
1144 sum(
1145 float(kwargs.get(f) or 0)
1146 for f in ("single_pilot_se", "single_pilot_me", "multi_pilot")
1147 ),
1148 1,
1149 )
1150 if abs(source_total - computed) >= 0.15:
1151 result.total_mismatch_warnings.append((row_num, source_total, computed))
1153 entries_to_add.append(Flight(**kwargs))
1155 result.imported = len(entries_to_add)
1156 for e in entries_to_add:
1157 db.session.add(e)
1159 # Opening balance — synthetic entry dated one day before earliest imported.
1160 # Only one opening-balance entry ever makes sense per pilot — re-importing
1161 # with opening-balance values filled in again must not create a second one.
1162 if opening_balance and any(v for v in opening_balance.values() if v):
1163 ob_exists = (
1164 db.session.query(Flight.id)
1165 .filter_by(pic_user_id=pilot_user_id, notes="Opening balance (imported)")
1166 .first()
1167 is not None
1168 )
1169 if ob_exists:
1170 result.duplicates.append(
1171 (
1172 0,
1173 _(
1174 "opening balance: an opening balance entry already "
1175 "exists for this pilot — not created again"
1176 ),
1177 )
1178 )
1179 else:
1180 earliest = (
1181 min(e.date for e in entries_to_add) if entries_to_add else date.today()
1182 )
1183 from datetime import timedelta as _td
1185 balance_date = earliest - _td(days=1)
1186 balance_entry = Flight(
1187 pic_user_id=pilot_user_id,
1188 import_batch_id=batch_id,
1189 source="import",
1190 date=balance_date,
1191 notes="Opening balance (imported)",
1192 night_time=opening_balance.get("night_time"),
1193 instrument_time=opening_balance.get("instrument_time"),
1194 single_pilot_se=opening_balance.get("single_pilot_se"),
1195 single_pilot_me=opening_balance.get("single_pilot_me"),
1196 multi_pilot=opening_balance.get("multi_pilot"),
1197 function_pic=opening_balance.get("function_pic"),
1198 function_copilot=opening_balance.get("function_copilot"),
1199 function_dual=opening_balance.get("function_dual"),
1200 function_instructor=opening_balance.get("function_instructor"),
1201 )
1202 db.session.add(balance_entry)
1203 result.has_opening_balance = True
1205 return result
1208# ── Near-match conflict detection (possible corrections) ───────────────────────
1210_CANDIDATE_MIN_SCORE = 3
1211_CANDIDATE_DURATION_TOLERANCE = 1.0
1214@dataclass
1215class ConflictRow:
1216 """A parsed row that isn't an exact duplicate but scores highly enough
1217 against one or more existing entries to plausibly be an edited version
1218 of one of them — needs a human decision, not a guess."""
1220 row_num: int
1221 kwargs: dict[str, Any]
1222 candidates: list[tuple[float, int]] # (score, existing_entry_id), best first
1225def _kwargs_duration(kwargs: dict[str, Any]) -> float | None:
1226 parts = [
1227 kwargs.get(f) for f in ("single_pilot_se", "single_pilot_me", "multi_pilot")
1228 ]
1229 vals = [float(p) for p in parts if p is not None]
1230 return round(sum(vals), 1) if vals else None
1233def _score_candidate(kwargs: dict[str, Any], existing: Any) -> float:
1234 """Score how likely *existing* (a Flight row) is the same real-world
1235 flight as *kwargs* (a freshly parsed row), across 7 points: registration,
1236 departure, arrival, departure time, arrival time, total duration,
1237 landings. A point only counts toward the score if both sides have data
1238 for it — missing data on either side is neutral, never a mismatch, so a
1239 row whose route/time isn't captured on one side can still be recognised
1240 via duration + landings alone. The two time points can be fractional —
1241 see services.time_band_matching.
1242 """
1243 from services.time_band_matching import ( # pyright: ignore[reportMissingImports]
1244 DEFAULT_OFFSET_HOURS,
1245 offset_ring_step_minutes,
1246 taxi_in_score,
1247 taxi_out_score,
1248 time_band_score,
1249 )
1251 score: float = 0.0
1253 reg_new = (kwargs.get("other_aircraft_registration") or "").strip().upper()
1254 reg_old = (existing.display_registration or "").strip().upper()
1255 if reg_new and reg_old and reg_new == reg_old:
1256 score += 1
1258 dep_new = (kwargs.get("departure_icao") or "").strip().upper()
1259 dep_old = (existing.departure_icao or "").strip().upper()
1260 if dep_new and dep_old and dep_new == dep_old:
1261 score += 1
1263 arr_new = (kwargs.get("arrival_icao") or "").strip().upper()
1264 arr_old = (existing.arrival_icao or "").strip().upper()
1265 if arr_new and arr_old and arr_new == arr_old:
1266 score += 1
1268 # Same-pair (existing.departure_time/arrival_time set) is scored tight,
1269 # expecting a near-exact repeat of the same real instant, with a ring
1270 # width of 1/3 of the managed aircraft's flight_counter_offset — pilots
1271 # log to a fixed precision (0.1h/6min at the 0.3h default), so a flat
1272 # few-minutes tolerance would put an ordinary rounding difference in the
1273 # wrong ring. Falls back to the model's own 0.3h default when this row
1274 # has no managed aircraft to read a real offset from. Cross-pair
1275 # (existing only has takeoff_time/landing_time — an airframe-import
1276 # placeholder, e.g. a hand-entered row from before this pilot ever ran
1277 # a logbook import) is scored by whether this time falls within the
1278 # taxi window that offset allows (see
1279 # services.time_band_matching.taxi_out_score) — this needs a *real*
1280 # managed-aircraft offset, so a standalone
1281 # existing row (no aircraft_id, e.g. a rental logged nowhere else) has
1282 # this comparison skipped entirely rather than guessing one.
1283 offset_hours = (
1284 float(existing.aircraft.flight_counter_offset) if existing.aircraft else None
1285 )
1286 same_pair_step = offset_ring_step_minutes(
1287 offset_hours if offset_hours is not None else DEFAULT_OFFSET_HOURS
1288 )
1290 dep_time_new = kwargs.get("departure_time")
1291 if existing.departure_time is not None:
1292 score += time_band_score(
1293 dep_time_new, (existing.departure_time,), same_pair_step
1294 )
1295 elif offset_hours is not None:
1296 score += taxi_out_score(dep_time_new, existing.takeoff_time, offset_hours)
1298 arr_time_new = kwargs.get("arrival_time")
1299 if existing.arrival_time is not None:
1300 score += time_band_score(arr_time_new, (existing.arrival_time,), same_pair_step)
1301 elif offset_hours is not None:
1302 score += taxi_in_score(existing.landing_time, arr_time_new, offset_hours)
1304 dur_new = _kwargs_duration(kwargs)
1305 dur_old = existing.total_flight_time
1306 if (
1307 dur_new is not None
1308 and dur_old is not None
1309 and abs(dur_new - dur_old) <= _CANDIDATE_DURATION_TOLERANCE
1310 ):
1311 score += 1
1313 land_day = kwargs.get("landings_day")
1314 land_night = kwargs.get("landings_night")
1315 if (
1316 land_day is not None
1317 and land_night is not None
1318 and existing.landings_day is not None
1319 and existing.landings_night is not None
1320 and land_day == existing.landings_day
1321 and land_night == existing.landings_night
1322 ):
1323 score += 1
1325 return score
1328def find_conflicting_rows(
1329 parsed: ParsedFile,
1330 mapping: dict[str, str],
1331 pilot_user_id: int,
1332 exclude_row_nums: set[int] | None = None,
1333) -> list[ConflictRow]:
1334 """Find rows that aren't an exact duplicate but plausibly match an
1335 existing entry closely enough (score >= _CANDIDATE_MIN_SCORE) to need a
1336 human decision: keep the existing entry, overwrite it with the new data,
1337 or import as a genuinely separate new entry.
1339 Skips subtotal rows, rows with an unparseable date, and exact duplicates
1340 (same as execute_import's own dedup — an unmodified re-upload scores the
1341 maximum on every point, so without this check it would incorrectly be
1342 routed to review instead of being silently skipped), plus any row_num in
1343 *exclude_row_nums* — typically rows already resolved in an earlier pass
1344 of this same review.
1346 Candidates also include *unclaimed* rows (pic_user_id AND
1347 second_crew_user_id both NULL) on a managed aircraft belonging to one of
1348 the pilot's own tenants — typically a hand-entered airframe-log row from
1349 before this pilot ever ran a logbook import. Without this, those rows
1350 are invisible to both this check and execute_import's own dedup (both
1351 only look at rows already linked to *pilot_user_id*), so re-importing a
1352 personal logbook can never discover that the real-world flight is
1353 already logged from the airframe side — it silently creates a second,
1354 duplicate Flight row instead of surfacing a conflict to resolve.
1355 """
1356 from models import ( # pyright: ignore[reportMissingImports]
1357 Aircraft,
1358 Flight,
1359 TenantUser,
1360 )
1361 from sqlalchemy import and_, or_ # pyright: ignore[reportMissingImports]
1363 exclude_row_nums = exclude_row_nums or set()
1364 date_idx = _date_col_index(parsed.norm_cols, mapping)
1365 col_index = {col: i for i, col in enumerate(parsed.norm_cols)}
1366 existing = _fetch_existing_dedup_index(pilot_user_id)
1367 conflicts: list[ConflictRow] = []
1369 tenant_ids = [
1370 row.tenant_id for row in TenantUser.query.filter_by(user_id=pilot_user_id).all()
1371 ]
1373 for row_num, row in enumerate(parsed.data_rows, start=1):
1374 if row_num in exclude_row_nums:
1375 continue
1376 if _is_subtotal_row(row, date_idx):
1377 continue
1378 date_val = _parse_row_date(row, mapping, col_index)
1379 if date_val is None:
1380 continue
1382 kwargs, _source_total, _parse_warnings = _build_entry_kwargs(
1383 row, mapping, col_index, date_val
1384 )
1385 if _contains_kwargs(existing, kwargs):
1386 continue # exact duplicate — execute_import's own dedup handles this
1388 same_day_filters = [
1389 Flight.pic_user_id == pilot_user_id,
1390 Flight.second_crew_user_id == pilot_user_id,
1391 ]
1392 if tenant_ids:
1393 same_day_filters.append(
1394 and_(
1395 Flight.pic_user_id.is_(None),
1396 Flight.second_crew_user_id.is_(None),
1397 Flight.aircraft_id.in_(
1398 Aircraft.query.with_entities(Aircraft.id).filter(
1399 Aircraft.tenant_id.in_(tenant_ids)
1400 )
1401 ),
1402 )
1403 )
1404 same_day = Flight.query.filter(
1405 or_(*same_day_filters),
1406 Flight.date == date_val,
1407 ).all()
1408 scored = [
1409 (score, existing.id)
1410 for existing in same_day
1411 if (score := _score_candidate(kwargs, existing)) >= _CANDIDATE_MIN_SCORE
1412 ]
1413 if scored:
1414 scored.sort(key=lambda t: -t[0])
1415 conflicts.append(
1416 ConflictRow(row_num=row_num, kwargs=kwargs, candidates=scored)
1417 )
1419 return conflicts
1422# ── Merging a row into an existing entry ─────────────────────────────────────
1424# Fields a merge never takes from the parsed row: the date and route are
1425# what matched it to the existing entry in the first place.
1426_MERGE_SKIP_FIELDS = frozenset({"date", "departure_icao", "arrival_icao"})
1427_OTHER_AIRCRAFT_FIELDS = frozenset(
1428 {"other_aircraft_type", "other_aircraft_type_icao", "other_aircraft_registration"}
1429)
1432def merge_into_existing(
1433 existing: Any, kwargs: dict[str, Any], pilot_user_id: int, *, fill_only: bool
1434) -> None:
1435 """Apply a parsed row's *kwargs* onto *existing* (a Flight), as the same
1436 real-world flight. Caller must commit.
1438 *fill_only* (an automatic merge with an entry logged from the airframe
1439 side) only fills fields that are still empty. Otherwise (the review
1440 step's "overwrite") the row's values replace the existing ones. Either
1441 way an empty cell in the file never clears data already on the entry —
1442 it means "not recorded here", e.g. the pilot logbook has no column for
1443 the airframe log's remarks — and a managed-aircraft row never gets the
1444 free-text other_aircraft_* fields (its aircraft comes from the Aircraft
1445 join, see link_entries_to_aircraft).
1446 """
1447 for field_name, value in kwargs.items():
1448 if value is None or field_name in _MERGE_SKIP_FIELDS:
1449 continue
1450 if existing.aircraft_id is not None and field_name in _OTHER_AIRCRAFT_FIELDS:
1451 continue
1452 current = getattr(existing, field_name)
1453 if field_name == "notes" and current and current != value:
1454 if fill_only and value not in current:
1455 existing.notes = f"{current}\n{value}"
1456 elif not fill_only:
1457 existing.notes = value
1458 continue
1459 if not fill_only or current is None:
1460 setattr(existing, field_name, value)
1462 ident: dict[str, Any] = dict(kwargs)
1463 _assign_pilot_identity(ident, pilot_user_id)
1464 existing.pic_user_id = ident.get("pic_user_id")
1465 existing.second_crew_user_id = ident.get("second_crew_user_id")
1466 existing.second_crew_role = ident.get("second_crew_role")
1467 _fill_engine_time(existing)
1470def _fill_engine_time(entry: Any) -> None:
1471 """A managed aircraft's engine_time (engine-run hours, summed for
1472 engine/propeller TBO) from the pilot's logged block time, when nothing
1473 better is known yet. Block time is engine time — the engine runs from
1474 block-off to block-on — whereas the airborne flight_time side is never
1475 derived from a personal logbook (see link_entries_to_aircraft)."""
1476 if entry.aircraft_id is None or entry.engine_time is not None:
1477 return
1478 parts = [entry.single_pilot_se, entry.single_pilot_me, entry.multi_pilot]
1479 vals = [float(p) for p in parts if p is not None]
1480 if vals:
1481 entry.engine_time = round(sum(vals), 1)
1484def _is_airframe_only(existing: Any) -> bool:
1485 """An entry logged only from the airframe side: on a managed aircraft,
1486 no pilot claimed it and no pilot figures recorded yet."""
1487 return (
1488 existing.aircraft_id is not None
1489 and existing.pic_user_id is None
1490 and existing.second_crew_user_id is None
1491 and existing.single_pilot_se is None
1492 and existing.single_pilot_me is None
1493 and existing.multi_pilot is None
1494 )
1497# Combined score of the two cross-pair time checks (block-off vs takeoff,
1498# block-on vs landing, each 0–1) needed to merge automatically: both inside
1499# their offset band, or one inside and the other within the first ring.
1500_AUTO_MERGE_MIN_TIME_SCORE = 1.75
1503def _is_strong_cross_match(kwargs: dict[str, Any], existing: Any) -> bool:
1504 """Same aircraft, same known route, and both clock times within the
1505 taxi window (the aircraft's block-to-airborne offset) of the airframe
1506 row's takeoff/landing times — strong enough to merge without asking."""
1507 from services.time_band_matching import ( # pyright: ignore[reportMissingImports]
1508 taxi_in_score,
1509 taxi_out_score,
1510 )
1512 reg_new = (kwargs.get("other_aircraft_registration") or "").strip().upper()
1513 if reg_new.replace("-", "") != (
1514 existing.display_registration or ""
1515 ).upper().replace("-", ""):
1516 return False
1517 dep, arr = (
1518 _place_key(kwargs.get("departure_icao")),
1519 _place_key(kwargs.get("arrival_icao")),
1520 )
1521 if not dep or not arr:
1522 return False
1523 if (dep, arr) != (
1524 _place_key(existing.departure_icao),
1525 _place_key(existing.arrival_icao),
1526 ):
1527 return False
1528 offset_hours = float(existing.aircraft.flight_counter_offset)
1529 score = taxi_out_score(
1530 kwargs.get("departure_time"), existing.takeoff_time, offset_hours
1531 ) + taxi_in_score(existing.landing_time, kwargs.get("arrival_time"), offset_hours)
1532 return score >= _AUTO_MERGE_MIN_TIME_SCORE
1535def split_auto_merges(
1536 conflicts: list[ConflictRow],
1537) -> tuple[list[tuple[ConflictRow, int]], list[ConflictRow]]:
1538 """Split near-match *conflicts* into (auto_merges, needs_review).
1540 A row merges automatically — no human decision needed — when exactly
1541 one of its candidates is an entry logged only from the airframe side
1542 (see _is_airframe_only) and that candidate is a strong match (see
1543 _is_strong_cross_match): the two sides then simply complement each
1544 other, there's nothing to choose between. Each existing entry is
1545 claimed by at most one row; anything else goes to review.
1546 """
1547 from models import Flight, db # pyright: ignore[reportMissingImports]
1549 auto: list[tuple[ConflictRow, int]] = []
1550 review: list[ConflictRow] = []
1551 claimed: set[int] = set()
1552 for conflict in conflicts:
1553 strong = [
1554 cid
1555 for _score, cid in conflict.candidates
1556 if (existing := db.session.get(Flight, cid)) is not None
1557 and _is_airframe_only(existing)
1558 and _is_strong_cross_match(conflict.kwargs, existing)
1559 ]
1560 if len(strong) == 1 and strong[0] not in claimed:
1561 claimed.add(strong[0])
1562 auto.append((conflict, strong[0]))
1563 else:
1564 review.append(conflict)
1565 return auto, review
1568def link_entries_to_aircraft(entries: list[Any]) -> int:
1569 """Promote each standalone Flight row (aircraft_id NULL,
1570 other_aircraft_registration set) in *entries* to a managed aircraft, for
1571 any whose registration matches an Aircraft belonging to one of the row's
1572 pic/second-crew pilot's own tenants. Returns the count promoted. Caller
1573 must commit.
1575 Unified-model note: this used to create a brand-new FlightEntry +
1576 FlightCrew and link the pilot's existing PilotLogbookEntry to it via
1577 flight_id. There's only one row now, so promotion is just updating
1578 aircraft_id (+ clearing the free-text other_aircraft_* fields) on the
1579 row that already exists — no second row, no separate crew row (the
1580 row's pic_user_id/pic_name are already whatever the CSV import set).
1581 """
1582 from models import ( # pyright: ignore[reportMissingImports]
1583 Aircraft,
1584 TenantUser,
1585 User,
1586 db,
1587 )
1589 def _norm_reg(reg: str) -> str:
1590 return reg.upper().replace("-", "").replace(" ", "")
1592 ac_by_tenant: dict[int, dict[str, Any]] = {}
1593 for ac in Aircraft.query.all():
1594 ac_by_tenant.setdefault(ac.tenant_id, {})[_norm_reg(ac.registration)] = ac
1596 pilot_tenant_ids: dict[int, set[int]] = {}
1598 def _tenants_for_pilot(pilot_user_id: int) -> set[int]:
1599 if pilot_user_id not in pilot_tenant_ids:
1600 pilot_tenant_ids[pilot_user_id] = {
1601 row.tenant_id
1602 for row in TenantUser.query.filter_by(user_id=pilot_user_id).all()
1603 }
1604 return pilot_tenant_ids[pilot_user_id]
1606 def _place_icao(place: str | None) -> str:
1607 if not place:
1608 return "ZZZZ"
1609 clean = re.sub(r"[^A-Z0-9]", "", place.upper())[:4]
1610 return clean if len(clean) == 4 else "ZZZZ"
1612 promoted = 0
1613 for entry in entries:
1614 if entry.aircraft_id is not None or not entry.other_aircraft_registration:
1615 continue
1616 pilot_user_id = entry.pic_user_id or entry.second_crew_user_id
1617 if pilot_user_id is None:
1618 continue
1619 norm_reg = _norm_reg(entry.other_aircraft_registration)
1620 ac = None
1621 for tid in _tenants_for_pilot(pilot_user_id):
1622 ac = ac_by_tenant.get(tid, {}).get(norm_reg)
1623 if ac is not None:
1624 break
1625 if ac is None:
1626 continue
1628 entry.aircraft_id = ac.id
1629 entry.departure_icao = _place_icao(entry.departure_icao)
1630 entry.arrival_icao = _place_icao(entry.arrival_icao)
1631 entry.other_aircraft_type = None
1632 entry.other_aircraft_type_icao = None
1633 entry.other_aircraft_registration = None
1634 entry.source = "logbook_import"
1636 # The airframe-side figures (flight_time_counter_*/flight_time —
1637 # airborne time) and the engine counter readings aren't known from
1638 # a personal logbook and stay NULL: no supposition, interpolation,
1639 # or guessed counter value gets invented here (a previous version
1640 # seeded a rough engine_time_counter estimate, which
1641 # services/component_limits.py would then have summed for TBO).
1642 # flight_time staying NULL is also what keeps the flight list's
1643 # needs-attention icon (source == "logbook_import" and flight_time
1644 # is None) lit until a human, or an airframe-log import, fills it
1645 # in. The pilot's block time is a real engine-run figure, though,
1646 # so it does go to engine_time.
1647 _fill_engine_time(entry)
1649 if not entry.pic_name:
1650 pilot_user = db.session.get(User, pilot_user_id)
1651 if pilot_user:
1652 entry.pic_name = pilot_user.name or pilot_user.email
1654 promoted += 1
1656 return promoted