(self, source_id: str, source: dict)
| 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: |
| 2250 | new_items = new_items[:10] |
| 2251 | print(f" INFO: Seed mode -- returning top {len(new_items)} recall entries") |
| 2252 | result = _make_update( |
| 2253 | source_id, source, "pmda_recalls", |
| 2254 | f"{len(new_items)} new PMDA Class {recall_class} recall(s) detected" |
| 2255 | ) |
| 2256 | result["new_items"] = new_items |
| 2257 | return result |
| 2258 | elif not prev_ids: |
| 2259 | print(f" INFO: Baseline established ({len(entries)} PMDA recall entries)") |
| 2260 | |
| 2261 | return None |
| 2262 |
nothing calls this directly
no test coverage detected