| 1006 | # --------------------------------------------------------------------------- |
| 1007 | |
| 1008 | class ECFRChecker: |
| 1009 | TITLES_URL = "https://www.ecfr.gov/api/versioner/v1/titles.json" |
| 1010 | |
| 1011 | def __init__(self, session: requests.Session, state: dict): |
| 1012 | self.session = session |
| 1013 | self.state = state |
| 1014 | |
| 1015 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 1016 | prev = self.state.get(source_id, {}) |
| 1017 | last_amendment_date = prev.get("last_amendment_date", "") |
| 1018 | try: |
| 1019 | resp = self.session.get(self.TITLES_URL, timeout=15) |
| 1020 | resp.raise_for_status() |
| 1021 | titles = resp.json().get("titles", []) |
| 1022 | title21 = None |
| 1023 | for t in titles: |
| 1024 | if t.get("number") == 21: |
| 1025 | title21 = t |
| 1026 | break |
| 1027 | if not title21: |
| 1028 | return None |
| 1029 | new_date = title21.get("latest_amended_on", "") or title21.get("up_to_date_as_of", "") |
| 1030 | self.state[source_id] = { |
| 1031 | "url": source["url"], |
| 1032 | "last_checked": datetime.now().isoformat(), |
| 1033 | "last_amendment_date": new_date, |
| 1034 | } |
| 1035 | if not last_amendment_date: |
| 1036 | print(f" INFO: eCFR baseline established (Title 21 last amended: {new_date})") |
| 1037 | return None |
| 1038 | if new_date and new_date > last_amendment_date: |
| 1039 | return _make_update( |
| 1040 | source_id, source, "ecfr_api", |
| 1041 | f"Title 21 CFR updated: {last_amendment_date} -> {new_date} (medical device Parts 800-898 may be affected)" |
| 1042 | ) |
| 1043 | return None |
| 1044 | except Exception as e: |
| 1045 | print(f" ERROR eCFR API: {e}") |
| 1046 | return None |
| 1047 | |
| 1048 | |
| 1049 | # --------------------------------------------------------------------------- |