Scrapes CDRH News and Updates page for regulatory announcements. Captures: town halls, guidance announcements, policy updates, and other CDRH communications relevant to medical device manufacturers.
| 1579 | # --------------------------------------------------------------------------- |
| 1580 | |
| 1581 | class CDRHNewsChecker: |
| 1582 | """Scrapes CDRH News and Updates page for regulatory announcements. |
| 1583 | |
| 1584 | Captures: town halls, guidance announcements, policy updates, and other |
| 1585 | CDRH communications relevant to medical device manufacturers. |
| 1586 | """ |
| 1587 | |
| 1588 | CDRH_URL = "https://www.fda.gov/medical-devices/medical-devices-news-and-events/cdrh-new-news-and-updates" |
| 1589 | |
| 1590 | _PRIORITY_KEYWORDS = re.compile( |
| 1591 | r"(?i)(guidance|final\s+rule|proposed\s+rule|safety\s+communication|" |
| 1592 | r"recall|cybersecurity|software|QMSR|510\(k\)|PMA|De\s*Novo|" |
| 1593 | r"UDI|labeling|AI|machine\s+learning|real.world|SaMD|" |
| 1594 | r"postmarket|premarket|clinical\s+investigation|town\s+hall)", |
| 1595 | ) |
| 1596 | |
| 1597 | def __init__(self, session: requests.Session, state: dict): |
| 1598 | self.session = session |
| 1599 | self.state = state |
| 1600 | |
| 1601 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 1602 | prev = self.state.get(source_id, {}) |
| 1603 | prev_titles = set(prev.get("seen_titles", [])) |
| 1604 | |
| 1605 | try: |
| 1606 | resp = self.session.get(self.CDRH_URL, timeout=30) |
| 1607 | if resp.status_code == 401: |
| 1608 | print(f" INFO: CDRH news page returned 401 (WAF block). " |
| 1609 | f"Falling back to OpenFDA guidance API coverage.") |
| 1610 | self.state[source_id] = { |
| 1611 | "url": source["url"], |
| 1612 | "last_checked": datetime.now().isoformat(), |
| 1613 | "seen_titles": list(prev_titles), |
| 1614 | "status": "waf_blocked", |
| 1615 | } |
| 1616 | return None |
| 1617 | resp.raise_for_status() |
| 1618 | except Exception as e: |
| 1619 | print(f" ERROR fetching CDRH news: {e}") |
| 1620 | return None |
| 1621 | |
| 1622 | entries = self._parse_entries(resp.text) |
| 1623 | if not entries: |
| 1624 | print(f" WARNING: No entries parsed from CDRH news page") |
| 1625 | return None |
| 1626 | |
| 1627 | print(f" INFO: Parsed {len(entries)} entries from CDRH news page") |
| 1628 | |
| 1629 | new_items = [] |
| 1630 | all_titles = list(prev_titles) |
| 1631 | |
| 1632 | for entry in entries: |
| 1633 | title = entry.get("title", "") |
| 1634 | if not title or title in prev_titles: |
| 1635 | continue |
| 1636 | all_titles.append(title) |
| 1637 | new_items.append({ |
| 1638 | "title": title, |