Blame for app/ingest/ctgov.py

"""ClinicalTrials.gov API v2 normalization, verified against NCT01169259.

One row is one registry study. Its canonical effect is the first comparative
primary analysis with an interpretable effect scale; all primary analyses are
retained so that endpoint selection and factorial comparisons are auditable.
"""

import re

from app.ingest.common import empty_study, integer, normalize_date, number, pmid


def parse_p_value(value) -> tuple[float | None, str | None]:
    if value is None:
        return None, None
    match = re.fullmatch(r"\s*(<=|>=|<|>|=|≤|≥)?\s*([\d.eE+\-]+)\s*", str(value))
    if not match:
        return None, None
    parsed = number(match.group(2))
    if parsed is None or not 0 <= parsed <= 1:
        return None, None
    return parsed, {"≤": "<=", "≥": ">="}.get(match.group(1), match.group(1) or "=")


def effect_type(value: str | None) -> str | None:
    cleaned = re.sub(r"[^a-z0-9]", "", (value or "").lower())
    aliases = {
        "hazardratio": "HR", "hazardratiohr": "HR", "hr": "HR",
        "oddsratio": "OR", "oddsratioor": "OR", "or": "OR",
        "riskratio": "RR", "relativerisk": "RR", "riskratiorr": "RR", "rr": "RR",
        "standardizedmeandifference": "SMD", "standardisedmeandifference": "SMD",
        "cohensd": "SMD", "hedgesg": "SMD", "smd": "SMD",
        "meandifference": "MD", "differenceofmeans": "MD", "differenceinmeans": "MD",
        "meandifferencenet": "MD", "md": "MD",
        "riskdifference": "RD", "differenceinproportions": "RD",
    }
    return aliases.get(cleaned)


_CONTROL_TITLE = re.compile(
    r"\b(?:placebo|sham|control|usual care|standard (?:of )?care|no (?:treatment|intervention)|"
    r"vehicle|wait(?:ing)?[- ]?list)\b",
    re.IGNORECASE,
)
_PARTICIPANT_UNITS = {"participants", "subjects", "patients"}
_CONTROL_ARM_TYPES = {"PLACEBO_COMPARATOR", "ACTIVE_COMPARATOR", "SHAM_COMPARATOR",
                      "NO_INTERVENTION"}


def _arm_pair(outcome: dict, selected: list[str],
              arm_types: dict[str, str] | None = None) -> tuple[str, str] | None:
    """(intervention, comparator) group IDs, or None when the orientation is not evident.

    A posted analysis fixes the order. Otherwise exactly one of two groups must be the
    control, by its registered arm type or, failing that, by its title.
    """
    if len(selected) == 2:
        return selected[0], selected[1]
    groups = [group for group in outcome.get("groups", []) if group.get("id")]
    if selected or len(groups) != 2:
        return None
    kinds = [(arm_types or {}).get(" ".join(group.get("title", "").casefold().split()))
             for group in groups]
    controls = [kind in _CONTROL_ARM_TYPES for kind in kinds]
    if None in kinds or sum(controls) != 1 or "EXPERIMENTAL" not in kinds:
        controls = [bool(_CONTROL_TITLE.search(group.get("title", ""))) for group in groups]
    if controls == [False, True]:
        return groups[0]["id"], groups[1]["id"]
    if controls == [True, False]:
        return groups[1]["id"], groups[0]["id"]
    return None


def arm_summary(outcome: dict, selected: list[str],
                arm_types: dict[str, str] | None = None) -> dict:
    """Posted per-arm results for one outcome: means with SDs, or participant counts.

    Only a single class and category is read, so repeated timepoints or multi-level
    categories are never collapsed into one comparison. Medians, geometric and
    model-adjusted means are left alone.
    """
    pair = _arm_pair(outcome, selected, arm_types)
    classes = outcome.get("classes") or []
    if pair is None or len(classes) != 1 or len(classes[0].get("categories") or []) != 1:
        return {}
    sizes: dict = {}
    for denominator in [*(classes[0].get("denoms") or []), *(outcome.get("denoms") or [])]:
        if str(denominator.get("units", "")).lower() in _PARTICIPANT_UNITS:
            sizes = {item.get("groupId"): integer(item.get("value"))
                     for item in denominator.get("counts", [])}
            break
    measured = {item.get("groupId"): item
                for item in classes[0]["categories"][0].get("measurements") or []}
    if any(sizes.get(group) is None or group not in measured for group in pair):
        return {}
    param = str(outcome.get("paramType", "")).upper()
    dispersion = str(outcome.get("dispersionType", "")).upper().replace(" ", "_")
    values = [number(measured[group].get("value")) for group in pair]
    summary = {"n_intervention": sizes[pair[0]], "n_comparator": sizes[pair[1]]}
    if param == "COUNT_OF_PARTICIPANTS":
        events = [integer(measured[group].get("value")) for group in pair]
        if None in events or any(event > sizes[group] for event, group in zip(events, pair)):
            return {}
        return {**summary, "events_intervention": events[0], "events_comparator": events[1]}
    if param != "MEAN" or dispersion not in {"STANDARD_DEVIATION", "STANDARD_ERROR"}:
        return {}
    spreads = [number(measured[group].get("spread")) for group in pair]
    if None in values or None in spreads or any(spread <= 0 for spread in spreads):
        return {}
    if dispersion == "STANDARD_ERROR":
        spreads = [spread * sizes[group] ** 0.5 for spread, group in zip(spreads, pair)]
    return {**summary, "mean_intervention": values[0], "mean_comparator": values[1],
            "sd_intervention": spreads[0], "sd_comparator": spreads[1]}


def _analysis_row(outcome: dict, analysis: dict, outcome_index: int, index: int) -> dict:
    groups = {group["id"]: group for group in outcome.get("groups", []) if group.get("id")}
    selected = list(dict.fromkeys(analysis.get("groupIds") or []))
    selected_groups = [groups[group_id] for group_id in selected if group_id in groups]
    p_value, operator = parse_p_value(analysis.get("pValue"))
    ci_percent = number(analysis.get("ciPctValue"))
    denominators = None
    for denominator in outcome.get("denoms", []):
        if str(denominator.get("units", "")).lower() not in _PARTICIPANT_UNITS:
            continue
        counts = {item.get("groupId"): integer(item.get("value"))
                  for item in denominator.get("counts", [])}
        if selected and all(counts.get(group_id) is not None for group_id in selected):
            denominators = sum(counts[group_id] for group_id in selected)
            break
    return {
        "outcome": outcome.get("title", ""), "outcome_unit": outcome.get("unitOfMeasure", ""),
        "outcome_index": outcome_index, "analysis_index": index, "outcome_type": "PRIMARY",
        "time_frame": outcome.get("timeFrame", ""),
        "estimate": number(analysis.get("paramValue")),
        "ci_low": number(analysis.get("ciLowerLimit")),
        "ci_high": number(analysis.get("ciUpperLimit")),
        "ci_level": ci_percent / 100 if ci_percent is not None and 0 < ci_percent < 100 else None,
        "ci_sides": analysis.get("ciNumSides"),
        "p_value": p_value, "p_value_operator": operator,
        "p_value_raw": analysis.get("pValue"),
        "effect_type": effect_type(analysis.get("paramType")),
        "effect_type_raw": analysis.get("paramType"),
        "statistical_method": analysis.get("statisticalMethod"),
        "analysis_group_ids": selected, "n": denominators,
        "intervention": selected_groups[0].get("title", "") if selected_groups else "",
        "comparator": " vs ".join(group.get("title", "") for group in selected_groups[1:]),
        "analysis_is_comparative": len(selected) >= 2,
        "non_inferiority_type": analysis.get("nonInferiorityType"),
    }


def flatten_trial(record: dict) -> dict:
    protocol = record.get("protocolSection") or {}
    identity = protocol.get("identificationModule") or {}
    nct_id = identity.get("nctId", "").upper()
    if not re.fullmatch(r"NCT\d{8}", nct_id):
        raise ValueError("ClinicalTrials.gov record requires an NCT identifier")
    study = empty_study(nct_id, "ctgov")
    status = protocol.get("statusModule") or {}
    design = protocol.get("designModule") or {}
    enrollment = design.get("enrollmentInfo") or {}
    arms_module = protocol.get("armsInterventionsModule") or {}
    arms = arms_module.get("armGroups") or []
    interventions = arms_module.get("interventions") or []
    description = protocol.get("descriptionModule") or {}
    conditions = (protocol.get("conditionsModule") or {}).get("conditions") or []
    eligibility = protocol.get("eligibilityModule") or {}
    primary = (protocol.get("outcomesModule") or {}).get("primaryOutcomes") or []
    references = (protocol.get("referencesModule") or {}).get("references") or []
    result_pmids = sorted({p for ref in references if ref.get("type") == "RESULT"
                           if (p := pmid(ref.get("pmid")))})
    completion = normalize_date((status.get("primaryCompletionDateStruct") or {}).get("date"))
    year = int(completion[:4]) if completion else None
    arm_types = {arm.get("type") for arm in arms}
    control_types = _CONTROL_ARM_TYPES
    has_control = True if arm_types & control_types else None
    intervention_model = (design.get("designInfo") or {}).get("interventionModel")
    if intervention_model == "SINGLE_GROUP" or (len(arms) == 1 and design.get("studyType") == "INTERVENTIONAL"):
        has_control = False
    enrollment_count = integer(enrollment.get("count"))
    study.update({
        "title": identity.get("briefTitle") or identity.get("officialTitle") or nct_id,
        "abstract": description.get("briefSummary") or description.get("detailedDescription") or "",
        "venue": "ClinicalTrials.gov", "url": f"https://clinicaltrials.gov/study/{nct_id}",
        "year": year, "publication_date": f"{year:04d}-01-01" if year else None,
        "population": "; ".join(conditions),
        "eligibility": eligibility.get("eligibilityCriteria", ""),
        "intervention": "; ".join(item.get("name", "") for item in interventions),
        "comparator": "; ".join(arm.get("label", "") for arm in arms if arm.get("type") in control_types),
        "outcome": "; ".join(item.get("measure", "") for item in primary),
        "nct_ids": [nct_id], "pmids": result_pmids, "result_pmids": result_pmids,
        # A registry RESULT reference is evidence of reporting even if its paper is outside our index.
        "has_linked_publication": bool(result_pmids),
        "overall_status": status.get("overallStatus"), "why_stopped": status.get("whyStopped"),
        "primary_completion_date": completion, "primary_completion_date_raw":
            (status.get("primaryCompletionDateStruct") or {}).get("date"),
        "has_results": record.get("hasResults") if isinstance(record.get("hasResults"), bool) else None,
        "enrollment_actual": enrollment_count if enrollment.get("type") == "ACTUAL" else None,
        "enrollment_planned": enrollment_count if enrollment.get("type") == "ESTIMATED" else None,
        "n": enrollment_count if enrollment.get("type") == "ACTUAL" else None,
        "has_control": has_control, "study_design": design.get("studyType"),
        "result_label": "no_result_stated", "classification_method": "registry",
    })
    measures = ((record.get("resultsSection") or {}).get("outcomeMeasuresModule") or {}).get("outcomeMeasures") or []
    analyses = [_analysis_row(outcome, analysis, outcome_index, index)
                for outcome_index, outcome in enumerate(measures)
                if outcome.get("type") == "PRIMARY"
                for index, analysis in enumerate(outcome.get("analyses") or [])]
    study["registry_analyses"] = analyses
    supported = [row for row in analyses if row["analysis_is_comparative"] and row["effect_type"]
                 and any(row.get(key) is not None for key in ("estimate", "ci_low", "p_value"))]
    comparative = [row for row in analyses if row["analysis_is_comparative"]]
    selected = next(iter(supported or comparative), None)
    if selected:
        for key, value in selected.items():
            if key in {"intervention", "comparator", "n"} and value in (None, ""):
                continue
            study[key] = value
        # Registry numbers retain an exact JSON source path rather than a fabricated quote.
        study["numeric_source"] = ("resultsSection.outcomeMeasuresModule.outcomeMeasures"
                                   f"[{selected['outcome_index']}].analyses[{selected['analysis_index']}]")
        study["evidence_span"] = ""
        study["has_control"] = True
        study["evidence_tier"] = "structured"
        study["selection_note"] = ("First comparative primary outcome with a supported effect scale; "
                                   "other primary analyses are retained in registry_analyses.")
    primaries = [(index, outcome) for index, outcome in enumerate(measures)
                 if outcome.get("type") == "PRIMARY"]
    if selected:
        # Arm-level numbers must describe the same outcome and arms as the chosen analysis.
        primaries = [(index, outcome) for index, outcome in primaries
                     if index == selected["outcome_index"]]
    arm_types = {" ".join(arm.get("label", "").casefold().split()): arm.get("type")
                 for arm in arms if arm.get("label")}
    for index, outcome in primaries:
        summary = arm_summary(outcome, selected["analysis_group_ids"] if selected else [],
                              arm_types)
        if not summary:
            continue
        study.update(summary)
        if not selected:
            titles = {group.get("id"): group.get("title", "") for group in outcome.get("groups", [])}
            pair = _arm_pair(outcome, [], arm_types)
            study.update({
                "outcome": outcome.get("title", ""), "outcome_unit": outcome.get("unitOfMeasure", ""),
                "intervention": titles.get(pair[0]) or study["intervention"],
                "comparator": titles.get(pair[1]) or study["comparator"],
                "n": summary["n_intervention"] + summary["n_comparator"], "has_control": True,
                "numeric_source": f"resultsSection.outcomeMeasuresModule.outcomeMeasures[{index}]",
                "selection_note": "First primary outcome posting two-arm summary results; "
                                  "no comparative analysis was posted.",
            })
        break
    return study