Coverage for app/sync_watcher.py: 100%
104 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 15:29 +0000
1"""
2Background polling thread — automatically imports documents that arrive via
3Syncthing (or any other file-sync tool) into the canonical folder structure.
5Canonical path layout (relative to UPLOAD_FOLDER):
6 {tenant_slug}/{aircraft_reg}/{category}/YYYY-MM-DD - title.ext
8Behaviour on discovery:
9 - File matches canonical structure AND aircraft is recognised in the tenant
10 → Document row created immediately (auto-import).
11 - File path is valid but aircraft/category cannot be resolved unambiguously
12 → PendingReconcile entry created for manual review in the UI.
13 - File already tracked (filename in documents table) or already pending
14 → Skipped.
16Enabled only when UPLOAD_FOLDER is set and the database is PostgreSQL
17(SQLite = dev/test; the watcher is skipped there to avoid confusion).
19Interval is configured via SYNC_SCAN_INTERVAL env var (default: 60 s).
20"""
22import contextlib
23import logging
24import mimetypes
25import os
26import re as _re
27import threading
28import time
29from datetime import date as _date
30from typing import Any
32from services.upload_layout import ( # pyright: ignore[reportMissingImports]
33 is_app_managed,
34)
36log = logging.getLogger("openhangar.sync_watcher")
38_CATEGORY_VALUES: set[str] | None = None
39_DATE_TITLE_RE = _re.compile(r"^(\d{4}-\d{2}-\d{2}) - (.+?)(\.[^.]+)?$")
42def _categories() -> set[str]:
43 global _CATEGORY_VALUES
44 if _CATEGORY_VALUES is None:
45 from models import DocCategory # pyright: ignore[reportMissingImports]
47 _CATEGORY_VALUES = set(DocCategory.ALL)
48 return _CATEGORY_VALUES
51def _scan_once(app: Any) -> None:
52 """Single scan pass — runs inside an app context.
54 Guarded by an advisory lock (see services.advisory_lock) so that only
55 one gunicorn worker scans per tick — without it, all four production
56 workers would race to import the same new file, since each builds its
57 own known_filenames snapshot before any of them commits.
58 """
59 from models import ( # pyright: ignore[reportMissingImports]
60 Aircraft,
61 Document,
62 PendingReconcile,
63 Tenant,
64 db,
65 )
66 from services.advisory_lock import (
67 advisory_lock_scope, # pyright: ignore[reportMissingImports]
68 )
70 folder = app.config.get("UPLOAD_FOLDER", "/data/uploads")
71 if not os.path.isdir(folder):
72 return
74 with app.app_context(), advisory_lock_scope(db, 7283910459) as acquired:
75 if not acquired:
76 log.info("sync_watcher: another worker holds the lock — skipping this scan")
77 return
79 # Build lookup tables once per scan
80 tenants = {
81 t.slug: t for t in Tenant.query.filter(Tenant.slug.isnot(None)).all()
82 }
83 if not tenants:
84 return
86 known_filenames: set[str] = {
87 doc.filename
88 for doc in Document.query.with_entities(Document.filename).all()
89 }
90 pending_filepaths: set[str] = {
91 pr.filepath
92 for pr in PendingReconcile.query.with_entities(
93 PendingReconcile.filepath
94 ).all()
95 }
97 for tenant_slug, tenant in tenants.items():
98 slug_dir = os.path.join(folder, tenant_slug)
99 if not os.path.isdir(slug_dir):
100 continue
102 # Drop pending entries whose file no longer exists on disk
103 for pr in PendingReconcile.query.filter_by(
104 tenant_id=tenant.id, reconciled_at=None, ignored=False
105 ).all():
106 if not os.path.exists(os.path.join(folder, pr.filepath)):
107 db.session.delete(pr)
108 pending_filepaths.discard(pr.filepath)
110 # Build registration → aircraft map for this tenant
111 aircraft_by_reg: dict[str, Any] = {
112 ac.registration.upper().replace("-", "").replace(" ", ""): ac
113 for ac in Aircraft.query.filter_by(tenant_id=tenant.id).all()
114 }
116 for dirpath, _dirs, filenames in os.walk(slug_dir):
117 for fname in filenames:
118 if fname.startswith((".", "_")):
119 continue
121 full = os.path.join(dirpath, fname)
122 relpath = os.path.relpath(full, folder).replace("\\", "/")
124 if relpath in known_filenames or relpath in pending_filepaths:
125 continue
126 if is_app_managed(relpath):
127 continue # photos, original GPS logs — not documents
129 _process_file(
130 app,
131 full,
132 relpath,
133 fname,
134 tenant,
135 aircraft_by_reg,
136 known_filenames,
137 pending_filepaths,
138 db,
139 Document,
140 PendingReconcile,
141 )
143 db.session.commit()
146def _process_file(
147 app: Any,
148 full_path: str,
149 relpath: str,
150 fname: str,
151 tenant: Any,
152 aircraft_by_reg: dict[str, Any],
153 known_filenames: set[str],
154 pending_filepaths: set[str],
155 db: Any,
156 Document: Any,
157 PendingReconcile: Any,
158) -> None:
159 """Decide whether to auto-import or queue for review."""
160 parts = relpath.split("/")
161 # Expected: slug / reg / category / filename (4 parts minimum)
162 if len(parts) < 4:
163 _queue_pending(
164 relpath, fname, None, None, None, None, tenant, db, PendingReconcile
165 )
166 pending_filepaths.add(relpath)
167 return
169 reg_raw = parts[1].upper().replace("-", "").replace(" ", "")
170 cat_str = parts[2]
171 filename_part = parts[3]
173 aircraft = aircraft_by_reg.get(reg_raw)
174 cat_lower = cat_str.lower()
175 category = cat_lower if cat_lower in _categories() else None
177 # Parse "YYYY-MM-DD - title.ext" from the filename
178 m = _DATE_TITLE_RE.match(filename_part)
179 title_hint: str | None = None
180 date_hint: _date | None = None
181 if m:
182 with contextlib.suppress(
183 ValueError
184 ): # regex matched date-like string but it's invalid (e.g. month 13); treat as no date
185 date_hint = _date.fromisoformat(m.group(1))
186 title_hint = m.group(2)
187 else:
188 title_hint = os.path.splitext(filename_part)[0]
190 if aircraft and category:
191 # Fully resolved — auto-import immediately
192 mime = mimetypes.guess_type(fname)[0] or "application/octet-stream"
193 size = None
194 with contextlib.suppress(
195 OSError
196 ): # file may have disappeared between scan and import; size stays None
197 size = os.path.getsize(full_path)
198 doc = Document(
199 aircraft_id=aircraft.id,
200 filename=relpath,
201 original_filename=fname,
202 mime_type=mime,
203 size_bytes=size,
204 title=title_hint,
205 category=category,
206 )
207 db.session.add(doc)
208 known_filenames.add(relpath)
209 log.info(
210 "sync_watcher: auto-imported %s → aircraft=%s category=%s",
211 relpath,
212 aircraft.registration,
213 category,
214 )
215 else:
216 # Ambiguous — queue for manual review
217 _queue_pending(
218 relpath,
219 fname,
220 aircraft,
221 category,
222 title_hint,
223 date_hint,
224 tenant,
225 db,
226 PendingReconcile,
227 )
228 pending_filepaths.add(relpath)
229 log.info(
230 "sync_watcher: queued for review %s (aircraft=%s category=%s)",
231 relpath,
232 aircraft.registration if aircraft else "?",
233 category or "?",
234 )
237def _queue_pending(
238 relpath: str,
239 fname: str,
240 aircraft: Any,
241 category: str | None,
242 title_hint: str | None,
243 date_hint: _date | None,
244 tenant: Any,
245 db: Any,
246 PendingReconcile: Any,
247) -> None:
248 pr = PendingReconcile(
249 tenant_id=tenant.id,
250 aircraft_id=aircraft.id if aircraft else None,
251 filepath=relpath,
252 category=category,
253 title_hint=title_hint or os.path.splitext(fname)[0],
254 date_hint=date_hint,
255 )
256 db.session.add(pr)
259def _watcher_loop(app: Any, interval: int) -> None:
260 log.info("sync_watcher: started (interval=%ds)", interval)
261 while True:
262 try:
263 _scan_once(app)
264 except Exception:
265 log.exception("sync_watcher: unhandled error during scan")
266 time.sleep(interval)
269def start_sync_watcher(app: Any) -> None:
270 """Start the background sync watcher thread (idempotent, daemon thread)."""
271 try:
272 interval = int(os.environ.get("OPENHANGAR_SYNC_SCAN_INTERVAL", "60"))
273 except ValueError:
274 interval = 60
276 t = threading.Thread(
277 target=_watcher_loop,
278 args=(app, interval),
279 daemon=True,
280 name="sync-watcher",
281 )
282 t.start()