Coverage for app/services/gps_archive.py: 100%

1"""Keep every uploaded GPS log file as-is, on disk. 

2 

3A GPS import only stores a simplified track in the database (at most 

4_MAX_TRACK_POINTS points, a handful of fields — see aircraft/gps_import.py). 

5The original files carry far more (one point per second, and for Garmin 

6data logs also airspeed, attitude, engine and fuel figures, …), so a copy of 

7each is kept under the uploads folder, where backups pick it up like any 

8other uploaded file, for future features to reprocess. Nothing about the 

9copy goes into the database. 

10 

11Layout (relative to UPLOAD_FOLDER, see services.upload_layout): 

12 <tenant-slug>/<REG>/gps_logs/<file> — uploaded for a managed aircraft 

13 _pilots/<user-id>/gps_logs/<file> — uploaded from a pilot's own logbook 

14 (or for an aircraft not managed here) 

15""" 

16 

17from __future__ import annotations 

18 

19import hashlib 

20import os 

21 

22from werkzeug.utils import secure_filename # pyright: ignore[reportMissingImports] 

23 

24from services.upload_layout import ( # pyright: ignore[reportMissingImports] 

25 GPS_ARCHIVE_DIRNAME, 

26 PILOT_ARCHIVE_ROOT, 

27) 

28 

29 

30def pilot_archive_parts(user_id: int) -> tuple[str, ...]: 

31 """Path components (under UPLOAD_FOLDER) of a pilot's GPS log archive.""" 

32 return (PILOT_ARCHIVE_ROOT, str(user_id), GPS_ARCHIVE_DIRNAME) 

33 

34 

35def _sha256_file(path: str) -> str: 

36 digest = hashlib.sha256() 

37 with open(path, "rb") as fh: 

38 for chunk in iter(lambda: fh.read(1024 * 1024), b""): 

39 digest.update(chunk) 

40 return digest.hexdigest() 

41 

42 

43def _find_identical(dest_dir: str, data: bytes, digest: str) -> str | None: 

44 """Path of a file already in *dest_dir* with exactly this content, if any. 

45 

46 Only files of the same size are hashed, so an upload usually costs no 

47 reads at all. 

48 """ 

49 size = len(data) 

50 for entry in os.scandir(dest_dir): 

51 if entry.name.startswith(".") or not entry.is_file(): 

52 continue 

53 if entry.stat().st_size == size and _sha256_file(entry.path) == digest: 

54 return entry.path 

55 return None 

56 

57 

58def archive_gps_file(data: bytes, original_filename: str, dest_dir: str) -> str: 

59 """Store *data* in *dest_dir* under its (sanitised) original name and 

60 return the stored file's full path. 

61 

62 A file whose content is already in *dest_dir* — under any name — stores 

63 nothing new; the existing copy's path is returned. This only governs the 

64 archive: the upload is parsed and offered for import either way (e.g. 

65 after its flight was deleted). A different file arriving under a name 

66 already taken — "flight.gpx" from two different days, say — gets a short 

67 content-hash suffix rather than overwriting the earlier one. 

68 """ 

69 os.makedirs(dest_dir, exist_ok=True) 

70 digest = hashlib.sha256(data).hexdigest() 

71 existing = _find_identical(dest_dir, data, digest) 

72 if existing: 

73 return existing 

74 name = secure_filename(original_filename) or "gps_log" 

75 path = os.path.join(dest_dir, name) 

76 if os.path.exists(path): 

77 stem, ext = os.path.splitext(name) 

78 path = os.path.join(dest_dir, f"{stem}-{digest[:8]}{ext}") 

79 if os.path.exists(path): # short-hash collision: use the full hash 

80 path = os.path.join(dest_dir, f"{stem}-{digest}{ext}") 

81 # Write under a dot-prefixed temporary name first (the document scan 

82 # ignores those) so an interrupted write never leaves a truncated file 

83 # under the real name. 

84 tmp_path = os.path.join(dest_dir, f".{os.path.basename(path)}.part") 

85 with open(tmp_path, "wb") as fh: 

86 fh.write(data) 

87 os.replace(tmp_path, path) 

88 return path