Checks OpenFDA device/recall.json for recently initiated recalls (all classes). This replaces the web-scraping FDASafetyPageChecker because FDA pages are protected by WAF (401). The OpenFDA recall API provides structured data about all device recalls initiated in the past 14 days.
| 1362 | # --------------------------------------------------------------------------- |
| 1363 | |
| 1364 | class FDASafetyRecallChecker: |
| 1365 | """Checks OpenFDA device/recall.json for recently initiated recalls (all classes). |
| 1366 | |
| 1367 | This replaces the web-scraping FDASafetyPageChecker because FDA pages are |
| 1368 | protected by WAF (401). The OpenFDA recall API provides structured data |
| 1369 | about all device recalls initiated in the past 14 days. |
| 1370 | |
| 1371 | Complements FDARecallChecker (enforcement API, Class I only) by covering |
| 1372 | Class II/III recalls and newly initiated events before enforcement action. |
| 1373 | """ |
| 1374 | |
| 1375 | RECALL_URL = "https://api.fda.gov/device/recall.json" |
| 1376 | |
| 1377 | def __init__(self, api_key: str, state: dict): |
| 1378 | self.api_key = api_key |
| 1379 | self.state = state |
| 1380 | |
| 1381 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 1382 | prev = self.state.get(source_id, {}) |
| 1383 | seen_ids = set(prev.get("seen_res_numbers", [])) |
| 1384 | |
| 1385 | lookback = (datetime.now() - timedelta(days=30)).strftime("%Y%m%d") |
| 1386 | today = datetime.now().strftime("%Y%m%d") |
| 1387 | |
| 1388 | search_q = f'event_date_initiated:[{lookback} TO {today}]' |
| 1389 | url = ( |
| 1390 | f"{self.RECALL_URL}" |
| 1391 | f"?search={requests.utils.quote(search_q)}" |
| 1392 | f"&sort=event_date_initiated:desc&limit=30" |
| 1393 | ) |
| 1394 | if self.api_key: |
| 1395 | url += f"&api_key={self.api_key}" |
| 1396 | |
| 1397 | try: |
| 1398 | resp = requests.get(url, timeout=20) |
| 1399 | if resp.status_code == 404: |
| 1400 | self.state[source_id] = { |
| 1401 | "url": source["url"], |
| 1402 | "last_checked": datetime.now().isoformat(), |
| 1403 | "seen_res_numbers": list(seen_ids), |
| 1404 | } |
| 1405 | return None |
| 1406 | resp.raise_for_status() |
| 1407 | results = resp.json().get("results", []) |
| 1408 | except Exception as e: |
| 1409 | print(f" ERROR OpenFDA Recall: {e}") |
| 1410 | return None |
| 1411 | |
| 1412 | new_items = [] |
| 1413 | all_ids = list(seen_ids) |
| 1414 | |
| 1415 | for r in results: |
| 1416 | res_num = r.get("product_res_number") or r.get("res_event_number", "") |
| 1417 | if not res_num or res_num in seen_ids: |
| 1418 | continue |
| 1419 | all_ids.append(res_num) |
| 1420 | |
| 1421 | firm = r.get("recalling_firm", "Unknown") |