|
import httpx, random |
|
|
|
_BASE_V2 = "https://clinicaltrials.gov/api/v2/studies" |
|
_BASE_V1 = "https://clinicaltrials.gov/api/query/study_fields" |
|
_HEADERS = { |
|
|
|
0: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " |
|
"(KHTML, like Gecko) Chrome/125 Safari/537.36", |
|
1: "Mozilla/5.0 (Macintosh; Intel Mac OS X 12_6) AppleWebKit/605.1.15 " |
|
"(KHTML, like Gecko) Version/16 Safari/605.1.15", |
|
2: "Mozilla/5.0 (X11; Linux x86_64) Gecko/20100101 Firefox/126.0", |
|
} |
|
|
|
async def _try_v2(term: str, n: int): |
|
p = { |
|
"query": term, |
|
"pageSize": n, |
|
"fields": ",".join([ |
|
"nctId", "briefTitle", "phase", "status", |
|
"startDate", "conditions", "interventions", |
|
]), |
|
} |
|
async with httpx.AsyncClient( |
|
headers={"User-Agent": _HEADERS[random.randint(0,2)]}, timeout=12 |
|
) as c: |
|
r = await c.get(_BASE_V2, params=p) |
|
if r.status_code == 403: |
|
raise RuntimeError("v2 blocked") |
|
r.raise_for_status() |
|
return r.json().get("studies", []) |
|
|
|
async def _try_v1(term: str, n: int): |
|
p = dict( |
|
expr=term, |
|
fields="NCTId,BriefTitle,Phase,OverallStatus,StartDate,Condition,InterventionName", |
|
max_rnk=n, min_rnk=1, fmt="json", |
|
) |
|
async with httpx.AsyncClient( |
|
headers={"User-Agent": _HEADERS[random.randint(0,2)]}, timeout=12 |
|
) as c: |
|
r = await c.get(_BASE_V1, params=p) |
|
r.raise_for_status() |
|
return r.json()["StudyFieldsResponse"]["StudyFields"] |
|
|
|
|
|
async def search_trials(term: str, max_studies: int = 20): |
|
try: |
|
return await _try_v2(term, max_studies) |
|
except Exception: |
|
try: |
|
return await _try_v1(term, max_studies) |
|
except Exception: |
|
return [] |
|
|