Checks Medsafe (NZ) Safety Communications for medical device alerts. Parses the HTML table with columns: Date, Communication, Product Type, Topic. Filters for Product Type containing 'Device'.
| 616 | # --------------------------------------------------------------------------- |
| 617 | |
| 618 | class MedsafeChecker: |
| 619 | """Checks Medsafe (NZ) Safety Communications for medical device alerts. |
| 620 | |
| 621 | Parses the HTML table with columns: Date, Communication, Product Type, Topic. |
| 622 | Filters for Product Type containing 'Device'. |
| 623 | """ |
| 624 | |
| 625 | def __init__(self, session: requests.Session, state: dict, |
| 626 | seed_mode: bool = False): |
| 627 | self.session = session |
| 628 | self.state = state |
| 629 | self.seed_mode = seed_mode |
| 630 | |
| 631 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 632 | url = source.get("url", "https://medsafe.govt.nz/safety/SafetyCommunications.asp") |
| 633 | prev = self.state.get(source_id, {}) |
| 634 | prev_ids = set(prev.get("seen_ids", [])) |
| 635 | |
| 636 | try: |
| 637 | resp = self.session.get(url, timeout=30, headers={ |
| 638 | "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/130.0", |
| 639 | }) |
| 640 | resp.raise_for_status() |
| 641 | except Exception as e: |
| 642 | print(f" ERROR Medsafe: {e}") |
| 643 | return None |
| 644 | |
| 645 | entries = self._parse_table(resp.text) |
| 646 | if not entries: |
| 647 | print(" WARNING: No entries from Medsafe Safety Communications") |
| 648 | return None |
| 649 | |
| 650 | print(f" INFO: Parsed {len(entries)} device entries from Medsafe") |
| 651 | |
| 652 | new_items = [] |
| 653 | all_ids = list(prev_ids) |
| 654 | |
| 655 | for entry in entries: |
| 656 | eid = entry.get("title", "") |
| 657 | if not eid or eid in prev_ids: |
| 658 | continue |
| 659 | all_ids.append(eid) |
| 660 | new_items.append(entry) |
| 661 | |
| 662 | self.state[source_id] = { |
| 663 | "url": url, |
| 664 | "last_checked": datetime.now().isoformat(), |
| 665 | "seen_ids": all_ids[-200:], |
| 666 | } |
| 667 | |
| 668 | if new_items and (prev_ids or self.seed_mode): |
| 669 | if self.seed_mode and not prev_ids: |
| 670 | new_items = new_items[:10] |
| 671 | print(f" INFO: Seed mode -- returning top {len(new_items)} Medsafe entries") |
| 672 | result = _make_update( |
| 673 | source_id, source, "medsafe_page", |
| 674 | f"{len(new_items)} new Medsafe device safety communication(s)" |
| 675 | ) |