Checks Health Canada recalls and safety alerts for medical devices. Uses the official Open Data JSON dataset from recalls-rappels.canada.ca. Medical device records are identified by 'Recall class' using 'Type I/II/III' (food/consumer products use 'Class 1/2/3' instead).
| 1819 | # --------------------------------------------------------------------------- |
| 1820 | |
| 1821 | class CanadaRecallsChecker: |
| 1822 | """Checks Health Canada recalls and safety alerts for medical devices. |
| 1823 | |
| 1824 | Uses the official Open Data JSON dataset from recalls-rappels.canada.ca. |
| 1825 | Medical device records are identified by 'Recall class' using 'Type I/II/III' |
| 1826 | (food/consumer products use 'Class 1/2/3' instead). |
| 1827 | """ |
| 1828 | |
| 1829 | DATA_URL = "https://recalls-rappels.canada.ca/sites/default/files/opendata-donneesouvertes/HCRSAMOpenData.json" |
| 1830 | |
| 1831 | _MEDICAL_DEVICE_CATEGORIES = { |
| 1832 | "Anaesthesiology", "Cardiovascular", "Chemistry", "Dental", |
| 1833 | "Ear, nose and throat", "Gastroenterology and urology", |
| 1834 | "General and plastic surgery", "General hospital and personal use", |
| 1835 | "In vitro diagnostics", "Microbiology", "Neurology", |
| 1836 | "Obstetrics and gynaecology", "Ophthalmology", "Orthopaedics", |
| 1837 | "Radiology", "Physical medicine", |
| 1838 | } |
| 1839 | |
| 1840 | def __init__(self, session: requests.Session, state: dict, |
| 1841 | seed_mode: bool = False): |
| 1842 | self.session = session |
| 1843 | self.state = state |
| 1844 | self.seed_mode = seed_mode |
| 1845 | |
| 1846 | _EXCLUDE_CATEGORIES = {"Drugs", "Alcoholic", "Dairy", "Herbs and spices", |
| 1847 | "Other", "Candy, confectionary, snacks and sweeten"} |
| 1848 | |
| 1849 | def _is_medical_device(self, record: dict) -> bool: |
| 1850 | category = record.get("Category", "") or "" |
| 1851 | if any(ec in category for ec in self._EXCLUDE_CATEGORIES): |
| 1852 | return False |
| 1853 | recall_class = record.get("Recall class", "") or "" |
| 1854 | if "Type " in recall_class: |
| 1855 | return True |
| 1856 | return any(mc in category for mc in self._MEDICAL_DEVICE_CATEGORIES) |
| 1857 | |
| 1858 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 1859 | prev = self.state.get(source_id, {}) |
| 1860 | prev_nids = set(str(n) for n in prev.get("seen_nids", [])) |
| 1861 | |
| 1862 | try: |
| 1863 | resp = self.session.get(self.DATA_URL, timeout=60) |
| 1864 | resp.raise_for_status() |
| 1865 | all_records = resp.json() |
| 1866 | except Exception as e: |
| 1867 | print(f" ERROR Canada Open Data: {e}") |
| 1868 | return None |
| 1869 | |
| 1870 | cutoff = (datetime.now() - timedelta(days=90)).strftime("%Y-%m-%d") |
| 1871 | md_records = [] |
| 1872 | for r in all_records: |
| 1873 | if not self._is_medical_device(r): |
| 1874 | continue |
| 1875 | updated = r.get("Last updated", "") or "" |
| 1876 | if updated >= cutoff: |
| 1877 | md_records.append(r) |
| 1878 |