src/context.py

"""Evidence ranking, deduplication, and truncation."""

SOURCE_PRIORITY = {
    "DMD_ExonSkip_Lookup": -2,
    "Variant_Therapy_Lookup": -2,
    "Supportive_Care_Lookup": -2,
    "ACMG_SF": -1,
    "ACMG_Guideline": -1,
    "CPIC": -1,
    "N1C_Eligibility": -1,
    "FDA": 0,
    "DailyMed": 0,
    "GeneReviews": 1,
    "ClinGen": 1,
    "ClinGen_Validity": 1,
    "ClinVar": 2,
    "Orphanet": 2,
    "GenCC": 2,
    "PharmGKB": 2,
    "HPO_MONDO": 2,
    "ClinicalTrials.gov": 3,
    "Ensembl": 4,
    "Ensembl_VEP": 4,
    "UniProt": 4,
    "ChEMBL": 5,
    "PubMed": 5,
    "PHAROS": 5,
    "DGIdb": 5,
    "Open Targets": 6,
    "OMIM": 7,
    "gnomAD": 8,
}


def _estimate_tokens(text: str) -> int:
    return len(text) // 4


def truncate_evidence(evidence: list[dict], max_tokens: int = 3000) -> list[dict]:
    """Rank, deduplicate, and truncate evidence to fit within token budget."""
    # Deduplicate by URL, keeping the first occurrence — but merge the
    # strongest `_disease_relevance` across duplicates. The same trial can be
    # retrieved by several queries (broad gene search vs. targeted condition
    # search) that score disease-relevance differently; keeping only the first
    # (often a low-relevance gene-query hit) would let a 0.0 duplicate mask a
    # 1.0 hit and wrongly disqualify the right trial (-> spurious "None").
    seen_urls: dict[str, dict] = {}
    unique = []
    for e in evidence:
        url = e["url"]
        if url not in seen_urls:
            seen_urls[url] = e
            unique.append(e)
        elif e.get("_disease_relevance", 0) > seen_urls[url].get("_disease_relevance", 0):
            seen_urls[url]["_disease_relevance"] = e.get("_disease_relevance")

    def sort_key(e):
        base = SOURCE_PRIORITY.get(e["source_name"], 99)
        snippet = e.get("snippet", "")
        if "maps to exon" in snippet or "DOMAIN MATCH" in snippet:
            return (-1, base)
        return (0, base)

    unique.sort(key=sort_key)

    selected = []
    total_tokens = 0
    for e in unique:
        snippet_tokens = _estimate_tokens(e["snippet"])
        if total_tokens + snippet_tokens > max_tokens:
            break
        selected.append(e)
        total_tokens += snippet_tokens

    return selected