(self, source_id: str, source: dict)
| 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, |
| 1639 | "link": entry.get("link", ""), |
| 1640 | "pub_date": entry.get("date", ""), |
| 1641 | "description": entry.get("type", "CDRH News"), |
| 1642 | }) |
| 1643 | |
| 1644 | self.state[source_id] = { |
| 1645 | "url": source["url"], |
| 1646 | "last_checked": datetime.now().isoformat(), |
| 1647 | "seen_titles": all_titles[-200:], |
| 1648 | } |
| 1649 | |
| 1650 | if new_items and prev_titles: |
| 1651 | result = _make_update( |
| 1652 | source_id, source, "fda_cdrh_news", |
| 1653 | f"{len(new_items)} new CDRH news/update(s) detected" |
| 1654 | ) |
| 1655 | result["new_items"] = new_items |
| 1656 | return result |
| 1657 | elif not prev_titles: |
| 1658 | print(f" INFO: Baseline established ({len(entries)} CDRH news entries)") |
nothing calls this directly
no test coverage detected