check_fix_helpers.py

"""Helpers for attaching structured fix proposals to QC warnings."""

from __future__ import annotations

from typing import Any, Iterable

from duo_terms import get_duo_term_metadata, normalize_duo_term_ids
from fact_descriptor_sync import (
    build_collection_descriptor_proposal,
    parse_collection_multi_value_field,
)
from fix_proposals import make_fix_proposal
from k_anonymity import donor_value_violates_k


MULTI_VALUE_COLLECTION_FIELDS = {
    "data_use",
    "type",
    "diagnosis_available",
    "materials",
    "sex",
}


def current_collection_field_value(collection: dict[str, Any], field: str) -> Any:
    """Return a normalized collection field value for fix export."""
    if field in MULTI_VALUE_COLLECTION_FIELDS:
        values = parse_collection_multi_value_field(collection.get(field))
        if field == "data_use":
            return normalize_duo_term_ids(values)
        return values
    value = collection.get(field)
    if value in (None, ""):
        return None
    return value


def make_collection_term_append_fix(
    *,
    update_id: str,
    module: str,
    collection: dict[str, Any],
    field: str,
    term_id: str,
    confidence: str,
    human_explanation: str,
    rationale: str = "",
    replace_required: bool = False,
    exclusive_group: str = "",
    blocking_reason: str = "",
):
    """Return a collection append fix for one ontology/code term."""
    current_value = current_collection_field_value(collection, field)
    return make_fix_proposal(
        update_id=update_id,
        module=module,
        entity_type="COLLECTION",
        entity_id=collection["id"],
        field=field,
        mode="append",
        confidence=confidence,
        current_value_at_export=current_value,
        proposed_value=[term_id],
        human_explanation=human_explanation,
        rationale=rationale,
        term_explanations=[get_duo_term_metadata(term_id)] if term_id.startswith("DUO:") else [],
        replace_required=replace_required,
        blocking_reason=blocking_reason,
        exclusive_group=exclusive_group,
    )


def make_collection_scalar_set_fix(
    *,
    update_id: str,
    module: str,
    collection: dict[str, Any],
    field: str,
    proposed_value: Any,
    confidence: str,
    human_explanation: str,
    rationale: str = "",
    blocking_reason: str = "",
):
    """Return a collection scalar set fix."""
    return make_fix_proposal(
        update_id=update_id,
        module=module,
        entity_type="COLLECTION",
        entity_id=collection["id"],
        field=field,
        mode="set",
        confidence=confidence,
        current_value_at_export=current_collection_field_value(collection, field),
        proposed_value=proposed_value,
        human_explanation=human_explanation,
        rationale=rationale,
        blocking_reason=blocking_reason,
    )


def make_collection_multi_value_fix(
    *,
    update_id: str,
    module: str,
    collection: dict[str, Any],
    field: str,
    proposed_values: Iterable[str],
    confidence: str,
    human_explanation: str,
    rationale: str = "",
    mode: str = "append",
    replace_required: bool = False,
    blocking_reason: str = "",
    exclusive_group: str = "",
):
    """Return a collection multi-value fix."""
    proposed = [value for value in proposed_values if value]
    return make_fix_proposal(
        update_id=update_id,
        module=module,
        entity_type="COLLECTION",
        entity_id=collection["id"],
        field=field,
        mode=mode,
        confidence=confidence,
        current_value_at_export=current_collection_field_value(collection, field),
        proposed_value=proposed,
        human_explanation=human_explanation,
        rationale=rationale,
        replace_required=replace_required,
        blocking_reason=blocking_reason,
        exclusive_group=exclusive_group,
    )


def build_fact_alignment_fix_proposals(collection: dict[str, Any], facts: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Return collection descriptor fixes derived conservatively from fact sheets."""
    proposal = build_collection_descriptor_proposal(collection, facts, replace_existing=False)
    fix_proposals = []
    change_map = {change["field"]: change for change in proposal["changes"]}
    field_notes = proposal.get("field_notes", {})

    for field, update_family in (
        ("diagnosis_available", "diagnoses"),
        ("materials", "materials"),
        ("sex", "clinical_profile"),
        ("age_low", "age"),
        ("age_high", "age"),
        ("age_unit", "age"),
        ("size", "counts"),
        ("number_of_donors", "counts"),
    ):
        change = change_map.get(field)
        if change is None:
            continue
        note_text = " ".join(field_notes.get(field, [])).strip()
        confidence = "certain" if update_family in {"diagnoses", "materials", "counts"} else "almost_certain"
        if update_family == "age" and note_text:
            confidence = "uncertain"
        if update_family == "counts":
            base_rationale = (
                "Counts proposal uses conservative normalization: totals are taken only from a single all-star aggregate fact row (sex='*', age_range='*', sample_type='*', disease='*'). "
                "Here, '*' means 'aggregated over all values', so this all-star row is the authoritative total. "
                "No count update is proposed when that all-star row is missing/duplicated or not numeric."
            )
        elif update_family == "age":
            base_rationale = (
                "Age proposal uses conservative normalization: Directory age labels/ranges are mapped to numeric bounds. "
                "For descriptor derivation, '*' and Unknown/Undefined age labels are treated as aggregated/unspecified buckets (not concrete age groups). "
                "Automatic updates only widen coverage unless explicit replace mode is used."
            )
        elif update_family == "materials":
            base_rationale = (
                "Materials proposal uses conservative normalization: for descriptor derivation, '*' is treated as an aggregated/unspecified sample_type bucket, not a concrete material value. "
                "NAV is treated as non-specific unless it is the only material signal."
            )
        elif update_family == "diagnoses":
            base_rationale = (
                "Diagnoses proposal uses conservative normalization: for descriptor derivation, '*' is treated as an aggregated/unspecified disease bucket, not a concrete diagnosis code. "
                "Existing broader ICD-10 metadata codes are preserved when they already cover more specific fact-sheet codes."
            )
        elif update_family == "clinical_profile":
            base_rationale = (
                "Sex proposal uses conservative normalization: for descriptor derivation, '*' is treated as an aggregated/unspecified sex bucket, not a concrete sex value. "
                "Only explicit fact-sheet sex values are compared."
            )
        else:
            base_rationale = (
                "Fact-sheet proposal uses conservative normalization: for descriptor derivation, '*' is treated as aggregated/unspecified rather than as a concrete category value."
            )
        fix_proposals.append(
            make_fix_proposal(
                update_id=f"{update_family}.{field}.from_facts",
                module="FT",
                entity_type="COLLECTION",
                entity_id=collection["id"],
                field=field,
                mode="append" if field in {"diagnosis_available", "materials", "sex"} else "set",
                confidence=confidence,
                current_value_at_export=change["current"],
                proposed_value=change["proposed"],
                human_explanation=(
                    f"Update collection field {field} from fact-sheet derived values."
                ),
                rationale=(
                    base_rationale
                    + (f" {note_text}" if note_text else "")
                ),
                blocking_reason=(
                    note_text if confidence == "uncertain" else ""
                ),
            )
        )
    return fix_proposals


def build_fact_k_anonymity_drop_fixes(
    collection: dict[str, Any],
    facts: list[dict[str, Any]],
    *,
    k_limit: int,
) -> list[dict[str, Any]]:
    """Return a fix proposal that drops fact-sheet rows violating donor k-anonymity."""
    violating_ids = []
    for fact in facts:
        if donor_value_violates_k(fact.get("number_of_donors"), k_limit) and fact.get("id"):
            violating_ids.append(str(fact["id"]))
    if not violating_ids:
        return []
    violating_ids = sorted(dict.fromkeys(violating_ids))
    return [
        make_fix_proposal(
            update_id=f"facts.k_anonymity.drop_rows_k{k_limit}",
            module="FT",
            entity_type="COLLECTION",
            entity_id=collection["id"],
            field="facts",
            mode="delete_rows",
            confidence="certain",
            current_value_at_export=violating_ids,
            proposed_value=violating_ids,
            human_explanation=(
                f"Delete fact-sheet rows that violate donor k-anonymity (number_of_donors < {k_limit})."
            ),
            rationale=(
                "This update drops only fact rows whose explicit donor count is greater than 0 and below the "
                f"k-anonymity threshold k={k_limit}. Rows with 0/empty donor values are not auto-dropped."
            ),
        )
    ]