Coverage for app/airworthiness_sync.py: 100%
108 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-31 10:13 +0000
1"""
2EASA Safety Publications Tool sync job.
4Queries the EASA AD search endpoint for each EASASourceNode, diffs against
5stored AirworthinessDocument records, and creates pending_review statuses for
6newly discovered documents on all aircraft that have the relevant component.
8Public API
9----------
10sync_all_nodes(app) — called by the background scheduler; syncs every node.
11sync_aircraft(ac) — called from the manual-trigger route; syncs only the
12 nodes that belong to this aircraft's components.
13"""
15import logging
16import re
17import time
18import urllib.error
19import urllib.parse
20import urllib.request
21from datetime import UTC, datetime
23from models import ( # pyright: ignore[reportMissingImports]
24 Aircraft,
25 AirworthinessDocStatus,
26 AirworthinessDocType,
27 AirworthinessDocument,
28 AirworthinessDocumentStatus,
29 EASASourceNode,
30 db,
31)
33_log = logging.getLogger(__name__)
35_EASA_SEARCH_URL = "https://ad.easa.europa.eu/search/advanced/result/"
36_REQUEST_TIMEOUT = 15
37_COURTESY_DELAY = 2.0 # seconds between requests
38_USER_AGENT = "OpenHangar/airworthiness-sync (+https://github.com/e2jk/OpenHangar)"
40# Matches "AD 2023-0048", "AD 2006-0345R", etc.
41_AD_RE = re.compile(r"\bAD\s+\d{4}-\d+[A-Z]*\b")
42# Matches "SIB 2024-01" etc.
43_SIB_RE = re.compile(r"\bSIB\s+\d{4}-\d+[A-Z]*\b")
46def _build_tree_path(node: EASASourceNode) -> str:
47 return (
48 f"{node.tc_holder_node_id}@@@@0@@{node.tc_holder_name}"
49 f"|||{node.type_node_id}@@{node.tc_holder_node_id}@@1@@{node.type_name}"
50 f"|||{node.model_node_id}@@{node.type_node_id}@@2@@{node.model_name}"
51 )
54def _fetch_references(node: EASASourceNode) -> list[tuple[str, str]]:
55 """
56 POST to the EASA search endpoint and return a list of (reference, doc_type)
57 tuples for all documents found. Raises requests.RequestException on failure.
58 """
59 payload = {
60 "fi_action": "advanced",
61 "fi_tree": _build_tree_path(node),
62 "fi_keyword": "",
63 "fi_date_start": "",
64 "fi_date_end": "",
65 "ps_src_tree": "",
66 "fi_notification": "N",
67 "is_default": "N",
68 "fi_basket[]": node.model_node_id,
69 }
70 data = urllib.parse.urlencode(payload).encode()
71 req = urllib.request.Request(
72 _EASA_SEARCH_URL,
73 data=data,
74 headers={"User-Agent": _USER_AGENT},
75 )
76 with urllib.request.urlopen( # nosec B310 # _EASA_SEARCH_URL is a hardcoded https:// constant
77 req, timeout=_REQUEST_TIMEOUT
78 ) as resp:
79 html = resp.read().decode()
81 refs: list[tuple[str, str]] = []
82 for m in _AD_RE.finditer(html):
83 refs.append((m.group().strip(), AirworthinessDocType.AD))
84 for m in _SIB_RE.finditer(html):
85 refs.append((m.group().strip(), AirworthinessDocType.SIB))
86 return refs
89def _easa_doc_url(reference: str) -> str:
90 slug = reference.replace(" ", "_").replace("/", "-")
91 return f"https://ad.easa.europa.eu/ad/{slug}"
94def _process_node(node: EASASourceNode) -> tuple[int, bool]:
95 """
96 Sync one node. Returns (new_docs_added, had_error).
97 Creates AirworthinessDocumentStatus records (pending_review) for each
98 aircraft that has a component referencing this node.
99 """
100 try:
101 refs = _fetch_references(node)
102 except Exception as exc: # noqa: BLE001 -- one node's fetch/parse failure
103 # (network error, timeout, unexpected EASA HTML) must not abort the
104 # sync of every other node in the batch; log and keep going.
105 _log.warning(
106 "EASA sync error for node %s (%s): %s", node.id, node.display_path, exc
107 )
108 node.consecutive_errors = (node.consecutive_errors or 0) + 1
109 db.session.commit()
110 return 0, True
112 # Existing references for this node
113 existing = {
114 d.reference
115 for d in AirworthinessDocument.query.filter_by(source_node_id=node.id).all()
116 }
118 aircraft_id = node.component.aircraft_id
120 added = 0
121 for reference, doc_type in refs:
122 if reference in existing:
123 continue
124 doc = AirworthinessDocument(
125 doc_type=doc_type,
126 reference=reference,
127 source_node_id=node.id,
128 doc_url=_easa_doc_url(reference),
129 )
130 db.session.add(doc)
131 db.session.flush()
133 # Create pending_review status for the aircraft
134 st = AirworthinessDocumentStatus(
135 aircraft_id=aircraft_id,
136 document_id=doc.id,
137 status=AirworthinessDocStatus.PENDING_REVIEW,
138 )
139 db.session.add(st)
140 added += 1
142 node.consecutive_errors = 0
143 node.last_synced_at = datetime.now(UTC)
144 db.session.commit()
145 return added, False
148def sync_aircraft(ac: Aircraft) -> tuple[int, int]:
149 """
150 Sync all EASA source nodes for the given aircraft.
151 Returns (total_new_docs, total_error_nodes).
152 """
153 total_added = 0
154 total_errors = 0
155 first = True
156 for comp in ac.components: # type: ignore[attr-defined]
157 for node in comp.easa_source_nodes:
158 if not first:
159 time.sleep(_COURTESY_DELAY)
160 first = False
161 added, had_error = _process_node(node)
162 total_added += added
163 if had_error:
164 total_errors += 1
165 return total_added, total_errors
168def sync_all_nodes(app: object) -> None:
169 """
170 Sync every EASASourceNode in the database. Called by the background
171 scheduler (once per 24 h). Logs a warning if a node has not synced
172 successfully in 72 h.
174 Guarded by an advisory lock (see services.advisory_lock) so that only one
175 gunicorn worker performs the sync per scheduled tick — without it, all
176 four production workers would hit the EASA endpoint independently and
177 each create their own copy of any new document. A session-scoped lock
178 (not a transaction-scoped one) is required here because _process_node
179 commits once per node, and a transaction-scoped lock would release at
180 the first commit.
181 """
182 import flask # pyright: ignore[reportMissingImports]
183 from services.advisory_lock import (
184 advisory_lock_scope, # pyright: ignore[reportMissingImports]
185 )
187 assert isinstance(app, flask.Flask)
189 with app.app_context(), advisory_lock_scope(db, 7283910458) as acquired:
190 if not acquired:
191 _log.info("EASA sync: another worker holds the lock — skipping this run")
192 return
194 nodes = EASASourceNode.query.all()
195 _log.info("EASA sync: starting sync for %d node(s)", len(nodes))
196 total_added = 0
197 total_errors = 0
198 skipped = 0
199 now = datetime.now(UTC)
200 first_processed = True
201 for node in nodes:
202 # Exponential backoff: after 2+ consecutive failures, wait before retrying.
203 # backoff = min(2^errors, 7) days from last successful sync.
204 errors = node.consecutive_errors or 0
205 if errors >= 2 and node.last_synced_at is not None:
206 backoff_days = min(2**errors, 7)
207 last = (
208 node.last_synced_at.replace(tzinfo=UTC)
209 if node.last_synced_at.tzinfo is None
210 else node.last_synced_at
211 )
212 if (now - last).days < backoff_days:
213 _log.info(
214 "EASA sync: skipping node %s (%s) — %d error(s), backoff %d day(s)",
215 node.id,
216 node.display_path,
217 errors,
218 backoff_days,
219 )
220 skipped += 1
221 continue
223 if not first_processed:
224 time.sleep(_COURTESY_DELAY)
225 first_processed = False
227 added, had_error = _process_node(node)
228 total_added += added
229 if had_error:
230 total_errors += 1
231 else:
232 _log.debug(
233 "EASA sync: node %s (%s) — %d new doc(s)",
234 node.id,
235 node.display_path,
236 added,
237 )
239 _log.info(
240 "EASA sync complete: %d new document(s), %d error(s), %d skipped (backoff)",
241 total_added,
242 total_errors,
243 skipped,
244 )
246 # Warn for nodes overdue (72 h without a successful sync)
247 from datetime import timedelta
249 cutoff = datetime.now(UTC) - timedelta(hours=72)
250 overdue = [
251 n
252 for n in nodes
253 if n.last_synced_at is None
254 or (
255 n.last_synced_at.replace(tzinfo=UTC)
256 if n.last_synced_at.tzinfo is None
257 else n.last_synced_at
258 )
259 < cutoff
260 ]
261 for node in overdue:
262 _log.warning(
263 "[AIRWORTHINESS] EASA sync overdue for node %s (%s) — last success: %s",
264 node.id,
265 node.display_path,
266 node.last_synced_at,
267 )