(self, source_id: str, source: dict)
| 1280 | self.seed_mode = seed_mode |
| 1281 | |
| 1282 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 1283 | url = source["url"] |
| 1284 | prev = self.state.get(source_id, {}) |
| 1285 | prev_ids = set(prev.get("seen_ids", [])) |
| 1286 | |
| 1287 | try: |
| 1288 | resp = self.session.get(url, timeout=30, headers={ |
| 1289 | "Accept": "application/json", |
| 1290 | }) |
| 1291 | resp.raise_for_status() |
| 1292 | data = resp.json() |
| 1293 | except Exception as e: |
| 1294 | print(f" ERROR TFDA OpenData: {e}") |
| 1295 | return None |
| 1296 | |
| 1297 | if not isinstance(data, list) or not data: |
| 1298 | print(" WARNING: TFDA OpenData returned empty or non-list response") |
| 1299 | return None |
| 1300 | |
| 1301 | print(f" INFO: Parsed {len(data)} entries from TFDA OpenData API") |
| 1302 | |
| 1303 | new_items = [] |
| 1304 | all_ids = list(prev_ids) |
| 1305 | |
| 1306 | for row in data: |
| 1307 | title = row.get("Title", row.get("title", "")) |
| 1308 | if not title: |
| 1309 | continue |
| 1310 | link = row.get("Link", row.get("link", "")) |
| 1311 | pub_date = row.get("Release date", row.get("PublishDate", "")) |
| 1312 | if pub_date: |
| 1313 | pub_date = pub_date[:10] |
| 1314 | |
| 1315 | eid = link or title |
| 1316 | if eid in prev_ids: |
| 1317 | continue |
| 1318 | all_ids.append(eid) |
| 1319 | |
| 1320 | if not link or not link.startswith("http"): |
| 1321 | link = "https://www.fda.gov.tw/" |
| 1322 | |
| 1323 | new_items.append({ |
| 1324 | "title": title, |
| 1325 | "link": link, |
| 1326 | "pub_date": pub_date, |
| 1327 | "description": f"TFDA medical device safety alert: {title[:200]}", |
| 1328 | }) |
| 1329 | |
| 1330 | self.state[source_id] = { |
| 1331 | "url": url, |
| 1332 | "last_checked": datetime.now().isoformat(), |
| 1333 | "seen_ids": all_ids[-500:], |
| 1334 | } |
| 1335 | |
| 1336 | if new_items and (prev_ids or self.seed_mode): |
| 1337 | if self.seed_mode and not prev_ids: |
| 1338 | new_items = new_items[:10] |
| 1339 | print(f" INFO: Seed mode -- returning top {len(new_items)} TFDA entries") |
nothing calls this directly
no test coverage detected