src/endopath/export.py

"""Project the confirmed case records onto the canonical concept list and render the cohort export
bundle (#101, docs/prd.md sections 3, 8.1).

The column spec is the canonical concept list (`mapping.concept_list()`, #91), not the built-in CAP
checklist: the export speaks concepts, so a second laboratory targeting the same concepts re-derives the
same table. `mapping.cap_checklist_coverage()` (#95) is the bridge that says which confirmed CAP field
fills which concept column. A concept no CAP field covers, and no derivation reaches, is absent, and an
absent value is never a blank: it carries a coded reason in a parallel `<concept>__absent_reason` column,
from the field-state vocabulary docs/prd.md section 4.3 defines (`not_applicable`, `not_stated`,
`indeterminate`, `determined_by_join`, `constrained`).

The two invariants this projection serves:

1. Numbers stay numbers. The stored quantity survives into its own column, and a derived category sits
   beside it in a separate column, recomputed from that quantity by the registry's named rule
   (`taxonomy.recompute`, #35), never stored in its place. A confirmed `myometrial_invasion_percent` of
   60.0 and the `half_or_more` category it derives to are two columns, so the export can still answer the
   other convention's question.
2. No value without its evidence. The field-level provenance carries each value's verbatim evidence quote,
   its page, and the pass that read it, and the bundle carries a checksum over every cited span per case,
   so a second centre auditing the row reads the same source material.

The bundle carries the schema the second centre re-derives against (the concept list and its checksum),
the version of the rule set and the extraction model that touched the values, the staging edition each
stage speaks (docs/prd.md section 4.2), and a per-case source-document reference with an evidence
checksum. The rendered CSV, XLSX, and JSON come from one projection, so the three files and the on-screen
preview never disagree.

## Relates to

- #40 owns the target-schema selector and the `derive_stage` set projection: choosing an arbitrary
  confirmed dictionary as the target, and rendering a `constrained` stage as the candidate set with the
  variable whose absence blocks narrowing. This module names the staging edition it emitted and reserves
  the `constrained` code for that projection; it does not itself enumerate the candidate set.
- #39 owns the deeper export provenance: the byte-level checksum of each scanned report and the positional
  region of every evidence span. This module carries the source-document reference and a checksum over the
  cited spans, the reproducible provenance available from the confirmed record alone.
- #35 owns the per-value coding-system resolution beyond the FIGO edition; this module cites each concept
  to the registry via its `cited_to` in the column spec.
"""

from __future__ import annotations

import base64
import csv
import hashlib
import io
import json
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Optional

from sqlalchemy import Connection, text

from endopath import llm_extraction, longitudinal, mapping, projection, taxonomy
from endopath.mapping import AssertionType, CompiledConcept
from endopath.schema import ChecklistField, FieldStatus
from endopath.validation import FieldState

# The coded-absence reason applied to a concept the confirmed record does not carry. A molecular finding
# absent from a report of this era arrives from the clinical record rather than the report, so it is
# `determined_by_join` (docs/prd.md sections 4.3, 8.3); every other uncovered concept, and a derived
# concept whose input primitives are absent, is `not_stated`, the searched-and-absent code.
_JOIN_SOURCED_CATEGORY = "molecular"

ABSENT_REASON_SUFFIX = "__absent_reason"


@dataclass(frozen=True)
class _Projected:
    """One concept's value on one case: a concrete value with no reason, or a coded absence with no
    value. `origin` records how the value was reached, for the column-spec descriptor."""

    value: Optional[Any]
    absent_reason: Optional[str]
    origin: str  # 'confirmed' | 'derived' | 'absent'


def _primitive(value: object) -> object:
    """An enum to its value, everything else as-is, so the export carries JSON primitives."""
    return value.value if hasattr(value, "value") else value


def _concept_labels() -> dict[str, str]:
    """Concept id to human label, read from the dependency-graph nodes the concept ids key on."""
    data = json.loads(mapping.DEPENDENCY_GRAPH_PATH.read_text(encoding="utf-8"))
    return {node["id"]: node["label"] for node in data["nodes"]}


def _field_value_and_state(field: ChecklistField) -> tuple[Optional[Any], FieldState]:
    """Reduce one confirmed CAP field to a value and a field state. A gated field is `not_applicable`; a
    flagged field was searched for and confirmed absent (`not_stated`); an applicable field with no value
    is `not_stated`; a field carrying a value is present. Silence is never the response to a miss."""
    if not field.applicable:
        return None, FieldState.NOT_APPLICABLE
    if field.status is FieldStatus.FLAGGED:
        return None, FieldState.NOT_STATED
    value = _primitive(field.value)
    if value is None:
        return None, FieldState.NOT_STATED
    return value, FieldState.PRESENT


def _project_case(
    checklist,
    *,
    concepts: list[CompiledConcept],
    coverage: dict[str, str],
) -> dict[str, _Projected]:
    """Project one confirmed checklist onto the concept list: seed the covered concepts from their CAP
    fields, recompute a derived concept whose inputs are all present, and code every remaining concept as
    a coded absence (invariant: no value without a reason)."""
    concept_to_field = {concept_id: field_name for field_name, concept_id in coverage.items()}
    projected: dict[str, _Projected] = {}

    # Pass 1: every covered concept, from its confirmed CAP field, so a derivation's inputs are available
    # before any derived concept is filled.
    for concept in concepts:
        field_name = concept_to_field.get(concept.concept_id)
        if field_name is None:
            continue
        value, state = _field_value_and_state(getattr(checklist, field_name))
        if state is FieldState.PRESENT:
            projected[concept.concept_id] = _Projected(value, None, "confirmed")
        else:
            projected[concept.concept_id] = _Projected(None, state.value, "absent")

    # Pass 2: the concepts no CAP field covers.
    for concept in concepts:
        cid = concept.concept_id
        if cid in projected:
            continue
        if concept.assertion_type is AssertionType.DERIVED and concept.derived is not None:
            projected[cid] = _derive(concept, projected)
        else:
            reason = (
                FieldState.DETERMINED_BY_JOIN
                if concept.category == _JOIN_SOURCED_CATEGORY
                else FieldState.NOT_STATED
            )
            projected[cid] = _Projected(None, reason.value, "absent")
    return projected


def _derive(concept: CompiledConcept, projected: dict[str, _Projected]) -> _Projected:
    """Recompute a derived concept from its inputs by its registry-owned rule (`taxonomy.recompute`,
    #35), the same discipline the B1 validation pass runs: recompute only when every graph input is
    present, so a derived category is computed from the primitive rather than trusted as stored. A
    derivation whose input a gate removed inherits `not_applicable`; one whose input is merely absent, or
    that the per-case pass does not recompute, is `not_stated`."""
    assert concept.derived is not None
    inputs = [projected.get(input_id) for input_id in concept.derived.inputs]
    if any(item is not None and item.absent_reason == FieldState.NOT_APPLICABLE.value for item in inputs):
        return _Projected(None, FieldState.NOT_APPLICABLE.value, "absent")
    if all(item is not None and item.value is not None for item in inputs):
        recomputed = taxonomy.recompute(concept.concept_id, [item.value for item in inputs])  # type: ignore[union-attr]
        if recomputed is not None:
            return _Projected(recomputed, None, "derived")
    return _Projected(None, FieldState.NOT_STATED.value, "absent")


def _evidence_checksum(checklist) -> str:
    """A SHA-256 over the case's cited evidence: the field, verbatim quote, page, and pass of every value
    that carries a span, in a canonical order. A second centre confirming the same values against the same
    reports reads the same spans and computes the same checksum (invariant 2)."""
    spans = []
    for field_name, field in sorted(checklist.all_fields().items()):
        if field.evidence is not None:
            spans.append(
                {
                    "field": field_name,
                    "quote": field.evidence.quote,
                    "page_number": field.evidence.page_number,
                    "source": field.evidence.source,
                }
            )
    canonical = json.dumps(spans, sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def _columns(concepts: list[CompiledConcept], labels: dict[str, str]) -> list[dict]:
    """The column spec: the case-level identifiers, then, per concept in concept-list order, a value
    column and its parallel absent-reason column. The descriptor carries the concept's category,
    assertion type, cardinality, and registry citation so the bundle and the screen render the spec
    without re-deriving it."""
    columns: list[dict] = [
        {"key": "case_barcode", "label": "Case barcode", "kind": "identifier"},
        {"key": "report_date", "label": "Report date", "kind": "identifier"},
        {"key": "staging_edition", "label": "Staging edition", "kind": "identifier"},
        {"key": "status", "label": "Case status", "kind": "identifier"},
    ]
    for concept in concepts:
        cid = concept.concept_id
        label = labels.get(cid, cid)
        columns.append(
            {
                "key": cid,
                "label": label,
                "kind": "value",
                "concept_id": cid,
                "category": concept.category,
                "assertion_type": concept.assertion_type.value,
                "cardinality": concept.cardinality.value,
                "cited_to": concept.cited_to,
                "derived": concept.derived.model_dump(mode="json") if concept.derived else None,
                "absent_reason_key": f"{cid}{ABSENT_REASON_SUFFIX}",
            }
        )
        columns.append(
            {
                "key": f"{cid}{ABSENT_REASON_SUFFIX}",
                "label": f"{label} absent reason",
                "kind": "absent_reason",
                "of": cid,
            }
        )
    columns.extend(_outcome_columns())
    return columns


def _outcome_columns() -> list[dict]:
    """The join-sourced outcome columns, appended after the concept columns: the disease outcome joined from
    the GDC clinical record rather than read from the pathology report (`longitudinal.py`). Each value column
    is marked `sourced_via: "join"` and paired with its parallel absent-reason column, so the same renderer
    and the same coded-absence discipline apply as for a concept column. The block keeps the report-concept
    schema and its checksum meaning "what the report answers"."""
    columns: list[dict] = []
    for spec in longitudinal.OUTCOME_COLUMNS:
        key = spec["key"]
        columns.append(
            {
                "key": key,
                "label": spec["label"],
                "kind": "value",
                "category": longitudinal.OUTCOME_CATEGORY,
                "assertion_type": spec["assertion_type"],
                "sourced_via": "join",
                "join_source": longitudinal.JOIN_SOURCE,
                "derived": spec.get("derived"),
                "note": spec.get("note"),
                "absent_reason_key": f"{key}{ABSENT_REASON_SUFFIX}",
            }
        )
        columns.append(
            {
                "key": f"{key}{ABSENT_REASON_SUFFIX}",
                "label": f"{spec['label']} absent reason",
                "kind": "absent_reason",
                "of": key,
            }
        )
    return columns


def _load_longitudinal(conn: Connection) -> dict[str, list[longitudinal.Record]]:
    """Every longitudinal record grouped by participant, in one read (issue #131's batched discipline). The
    outcome projection joins by participant barcode, and a per-case query would reintroduce the N+1 the
    cohort export was restructured to avoid. The table holds a few thousand small rows, so one unfiltered
    read is cheaper than a query per case."""
    by_participant: dict[str, list[longitudinal.Record]] = {}
    rows = conn.execute(
        text(
            "SELECT participant_id, record_id, record_type, structured_data FROM longitudinal_records"
        )
    ).mappings().fetchall()
    for row in rows:
        data: dict = {}
        if row["structured_data"]:
            try:
                data = json.loads(row["structured_data"])
            except (json.JSONDecodeError, TypeError):
                data = {}
        by_participant.setdefault(row["participant_id"], []).append(
            longitudinal.Record(
                record_type=row["record_type"], record_id=row["record_id"], data=data
            )
        )
    return by_participant


def _load_longitudinal_joins(conn: Connection) -> dict[str, list[dict]]:
    """Every reviewer join decision grouped by case barcode, in one read (issue #131's batched discipline).
    #180 writes these from the Case Review panel: a per-record disposition (`confirmed` / `rejected`) with
    the confirmer's identity and role. The export and the cohort endpoint read them to drop rejected records
    and to attribute the confirmed linkages. A record with no row here has not been reviewed, not rejected."""
    by_case: dict[str, list[dict]] = {}
    rows = conn.execute(
        text(
            "SELECT case_barcode, record_id, disposition, confirmed_by, role, confirmed_at "
            "FROM longitudinal_joins"
        )
    ).mappings().fetchall()
    for row in rows:
        by_case.setdefault(row["case_barcode"], []).append(
            {
                "record_id": row["record_id"],
                "disposition": row["disposition"],
                "confirmed_by": row["confirmed_by"],
                "role": row["role"],
                "confirmed_at": row["confirmed_at"],
            }
        )
    return by_case


def _rejected_record_ids(case_joins: list[dict]) -> set[str]:
    """The record ids a reviewer explicitly rejected for this case. The barcode join is deterministic, so a
    record with no join row is unreviewed, not rejected: only an explicit `rejected` disposition excludes a
    record from the projection, leaving the default (no decision) projecting exactly as before."""
    return {j["record_id"] for j in case_joins if j.get("disposition") == "rejected"}


def _bundle(
    concepts: list[CompiledConcept],
    sources: list[dict],
    *,
    staging_edition: Optional[str],
    per_case_editions: list[str],
) -> dict:
    """The provenance bundle (docs/prd.md section 3): the schema the second centre re-derives against and
    its checksum, the rule-set and extraction-model versions that touched the values, the staging edition
    each stage speaks, the coding systems, and the per-case source references."""
    schema_checksum = hashlib.sha256(mapping.concept_list_json().encode("utf-8")).hexdigest()
    rule_set_checksum = hashlib.sha256(
        mapping.DEPENDENCY_GRAPH_PATH.read_bytes()
    ).hexdigest()
    named_rules = sorted(
        {
            concept.derived.rule
            for concept in concepts
            if concept.derived is not None
        }
    )
    return {
        "target_schema": {
            "name": "CAP/ICCR endometrial carcinoma synoptic, projected onto the canonical concept list",
            "concept_count": len(concepts),
            "schema_checksum": schema_checksum,
        },
        "staging_edition": {
            "selected": staging_edition or "as_recorded",
            "per_case_editions": sorted(set(per_case_editions)),
        },
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "rule_versions": {
            "dependency_graph_checksum": rule_set_checksum,
            "named_rules": named_rules,
        },
        "model_versions": {"extraction": llm_extraction.MODEL},
        "coding_systems": {
            "stage": "FIGO; the edition each stage speaks is the per-case staging_edition column",
            "value_sets": (
                "CAP, ICCR, WHO, and AJCC reference standards; each concept cites its registry source "
                "in the column spec's cited_to"
            ),
        },
        "sources": sources,
        "reproduction": (
            "A laboratory holding the same reports and confirming the same values against this concept "
            "list re-derives an identical table. Every absent value carries a coded reason from "
            "docs/prd.md section 4.3, so a missing value is never a bare blank."
        ),
    }


def _provenance(case_barcode: str, checklist) -> list[dict]:
    """One field-level provenance record per checklist field: the value, its confidence and status, the
    coded-absence markers, and the verbatim evidence span (docs/prd.md section 3 case record)."""
    records: list[dict] = []
    for field_name, field in checklist.all_fields().items():
        evidence = field.evidence
        records.append(
            {
                "case_barcode": case_barcode,
                "field_name": field_name,
                "value": _primitive(field.value),
                "confidence": field.confidence,
                "status": field.status.value,
                "applicable": field.applicable,
                "not_applicable_reason": field.not_applicable_reason,
                "evidence_quote": evidence.quote if evidence else None,
                "evidence_char_start": evidence.char_start if evidence else None,
                "evidence_char_end": evidence.char_end if evidence else None,
                "evidence_page": evidence.page_number if evidence else None,
                "evidence_source": evidence.source if evidence else None,
            }
        )
    return records


# The export is the confirmed dataset (#231, decision D1): the API exports only cases a reviewer has
# confirmed (or already exported), so the screen's "Confirmed records" title is truthful. build_export
# still projects any set when `statuses` is None, for internal callers and the projection tests.
EXPORTABLE_STATUSES: tuple[str, ...] = ("confirmed", "exported")


def build_export(
    conn: Connection,
    *,
    staging_edition: Optional[str] = None,
    project: Optional[str] = None,
    statuses: Optional[tuple[str, ...]] = None,
) -> dict:
    """Project every case in the store onto the concept list and assemble the cohort export.

    Returns the column spec, one projected row per case, the field-level provenance, and the reproduction
    bundle. `staging_edition` names the edition the researcher chose at export (docs/prd.md section 4.2);
    the base projection carries the edition each case recorded, and #40 owns re-projecting a stage under a
    chosen edition as a candidate set. `project` scopes the export to one project's cases (issue #149);
    omitted, every case in the store is projected. `statuses`, when given, keeps only cases in those
    statuses; the API passes EXPORTABLE_STATUSES so the served export is the confirmed dataset (#231)."""
    from endopath import storage

    concepts = mapping.compile_concepts()
    coverage = mapping.cap_checklist_coverage()
    labels = _concept_labels()

    rows: list[dict] = []
    provenance: list[dict] = []
    sources: list[dict] = []
    join_sources: list[dict] = []
    per_case_editions: list[str] = []

    # The joined GDC outcome records, grouped by participant, in one read so the per-case outcome projection
    # stays free of the N+1 the cohort export was restructured to avoid (issue #131).
    longitudinal_by_participant = _load_longitudinal(conn)
    # The reviewer join decisions (#180), so a rejected record is excluded from the projection and the
    # confirmed linkages are attributed in the bundle (#217).
    longitudinal_joins_by_case = _load_longitudinal_joins(conn)

    # One batched read of the whole cohort rather than a get_case per barcode (issue #131): a per-case
    # N+1 is ~100s over remote Postgres and 500s the export; this is a few queries. Scoped to the project
    # when one is named (issue #149).
    for case, patient_filename in storage.load_all_cases(conn, project=project):
        if statuses is not None and case.status.value not in statuses:
            continue  # the export is the confirmed dataset (#231): an unreviewed case is not a record
        projected = _project_case(case.checklist, concepts=concepts, coverage=coverage)

        row: dict[str, Any] = {
            "case_barcode": case.case_barcode,
            "report_date": case.report_date.isoformat() if case.report_date else None,
            "staging_edition": case.staging_edition.value,
            "status": case.status.value,
        }
        for concept in concepts:
            item = projected[concept.concept_id]
            row[concept.concept_id] = item.value
            row[f"{concept.concept_id}{ABSENT_REASON_SUFFIX}"] = item.absent_reason

        # Re-project the FIGO stage under the chosen edition (#40). derive_stage returns a determined
        # code, a constrained candidate set, or an indeterminate reason, each rendered distinctly rather
        # than collapsed to one confident stage. The raw quantities stay in their own columns (invariant 1).
        target_edition = staging_edition or case.staging_edition.value
        stage_verdict = projection.project_stage(case.checklist, target_edition)
        stage_value, stage_reason = projection.stage_cell(stage_verdict)
        row["figo_stage"] = stage_value
        row[f"figo_stage{ABSENT_REASON_SUFFIX}"] = stage_reason
        row["figo_stage__projection"] = stage_verdict

        # The join-sourced outcome block: the disease outcome joined from the GDC clinical record by
        # participant barcode, projected onto the outcome columns (`longitudinal.py`). Every cell is a value
        # or a coded absence, never a bare blank; a participant the join did not reach reads determined_by_join.
        participant_id = (
            case.case_barcode[:12] if len(case.case_barcode) >= 12 else case.case_barcode
        )
        outcome_records = longitudinal_by_participant.get(participant_id, [])
        # Drop records a reviewer rejected (#217): a rejection removes the record from the outcome columns
        # and from the bundle's contributing ids, so a linkage the reviewer disowned has no effect. A record
        # with no decision is unreviewed, not rejected, and still projects (see `_rejected_record_ids`).
        case_joins = longitudinal_joins_by_case.get(case.case_barcode, [])
        rejected = _rejected_record_ids(case_joins)
        if rejected:
            outcome_records = [r for r in outcome_records if r.record_id not in rejected]
        outcomes = longitudinal.project_outcomes(outcome_records)
        for key in longitudinal.OUTCOME_COLUMN_KEYS:
            cell = outcomes[key]
            row[key] = cell.value
            row[f"{key}{ABSENT_REASON_SUFFIX}"] = cell.absent_reason
        source_entry = {
            "case_barcode": case.case_barcode,
            "participant_id": participant_id,
            "record_ids": longitudinal.contributing_record_ids(outcome_records),
        }
        # The reviewer-attributed provenance behind the join-sourced columns (#176, #217): who decided each
        # linkage, in what role, with what disposition. Present only when the case carries a decision.
        if case_joins:
            source_entry["linkage_decisions"] = case_joins
        join_sources.append(source_entry)
        rows.append(row)

        provenance.extend(_provenance(case.case_barcode, case.checklist))
        sources.append(
            {
                "case_barcode": case.case_barcode,
                "source_document": patient_filename,
                "evidence_checksum": _evidence_checksum(case.checklist),
            }
        )
        per_case_editions.append(case.staging_edition.value)

    from endopath import provenance as provenance_mod

    bundle = _bundle(
        concepts,
        sources,
        staging_edition=staging_edition,
        per_case_editions=per_case_editions,
    )
    # The per-call LLM ledger (#116): how many model calls the cohort cost, in tokens and mean latency,
    # so a second centre sees what produced the values, not only the values.
    bundle["llm_calls"] = provenance_mod.summary(conn)
    # The join provenance for the outcome columns (invariant 2 for a join-sourced value): the source of the
    # join, the outcome columns it fills, the named derivation rule, the survival gap this pull carries, and
    # the GDC record ids that fed each case's outcomes.
    bundle["join_sources"] = {
        "source": longitudinal.JOIN_SOURCE,
        "outcome_columns": longitudinal.OUTCOME_COLUMN_KEYS,
        "derivation_rules": [longitudinal.RECURRENCE_RULE, longitudinal.OVERALL_SURVIVAL_RULE],
        "data_quality": (
            "therapeutic_agents reads Tamoxifen for nearly every case. This is faithful to the GDC/TCGA "
            "harmonized clinical record, a known source artifact, not an error in this pipeline; it is "
            "surfaced with its provenance rather than silently dropped. The received_chemotherapy, "
            "received_radiation, and received_hormone_therapy flags read administration, not record "
            'presence: GDC emits a treatment_<type> template record marked treatment_or_therapy = "no" on '
            "cases where the therapy was considered but not given (most often Radiation Therapy, NOS), so a "
            'flag is True only where a matching record is marked "yes", False on an explicit "no", and '
            'not_stated where the matching records carry only "unknown". Reading presence instead reports '
            "radiation on 543/546 cases (99%); administration reports 264 (48%)."
        ),
        "per_case": join_sources,
    }
    return {
        "columns": _columns(concepts, labels),
        "rows": rows,
        "provenance": provenance,
        "bundle": bundle,
    }


# --- rendering the three files from one projection ------------------------------------------------


def _cell(value: Any) -> Any:
    """A projected value rendered for a spreadsheet cell: None to an empty string, a bool kept as a bool,
    everything else as-is. An absent value's cell is empty; its reason rides the parallel column."""
    if value is None:
        return ""
    return value


def to_csv(export: dict) -> str:
    """The projected table as CSV: the column keys as the header, one row per case, an empty cell for an
    absent value with its reason in the parallel column."""
    keys = [column["key"] for column in export["columns"]]
    buffer = io.StringIO()
    writer = csv.DictWriter(buffer, fieldnames=keys, extrasaction="ignore")
    writer.writeheader()
    for row in export["rows"]:
        writer.writerow({key: _cell(row.get(key)) for key in keys})
    return buffer.getvalue()


def to_json(export: dict) -> str:
    """The whole export as self-describing JSON: the bundle, the column spec, the rows, and the field-level
    provenance, so the file alone re-derives the table."""
    return json.dumps(
        {
            "bundle": export["bundle"],
            "columns": export["columns"],
            "rows": export["rows"],
            "provenance": export["provenance"],
        },
        indent=2,
        default=str,
    )


def to_xlsx_bytes(export: dict) -> bytes:
    """The projected table as an XLSX workbook: a `cases` sheet of the projected rows, a `provenance`
    sheet of the field-level evidence, and a `bundle` sheet of the reproduction metadata as key/value
    rows."""
    from openpyxl import Workbook

    workbook = Workbook()

    cases = workbook.active
    cases.title = "cases"
    keys = [column["key"] for column in export["columns"]]
    cases.append([column["label"] for column in export["columns"]])
    for row in export["rows"]:
        cases.append([_cell(row.get(key)) for key in keys])

    provenance_sheet = workbook.create_sheet("provenance")
    provenance_headers = [
        "case_barcode",
        "field_name",
        "value",
        "confidence",
        "status",
        "applicable",
        "not_applicable_reason",
        "evidence_quote",
        "evidence_page",
        "evidence_source",
    ]
    provenance_sheet.append(provenance_headers)
    for record in export["provenance"]:
        provenance_sheet.append([_cell(record.get(header)) for header in provenance_headers])

    bundle_sheet = workbook.create_sheet("bundle")
    bundle_sheet.append(["key", "value"])
    for key, value in _flatten(export["bundle"]):
        bundle_sheet.append([key, value])

    buffer = io.BytesIO()
    workbook.save(buffer)
    return buffer.getvalue()


def _flatten(value: Any, prefix: str = "") -> list[tuple[str, str]]:
    """Flatten the nested bundle into dotted key/value rows for the XLSX bundle sheet."""
    rows: list[tuple[str, str]] = []
    if isinstance(value, dict):
        for key, inner in value.items():
            rows.extend(_flatten(inner, f"{prefix}.{key}" if prefix else str(key)))
    elif isinstance(value, list):
        if all(not isinstance(item, (dict, list)) for item in value):
            rows.append((prefix, ", ".join(str(item) for item in value)))
        else:
            for index, item in enumerate(value):
                rows.extend(_flatten(item, f"{prefix}[{index}]"))
    else:
        rows.append((prefix, "" if value is None else str(value)))
    return rows


def render_files(export: dict) -> dict:
    """The three download files from one projection: CSV and JSON as text, XLSX as base64, each with its
    filename and media type. One projection feeds the preview and all three files, so they never diverge."""
    return {
        "csv": {
            "filename": "endometrial_cohort.csv",
            "media_type": "text/csv",
            "content": to_csv(export),
        },
        "json": {
            "filename": "endometrial_cohort.json",
            "media_type": "application/json",
            "content": to_json(export),
        },
        "xlsx": {
            "filename": "endometrial_cohort.xlsx",
            "media_type": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            "content_base64": base64.b64encode(to_xlsx_bytes(export)).decode("ascii"),
        },
    }