Create ctx.py

core/sources/ctx.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+import re
+import httpx
+
+from ..config import settings
+from ..validate import normalize_cas
+
+DTXSID_IN_HTML = re.compile(r"DTXSID\d{7,}")
+
+async def resolve_dtxsid_from_dashboard(search: str, http: httpx.AsyncClient) -> str | None:
+    url = f"https://comptox.epa.gov/dashboard/dsstoxdb/results?search={httpx.utils.quote(search, safe='')}"
+    r = await http.get(url, headers={"accept": "text/html"})
+    r.raise_for_status()
+    m = DTXSID_IN_HTML.search(r.text)
+    return m.group(0) if m else None
+
+async def genetox_summary_by_dtxsid(dtxsid: str, http: httpx.AsyncClient) -> dict:
+    # confirmed working endpoint pattern
+    url = f"{settings.ctx_base_url}/hazard/genetox/summary/search/by-dtxsid/{httpx.utils.quote(dtxsid, safe='')}"
+    headers = {"accept": "application/json"}
+    if settings.ctx_api_key:
+        headers["x-api-key"] = settings.ctx_api_key
+
+    r = await http.get(url, headers=headers)
+    if r.status_code == 401:
+        return {"ok": False, "error": "CTX 401: missing/invalid x-api-key (set CTX_API_KEY secret).", "status": 401, "url": url}
+    if r.status_code >= 400:
+        return {"ok": False, "error": f"CTX {r.status_code}: {r.text[:600]}", "status": r.status_code, "url": url}
+    return {"ok": True, "summary": r.json(), "status": r.status_code, "url": url}
+
+async def fetch_ctx_genetox(cas: str, http: httpx.AsyncClient) -> dict:
+    q = (cas or "").strip()
+    if not q:
+        return {"ok": False, "error": "Empty CAS."}
+
+    dtxsid = await resolve_dtxsid_from_dashboard(normalize_cas(q), http)
+    if not dtxsid:
+        return {
+            "ok": False,
+            "error": "No DTXSID found for this CAS via dashboard search.",
+            "resolveUrl": f"https://comptox.epa.gov/dashboard/dsstoxdb/results?search={httpx.utils.quote(q, safe='')}",
+        }
+
+    s = await genetox_summary_by_dtxsid(dtxsid, http)
+    return {"ok": s.get("ok", False), "dtxsid": dtxsid, **s}