Recipe 5.2: Python Implementation Example

pip install boto3 jellyfish requests python-dateutil

import hashlib
import json
import logging
import math
import re
import unicodedata
import uuid
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from typing import Optional

import boto3
import jellyfish
import requests
from boto3.dynamodb.conditions import Key
from botocore.config import Config
from dateutil import parser as dateparser

# Structured logging. In production, ship JSON-formatted records to
# CloudWatch Logs Insights. Provider data is less sensitive than
# patient data but license numbers and personal addresses still
# warrant care; log structural metadata only (record IDs, scores,
# routing decisions, NPIs as identifiers), never raw demographic
# fields.
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Adaptive retry handles throttling from DynamoDB, EventBridge,
# CloudWatch, and the NPI Registry API. Real-time onboarding has
# a tight latency budget; transient throttling should not fail the
# whole NPI assignment.
BOTO3_RETRY_CONFIG = Config(retries={"max_attempts": 5, "mode": "adaptive"})

# Module-level clients. Reused across Lambda invocations in warm
# containers so each invocation does not pay the connection cost.
REGION = "us-east-1"
dynamodb = boto3.resource("dynamodb", region_name=REGION, config=BOTO3_RETRY_CONFIG)
s3_client = boto3.client("s3", region_name=REGION, config=BOTO3_RETRY_CONFIG)
eventbridge_client = boto3.client("events", region_name=REGION, config=BOTO3_RETRY_CONFIG)
cloudwatch_client = boto3.client("cloudwatch", region_name=REGION, config=BOTO3_RETRY_CONFIG)

# --- Resource Names ---
# Fill these in with your actual resource names. The demo prints
# what it would write rather than failing if the resources do not
# exist; see the run_demo() function at the bottom.
ASSIGNMENT_TABLE      = "provider-npi-assignment"
SCHEDULE_TABLE        = "verification-schedule"
REVIEW_QUEUE_TABLE    = "provider-review-queue"
AUDIT_BUCKET          = "my-provider-matching-audit"
EVENTS_BUS_NAME       = "provider-npi-events"
CLOUDWATCH_NAMESPACE  = "Provider/NPIMatching"

# Deploy-time guardrail. Validates every resource-name constant so a
# missing value produces an actionable assertion message rather than a
# downstream boto3 ValidationException.
assert ASSIGNMENT_TABLE != "", "ASSIGNMENT_TABLE must be set before deploying."
assert SCHEDULE_TABLE != "", "SCHEDULE_TABLE must be set before deploying."
assert REVIEW_QUEUE_TABLE != "", "REVIEW_QUEUE_TABLE must be set before deploying."
assert AUDIT_BUCKET != "", "AUDIT_BUCKET must be set before deploying."
assert EVENTS_BUS_NAME != "", "EVENTS_BUS_NAME must be set before deploying."
assert CLOUDWATCH_NAMESPACE != "", "CLOUDWATCH_NAMESPACE must be set before deploying."

# --- NPI Registry API ---
# The public NPI Registry API is published by CMS at this base URL.
# It is unauthenticated, free to use, and rate-limited. For batch
# workloads, do not iterate API calls; use the monthly Downloadable
# File instead. 
NPI_REGISTRY_BASE_URL = "https://npiregistry.cms.hhs.gov/api/"
NPI_REGISTRY_API_VERSION = "2.1"
NPI_REGISTRY_TIMEOUT_SECONDS = 5
NPI_REGISTRY_MAX_RESULTS_PER_QUERY = 200
# The public NPI Registry API supports `limit` up to 200 per query.
# Using the maximum reduces the chance of silently truncating
# common-surname searches. For queries that hit the cap, consider
# adding `skip`-based pagination in `_npi_registry_lookup`.

# --- Versioning ---
NORMALIZER_VERSION = "norm-provider-v1.0"
MODEL_VERSION      = "fs-provider-v1.0"

# --- Routing Thresholds and Margin ---
# The composite score is a Fellegi-Sunter log-likelihood ratio
# (sum of per-field log(m/u) and log((1-m)/(1-u)) terms). Higher
# is more confident match. The margin requirement is what catches
# the "two Sarah Patels in California" confounder that pure
# absolute thresholds miss: a top score that barely beats the
# runner-up should not auto-attach even if it clears the high
# threshold. Tune all three against a labeled gold set.
HIGH_THRESHOLD = Decimal("8.0")    # at or above: candidate for auto-attach
LOW_THRESHOLD  = Decimal("-2.0")   # at or below: auto-non-match
MIN_MARGIN     = Decimal("3.0")    # required gap between top and runner-up

# --- Re-Verification Cadence ---
# Network adequacy regulations commonly require verification every
# 90 days; the architecture should support per-segment cadences in
# production (Medicare Advantage stricter, Medicaid varying by
# state). The demo uses a single cadence for clarity.
VERIFICATION_CADENCE_DAYS = 90

# --- Candidate Cap ---
# Internal records that produce more than this many candidates
# (very common surnames in dense states with no license-number
# field) route directly to review rather than auto-deciding. The
# cap protects against false-positive auto-attaches in low-
# information cases.
MAX_CANDIDATES_BEFORE_REVIEW = 50

# --- Fellegi-Sunter Probabilities ---
#
# m_probabilities[field][comparison_level] = P(observe this comparison
#   level | the internal record and the registry record are about the
#   same provider).
# u_probabilities[field][comparison_level] = P(observe this comparison
#   level | the records are about different providers).
#
# Provider data is cleaner than patient data and the registry is
# authoritative, so the m values for stable fields (license number,
# legal name) are higher and the u values are lower than the
# corresponding patient-matching priors. License-number-plus-state
# match is the strongest signal in the pipeline.
M_PROBABILITIES = {
    "first_name": {
        "exact":             Decimal("0.90"),
        "jaro_winkler_high": Decimal("0.06"),
        "metaphone_match":   Decimal("0.02"),
        "other_name_match":  Decimal("0.01"),  # via NPPES other-names field
        "mismatch":          Decimal("0.01"),
    },
    "last_name": {
        "exact":             Decimal("0.85"),
        "damerau_high":      Decimal("0.06"),
        "metaphone_match":   Decimal("0.04"),
        "other_name_match":  Decimal("0.04"),  # legal name change pattern
        "mismatch":          Decimal("0.01"),
    },
    "credential": {
        "exact_set_match":   Decimal("0.70"),
        "subset_match":      Decimal("0.25"),  # internal subset of registry
        "mismatch":          Decimal("0.04"),
        "one_null":          Decimal("0.01"),
    },
    "license": {
        "exact_number_and_state": Decimal("0.95"),
        "number_only_match":      Decimal("0.02"),
        "state_only_match":       Decimal("0.01"),
        "mismatch":               Decimal("0.01"),
        "one_null":               Decimal("0.01"),
    },
    "taxonomy": {
        "primary_match":     Decimal("0.65"),
        "any_match":         Decimal("0.20"),
        "parent_match":      Decimal("0.10"),  # NUCC parent taxonomy
        "mismatch":          Decimal("0.04"),
        "internal_unknown":  Decimal("0.01"),
    },
    "address": {
        "exact":             Decimal("0.40"),
        "same_zip":          Decimal("0.30"),
        "same_state":        Decimal("0.20"),
        "mismatch":          Decimal("0.05"),
        "one_null":          Decimal("0.05"),
    },
    "phone": {
        "exact":             Decimal("0.35"),
        "last_4_match":      Decimal("0.15"),
        "mismatch":          Decimal("0.30"),
        "one_null":          Decimal("0.20"),
    },
}

U_PROBABILITIES = {
    "first_name": {
        "exact":             Decimal("0.005"),
        "jaro_winkler_high": Decimal("0.02"),
        "metaphone_match":   Decimal("0.05"),
        "other_name_match":  Decimal("0.005"),
        "mismatch":          Decimal("0.92"),
    },
    "last_name": {
        "exact":             Decimal("0.002"),
        "damerau_high":      Decimal("0.005"),
        "metaphone_match":   Decimal("0.01"),
        "other_name_match":  Decimal("0.003"),
        "mismatch":          Decimal("0.98"),
    },
    "credential": {
        "exact_set_match":   Decimal("0.10"),
        "subset_match":      Decimal("0.10"),
        "mismatch":          Decimal("0.78"),
        "one_null":          Decimal("0.02"),
    },
    "license": {
        "exact_number_and_state": Decimal("0.00001"),  # essentially unique
        "number_only_match":      Decimal("0.0001"),
        "state_only_match":       Decimal("0.05"),
        "mismatch":               Decimal("0.94989"),
        "one_null":               Decimal("0.001"),
    },
    "taxonomy": {
        "primary_match":     Decimal("0.05"),
        "any_match":         Decimal("0.10"),
        "parent_match":      Decimal("0.05"),
        "mismatch":          Decimal("0.78"),
        "internal_unknown":  Decimal("0.02"),
    },
    "address": {
        "exact":             Decimal("0.001"),
        "same_zip":          Decimal("0.05"),
        "same_state":        Decimal("0.10"),
        "mismatch":          Decimal("0.83"),
        "one_null":          Decimal("0.019"),
    },
    "phone": {
        "exact":             Decimal("0.0005"),
        "last_4_match":      Decimal("0.005"),
        "mismatch":          Decimal("0.90"),
        "one_null":          Decimal("0.0945"),
    },
}

def _to_decimal(value) -> Decimal:
    """Coerce numeric input into Decimal for DynamoDB."""
    if isinstance(value, Decimal):
        return value
    return Decimal(str(value))

def _now_iso() -> str:
    """UTC timestamp in ISO 8601 format. Always UTC; never local time."""
    return datetime.now(timezone.utc).isoformat()

SUFFIX_CANONICAL = {
    "jr": "jr", "jr.": "jr", "junior": "jr",
    "sr": "sr", "sr.": "sr", "senior": "sr",
    "ii": "ii", "iii": "iii", "iv": "iv",
}

# A small, illustrative map from common internal specialty values
# to NUCC taxonomy codes. Real systems maintain a much larger and
# institution-specific map, versioned and code-managed. The
# `unknown_taxonomy` sentinel marks specialties that cannot be
# mapped; downstream comparators handle it as a non-information.
INTERNAL_SPECIALTY_TO_NUCC = {
    "family medicine":      "207Q00000X",
    "internal medicine":    "207R00000X",
    "pediatrics":           "208000000X",
    "cardiology":           "207RC0000X",
    "psychiatry":           "2084P0800X",
    "general surgery":      "208600000X",
    "obstetrics gynecology": "207V00000X",
    "emergency medicine":   "207P00000X",
    "anesthesiology":       "207L00000X",
    "radiology":            "2085R0202X",
    "dermatology":          "207N00000X",
    "neurology":            "2084N0400X",
    "oncology":             "207RX0202X",
    "orthopedics":          "207X00000X",
    "primary care":         "207Q00000X",  # commonly used as a synonym
}

# Reverse map: NUCC code -> description string for the NPI Registry API.
# The API accepts a `taxonomy_description` parameter, not a code.
NUCC_CODE_TO_DESCRIPTION = {v: k.title() for k, v in INTERNAL_SPECIALTY_TO_NUCC.items()
                            if k != "primary care"}  # avoid duplicates

def _strip_diacritics(s: str) -> str:
    """Strip combining diacritical marks for case-insensitive matching."""
    if not s:
        return ""
    nfkd = unicodedata.normalize("NFKD", s)
    return "".join(c for c in nfkd if not unicodedata.combining(c))

def _normalize_name(raw: Optional[str]) -> str:
    """Trim, case-fold, strip diacritics, collapse whitespace; preserve hyphens."""
    if not raw:
        return ""
    s = _strip_diacritics(raw)
    s = s.lower().strip()
    s = re.sub(r"[^a-z\- ]", " ", s)
    s = re.sub(r"\s+", " ", s).strip()
    return s

def _normalize_organization_name(raw: Optional[str]) -> str:
    """
    Org names need slightly different handling than person names:
    keep digits (some clinics include them), keep periods (St., Inc.),
    but uppercase and collapse whitespace.
    """
    if not raw:
        return ""
    s = _strip_diacritics(raw).upper().strip()
    s = re.sub(r"\s+", " ", s)
    return s

def _normalize_suffix(raw: Optional[str]) -> str:
    if not raw:
        return ""
    return SUFFIX_CANONICAL.get(raw.lower().strip(), "")

def _parse_credential_string(raw: Optional[str]) -> list:
    """
    NPPES stores credentials as a free-text field with a wide variety
    of separators ("MD", "MD, MPH", "DO,FACP", "MD/PHD"). Parse into
    an ordered, deduplicated list of normalized credential codes.
    Returns an empty list if the input is missing or unparseable.
    """
    if not raw:
        return []
    parts = re.split(r"[,/;|]+|\s+", raw.upper())
    cleaned = []
    for p in parts:
        p = re.sub(r"[^A-Z]", "", p)  # strip dots, dashes, etc.
        if p and p not in cleaned:
            cleaned.append(p)
    return cleaned

def _double_metaphone(s: str) -> str:
    """
    Phonetic encoding for blocking and as a comparator level.
    Uses jellyfish.metaphone (original metaphone, not double metaphone).
    For production, adopt the `metaphone` PyPI package which provides
    true double metaphone with secondary-code matching (relevant for
    equity: names from non-English naming conventions often produce
    useful secondary codes that original metaphone discards).
    """
    if not s:
        return ""
    return jellyfish.metaphone(s) or ""

def _normalize_phone(raw: Optional[str]) -> str:
    """Strip non-digit characters; drop leading 1 to produce a 10-digit number."""
    if not raw:
        return ""
    digits = re.sub(r"\D", "", raw)
    if len(digits) == 11 and digits.startswith("1"):
        digits = digits[1:]
    return digits

def _normalize_license_number(raw: Optional[str]) -> str:
    """
    Strip whitespace and normalize case. Some sources include state
    prefixes ("CA-12345", "TX 67890"); strip them so the comparator
    sees the bare number. License-state is tracked separately.
    """
    if not raw:
        return ""
    s = raw.upper().strip()
    s = re.sub(r"\s+", "", s)
    return s


def _license_in_candidate(candidate: dict, target_license: str) -> bool:
    """
    Check whether a candidate's NPPES record contains a matching
    license number. The NPPES API response nests license info inside
    the taxonomies array; this helper checks both the structured
    taxonomy entries and any top-level license fields.
    """
    for tax in candidate.get("taxonomies", []):
        lic = (tax.get("license", "") or "").upper().strip()
        if lic and lic == target_license:
            return True
    return False

def _normalize_address(raw: dict) -> dict:
    """
    Coarse address normalization. A real system pipes this through
    a CASS-certified product (SmartyStreets, Melissa, the USPS API)
    that returns a canonical USPS form, deliverability flag, and
    ZIP+4. The block below handles the obvious cases.
    """
    if not raw:
        return {"line1": "", "line2": "", "city": "", "state": "", "zip": "",
                "zip5": "", "canonical": ""}

    def _u(v):
        return (v or "").upper().strip()

    line1 = _u(raw.get("line1"))
    line2 = _u(raw.get("line2"))
    city  = _u(raw.get("city"))
    state = _u(raw.get("state"))[:2]  # USPS two-letter
    zip_full = re.sub(r"[^\d-]", "", raw.get("zip", "") or "")

    # Common abbreviations.
    for full, abbr in [
        (r"\bSTREET\b", "ST"), (r"\bAVENUE\b", "AVE"),
        (r"\bROAD\b", "RD"), (r"\bAPARTMENT\b", "APT"),
        (r"\bSUITE\b", "STE"), (r"\bDRIVE\b", "DR"),
        (r"\bBOULEVARD\b", "BLVD"), (r"\bLANE\b", "LN"),
    ]:
        line1 = re.sub(full, abbr, line1)
        line2 = re.sub(full, abbr, line2)

    line1 = re.sub(r"\s+", " ", line1)
    line2 = re.sub(r"\s+", " ", line2)
    canonical = " ".join(p for p in [line1, line2, city, state, zip_full] if p)
    zip5 = (zip_full.split("-")[0] if zip_full else "")[:5]

    return {
        "line1": line1, "line2": line2, "city": city, "state": state,
        "zip": zip_full, "zip5": zip5, "canonical": canonical,
    }

def _parse_iso_date(raw) -> Optional[str]:
    """Parse a date-like string into canonical YYYY-MM-DD; None on failure."""
    if not raw:
        return None
    try:
        return dateparser.parse(str(raw)).strftime("%Y-%m-%d")
    except (ValueError, dateparser.ParserError, OverflowError, TypeError):
        return None

def _parse_license_entries(raw_nppes: dict) -> list:
    """
    NPPES allows up to 15 license-and-taxonomy entries per NPI
    (covering different states or different license types). The
    API returns them as a list of dicts with the fields below; the
    bulk file flattens them into numbered columns. Either way,
    parse into a list of normalized {license_number, license_state,
    taxonomy_code, taxonomy_description, is_primary_taxonomy} dicts.
    """
    entries = []
    for raw_entry in (raw_nppes.get("taxonomies") or []):
        entries.append({
            "license_number": _normalize_license_number(raw_entry.get("license")),
            "license_state":  (raw_entry.get("state") or "").upper().strip()[:2],
            "taxonomy_code":  (raw_entry.get("code") or "").strip(),
            "taxonomy_description": (raw_entry.get("desc") or "").strip(),
            "is_primary_taxonomy": bool(raw_entry.get("primary")),
        })
    return entries

def normalize_nppes_record(raw_nppes: dict, file_release_date: str = "") -> dict:
    """
    Apply the full normalization pipeline to one NPPES record.
    Output is the canonical form used downstream.

    Expected input keys (subset of NPI Registry API response):
      number, enumeration_type, last_updated_epoch, basic, addresses,
      taxonomies, identifiers, other_names
    """
    # The NPI Registry API nests fields under a "basic" object plus
    # parallel "addresses" and "taxonomies" lists. Extract once.
    basic = raw_nppes.get("basic", {}) or {}
    addresses = raw_nppes.get("addresses", []) or []
    other_names_list = raw_nppes.get("other_names", []) or []

    # Practice and mailing addresses come from the "addresses" list.
    practice = next(
        (a for a in addresses if a.get("address_purpose") == "LOCATION"),
        {},
    )
    mailing = next(
        (a for a in addresses if a.get("address_purpose") == "MAILING"),
        {},
    )

    practice_addr = _normalize_address({
        "line1": practice.get("address_1"),
        "line2": practice.get("address_2"),
        "city":  practice.get("city"),
        "state": practice.get("state"),
        "zip":   practice.get("postal_code"),
    })
    mailing_addr = _normalize_address({
        "line1": mailing.get("address_1"),
        "line2": mailing.get("address_2"),
        "city":  mailing.get("city"),
        "state": mailing.get("state"),
        "zip":   mailing.get("postal_code"),
    })

    enumeration_type = raw_nppes.get("enumeration_type", "")
    entity_type_code = "1" if "NPI-1" in enumeration_type else (
        "2" if "NPI-2" in enumeration_type else ""
    )

    deactivation_date = _parse_iso_date(basic.get("deactivation_date"))
    is_active = (deactivation_date is None)

    # Type-1 (individual) name fields.
    first_name = _normalize_name(basic.get("first_name")) if entity_type_code == "1" else ""
    last_name  = _normalize_name(basic.get("last_name"))  if entity_type_code == "1" else ""
    middle_name = _normalize_name(basic.get("middle_name")) if entity_type_code == "1" else ""
    suffix = _normalize_suffix(basic.get("name_suffix"))
    credential_string = _parse_credential_string(basic.get("credential"))

    # Type-2 (organizational) name fields.
    legal_business_name = (
        _normalize_organization_name(basic.get("organization_name"))
        if entity_type_code == "2" else ""
    )

    # Other names: previous legal names with a typed reason. Critical
    # for matching across name changes.
    other_names = []
    for o in other_names_list:
        other_names.append({
            "first_name": _normalize_name(o.get("first_name")),
            "last_name":  _normalize_name(o.get("last_name")),
            "type_code":  o.get("type"),
        })

    licenses = _parse_license_entries(raw_nppes)
    taxonomies = [
        {
            "nucc_code":     entry["taxonomy_code"],
            "description":   entry["taxonomy_description"],
            "is_primary":    entry["is_primary_taxonomy"],
            "license_state": entry["license_state"],
        }
        for entry in licenses if entry["taxonomy_code"]
    ]
    primary_taxonomy = next(
        (t["nucc_code"] for t in taxonomies if t["is_primary"]),
        (taxonomies[0]["nucc_code"] if taxonomies else ""),
    )

    return {
        "npi":                  str(raw_nppes.get("number", "")),
        "entity_type_code":     entity_type_code,
        "is_active":            is_active,
        "deactivation_date":    deactivation_date,
        "enumeration_date":     _parse_iso_date(basic.get("enumeration_date")),
        "last_update_date":     _parse_iso_date(basic.get("last_updated")),
        # Type-1 individual fields.
        "first_name":           first_name,
        "last_name":            last_name,
        "middle_name":          middle_name,
        "name_suffix":          suffix,
        "first_name_metaphone": _double_metaphone(first_name),
        "last_name_metaphone":  _double_metaphone(last_name),
        "credential_string":    credential_string,
        "other_names":          other_names,
        # Type-2 organizational fields.
        "legal_business_name":  legal_business_name,
        # Practice address (what members care about).
        "practice_address":     practice_addr,
        "mailing_address":      mailing_addr,
        "practice_zip5":        practice_addr["zip5"],
        "practice_state":       practice_addr["state"],
        # Phone / fax.
        "practice_phone":       _normalize_phone(practice.get("telephone_number")),
        "practice_fax":         _normalize_phone(practice.get("fax_number")),
        # License and taxonomy.
        "licenses":             licenses,
        "taxonomies":           taxonomies,
        "primary_taxonomy":     primary_taxonomy,
        # Provenance.
        "source":               "nppes",
        "nppes_file_release_date": file_release_date,
        "normalized_at":        _now_iso(),
        "normalizer_version":   NORMALIZER_VERSION,
    }

def normalize_internal_provider(raw_internal: dict) -> dict:
    """
    Apply the normalization pipeline to one internal provider record.
    The schema below is illustrative; in production a small adapter
    maps from the credentialing-system schema to these canonical
    field names.
    """
    first_name = _normalize_name(raw_internal.get("first_name"))
    last_name  = _normalize_name(raw_internal.get("last_name"))
    middle_name = _normalize_name(raw_internal.get("middle_name"))
    suffix = _normalize_suffix(raw_internal.get("suffix"))
    credential_string = _parse_credential_string(raw_internal.get("credentials"))

    # Address.
    practice_addr = _normalize_address({
        "line1": raw_internal.get("address_line_1"),
        "line2": raw_internal.get("address_line_2"),
        "city":  raw_internal.get("city"),
        "state": raw_internal.get("state"),
        "zip":   raw_internal.get("zip"),
    })

    # Specialty -> NUCC mapping.
    raw_specialty = (raw_internal.get("specialty") or "").lower().strip()
    primary_taxonomy = INTERNAL_SPECIALTY_TO_NUCC.get(
        raw_specialty,
        "unknown_taxonomy",
    )

    # License (single license is typical in internal records; some
    # systems carry multiple).
    licenses = []
    if raw_internal.get("license_number"):
        licenses.append({
            "license_number": _normalize_license_number(
                raw_internal.get("license_number")),
            "license_state":  (raw_internal.get("license_state") or "").upper()[:2],
            "taxonomy_code":  primary_taxonomy if primary_taxonomy != "unknown_taxonomy" else "",
            "is_primary_taxonomy": True,
        })

    # Match mode: do we already have an NPI on file, or do we need
    # to search for one?
    existing_npi = raw_internal.get("npi")
    has_existing = bool(existing_npi)

    # is_active flag for the internal record itself; a deactivated
    # internal record is rare but the matcher should respect it.
    is_active = bool(raw_internal.get("is_active", True))

    return {
        "internal_provider_id": raw_internal.get("provider_id", ""),
        "match_mode":           "confirm" if has_existing else "search",
        "has_existing_npi":     has_existing,
        "existing_npi":         str(existing_npi) if existing_npi else "",
        "first_name":           first_name,
        "last_name":            last_name,
        "middle_name":          middle_name,
        "name_suffix":          suffix,
        "first_name_metaphone": _double_metaphone(first_name),
        "last_name_metaphone":  _double_metaphone(last_name),
        "credential_string":    credential_string,
        "primary_taxonomy":     primary_taxonomy,
        "practice_address":     practice_addr,
        "practice_phone":       _normalize_phone(raw_internal.get("phone")),
        "licenses":              licenses,
        "is_active":             is_active,
        "normalized_at":         _now_iso(),
        "normalizer_version":    NORMALIZER_VERSION,
    }

def _npi_registry_lookup(params: dict) -> list:
    """
    Talk to the public NPI Registry API. The API supports searches
    by NPI number, name, location, taxonomy, and a few other fields.
    Returns a list of normalized NPPES records. Empty list on
    network failure rather than raising; the caller decides how to
    handle a partial candidate set.

    Production note: the API is rate-limited. For batch workloads,
    do not iterate API calls; use the monthly Downloadable File.
    """
    request_params = {"version": NPI_REGISTRY_API_VERSION,
                       "limit": NPI_REGISTRY_MAX_RESULTS_PER_QUERY,
                       **params}
    try:
        resp = requests.get(
            NPI_REGISTRY_BASE_URL,
            params=request_params,
            timeout=NPI_REGISTRY_TIMEOUT_SECONDS,
        )
        resp.raise_for_status()
        body = resp.json()
    except (requests.RequestException, ValueError) as exc:
        logger.warning(
            "npi registry api call failed",
            extra={"params": request_params, "error": str(exc)},
        )
        return []

    results = body.get("results") or []
    return [normalize_nppes_record(r) for r in results]

def _candidate_key(c: dict) -> str:
    """Stable identity for a candidate; used to deduplicate across passes."""
    return c.get("npi", "")

def generate_candidates(internal_record: dict) -> list:
    """
    Run multiple registry-lookup passes and return the deduplicated
    union. Each pass plays the role of a blocking strategy from
    recipe 5.1, adapted to a one-sided lookup against an external
    authoritative source.
    """
    candidates_by_npi = {}

    def _add(c, pass_name):
        npi = _candidate_key(c)
        if not npi:
            return
        if npi not in candidates_by_npi:
            c["_blocking_passes"] = [pass_name]
            candidates_by_npi[npi] = c
        else:
            existing = candidates_by_npi[npi]
            if pass_name not in existing["_blocking_passes"]:
                existing["_blocking_passes"].append(pass_name)

    # Pass 0: existing-NPI confirmation. If the internal record
    # already claims an NPI, look it up directly. We still run the
    # search passes below so we can detect the case where the
    # internal record's NPI is wrong (a real failure mode in
    # legacy data).
    if internal_record["has_existing_npi"]:
        results = _npi_registry_lookup({"number": internal_record["existing_npi"]})
        for c in results:
            _add(c, "existing_npi_confirmation")

    # Pass 1: license-number + license-state. Highest information.
    # The API has a `taxonomy_description` parameter but no direct
    # "license number" parameter; license-number search is done by
    # exact string match on the license field within taxonomies,
    # so we issue a name-plus-state query and filter client-side
    # for license-number match. This pass is "license-anchored"
    # because of the client-side filter, not the API query shape.
    # In production with the bulk file, the license-number lookup
    # is a direct index hit. 
    if internal_record["licenses"]:
        for license_entry in internal_record["licenses"]:
            if not license_entry["license_number"]:
                continue
            # API does not support direct license-number search; we
            # do a name-plus-state pass and filter client-side for
            # license-number match.
            results = _npi_registry_lookup({
                "first_name": internal_record["first_name"],
                "last_name":  internal_record["last_name"],
                "state":      license_entry["license_state"],
                "enumeration_type": "NPI-1",
            })
            # Client-side license-number filter: only keep candidates
            # whose NPPES license entries contain a matching number.
            target_license = license_entry["license_number"].upper().strip()
            for c in results:
                candidate_licenses = [
                    lic.get("number", "").upper().strip()
                    for lic in c.get("taxonomies", [])
                    if lic.get("license", "")
                ]
                # Also check the flattened license field if present
                if target_license in candidate_licenses or _license_in_candidate(c, target_license):
                    _add(c, "license_number_state")

    # Pass 2: last-name (with metaphone fallback) + first-name initial
    # + practice state.
    if internal_record["last_name"] and internal_record["practice_address"]["state"]:
        results = _npi_registry_lookup({
            "last_name":  internal_record["last_name"],
            "first_name": internal_record["first_name"],
            "state":      internal_record["practice_address"]["state"],
            "enumeration_type": "NPI-1",
        })
        for c in results:
            _add(c, "last_name_first_name_state")

    # Pass 3: last-name + primary taxonomy + state.
    # The API accepts a `taxonomy_description` string parameter.
    # We look up the description from the internal record's primary
    # taxonomy code using a NUCC-code-to-description map so this
    # pass produces a candidate set distinct from Pass 2 when the
    # internal record has a strong taxonomy signal.
    if (internal_record["primary_taxonomy"] != "unknown_taxonomy"
            and internal_record["last_name"]
            and internal_record["practice_address"]["state"]):
        # Map the NUCC code to a description string the API understands.
        taxonomy_desc = NUCC_CODE_TO_DESCRIPTION.get(
            internal_record["primary_taxonomy"], "")
        results = _npi_registry_lookup({
            "last_name":     internal_record["last_name"],
            "taxonomy_description": taxonomy_desc,
            "state":         internal_record["practice_address"]["state"],
            "enumeration_type": "NPI-1",
        })
        for c in results:
            _add(c, "last_name_taxonomy_state")

    # Pass 4: practice ZIP5 + last-name initial.
    if (internal_record["last_name"]
            and internal_record["practice_address"]["zip5"]):
        results = _npi_registry_lookup({
            "postal_code": internal_record["practice_address"]["zip5"],
            "last_name":   internal_record["last_name"],
            "enumeration_type": "NPI-1",
        })
        for c in results:
            _add(c, "zip_lastname")

    # (We deliberately omit the phone-last-4 pass from this demo
    # because the public API does not expose phone search; in
    # production, the OpenSearch-backed bulk-file index supports it.)

    candidates = list(candidates_by_npi.values())

    # Cap candidate set size; route oversized cases to review rather
    # than auto-decide. Most internal records produce <50 candidates;
    # outliers (very common surnames in dense states) can produce
    # hundreds, which is a signal that auto-decision is unsafe.
    if len(candidates) > MAX_CANDIDATES_BEFORE_REVIEW:
        logger.info(
            "candidate set oversized; will route to review",
            extra={
                "internal_provider_id": internal_record["internal_provider_id"],
                "candidate_count": len(candidates),
            },
        )
        # Tag the internal record so the caller knows to route to review.
        internal_record["_oversized_candidate_set"] = True
        return candidates[:MAX_CANDIDATES_BEFORE_REVIEW]

    return candidates

def _compare_first_name(a_record: dict, candidate: dict) -> str:
    a_first = a_record["first_name"]
    c_first = candidate["first_name"]
    if not a_first or not c_first:
        return "mismatch"
    if a_first == c_first:
        return "exact"

    jw = jellyfish.jaro_winkler_similarity(a_first, c_first)
    if jw >= 0.92:
        return "jaro_winkler_high"

    if a_record["first_name_metaphone"] and (
        a_record["first_name_metaphone"] == candidate["first_name_metaphone"]
    ):
        return "metaphone_match"

    # Other-names check: NPPES carries previous legal names with a
    # type code. If the internal first name matches an other-name
    # entry on the candidate, count it as a name-change match.
    for o in candidate.get("other_names", []):
        if o.get("first_name") == a_first:
            return "other_name_match"

    return "mismatch"

def _compare_last_name(a_record: dict, candidate: dict) -> str:
    a_last = a_record["last_name"]
    c_last = candidate["last_name"]
    if not a_last or not c_last:
        return "mismatch"
    if a_last == c_last:
        return "exact"

    distance = jellyfish.damerau_levenshtein_distance(a_last, c_last)
    longest = max(len(a_last), len(c_last))
    similarity = 1.0 - (distance / longest) if longest else 0.0
    if similarity >= 0.85:
        return "damerau_high"

    if a_record["last_name_metaphone"] and (
        a_record["last_name_metaphone"] == candidate["last_name_metaphone"]
    ):
        return "metaphone_match"

    # Other-names check on last name (legal name change pattern).
    for o in candidate.get("other_names", []):
        if o.get("last_name") == a_last:
            return "other_name_match"

    return "mismatch"

def _compare_credentials(internal_creds: list, candidate_creds: list) -> str:
    if not internal_creds or not candidate_creds:
        return "one_null"
    set_internal = set(internal_creds)
    set_candidate = set(candidate_creds)
    if set_internal == set_candidate:
        return "exact_set_match"
    # Subset pattern: internal record has an outdated subset of the
    # candidate's credential list, or vice versa. Common when the
    # provider has earned an additional credential the internal
    # record has not picked up yet.
    if set_internal & set_candidate and (
        set_internal.issubset(set_candidate) or set_candidate.issubset(set_internal)
    ):
        return "subset_match"
    return "mismatch"

def _compare_license_set(internal_licenses: list, candidate_licenses: list) -> str:
    """
    The strongest comparator in the pipeline. Exact match on
    license-number-plus-state is essentially conclusive (license
    numbers are state-issued and unique within the state).
    """
    if not internal_licenses or not candidate_licenses:
        return "one_null"

    for il in internal_licenses:
        for cl in candidate_licenses:
            if not il["license_number"] or not cl["license_number"]:
                continue
            same_state = (il["license_state"] and il["license_state"] == cl["license_state"])
            same_number = (il["license_number"] == cl["license_number"])
            if same_number and same_state:
                return "exact_number_and_state"
            if same_number and not same_state:
                return "number_only_match"

    # No license-number match. Check if at least one license-state
    # is shared (weak signal; many providers practice in many states).
    internal_states = {il["license_state"] for il in internal_licenses if il["license_state"]}
    candidate_states = {cl["license_state"] for cl in candidate_licenses if cl["license_state"]}
    if internal_states & candidate_states:
        return "state_only_match"

    return "mismatch"

def _compare_taxonomy(internal_primary: str, candidate_taxonomies: list) -> str:
    if internal_primary == "unknown_taxonomy" or not internal_primary:
        return "internal_unknown"
    if not candidate_taxonomies:
        return "internal_unknown"

    candidate_codes = {t["nucc_code"] for t in candidate_taxonomies if t.get("nucc_code")}
    candidate_primary = next(
        (t["nucc_code"] for t in candidate_taxonomies if t.get("is_primary")),
        "",
    )

    if internal_primary == candidate_primary:
        return "primary_match"
    if internal_primary in candidate_codes:
        return "any_match"
    # Parent-class match: the first three characters of the NUCC
    # code identify the broad classification (e.g., 207 = Allopathic
    # & Osteopathic Physicians, 208 = ...). A real implementation
    # uses the full NUCC hierarchy table for accurate parent-class
    # matching; this is a placeholder.
    internal_class = internal_primary[:3]
    candidate_classes = {c[:3] for c in candidate_codes}
    if internal_class in candidate_classes:
        return "parent_match"
    return "mismatch"

def _compare_address(internal_addr: dict, candidate_addr: dict) -> str:
    if not internal_addr or not candidate_addr:
        return "one_null"
    if not internal_addr.get("canonical") or not candidate_addr.get("canonical"):
        return "one_null"
    if internal_addr["canonical"] == candidate_addr["canonical"]:
        return "exact"
    if internal_addr.get("zip5") and internal_addr["zip5"] == candidate_addr.get("zip5"):
        return "same_zip"
    if internal_addr.get("state") and internal_addr["state"] == candidate_addr.get("state"):
        return "same_state"
    return "mismatch"

def _compare_phone(internal_phone: str, candidate_phone: str) -> str:
    if not internal_phone or not candidate_phone:
        return "one_null"
    if internal_phone == candidate_phone:
        return "exact"
    if internal_phone[-4:] == candidate_phone[-4:]:
        return "last_4_match"
    return "mismatch"

def _candidate_has_matching_license_state(internal_record: dict, candidate: dict) -> bool:
    """Hard filter helper: does the candidate have any license in any of the internal record's license states?"""
    internal_states = {
        (il.get("license_state") or "").upper()
        for il in internal_record.get("licenses", [])
        if il.get("license_state")
    }
    if not internal_states:
        return True  # no internal license state to enforce
    candidate_states = {
        (cl.get("license_state") or "").upper()
        for cl in candidate.get("licenses", [])
        if cl.get("license_state")
    }
    return bool(internal_states & candidate_states)

def _log_likelihood_ratio(field: str, level: str) -> Decimal:
    """Per-field, per-comparison-level log-likelihood-ratio contribution. Same pattern as recipe 5.1."""
    m = M_PROBABILITIES.get(field, {}).get(level)
    u = U_PROBABILITIES.get(field, {}).get(level)
    if m is None or u is None:
        return Decimal("0")
    if level in {"one_null", "internal_unknown"}:
        return Decimal("0")
    if m <= 0 or u <= 0:
        return Decimal("0")
    return _to_decimal(math.log(float(m) / float(u)))

def _drift_relevant_snapshot(candidate: dict) -> dict:
    """Pick the registry fields most likely to drift between re-verifications."""
    return {
        "practice_address":  candidate.get("practice_address", {}).get("canonical", ""),
        "practice_phone":    candidate.get("practice_phone", ""),
        "primary_taxonomy":  candidate.get("primary_taxonomy", ""),
        "all_taxonomies":    [t.get("nucc_code") for t in candidate.get("taxonomies", [])],
        "is_active":         candidate.get("is_active", True),
        "deactivation_date": candidate.get("deactivation_date"),
        "last_update_date":  candidate.get("last_update_date"),
    }

def score_candidates(internal_record: dict, candidates: list) -> list:
    """
    Apply hard filters, score the survivors with per-field
    comparators and a Fellegi-Sunter combiner, return the scored
    list sorted descending by composite score (best candidate
    first; second is the runner-up the router uses for the margin
    requirement).
    """
    scored = []

    for candidate in candidates:
        # Hard filter 1: deactivation. A deactivated NPI almost
        # never matches an active internal record. Filter out
        # unless the internal record is also marked inactive.
        if (not candidate.get("is_active", True)
                and internal_record.get("is_active", True)):
            logger.debug(
                "filtered deactivated candidate",
                extra={"npi": candidate.get("npi")},
            )
            continue

        # Hard filter 2: type mismatch. We are matching individual
        # providers; reject Type-2 (organizational) NPIs.
        if (internal_record["match_mode"] == "search"
                and candidate.get("entity_type_code") != "1"):
            logger.debug(
                "filtered type-mismatch candidate",
                extra={"npi": candidate.get("npi")},
            )
            continue

        # Hard filter 3: license-state mismatch. When the internal
        # record has explicit license states, reject candidates that
        # do not share at least one license state.
        if (internal_record.get("licenses")
                and not _candidate_has_matching_license_state(internal_record, candidate)):
            logger.debug(
                "filtered license-state-mismatch candidate",
                extra={"npi": candidate.get("npi")},
            )
            continue

        # Per-field comparison levels.
        field_comparisons = {
            "first_name": _compare_first_name(internal_record, candidate),
            "last_name":  _compare_last_name(internal_record, candidate),
            "credential": _compare_credentials(
                internal_record["credential_string"],
                candidate["credential_string"],
            ),
            "license":    _compare_license_set(
                internal_record["licenses"], candidate["licenses"],
            ),
            "taxonomy":   _compare_taxonomy(
                internal_record["primary_taxonomy"], candidate["taxonomies"],
            ),
            "address":    _compare_address(
                internal_record["practice_address"], candidate["practice_address"],
            ),
            "phone":      _compare_phone(
                internal_record["practice_phone"], candidate["practice_phone"],
            ),
        }

        per_field_log_ratios = {
            field: _log_likelihood_ratio(field, level)
            for field, level in field_comparisons.items()
        }
        composite = sum(per_field_log_ratios.values(), Decimal("0"))
        match_probability = Decimal(
            str(1.0 / (1.0 + math.exp(-float(composite))))
        )

        scored.append({
            "internal_provider_id": internal_record["internal_provider_id"],
            "candidate_npi":        candidate["npi"],
            "composite_score":      composite,
            "match_probability":    match_probability,
            "field_comparisons":    field_comparisons,
            "per_field_log_ratios": per_field_log_ratios,
            "blocking_passes":      candidate.get("_blocking_passes", []),
            "candidate_snapshot":   _drift_relevant_snapshot(candidate),
            "scored_at":            _now_iso(),
            "model_version":        MODEL_VERSION,
        })

    scored.sort(key=lambda r: r["composite_score"], reverse=True)
    return scored

def _serialize_for_dynamodb(obj):
    """Recursive serialization helper. Same pattern as recipe 5.1."""
    if isinstance(obj, dict):
        return {k: _serialize_for_dynamodb(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple)):
        return [_serialize_for_dynamodb(v) for v in obj]
    if isinstance(obj, float):
        return Decimal(str(obj))
    return obj

def _write_audit_archive(record: dict, partition: str) -> None:
    """Write a JSON-serialized audit record to S3, partitioned by decision and date."""
    today = datetime.now(timezone.utc).strftime("%Y/%m/%d")
    audit_key = f"audit/{partition}/{today}/{uuid.uuid4()}.json"
    body = json.dumps(record, default=str).encode("utf-8")
    try:
        s3_client.put_object(
            Bucket=AUDIT_BUCKET,
            Key=audit_key,
            Body=body,
            ServerSideEncryption="aws:kms",
        )
    except Exception as exc:
        logger.error(
            "audit archive write failed",
            extra={"partition": partition, "error": str(exc)},
        )

def _emit_metric(metric_name: str, value: float, dimensions: dict = None) -> None:
    try:
        cloudwatch_client.put_metric_data(
            Namespace=CLOUDWATCH_NAMESPACE,
            MetricData=[{
                "MetricName": metric_name,
                "Value": value,
                "Unit": "Count",
                "Dimensions": [
                    {"Name": k, "Value": v} for k, v in (dimensions or {}).items()
                ],
            }],
        )
    except Exception as exc:
        logger.warning("metric emit failed", extra={"metric": metric_name, "error": str(exc)})

def route_match(
    internal_record: dict,
    scored_candidates: list,
    high_threshold: Decimal = HIGH_THRESHOLD,
    low_threshold: Decimal = LOW_THRESHOLD,
    min_margin: Decimal = MIN_MARGIN,
) -> dict:
    """
    Decide auto_attach / review / auto_non_match. Write the outcome
    to the audit archive. For review, write a queue item to
    DynamoDB. Returns a small dict with the routing decision and
    the chosen candidate (if any).
    """
    if internal_record.get("_oversized_candidate_set"):
        # Oversized candidate sets always go to review; the matcher
        # cannot reliably auto-decide.
        decision = "review"
        reason = "oversized_candidate_set"
        chosen = None
        runner_up = None
    elif not scored_candidates:
        decision = "review"
        reason = "no_viable_candidates"
        chosen = None
        runner_up = None
    else:
        chosen = scored_candidates[0]
        runner_up = scored_candidates[1] if len(scored_candidates) > 1 else None
        margin = (
            chosen["composite_score"] - runner_up["composite_score"]
            if runner_up else Decimal("999")
        )

        if (chosen["composite_score"] >= high_threshold
                and margin >= min_margin):
            decision = "auto_attach"
            reason = "high_score_and_margin"
        elif chosen["composite_score"] <= low_threshold:
            decision = "auto_non_match"
            reason = "no_match_in_registry"
        elif chosen["composite_score"] < high_threshold:
            decision = "review"
            reason = "borderline_score"
        else:
            # Cleared the high threshold but failed the margin test.
            decision = "review"
            reason = "narrow_margin"

    # Audit archive write for every decision (not just review).
    audit_record = _serialize_for_dynamodb({
        "internal_provider_id": internal_record["internal_provider_id"],
        "routing_decision":     decision,
        "reason":               reason,
        "chosen_candidate":     chosen,
        "runner_up_candidate":  runner_up,
        "high_threshold":       high_threshold,
        "low_threshold":        low_threshold,
        "min_margin":           min_margin,
        "all_scored_candidates": scored_candidates[:5],  # top 5 for context
        "decided_at":           _now_iso(),
    })
    _write_audit_archive(audit_record, partition=decision)
    _emit_metric("RoutingDecision", 1.0, dimensions={"Decision": decision, "Reason": reason})

    if decision == "review":
        # Compute priority so the credentialing-team UI can sort
        # the queue by urgency rather than arrival order.
        # Priority logic: narrow-margin cases (two strong candidates)
        # are higher priority than borderline-score cases (weak
        # candidates) because they are more likely to be resolvable
        # with a quick license-number lookup.
        if reason == "narrow_margin":
            priority = "high"
        elif reason == "existing_npi_disputed" or reason == "existing_npi_not_resolved":
            priority = "high"
        elif chosen and chosen.get("match_probability", 0) > 0.8:
            priority = "medium"
        else:
            priority = "low"

        review_item = _serialize_for_dynamodb({
            "queue_id":           "default",  # tier by team in production
            "candidate_pair_id":  str(uuid.uuid4()),
            "internal_provider_id": internal_record["internal_provider_id"],
            "internal_record_snapshot": internal_record,
            "scored_candidates_snapshot": scored_candidates[:5],
            "reason":             reason,
            "priority":           priority,
            "queued_at":          _now_iso(),
            "review_status":      "pending",
            "model_version":      MODEL_VERSION,
        })
        try:
            dynamodb.Table(REVIEW_QUEUE_TABLE).put_item(Item=review_item)
        except Exception as exc:
            logger.error(
                "review queue write failed",
                extra={"internal_provider_id": internal_record["internal_provider_id"],
                       "error": str(exc)},
            )

    return {
        "decision": decision,
        "reason":   reason,
        "chosen":   chosen,
        "runner_up": runner_up,
    }

def attach_npi(internal_record: dict, matched_candidate_score: dict,
                decision_metadata: dict) -> None:
    """
    Write the assignment, schedule re-verification, write the audit
    record, and emit the assignment event.

    decision_metadata: who or what decided (auto_attach pipeline, or
    review_decision_by_user_X), the score, the model version, and
    the timestamp.
    """
    matched_npi = matched_candidate_score["candidate_npi"]
    drift_snapshot = matched_candidate_score["candidate_snapshot"]

    next_verify_date = (
        datetime.now(timezone.utc).date() + timedelta(days=VERIFICATION_CADENCE_DAYS)
    ).isoformat()

    assignment_item = _serialize_for_dynamodb({
        "internal_provider_id":  internal_record["internal_provider_id"],
        "matched_npi":           matched_npi,
        "match_score":           matched_candidate_score["composite_score"],
        "match_probability":     matched_candidate_score["match_probability"],
        "field_comparisons":     matched_candidate_score["field_comparisons"],
        "match_method":          decision_metadata.get("method"),
        "decided_by":            decision_metadata.get("decided_by"),
        "decided_at":            decision_metadata.get("decided_at"),
        "model_version":         matched_candidate_score["model_version"],
        "drift_snapshot":        drift_snapshot,
        "last_verified_at":      _now_iso(),
        "next_verification_due_at": next_verify_date,
        "active":                True,
    })
    try:
        dynamodb.Table(ASSIGNMENT_TABLE).put_item(Item=assignment_item)
    except Exception as exc:
        logger.error(
            "assignment write failed",
            extra={
                "internal_provider_id": internal_record["internal_provider_id"],
                "matched_npi": matched_npi,
                "error": str(exc),
            },
        )
        # In production: DLQ + alarm. Do not silently lose the assignment.
        return

    # Schedule the next re-verification. The schedule table is keyed
    # by (verification_due_date, internal_provider_id) so the daily
    # job can pull due-or-overdue records efficiently.
    schedule_item = _serialize_for_dynamodb({
        "verification_due_date":  next_verify_date,
        "internal_provider_id":   internal_record["internal_provider_id"],
        "matched_npi":            matched_npi,
        "scheduled_at":           _now_iso(),
    })
    try:
        dynamodb.Table(SCHEDULE_TABLE).put_item(Item=schedule_item)
    except Exception as exc:
        logger.error(
            "schedule write failed",
            extra={
                "internal_provider_id": internal_record["internal_provider_id"],
                "error": str(exc),
            },
        )

    # Audit archive entry for the attachment.
    audit_record = _serialize_for_dynamodb({
        "attachment_id":         str(uuid.uuid4()),
        "internal_provider_id":  internal_record["internal_provider_id"],
        "matched_npi":           matched_npi,
        "decision_metadata":     decision_metadata,
        "drift_snapshot":        drift_snapshot,
        "attached_at":           _now_iso(),
    })
    _write_audit_archive(audit_record, partition="attached")

    # Emit the assignment event.
    try:
        eventbridge_client.put_events(Entries=[{
            "Source":       "provider-npi-matching",
            "DetailType":   "npi_attached",
            "EventBusName": EVENTS_BUS_NAME,
            "Detail": json.dumps({
                "internal_provider_id": internal_record["internal_provider_id"],
                "matched_npi":          matched_npi,
                "match_score":          str(matched_candidate_score["composite_score"]),
                "attached_at":          audit_record["attached_at"],
            }, default=str),
        }])
    except Exception as exc:
        logger.error("assignment event emit failed", extra={"error": str(exc)})

    _emit_metric("NPIAttached", 1.0,
                  dimensions={"Method": decision_metadata.get("method", "unknown")})

def _compare_drift_snapshot(previous: dict, current_candidate: dict) -> dict:
    """Detect drift between the stored snapshot and the latest registry record."""
    current = _drift_relevant_snapshot(current_candidate)
    drift = {
        "address_changed":      previous.get("practice_address") != current.get("practice_address"),
        "phone_changed":        previous.get("practice_phone") != current.get("practice_phone"),
        "taxonomy_changed":     previous.get("primary_taxonomy") != current.get("primary_taxonomy"),
        "deactivation_changed": previous.get("is_active") != current.get("is_active"),
        "previous_snapshot":    previous,
        "current_snapshot":     current,
    }
    drift["any_drift"] = any([
        drift["address_changed"], drift["phone_changed"],
        drift["taxonomy_changed"], drift["deactivation_changed"],
    ])
    return drift

def re_verify_npi(internal_provider_id: str) -> dict:
    """
    Pull the current assignment, fetch the latest registry record,
    detect drift, update the snapshot, reschedule, emit drift events
    as appropriate. Returns a summary dict for logging and metrics.
    """
    try:
        resp = dynamodb.Table(ASSIGNMENT_TABLE).get_item(
            Key={"internal_provider_id": internal_provider_id},
        )
    except Exception as exc:
        logger.error("assignment read failed", extra={"error": str(exc)})
        return {"status": "read_failed"}

    assignment = resp.get("Item")
    if not assignment:
        return {"status": "no_assignment"}

    matched_npi = assignment["matched_npi"]
    previous_snapshot = assignment.get("drift_snapshot", {})

    # Fetch the current registry record for this NPI.
    current_candidates = _npi_registry_lookup({"number": matched_npi})
    if not current_candidates:
        # The registry returned nothing for an NPI it should know
        # about; treat this as a deactivation-or-removal event.
        logger.warning(
            "registry returned no record for known NPI",
            extra={"matched_npi": matched_npi,
                   "internal_provider_id": internal_provider_id},
        )
        return {"status": "registry_missing", "matched_npi": matched_npi}

    current_candidate = current_candidates[0]
    drift = _compare_drift_snapshot(previous_snapshot, current_candidate)

    # Update the assignment with the new snapshot and re-verification
    # metadata. Reschedule the next verification.
    next_verify_date = (
        datetime.now(timezone.utc).date() + timedelta(days=VERIFICATION_CADENCE_DAYS)
    ).isoformat()
    new_snapshot = _drift_relevant_snapshot(current_candidate)

    try:
        dynamodb.Table(ASSIGNMENT_TABLE).update_item(
            Key={"internal_provider_id": internal_provider_id},
            UpdateExpression=(
                "SET drift_snapshot = :snap, "
                "last_verified_at = :ts, "
                "next_verification_due_at = :due"
            ),
            ExpressionAttributeValues={
                ":snap": _serialize_for_dynamodb(new_snapshot),
                ":ts":   _now_iso(),
                ":due":  next_verify_date,
            },
        )
        # Schedule next re-verification. The table is keyed on
        # internal_provider_id so each provider has at most one row.
        # This Put overwrites the prior row, preventing stale-entry
        # accumulation. The GSI on verification_due_date allows the
        # daily job to query due records efficiently.
        dynamodb.Table(SCHEDULE_TABLE).put_item(Item=_serialize_for_dynamodb({
            "verification_due_date": next_verify_date,
            "internal_provider_id":  internal_provider_id,
            "matched_npi":           matched_npi,
            "scheduled_at":          _now_iso(),
        }))
    except Exception as exc:
        logger.error("re-verify writes failed", extra={"error": str(exc)})

    # Surface drift events.
    if drift["deactivation_changed"] and not new_snapshot.get("is_active"):
        try:
            eventbridge_client.put_events(Entries=[{
                "Source":       "provider-npi-matching",
                "DetailType":   "npi_deactivated",
                "EventBusName": EVENTS_BUS_NAME,
                "Detail": json.dumps({
                    "internal_provider_id": internal_provider_id,
                    "matched_npi":          matched_npi,
                    "deactivation_date":    new_snapshot.get("deactivation_date"),
                }, default=str),
            }])
        except Exception as exc:
            logger.error("deactivation event emit failed", extra={"error": str(exc)})

    if drift["address_changed"]:
        try:
            eventbridge_client.put_events(Entries=[{
                "Source":       "provider-npi-matching",
                "DetailType":   "practice_address_changed",
                "EventBusName": EVENTS_BUS_NAME,
                "Detail": json.dumps({
                    "internal_provider_id": internal_provider_id,
                    "matched_npi":          matched_npi,
                    "old_address":          previous_snapshot.get("practice_address"),
                    "new_address":          new_snapshot.get("practice_address"),
                }, default=str),
            }])
        except Exception as exc:
            logger.error("address-change event emit failed", extra={"error": str(exc)})

    # Emit taxonomy_changed event. Taxonomy drift is a directory event
    # and a network-adequacy event (per-specialty provider counts
    # change when a provider's primary taxonomy changes).
    if drift["taxonomy_changed"]:
        try:
            eventbridge_client.put_events(Entries=[{
                "Source":       "provider-npi-matching",
                "DetailType":   "taxonomy_changed",
                "EventBusName": EVENTS_BUS_NAME,
                "Detail": json.dumps({
                    "internal_provider_id": internal_provider_id,
                    "matched_npi":          matched_npi,
                    "old_primary_taxonomy": previous_snapshot.get("primary_taxonomy"),
                    "new_primary_taxonomy": new_snapshot.get("primary_taxonomy"),
                    "old_all_taxonomies":   previous_snapshot.get("all_taxonomies", []),
                    "new_all_taxonomies":   new_snapshot.get("all_taxonomies", []),
                }, default=str),
            }])
        except Exception as exc:
            logger.error("taxonomy-change event emit failed", extra={"error": str(exc)})

    _write_audit_archive(_serialize_for_dynamodb({
        "verification_id":       str(uuid.uuid4()),
        "internal_provider_id":  internal_provider_id,
        "matched_npi":           matched_npi,
        "drift":                 drift,
        "verified_at":           _now_iso(),
    }), partition="re_verified")

    _emit_metric("ReVerification", 1.0,
                  dimensions={"AnyDrift": "true" if drift["any_drift"] else "false"})

    return {"status": "ok", "drift": drift, "matched_npi": matched_npi}

def run_match_pipeline_for_provider(raw_internal: dict) -> dict:
    """
    End-to-end match pipeline for a single internal provider record.
    Returns a small summary dict for the demo to print.
    """
    summary = {
        "internal_provider_id": raw_internal.get("provider_id", ""),
        "candidate_count":      0,
        "decision":             "",
        "reason":               "",
        "matched_npi":          None,
        "match_score":          None,
    }

    # Step 2: normalize the internal record. (We run normalization
    # of the registry side inline below, since each candidate comes
    # from an API response.)
    internal_normalized = normalize_internal_provider(raw_internal)

    # Step 3: generate candidates.
    candidates = generate_candidates(internal_normalized)
    summary["candidate_count"] = len(candidates)

    # Step 4: score the candidates.
    scored = score_candidates(internal_normalized, candidates)

    # Step 5: route by threshold and margin.
    routing = route_match(internal_normalized, scored)
    summary["decision"] = routing["decision"]
    summary["reason"]   = routing["reason"]

    if routing["decision"] == "auto_attach":
        # Step 6: attach the NPI and schedule re-verification.
        decision_metadata = {
            "method":      "auto_attach",
            "decided_by":  "matching_pipeline",
            "decided_at":  _now_iso(),
            "score":       routing["chosen"]["composite_score"],
            "model_version": routing["chosen"]["model_version"],
        }
        try:
            attach_npi(internal_normalized, routing["chosen"], decision_metadata)
            summary["matched_npi"]  = routing["chosen"]["candidate_npi"]
            summary["match_score"]  = float(routing["chosen"]["composite_score"])
        except Exception as exc:
            logger.error(
                "attach failed",
                extra={"internal_provider_id": internal_normalized["internal_provider_id"],
                       "error": str(exc)},
            )

    return summary

def run_demo():
    """
    Run the full pipeline against a small synthetic roster. This
    talks to the real public NPI Registry API for candidate
    generation, so the candidate counts in the printed output will
    vary from run to run (the registry data is real and updated
    monthly, even though the internal records below are fictional).
    """
    print("=" * 70)
    print("Provider NPI Matching Demo")
    print("=" * 70)
    print()
    print("Note: this demo calls the public NPI Registry API for "
          "candidate generation.")
    print("The internal provider records below are fictional. The "
          "matched NPIs are real")
    print("registry entries that happen to match the synthetic "
          "demographics.")
    print()

    for record in SYNTHETIC_INTERNAL_PROVIDERS:
        print("-" * 70)
        print(f"Internal record: {record['provider_id']} "
              f"({record['first_name']} {record['last_name']}, "
              f"{record.get('specialty', 'unknown specialty')})")
        result = run_match_pipeline_for_provider(record)
        print(f"  candidates: {result['candidate_count']}")
        print(f"  decision:   {result['decision']} ({result['reason']})")
        if result["matched_npi"]:
            print(f"  matched:    NPI {result['matched_npi']}  "
                  f"score={result['match_score']:.2f}")

SYNTHETIC_INTERNAL_PROVIDERS = [
    # Internal record with full data: license number, state, taxonomy,
    # address. License-anchored auto-attach is the expected outcome
    # when the registry data is current.
    {
        "provider_id":   "provider-internal-00874",
        "first_name":    "Sarah",
        "last_name":     "Patel",
        "credentials":   "MD",
        "specialty":     "family medicine",
        "license_number": "MD-87543",
        "license_state":  "CA",
        "address_line_1": "1421 Elm Street Apt 4",
        "city":          "Anytown",
        "state":         "CA",
        "zip":           "94555",
        "phone":         "(555) 123-4567",
        "is_active":     True,
    },
    # Internal record without license number; matcher has to find
    # the NPI via name+state+taxonomy. More likely to land in review.
    {
        "provider_id":   "provider-internal-01205",
        "first_name":    "John",
        "last_name":     "Smith",
        "credentials":   "MD, MPH",
        "specialty":     "internal medicine",
        "address_line_1": "200 Pine Lane",
        "city":          "Anytown",
        "state":         "TX",
        "zip":           "75001",
        "phone":         "555-444-9090",
        "is_active":     True,
    },
    # Internal record with an existing NPI on file. Confirmation path,
    # not search. Note: this record omits the `npi` field so the demo
    # exercises the search path (license-anchored auto-attach) rather
    # than the confirmation path with a placeholder NPI that would
    # mismatch against the real registry. In production, the
    # confirmation path runs when the credentialing system provides
    # an NPI at onboarding; the search path runs for legacy records
    # without a captured NPI.
    {
        "provider_id":   "provider-internal-02488",
        "first_name":    "Maria",
        "last_name":     "Hernandez",
        "credentials":   "DO",
        "specialty":     "pediatrics",
        "license_number": "DO-22119",
        "license_state":  "NY",
        "address_line_1": "789 Oak Avenue",
        "city":          "Brooklyn",
        "state":         "NY",
        "zip":           "11201",
        "phone":         "718-555-1212",
        "is_active":     True,
    },
]

if __name__ == "__main__":
    run_demo()