Fetch FDA guidance documents using curl-cffi to bypass Akamai bot detection.
| 87 | "fda-remanufacturing-2024": "https://www.fda.gov/media/150141/download", |
| 88 | } |
| 89 | |
| 90 | |
| 91 | class FDAFetcher: |
| 92 | """Fetch FDA guidance documents using curl-cffi to bypass Akamai bot detection.""" |
| 93 | |
| 94 | FDA_GUIDANCE_BASE = "https://www.fda.gov/regulatory-information/search-fda-guidance-documents/" |
| 95 | FDA_MEDIA_PATTERN = re.compile(r"https://www\.fda\.gov/media/(\d+)/download") |
| 96 | IMPERSONATE = "chrome124" |
| 97 | |
| 98 | @staticmethod |
| 99 | def _cffi_get(url: str, timeout: int = 30, **kwargs) -> Optional[object]: |
| 100 | """Use curl-cffi with Chrome TLS impersonation, bypassing proxy for FDA. |
| 101 | Akamai bot detection blocks requests from proxy/VPN IPs. |
| 102 | """ |
| 103 | if cffi_requests is not None: |
| 104 | saved = {k: os.environ.pop(k, None) for k in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY", "all_proxy", "ALL_PROXY")} |
| 105 | try: |
| 106 | session = cffi_requests.Session(impersonate=FDAFetcher.IMPERSONATE) |
| 107 | return session.get(url, timeout=timeout, allow_redirects=True, **kwargs) |
| 108 | finally: |
| 109 | for k, v in saved.items(): |
| 110 | if v is not None: |
| 111 | os.environ[k] = v |
| 112 | return requests.get(url, headers=REQUEST_HEADERS, timeout=timeout, allow_redirects=True, **kwargs) |
| 113 | |
| 114 | @staticmethod |
| 115 | def discover_pdf_url(source_url: str) -> Optional[str]: |
| 116 | """Try to discover the PDF download URL from an FDA guidance page.""" |
| 117 | try: |
| 118 | resp = FDAFetcher._cffi_get(source_url, timeout=30) |
| 119 | resp.raise_for_status() |
| 120 | content_type = resp.headers.get("content-type", "") |
| 121 | if "pdf" in content_type.lower(): |
| 122 | return source_url |
| 123 | |
| 124 | text = resp.text |
| 125 | pdf_links = re.findall( |
| 126 | r'href=["\']?(https://www\.fda\.gov/media/\d+/download)["\']?', |
| 127 | text, |
| 128 | ) |
| 129 | if pdf_links: |
| 130 | return pdf_links[0] |
| 131 | |
| 132 | media_links = re.findall( |
| 133 | r'href=["\']?(/media/\d+/download)["\']?', |
| 134 | text, |
| 135 | ) |
| 136 | if media_links: |
| 137 | return f"https://www.fda.gov{media_links[0]}" |
| 138 | |
| 139 | except Exception as e: |
| 140 | logger.warning(f" Failed to discover PDF URL from {source_url}: {e}") |
| 141 | return None |
| 142 | |
| 143 | @staticmethod |
| 144 | def fetch_pdf(pdf_url: str, source_url: str = "") -> Optional[bytes]: |
| 145 | """Fetch PDF using curl-cffi with Chrome TLS impersonation.""" |
| 146 | try: |
nothing calls this directly
no outgoing calls
no test coverage detected