(self, source_id: str, source: dict)
| 1484 | self.state = state |
| 1485 | |
| 1486 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 1487 | prev = self.state.get(source_id, {}) |
| 1488 | last_recall_nums = set(prev.get("seen_recall_numbers", [])) |
| 1489 | |
| 1490 | two_weeks_ago = (datetime.now() - timedelta(days=14)).strftime("%Y%m%d") |
| 1491 | today = datetime.now().strftime("%Y%m%d") |
| 1492 | |
| 1493 | search_q = ( |
| 1494 | f'classification:"Class I"' |
| 1495 | f' AND report_date:[{two_weeks_ago} TO {today}]' |
| 1496 | ) |
| 1497 | url = f"{self.ENFORCEMENT_URL}?search={requests.utils.quote(search_q)}&limit=20" |
| 1498 | if self.api_key: |
| 1499 | url += f"&api_key={self.api_key}" |
| 1500 | |
| 1501 | try: |
| 1502 | resp = requests.get(url, timeout=20) |
| 1503 | if resp.status_code == 404: |
| 1504 | self.state[source_id] = { |
| 1505 | "url": source["url"], |
| 1506 | "last_checked": datetime.now().isoformat(), |
| 1507 | "seen_recall_numbers": list(last_recall_nums), |
| 1508 | } |
| 1509 | return None |
| 1510 | resp.raise_for_status() |
| 1511 | results = resp.json().get("results", []) |
| 1512 | except Exception as e: |
| 1513 | print(f" ERROR OpenFDA Enforcement: {e}") |
| 1514 | return None |
| 1515 | |
| 1516 | new_items = [] |
| 1517 | all_recall_nums = list(last_recall_nums) |
| 1518 | |
| 1519 | for r in results: |
| 1520 | recall_num = r.get("recall_number", "") |
| 1521 | if not recall_num or recall_num in last_recall_nums: |
| 1522 | continue |
| 1523 | all_recall_nums.append(recall_num) |
| 1524 | |
| 1525 | product_desc = r.get("product_description", "")[:200] |
| 1526 | reason = r.get("reason_for_recall", "") |
| 1527 | firm = r.get("recalling_firm", "") |
| 1528 | report_date = r.get("report_date", "") |
| 1529 | quantity = r.get("product_quantity", "") |
| 1530 | |
| 1531 | title = f"Class I Recall: {firm} - {product_desc[:80]}" |
| 1532 | description = ( |
| 1533 | f"Recall #{recall_num} by {firm}. " |
| 1534 | f"Reason: {reason[:200]}. " |
| 1535 | f"Quantity: {quantity}." |
| 1536 | ) |
| 1537 | |
| 1538 | new_items.append({ |
| 1539 | "title": title, |
| 1540 | "link": f"https://www.accessdata.fda.gov/scripts/cdrh/cfdocs/cfres/res.cfm?id={recall_num}", |
| 1541 | "pub_date": self._format_date(report_date), |
| 1542 | "description": description, |
| 1543 | "recall_number": recall_num, |
nothing calls this directly
no test coverage detected