avoigt1121
Claude Opus 5.5
feat(trials): show the date ClinicalTrials.gov processed the data, and our changes
e93b289 | """trials_for_genes — ClinicalTrials.gov studies whose intervention matches the gene. | |
| Memo decision 3: ToolUniverse for interactive lookups; lit-agent's own harvester keeps the | |
| weekly digest. ToolUniverse narrows an intervention search to the intervention's name and other | |
| name (Phase 0 finding 5: "CD40" gives the one pancreas-specific mitazalimab trial, where a broad | |
| API query also returns two solid-tumour basket trials). That rewrite is helpful but silent, so | |
| every payload states it: ``query_semantics`` in words, and ``executed_queries`` with the exact | |
| query ClinicalTrials.gov ran, taken from the response itself and pinned by a recorded fixture. | |
| ClinicalTrials.gov's terms ask a redistributor to show the date ClinicalTrials.gov processed the | |
| data and to state any changes: once a search has answered, one more call reads that date | |
| (``dataTimestamp``, src/direct_calls.py) and the payload carries the notice of | |
| :mod:`src.ctgov_terms` in ``source_terms`` and as the narrative's last sentence. Each row also | |
| carries ``ctgov_data_processed_on``, so a relay that shows only rows keeps the date. | |
| """ | |
| from __future__ import annotations | |
| from typing import Any | |
| from src import ctgov_terms, payloads | |
| from src.endpoints._common import Budget, Collector, according_to, get_client, stamped | |
| from src.identifiers import parse_disease, parse_gene_list | |
| ENDPOINT = "trials_for_genes" | |
| CT_SEARCH = "ClinicalTrials_search_studies" | |
| DEFAULT_CONDITION = "pancreatic cancer" | |
| DEFAULT_STATUS = "RECRUITING" | |
| STATUSES = ( | |
| "RECRUITING", | |
| "NOT_YET_RECRUITING", | |
| "ACTIVE_NOT_RECRUITING", | |
| "ENROLLING_BY_INVITATION", | |
| "COMPLETED", | |
| "ANY", | |
| ) | |
| PAGE_SIZE = 100 | |
| QUERY_SEMANTICS = ( | |
| "A study is listed when its intervention name or intervention other-name matches the gene " | |
| "symbol (ToolUniverse's rewrite of the ClinicalTrials.gov query). Studies that mention the " | |
| "gene only in their title, outcomes or eligibility are not listed, and a match on the name is " | |
| "not evidence that the intervention acts on the gene's product." | |
| ) | |
| def trials_for_genes(genes: str = "", condition: str = "", status: str = "") -> dict[str, Any]: | |
| parsed = parse_gene_list(genes, allow_ensembl=False) | |
| if not parsed.ok: | |
| return payloads.refused(ENDPOINT, parsed.reason, parsed.message) | |
| cond = parse_disease(condition or DEFAULT_CONDITION) | |
| if not cond.ok: | |
| return payloads.refused(ENDPOINT, cond.reason, cond.message) | |
| if cond.identifiers[0].kind != "disease_name": | |
| return payloads.refused( | |
| ENDPOINT, | |
| "invalid_identifier", | |
| "Give the condition as a name, e.g. 'pancreatic cancer'.", | |
| ) | |
| wanted = (status or DEFAULT_STATUS).strip().upper() | |
| if wanted not in STATUSES: | |
| return payloads.refused( | |
| ENDPOINT, | |
| "invalid_argument", | |
| f"status must be one of {', '.join(STATUSES)}; got {wanted[:24]!r}.", | |
| ) | |
| condition_name = cond.identifiers[0].value | |
| client, budget, out = get_client(), Budget(), Collector() | |
| executed: list[dict[str, Any]] = [] | |
| per_gene: list[tuple[str, int, int]] = [] | |
| symbols = [i.value for i in parsed.identifiers] | |
| for idx, gene in enumerate(symbols): | |
| if budget.spent: | |
| out.out_of_budget(symbols[idx:]) | |
| break | |
| args: dict[str, Any] = { | |
| "query_cond": condition_name, | |
| "query_intr": gene, | |
| "page_size": PAGE_SIZE, | |
| } | |
| if wanted != "ANY": | |
| args["filter_status"] = wanted | |
| result = out.call(client, budget, CT_SEARCH, args) | |
| if result is None: | |
| continue | |
| data = result.data | |
| executed.append({"gene": gene, "executed_query": data.get("executed_query")}) | |
| studies = data.get("studies") or [] | |
| for study in studies: | |
| nct = study["nct_id"] | |
| out.rows.append( | |
| stamped( | |
| { | |
| "gene": gene, | |
| "nct_id": nct, | |
| "title": study.get("brief_title"), | |
| "status": study.get("status"), | |
| "phases": study.get("phases") or [], | |
| "study_type": study.get("study_type"), | |
| "conditions": study.get("conditions") or [], | |
| "interventions": study.get("interventions") or [], | |
| "sponsor": study.get("sponsor"), | |
| "start_date": study.get("start_date"), | |
| "completion_date": study.get("completion_date"), | |
| }, | |
| result, | |
| ref=nct, | |
| url=f"https://clinicaltrials.gov/study/{nct}", | |
| ) | |
| ) | |
| per_gene.append((gene, len(studies), int(data.get("total_count") or len(studies)))) | |
| if not out.rows and out.source_errors and not per_gene: | |
| return payloads.error( | |
| ENDPOINT, | |
| "all_sources_failed", | |
| "ClinicalTrials.gov did not answer; nothing was looked up.", | |
| source_errors=out.source_errors, | |
| ) | |
| terms = _terms_notice(client, budget, out) if per_gene else None | |
| if terms is not None: | |
| for row in out.rows: | |
| row["ctgov_data_processed_on"] = terms["data_processed_on"] | |
| lead = according_to(["clinicaltrials"], out.retrieved) | |
| kind = "of any status" if wanted == "ANY" else wanted | |
| clauses = [] | |
| for gene, shown, total in per_gene: | |
| ids = [r["nct_id"] for r in out.rows if r["gene"] == gene] | |
| listed = f" ({', '.join(ids[:10])}" + ( | |
| f" and {len(ids) - 10} more" if len(ids) > 10 else "" | |
| ) | |
| listed += ")" if ids else "" | |
| more = f", showing the first {shown}" if total > shown else "" | |
| noun = "study" if total == 1 else "studies" | |
| clauses.append(f"{gene}: {total} {kind} {noun}{listed}{more}") | |
| sentences = [] | |
| if clauses: | |
| sentences.append( | |
| f"{lead}, studies for '{condition_name}' with an intervention matching each gene — " | |
| + "; ".join(clauses) | |
| + "." | |
| ) | |
| sentences.append("Matching is by intervention name only; see query_semantics.") | |
| if out.source_errors: | |
| sentences.append( | |
| "Source problems: " | |
| + "; ".join(f"{e['source']} {e['reason']}" for e in out.source_errors) | |
| + "." | |
| ) | |
| if terms is not None: | |
| sentences.append(terms["sentence"]) | |
| return payloads.ok( | |
| ENDPOINT, | |
| query={"genes": symbols, "condition": condition_name, "status": wanted}, | |
| rows=out.rows, | |
| sources_used=out.sources_used or {"clinicaltrials"}, | |
| narrative=" ".join(sentences), | |
| unresolved=out.unresolved, | |
| source_errors=out.source_errors, | |
| query_semantics=QUERY_SEMANTICS, | |
| executed_queries=executed, | |
| source_terms=[terms] if terms is not None else [], | |
| ) | |
| def _terms_notice(client, budget: Budget, out: Collector) -> dict[str, Any]: | |
| """The ClinicalTrials.gov notice, with the processed date read from ClinicalTrials.gov. | |
| A failure here never hides the studies already found: the notice says the date could not be | |
| read and why, and the failure is not a ``source_errors`` entry (the search itself answered). | |
| """ | |
| if budget.spent: | |
| return ctgov_terms.notice(None, "the call budget ran out before it could be read") | |
| result, err = out.attempt(client, budget, ctgov_terms.DATE_TOOL, {}) | |
| if err is not None: | |
| return ctgov_terms.notice(None, f"{err.reason}: {err.detail}") | |
| return ctgov_terms.notice(result.data["dataTimestamp"]) | |