Coverage for app/pilots/logbook_import.py: 39%
783 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-13 03:30 +0000
1"""Pilot logbook import — parsing, normalisation, mapping, and execution."""
3from __future__ import annotations
5import copy
6import csv
7import hashlib
8import io
9import json
10import math
11import re
12import threading
13from collections import OrderedDict
14from collections.abc import Callable
15from dataclasses import dataclass, field
16from datetime import date, datetime, time, timedelta
17from decimal import Decimal
18from typing import Any
20import openpyxl # pyright: ignore[reportMissingImports]
21from flask_babel import gettext as _ # pyright: ignore[reportMissingImports]
22from flask_babel import pgettext # pyright: ignore[reportMissingImports]
24# ── Constants ─────────────────────────────────────────────────────────────────
26TARGET_FIELDS: list[str] = [
27 "date",
28 "aircraft_type",
29 "aircraft_registration",
30 "departure_place",
31 "departure_time",
32 "arrival_place",
33 "arrival_time",
34 "pic_name",
35 "night_time",
36 "instrument_time",
37 "cross_country",
38 "landings_day",
39 "landings_night",
40 "single_pilot_se",
41 "single_pilot_me",
42 "multi_pilot",
43 "function_pic",
44 "function_copilot",
45 "function_dual",
46 "function_instructor",
47 "remarks",
48 # virtual target — used for import validation only, not stored
49 "total_flight_time_check",
50]
52# Normalised source column name → target field.
53# Positional disambiguation suffixes (_2, _3 …) are applied before lookup,
54# so "time" is the first TIME column (departure) and "time_2" is the second (arrival).
55_ALIASES: dict[str, str] = {
56 # EASA standard logbook column names (normalised)
57 "date dd/mm/yy": "date",
58 "date": "date",
59 "aircraft type": "aircraft_type",
60 "type": "aircraft_type",
61 "aircraft registration number": "aircraft_registration",
62 "registration": "aircraft_registration",
63 "reg": "aircraft_registration",
64 "from": "departure_place",
65 "departure": "departure_place",
66 "dep": "departure_place",
67 "time": "departure_time", # first TIME → departure
68 "time_2": "arrival_time", # second TIME → arrival
69 "departure time": "departure_time",
70 "arrival time": "arrival_time",
71 "to": "arrival_place",
72 "arrival": "arrival_place",
73 "arr": "arrival_place",
74 "pic name": "pic_name",
75 "pic name (if not student pilot)": "pic_name",
76 "captain": "pic_name",
77 # Landings group — first DAY / NIGHT pair
78 "day": "landings_day",
79 "night": "landings_night",
80 # Operational conditions group — second DAY / NIGHT pair (positional suffix)
81 "day_2": "ignore", # cross-country day — not a logbook field
82 "night_2": "night_time", # night flying time
83 # Aircraft category
84 "se": "single_pilot_se",
85 "single engine": "single_pilot_se",
86 "single pilot se": "single_pilot_se",
87 "me": "single_pilot_me",
88 "multi engine": "single_pilot_me",
89 "single pilot me": "single_pilot_me",
90 "multi pilot": "multi_pilot",
91 "mp": "multi_pilot",
92 # Pilot function
93 "pic": "function_pic",
94 "p1": "function_pic",
95 "co-pic": "function_copilot",
96 "co-pilot": "function_copilot",
97 "copilot": "function_copilot",
98 "p2": "function_copilot",
99 "dual received": "function_dual",
100 "dual": "function_dual",
101 "student": "function_dual",
102 "instructor": "function_instructor",
103 "fi": "function_instructor",
104 # Cross-country time
105 "cross-country": "cross_country",
106 "cross country": "cross_country",
107 "x-country": "cross_country",
108 "xc": "cross_country",
109 # Total flight time — imported only to validate against computed sum
110 "total flight time": "total_flight_time_check",
111 # Catch-all
112 "total": "ignore",
113 "no. istr. appr.": "ignore",
114 "ifr approaches": "ignore",
115 "page": "ignore",
116 "line": "ignore",
117 "remarks": "remarks",
118 "notes": "remarks",
119 "comments": "remarks",
120 "instrument time": "instrument_time",
121 "ifr": "instrument_time",
122 "instrument": "instrument_time",
123 "night time": "night_time",
124 # Group-prefixed column names (e.g. Belgian EASA logbook with a span header row)
125 "departure & arrival from": "departure_place",
126 "departure & arrival time": "departure_time",
127 "departure & arrival time_2": "arrival_time",
128 "departure & arrival to": "arrival_place",
129 "aircraft category se": "single_pilot_se",
130 "aircraft category me": "single_pilot_me",
131 "operational conditions cross-country": "cross_country",
132 "operational conditions day": "ignore",
133 "operational conditions night": "night_time",
134 "pilot function pic": "function_pic",
135 "pilot function co-pic": "function_copilot",
136 "pilot function dual received": "function_dual",
137 "page subtotals total flight time": "ignore",
138 "page subtotals total flight time_2": "ignore",
139 "page subtotals pic": "ignore",
140 "page subtotals dual": "ignore",
141 "page subtotals night": "ignore",
142 "page subtotals landings – day": "ignore",
143 "page subtotals landings – night": "ignore",
144 "page subtotals formated date": "ignore",
145 "page subtotals formated type": "ignore",
146 "page subtotals days since flight": "ignore",
147 "page subtotals daytime duration": "ignore",
148 "page subtotals non pic or dual": "ignore",
149 "formated date": "ignore",
150 "formated type": "ignore",
151 "days since flight": "ignore",
152 "daytime duration": "ignore",
153 "non pic or dual": "ignore",
154 # Landings group (group-prefixed)
155 "landings day": "landings_day",
156 "landings night": "landings_night",
157 # Aircraft type (multiline cell name normalised to single space)
158 "aircraft type name, model, variant": "aircraft_type",
159}
161# ── Data structures ───────────────────────────────────────────────────────────
164@dataclass
165class ParsedFile:
166 """Result of parsing an uploaded logbook file."""
168 norm_cols: list[str] # normalised + disambiguated column keys
169 raw_cols: list[str] # original column labels (for display)
170 header_row_index: int # 0-based row index of the detected header
171 data_rows: list[list[Any]] # all rows after the header (including subtotals)
172 fingerprint: str # SHA-256 of norm_cols
175@dataclass
176class MappingProposal:
177 """Proposed column mapping and how it was derived."""
179 mapping: dict[str, str] # norm_col_key → target_field or "ignore"
180 match_type: str # "exact", "fuzzy", "alias"
181 fuzzy_score: float = 0.0 # 0.0–1.0, only meaningful for "fuzzy"
182 matched_mapping_id: int | None = None
185@dataclass
186class ImportResult:
187 """Summary returned after executing an import."""
189 imported: int = 0
190 subtotals: int = 0
191 skipped: list[tuple[int, str]] = field(default_factory=list) # (row_num, reason)
192 # (row_num, reason) — rows that matched an entry already in the pilot's logbook
193 duplicates: list[tuple[int, str]] = field(default_factory=list)
194 # (row_num, source_col, target_field, repr(raw)) — non-empty cells that couldn't parse
195 parse_warnings: list[tuple[int, str, str, str]] = field(default_factory=list)
196 # (row_num, source_total, computed_total) — rows where total ≠ sum of components
197 total_mismatch_warnings: list[tuple[int, float, float]] = field(
198 default_factory=list
199 )
200 has_opening_balance: bool = False
203# ── Normalisation ─────────────────────────────────────────────────────────────
206def _norm(text: str) -> str:
207 """Strip, collapse whitespace, lower-case."""
208 return re.sub(r"\s+", " ", str(text).strip().lower())
211def _disambiguate(names: list[str]) -> list[str]:
212 """Append _2, _3 … to duplicate names to make them unique."""
213 seen: dict[str, int] = {}
214 result: list[str] = []
215 for n in names:
216 if n in seen:
217 seen[n] += 1
218 result.append(f"{n}_{seen[n]}")
219 else:
220 seen[n] = 1
221 result.append(n)
222 return result
225def _fingerprint(norm_cols: list[str]) -> str:
226 payload = json.dumps(norm_cols, ensure_ascii=False, sort_keys=False)
227 return hashlib.sha256(payload.encode()).hexdigest()
230# ── Header detection ──────────────────────────────────────────────────────────
233def _is_header_row(row: list[Any]) -> bool:
234 """True if ≥ 50 % of non-empty cells are non-numeric strings and ≥ 4 non-empty."""
235 non_empty = [c for c in row if c is not None and str(c).strip()]
236 if len(non_empty) < 4:
237 return False
238 string_like = [
239 c for c in non_empty if isinstance(c, str) and not _is_numeric_str(str(c))
240 ]
241 return len(string_like) / len(non_empty) >= 0.5
244def _is_numeric_str(s: str) -> bool:
245 try:
246 float(s)
247 return True
248 except ValueError:
249 return False
252def _header_alias_score(row: list[Any]) -> int:
253 """Count how many cells in *row* match a known alias."""
254 return sum(1 for c in row if c is not None and _norm(str(c)) in _ALIASES)
257def _find_header_row(rows: list[list[Any]], max_scan: int = 20) -> int | None:
258 """Return 0-based index of the best header row within the first max_scan rows.
260 Many logbook templates have a group-label row (e.g. "DEPARTURE & ARRIVAL",
261 "LANDINGS") above the actual column-header row. Both pass _is_header_row,
262 but the actual header row has far more alias matches. We score every
263 candidate and return the one with the highest score; ties go to the earlier
264 row. If no alias matches are found we fall back to the first text-like row.
265 """
266 best_idx: int | None = None
267 best_score = -1
268 first_text_row: int | None = None
270 for i, row in enumerate(rows[:max_scan]):
271 if not _is_header_row(row):
272 continue
273 if first_text_row is None:
274 first_text_row = i
275 score = _header_alias_score(row)
276 if score > best_score:
277 best_score = score
278 best_idx = i
280 return best_idx if (best_idx is not None and best_score > 0) else first_text_row
283def _trim_trailing_empty_cols(
284 all_rows: list[list[Any]], max_scan: int = 50
285) -> list[list[Any]]:
286 """Trim columns beyond the rightmost non-empty value in the first max_scan rows.
288 Excel templates often declare thousands of formatted-but-empty columns.
289 Without trimming, every empty cell becomes a separate mapping entry.
290 """
291 max_col = 0
292 for row in all_rows[:max_scan]:
293 for j in range(len(row) - 1, -1, -1):
294 if row[j] is not None and str(row[j]).strip():
295 max_col = max(max_col, j + 1)
296 break
297 if max_col == 0:
298 return (
299 all_rows # pragma: no cover — blank file rejected by header detection first
300 )
301 return [row[:max_col] for row in all_rows]
304# ── Group-header detection ────────────────────────────────────────────────────
307def _merge_label_map(ws: Any) -> dict[tuple[int, int], str]:
308 """Return {(0-based row, 0-based col): group_label} for every merged region.
310 The label is the value of the top-left cell of each merged range.
311 Every cell in the range gets the same label so downstream code can do a
312 simple lookup without forward-fill.
313 """
314 result: dict[tuple[int, int], str] = {}
315 for mc in ws.merged_cells.ranges:
316 val = ws.cell(mc.min_row, mc.min_col).value
317 if val is None or not str(val).strip():
318 continue
319 label = str(val).strip()
320 for r in range(mc.min_row - 1, mc.max_row):
321 for c in range(mc.min_col - 1, mc.max_col):
322 result[(r, c)] = label
323 return result
326def _group_labels_from_map(
327 merge_map: dict[tuple[int, int], str], row_idx: int, width: int
328) -> list[str]:
329 """Return per-column group labels for *row_idx* using exact merge metadata."""
330 return [merge_map.get((row_idx, c), "") for c in range(width)]
333def _group_labels_heuristic(row: list[Any], width: int) -> list[str] | None:
334 """Fallback for CSV: forward-fill a sparse row to infer group labels.
336 Requires at least two non-empty values with a gap between them (so a single
337 spanning title cell does not fire). Returns None if the pattern is absent.
338 """
339 padded: list[Any] = list(row[:width]) + [None] * max(0, width - len(row))
341 prev_nonempty_idx: int | None = None
342 has_span = False
343 for i, val in enumerate(padded):
344 if val is not None and str(val).strip():
345 if prev_nonempty_idx is not None and i > prev_nonempty_idx + 1:
346 has_span = True
347 break
348 prev_nonempty_idx = i
349 if not has_span:
350 return None
352 result: list[str] = []
353 current = ""
354 for val in padded:
355 if val is not None and str(val).strip():
356 current = str(val).strip()
357 result.append(current)
358 return result
361def _apply_group_labels(group_labels: list[str], raw_header: list[str]) -> list[str]:
362 """Prepend each non-empty group label to the corresponding column header."""
363 result = []
364 for i, col in enumerate(raw_header):
365 label = group_labels[i] if i < len(group_labels) else ""
366 if label and col:
367 result.append(f"{label} {col}")
368 elif label:
369 result.append(label)
370 else:
371 result.append(col)
372 return result
375# ── File parsing ──────────────────────────────────────────────────────────────
378def parse_file(data: bytes, filename: str) -> ParsedFile:
379 """Parse bytes from an uploaded file into a ParsedFile.
381 Raises ValueError with a user-friendly message on format errors.
382 """
383 ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
384 if ext in ("xlsx", "xls"):
385 return _parse_excel_cached(data, filename)
386 if ext == "csv":
387 return _parse_csv(data, filename)
388 raise ValueError(
389 f"Unsupported file format: .{ext} — please upload a .csv or .xlsx file."
390 )
393# The import flow re-parses the same uploaded workbook on every step — the
394# mapping page, execute, and twice per resolved row on the review page (the
395# resolve POST, then the review page it redirects to). openpyxl needs ~1 s
396# for a few hundred logbook rows (it loads the workbook twice, see
397# _parse_excel), so each review click used to cost seconds of server time.
398# A small per-process cache keyed on the file's content hash makes the
399# repeats free. Keyed on the active locale's preferred sheet names too,
400# since they decide which sheet gets picked. Entries are deep-copied in
401# and out so a caller mutating its ParsedFile can never affect another's.
402_EXCEL_PARSE_CACHE_SIZE = 4
403_excel_parse_cache: OrderedDict[tuple[str, str, frozenset[str]], ParsedFile] = (
404 OrderedDict()
405)
406_excel_parse_cache_lock = threading.Lock()
409def _parse_excel_cached(data: bytes, filename: str) -> ParsedFile:
410 key = (hashlib.sha256(data).hexdigest(), filename, _preferred_sheet_names())
411 with _excel_parse_cache_lock:
412 cached = _excel_parse_cache.get(key)
413 if cached is not None:
414 _excel_parse_cache.move_to_end(key)
415 return copy.deepcopy(cached)
416 parsed = _parse_excel(data, filename)
417 with _excel_parse_cache_lock:
418 _excel_parse_cache[key] = copy.deepcopy(parsed)
419 while len(_excel_parse_cache) > _EXCEL_PARSE_CACHE_SIZE:
420 _excel_parse_cache.popitem(last=False)
421 return parsed
424# English-only Excel tab names used to identify the logbook sheet.
425# Locale-specific equivalents live in the translation catalogue, keyed under
426# the "excel tab name" pgettext context (see _preferred_sheet_names below).
427_PREFERRED_SHEET_NAMES_EN: frozenset[str] = frozenset(
428 {
429 "logbook",
430 "log",
431 "flights",
432 "flight log",
433 "flights log",
434 "journal",
435 }
436)
439def _preferred_sheet_names() -> frozenset[str]:
440 """Return preferred sheet names merged with their active-locale translations.
442 Uses pgettext with a dedicated context rather than plain _() — these
443 keywords are matched against user-supplied Excel tab names, not
444 displayed anywhere in the UI, and several of them (logbook/flight log,
445 log/journal) intentionally translate to the same word in French. A
446 context keeps that expected collision scoped to this lookup, so it can
447 never silently mask a real duplicate-translation issue if the same
448 English word is ever used as an actual UI label elsewhere.
450 The context string is repeated as a literal at each call site (not a
451 shared constant) because pybabel's extractor is a static source scanner —
452 it can only resolve literal string arguments, not a variable reference.
453 """
454 t_logbook = pgettext("excel tab name", "logbook")
455 t_log = pgettext("excel tab name", "log")
456 t_flights = pgettext("excel tab name", "flights")
457 t_flight_log = pgettext("excel tab name", "flight log")
458 t_flights_log = pgettext("excel tab name", "flights log")
459 t_journal = pgettext("excel tab name", "journal")
460 translated = frozenset(
461 {t_logbook, t_log, t_flights, t_flight_log, t_flights_log, t_journal}
462 )
463 return _PREFERRED_SHEET_NAMES_EN | frozenset(s.strip().lower() for s in translated)
466def _pick_best_excel_sheet(wb: Any) -> tuple[str, list[list[Any]]]:
467 """Return (sheet_name, all_rows) for the most likely logbook sheet.
469 Prefers sheets whose name matches a known logbook keyword (fast path, reads
470 only the chosen sheet). When no name matches, reads every sheet and scores
471 by alias matches in the detected header, returning the highest-scoring sheet
472 together with its rows. Falls back to the first sheet when scoring is
473 inconclusive.
475 Rows are returned here because openpyxl read-only worksheets use a one-shot
476 generator — calling iter_rows() twice on the same worksheet yields nothing on
477 the second call.
478 """
479 preferred = _preferred_sheet_names()
480 for ws in wb.worksheets:
481 if ws.title.strip().lower() in preferred:
482 return ws.title, [list(row) for row in ws.iter_rows(values_only=True)]
484 best_name = wb.worksheets[0].title if wb.worksheets else ""
485 best_score = -1
486 best_rows: list[list[Any]] = []
487 for ws in wb.worksheets:
488 rows = [list(row) for row in ws.iter_rows(values_only=True)]
489 trimmed = _trim_trailing_empty_cols(list(rows))
490 hi = _find_header_row(trimmed)
491 score = _header_alias_score(trimmed[hi]) if hi is not None else 0
492 if score > best_score:
493 best_score = score
494 best_name = ws.title
495 best_rows = rows
496 return best_name, best_rows
499def _parse_excel(data: bytes, filename: str) -> ParsedFile:
500 try:
501 wb_ro = openpyxl.load_workbook(io.BytesIO(data), read_only=True, data_only=True)
502 except Exception as exc:
503 raise ValueError(f"Could not open Excel file: {exc}") from exc
504 sheet_name, all_rows = _pick_best_excel_sheet(wb_ro)
505 wb_ro.close()
507 # Second load (non-read-only) to access merged cell ranges for exact group labels.
508 excel_merge_map: dict[tuple[int, int], str] | None = None
509 try:
510 wb_full = openpyxl.load_workbook(io.BytesIO(data), data_only=True)
511 excel_merge_map = _merge_label_map(wb_full[sheet_name])
512 wb_full.close()
513 except Exception: # noqa: S110, BLE001 # fall back to heuristic in _build_parsed_file
514 pass
516 return _build_parsed_file(all_rows, filename, excel_merge_map=excel_merge_map)
519def _parse_csv(data: bytes, filename: str) -> ParsedFile:
520 try:
521 text = data.decode("utf-8-sig")
522 except UnicodeDecodeError:
523 text = data.decode("latin-1")
525 sample = text[:4096]
526 try:
527 dialect = csv.Sniffer().sniff(sample)
528 except csv.Error:
529 dialect = csv.excel
531 reader = csv.reader(io.StringIO(text), dialect)
532 try:
533 all_rows: list[list[Any]] = list(reader)
534 except csv.Error as exc:
535 raise ValueError(f"Could not parse CSV file: {exc}") from exc
536 return _build_parsed_file(all_rows, filename)
539def _build_parsed_file(
540 all_rows: list[list[Any]],
541 filename: str,
542 excel_merge_map: dict[tuple[int, int], str] | None = None,
543) -> ParsedFile:
544 all_rows = _trim_trailing_empty_cols(all_rows)
545 header_idx = _find_header_row(all_rows)
546 if header_idx is None:
547 raise ValueError(
548 "Could not detect a header row in this file. "
549 "Make sure the file contains column names."
550 )
552 raw_header = [str(c).strip() if c is not None else "" for c in all_rows[header_idx]]
554 # Prepend group labels from the row above the header, if one exists
555 if header_idx > 0:
556 width = len(raw_header)
557 if excel_merge_map is not None:
558 # Excel: exact boundaries from merged cell metadata
559 _gl = _group_labels_from_map(excel_merge_map, header_idx - 1, width)
560 group_labels: list[str] | None = _gl if any(_gl) else None
561 else:
562 # CSV: heuristic forward-fill (merged cell info unavailable)
563 group_labels = _group_labels_heuristic(all_rows[header_idx - 1], width)
564 if group_labels is not None:
565 raw_header = _apply_group_labels(group_labels, raw_header)
567 norm_raw = [_norm(c) for c in raw_header]
568 norm_cols = _disambiguate(norm_raw)
569 data_rows = all_rows[header_idx + 1 :]
570 # Spreadsheets often carry a tail of formatted-but-empty rows; they'd
571 # otherwise each be reported as an ignored "subtotal" row. Only the
572 # trailing run is dropped, so row numbers stay those of the file.
573 while data_rows and not any(_is_nonempty(v) for v in data_rows[-1]):
574 data_rows.pop()
576 return ParsedFile(
577 norm_cols=norm_cols,
578 raw_cols=raw_header,
579 header_row_index=header_idx,
580 data_rows=data_rows,
581 fingerprint=_fingerprint(norm_cols),
582 )
585# ── Mapping proposal ──────────────────────────────────────────────────────────
588def propose_mapping(
589 parsed: ParsedFile,
590 existing_mappings: list[Any], # list[LogbookImportMapping]
591) -> MappingProposal:
592 """Return the best mapping proposal for *parsed*, checking saved mappings first."""
593 # 1. Exact fingerprint match
594 for m in existing_mappings:
595 if m.source_fingerprint == parsed.fingerprint:
596 return MappingProposal(
597 mapping=json.loads(m.column_mapping),
598 match_type="exact",
599 matched_mapping_id=m.id,
600 )
602 # 2. Fuzzy match — best overlap among saved mappings
603 best_score = 0.0
604 best_m = None
605 new_set = set(parsed.norm_cols)
606 for m in existing_mappings:
607 saved_cols: list[str] = json.loads(m.source_columns)
608 saved_set = set(saved_cols)
609 if not saved_set:
610 continue
611 overlap = len(new_set & saved_set)
612 score = overlap / max(len(new_set), len(saved_set))
613 if score > best_score:
614 best_score = score
615 best_m = m
617 if best_m is not None and best_score >= 0.6:
618 saved_map: dict[str, str] = json.loads(best_m.column_mapping)
619 # Build a mapping for the new columns, falling back to alias for unmatched ones
620 merged = _alias_mapping(parsed.norm_cols)
621 for col in parsed.norm_cols:
622 if col in saved_map:
623 merged[col] = saved_map[col]
624 return MappingProposal(
625 mapping=merged,
626 match_type="fuzzy",
627 fuzzy_score=best_score,
628 matched_mapping_id=best_m.id,
629 )
631 # 3. Alias-only auto-mapping
632 return MappingProposal(
633 mapping=_alias_mapping(parsed.norm_cols),
634 match_type="alias",
635 )
638def _alias_mapping(norm_cols: list[str]) -> dict[str, str]:
639 """Apply the built-in alias table; unknown columns default to 'ignore'."""
640 result: dict[str, str] = {}
641 for col in norm_cols:
642 result[col] = _ALIASES.get(col, "ignore")
643 return result
646# ── Subtotal detection ────────────────────────────────────────────────────────
649def _is_subtotal_row(row: list[Any], date_col_idx: int | None) -> bool:
650 if date_col_idx is None:
651 return False
652 if date_col_idx >= len(row):
653 return True
654 val = row[date_col_idx]
655 if isinstance(val, timedelta):
656 return True
657 if val is None or (isinstance(val, str) and not val.strip()):
658 return True
659 return bool(isinstance(val, str) and "total" in val.lower())
662# ── Value parsing ─────────────────────────────────────────────────────────────
665def parse_date_value(val: Any) -> date | None:
666 if isinstance(val, datetime):
667 return val.date()
668 if isinstance(val, date):
669 return val
670 if isinstance(val, (int, float)):
671 # Excel serial date number — openpyxl returns datetime; guard anyway
672 return None
673 if not isinstance(val, str):
674 return None
675 s = val.strip()
676 if not s:
677 return None
678 for fmt in ("%d/%m/%y", "%d/%m/%Y", "%Y-%m-%d", "%m/%d/%Y", "%d-%m-%Y"):
679 try:
680 return datetime.strptime(s, fmt).date()
681 except ValueError:
682 pass
683 return None
686def parse_time_value(val: Any) -> time | None:
687 """Parse a time-of-day value (HH:MM or Python time)."""
688 if isinstance(val, time):
689 return val
690 if isinstance(val, datetime):
691 return val.time()
692 if not isinstance(val, str):
693 return None
694 s = val.strip()
695 if not s:
696 return None
697 m = re.match(r"^(\d{1,2}):(\d{2})$", s)
698 if m:
699 try:
700 return time(int(m.group(1)), int(m.group(2)))
701 except ValueError:
702 return None
703 return None
706def parse_duration_value(val: Any) -> float | None:
707 """Parse a duration into decimal hours (e.g. time(0,42) → 0.7, '1:24' → 1.4)."""
708 if isinstance(val, timedelta):
709 hours = val.total_seconds() / 3600
710 return round(hours, 1) if hours >= 0 else None
711 if isinstance(val, time):
712 return round(val.hour + val.minute / 60, 1)
713 if isinstance(val, datetime):
714 # Excel sometimes returns a dummy date + the time-of-day
715 t = val.time()
716 return round(t.hour + t.minute / 60, 1)
717 if isinstance(val, (int, float)):
718 if not math.isfinite(val) or val < 0:
719 return None
720 return round(float(val), 1)
721 if isinstance(val, str):
722 s = val.strip()
723 if not s:
724 return None
725 m = re.match(r"^(\d+):(\d{2})$", s)
726 if m:
727 return round(int(m.group(1)) + int(m.group(2)) / 60, 1)
728 try:
729 v = float(s)
730 return round(v, 1) if math.isfinite(v) and v >= 0 else None
731 except ValueError:
732 return None
733 return None
736def parse_int_value(val: Any) -> int | None:
737 if isinstance(val, int):
738 return val if val >= 0 else None
739 if isinstance(val, float):
740 try:
741 return int(val) if val >= 0 else None
742 except (ValueError, OverflowError):
743 return None
744 if isinstance(val, str):
745 s = val.strip()
746 if not s:
747 return None
748 try:
749 n = int(float(s))
750 except (ValueError, OverflowError):
751 return None
752 return n if n >= 0 else None
753 return None
756# Transcribed logbooks (paper → spreadsheet) mark a cell deliberately left
757# blank on paper with this literal placeholder rather than a truly empty
758# cell, to tell "known blank" apart from "the transcription missed it" —
759# e.g. an airframe log's intermediate counter on a two-leg engine run, or a
760# pilot log's block times on a flight cancelled before takeoff. It is read
761# as a blank cell everywhere a mapped cell is read, by both the pilot and
762# the airframe import: no value, no parse warning, no "[empty]" text stored.
763# (The subtotal check on the date column deliberately sees the raw marker,
764# so a row whose date is "[empty]" is reported as an unparseable date
765# rather than silently dropped as a subtotal line.)
766BLANK_CELL_MARKER = "[empty]"
769def is_blank_cell(raw: Any) -> bool:
770 """True for None, a whitespace-only string, or the BLANK_CELL_MARKER."""
771 if raw is None:
772 return True
773 s = str(raw).strip()
774 return not s or s.casefold() == BLANK_CELL_MARKER
777def _is_nonempty(val: Any) -> bool:
778 """Return True when val carries a real value (not blank, see is_blank_cell)."""
779 return not is_blank_cell(val)
782# Map target field → (human-readable type name, parser function)
783_FIELD_TYPE: dict[str, tuple[str, Callable[[Any], Any]]] = {
784 "date": ("date", parse_date_value),
785 "departure_time": ("time", parse_time_value),
786 "arrival_time": ("time", parse_time_value),
787 "night_time": ("duration", parse_duration_value),
788 "instrument_time": ("duration", parse_duration_value),
789 "cross_country": ("duration", parse_duration_value),
790 "total_flight_time_check": ("duration", parse_duration_value),
791 "single_pilot_se": ("duration", parse_duration_value),
792 "single_pilot_me": ("duration", parse_duration_value),
793 "multi_pilot": ("duration", parse_duration_value),
794 "function_pic": ("duration", parse_duration_value),
795 "function_copilot": ("duration", parse_duration_value),
796 "function_dual": ("duration", parse_duration_value),
797 "function_instructor": ("duration", parse_duration_value),
798 "landings_day": ("integer", parse_int_value),
799 "landings_night": ("integer", parse_int_value),
800}
802_HINT_SAMPLE_ROWS = 5
805def type_hints(parsed: ParsedFile, mapping: dict[str, str]) -> dict[str, str]:
806 """Return {col: hint_text} for columns where sample data doesn't match the proposed type.
808 Samples up to _HINT_SAMPLE_ROWS rows. Returns a hint when non-empty values are
809 present but any of them fail to parse — indicating a likely mapping mismatch.
810 """
811 col_index = {col: i for i, col in enumerate(parsed.norm_cols)}
812 hints: dict[str, str] = {}
813 for col, target in mapping.items():
814 if target not in _FIELD_TYPE:
815 continue
816 type_name, parser = _FIELD_TYPE[target]
817 idx = col_index.get(col)
818 if idx is None:
819 continue
820 sample = [
821 row[idx]
822 for row in parsed.data_rows[:_HINT_SAMPLE_ROWS]
823 if idx < len(row) and _is_nonempty(row[idx])
824 ]
825 if not sample:
826 continue
827 failed = [v for v in sample if parser(v) is None]
828 if failed:
829 example = str(failed[0])[:30]
830 hints[col] = (
831 f"Sample data doesn't look like a {type_name} (e.g. {example!r})"
832 )
833 return hints
836# ── Preview rows ──────────────────────────────────────────────────────────────
839def preview_rows(
840 parsed: ParsedFile,
841 mapping: dict[str, str],
842 n: int = 5,
843) -> list[dict[str, Any]]:
844 """Return up to *n* non-subtotal data rows mapped to target field names."""
845 date_idx = _date_col_index(parsed.norm_cols, mapping)
846 result: list[dict[str, Any]] = []
847 for row in parsed.data_rows:
848 if _is_subtotal_row(row, date_idx):
849 continue
850 mapped: dict[str, Any] = {}
851 for i, col in enumerate(parsed.norm_cols):
852 target = mapping.get(col, "ignore")
853 if target == "ignore":
854 continue
855 val = row[i] if i < len(row) else None
856 mapped[target] = None if is_blank_cell(val) else val
857 result.append(mapped)
858 if len(result) >= n:
859 break
860 return result
863# Mapped fields holding a duration (pilot import, plus the airframe
864# import's flight_time): a time-of-day cell mapped to one is a duration
865# typed into an Excel time cell (e.g. 2:54 of SE time), so the preview
866# shows it as hours:minutes ("2:54"), not as a clock time ("02:54").
867_PREVIEW_DURATION_TARGETS = frozenset(
868 {t for t, (type_name, _parser) in _FIELD_TYPE.items() if type_name == "duration"}
869 | {"flight_time"}
870)
873def review_values_differ(old: Any, new: Any, places: int = 2) -> bool:
874 """For the import review pages' side-by-side comparison: do an existing
875 value and the file's value differ as shown? Numbers are compared as
876 numbers rounded to *places* — the database gives Decimal('1493.3') and
877 the file 1493.3 (float), which are not == to each other — and text
878 ignoring surrounding whitespace."""
879 if old is None or new is None:
880 return old is not new
881 numbers = (int, float, Decimal)
882 if isinstance(old, numbers) and isinstance(new, numbers):
883 return round(float(old), places) != round(float(new), places)
884 if isinstance(old, str) and isinstance(new, str):
885 return old.strip() != new.strip()
886 return bool(old != new)
889def format_preview_cell(val: Any, target: str | None = None) -> str:
890 """A raw spreadsheet cell as the mapping-page preview shows it — the way
891 the spreadsheet displays it rather than Python's repr: an Excel date
892 cell (read back as a midnight datetime) as just the date, a time
893 without its ":00" seconds, a duration as hours:minutes — including a
894 duration typed into a time cell, when *target* (the field the column
895 is mapped to) is a duration. Display only; the import itself parses
896 the raw value (see _FIELD_TYPE)."""
897 if val is None:
898 return ""
899 if isinstance(val, time) and target in _PREVIEW_DURATION_TARGETS:
900 return f"{val.hour}:{val.minute:02d}"
901 if isinstance(val, datetime):
902 if val.time() == time(0, 0):
903 return val.date().isoformat()
904 return val.strftime(
905 "%Y-%m-%d %H:%M" if val.second == 0 else "%Y-%m-%d %H:%M:%S"
906 )
907 if isinstance(val, date):
908 return val.isoformat()
909 if isinstance(val, time):
910 return val.strftime("%H:%M" if val.second == 0 else "%H:%M:%S")
911 if isinstance(val, timedelta):
912 total_minutes = round(val.total_seconds() / 60)
913 sign = "-" if total_minutes < 0 else ""
914 hours, minutes = divmod(abs(total_minutes), 60)
915 return f"{sign}{hours}:{minutes:02d}"
916 return str(val)
919def _date_col_index(norm_cols: list[str], mapping: dict[str, str]) -> int | None:
920 for i, col in enumerate(norm_cols):
921 if mapping.get(col) == "date":
922 return i
923 return None
926# ── Import execution ──────────────────────────────────────────────────────────
929def _row_get(row: list[Any], col_index: dict[str, int], col: str) -> Any:
930 i = col_index.get(col)
931 val = row[i] if i is not None and i < len(row) else None
932 return None if is_blank_cell(val) else val
935def _parse_row_date(
936 row: list[Any], mapping: dict[str, str], col_index: dict[str, int]
937) -> date | None:
938 for col, target in mapping.items():
939 if target == "date":
940 return parse_date_value(_row_get(row, col_index, col))
941 return None
944def _build_entry_kwargs(
945 row: list[Any],
946 mapping: dict[str, str],
947 col_index: dict[str, int],
948 date_val: date,
949) -> tuple[dict[str, Any], float | None, list[tuple[str, str, str]]]:
950 """Build Flight (standalone, aircraft_id NULL) field kwargs for one data
951 row (identity/batch fields like pic_user_id are the caller's
952 responsibility to add).
954 Returns (kwargs, source_total_flight_time_check, parse_warnings), where
955 parse_warnings is a list of (col, target, raw_repr) for non-empty cells
956 that couldn't be parsed as their target field's type. Shared by the
957 normal import pass and the near-match conflict finder below, so the two
958 can never compute a row's fields differently.
959 """
960 kwargs: dict[str, Any] = {"date": date_val}
961 source_total: float | None = None
962 parse_warnings: list[tuple[str, str, str]] = []
964 for col, target in mapping.items():
965 if target in ("ignore", "date"):
966 continue
967 raw = _row_get(row, col_index, col)
968 if target == "total_flight_time_check":
969 if _is_nonempty(raw):
970 source_total = parse_duration_value(raw)
971 continue
972 if target in _FIELD_TYPE:
973 # NB: unpack into _type_name, not _ — callers of this module also
974 # call gettext (imported as `_`) in the same function scope;
975 # reassigning `_` would shadow it for the rest of that function
976 # (Python's function-scoping, not block-scoping).
977 _type_name, parser = _FIELD_TYPE[target]
978 parsed_val = parser(raw)
979 if parsed_val is None and _is_nonempty(raw):
980 parse_warnings.append((col, target, repr(str(raw)[:40])))
981 kwargs[target] = parsed_val
982 elif target == "aircraft_type":
983 val = str(raw).strip() if raw is not None else None
984 kwargs["other_aircraft_type"] = val
985 if val:
986 from utils import (
987 resolve_aircraft_type_icao, # pyright: ignore[reportMissingImports]
988 )
990 kwargs["other_aircraft_type_icao"] = resolve_aircraft_type_icao(val)
991 elif target == "aircraft_registration":
992 kwargs["other_aircraft_registration"] = (
993 str(raw).strip() if raw is not None else None
994 )
995 elif target == "departure_place":
996 kwargs["departure_icao"] = str(raw).strip() if raw is not None else None
997 elif target == "arrival_place":
998 kwargs["arrival_icao"] = str(raw).strip() if raw is not None else None
999 elif target == "pic_name":
1000 kwargs["pic_name"] = str(raw).strip() if raw is not None else None
1001 elif target == "remarks":
1002 kwargs["notes"] = str(raw).strip() if raw is not None else None
1004 return kwargs, source_total, parse_warnings
1007def _num(v: Any) -> float | None:
1008 # Numeric columns come back as decimal.Decimal; the freshly parsed side
1009 # is always plain float (parse_duration_value). Decimal('0.7') != 0.7
1010 # (float) directly — comparing Decimals converted from float literals
1011 # hits binary-rounding mismatches — so normalise both sides to float
1012 # before building/comparing keys.
1013 return None if v is None else float(v)
1016def _dup_key(kwargs: dict[str, Any]) -> tuple[Any, ...]:
1017 """Coarse exact-duplicate key: date + aircraft + duration + landings.
1018 Necessary but not sufficient for a duplicate — see DedupIndex, which
1019 additionally requires the route and departure time to be compatible.
1020 Shared by execute_import (silently skips exact matches) and
1021 find_conflicting_rows (must not also flag them as a near-match conflict
1022 needing review)."""
1023 return (
1024 kwargs["date"],
1025 kwargs.get("other_aircraft_registration"),
1026 kwargs.get("single_pilot_se"),
1027 kwargs.get("single_pilot_me"),
1028 kwargs.get("multi_pilot"),
1029 kwargs.get("landings_day"),
1030 kwargs.get("landings_night"),
1031 )
1034def _place_key(place: str | None) -> str | None:
1035 """Normalise a departure/arrival place for duplicate matching — the same
1036 4-char cleanup link_entries_to_aircraft applies when promoting a row to
1037 a managed aircraft, so a promoted row still matches its source row.
1038 None (unknown) for anything that doesn't reduce to a 4-char code."""
1039 if not place:
1040 return None
1041 clean = re.sub(r"[^A-Z0-9]", "", place.upper())[:4]
1042 return clean if len(clean) == 4 and clean != "ZZZZ" else None
1045def _dup_detail(
1046 departure: str | None, arrival: str | None, departure_time: time | None
1047) -> tuple[str | None, str | None, time | None]:
1048 return (_place_key(departure), _place_key(arrival), departure_time)
1051def _details_compatible(a: tuple[Any, ...], b: tuple[Any, ...]) -> bool:
1052 """Equal wherever both sides know the value — a missing route or time on
1053 either side is neutral, never a mismatch."""
1054 return all(x is None or y is None or x == y for x, y in zip(a, b, strict=True))
1057class DedupIndex:
1058 """Exact-duplicate lookup shared by the pilot and airframe logbook
1059 imports: a row is a duplicate of an existing entry when their coarse
1060 key matches AND their details are compatible (see _details_compatible)
1061 — for the pilot import, departure, arrival and departure time; for the
1062 airframe import, takeoff/landing time and counter readings.
1064 The coarse key alone is not enough: two genuinely different flights on
1065 the same day in the same aircraft routinely share duration and landings
1066 (an out-and-back pair of 0.5 h legs, or a day of identical training
1067 circuits), and keying on it alone silently dropped them as
1068 "duplicates" — or matched a row to the wrong flight of that day. The
1069 details can't simply be added to the key either, since entries where
1070 they genuinely aren't captured (a hand-entered or backfilled row) would
1071 then never match at all.
1072 """
1074 def __init__(self) -> None:
1075 self._by_key: dict[tuple[Any, ...], list[tuple[Any, ...]]] = {}
1077 def add(self, key: tuple[Any, ...], detail: tuple[Any, ...]) -> None:
1078 self._by_key.setdefault(key, []).append(detail)
1080 def contains(self, key: tuple[Any, ...], detail: tuple[Any, ...]) -> bool:
1081 return any(
1082 _details_compatible(detail, other) for other in self._by_key.get(key, [])
1083 )
1086def _add_kwargs(index: DedupIndex, kwargs: dict[str, Any]) -> None:
1087 index.add(_dup_key(kwargs), _kwargs_dup_detail(kwargs))
1090def _contains_kwargs(index: DedupIndex, kwargs: dict[str, Any]) -> bool:
1091 return index.contains(_dup_key(kwargs), _kwargs_dup_detail(kwargs))
1094def _kwargs_dup_detail(kwargs: dict[str, Any]) -> tuple[Any, ...]:
1095 return _dup_detail(
1096 kwargs.get("departure_icao"),
1097 kwargs.get("arrival_icao"),
1098 kwargs.get("departure_time"),
1099 )
1102def _fetch_existing_dedup_index(pilot_user_id: int) -> DedupIndex:
1103 from models import Aircraft, Flight, db # pyright: ignore[reportMissingImports]
1104 from sqlalchemy import or_ # pyright: ignore[reportMissingImports]
1106 # Registration is Aircraft.registration for a managed-aircraft row, or
1107 # the free-text other_aircraft_registration for a standalone one.
1108 rows = (
1109 db.session.query(
1110 Flight.date,
1111 Flight.other_aircraft_registration,
1112 Aircraft.registration,
1113 Flight.single_pilot_se,
1114 Flight.single_pilot_me,
1115 Flight.multi_pilot,
1116 Flight.landings_day,
1117 Flight.landings_night,
1118 Flight.departure_icao,
1119 Flight.arrival_icao,
1120 Flight.departure_time,
1121 )
1122 .outerjoin(Aircraft, Aircraft.id == Flight.aircraft_id)
1123 .filter(
1124 or_(
1125 Flight.pic_user_id == pilot_user_id,
1126 Flight.second_crew_user_id == pilot_user_id,
1127 )
1128 )
1129 )
1130 index = DedupIndex()
1131 for row in rows:
1132 index.add(
1133 (
1134 row[0],
1135 row[2] or row[1],
1136 _num(row[3]),
1137 _num(row[4]),
1138 _num(row[5]),
1139 row[6],
1140 row[7],
1141 ),
1142 _dup_detail(row[8], row[9], row[10]),
1143 )
1144 return index
1147def _assign_pilot_identity(kwargs: dict[str, Any], pilot_user_id: int) -> None:
1148 """Put *pilot_user_id* into the correct crew slot for one row (mutates
1149 *kwargs* in place), based on its function_* breakdown.
1151 A personal-logbook row's "PIC name" column (already in kwargs["pic_name"]
1152 by the time this runs) records the actual PIC's name as free text,
1153 independent of which slot the importing pilot occupies on that row. When
1154 the row shows dual-received or copilot time and no PIC time, the
1155 importing pilot was not the PIC — they belong in the second_crew_* slot
1156 (as student or copilot) instead, leaving pic_user_id unset so pic_name's
1157 free-text name remains the row's only PIC identity. Any other case (PIC
1158 time logged, instructor time logged, or no function breakdown at all —
1159 e.g. a source file with no "Pilot function" column) keeps pic_user_id as
1160 the importing pilot, matching this app's original single-slot behaviour.
1161 """
1162 from models import CrewRole # pyright: ignore[reportMissingImports]
1164 function_pic = kwargs.get("function_pic")
1165 function_dual = kwargs.get("function_dual")
1166 function_copilot = kwargs.get("function_copilot")
1168 if not function_pic and function_dual:
1169 kwargs["second_crew_user_id"] = pilot_user_id
1170 kwargs["second_crew_role"] = CrewRole.STUDENT
1171 elif not function_pic and function_copilot:
1172 kwargs["second_crew_user_id"] = pilot_user_id
1173 kwargs["second_crew_role"] = CrewRole.COPILOT
1174 else:
1175 kwargs["pic_user_id"] = pilot_user_id
1178def execute_import(
1179 parsed: ParsedFile,
1180 mapping: dict[str, str],
1181 pilot_user_id: int,
1182 batch_id: int,
1183 opening_balance: dict[str, Any] | None = None,
1184 skip_row_nums: set[int] | None = None,
1185) -> ImportResult:
1186 """Create standalone Flight rows (aircraft_id NULL) from *parsed* using
1187 *mapping*.
1189 Returns an ImportResult describing what happened. Entries are added to
1190 db.session but NOT committed — the caller commits after also saving the
1191 batch/mapping records. *skip_row_nums* (1-based, matching the row
1192 numbering used throughout this module) lets the caller carve out rows
1193 it's handling separately — e.g. near-match conflicts routed to the
1194 interactive review step in app/pilots/routes.py — so they're excluded
1195 entirely from this pass (not counted as imported, duplicate, or skipped).
1196 """
1197 from models import Flight, db # pyright: ignore[reportMissingImports]
1199 result = ImportResult()
1200 date_idx = _date_col_index(parsed.norm_cols, mapping)
1201 col_index = {col: i for i, col in enumerate(parsed.norm_cols)}
1202 skip_row_nums = skip_row_nums or set()
1204 entries_to_add: list[Flight] = []
1206 # Duplicate detection: re-importing the same file — either by mistake, or
1207 # deliberately after appending new rows to the source spreadsheet, which
1208 # is the expected workflow for keeping an import-based logbook current —
1209 # must only add genuinely new rows, not double up everything already
1210 # imported. Match on date + aircraft + the duration/landings figures,
1211 # with departure/arrival place & time only required to agree where both
1212 # sides actually record them — see DedupIndex.
1213 existing = _fetch_existing_dedup_index(pilot_user_id)
1215 for row_num, row in enumerate(parsed.data_rows, start=1):
1216 if row_num in skip_row_nums:
1217 continue
1218 if _is_subtotal_row(row, date_idx):
1219 result.subtotals += 1
1220 continue
1222 date_val = _parse_row_date(row, mapping, col_index)
1223 if date_val is None:
1224 raw_date = (
1225 row[date_idx] if date_idx is not None and date_idx < len(row) else None
1226 )
1227 result.skipped.append((row_num, f"unparseable date: {raw_date!r}"))
1228 continue
1230 kwargs, source_total, parse_warnings = _build_entry_kwargs(
1231 row, mapping, col_index, date_val
1232 )
1233 _assign_pilot_identity(kwargs, pilot_user_id)
1234 kwargs["import_batch_id"] = batch_id
1235 kwargs["source"] = "import"
1236 for col, target, raw_repr in parse_warnings:
1237 result.parse_warnings.append((row_num, col, target, raw_repr))
1239 if _contains_kwargs(existing, kwargs):
1240 result.duplicates.append(
1241 (
1242 row_num,
1243 _(
1244 "matches an entry already in your logbook "
1245 "(same date, aircraft, duration and landings)"
1246 ),
1247 )
1248 )
1249 continue
1250 _add_kwargs(existing, kwargs)
1252 if source_total is not None:
1253 computed = round(
1254 sum(
1255 float(kwargs.get(f) or 0)
1256 for f in ("single_pilot_se", "single_pilot_me", "multi_pilot")
1257 ),
1258 1,
1259 )
1260 if abs(source_total - computed) >= 0.15:
1261 result.total_mismatch_warnings.append((row_num, source_total, computed))
1263 entries_to_add.append(Flight(**kwargs))
1265 result.imported = len(entries_to_add)
1266 for e in entries_to_add:
1267 db.session.add(e)
1269 # Opening balance — synthetic entry dated one day before earliest imported.
1270 # Only one opening-balance entry ever makes sense per pilot — re-importing
1271 # with opening-balance values filled in again must not create a second one.
1272 if opening_balance and any(v for v in opening_balance.values() if v):
1273 ob_exists = (
1274 db.session.query(Flight.id)
1275 .filter_by(pic_user_id=pilot_user_id, notes="Opening balance (imported)")
1276 .first()
1277 is not None
1278 )
1279 if ob_exists:
1280 result.duplicates.append(
1281 (
1282 0,
1283 _(
1284 "opening balance: an opening balance entry already "
1285 "exists for this pilot — not created again"
1286 ),
1287 )
1288 )
1289 else:
1290 earliest = (
1291 min(e.date for e in entries_to_add) if entries_to_add else date.today()
1292 )
1293 from datetime import timedelta as _td
1295 balance_date = earliest - _td(days=1)
1296 balance_entry = Flight(
1297 pic_user_id=pilot_user_id,
1298 import_batch_id=batch_id,
1299 source="import",
1300 date=balance_date,
1301 notes="Opening balance (imported)",
1302 night_time=opening_balance.get("night_time"),
1303 instrument_time=opening_balance.get("instrument_time"),
1304 single_pilot_se=opening_balance.get("single_pilot_se"),
1305 single_pilot_me=opening_balance.get("single_pilot_me"),
1306 multi_pilot=opening_balance.get("multi_pilot"),
1307 function_pic=opening_balance.get("function_pic"),
1308 function_copilot=opening_balance.get("function_copilot"),
1309 function_dual=opening_balance.get("function_dual"),
1310 function_instructor=opening_balance.get("function_instructor"),
1311 )
1312 db.session.add(balance_entry)
1313 result.has_opening_balance = True
1315 return result
1318# ── Near-match conflict detection (possible corrections) ───────────────────────
1320_CANDIDATE_MIN_SCORE = 3
1321_CANDIDATE_DURATION_TOLERANCE = 1.0
1324@dataclass
1325class ConflictRow:
1326 """A parsed row that isn't an exact duplicate but scores highly enough
1327 against one or more existing entries to plausibly be an edited version
1328 of one of them — needs a human decision, not a guess."""
1330 row_num: int
1331 kwargs: dict[str, Any]
1332 candidates: list[tuple[float, int]] # (score, existing_entry_id), best first
1335def _kwargs_duration(kwargs: dict[str, Any]) -> float | None:
1336 parts = [
1337 kwargs.get(f) for f in ("single_pilot_se", "single_pilot_me", "multi_pilot")
1338 ]
1339 vals = [float(p) for p in parts if p is not None]
1340 return round(sum(vals), 1) if vals else None
1343def _score_candidate(kwargs: dict[str, Any], existing: Any) -> float:
1344 """Score how likely *existing* (a Flight row) is the same real-world
1345 flight as *kwargs* (a freshly parsed row), across 7 points: registration,
1346 departure, arrival, departure time, arrival time, total duration,
1347 landings. A point only counts toward the score if both sides have data
1348 for it — missing data on either side is neutral, never a mismatch, so a
1349 row whose route/time isn't captured on one side can still be recognised
1350 via duration + landings alone. The two time points can be fractional —
1351 see services.time_band_matching.
1352 """
1353 from services.time_band_matching import ( # pyright: ignore[reportMissingImports]
1354 DEFAULT_OFFSET_HOURS,
1355 offset_ring_step_minutes,
1356 taxi_in_score,
1357 taxi_out_score,
1358 time_band_score,
1359 )
1361 score: float = 0.0
1363 reg_new = (kwargs.get("other_aircraft_registration") or "").strip().upper()
1364 reg_old = (existing.display_registration or "").strip().upper()
1365 if reg_new and reg_old and reg_new == reg_old:
1366 score += 1
1368 dep_new = (kwargs.get("departure_icao") or "").strip().upper()
1369 dep_old = (existing.departure_icao or "").strip().upper()
1370 if dep_new and dep_old and dep_new == dep_old:
1371 score += 1
1373 arr_new = (kwargs.get("arrival_icao") or "").strip().upper()
1374 arr_old = (existing.arrival_icao or "").strip().upper()
1375 if arr_new and arr_old and arr_new == arr_old:
1376 score += 1
1378 # Same-pair (existing.departure_time/arrival_time set) is scored tight,
1379 # expecting a near-exact repeat of the same real instant, with a ring
1380 # width of 1/3 of the managed aircraft's flight_counter_offset — pilots
1381 # log to a fixed precision (0.1h/6min at the 0.3h default), so a flat
1382 # few-minutes tolerance would put an ordinary rounding difference in the
1383 # wrong ring. Falls back to the model's own 0.3h default when this row
1384 # has no managed aircraft to read a real offset from. Cross-pair
1385 # (existing only has takeoff_time/landing_time — an airframe-import
1386 # placeholder, e.g. a hand-entered row from before this pilot ever ran
1387 # a logbook import) is scored by whether this time falls within the
1388 # taxi window that offset allows (see
1389 # services.time_band_matching.taxi_out_score) — this needs a *real*
1390 # managed-aircraft offset, so a standalone
1391 # existing row (no aircraft_id, e.g. a rental logged nowhere else) has
1392 # this comparison skipped entirely rather than guessing one.
1393 offset_hours = (
1394 float(existing.aircraft.flight_counter_offset) if existing.aircraft else None
1395 )
1396 same_pair_step = offset_ring_step_minutes(
1397 offset_hours if offset_hours is not None else DEFAULT_OFFSET_HOURS
1398 )
1400 dep_time_new = kwargs.get("departure_time")
1401 if existing.departure_time is not None:
1402 score += time_band_score(
1403 dep_time_new, (existing.departure_time,), same_pair_step
1404 )
1405 elif offset_hours is not None:
1406 score += taxi_out_score(dep_time_new, existing.takeoff_time, offset_hours)
1408 arr_time_new = kwargs.get("arrival_time")
1409 if existing.arrival_time is not None:
1410 score += time_band_score(arr_time_new, (existing.arrival_time,), same_pair_step)
1411 elif offset_hours is not None:
1412 score += taxi_in_score(existing.landing_time, arr_time_new, offset_hours)
1414 dur_new = _kwargs_duration(kwargs)
1415 dur_old = existing.total_flight_time
1416 if (
1417 dur_new is not None
1418 and dur_old is not None
1419 and abs(dur_new - dur_old) <= _CANDIDATE_DURATION_TOLERANCE
1420 ):
1421 score += 1
1423 land_day = kwargs.get("landings_day")
1424 land_night = kwargs.get("landings_night")
1425 if (
1426 land_day is not None
1427 and land_night is not None
1428 and existing.landings_day is not None
1429 and existing.landings_night is not None
1430 and land_day == existing.landings_day
1431 and land_night == existing.landings_night
1432 ):
1433 score += 1
1435 return score
1438def find_conflicting_rows(
1439 parsed: ParsedFile,
1440 mapping: dict[str, str],
1441 pilot_user_id: int,
1442 exclude_row_nums: set[int] | None = None,
1443) -> list[ConflictRow]:
1444 """Find rows that aren't an exact duplicate but plausibly match an
1445 existing entry closely enough (score >= _CANDIDATE_MIN_SCORE) to need a
1446 human decision: keep the existing entry, overwrite it with the new data,
1447 or import as a genuinely separate new entry.
1449 Skips subtotal rows, rows with an unparseable date, and exact duplicates
1450 (same as execute_import's own dedup — an unmodified re-upload scores the
1451 maximum on every point, so without this check it would incorrectly be
1452 routed to review instead of being silently skipped), plus any row_num in
1453 *exclude_row_nums* — typically rows already resolved in an earlier pass
1454 of this same review.
1456 Candidates also include *unclaimed* rows (pic_user_id AND
1457 second_crew_user_id both NULL) on a managed aircraft belonging to one of
1458 the pilot's own tenants — typically a hand-entered airframe-log row from
1459 before this pilot ever ran a logbook import. Without this, those rows
1460 are invisible to both this check and execute_import's own dedup (both
1461 only look at rows already linked to *pilot_user_id*), so re-importing a
1462 personal logbook can never discover that the real-world flight is
1463 already logged from the airframe side — it silently creates a second,
1464 duplicate Flight row instead of surfacing a conflict to resolve.
1465 """
1466 from models import ( # pyright: ignore[reportMissingImports]
1467 Aircraft,
1468 Flight,
1469 TenantUser,
1470 )
1471 from sqlalchemy import and_, or_ # pyright: ignore[reportMissingImports]
1473 exclude_row_nums = exclude_row_nums or set()
1474 date_idx = _date_col_index(parsed.norm_cols, mapping)
1475 col_index = {col: i for i, col in enumerate(parsed.norm_cols)}
1476 existing = _fetch_existing_dedup_index(pilot_user_id)
1477 conflicts: list[ConflictRow] = []
1479 tenant_ids = [
1480 row.tenant_id for row in TenantUser.query.filter_by(user_id=pilot_user_id).all()
1481 ]
1483 for row_num, row in enumerate(parsed.data_rows, start=1):
1484 if row_num in exclude_row_nums:
1485 continue
1486 if _is_subtotal_row(row, date_idx):
1487 continue
1488 date_val = _parse_row_date(row, mapping, col_index)
1489 if date_val is None:
1490 continue
1492 kwargs, _source_total, _parse_warnings = _build_entry_kwargs(
1493 row, mapping, col_index, date_val
1494 )
1495 if _contains_kwargs(existing, kwargs):
1496 continue # exact duplicate — execute_import's own dedup handles this
1498 same_day_filters = [
1499 Flight.pic_user_id == pilot_user_id,
1500 Flight.second_crew_user_id == pilot_user_id,
1501 ]
1502 if tenant_ids:
1503 same_day_filters.append(
1504 and_(
1505 Flight.pic_user_id.is_(None),
1506 Flight.second_crew_user_id.is_(None),
1507 Flight.aircraft_id.in_(
1508 Aircraft.query.with_entities(Aircraft.id).filter(
1509 Aircraft.tenant_id.in_(tenant_ids)
1510 )
1511 ),
1512 )
1513 )
1514 same_day = Flight.query.filter(
1515 or_(*same_day_filters),
1516 Flight.date == date_val,
1517 ).all()
1518 scored = []
1519 for candidate in same_day:
1520 score = _score_candidate(kwargs, candidate)
1521 # A cancelled-before-takeoff engine run may have nothing but
1522 # the route to score on — still offer it (and let
1523 # split_auto_merges merge it).
1524 if _is_no_takeoff_match(kwargs, candidate):
1525 score = max(score, float(_CANDIDATE_MIN_SCORE))
1526 if score >= _CANDIDATE_MIN_SCORE:
1527 scored.append((score, candidate.id))
1528 if scored:
1529 scored.sort(key=lambda t: -t[0])
1530 conflicts.append(
1531 ConflictRow(row_num=row_num, kwargs=kwargs, candidates=scored)
1532 )
1534 return conflicts
1537# ── Merging a row into an existing entry ─────────────────────────────────────
1539# Fields a merge never takes from the parsed row: the date and route are
1540# what matched it to the existing entry in the first place.
1541_MERGE_SKIP_FIELDS = frozenset({"date", "departure_icao", "arrival_icao"})
1542_OTHER_AIRCRAFT_FIELDS = frozenset(
1543 {"other_aircraft_type", "other_aircraft_type_icao", "other_aircraft_registration"}
1544)
1547def merge_into_existing(
1548 existing: Any, kwargs: dict[str, Any], pilot_user_id: int, *, fill_only: bool
1549) -> None:
1550 """Apply a parsed row's *kwargs* onto *existing* (a Flight), as the same
1551 real-world flight. Caller must commit.
1553 *fill_only* (an automatic merge with an entry logged from the airframe
1554 side) only fills fields that are still empty. Otherwise (the review
1555 step's "overwrite") the row's values replace the existing ones. Either
1556 way an empty cell in the file never clears data already on the entry —
1557 it means "not recorded here", e.g. the pilot logbook has no column for
1558 the airframe log's remarks — and a managed-aircraft row never gets the
1559 free-text other_aircraft_* fields (its aircraft comes from the Aircraft
1560 join, see link_entries_to_aircraft).
1561 """
1562 for field_name, value in kwargs.items():
1563 if value is None or field_name in _MERGE_SKIP_FIELDS:
1564 continue
1565 if existing.aircraft_id is not None and field_name in _OTHER_AIRCRAFT_FIELDS:
1566 continue
1567 current = getattr(existing, field_name)
1568 if field_name == "notes" and current and current != value:
1569 if fill_only and value not in current:
1570 existing.notes = f"{current}\n{value}"
1571 elif not fill_only:
1572 existing.notes = value
1573 continue
1574 if not fill_only or current is None:
1575 setattr(existing, field_name, value)
1577 ident: dict[str, Any] = dict(kwargs)
1578 _assign_pilot_identity(ident, pilot_user_id)
1579 existing.pic_user_id = ident.get("pic_user_id")
1580 existing.second_crew_user_id = ident.get("second_crew_user_id")
1581 existing.second_crew_role = ident.get("second_crew_role")
1582 _fill_engine_time(existing)
1585def _fill_engine_time(entry: Any) -> None:
1586 """A managed aircraft's engine_time (engine-run hours, summed for
1587 engine/propeller TBO) from the pilot's logged block time, when nothing
1588 better is known yet. Block time is engine time — the engine runs from
1589 block-off to block-on — whereas the airborne flight_time side is never
1590 derived from a personal logbook (see link_entries_to_aircraft)."""
1591 if entry.aircraft_id is None or entry.engine_time is not None:
1592 return
1593 parts = [entry.single_pilot_se, entry.single_pilot_me, entry.multi_pilot]
1594 vals = [float(p) for p in parts if p is not None]
1595 if vals:
1596 entry.engine_time = round(sum(vals), 1)
1599def _is_airframe_only(existing: Any) -> bool:
1600 """An entry logged only from the airframe side: on a managed aircraft,
1601 no pilot claimed it and no pilot figures recorded yet."""
1602 return (
1603 existing.aircraft_id is not None
1604 and existing.pic_user_id is None
1605 and existing.second_crew_user_id is None
1606 and existing.single_pilot_se is None
1607 and existing.single_pilot_me is None
1608 and existing.multi_pilot is None
1609 )
1612# Combined score of the two cross-pair time checks (block-off vs takeoff,
1613# block-on vs landing, each 0–1) needed to merge automatically: both inside
1614# their offset band, or one inside and the other within the first ring.
1615_AUTO_MERGE_MIN_TIME_SCORE = 1.75
1618def _is_strong_cross_match(kwargs: dict[str, Any], existing: Any) -> bool:
1619 """Same aircraft, same known route, and both clock times within the
1620 taxi window (the aircraft's block-to-airborne offset) of the airframe
1621 row's takeoff/landing times — strong enough to merge without asking."""
1622 from services.time_band_matching import ( # pyright: ignore[reportMissingImports]
1623 taxi_in_score,
1624 taxi_out_score,
1625 )
1627 reg_new = (kwargs.get("other_aircraft_registration") or "").strip().upper()
1628 if reg_new.replace("-", "") != (
1629 existing.display_registration or ""
1630 ).upper().replace("-", ""):
1631 return False
1632 dep, arr = (
1633 _place_key(kwargs.get("departure_icao")),
1634 _place_key(kwargs.get("arrival_icao")),
1635 )
1636 if not dep or not arr:
1637 return False
1638 if (dep, arr) != (
1639 _place_key(existing.departure_icao),
1640 _place_key(existing.arrival_icao),
1641 ):
1642 return False
1643 offset_hours = float(existing.aircraft.flight_counter_offset)
1644 score = taxi_out_score(
1645 kwargs.get("departure_time"), existing.takeoff_time, offset_hours
1646 ) + taxi_in_score(existing.landing_time, kwargs.get("arrival_time"), offset_hours)
1647 return score >= _AUTO_MERGE_MIN_TIME_SCORE
1650def _is_no_takeoff_match(kwargs: dict[str, Any], existing: Any) -> bool:
1651 """*kwargs* (a parsed pilot-log row) and *existing* (a flight logged only
1652 from the airframe side, on the same registration) are the same engine
1653 run that never took off — see services.no_takeoff_matching."""
1654 from services.no_takeoff_matching import ( # pyright: ignore[reportMissingImports]
1655 PilotSide,
1656 airframe_side_of_flight,
1657 is_no_takeoff_pair,
1658 )
1660 if not _is_airframe_only(existing):
1661 return False
1662 reg_new = (kwargs.get("other_aircraft_registration") or "").upper()
1663 reg_old = (existing.display_registration or "").upper()
1664 if not reg_new or reg_new.replace("-", "") != reg_old.replace("-", ""):
1665 return False
1666 pilot = PilotSide(
1667 departure_icao=kwargs.get("departure_icao"),
1668 arrival_icao=kwargs.get("arrival_icao"),
1669 departure_time=kwargs.get("departure_time"),
1670 arrival_time=kwargs.get("arrival_time"),
1671 block_time=_kwargs_duration(kwargs),
1672 landings_day=kwargs.get("landings_day"),
1673 landings_night=kwargs.get("landings_night"),
1674 )
1675 return is_no_takeoff_pair(airframe_side_of_flight(existing), pilot)
1678def plan_bulk_overwrites(
1679 conflicts: list[Any], resolved: dict[str, str]
1680) -> tuple[list[tuple[Any, int]], list[Any]]:
1681 """For the review page's "overwrite all remaining" action: pair each
1682 remaining conflict row (a ConflictRow or AirframeConflictRow — anything
1683 with .candidates sorted best match first) with its best-scoring
1684 candidate. Returns (planned, left_for_review).
1686 Two rows never overwrite the same existing flight: a candidate already
1687 overwritten during this review (a decision in *resolved*) or claimed by
1688 an earlier row in this pass is off limits, and a row whose best match
1689 is taken is left for a manual decision rather than falling back to a
1690 weaker candidate that may well be the wrong flight.
1691 """
1692 claimed = {
1693 int(decision.split(":", 1)[1])
1694 for decision in resolved.values()
1695 if decision.startswith("overwrite:")
1696 }
1697 planned: list[tuple[Any, int]] = []
1698 left: list[Any] = []
1699 for conflict in conflicts:
1700 best_id = conflict.candidates[0][1]
1701 if best_id in claimed:
1702 left.append(conflict)
1703 continue
1704 claimed.add(best_id)
1705 planned.append((conflict, best_id))
1706 return planned, left
1709def split_auto_merges(
1710 conflicts: list[ConflictRow],
1711) -> tuple[list[tuple[ConflictRow, int]], list[ConflictRow]]:
1712 """Split near-match *conflicts* into (auto_merges, needs_review).
1714 A row merges automatically — no human decision needed — when exactly
1715 one of its candidates is an entry logged only from the airframe side
1716 (see _is_airframe_only) and that candidate is a strong match (see
1717 _is_strong_cross_match, or _is_no_takeoff_match for an engine run that
1718 never took off): the two sides then simply complement each
1719 other, there's nothing to choose between. Each existing entry is
1720 claimed by at most one row; anything else goes to review.
1721 """
1722 from models import Flight, db # pyright: ignore[reportMissingImports]
1724 auto: list[tuple[ConflictRow, int]] = []
1725 review: list[ConflictRow] = []
1726 claimed: set[int] = set()
1727 for conflict in conflicts:
1728 strong = [
1729 cid
1730 for _score, cid in conflict.candidates
1731 if (existing := db.session.get(Flight, cid)) is not None
1732 and _is_airframe_only(existing)
1733 and (
1734 _is_strong_cross_match(conflict.kwargs, existing)
1735 or _is_no_takeoff_match(conflict.kwargs, existing)
1736 )
1737 ]
1738 if len(strong) == 1 and strong[0] not in claimed:
1739 claimed.add(strong[0])
1740 auto.append((conflict, strong[0]))
1741 else:
1742 review.append(conflict)
1743 return auto, review
1746def link_entries_to_aircraft(entries: list[Any]) -> int:
1747 """Promote each standalone Flight row (aircraft_id NULL,
1748 other_aircraft_registration set) in *entries* to a managed aircraft, for
1749 any whose registration matches an Aircraft belonging to one of the row's
1750 pic/second-crew pilot's own tenants. Returns the count promoted. Caller
1751 must commit.
1753 Unified-model note: this used to create a brand-new FlightEntry +
1754 FlightCrew and link the pilot's existing PilotLogbookEntry to it via
1755 flight_id. There's only one row now, so promotion is just updating
1756 aircraft_id (+ clearing the free-text other_aircraft_* fields) on the
1757 row that already exists — no second row, no separate crew row (the
1758 row's pic_user_id/pic_name are already whatever the CSV import set).
1759 """
1760 from models import ( # pyright: ignore[reportMissingImports]
1761 Aircraft,
1762 TenantUser,
1763 User,
1764 db,
1765 )
1767 def _norm_reg(reg: str) -> str:
1768 return reg.upper().replace("-", "").replace(" ", "")
1770 ac_by_tenant: dict[int, dict[str, Any]] = {}
1771 for ac in Aircraft.query.all():
1772 ac_by_tenant.setdefault(ac.tenant_id, {})[_norm_reg(ac.registration)] = ac
1774 pilot_tenant_ids: dict[int, set[int]] = {}
1776 def _tenants_for_pilot(pilot_user_id: int) -> set[int]:
1777 if pilot_user_id not in pilot_tenant_ids:
1778 pilot_tenant_ids[pilot_user_id] = {
1779 row.tenant_id
1780 for row in TenantUser.query.filter_by(user_id=pilot_user_id).all()
1781 }
1782 return pilot_tenant_ids[pilot_user_id]
1784 def _place_icao(place: str | None) -> str:
1785 if not place:
1786 return "ZZZZ"
1787 clean = re.sub(r"[^A-Z0-9]", "", place.upper())[:4]
1788 return clean if len(clean) == 4 else "ZZZZ"
1790 promoted = 0
1791 for entry in entries:
1792 if entry.aircraft_id is not None or not entry.other_aircraft_registration:
1793 continue
1794 pilot_user_id = entry.pic_user_id or entry.second_crew_user_id
1795 if pilot_user_id is None:
1796 continue
1797 norm_reg = _norm_reg(entry.other_aircraft_registration)
1798 ac = None
1799 for tid in _tenants_for_pilot(pilot_user_id):
1800 ac = ac_by_tenant.get(tid, {}).get(norm_reg)
1801 if ac is not None:
1802 break
1803 if ac is None:
1804 continue
1806 entry.aircraft_id = ac.id
1807 entry.departure_icao = _place_icao(entry.departure_icao)
1808 entry.arrival_icao = _place_icao(entry.arrival_icao)
1809 entry.other_aircraft_type = None
1810 entry.other_aircraft_type_icao = None
1811 entry.other_aircraft_registration = None
1812 entry.source = "logbook_import"
1814 # The airframe-side figures (flight_time_counter_*/flight_time —
1815 # airborne time) and the engine counter readings aren't known from
1816 # a personal logbook and stay NULL: no supposition, interpolation,
1817 # or guessed counter value gets invented here (a previous version
1818 # seeded a rough engine_time_counter estimate, which
1819 # services/component_limits.py would then have summed for TBO).
1820 # flight_time staying NULL is also what keeps the flight list's
1821 # needs-attention icon (source == "logbook_import" and flight_time
1822 # is None) lit until a human, or an airframe-log import, fills it
1823 # in. The pilot's block time is a real engine-run figure, though,
1824 # so it does go to engine_time.
1825 _fill_engine_time(entry)
1827 if not entry.pic_name:
1828 pilot_user = db.session.get(User, pilot_user_id)
1829 if pilot_user:
1830 entry.pic_name = pilot_user.name or pilot_user.email
1832 promoted += 1
1834 return promoted