Coverage for app/sync_watcher.py: 100%
101 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-31 10:13 +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
32log = logging.getLogger("openhangar.sync_watcher")
34_CATEGORY_VALUES: set[str] | None = None
35_DATE_TITLE_RE = _re.compile(r"^(\d{4}-\d{2}-\d{2}) - (.+?)(\.[^.]+)?$")
38def _categories() -> set[str]:
39 global _CATEGORY_VALUES
40 if _CATEGORY_VALUES is None:
41 from models import DocCategory # pyright: ignore[reportMissingImports]
43 _CATEGORY_VALUES = set(DocCategory.ALL)
44 return _CATEGORY_VALUES
47def _scan_once(app: Any) -> None:
48 """Single scan pass — runs inside an app context.
50 Guarded by an advisory lock (see services.advisory_lock) so that only
51 one gunicorn worker scans per tick — without it, all four production
52 workers would race to import the same new file, since each builds its
53 own known_filenames snapshot before any of them commits.
54 """
55 from models import ( # pyright: ignore[reportMissingImports]
56 Aircraft,
57 Document,
58 PendingReconcile,
59 Tenant,
60 db,
61 )
62 from services.advisory_lock import (
63 advisory_lock_scope, # pyright: ignore[reportMissingImports]
64 )
66 folder = app.config.get("UPLOAD_FOLDER", "/data/uploads")
67 if not os.path.isdir(folder):
68 return
70 with app.app_context(), advisory_lock_scope(db, 7283910459) as acquired:
71 if not acquired:
72 log.info("sync_watcher: another worker holds the lock — skipping this scan")
73 return
75 # Build lookup tables once per scan
76 tenants = {
77 t.slug: t for t in Tenant.query.filter(Tenant.slug.isnot(None)).all()
78 }
79 if not tenants:
80 return
82 known_filenames: set[str] = {
83 doc.filename
84 for doc in Document.query.with_entities(Document.filename).all()
85 }
86 pending_filepaths: set[str] = {
87 pr.filepath
88 for pr in PendingReconcile.query.with_entities(
89 PendingReconcile.filepath
90 ).all()
91 }
93 for tenant_slug, tenant in tenants.items():
94 slug_dir = os.path.join(folder, tenant_slug)
95 if not os.path.isdir(slug_dir):
96 continue
98 # Drop pending entries whose file no longer exists on disk
99 for pr in PendingReconcile.query.filter_by(
100 tenant_id=tenant.id, reconciled_at=None, ignored=False
101 ).all():
102 if not os.path.exists(os.path.join(folder, pr.filepath)):
103 db.session.delete(pr)
104 pending_filepaths.discard(pr.filepath)
106 # Build registration → aircraft map for this tenant
107 aircraft_by_reg: dict[str, Any] = {
108 ac.registration.upper().replace("-", "").replace(" ", ""): ac
109 for ac in Aircraft.query.filter_by(tenant_id=tenant.id).all()
110 }
112 for dirpath, _dirs, filenames in os.walk(slug_dir):
113 for fname in filenames:
114 if fname.startswith((".", "_")):
115 continue
117 full = os.path.join(dirpath, fname)
118 relpath = os.path.relpath(full, folder).replace("\\", "/")
120 if relpath in known_filenames or relpath in pending_filepaths:
121 continue
123 _process_file(
124 app,
125 full,
126 relpath,
127 fname,
128 tenant,
129 aircraft_by_reg,
130 known_filenames,
131 pending_filepaths,
132 db,
133 Document,
134 PendingReconcile,
135 )
137 db.session.commit()
140def _process_file(
141 app: Any,
142 full_path: str,
143 relpath: str,
144 fname: str,
145 tenant: Any,
146 aircraft_by_reg: dict[str, Any],
147 known_filenames: set[str],
148 pending_filepaths: set[str],
149 db: Any,
150 Document: Any,
151 PendingReconcile: Any,
152) -> None:
153 """Decide whether to auto-import or queue for review."""
154 parts = relpath.split("/")
155 # Expected: slug / reg / category / filename (4 parts minimum)
156 if len(parts) < 4:
157 _queue_pending(
158 relpath, fname, None, None, None, None, tenant, db, PendingReconcile
159 )
160 pending_filepaths.add(relpath)
161 return
163 reg_raw = parts[1].upper().replace("-", "").replace(" ", "")
164 cat_str = parts[2]
165 filename_part = parts[3]
167 aircraft = aircraft_by_reg.get(reg_raw)
168 cat_lower = cat_str.lower()
169 category = cat_lower if cat_lower in _categories() else None
171 # Parse "YYYY-MM-DD - title.ext" from the filename
172 m = _DATE_TITLE_RE.match(filename_part)
173 title_hint: str | None = None
174 date_hint: _date | None = None
175 if m:
176 with contextlib.suppress(
177 ValueError
178 ): # regex matched date-like string but it's invalid (e.g. month 13); treat as no date
179 date_hint = _date.fromisoformat(m.group(1))
180 title_hint = m.group(2)
181 else:
182 title_hint = os.path.splitext(filename_part)[0]
184 if aircraft and category:
185 # Fully resolved — auto-import immediately
186 mime = mimetypes.guess_type(fname)[0] or "application/octet-stream"
187 size = None
188 with contextlib.suppress(
189 OSError
190 ): # file may have disappeared between scan and import; size stays None
191 size = os.path.getsize(full_path)
192 doc = Document(
193 aircraft_id=aircraft.id,
194 filename=relpath,
195 original_filename=fname,
196 mime_type=mime,
197 size_bytes=size,
198 title=title_hint,
199 category=category,
200 )
201 db.session.add(doc)
202 known_filenames.add(relpath)
203 log.info(
204 "sync_watcher: auto-imported %s → aircraft=%s category=%s",
205 relpath,
206 aircraft.registration,
207 category,
208 )
209 else:
210 # Ambiguous — queue for manual review
211 _queue_pending(
212 relpath,
213 fname,
214 aircraft,
215 category,
216 title_hint,
217 date_hint,
218 tenant,
219 db,
220 PendingReconcile,
221 )
222 pending_filepaths.add(relpath)
223 log.info(
224 "sync_watcher: queued for review %s (aircraft=%s category=%s)",
225 relpath,
226 aircraft.registration if aircraft else "?",
227 category or "?",
228 )
231def _queue_pending(
232 relpath: str,
233 fname: str,
234 aircraft: Any,
235 category: str | None,
236 title_hint: str | None,
237 date_hint: _date | None,
238 tenant: Any,
239 db: Any,
240 PendingReconcile: Any,
241) -> None:
242 pr = PendingReconcile(
243 tenant_id=tenant.id,
244 aircraft_id=aircraft.id if aircraft else None,
245 filepath=relpath,
246 category=category,
247 title_hint=title_hint or os.path.splitext(fname)[0],
248 date_hint=date_hint,
249 )
250 db.session.add(pr)
253def _watcher_loop(app: Any, interval: int) -> None:
254 log.info("sync_watcher: started (interval=%ds)", interval)
255 while True:
256 try:
257 _scan_once(app)
258 except Exception:
259 log.exception("sync_watcher: unhandled error during scan")
260 time.sleep(interval)
263def start_sync_watcher(app: Any) -> None:
264 """Start the background sync watcher thread (idempotent, daemon thread)."""
265 try:
266 interval = int(os.environ.get("OPENHANGAR_SYNC_SCAN_INTERVAL", "60"))
267 except ValueError:
268 interval = 60
270 t = threading.Thread(
271 target=_watcher_loop,
272 args=(app, interval),
273 daemon=True,
274 name="sync-watcher",
275 )
276 t.start()