Checks MFDS (Korea) English news page for medical device regulatory updates. Korea is transitioning from MFDS to DMPA (2026 reform). Captures: KGMP updates, device classification changes, regulatory reforms.
| 2318 | # --------------------------------------------------------------------------- |
| 2319 | |
| 2320 | class MFDSChecker: |
| 2321 | """Checks MFDS (Korea) English news page for medical device regulatory updates. |
| 2322 | |
| 2323 | Korea is transitioning from MFDS to DMPA (2026 reform). |
| 2324 | Captures: KGMP updates, device classification changes, regulatory reforms. |
| 2325 | """ |
| 2326 | |
| 2327 | def __init__(self, session: requests.Session, state: dict, |
| 2328 | seed_mode: bool = False): |
| 2329 | self.session = session |
| 2330 | self.state = state |
| 2331 | self.seed_mode = seed_mode |
| 2332 | |
| 2333 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 2334 | url = source["url"] |
| 2335 | prev = self.state.get(source_id, {}) |
| 2336 | prev_titles = set(prev.get("seen_titles", [])) |
| 2337 | title_filter = source.get("title_filter") |
| 2338 | |
| 2339 | try: |
| 2340 | resp = self.session.get(url, timeout=30) |
| 2341 | resp.raise_for_status() |
| 2342 | except Exception as e: |
| 2343 | print(f" ERROR MFDS: {e}") |
| 2344 | return None |
| 2345 | |
| 2346 | base_url = self._detect_base_url(url) |
| 2347 | entries = self._parse_news(resp.text, base_url) |
| 2348 | if not entries: |
| 2349 | print(f" WARNING: No entries from MFDS page ({url})") |
| 2350 | return None |
| 2351 | |
| 2352 | print(f" INFO: Parsed {len(entries)} entries from MFDS page") |
| 2353 | |
| 2354 | if title_filter: |
| 2355 | before = len(entries) |
| 2356 | entries = [e for e in entries |
| 2357 | if re.search(title_filter, e.get("title", ""), re.IGNORECASE)] |
| 2358 | print(f" INFO: {len(entries)} entries after title filter (from {before})") |
| 2359 | |
| 2360 | new_items = [] |
| 2361 | all_titles = list(prev_titles) |
| 2362 | |
| 2363 | for entry in entries: |
| 2364 | title = entry.get("title", "") |
| 2365 | if not title or title in prev_titles: |
| 2366 | continue |
| 2367 | all_titles.append(title) |
| 2368 | new_items.append(entry) |
| 2369 | |
| 2370 | self.state[source_id] = { |
| 2371 | "url": url, |
| 2372 | "last_checked": datetime.now().isoformat(), |
| 2373 | "seen_titles": all_titles[-300:], |
| 2374 | } |
| 2375 | |
| 2376 | if new_items and (prev_titles or self.seed_mode): |
| 2377 | if self.seed_mode and not prev_titles: |