src/endopath/gdc_longitudinal.py

"""Map a GDC case record to the `longitudinal_records` rows the app stores (#177).

The longitudinal source is the GDC clinical record for each TCGA-UCEC participant: the follow-ups (with the
disease response and the recurrence flag), the treatments, the observation time, and the demographic (which
carries vital status and days to death). This module is the pure mapping from one GDC `/cases` hit to the
rows the store holds; `scripts/ingest_longitudinal.py` does the HTTP pull and the write.

Two properties this mapping fixes over the earlier CSV load:

1. Every record is keyed on its GDC-native UUID (`follow_up_id`, `treatment_id`, `demographic_id`), not a
   synthesized `{participant}_{type}_{index}` string. That string collided (a treatment index resets per
   diagnosis block), so the earlier load, keyed on a PRIMARY KEY, was lossy. A UUID key is unique and makes
   the load idempotent.

2. The demographic entity is pulled, so survival arrives: `vital_status` and `days_to_death`, which the
   earlier CSV omitted entirely. That turns the survival columns of the export from a coded-absence
   placeholder into real values (`docs/prd.md` sections 4.1, 4.3).

The `structured_data` column keeps the raw GDC entity as JSON, for provenance and so the export projection
(`longitudinal.py`) reads the same field names GDC uses.
"""

from __future__ import annotations

from typing import Any, Optional

# The GDC `/cases` fields to request, as explicit dotted paths. `expand` triggers a GDC 500 on this project,
# so every leaf is named. Follow-up carries the recurrence signal; the demographic carries survival.
GDC_FIELDS: list[str] = [
    "submitter_id",
    "demographic.demographic_id",
    "demographic.vital_status",
    "demographic.days_to_death",
    "diagnoses.diagnosis_id",
    "diagnoses.days_to_last_follow_up",
    "diagnoses.treatments.treatment_id",
    "diagnoses.treatments.submitter_id",
    "diagnoses.treatments.treatment_type",
    "diagnoses.treatments.therapeutic_agents",
    "diagnoses.treatments.days_to_treatment_start",
    "diagnoses.treatments.days_to_treatment_end",
    "diagnoses.treatments.treatment_intent_type",
    "diagnoses.treatments.treatment_or_therapy",
    "follow_ups.follow_up_id",
    "follow_ups.days_to_follow_up",
    "follow_ups.disease_response",
    "follow_ups.progression_or_recurrence",
    "follow_ups.progression_or_recurrence_type",
    "follow_ups.progression_or_recurrence_anatomic_site",
    "follow_ups.days_to_recurrence",
    "follow_ups.days_to_progression",
]


def normalize_treatment_type(treatment_type: str) -> str:
    """The `record_type` for a treatment: `treatment_` plus the GDC `treatment_type` lowercased with spaces
    as underscores, so "Radiation Therapy, NOS" becomes `treatment_radiation_therapy,_nos`, matching the
    record-type vocabulary the export and the completeness panel already read."""
    return "treatment_" + treatment_type.strip().lower().replace(" ", "_")


def _num(value: Any) -> Optional[float]:
    if value is None or value == "":
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _int(value: Any) -> Optional[int]:
    n = _num(value)
    return int(n) if n is not None else None


def _row(
    *,
    record_id: str,
    participant_id: str,
    record_type: str,
    record_index: int,
    data: dict,
    ingested_at: str,
    days_to_followup: Optional[int] = None,
    days_to_treatment: Optional[int] = None,
    days_to_recurrence: Optional[int] = None,
    recurrence_type: Optional[str] = None,
    therapy_type: Optional[str] = None,
) -> dict:
    """One `longitudinal_records` row. The flat day/type columns mirror the earlier shape for the read
    endpoint; `structured_data` holds the raw GDC entity."""
    return {
        "record_id": record_id,
        "participant_id": participant_id,
        "case_barcode": participant_id,
        "record_type": record_type,
        "record_index": record_index,
        "days_to_followup": days_to_followup,
        "days_to_treatment": days_to_treatment,
        "days_to_recurrence": days_to_recurrence,
        "days_to_lost_to_followup": None,
        "biopsy_result": None,
        "followup_treatment_success": None,
        "recurrence_type": recurrence_type,
        "therapy_type": therapy_type,
        "lost_to_followup": None,
        "structured_data": data,
        "ingested_at": ingested_at,
    }


def records_from_case(case: dict, *, ingested_at: str) -> list[dict]:
    """Every `longitudinal_records` row for one GDC case hit.

    Emits one `follow_up` per GDC follow-up (keyed on `follow_up_id`), one `treatment_<type>` per treatment
    (keyed on `treatment_id`), one derived per-participant `last_followup` (the observation time), and one
    `demographic` carrying survival. A follow-up or treatment GDC did not return is simply absent; the row is
    never fabricated. `structured_data` is left as the raw GDC dict for provenance.
    """
    participant_id = case.get("submitter_id")
    if not participant_id:
        return []

    rows: list[dict] = []
    index = 0

    follow_ups = case.get("follow_ups") or []
    for follow_up in follow_ups:
        fid = follow_up.get("follow_up_id")
        if not fid:
            continue
        index += 1
        rows.append(
            _row(
                record_id=fid,
                participant_id=participant_id,
                record_type="follow_up",
                record_index=index,
                data=follow_up,
                ingested_at=ingested_at,
                days_to_followup=_int(follow_up.get("days_to_follow_up")),
                days_to_recurrence=_int(follow_up.get("days_to_recurrence")),
                recurrence_type=follow_up.get("progression_or_recurrence_type"),
            )
        )

    for diagnosis in case.get("diagnoses") or []:
        for treatment in diagnosis.get("treatments") or []:
            tid = treatment.get("treatment_id")
            treatment_type = treatment.get("treatment_type")
            if not tid or not treatment_type:
                continue
            index += 1
            rows.append(
                _row(
                    record_id=tid,
                    participant_id=participant_id,
                    record_type=normalize_treatment_type(treatment_type),
                    record_index=index,
                    data=treatment,
                    ingested_at=ingested_at,
                    days_to_treatment=_int(treatment.get("days_to_treatment_start")),
                    therapy_type=treatment_type,
                )
            )

    # The observation/censoring time: the largest days-to-last-follow-up any diagnosis carries, or the latest
    # follow-up day. One derived record per participant, keyed deterministically.
    last_day = _last_follow_up_day(case)
    if last_day is not None:
        index += 1
        rows.append(
            _row(
                record_id=f"{participant_id}_last_followup",
                participant_id=participant_id,
                record_type="last_followup",
                record_index=index,
                data={"days_to_last_follow_up": last_day},
                ingested_at=ingested_at,
                days_to_followup=last_day,
            )
        )

    # Survival: the demographic entity, pulled so the export's survival columns carry real values rather than
    # a coded-absence placeholder. Keyed on the GDC demographic UUID when present.
    demographic = case.get("demographic") or {}
    if demographic.get("vital_status") is not None or demographic.get("days_to_death") is not None:
        demographic_id = demographic.get("demographic_id") or f"{participant_id}_demographic"
        index += 1
        rows.append(
            _row(
                record_id=demographic_id,
                participant_id=participant_id,
                record_type="demographic",
                record_index=index,
                data=demographic,
                ingested_at=ingested_at,
            )
        )

    return rows


def _last_follow_up_day(case: dict) -> Optional[int]:
    candidates: list[int] = []
    for diagnosis in case.get("diagnoses") or []:
        day = _int(diagnosis.get("days_to_last_follow_up"))
        if day is not None:
            candidates.append(day)
    for follow_up in case.get("follow_ups") or []:
        day = _int(follow_up.get("days_to_follow_up"))
        if day is not None:
            candidates.append(day)
    return max(candidates) if candidates else None