(self, source_id: str, source: dict)
| 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 | ) |
| 676 | result["new_items"] = new_items |
| 677 | return result |
| 678 | elif not prev_ids: |
| 679 | print(f" INFO: Baseline established ({len(entries)} Medsafe entries)") |
| 680 | |
| 681 | return None |
| 682 | |
| 683 | @staticmethod |
| 684 | def _parse_table(html: str) -> list[dict]: |
nothing calls this directly
no test coverage detected