Checks PMDA (Japan) English pages for medical device safety & regulatory updates. Parses tabular pages: Medical Safety Information, Revisions of PRECAUTIONS, Alert for Proper Use of Medical Devices, etc.
| 2055 | |
| 2056 | |
| 2057 | class PMDAChecker: |
| 2058 | """Checks PMDA (Japan) English pages for medical device safety & regulatory updates. |
| 2059 | |
| 2060 | Parses tabular pages: Medical Safety Information, Revisions of PRECAUTIONS, |
| 2061 | Alert for Proper Use of Medical Devices, etc. |
| 2062 | """ |
| 2063 | |
| 2064 | def __init__(self, session: requests.Session, state: dict, |
| 2065 | seed_mode: bool = False): |
| 2066 | self.session = session |
| 2067 | self.state = state |
| 2068 | self.seed_mode = seed_mode |
| 2069 | |
| 2070 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 2071 | url = source["url"] |
| 2072 | prev = self.state.get(source_id, {}) |
| 2073 | prev_titles = set(prev.get("seen_titles", [])) |
| 2074 | |
| 2075 | try: |
| 2076 | resp = self.session.get(url, timeout=30) |
| 2077 | resp.raise_for_status() |
| 2078 | except Exception as e: |
| 2079 | print(f" ERROR PMDA: {e}") |
| 2080 | return None |
| 2081 | |
| 2082 | entries = self._parse_table_page(resp.text, url) |
| 2083 | if not entries: |
| 2084 | print(f" WARNING: No entries from PMDA page ({url})") |
| 2085 | return None |
| 2086 | |
| 2087 | print(f" INFO: Parsed {len(entries)} entries from PMDA page") |
| 2088 | |
| 2089 | new_items = [] |
| 2090 | all_titles = list(prev_titles) |
| 2091 | |
| 2092 | for entry in entries: |
| 2093 | title = entry.get("title", "") |
| 2094 | if not title or title in prev_titles: |
| 2095 | continue |
| 2096 | all_titles.append(title) |
| 2097 | new_items.append(entry) |
| 2098 | |
| 2099 | self.state[source_id] = { |
| 2100 | "url": url, |
| 2101 | "last_checked": datetime.now().isoformat(), |
| 2102 | "seen_titles": all_titles[-300:], |
| 2103 | } |
| 2104 | |
| 2105 | if new_items and (prev_titles or self.seed_mode): |
| 2106 | if self.seed_mode and not prev_titles: |
| 2107 | new_items = new_items[:10] |
| 2108 | print(f" INFO: Seed mode -- returning top {len(new_items)} entries as initial news") |
| 2109 | result = _make_update( |
| 2110 | source_id, source, "pmda_page", |
| 2111 | f"{len(new_items)} new PMDA update(s) detected" |
| 2112 | ) |
| 2113 | result["new_items"] = new_items |
| 2114 | return result |