(self, source_id: str, source: dict)
| 958 | self.state = state |
| 959 | |
| 960 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 961 | prev = self.state.get(source_id, {}) |
| 962 | last_checked = prev.get("last_checked") |
| 963 | date_filter = "" |
| 964 | if last_checked: |
| 965 | try: |
| 966 | dt = datetime.fromisoformat(last_checked) |
| 967 | date_filter = f"+AND+effective_date:[{dt.strftime('%Y%m%d')}+TO+*]" |
| 968 | except Exception: |
| 969 | pass |
| 970 | params = { |
| 971 | "search": f'product_type:"DEVICE"+AND+status:"Final"{date_filter}', |
| 972 | "limit": 20, "sort": "effective_date:desc", |
| 973 | } |
| 974 | if self.api_key: |
| 975 | params["api_key"] = self.api_key |
| 976 | try: |
| 977 | resp = requests.get(self.GUIDANCE_URL, params=params, timeout=20) |
| 978 | if resp.status_code == 404: |
| 979 | self.state[source_id] = {"last_checked": datetime.now().isoformat()} |
| 980 | return None |
| 981 | resp.raise_for_status() |
| 982 | results = resp.json().get("results", []) |
| 983 | self.state[source_id] = {"last_checked": datetime.now().isoformat()} |
| 984 | if not results: |
| 985 | return None |
| 986 | new_items = [] |
| 987 | for r in results: |
| 988 | doc_id = r.get("id", "") |
| 989 | url = r.get("url", "") or ( |
| 990 | f"https://www.fda.gov/regulatory-information/" |
| 991 | f"search-fda-guidance-documents/{doc_id}" if doc_id else "") |
| 992 | new_items.append({"title": r.get("title", ""), "link": url, |
| 993 | "pub_date": r.get("effective_date", ""), "doc_id": doc_id}) |
| 994 | result = _make_update(source_id, source, "openfda_guidance", |
| 995 | f"{len(new_items)} new/updated FDA Final guidance via OpenFDA API") |
| 996 | result["new_items"] = new_items |
| 997 | return result |
| 998 | except Exception as e: |
| 999 | print(f" ERROR OpenFDA: {e}") |
| 1000 | return None |
| 1001 | |
| 1002 | |
| 1003 | # --------------------------------------------------------------------------- |
nothing calls this directly
no test coverage detected