Coverage for app/pilots/logbook_import.py: 41%
620 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-01 12:21 +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 dataclasses import dataclass, field
12from datetime import date, datetime, time, timedelta
13from typing import Any, Callable
15import openpyxl # pyright: ignore[reportMissingImports]
16from flask_babel import gettext as _ # pyright: ignore[reportMissingImports]
17from flask_babel import pgettext # pyright: ignore[reportMissingImports]
19# ── Constants ─────────────────────────────────────────────────────────────────
21TARGET_FIELDS: list[str] = [
22 "date",
23 "aircraft_type",
24 "aircraft_registration",
25 "departure_place",
26 "departure_time",
27 "arrival_place",
28 "arrival_time",
29 "pic_name",
30 "night_time",
31 "instrument_time",
32 "cross_country",
33 "landings_day",
34 "landings_night",
35 "single_pilot_se",
36 "single_pilot_me",
37 "multi_pilot",
38 "function_pic",
39 "function_copilot",
40 "function_dual",
41 "function_instructor",
42 "remarks",
43 # virtual target — used for import validation only, not stored
44 "total_flight_time_check",
45]
47# Normalised source column name → target field.
48# Positional disambiguation suffixes (_2, _3 …) are applied before lookup,
49# so "time" is the first TIME column (departure) and "time_2" is the second (arrival).
50_ALIASES: dict[str, str] = {
51 # EASA standard logbook column names (normalised)
52 "date dd/mm/yy": "date",
53 "date": "date",
54 "aircraft type": "aircraft_type",
55 "type": "aircraft_type",
56 "aircraft registration number": "aircraft_registration",
57 "registration": "aircraft_registration",
58 "reg": "aircraft_registration",
59 "from": "departure_place",
60 "departure": "departure_place",
61 "dep": "departure_place",
62 "time": "departure_time", # first TIME → departure
63 "time_2": "arrival_time", # second TIME → arrival
64 "departure time": "departure_time",
65 "arrival time": "arrival_time",
66 "to": "arrival_place",
67 "arrival": "arrival_place",
68 "arr": "arrival_place",
69 "pic name": "pic_name",
70 "pic name (if not student pilot)": "pic_name",
71 "captain": "pic_name",
72 # Landings group — first DAY / NIGHT pair
73 "day": "landings_day",
74 "night": "landings_night",
75 # Operational conditions group — second DAY / NIGHT pair (positional suffix)
76 "day_2": "ignore", # cross-country day — not a logbook field
77 "night_2": "night_time", # night flying time
78 # Aircraft category
79 "se": "single_pilot_se",
80 "single engine": "single_pilot_se",
81 "single pilot se": "single_pilot_se",
82 "me": "single_pilot_me",
83 "multi engine": "single_pilot_me",
84 "single pilot me": "single_pilot_me",
85 "multi pilot": "multi_pilot",
86 "mp": "multi_pilot",
87 # Pilot function
88 "pic": "function_pic",
89 "p1": "function_pic",
90 "co-pic": "function_copilot",
91 "co-pilot": "function_copilot",
92 "copilot": "function_copilot",
93 "p2": "function_copilot",
94 "dual received": "function_dual",
95 "dual": "function_dual",
96 "student": "function_dual",
97 "instructor": "function_instructor",
98 "fi": "function_instructor",
99 # Cross-country time
100 "cross-country": "cross_country",
101 "cross country": "cross_country",
102 "x-country": "cross_country",
103 "xc": "cross_country",
104 # Total flight time — imported only to validate against computed sum
105 "total flight time": "total_flight_time_check",
106 # Catch-all
107 "total": "ignore",
108 "no. istr. appr.": "ignore",
109 "ifr approaches": "ignore",
110 "page": "ignore",
111 "line": "ignore",
112 "remarks": "remarks",
113 "notes": "remarks",
114 "comments": "remarks",
115 "instrument time": "instrument_time",
116 "ifr": "instrument_time",
117 "instrument": "instrument_time",
118 "night time": "night_time",
119 # Group-prefixed column names (e.g. Belgian EASA logbook with a span header row)
120 "departure & arrival from": "departure_place",
121 "departure & arrival time": "departure_time",
122 "departure & arrival time_2": "arrival_time",
123 "departure & arrival to": "arrival_place",
124 "aircraft category se": "single_pilot_se",
125 "aircraft category me": "single_pilot_me",
126 "operational conditions cross-country": "cross_country",
127 "operational conditions day": "ignore",
128 "operational conditions night": "night_time",
129 "pilot function pic": "function_pic",
130 "pilot function co-pic": "function_copilot",
131 "pilot function dual received": "function_dual",
132 "page subtotals total flight time": "ignore",
133 "page subtotals total flight time_2": "ignore",
134 "page subtotals pic": "ignore",
135 "page subtotals dual": "ignore",
136 "page subtotals night": "ignore",
137 "page subtotals landings – day": "ignore",
138 "page subtotals landings – night": "ignore",
139 "page subtotals formated date": "ignore",
140 "page subtotals formated type": "ignore",
141 "page subtotals days since flight": "ignore",
142 "page subtotals daytime duration": "ignore",
143 "page subtotals non pic or dual": "ignore",
144 "formated date": "ignore",
145 "formated type": "ignore",
146 "days since flight": "ignore",
147 "daytime duration": "ignore",
148 "non pic or dual": "ignore",
149 # Landings group (group-prefixed)
150 "landings day": "landings_day",
151 "landings night": "landings_night",
152 # Aircraft type (multiline cell name normalised to single space)
153 "aircraft type name, model, variant": "aircraft_type",
154}
156# ── Data structures ───────────────────────────────────────────────────────────
159@dataclass
160class ParsedFile:
161 """Result of parsing an uploaded logbook file."""
163 norm_cols: list[str] # normalised + disambiguated column keys
164 raw_cols: list[str] # original column labels (for display)
165 header_row_index: int # 0-based row index of the detected header
166 data_rows: list[list[Any]] # all rows after the header (including subtotals)
167 fingerprint: str # SHA-256 of norm_cols
170@dataclass
171class MappingProposal:
172 """Proposed column mapping and how it was derived."""
174 mapping: dict[str, str] # norm_col_key → target_field or "ignore"
175 match_type: str # "exact", "fuzzy", "alias"
176 fuzzy_score: float = 0.0 # 0.0–1.0, only meaningful for "fuzzy"
177 matched_mapping_id: int | None = None
180@dataclass
181class ImportResult:
182 """Summary returned after executing an import."""
184 imported: int = 0
185 subtotals: int = 0
186 skipped: list[tuple[int, str]] = field(default_factory=list) # (row_num, reason)
187 # (row_num, reason) — rows that matched an entry already in the pilot's logbook
188 duplicates: list[tuple[int, str]] = field(default_factory=list)
189 # (row_num, source_col, target_field, repr(raw)) — non-empty cells that couldn't parse
190 parse_warnings: list[tuple[int, str, str, str]] = field(default_factory=list)
191 # (row_num, source_total, computed_total) — rows where total ≠ sum of components
192 total_mismatch_warnings: list[tuple[int, float, float]] = field(
193 default_factory=list
194 )
195 has_opening_balance: bool = False
198# ── Normalisation ─────────────────────────────────────────────────────────────
201def _norm(text: str) -> str:
202 """Strip, collapse whitespace, lower-case."""
203 return re.sub(r"\s+", " ", str(text).strip().lower())
206def _disambiguate(names: list[str]) -> list[str]:
207 """Append _2, _3 … to duplicate names to make them unique."""
208 seen: dict[str, int] = {}
209 result: list[str] = []
210 for n in names:
211 if n in seen:
212 seen[n] += 1
213 result.append(f"{n}_{seen[n]}")
214 else:
215 seen[n] = 1
216 result.append(n)
217 return result
220def _fingerprint(norm_cols: list[str]) -> str:
221 payload = json.dumps(norm_cols, ensure_ascii=False, sort_keys=False)
222 return hashlib.sha256(payload.encode()).hexdigest()
225# ── Header detection ──────────────────────────────────────────────────────────
228def _is_header_row(row: list[Any]) -> bool:
229 """True if ≥ 50 % of non-empty cells are non-numeric strings and ≥ 4 non-empty."""
230 non_empty = [c for c in row if c is not None and str(c).strip()]
231 if len(non_empty) < 4:
232 return False
233 string_like = [
234 c for c in non_empty if isinstance(c, str) and not _is_numeric_str(str(c))
235 ]
236 return len(string_like) / len(non_empty) >= 0.5
239def _is_numeric_str(s: str) -> bool:
240 try:
241 float(s)
242 return True
243 except ValueError:
244 return False
247def _header_alias_score(row: list[Any]) -> int:
248 """Count how many cells in *row* match a known alias."""
249 return sum(1 for c in row if c is not None and _norm(str(c)) in _ALIASES)
252def _find_header_row(rows: list[list[Any]], max_scan: int = 20) -> int | None:
253 """Return 0-based index of the best header row within the first max_scan rows.
255 Many logbook templates have a group-label row (e.g. "DEPARTURE & ARRIVAL",
256 "LANDINGS") above the actual column-header row. Both pass _is_header_row,
257 but the actual header row has far more alias matches. We score every
258 candidate and return the one with the highest score; ties go to the earlier
259 row. If no alias matches are found we fall back to the first text-like row.
260 """
261 best_idx: int | None = None
262 best_score = -1
263 first_text_row: int | None = None
265 for i, row in enumerate(rows[:max_scan]):
266 if not _is_header_row(row):
267 continue
268 if first_text_row is None:
269 first_text_row = i
270 score = _header_alias_score(row)
271 if score > best_score:
272 best_score = score
273 best_idx = i
275 return best_idx if (best_idx is not None and best_score > 0) else first_text_row
278def _trim_trailing_empty_cols(
279 all_rows: list[list[Any]], max_scan: int = 50
280) -> list[list[Any]]:
281 """Trim columns beyond the rightmost non-empty value in the first max_scan rows.
283 Excel templates often declare thousands of formatted-but-empty columns.
284 Without trimming, every empty cell becomes a separate mapping entry.
285 """
286 max_col = 0
287 for row in all_rows[:max_scan]:
288 for j in range(len(row) - 1, -1, -1):
289 if row[j] is not None and str(row[j]).strip():
290 if j + 1 > max_col:
291 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 # fall back to heuristic in _build_parsed_file
479 pass
481 return _build_parsed_file(all_rows, filename, excel_merge_map=excel_merge_map)
484def _parse_csv(data: bytes, filename: str) -> ParsedFile:
485 try:
486 text = data.decode("utf-8-sig")
487 except UnicodeDecodeError:
488 text = data.decode("latin-1")
490 sample = text[:4096]
491 try:
492 dialect = csv.Sniffer().sniff(sample)
493 except csv.Error:
494 dialect = csv.excel
496 reader = csv.reader(io.StringIO(text), dialect)
497 try:
498 all_rows: list[list[Any]] = list(reader)
499 except csv.Error as exc:
500 raise ValueError(f"Could not parse CSV file: {exc}") from exc
501 return _build_parsed_file(all_rows, filename)
504def _build_parsed_file(
505 all_rows: list[list[Any]],
506 filename: str,
507 excel_merge_map: dict[tuple[int, int], str] | None = None,
508) -> ParsedFile:
509 all_rows = _trim_trailing_empty_cols(all_rows)
510 header_idx = _find_header_row(all_rows)
511 if header_idx is None:
512 raise ValueError(
513 "Could not detect a header row in this file. "
514 "Make sure the file contains column names."
515 )
517 raw_header = [str(c).strip() if c is not None else "" for c in all_rows[header_idx]]
519 # Prepend group labels from the row above the header, if one exists
520 if header_idx > 0:
521 width = len(raw_header)
522 if excel_merge_map is not None:
523 # Excel: exact boundaries from merged cell metadata
524 _gl = _group_labels_from_map(excel_merge_map, header_idx - 1, width)
525 group_labels: list[str] | None = _gl if any(_gl) else None
526 else:
527 # CSV: heuristic forward-fill (merged cell info unavailable)
528 group_labels = _group_labels_heuristic(all_rows[header_idx - 1], width)
529 if group_labels is not None:
530 raw_header = _apply_group_labels(group_labels, raw_header)
532 norm_raw = [_norm(c) for c in raw_header]
533 norm_cols = _disambiguate(norm_raw)
534 data_rows = all_rows[header_idx + 1 :]
536 return ParsedFile(
537 norm_cols=norm_cols,
538 raw_cols=raw_header,
539 header_row_index=header_idx,
540 data_rows=data_rows,
541 fingerprint=_fingerprint(norm_cols),
542 )
545# ── Mapping proposal ──────────────────────────────────────────────────────────
548def propose_mapping(
549 parsed: ParsedFile,
550 existing_mappings: list[Any], # list[LogbookImportMapping]
551) -> MappingProposal:
552 """Return the best mapping proposal for *parsed*, checking saved mappings first."""
553 # 1. Exact fingerprint match
554 for m in existing_mappings:
555 if m.source_fingerprint == parsed.fingerprint:
556 return MappingProposal(
557 mapping=json.loads(m.column_mapping),
558 match_type="exact",
559 matched_mapping_id=m.id,
560 )
562 # 2. Fuzzy match — best overlap among saved mappings
563 best_score = 0.0
564 best_m = None
565 new_set = set(parsed.norm_cols)
566 for m in existing_mappings:
567 saved_cols: list[str] = json.loads(m.source_columns)
568 saved_set = set(saved_cols)
569 if not saved_set:
570 continue
571 overlap = len(new_set & saved_set)
572 score = overlap / max(len(new_set), len(saved_set))
573 if score > best_score:
574 best_score = score
575 best_m = m
577 if best_m is not None and best_score >= 0.6:
578 saved_map: dict[str, str] = json.loads(best_m.column_mapping)
579 # Build a mapping for the new columns, falling back to alias for unmatched ones
580 merged = _alias_mapping(parsed.norm_cols)
581 for col in parsed.norm_cols:
582 if col in saved_map:
583 merged[col] = saved_map[col]
584 return MappingProposal(
585 mapping=merged,
586 match_type="fuzzy",
587 fuzzy_score=best_score,
588 matched_mapping_id=best_m.id,
589 )
591 # 3. Alias-only auto-mapping
592 return MappingProposal(
593 mapping=_alias_mapping(parsed.norm_cols),
594 match_type="alias",
595 )
598def _alias_mapping(norm_cols: list[str]) -> dict[str, str]:
599 """Apply the built-in alias table; unknown columns default to 'ignore'."""
600 result: dict[str, str] = {}
601 for col in norm_cols:
602 result[col] = _ALIASES.get(col, "ignore")
603 return result
606# ── Subtotal detection ────────────────────────────────────────────────────────
609def _is_subtotal_row(row: list[Any], date_col_idx: int | None) -> bool:
610 if date_col_idx is None:
611 return False
612 if date_col_idx >= len(row):
613 return True
614 val = row[date_col_idx]
615 if isinstance(val, timedelta):
616 return True
617 if val is None or (isinstance(val, str) and not val.strip()):
618 return True
619 if isinstance(val, str) and "total" in val.lower():
620 return True
621 return False
624# ── Value parsing ─────────────────────────────────────────────────────────────
627def parse_date_value(val: Any) -> date | None:
628 if isinstance(val, datetime):
629 return val.date()
630 if isinstance(val, date):
631 return val
632 if isinstance(val, (int, float)):
633 # Excel serial date number — openpyxl returns datetime; guard anyway
634 return None
635 if not isinstance(val, str):
636 return None
637 s = val.strip()
638 if not s:
639 return None
640 for fmt in ("%d/%m/%y", "%d/%m/%Y", "%Y-%m-%d", "%m/%d/%Y", "%d-%m-%Y"):
641 try:
642 return datetime.strptime(s, fmt).date()
643 except ValueError:
644 pass
645 return None
648def parse_time_value(val: Any) -> time | None:
649 """Parse a time-of-day value (HH:MM or Python time)."""
650 if isinstance(val, time):
651 return val
652 if isinstance(val, datetime):
653 return val.time()
654 if not isinstance(val, str):
655 return None
656 s = val.strip()
657 if not s:
658 return None
659 m = re.match(r"^(\d{1,2}):(\d{2})$", s)
660 if m:
661 try:
662 return time(int(m.group(1)), int(m.group(2)))
663 except ValueError:
664 return None
665 return None
668def parse_duration_value(val: Any) -> float | None:
669 """Parse a duration into decimal hours (e.g. time(0,42) → 0.7, '1:24' → 1.4)."""
670 if isinstance(val, timedelta):
671 hours = val.total_seconds() / 3600
672 return round(hours, 1) if hours >= 0 else None
673 if isinstance(val, time):
674 return round(val.hour + val.minute / 60, 1)
675 if isinstance(val, datetime):
676 # Excel sometimes returns a dummy date + the time-of-day
677 t = val.time()
678 return round(t.hour + t.minute / 60, 1)
679 if isinstance(val, (int, float)):
680 if not math.isfinite(val) or val < 0:
681 return None
682 return round(float(val), 1)
683 if isinstance(val, str):
684 s = val.strip()
685 if not s:
686 return None
687 m = re.match(r"^(\d+):(\d{2})$", s)
688 if m:
689 return round(int(m.group(1)) + int(m.group(2)) / 60, 1)
690 try:
691 v = float(s)
692 return round(v, 1) if math.isfinite(v) and v >= 0 else None
693 except ValueError:
694 return None
695 return None
698def parse_int_value(val: Any) -> int | None:
699 if isinstance(val, int):
700 return val if val >= 0 else None
701 if isinstance(val, float):
702 try:
703 return int(val) if val >= 0 else None
704 except (ValueError, OverflowError):
705 return None
706 if isinstance(val, str):
707 s = val.strip()
708 if not s:
709 return None
710 try:
711 n = int(float(s))
712 except (ValueError, OverflowError):
713 return None
714 return n if n >= 0 else None
715 return None
718def _is_nonempty(val: Any) -> bool:
719 """Return True when val carries a real value (not None, not blank string)."""
720 if val is None:
721 return False
722 if isinstance(val, str):
723 return bool(val.strip())
724 return True
727# Map target field → (human-readable type name, parser function)
728_FIELD_TYPE: dict[str, tuple[str, Callable[[Any], Any]]] = {
729 "date": ("date", parse_date_value),
730 "departure_time": ("time", parse_time_value),
731 "arrival_time": ("time", parse_time_value),
732 "night_time": ("duration", parse_duration_value),
733 "instrument_time": ("duration", parse_duration_value),
734 "cross_country": ("duration", parse_duration_value),
735 "total_flight_time_check": ("duration", parse_duration_value),
736 "single_pilot_se": ("duration", parse_duration_value),
737 "single_pilot_me": ("duration", parse_duration_value),
738 "multi_pilot": ("duration", parse_duration_value),
739 "function_pic": ("duration", parse_duration_value),
740 "function_copilot": ("duration", parse_duration_value),
741 "function_dual": ("duration", parse_duration_value),
742 "function_instructor": ("duration", parse_duration_value),
743 "landings_day": ("integer", parse_int_value),
744 "landings_night": ("integer", parse_int_value),
745}
747_HINT_SAMPLE_ROWS = 5
750def type_hints(parsed: ParsedFile, mapping: dict[str, str]) -> dict[str, str]:
751 """Return {col: hint_text} for columns where sample data doesn't match the proposed type.
753 Samples up to _HINT_SAMPLE_ROWS rows. Returns a hint when non-empty values are
754 present but any of them fail to parse — indicating a likely mapping mismatch.
755 """
756 col_index = {col: i for i, col in enumerate(parsed.norm_cols)}
757 hints: dict[str, str] = {}
758 for col, target in mapping.items():
759 if target not in _FIELD_TYPE:
760 continue
761 type_name, parser = _FIELD_TYPE[target]
762 idx = col_index.get(col)
763 if idx is None:
764 continue
765 sample = [
766 row[idx]
767 for row in parsed.data_rows[:_HINT_SAMPLE_ROWS]
768 if idx < len(row) and _is_nonempty(row[idx])
769 ]
770 if not sample:
771 continue
772 failed = [v for v in sample if parser(v) is None]
773 if failed:
774 example = str(failed[0])[:30]
775 hints[col] = (
776 f"Sample data doesn't look like a {type_name} (e.g. {example!r})"
777 )
778 return hints
781# ── Preview rows ──────────────────────────────────────────────────────────────
784def preview_rows(
785 parsed: ParsedFile,
786 mapping: dict[str, str],
787 n: int = 5,
788) -> list[dict[str, Any]]:
789 """Return up to *n* non-subtotal data rows mapped to target field names."""
790 date_idx = _date_col_index(parsed.norm_cols, mapping)
791 result: list[dict[str, Any]] = []
792 for row in parsed.data_rows:
793 if _is_subtotal_row(row, date_idx):
794 continue
795 mapped: dict[str, Any] = {}
796 for i, col in enumerate(parsed.norm_cols):
797 target = mapping.get(col, "ignore")
798 if target == "ignore":
799 continue
800 mapped[target] = row[i] if i < len(row) else None
801 result.append(mapped)
802 if len(result) >= n:
803 break
804 return result
807def _date_col_index(norm_cols: list[str], mapping: dict[str, str]) -> int | None:
808 for i, col in enumerate(norm_cols):
809 if mapping.get(col) == "date":
810 return i
811 return None
814# ── Import execution ──────────────────────────────────────────────────────────
817def _row_get(row: list[Any], col_index: dict[str, int], col: str) -> Any:
818 i = col_index.get(col)
819 return row[i] if i is not None and i < len(row) else None
822def _parse_row_date(
823 row: list[Any], mapping: dict[str, str], col_index: dict[str, int]
824) -> date | None:
825 for col, target in mapping.items():
826 if target == "date":
827 return parse_date_value(_row_get(row, col_index, col))
828 return None
831def _build_entry_kwargs(
832 row: list[Any],
833 mapping: dict[str, str],
834 col_index: dict[str, int],
835 date_val: date,
836) -> tuple[dict[str, Any], float | None, list[tuple[str, str, str]]]:
837 """Build Flight (standalone, aircraft_id NULL) field kwargs for one data
838 row (identity/batch fields like pic_user_id are the caller's
839 responsibility to add).
841 Returns (kwargs, source_total_flight_time_check, parse_warnings), where
842 parse_warnings is a list of (col, target, raw_repr) for non-empty cells
843 that couldn't be parsed as their target field's type. Shared by the
844 normal import pass and the near-match conflict finder below, so the two
845 can never compute a row's fields differently.
846 """
847 kwargs: dict[str, Any] = {"date": date_val}
848 source_total: float | None = None
849 parse_warnings: list[tuple[str, str, str]] = []
851 for col, target in mapping.items():
852 if target in ("ignore", "date"):
853 continue
854 raw = _row_get(row, col_index, col)
855 if target == "total_flight_time_check":
856 if _is_nonempty(raw):
857 source_total = parse_duration_value(raw)
858 continue
859 if target in _FIELD_TYPE:
860 # NB: unpack into _type_name, not _ — callers of this module also
861 # call gettext (imported as `_`) in the same function scope;
862 # reassigning `_` would shadow it for the rest of that function
863 # (Python's function-scoping, not block-scoping).
864 _type_name, parser = _FIELD_TYPE[target]
865 parsed_val = parser(raw)
866 if parsed_val is None and _is_nonempty(raw):
867 parse_warnings.append((col, target, repr(str(raw)[:40])))
868 kwargs[target] = parsed_val
869 elif target == "aircraft_type":
870 val = str(raw).strip() if raw is not None else None
871 kwargs["other_aircraft_type"] = val
872 if val:
873 from utils import resolve_aircraft_type_icao # pyright: ignore[reportMissingImports]
875 kwargs["other_aircraft_type_icao"] = resolve_aircraft_type_icao(val)
876 elif target == "aircraft_registration":
877 kwargs["other_aircraft_registration"] = (
878 str(raw).strip() if raw is not None else None
879 )
880 elif target == "departure_place":
881 kwargs["departure_icao"] = str(raw).strip() if raw is not None else None
882 elif target == "arrival_place":
883 kwargs["arrival_icao"] = str(raw).strip() if raw is not None else None
884 elif target == "pic_name":
885 kwargs["pic_name"] = str(raw).strip() if raw is not None else None
886 elif target == "remarks":
887 kwargs["notes"] = str(raw).strip() if raw is not None else None
889 return kwargs, source_total, parse_warnings
892def _num(v: Any) -> float | None:
893 # Numeric columns come back as decimal.Decimal; the freshly parsed side
894 # is always plain float (parse_duration_value). Decimal('0.7') != 0.7
895 # (float) directly — comparing Decimals converted from float literals
896 # hits binary-rounding mismatches — so normalise both sides to float
897 # before building/comparing keys.
898 return None if v is None else float(v)
901def _dup_key(kwargs: dict[str, Any]) -> tuple[Any, ...]:
902 """Exact-duplicate key: date + aircraft + duration + landings. Shared by
903 execute_import (silently skips exact matches) and find_conflicting_rows
904 (must not also flag them as a near-match conflict needing review)."""
905 return (
906 kwargs["date"],
907 kwargs.get("other_aircraft_registration"),
908 kwargs.get("single_pilot_se"),
909 kwargs.get("single_pilot_me"),
910 kwargs.get("multi_pilot"),
911 kwargs.get("landings_day"),
912 kwargs.get("landings_night"),
913 )
916def _fetch_existing_dedup_keys(pilot_user_id: int) -> set[tuple[Any, ...]]:
917 from sqlalchemy import or_ # pyright: ignore[reportMissingImports]
919 from models import Aircraft, Flight, db # pyright: ignore[reportMissingImports]
921 # Registration is Aircraft.registration for a managed-aircraft row, or
922 # the free-text other_aircraft_registration for a standalone one.
923 rows = (
924 db.session.query(
925 Flight.date,
926 Flight.other_aircraft_registration,
927 Aircraft.registration,
928 Flight.single_pilot_se,
929 Flight.single_pilot_me,
930 Flight.multi_pilot,
931 Flight.landings_day,
932 Flight.landings_night,
933 )
934 .outerjoin(Aircraft, Aircraft.id == Flight.aircraft_id)
935 .filter(
936 or_(
937 Flight.pic_user_id == pilot_user_id,
938 Flight.second_crew_user_id == pilot_user_id,
939 )
940 )
941 )
942 return {
943 (
944 row[0],
945 row[2] or row[1],
946 _num(row[3]),
947 _num(row[4]),
948 _num(row[5]),
949 row[6],
950 row[7],
951 )
952 for row in rows
953 }
956def _assign_pilot_identity(kwargs: dict[str, Any], pilot_user_id: int) -> None:
957 """Put *pilot_user_id* into the correct crew slot for one row (mutates
958 *kwargs* in place), based on its function_* breakdown.
960 A personal-logbook row's "PIC name" column (already in kwargs["pic_name"]
961 by the time this runs) records the actual PIC's name as free text,
962 independent of which slot the importing pilot occupies on that row. When
963 the row shows dual-received or copilot time and no PIC time, the
964 importing pilot was not the PIC — they belong in the second_crew_* slot
965 (as student or copilot) instead, leaving pic_user_id unset so pic_name's
966 free-text name remains the row's only PIC identity. Any other case (PIC
967 time logged, instructor time logged, or no function breakdown at all —
968 e.g. a source file with no "Pilot function" column) keeps pic_user_id as
969 the importing pilot, matching this app's original single-slot behaviour.
970 """
971 from models import CrewRole # pyright: ignore[reportMissingImports]
973 function_pic = kwargs.get("function_pic")
974 function_dual = kwargs.get("function_dual")
975 function_copilot = kwargs.get("function_copilot")
977 if not function_pic and function_dual:
978 kwargs["second_crew_user_id"] = pilot_user_id
979 kwargs["second_crew_role"] = CrewRole.STUDENT
980 elif not function_pic and function_copilot:
981 kwargs["second_crew_user_id"] = pilot_user_id
982 kwargs["second_crew_role"] = CrewRole.COPILOT
983 else:
984 kwargs["pic_user_id"] = pilot_user_id
987def execute_import(
988 parsed: ParsedFile,
989 mapping: dict[str, str],
990 pilot_user_id: int,
991 batch_id: int,
992 opening_balance: dict[str, Any] | None = None,
993 skip_row_nums: set[int] | None = None,
994) -> ImportResult:
995 """Create standalone Flight rows (aircraft_id NULL) from *parsed* using
996 *mapping*.
998 Returns an ImportResult describing what happened. Entries are added to
999 db.session but NOT committed — the caller commits after also saving the
1000 batch/mapping records. *skip_row_nums* (1-based, matching the row
1001 numbering used throughout this module) lets the caller carve out rows
1002 it's handling separately — e.g. near-match conflicts routed to the
1003 interactive review step in app/pilots/routes.py — so they're excluded
1004 entirely from this pass (not counted as imported, duplicate, or skipped).
1005 """
1006 from models import Flight, db # pyright: ignore[reportMissingImports]
1008 result = ImportResult()
1009 date_idx = _date_col_index(parsed.norm_cols, mapping)
1010 col_index = {col: i for i, col in enumerate(parsed.norm_cols)}
1011 skip_row_nums = skip_row_nums or set()
1013 entries_to_add: list[Flight] = []
1015 # Duplicate detection: re-importing the same file — either by mistake, or
1016 # deliberately after appending new rows to the source spreadsheet, which
1017 # is the expected workflow for keeping an import-based logbook current —
1018 # must only add genuinely new rows, not double up everything already
1019 # imported. Match on date + aircraft + the duration/landings figures
1020 # rather than departure/arrival place & time: those are the fields that
1021 # actually drive currency and compliance totals, and they stay populated
1022 # even for entries whose route/time genuinely isn't captured (e.g. a
1023 # hand-entered or backfilled logbook row) — a key that depended on
1024 # route/time would silently fail to match those.
1025 existing_keys = _fetch_existing_dedup_keys(pilot_user_id)
1027 for row_num, row in enumerate(parsed.data_rows, start=1):
1028 if row_num in skip_row_nums:
1029 continue
1030 if _is_subtotal_row(row, date_idx):
1031 result.subtotals += 1
1032 continue
1034 date_val = _parse_row_date(row, mapping, col_index)
1035 if date_val is None:
1036 raw_date = (
1037 row[date_idx] if date_idx is not None and date_idx < len(row) else None
1038 )
1039 result.skipped.append((row_num, f"unparseable date: {raw_date!r}"))
1040 continue
1042 kwargs, source_total, parse_warnings = _build_entry_kwargs(
1043 row, mapping, col_index, date_val
1044 )
1045 _assign_pilot_identity(kwargs, pilot_user_id)
1046 kwargs["import_batch_id"] = batch_id
1047 kwargs["source"] = "import"
1048 for col, target, raw_repr in parse_warnings:
1049 result.parse_warnings.append((row_num, col, target, raw_repr))
1051 dup_key = _dup_key(kwargs)
1052 if dup_key in existing_keys:
1053 result.duplicates.append(
1054 (
1055 row_num,
1056 _(
1057 "matches an entry already in your logbook "
1058 "(same date, aircraft, duration and landings)"
1059 ),
1060 )
1061 )
1062 continue
1063 existing_keys.add(dup_key)
1065 if source_total is not None:
1066 computed = round(
1067 sum(
1068 float(kwargs.get(f) or 0)
1069 for f in ("single_pilot_se", "single_pilot_me", "multi_pilot")
1070 ),
1071 1,
1072 )
1073 if abs(source_total - computed) >= 0.15:
1074 result.total_mismatch_warnings.append((row_num, source_total, computed))
1076 entries_to_add.append(Flight(**kwargs))
1078 result.imported = len(entries_to_add)
1079 for e in entries_to_add:
1080 db.session.add(e)
1082 # Opening balance — synthetic entry dated one day before earliest imported.
1083 # Only one opening-balance entry ever makes sense per pilot — re-importing
1084 # with opening-balance values filled in again must not create a second one.
1085 if opening_balance and any(v for v in opening_balance.values() if v):
1086 ob_exists = (
1087 db.session.query(Flight.id)
1088 .filter_by(pic_user_id=pilot_user_id, notes="Opening balance (imported)")
1089 .first()
1090 is not None
1091 )
1092 if ob_exists:
1093 result.duplicates.append(
1094 (
1095 0,
1096 _(
1097 "opening balance: an opening balance entry already "
1098 "exists for this pilot — not created again"
1099 ),
1100 )
1101 )
1102 else:
1103 earliest = (
1104 min(e.date for e in entries_to_add) if entries_to_add else date.today()
1105 )
1106 from datetime import timedelta as _td
1108 balance_date = earliest - _td(days=1)
1109 balance_entry = Flight(
1110 pic_user_id=pilot_user_id,
1111 import_batch_id=batch_id,
1112 source="import",
1113 date=balance_date,
1114 notes="Opening balance (imported)",
1115 night_time=opening_balance.get("night_time"),
1116 instrument_time=opening_balance.get("instrument_time"),
1117 single_pilot_se=opening_balance.get("single_pilot_se"),
1118 single_pilot_me=opening_balance.get("single_pilot_me"),
1119 multi_pilot=opening_balance.get("multi_pilot"),
1120 function_pic=opening_balance.get("function_pic"),
1121 function_copilot=opening_balance.get("function_copilot"),
1122 function_dual=opening_balance.get("function_dual"),
1123 function_instructor=opening_balance.get("function_instructor"),
1124 )
1125 db.session.add(balance_entry)
1126 result.has_opening_balance = True
1128 return result
1131# ── Near-match conflict detection (possible corrections) ───────────────────────
1133_CANDIDATE_MIN_SCORE = 3
1134_CANDIDATE_TIME_TOLERANCE_MINUTES = 120
1135_CANDIDATE_DURATION_TOLERANCE = 1.0
1138@dataclass
1139class ConflictRow:
1140 """A parsed row that isn't an exact duplicate but scores highly enough
1141 against one or more existing entries to plausibly be an edited version
1142 of one of them — needs a human decision, not a guess."""
1144 row_num: int
1145 kwargs: dict[str, Any]
1146 candidates: list[tuple[int, int]] # (score, existing_entry_id), best first
1149def _kwargs_duration(kwargs: dict[str, Any]) -> float | None:
1150 parts = [
1151 kwargs.get(f) for f in ("single_pilot_se", "single_pilot_me", "multi_pilot")
1152 ]
1153 vals = [float(p) for p in parts if p is not None]
1154 return round(sum(vals), 1) if vals else None
1157def _time_close(t1: time | None, t2: time | None, tolerance_minutes: int) -> bool:
1158 if t1 is None or t2 is None:
1159 return False
1160 m1 = t1.hour * 60 + t1.minute
1161 m2 = t2.hour * 60 + t2.minute
1162 return abs(m1 - m2) <= tolerance_minutes
1165def _score_candidate(kwargs: dict[str, Any], existing: Any) -> int:
1166 """Score how likely *existing* (a Flight row) is the same real-world
1167 flight as *kwargs* (a freshly parsed row), across 7 points: registration,
1168 departure, arrival, departure time, arrival time, total duration,
1169 landings. A point only counts toward the score if both sides have data
1170 for it — missing data on either side is neutral, never a mismatch, so a
1171 row whose route/time isn't captured on one side can still be recognised
1172 via duration + landings alone.
1173 """
1174 score = 0
1176 reg_new = (kwargs.get("other_aircraft_registration") or "").strip().upper()
1177 reg_old = (existing.display_registration or "").strip().upper()
1178 if reg_new and reg_old and reg_new == reg_old:
1179 score += 1
1181 dep_new = (kwargs.get("departure_icao") or "").strip().upper()
1182 dep_old = (existing.departure_icao or "").strip().upper()
1183 if dep_new and dep_old and dep_new == dep_old:
1184 score += 1
1186 arr_new = (kwargs.get("arrival_icao") or "").strip().upper()
1187 arr_old = (existing.arrival_icao or "").strip().upper()
1188 if arr_new and arr_old and arr_new == arr_old:
1189 score += 1
1191 if _time_close(
1192 kwargs.get("departure_time"),
1193 existing.departure_time,
1194 _CANDIDATE_TIME_TOLERANCE_MINUTES,
1195 ):
1196 score += 1
1198 if _time_close(
1199 kwargs.get("arrival_time"),
1200 existing.arrival_time,
1201 _CANDIDATE_TIME_TOLERANCE_MINUTES,
1202 ):
1203 score += 1
1205 dur_new = _kwargs_duration(kwargs)
1206 dur_old = existing.total_flight_time
1207 if (
1208 dur_new is not None
1209 and dur_old is not None
1210 and abs(dur_new - dur_old) <= _CANDIDATE_DURATION_TOLERANCE
1211 ):
1212 score += 1
1214 land_day = kwargs.get("landings_day")
1215 land_night = kwargs.get("landings_night")
1216 if (
1217 land_day is not None
1218 and land_night is not None
1219 and existing.landings_day is not None
1220 and existing.landings_night is not None
1221 and land_day == existing.landings_day
1222 and land_night == existing.landings_night
1223 ):
1224 score += 1
1226 return score
1229def find_conflicting_rows(
1230 parsed: ParsedFile,
1231 mapping: dict[str, str],
1232 pilot_user_id: int,
1233 exclude_row_nums: set[int] | None = None,
1234) -> list[ConflictRow]:
1235 """Find rows that aren't an exact duplicate but plausibly match an
1236 existing entry closely enough (score >= _CANDIDATE_MIN_SCORE) to need a
1237 human decision: keep the existing entry, overwrite it with the new data,
1238 or import as a genuinely separate new entry.
1240 Skips subtotal rows, rows with an unparseable date, and exact duplicates
1241 (same as execute_import's own dedup — an unmodified re-upload scores the
1242 maximum on every point, so without this check it would incorrectly be
1243 routed to review instead of being silently skipped), plus any row_num in
1244 *exclude_row_nums* — typically rows already resolved in an earlier pass
1245 of this same review.
1247 Candidates also include *unclaimed* rows (pic_user_id AND
1248 second_crew_user_id both NULL) on a managed aircraft belonging to one of
1249 the pilot's own tenants — typically a hand-entered airframe-log row from
1250 before this pilot ever ran a logbook import. Without this, those rows
1251 are invisible to both this check and execute_import's own dedup (both
1252 only look at rows already linked to *pilot_user_id*), so re-importing a
1253 personal logbook can never discover that the real-world flight is
1254 already logged from the airframe side — it silently creates a second,
1255 duplicate Flight row instead of surfacing a conflict to resolve.
1256 """
1257 from sqlalchemy import and_, or_ # pyright: ignore[reportMissingImports]
1259 from models import Aircraft, Flight, TenantUser # pyright: ignore[reportMissingImports]
1261 exclude_row_nums = exclude_row_nums or set()
1262 date_idx = _date_col_index(parsed.norm_cols, mapping)
1263 col_index = {col: i for i, col in enumerate(parsed.norm_cols)}
1264 existing_keys = _fetch_existing_dedup_keys(pilot_user_id)
1265 conflicts: list[ConflictRow] = []
1267 tenant_ids = [
1268 row.tenant_id for row in TenantUser.query.filter_by(user_id=pilot_user_id).all()
1269 ]
1271 for row_num, row in enumerate(parsed.data_rows, start=1):
1272 if row_num in exclude_row_nums:
1273 continue
1274 if _is_subtotal_row(row, date_idx):
1275 continue
1276 date_val = _parse_row_date(row, mapping, col_index)
1277 if date_val is None:
1278 continue
1280 kwargs, _source_total, _parse_warnings = _build_entry_kwargs(
1281 row, mapping, col_index, date_val
1282 )
1283 if _dup_key(kwargs) in existing_keys:
1284 continue # exact duplicate — execute_import's own dedup handles this
1286 same_day_filters = [
1287 Flight.pic_user_id == pilot_user_id,
1288 Flight.second_crew_user_id == pilot_user_id,
1289 ]
1290 if tenant_ids:
1291 same_day_filters.append(
1292 and_(
1293 Flight.pic_user_id.is_(None),
1294 Flight.second_crew_user_id.is_(None),
1295 Flight.aircraft_id.in_(
1296 Aircraft.query.with_entities(Aircraft.id).filter(
1297 Aircraft.tenant_id.in_(tenant_ids)
1298 )
1299 ),
1300 )
1301 )
1302 same_day = Flight.query.filter(
1303 or_(*same_day_filters),
1304 Flight.date == date_val,
1305 ).all()
1306 scored = [
1307 (score, existing.id)
1308 for existing in same_day
1309 if (score := _score_candidate(kwargs, existing)) >= _CANDIDATE_MIN_SCORE
1310 ]
1311 if scored:
1312 scored.sort(key=lambda t: -t[0])
1313 conflicts.append(
1314 ConflictRow(row_num=row_num, kwargs=kwargs, candidates=scored)
1315 )
1317 return conflicts
1320def link_entries_to_aircraft(entries: list[Any]) -> int:
1321 """Promote each standalone Flight row (aircraft_id NULL,
1322 other_aircraft_registration set) in *entries* to a managed aircraft, for
1323 any whose registration matches an Aircraft belonging to one of the row's
1324 pic/second-crew pilot's own tenants. Returns the count promoted. Caller
1325 must commit.
1327 Unified-model note: this used to create a brand-new FlightEntry +
1328 FlightCrew and link the pilot's existing PilotLogbookEntry to it via
1329 flight_id. There's only one row now, so promotion is just updating
1330 aircraft_id (+ clearing the free-text other_aircraft_* fields) on the
1331 row that already exists — no second row, no separate crew row (the
1332 row's pic_user_id/pic_name are already whatever the CSV import set).
1333 """
1334 from models import Aircraft, TenantUser, User, db # pyright: ignore[reportMissingImports]
1336 def _norm_reg(reg: str) -> str:
1337 return reg.upper().replace("-", "").replace(" ", "")
1339 ac_by_tenant: dict[int, dict[str, Any]] = {}
1340 for ac in Aircraft.query.all():
1341 ac_by_tenant.setdefault(ac.tenant_id, {})[_norm_reg(ac.registration)] = ac
1343 pilot_tenant_ids: dict[int, set[int]] = {}
1345 def _tenants_for_pilot(pilot_user_id: int) -> set[int]:
1346 if pilot_user_id not in pilot_tenant_ids:
1347 pilot_tenant_ids[pilot_user_id] = {
1348 row.tenant_id
1349 for row in TenantUser.query.filter_by(user_id=pilot_user_id).all()
1350 }
1351 return pilot_tenant_ids[pilot_user_id]
1353 def _place_icao(place: str | None) -> str:
1354 if not place:
1355 return "ZZZZ"
1356 clean = re.sub(r"[^A-Z0-9]", "", place.upper())[:4]
1357 return clean if len(clean) == 4 else "ZZZZ"
1359 promoted = 0
1360 for entry in entries:
1361 if entry.aircraft_id is not None or not entry.other_aircraft_registration:
1362 continue
1363 pilot_user_id = entry.pic_user_id or entry.second_crew_user_id
1364 if pilot_user_id is None:
1365 continue
1366 norm_reg = _norm_reg(entry.other_aircraft_registration)
1367 ac = None
1368 for tid in _tenants_for_pilot(pilot_user_id):
1369 ac = ac_by_tenant.get(tid, {}).get(norm_reg)
1370 if ac is not None:
1371 break
1372 if ac is None:
1373 continue
1375 if entry.departure_time is not None:
1376 dummy = datetime.combine(date.min, entry.departure_time) - timedelta(
1377 hours=float(ac.flight_counter_offset)
1378 )
1379 entry.departure_time = dummy.time()
1381 entry.aircraft_id = ac.id
1382 entry.departure_icao = _place_icao(entry.departure_icao)
1383 entry.arrival_icao = _place_icao(entry.arrival_icao)
1384 entry.other_aircraft_type = None
1385 entry.other_aircraft_type_icao = None
1386 entry.other_aircraft_registration = None
1387 entry.source = "logbook_import"
1389 # Step 2 (docs/backlog.md "reconcile imports from either side"): the
1390 # true airframe-side fields (flight_time_counter_*/flight_time,
1391 # engine_time_counter_*/engine_time — the maintenance-relevant hour
1392 # figures) aren't known from a personal logbook and stay NULL, same
1393 # as everywhere else in this app "if a field is not filled, it's not
1394 # filled" — no supposition, interpolation, or guessed value gets
1395 # invented here. That's also what keeps the flight list's
1396 # needs-attention icon (source == "logbook_import" and flight_time
1397 # is None) lit until a human fills these in from a real airframe
1398 # source. (A previous version of this function seeded a rough
1399 # engine_time_counter estimate here, on the assumption that nothing
1400 # read it for maintenance tracking — services/component_limits.py
1401 # now sums engine_time_counter for engine/propeller TBO, so that
1402 # assumption no longer holds and the guess was removed.)
1404 if not entry.pic_name:
1405 pilot_user = db.session.get(User, pilot_user_id)
1406 if pilot_user:
1407 entry.pic_name = pilot_user.name or pilot_user.email
1409 promoted += 1
1411 return promoted