Signal/python-backend/core/dedup.py

"""
dedup.py
Signal — STTIL Solutions

Groups normalized shipment rows into deduped shipments and gradeable coverage
lines. This is the structural foundation of the readiness model: it turns a
flat list of CSV-derived rows into the patient -> device -> shipment hierarchy
the worklist needs, BEFORE any grading happens.

Design: docs/dedup-design-2026-06-24.md (Pi, locked by Kisa 2026-06-24)
        docs/readiness-model-brief-2026-06-23.md sections 3, 5, 7

LOCKED DEDUP KEY: (patient_id, device_type, date_of_service)
    order_number is a displayed attribute, NEVER part of the dedup key.
    Monthly resupplies are separate dates of service, NOT duplicates.

PLAN TYPE: carried through verbatim from the client-mapped field. It is NEVER
    derived from the payer name. None means "plan type needed" -> keyed as
    'unknown' so the line still forms but cannot be graded green later.

PHI CONTRACT: operates only on patient_id, device_type, shipment_date,
    quantity, payer (display), plan_type, component, order_number, hcpcs, and
    the csv_* doc-status fields. No names, SSNs, DOBs, or contact fields.
"""

from __future__ import annotations

import hashlib
import logging
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import date, datetime
from typing import Optional

from core.coverage_calculator import ShipmentRecord, CoverageFlag

logger = logging.getLogger(__name__)


def _hash_pid(patient_id: str) -> str:
    """Short hashed patient_id for log lines — raw patient_id is never logged."""
    return hashlib.sha256(patient_id.encode()).hexdigest()[:12]


def _as_date(value) -> date:
    """Coerce a datetime to a pure date so same-day timestamps group together."""
    return value.date() if isinstance(value, datetime) else value

# CGM device types graded this phase. Pumps and other DMEPOS categories are
# preserved and displayed but not graded (gradeable=False). Mirrors the CGM
# devices in payer_rules.json; moves into addressable config in a later step.
CGM_DEVICE_TYPES = frozenset(
    {"dexcom_g6", "dexcom_g7", "freestyle_libre_2", "freestyle_libre_3"}
)

UNKNOWN_PLAN_TYPE = "unknown"


def is_cgm_device(device_type: str) -> bool:
    """True only for CGM device types graded in the current phase."""
    return device_type in CGM_DEVICE_TYPES


@dataclass
class MergedShipment:
    """One deduped shipment: all rows sharing (patient_id, device_type, DOS)."""
    patient_id: str
    device_type: str
    shipment_date: date
    total_quantity: int
    components: list[str]                 # combined, deduped, first-seen order
    order_numbers: list[str]              # collected, deduped — display only
    hcpcs_codes: list[str]                # collected, deduped — display only
    payer: str                           # raw display NAME, latest non-null
    plan_type: Optional[str]             # client-mapped; NEVER from payer name
    # csv_* doc fields — latest non-null wins across the group
    csv_visit_date: Optional[date] = None
    csv_swo_status: Optional[str] = None
    csv_pecos_verified: Optional[str] = None
    csv_pa_status: Optional[str] = None
    csv_diagnosis_on_file: Optional[str] = None
    csv_transfer_from: Optional[str] = None


@dataclass(frozen=True)
class CoverageLineKey:
    patient_id: str
    device_type: str
    plan_type_id: str


@dataclass
class CoverageLine:
    """A gradeable unit: one patient's one device under one plan type.

    line_status and timing_flag are computed later by the grading step
    (gradeLine); a freshly assigned line is intentionally ungraded.
    """
    patient_id: str
    device_type: str
    plan_type: str
    gradeable: bool
    current_shipment: MergedShipment
    prior_shipments: list[MergedShipment] = field(default_factory=list)
    line_status: Optional[str] = None
    timing_flag: Optional[CoverageFlag] = None


def _qty(record: ShipmentRecord) -> int:
    """A row without a quantity is still a shipment -> default to 1."""
    return 1 if record.quantity is None else record.quantity


def _merge_quantities(rows: list[ShipmentRecord]) -> int:
    """Sum quantities across DISTINCT component configs, max within a config.

    Within one component config, identical rows are likely data-entry
    duplicates, so take the MAX (guards against doubling). Across distinct
    configs (a real component split, e.g. sensor + transmitter), SUM.

    A disagreement within one config (e.g. qty 3 and qty 100 for the same
    component) collapses to the max and leaves a warning, so silently dropping
    quantity always leaves an audit breadcrumb.
    """
    by_config: dict[frozenset, list[int]] = defaultdict(list)
    for row in rows:
        by_config[frozenset({row.component})].append(_qty(row))

    total = 0
    for config, qtys in by_config.items():
        if len(set(qtys)) > 1:
            logger.warning(
                "Quantity disagreement in dedup group patient=%s device=%s "
                "component=%s qtys=%s -> taking max",
                _hash_pid(rows[0].patient_id), rows[0].device_type,
                set(config), qtys,
            )
        total += max(qtys)
    return total


def _merge_plan_type(rows: list[ShipmentRecord]) -> Optional[str]:
    """Plan type is grading-critical and must NEVER be guessed.

    If a dedup group carries more than one distinct non-null plan type, the data
    is ambiguous, so return None ('plan type needed') rather than pick one.
    """
    distinct = {r.plan_type for r in rows if r.plan_type}
    if len(distinct) > 1:
        logger.warning(
            "Conflicting plan_type in dedup group patient=%s device=%s: %s "
            "-> routing to 'plan type needed'",
            _hash_pid(rows[0].patient_id), rows[0].device_type, sorted(distinct),
        )
        return None
    return next(iter(distinct)) if distinct else None


def _dedup_preserve_order(values: list[Optional[str]]) -> list[str]:
    """Collect non-empty values, dedup, preserve first-seen order."""
    seen: dict[str, None] = {}
    for v in values:
        if v:
            seen.setdefault(v, None)
    return list(seen.keys())


def _latest_non_null(rows: list[ShipmentRecord], attr: str):
    """Last meaningful value for an attribute, in ingestion order.

    Empty / whitespace-only strings are treated as null so a blank later row
    never clobbers an earlier real value (does not rely on upstream nulling).
    """
    result = None
    for row in rows:
        value = getattr(row, attr)
        if value is None:
            continue
        if isinstance(value, str) and value.strip() == "":
            continue
        result = value
    return result


def dedup(records: list[ShipmentRecord]) -> list[MergedShipment]:
    """Group rows by (patient_id, device_type, shipment_date) into MergedShipments."""
    groups: dict[tuple, list[ShipmentRecord]] = {}
    for record in records:
        key = (record.patient_id, record.device_type, _as_date(record.shipment_date))
        groups.setdefault(key, []).append(record)

    merged: list[MergedShipment] = []
    for (patient_id, device_type, shipment_date), rows in groups.items():
        merged.append(
            MergedShipment(
                patient_id=patient_id,
                device_type=device_type,
                shipment_date=shipment_date,
                total_quantity=_merge_quantities(rows),
                components=_dedup_preserve_order([r.component for r in rows]),
                order_numbers=_dedup_preserve_order([r.order_number for r in rows]),
                hcpcs_codes=_dedup_preserve_order([r.hcpcs for r in rows]),
                payer=_latest_non_null(rows, "payer") or "",
                plan_type=_merge_plan_type(rows),
                csv_visit_date=_latest_non_null(rows, "csv_visit_date"),
                csv_swo_status=_latest_non_null(rows, "csv_swo_status"),
                csv_pecos_verified=_latest_non_null(rows, "csv_pecos_verified"),
                csv_pa_status=_latest_non_null(rows, "csv_pa_status"),
                csv_diagnosis_on_file=_latest_non_null(rows, "csv_diagnosis_on_file"),
                csv_transfer_from=_latest_non_null(rows, "csv_transfer_from"),
            )
        )

    merged.sort(key=lambda m: (m.patient_id, m.device_type, m.shipment_date))
    return merged


def assign_to_coverage_lines(
    merged_shipments: list[MergedShipment],
) -> dict[CoverageLineKey, CoverageLine]:
    """Group MergedShipments into CoverageLines by (patient_id, device_type, plan_type).

    plan_type is the client-mapped field. None is keyed as 'unknown' (plan type
    needed) so the line still forms but cannot be graded green later. The latest
    DOS in each line is the current shipment; earlier ones are collapsed history.
    """
    by_key: dict[CoverageLineKey, list[MergedShipment]] = {}
    for shipment in merged_shipments:
        key = CoverageLineKey(
            patient_id=shipment.patient_id,
            device_type=shipment.device_type,
            plan_type_id=shipment.plan_type or UNKNOWN_PLAN_TYPE,
        )
        by_key.setdefault(key, []).append(shipment)

    lines: dict[CoverageLineKey, CoverageLine] = {}
    for key, shipments in by_key.items():
        shipments.sort(key=lambda s: s.shipment_date, reverse=True)
        lines[key] = CoverageLine(
            patient_id=key.patient_id,
            device_type=key.device_type,
            plan_type=key.plan_type_id,
            gradeable=is_cgm_device(key.device_type),
            current_shipment=shipments[0],
            prior_shipments=shipments[1:],
            line_status=None,
            timing_flag=None,
        )

    return lines