Checks PMDA Japanese recall list pages for medical device recalls. Parses the structured HTML tables at info.pmda.go.jp with device name, manufacturer, recall class and publication date.
| 2190 | |
| 2191 | |
| 2192 | class PMDARecallsChecker: |
| 2193 | """Checks PMDA Japanese recall list pages for medical device recalls. |
| 2194 | |
| 2195 | Parses the structured HTML tables at info.pmda.go.jp with device name, |
| 2196 | manufacturer, recall class and publication date. |
| 2197 | """ |
| 2198 | |
| 2199 | def __init__(self, session: requests.Session, state: dict, |
| 2200 | seed_mode: bool = False): |
| 2201 | self.session = session |
| 2202 | self.state = state |
| 2203 | self.seed_mode = seed_mode |
| 2204 | |
| 2205 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 2206 | url = source["url"] |
| 2207 | if source.get("dynamic_year"): |
| 2208 | yy = str(datetime.now().year % 100).zfill(2) |
| 2209 | url = url.replace("{yy}", yy) |
| 2210 | recall_class = source.get("recall_class", "") |
| 2211 | prev = self.state.get(source_id, {}) |
| 2212 | prev_ids = set(prev.get("seen_ids", [])) |
| 2213 | |
| 2214 | try: |
| 2215 | resp = self.session.get(url, timeout=30) |
| 2216 | resp.raise_for_status() |
| 2217 | except Exception as e: |
| 2218 | print(f" ERROR PMDA Recalls: {e}") |
| 2219 | return None |
| 2220 | |
| 2221 | entries = self._parse_recall_table(resp.text, recall_class) |
| 2222 | if not entries: |
| 2223 | print(f" WARNING: No entries from PMDA recall page") |
| 2224 | return None |
| 2225 | |
| 2226 | print(f" INFO: Parsed {len(entries)} Class {recall_class} recall entries from PMDA") |
| 2227 | |
| 2228 | new_items = [] |
| 2229 | all_ids = list(prev_ids) |
| 2230 | cutoff = (datetime.now() - timedelta(days=60)).strftime("%Y-%m-%d") |
| 2231 | |
| 2232 | for entry in entries: |
| 2233 | eid = entry.get("id", "") |
| 2234 | if not eid or eid in prev_ids: |
| 2235 | continue |
| 2236 | all_ids.append(eid) |
| 2237 | pub = entry.get("pub_date", "") |
| 2238 | if pub and pub < cutoff: |
| 2239 | continue |
| 2240 | new_items.append(entry) |
| 2241 | |
| 2242 | self.state[source_id] = { |
| 2243 | "url": url, |
| 2244 | "last_checked": datetime.now().isoformat(), |
| 2245 | "seen_ids": all_ids[-500:], |
| 2246 | } |
| 2247 | |
| 2248 | if new_items and (prev_ids or self.seed_mode): |
| 2249 | if self.seed_mode and not prev_ids: |