doczyai-pipelines/src/parent_child/qc.py

"""
Parent-Child Contract Mapping and Ranking System
"""

import json
import re
import sys
import time
import logging
import hashlib
import uuid
from collections import defaultdict
from typing import Dict, List, DefaultDict, Any, Optional, Tuple, Set, Callable
from datetime import datetime
from dataclasses import dataclass, field
from enum import Enum

import numpy as np
import pandas as pd

# ============================================================
# Logging Configuration
# ============================================================

logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Suppress AWS SDK logging
logging.getLogger("botocore").setLevel(logging.WARNING)
logging.getLogger("boto3").setLevel(logging.WARNING)


# ============================================================
# Constants & Enums
# ============================================================


class Strategy(Enum):
    """Supported strategies for parent/child operations."""

    KEYWORD = "keyword"
    FIELD_VALUE = "field_value"
    SOURCE_BOOLEAN = "source_boolean"
    SOURCE_LABEL = "source_label"
    SINGLE_FIELD = "single_field"
    MULTI_IDENTIFIER = "multi_identifier"


# Legal Entity Suffixes
LEGAL_SUFFIXES = [
    "inc",
    "inc.",
    "llc",
    "l.l.c.",
    "corp",
    "corp.",
    "co",
    "co.",
    "ltd",
    "ltd.",
    "company",
    "corporation",
    "limited",
    "dba",
    "d/b/a",
]
SUFFIX_REGEX = re.compile(
    r"\b(" + "|".join(re.escape(s) for s in LEGAL_SUFFIXES) + r")\b",
    flags=re.IGNORECASE,
)

# Keywords indicating child/amendment contracts
TITLE_KEYWORDS = [
    "amendment",
    "addendum",
    "exhibit",
    "attachment",
    "renewal",
    "extension",
    "modification",
    "letter",
    "rate letter",
    "notice",
    "notification",
]
TITLE_REGEX = re.compile(r"(?i)\b(" + "|".join(TITLE_KEYWORDS) + r")\b")

# CareSource-specific keywords
CARESOURCE_NON_PARENT_KEYWORDS = [
    "addendum",
    "amendment",
    "tracking sheet",
    "letter",
    "notice",
    "memo",
    "form",
    "term sheet",
    "request",
    "exhibit",
    "attachment",
    "renewal",
    "extension",
    "modification",
]

# BCBS-specific keywords
BCBS_NON_PARENT_KEYWORDS = [
    "addendum",
    "amendment",
    "tracking sheet",
    "letter",
    "notice",
    "memo",
    "form",
    "term sheet",
    "request",
    "exhibit",
    "attachment",
    "renewal",
    "extension",
    "modification",
]

# Default stopwords for text cleaning
DEFAULT_STOPWORDS = [
    "the",
    "a",
    "an",
    "and",
    "or",
    "for",
    "of",
    "to",
    "in",
    "on",
    "at",
    "by",
    "with",
    "agreement",
    "contract",
]

# Noise words for similarity calculations
NOISE_WORDS = {
    "the",
    "a",
    "an",
    "and",
    "or",
    "for",
    "of",
    "to",
    "in",
    "on",
    "at",
    "by",
    "with",
}


# ============================================================
# Utility Functions
# ============================================================


def resolve_column(
    df: pd.DataFrame, col_spec: Any, required: bool = False, label: str = ""
) -> Optional[str]:
    """Resolve column name from various specifications."""
    if col_spec is None:
        if required:
            raise ValueError(f"Missing required column spec for {label}")
        return None

    if isinstance(col_spec, (list, tuple)):
        for c in col_spec:
            if isinstance(c, str) and c in df.columns:
                return c
        if required:
            raise ValueError(
                f"None of candidate columns exist for {label}: {list(col_spec)}"
            )
        return None

    if isinstance(col_spec, str):
        if col_spec in df.columns:
            return col_spec
        if required:
            raise ValueError(f"Required column missing for {label}: {col_spec}")
        return None

    if required:
        raise ValueError(f"Invalid column spec for {label}: {col_spec}")
    return None


def to_string(series: pd.Series) -> pd.Series:
    """Convert series to string, filling NaN with empty string."""
    return series.fillna("").astype("string")


def to_datetime(series: pd.Series) -> pd.Series:
    """Convert series to datetime, coercing errors to NaT."""
    return pd.to_datetime(series, errors="coerce")


# ============================================================
# Text Processing
# ============================================================


class TextProcessor:
    """Handles text cleaning and normalization."""

    @staticmethod
    def clean_text(series: pd.Series) -> pd.Series:
        """Remove special chars, normalize whitespace."""
        text = to_string(series)
        text = text.str.replace(r"[^0-9A-Za-z\s]+", " ", regex=True)
        text = text.str.replace(r"\s+", " ", regex=True)
        text = text.str.strip()
        return text

    @staticmethod
    def clean_text_lowercase(series: pd.Series) -> pd.Series:
        """Clean text and convert to lowercase."""
        return TextProcessor.clean_text(series).str.lower()

    @staticmethod
    def normalize_single_identifier(value: str) -> str:
        """Normalize TIN/NPI identifier."""
        if not value:
            return ""
        normalized = re.sub(r"[^0-9A-Za-z]", "", value.strip())
        return normalized.upper()

    @staticmethod
    def _identifier_parts(val_str: str) -> List[str]:
        """Split a raw identifier value into individual IDs.

        Handles:
        - JSON lists: '["275440611", "274211365"]'  → ['275440611', '274211365']
        - Pipe-separated: '275440611|274211365'      → ['275440611', '274211365']
        - Plain scalars: '275440611'                 → ['275440611']

        JSON-list parsing keeps multi-TIN contracts readable in the grouping
        key — without it, `normalize_single_identifier` strips the brackets
        and commas and concatenates every digit into one blob.
        """
        if val_str.startswith("[") and val_str.endswith("]"):
            try:
                parsed = json.loads(val_str)
                if isinstance(parsed, list):
                    return [str(p) for p in parsed]
            except (json.JSONDecodeError, ValueError):
                pass
        if "|" in val_str:
            return val_str.split("|")
        return [val_str]

    @staticmethod
    def normalize_identifier(value: Any) -> str:
        """Normalize TIN/NPI identifiers.

        Accepts JSON-list, pipe-separated, or scalar input. Returns the
        unique IDs joined by `|`, sorted for deterministic bucketing.
        """
        if value is None or pd.isna(value):
            return ""

        val_str = str(value).strip()
        if not val_str:
            return ""

        parts = TextProcessor._identifier_parts(val_str)
        if len(parts) > 1:
            normalized = [TextProcessor.normalize_single_identifier(p) for p in parts]
            unique = sorted({p for p in normalized if p})
            return "|".join(unique) if unique else ""

        return TextProcessor.normalize_single_identifier(val_str)

    @staticmethod
    def normalize_identifier_series(series: pd.Series) -> pd.Series:
        """Apply identifier normalization to series."""
        return series.apply(TextProcessor.normalize_identifier).astype("string")

    @staticmethod
    def get_identifier_values(value: Any) -> Set[str]:
        """Extract all individual identifier values.

        Handles JSON-list, pipe-separated, and scalar input.
        """
        if value is None or pd.isna(value):
            return set()

        val_str = str(value).strip()
        if not val_str:
            return set()

        parts = TextProcessor._identifier_parts(val_str)
        if len(parts) > 1:
            normalized = [TextProcessor.normalize_single_identifier(p) for p in parts]
            return {p for p in normalized if p}

        normalized = TextProcessor.normalize_single_identifier(val_str)
        return {normalized} if normalized else set()

    @staticmethod
    def normalize_label(series: pd.Series) -> pd.Series:
        """Normalize parent/child/orphan labels."""
        text = series.astype("string").str.strip().str.lower()
        variations = {
            "parent": "Parent",
            "child": "Child",
            "orphan": "Orphan",
            "standalone": "StandAlone",
            "stand alone": "StandAlone",
            "stand_alone": "StandAlone",
            "stand-alone": "StandAlone",
        }
        text = text.replace(variations)
        text = text.replace("", pd.NA)
        return text

    @staticmethod
    def get_word_tokens(text: Any) -> Set[str]:
        """Extract and clean word tokens."""
        if text is None or pd.isna(text):
            return set()
        t = str(text).lower().strip()
        if not t:
            return set()
        t = re.sub(r"[^0-9a-z\s]+", " ", t)
        t = re.sub(r"\s+", " ", t).strip()
        if not t:
            return set()
        tokens = set(t.split(" "))
        return {x for x in tokens if x and x not in NOISE_WORDS}

    @staticmethod
    def calculate_text_similarity(text_a: Any, text_b: Any) -> float:
        """Calculate text similarity as overlap ratio."""
        tokens_a = TextProcessor.get_word_tokens(text_a)
        tokens_b = TextProcessor.get_word_tokens(text_b)
        if not tokens_a or not tokens_b:
            return 0.0
        overlap = len(tokens_a & tokens_b)
        total = max(len(tokens_a), len(tokens_b))
        return float(overlap) / float(total)

    @staticmethod
    def has_text_match(text_a: Any, text_b: Any) -> bool:
        """Check if one text contains or equals the other (case-insensitive)."""
        a = "" if (text_a is None or pd.isna(text_a)) else str(text_a)
        b = "" if (text_b is None or pd.isna(text_b)) else str(text_b)
        if not a or not b:
            return False
        a_lower = a.lower().strip()
        b_lower = b.lower().strip()
        return (a_lower in b_lower) or (b_lower in a_lower)


class TextCleaner:
    """Field-specific text cleaning with configurable rules."""

    def clean_field(self, field_name: str, series: pd.Series, rules: Dict) -> pd.Series:
        """Apply field-specific cleaning rules."""
        cleaned = TextProcessor.clean_text_lowercase(series)

        fields = rules.get("fields", {}) if isinstance(rules, dict) else {}
        field_rules = fields.get(field_name, {}) if isinstance(fields, dict) else {}

        # Remove stopwords
        if field_rules.get("remove_stopwords"):
            stopwords = set(field_rules.get("stopwords", []))
            if stopwords:

                def drop_stopwords(text: Any) -> str:
                    if text is None or pd.isna(text) or not str(text).strip():
                        return ""
                    words = [w for w in str(text).split() if w not in stopwords]
                    return " ".join(words)

                cleaned = cleaned.apply(drop_stopwords)

        # Remove suffixes
        if field_rules.get("remove_suffixes"):
            suffixes = field_rules.get("suffixes", [])
            if suffixes:
                suffix_pattern = (
                    r"\b(" + "|".join(re.escape(str(s)) for s in suffixes) + r")\b"
                )
                cleaned = cleaned.str.replace(suffix_pattern, "", regex=True)
                cleaned = cleaned.str.replace(r"\s+", " ", regex=True).str.strip()

        return cleaned


# ============================================================
# Field Derivation
# ============================================================


class FieldDeriver:
    """Derives new fields from source data."""

    def __init__(self):
        self.pattern_cache: Dict[str, re.Pattern] = {}

    def get_cached_regex(self, pattern: str) -> re.Pattern:
        """Get cached compiled regex."""
        if pattern not in self.pattern_cache:
            self.pattern_cache[pattern] = re.compile(pattern)
        return self.pattern_cache[pattern]

    def build_fields(
        self,
        df: pd.DataFrame,
        derive_config: List[Dict],
        text_cleaner: TextCleaner,
        cleaning_rules: Dict,
    ) -> pd.DataFrame:
        """Apply field derivation rules to dataframe."""
        if not derive_config:
            return df

        for rule in derive_config:
            target_col = rule.get("target")
            source_col = rule.get("source")
            rule_type = rule.get("mode")

            if (
                not target_col
                or not source_col
                or source_col not in df.columns
                or not rule_type
            ):
                continue

            source_data = df[source_col]

            if rule_type == "value_map":
                self._apply_value_map(df, target_col, source_data, rule)
            elif rule_type == "regex_map":
                self._apply_regex_map(df, target_col, source_data, rule)
            elif rule_type == "regex_contains_any":
                self._apply_regex_contains(df, target_col, source_data, rule)
            elif rule_type == "copy_clean":
                self._apply_copy_clean(
                    df, target_col, source_data, rule, text_cleaner, cleaning_rules
                )

        return df

    def _apply_value_map(
        self, df: pd.DataFrame, target_col: str, source_data: pd.Series, rule: Dict
    ) -> None:
        """Apply value mapping rule."""
        value_map = rule.get("map", {})
        default = rule.get("default", pd.NA)
        normalized = to_string(source_data).str.lower().str.strip()
        df[target_col] = (
            normalized.map(lambda x: value_map.get(str(x), default))
            .fillna(default)
            .astype("string")
        )

    def _apply_regex_map(
        self, df: pd.DataFrame, target_col: str, source_data: pd.Series, rule: Dict
    ) -> None:
        """Apply regex mapping rule."""
        patterns = rule.get("patterns", [])
        default = rule.get("default", pd.NA)
        text = to_string(source_data).str.lower()

        result = pd.Series([default] * len(df), index=df.index, dtype="string")
        for pattern_rule in patterns:
            pattern_str = pattern_rule.get("pattern")
            value = pattern_rule.get("value")
            if not pattern_str:
                continue
            compiled = self.get_cached_regex(pattern_str)
            matches = text.str.contains(compiled, na=False, regex=True)
            result.loc[matches & result.isna()] = value
        df[target_col] = result

    def _apply_regex_contains(
        self, df: pd.DataFrame, target_col: str, source_data: pd.Series, rule: Dict
    ) -> None:
        """Apply regex contains rule."""
        patterns = rule.get("patterns", [])
        if patterns:
            combined_pattern = "|".join(patterns)
            text = to_string(source_data).str.lower()
            df[target_col] = text.str.contains(combined_pattern, na=False, regex=True)
        else:
            df[target_col] = False

    def _apply_copy_clean(
        self,
        df: pd.DataFrame,
        target_col: str,
        source_data: pd.Series,
        rule: Dict,
        text_cleaner: TextCleaner,
        cleaning_rules: Dict,
    ) -> None:
        """Apply copy and clean rule."""
        clean_type = rule.get("cleaner", "lower_clean")
        if clean_type == "upper_clean":
            df[target_col] = TextProcessor.clean_text(
                to_string(source_data)
            ).str.upper()
        elif clean_type == "raw":
            df[target_col] = to_string(source_data)
        elif clean_type == "identifier":
            df[target_col] = TextProcessor.normalize_identifier_series(source_data)
        else:
            df[target_col] = text_cleaner.clean_field(
                target_col, to_string(source_data), cleaning_rules
            )


# ============================================================
# Grouping
# ============================================================


class GroupKeyBuilder:
    """Builds grouping keys to identify contract families."""

    def build_grouping_key(
        self, df: pd.DataFrame, config: Dict
    ) -> Tuple[pd.Series, pd.Series]:
        """Build grouping key and tier level."""
        strategy_type = config.get("strategy_type", "single_field")

        if strategy_type == "single_field":
            return self._build_single_field(df, config)
        elif strategy_type == "multi_identifier":
            return self._build_multi_identifier(df, config)
        else:
            raise ValueError(f"Unknown grouping strategy: {strategy_type}")

    def _build_single_field(
        self, df: pd.DataFrame, config: Dict
    ) -> Tuple[pd.Series, pd.Series]:
        """Build grouping key from single field."""
        field = config.get("field")
        if not field or field not in df.columns:
            raise ValueError(f"Grouping field missing or not found: {field}")
        grouping_key = df[field].astype("string")
        tier = pd.Series([1] * len(df), index=df.index, dtype=np.int8)
        return grouping_key, tier

    def _build_multi_identifier(
        self, df: pd.DataFrame, config: Dict
    ) -> Tuple[pd.Series, pd.Series]:
        """Build grouping key from multiple identifiers."""
        identifiers = config.get("identifiers", [])
        if not identifiers or not isinstance(identifiers, list):
            raise ValueError("multi_identifier requires list 'identifiers'")

        # Ensure columns exist
        for col in identifiers:
            if col not in df.columns:
                df[col] = pd.Series([pd.NA] * len(df), index=df.index, dtype="string")

        # Count identifiers present for each row
        present = [
            df[col].fillna("").astype("string").ne("").to_numpy() for col in identifiers
        ]
        id_count = np.zeros(len(df), dtype=np.int8)
        for has_value in present:
            id_count += has_value.astype(np.int8)

        # Assign tier based on identifier count
        tier = np.zeros(len(df), dtype=np.int8)
        tier[id_count == len(identifiers)] = 1
        tier[(id_count >= 2) & (tier == 0)] = 2
        tier[(id_count == 1) & (tier == 0)] = 3

        # Build keys for each tier
        grouping_key = self._build_tier_keys(df, identifiers, tier)

        return grouping_key, pd.Series(tier, index=df.index, dtype=np.int8)

    def _build_tier_keys(
        self, df: pd.DataFrame, identifiers: List[str], tier: np.ndarray
    ) -> pd.Series:
        """Build keys for each tier level."""
        grouping_key = pd.Series([pd.NA] * len(df), index=df.index, dtype="string")

        # Tier 1: all identifiers
        mask_tier1 = tier == 1
        if mask_tier1.any():
            parts = [
                col + ":" + df.loc[mask_tier1, col].astype("string")
                for col in identifiers
            ]
            grouping_key.loc[mask_tier1] = (
                parts[0] if len(parts) == 1 else parts[0].str.cat(parts[1:], sep="|")
            )

        # Tier 2: first 2 available identifiers
        mask_tier2 = tier == 2
        if mask_tier2.any():
            keys = self._build_tier_keys_for_rows(
                df[mask_tier2], identifiers, max_count=2
            )
            grouping_key.loc[mask_tier2] = keys

        # Tier 3: first available identifier
        mask_tier3 = tier == 3
        if mask_tier3.any():
            keys = self._build_tier_keys_for_rows(
                df[mask_tier3], identifiers, max_count=1
            )
            grouping_key.loc[mask_tier3] = keys

        return grouping_key

    @staticmethod
    def _build_tier_keys_for_rows(
        df_subset: pd.DataFrame, identifiers: List[str], max_count: int
    ) -> List[str]:
        """Build keys for a subset of rows with limited identifier count."""
        keys = []
        for idx in df_subset.index:
            picked = []
            for col in identifiers:
                val = df_subset.at[idx, col]
                val_str = "" if (val is None or pd.isna(val)) else str(val)
                if val_str:
                    picked.append(f"{col}:{val_str}")
                    if len(picked) == max_count:
                        break
            keys.append("|".join(picked) if picked else pd.NA)
        return keys


# ============================================================
# Parent Identification
# ============================================================


class ParentIdentifier:
    """Identifies parent contracts using various strategies."""

    def __init__(self):
        self.pattern_cache: Dict[str, re.Pattern] = {}

    def find_parents(self, df: pd.DataFrame, config: Dict) -> pd.Series:
        """Identify parent contracts."""
        strategy_type = config.get("strategy_type", "keyword")

        if strategy_type == "keyword":
            return self._identify_by_keyword(df, config)
        elif strategy_type == "field_value":
            return self._identify_by_field_value(df, config)
        elif strategy_type == "source_boolean":
            return self._identify_by_boolean(df, config)
        elif strategy_type == "source_label":
            return self._identify_by_label(df, config)
        elif strategy_type == "bcbs_custom":
            return self._identify_bcbs_custom(df, config)
        else:
            raise ValueError(f"Unknown parent identification strategy: {strategy_type}")

    def _identify_by_keyword(self, df: pd.DataFrame, config: Dict) -> pd.Series:
        """Identify parents by excluding keywords.

        `field` (typically the cleaned contract title) is the primary signal.
        `fallback_field` (typically the filename) is consulted only when the
        primary field is blank/missing for a row.
        """
        field = config.get("field")
        exclude_keywords = config.get("exclude_keywords", [])
        fallback_field = config.get("fallback_field")
        if not field or field not in df.columns:
            raise ValueError(f"keyword strategy missing field: {field}")
        if not exclude_keywords:
            return pd.Series([True] * len(df), index=df.index)

        pattern_str = (
            r"\b(?:" + "|".join(re.escape(str(k)) for k in exclude_keywords) + r")\b"
        )
        if pattern_str not in self.pattern_cache:
            self.pattern_cache[pattern_str] = re.compile(pattern_str, re.IGNORECASE)

        compiled = self.pattern_cache[pattern_str]
        primary = df[field].astype("string").fillna("").str.strip()
        has_keyword = primary.str.contains(compiled, na=False, regex=True)

        if fallback_field and fallback_field in df.columns:
            primary_missing = (primary == "") | (primary.str.lower() == "nan")
            # Normalize the fallback (typically FILE_NAME) so regex word
            # boundaries land cleanly. Digits, underscores and punctuation
            # are regex word chars, so `\bamd\b` won't match "AMD1" or
            # "_AMD#" without this preprocessing step.
            fallback_has_kw = (
                df[fallback_field]
                .astype("string")
                .fillna("")
                .str.replace(r"[^a-zA-Z]+", " ", regex=True)
                .str.contains(compiled, na=False, regex=True)
            )
            has_keyword = has_keyword | (primary_missing & fallback_has_kw)

        return ~has_keyword

    def _identify_by_field_value(self, df: pd.DataFrame, config: Dict) -> pd.Series:
        """Identify parents by field value match."""
        field = config.get("field")
        target_value = config.get("target_value")
        if not field or field not in df.columns:
            raise ValueError(f"field_value strategy missing field: {field}")
        return df[field].astype("string") == str(target_value)

    def _identify_by_boolean(self, df: pd.DataFrame, config: Dict) -> pd.Series:
        """Identify parents from boolean field."""
        field = config.get("field")
        if not field or field not in df.columns:
            raise ValueError(f"source_boolean strategy missing field: {field}")
        return df[field].fillna(False).astype(bool)

    def _identify_by_label(self, df: pd.DataFrame, config: Dict) -> pd.Series:
        """Identify parents from label field."""
        field = config.get("field")
        if not field or field not in df.columns:
            raise ValueError(f"source_label strategy missing field: {field}")
        labels = TextProcessor.normalize_label(df[field])
        return (labels == "PARENT").fillna(False)

    def _identify_bcbs_custom(self, df: pd.DataFrame, config: Dict) -> pd.Series:
        """
        BCBS-specific parent identification matching original bcbs_parent_child_mapping.py logic.

        Rules:
        1. effective_date_rank == 1 AND title is not null -> Parent
        2. "agreement" in title but NOT "amendment/addendum" -> Parent
        3. LOB-based: has LOB AND title not null AND no "amendment/addendum" -> Parent
        4. Exclude if title/filename contains "amendment" or "addendum"
        """
        title_col = config.get("field", "title_clean")
        filename_col = config.get("filename_field", "id_clean")
        grouping_key_col = "grouping_key"
        eff_date_col = "eff_date"
        lob_col = config.get("lob_field", "_lob")

        # Initialize all as non-parent
        is_parent = pd.Series([False] * len(df), index=df.index)

        # Get title and ensure it's string
        title = (
            df[title_col].fillna("").astype(str).str.lower()
            if title_col in df.columns
            else pd.Series([""] * len(df), index=df.index)
        )
        filename = (
            df[filename_col].fillna("").astype(str).str.lower()
            if filename_col in df.columns
            else pd.Series([""] * len(df), index=df.index)
        )

        # Check for amendment/addendum in title or filename
        amend_pattern = r"\b(amendment|addendum)\b"
        has_amendment_title = title.str.contains(amend_pattern, na=False, regex=True)
        has_amendment_filename = filename.str.contains(
            amend_pattern, na=False, regex=True
        )
        has_amendment = has_amendment_title | has_amendment_filename

        # Check for "agreement" in title
        has_agreement = title.str.contains(r"\bagreement\b", na=False, regex=True)

        # Check for non-empty title
        has_title = title.str.strip() != ""

        # Compute effective_date_rank within each group
        if grouping_key_col in df.columns and eff_date_col in df.columns:
            eff_date_rank = df.groupby(grouping_key_col)[eff_date_col].rank(
                method="dense", ascending=True
            )
        else:
            eff_date_rank = pd.Series([1] * len(df), index=df.index)

        # Rule 1: effective_date_rank == 1 AND title not null -> Parent
        case1 = (eff_date_rank == 1) & has_title
        is_parent = is_parent | case1

        # Rule 2: "agreement" in title but NOT "amendment/addendum" AND title not null -> Parent
        case2 = has_agreement & ~has_amendment & has_title
        is_parent = is_parent | case2

        # Rule 3: LOB-based - has LOB AND title not null AND no "amendment/addendum" -> Parent
        if lob_col in df.columns:
            has_lob = df[lob_col].fillna("").astype(str).str.strip() != ""
            case3 = has_lob & has_title & ~has_amendment
            is_parent = is_parent | case3

        # Final exclusion: if amendment/addendum is in title OR filename -> NOT Parent
        is_parent = is_parent & ~has_amendment

        return is_parent


# ============================================================
# Child Assignment (continues in next message due to length)
# ============================================================


class ChildAssigner:
    """Multi-pass child-to-parent assignment."""

    def build_parent_index(
        self, df: pd.DataFrame, index_fields: List[str]
    ) -> Dict[str, DefaultDict[str, List[int]]]:
        """Build parent lookup index by identifier fields."""
        maps: Dict[str, DefaultDict[str, List[int]]] = {
            f: defaultdict(list) for f in index_fields
        }
        parents = df[df["is_parent"]]

        for f in index_fields:
            if f not in df.columns:
                continue

            for idx in parents.index:
                val = parents.at[idx, f]
                if val is None or pd.isna(val):
                    continue

                individual_values = TextProcessor.get_identifier_values(val)
                for individual_val in individual_values:
                    if individual_val:
                        maps[f][individual_val].append(idx)

        return maps

    def find_candidates(
        self,
        parent_index: Dict[str, DefaultDict[str, List[int]]],
        child_row: pd.Series,
        index_fields: List[str],
    ) -> List[int]:
        """Find candidate parents for a child."""
        cand: Set[int] = set()
        for f in index_fields:
            if f not in child_row.index:
                continue
            v = child_row[f]
            if v is None or pd.isna(v):
                continue

            child_values = TextProcessor.get_identifier_values(v)
            for individual_val in child_values:
                if individual_val:
                    cand.update(parent_index[f].get(individual_val, []))

        return list(cand)

    @staticmethod
    def get_field_value(row: pd.Series, col: str) -> Any:
        """Safely get field value from row."""
        if not col:
            return None
        return row.get(col, None)

    @staticmethod
    def is_date_valid(parent_date: Any, child_date: Any) -> bool:
        """Check if parent effective date is before or equal to child date."""
        parent_valid = pd.notna(parent_date)
        child_valid = pd.notna(child_date)

        if parent_valid and child_valid:
            return parent_date <= child_date
        elif not parent_valid and child_valid:
            return False
        else:
            return True

    def check_constraints(
        self,
        row: pd.Series,
        parent: Dict[str, Any],
        pass_config: Dict,
        constraints: Dict,
    ) -> bool:
        """Verify assignment constraints between child and parent."""
        col_state = constraints.get("state")
        col_payer = constraints.get("payer")
        col_commodity = constraints.get("commodity")
        col_eff = constraints.get("date")
        col_lob = constraints.get("lob")
        col_provider_type = constraints.get("provider_type")

        if pass_config.get("state") and col_state:
            val_a = self.get_field_value(row, col_state)
            val_b = parent.get(col_state)
            if (
                pd.notna(val_a)
                and pd.notna(val_b)
                and str(val_a)
                and str(val_b)
                and str(val_a) != str(val_b)
            ):
                return False

        if pass_config.get("payer") and col_payer:
            val_a = self.get_field_value(row, col_payer)
            val_b = parent.get(col_payer)
            if (
                pd.notna(val_a)
                and pd.notna(val_b)
                and str(val_a)
                and str(val_b)
                and str(val_a) != str(val_b)
            ):
                return False

        if pass_config.get("commodity") and col_commodity:
            val_a = self.get_field_value(row, col_commodity)
            val_b = parent.get(col_commodity)
            if (
                pd.notna(val_a)
                and pd.notna(val_b)
                and str(val_a)
                and str(val_b)
                and str(val_a) != str(val_b)
            ):
                return False

        # LOB matching (for BCBS: extracted_lob must match)
        if pass_config.get("lob") and col_lob:
            val_a = self.get_field_value(row, col_lob)
            val_b = parent.get(col_lob)
            if (
                pd.notna(val_a)
                and pd.notna(val_b)
                and str(val_a)
                and str(val_b)
                and str(val_a) != str(val_b)
            ):
                return False

        # Provider Type matching (for BCBS: consolidated_provider_type must match)
        if pass_config.get("provider_type") and col_provider_type:
            val_a = self.get_field_value(row, col_provider_type)
            val_b = parent.get(col_provider_type)
            if (
                pd.notna(val_a)
                and pd.notna(val_b)
                and str(val_a)
                and str(val_b)
                and str(val_a) != str(val_b)
            ):
                return False

        if pass_config.get("date") and col_eff:
            if not self.is_date_valid(
                parent.get(col_eff), self.get_field_value(row, col_eff)
            ):
                return False

        return True

    def assign_children(
        self, df: pd.DataFrame, config: Dict, parent_cache: Dict[int, Dict[str, Any]]
    ) -> pd.DataFrame:
        """Execute multi-pass child-to-parent assignment."""
        # Ensure columns exist
        if "assigned_parent_idx" not in df.columns:
            df["assigned_parent_idx"] = pd.Series(
                [pd.NA] * len(df), index=df.index, dtype="Int64"
            )
        if "assignment_pass" not in df.columns:
            df["assignment_pass"] = pd.Series(
                [pd.NA] * len(df), index=df.index, dtype="Int8"
            )
        if "assignment_reasoning" not in df.columns:
            df["assignment_reasoning"] = pd.Series(
                [""] * len(df), index=df.index, dtype="string"
            )

        rules = config.get("assignment_rules", {}) or {}
        pass1 = rules.get("pass1", {}) or {}
        pass2 = rules.get("pass2", {}) or {}
        pass3 = rules.get("pass3", {}) or {}

        index_fields = [
            f
            for f in (config.get("candidate_index_fields", []) or [])
            if isinstance(f, str) and f
        ]
        constraints = config.get("assignment_constraints", {}) or {}

        col_title = constraints.get("title")
        col_eff = constraints.get("date")

        parent_index = self.build_parent_index(df, index_fields)
        children_idx = df.index[(~df["is_parent"]) & (df["processable"])].tolist()

        if not children_idx:
            return df

        # Pass 1: Strict constraints + title matching
        self._assign_pass1(
            df,
            children_idx,
            parent_index,
            parent_cache,
            col_title,
            col_eff,
            pass1,
            constraints,
        )

        # Pass 2: Title-based matching
        remaining = df.index[
            (~df["is_parent"]) & (df["processable"]) & (df["assignment_pass"].isna())
        ].tolist()
        if remaining and pass2:
            self._assign_pass2(
                df,
                remaining,
                parent_index,
                parent_cache,
                col_title,
                col_eff,
                pass2,
                constraints,
            )

        # Pass 3: Fallback (unique parent, unique payer)
        remaining = df.index[
            (~df["is_parent"]) & (df["processable"]) & (df["assignment_pass"].isna())
        ].tolist()
        if remaining and pass3:
            self._assign_pass3(
                df, remaining, parent_index, parent_cache, col_eff, pass3, constraints
            )

        return df

    def _assign_pass1(
        self,
        df: pd.DataFrame,
        children_idx: List[int],
        parent_index: Dict,
        parent_cache: Dict,
        col_title: str,
        col_eff: str,
        pass1: Dict,
        constraints: Dict,
    ) -> None:
        """Pass 1: Strict constraints + title matching."""
        for cidx in children_idx:
            row = df.loc[cidx]
            cand = self.find_candidates(
                parent_index, row, [f for f in parent_index.keys() if f]
            )
            if not cand:
                continue

            # Filter by strict constraints
            filtered = [
                pidx
                for pidx in cand
                if self.check_constraints(
                    row, parent_cache.get(pidx, {}), pass1, constraints
                )
            ]

            if not filtered:
                continue

            # Title matching if enabled
            if pass1.get("title") and col_title:
                self._apply_title_priority(
                    df, cidx, filtered, parent_cache, col_title, pass_num=1
                )
            else:
                # No title matching, use highest rank
                best = max(
                    filtered,
                    key=lambda x: int(parent_cache.get(x, {}).get("parent_rank", 0)),
                )
                best_rank = int(parent_cache.get(best, {}).get("parent_rank", 0))
                df.at[cidx, "assigned_parent_idx"] = best
                df.at[cidx, "assignment_pass"] = 1
                df.at[cidx, "assignment_reasoning"] = (
                    f"Pass 1: strict constraints (rank={best_rank})"
                )

    def _assign_pass2(
        self,
        df: pd.DataFrame,
        remaining: List[int],
        parent_index: Dict,
        parent_cache: Dict,
        col_title: str,
        col_eff: str,
        pass2: Dict,
        constraints: Dict,
    ) -> None:
        """Pass 2: Title-based matching."""
        col_state = constraints.get("state")
        col_payer = constraints.get("payer")

        for cidx in remaining:
            row = df.loc[cidx]
            cand = self.find_candidates(
                parent_index, row, [f for f in parent_index.keys() if f]
            )
            if not cand:
                continue

            child_eff = row.get(col_eff, pd.NaT) if col_eff else pd.NaT
            child_title = row.get(col_title, "") if col_title else ""

            substring_matches = []
            similarity_matches = []

            for pidx in cand:
                p = parent_cache.get(pidx)
                if not p:
                    continue

                # Date constraint
                if pass2.get("date") and col_eff and pd.notna(child_eff):
                    parent_eff = p.get(col_eff, pd.NaT)
                    if pd.notna(parent_eff) and parent_eff > child_eff:
                        continue

                # Title matching
                if pass2.get("title") and col_title:
                    parent_title = p.get(col_title, "")
                    is_substring = TextProcessor.has_text_match(
                        parent_title, child_title
                    )
                    similarity = TextProcessor.calculate_text_similarity(
                        parent_title, child_title
                    )
                    parent_rank = int(p.get("parent_rank", 0))

                    if is_substring:
                        substring_matches.append((parent_rank, similarity, pidx))
                    else:
                        threshold = float(pass2.get("title_similarity_min", 0.50))
                        if similarity >= threshold:
                            similarity_matches.append((similarity, parent_rank, pidx))

            if substring_matches:
                substring_matches.sort(reverse=True)
                best_rank, best_sim, best_idx = substring_matches[0]
                df.at[cidx, "assigned_parent_idx"] = best_idx
                df.at[cidx, "assignment_pass"] = 2
                df.at[cidx, "assignment_reasoning"] = (
                    f"Pass 2: title substring match (rank={best_rank})"
                )
            elif similarity_matches:
                similarity_matches.sort(reverse=True)
                best_sim, best_rank, best_idx = similarity_matches[0]
                df.at[cidx, "assigned_parent_idx"] = best_idx
                df.at[cidx, "assignment_pass"] = 2
                df.at[cidx, "assignment_reasoning"] = (
                    f"Pass 2: title similarity (sim={best_sim:.2f},rank={best_rank})"
                )

    def _assign_pass3(
        self,
        df: pd.DataFrame,
        remaining: List[int],
        parent_index: Dict,
        parent_cache: Dict,
        col_eff: str,
        pass3: Dict,
        constraints: Dict,
    ) -> None:
        """Pass 3: Fallback (unique parent, unique payer)."""
        col_payer = constraints.get("payer")

        for cidx in remaining:
            row = df.loc[cidx]
            cand = self.find_candidates(
                parent_index, row, [f for f in parent_index.keys() if f]
            )
            if not cand:
                continue

            child_eff = row.get(col_eff, pd.NaT) if col_eff else pd.NaT

            # Filter by date, fallback to all candidates if none are date-valid
            date_valid_cand = [
                pidx
                for pidx in cand
                if self.is_date_valid(
                    parent_cache.get(pidx, {}).get(col_eff, pd.NaT), child_eff
                )
            ]

            if not date_valid_cand:
                # Fallback: use all candidates when date filtering eliminates
                # everyone — the child may predate the parent (e.g., an older
                # amendment linked to a newer base agreement)
                date_valid_cand = cand

            # Unique parent
            if pass3.get("unique_parent") and len(date_valid_cand) == 1:
                df.at[cidx, "assigned_parent_idx"] = date_valid_cand[0]
                df.at[cidx, "assignment_pass"] = 3
                df.at[cidx, "assignment_reasoning"] = (
                    "Pass 3: unique parent (date valid)"
                )
                continue

            # Unique payer
            if pass3.get("unique_payer") and col_payer:
                child_payer = row.get(col_payer, "")
                child_payer = (
                    ""
                    if (child_payer is None or pd.isna(child_payer))
                    else str(child_payer)
                )
                if child_payer:
                    payer_matches = [
                        pidx
                        for pidx in date_valid_cand
                        if str(parent_cache.get(pidx, {}).get(col_payer, ""))
                        == child_payer
                    ]
                    if len(payer_matches) == 1:
                        df.at[cidx, "assigned_parent_idx"] = payer_matches[0]
                        df.at[cidx, "assignment_pass"] = 3
                        df.at[cidx, "assignment_reasoning"] = (
                            "Pass 3: unique payer (date valid)"
                        )

    @staticmethod
    def _apply_title_priority(
        df: pd.DataFrame,
        cidx: int,
        filtered: List[int],
        parent_cache: Dict,
        col_title: str,
        pass_num: int,
    ) -> None:
        """Apply priority logic: substring matches before similarity."""
        child_title = df.loc[cidx, col_title] if col_title in df.columns else ""

        substring_matches = []
        non_substring = []

        for pidx in filtered:
            parent_title = parent_cache.get(pidx, {}).get(col_title, "")
            parent_rank = int(parent_cache.get(pidx, {}).get("parent_rank", 0))
            is_substring = TextProcessor.has_text_match(parent_title, child_title)
            similarity = TextProcessor.calculate_text_similarity(
                parent_title, child_title
            )

            if is_substring:
                substring_matches.append((parent_rank, similarity, pidx))
            else:
                non_substring.append((parent_rank, similarity, pidx))

        if substring_matches:
            substring_matches.sort(reverse=True)
            best_rank, best_sim, best_idx = substring_matches[0]
            df.at[cidx, "assigned_parent_idx"] = best_idx
            df.at[cidx, "assignment_pass"] = pass_num
            df.at[cidx, "assignment_reasoning"] = (
                f"Pass {pass_num}: title substring (rank={best_rank})"
            )
        elif non_substring:
            non_substring.sort(reverse=True)
            best_rank, best_sim, best_idx = non_substring[0]
            df.at[cidx, "assigned_parent_idx"] = best_idx
            df.at[cidx, "assignment_pass"] = pass_num
            df.at[cidx, "assignment_reasoning"] = (
                f"Pass {pass_num}: rank only (rank={best_rank})"
            )


# ============================================================
# BCBS-Specific Child Assignment (5 Rules)
# ============================================================


class BCBSChildAssigner:
    """BCBS-specific child-to-parent assignment implementing 5 rules from original logic."""

    @staticmethod
    def is_missing_lob(lob_value: Any) -> bool:
        """Check if LOB value is missing or empty."""
        if lob_value is None or pd.isna(lob_value):
            return True
        return str(lob_value).strip() == ""

    @staticmethod
    def get_str(value: Any) -> str:
        """Safely convert value to lowercase string."""
        if value is None or pd.isna(value):
            return ""
        return str(value).lower().strip()

    def assign_children_bcbs(self, df: pd.DataFrame, config: Dict) -> pd.DataFrame:
        """
        BCBS-specific child assignment implementing 5 rules:
        Rule 0: Professional logic (title substring + keyword)
        Rule 1: LOB + Provider Type exact match
        Rule 2: Title substring match (for missing LOB)
        Rule 3: Provider Type Count fallback
        Rule 4: Commercial LOB fallback
        Rule 5: Latest parent fallback
        """
        # Ensure columns exist
        if "assigned_parent_idx" not in df.columns:
            df["assigned_parent_idx"] = pd.Series(
                [pd.NA] * len(df), index=df.index, dtype="Int64"
            )
        if "assignment_pass" not in df.columns:
            df["assignment_pass"] = pd.Series(
                [pd.NA] * len(df), index=df.index, dtype="Int8"
            )
        if "assignment_reasoning" not in df.columns:
            df["assignment_reasoning"] = pd.Series(
                [""] * len(df), index=df.index, dtype="string"
            )

        constraints = config.get("assignment_constraints", {}) or {}
        col_title = constraints.get("title", "title_clean")
        col_eff = constraints.get("date", "eff_date")
        col_lob = constraints.get("lob", "_lob")
        col_provider_type = constraints.get("provider_type", "_provider_type")

        # Build parent index for cross-tier candidate lookup
        index_fields = [
            f
            for f in (config.get("candidate_index_fields", []) or [])
            if isinstance(f, str) and f
        ]
        child_assigner = ChildAssigner()
        parent_index = child_assigner.build_parent_index(df, index_fields)

        all_parents = df[df["is_parent"]].copy()
        all_children = df[~df["is_parent"] & df["processable"]].copy()

        if all_parents.empty or all_children.empty:
            return df

        # Build parent cache for all parents
        parent_cache = {}
        for pidx in all_parents.index:
            parent_cache[pidx] = {
                "title": (
                    self.get_str(all_parents.at[pidx, col_title])
                    if col_title in all_parents.columns
                    else ""
                ),
                "eff_date": (
                    all_parents.at[pidx, col_eff]
                    if col_eff in all_parents.columns
                    else pd.NaT
                ),
                "lob": (
                    self.get_str(all_parents.at[pidx, col_lob])
                    if col_lob in all_parents.columns
                    else ""
                ),
                "provider_type": (
                    self.get_str(all_parents.at[pidx, col_provider_type])
                    if col_provider_type in all_parents.columns
                    else ""
                ),
                "parent_rank": (
                    int(all_parents.at[pidx, "parent_rank"])
                    if "parent_rank" in all_parents.columns
                    else 0
                ),
            }

        # Process each child — find candidates via index lookup (cross-tier)
        for cidx in all_children.index:
            child_row = df.loc[cidx]
            child_title = (
                self.get_str(df.at[cidx, col_title]) if col_title in df.columns else ""
            )
            child_eff = df.at[cidx, col_eff] if col_eff in df.columns else pd.NaT
            child_lob = (
                self.get_str(df.at[cidx, col_lob]) if col_lob in df.columns else ""
            )
            child_ptype = (
                self.get_str(df.at[cidx, col_provider_type])
                if col_provider_type in df.columns
                else ""
            )

            candidate_idx = None
            rule_used = ""

            # Find eligible parents via cross-tier candidate lookup
            cand_indices = child_assigner.find_candidates(
                parent_index, child_row, index_fields
            )
            eligible_parents = [p for p in cand_indices if p in parent_cache]

            if not eligible_parents:
                continue

            # Rule 0: Professional logic
            if "professional" in child_title and candidate_idx is None:
                for pidx in sorted(
                    eligible_parents,
                    key=lambda x: parent_cache[x]["eff_date"] or pd.Timestamp.min,
                    reverse=True,
                ):
                    p_title = parent_cache[pidx]["title"]
                    if p_title and len(p_title) > 0 and p_title in child_title:
                        candidate_idx = pidx
                        rule_used = "Rule 0a: Professional subset match"
                        break

                if candidate_idx is None:
                    for pidx in sorted(
                        eligible_parents,
                        key=lambda x: parent_cache[x]["eff_date"] or pd.Timestamp.min,
                        reverse=True,
                    ):
                        if "professional" in parent_cache[pidx]["title"]:
                            candidate_idx = pidx
                            rule_used = "Rule 0b: Professional keyword"
                            break

            # Rule 1: LOB + Provider Type match
            if (
                candidate_idx is None
                and not self.is_missing_lob(child_lob)
                and child_ptype
            ):
                for pidx in sorted(
                    eligible_parents,
                    key=lambda x: parent_cache[x]["eff_date"] or pd.Timestamp.min,
                    reverse=True,
                ):
                    p = parent_cache[pidx]
                    if (
                        p["lob"] == child_lob
                        and p["provider_type"]
                        and p["provider_type"] == child_ptype
                    ):
                        candidate_idx = pidx
                        rule_used = "Rule 1: LOB + Provider Type match"
                        break

            # Rule 2: Title substring match (for missing LOB)
            if candidate_idx is None and self.is_missing_lob(child_lob):
                for pidx in sorted(
                    eligible_parents,
                    key=lambda x: parent_cache[x]["eff_date"] or pd.Timestamp.min,
                    reverse=True,
                ):
                    p_title = parent_cache[pidx]["title"]
                    if p_title and len(p_title) > 0 and p_title in child_title:
                        candidate_idx = pidx
                        rule_used = "Rule 2: Title substring (missing LOB)"
                        break

            # Rule 3: Provider Type logic
            if candidate_idx is None:
                if child_ptype:
                    for pidx in sorted(
                        eligible_parents,
                        key=lambda x: parent_cache[x]["eff_date"] or pd.Timestamp.min,
                        reverse=True,
                    ):
                        p = parent_cache[pidx]
                        if p["lob"] == child_lob and p["provider_type"] == child_ptype:
                            candidate_idx = pidx
                            rule_used = "Rule 3a: Provider Type match"
                            break
                else:
                    lob_matches = [
                        pidx
                        for pidx in eligible_parents
                        if parent_cache[pidx]["lob"] == child_lob
                    ]
                    if lob_matches:
                        facility_matches = [
                            pidx
                            for pidx in lob_matches
                            if "facility" in parent_cache[pidx]["provider_type"]
                        ]
                        if facility_matches:
                            candidate_idx = sorted(
                                facility_matches,
                                key=lambda x: parent_cache[x]["eff_date"]
                                or pd.Timestamp.min,
                                reverse=True,
                            )[0]
                            rule_used = "Rule 3b: Provider Type Count (facility)"
                        else:
                            candidate_idx = sorted(
                                lob_matches,
                                key=lambda x: parent_cache[x]["eff_date"]
                                or pd.Timestamp.min,
                                reverse=True,
                            )[0]
                            rule_used = "Rule 3b: Provider Type Count fallback"

            # Rule 4: Commercial LOB fallback
            if candidate_idx is None:
                comm_matches = [
                    pidx
                    for pidx in eligible_parents
                    if "comm" in parent_cache[pidx]["lob"]
                ]
                if comm_matches:
                    candidate_idx = sorted(
                        comm_matches,
                        key=lambda x: parent_cache[x]["eff_date"] or pd.Timestamp.min,
                        reverse=True,
                    )[0]
                    rule_used = "Rule 4: Commercial fallback"

            # Rule 5: Latest parent fallback
            if candidate_idx is None and eligible_parents:
                candidate_idx = sorted(
                    eligible_parents,
                    key=lambda x: parent_cache[x]["eff_date"] or pd.Timestamp.min,
                    reverse=True,
                )[0]
                rule_used = "Rule 5: Latest parent fallback"

            # Assign the candidate
            if candidate_idx is not None:
                p_rank = parent_cache[candidate_idx]["parent_rank"]
                df.at[cidx, "assigned_parent_idx"] = candidate_idx
                df.at[cidx, "assignment_pass"] = 1
                df.at[cidx, "assignment_reasoning"] = f"{rule_used} (rank={p_rank})"

        return df


# ============================================================
# Configuration Management
# ============================================================


@dataclass
class ClientConfig:
    """Client configuration factory."""

    name: str
    engine_type: str
    columns: Dict[str, Any]
    prep_fields: Dict[str, Dict] = field(default_factory=dict)
    cleaning_rules: Dict = field(default_factory=dict)
    derive_fields: List[Dict] = field(default_factory=list)
    grouping: Dict = field(default_factory=dict)
    parent_identification: Dict = field(default_factory=dict)
    candidate_index_fields: List[str] = field(default_factory=list)
    assignment_constraints: Dict = field(default_factory=dict)
    assignment_rules: Dict = field(default_factory=dict)
    validation: Dict = field(default_factory=dict)
    rank_output: Dict = field(default_factory=dict)
    qa_flags: Dict = field(default_factory=dict)
    keep_work_cols: bool = False

    def to_dict(self) -> Dict:
        """Convert to dictionary for engine compatibility."""
        return {
            "engine_type": self.engine_type,
            "columns": self.columns,
            "prep_fields": self.prep_fields,
            "cleaning_rules": self.cleaning_rules,
            "derive_fields": self.derive_fields,
            "grouping": self.grouping,
            "parent_identification": self.parent_identification,
            "candidate_index_fields": self.candidate_index_fields,
            "assignment_constraints": self.assignment_constraints,
            "assignment_rules": self.assignment_rules,
            "validation": self.validation,
            "rank_output": self.rank_output,
            "qa_flags": self.qa_flags,
            "keep_work_cols": self.keep_work_cols,
        }


class ConfigFactory:
    """Factory for creating client configurations."""

    @staticmethod
    def create_molina() -> ClientConfig:
        """Create Molina configuration."""
        return ClientConfig(
            name="molina",
            engine_type="parent_child",
            columns={
                "id": "FILE_NAME",
                "title": "contract_title_cleaned",
                "effective_date": "fixed_effective_date",
            },
            prep_fields={
                "_tin": {
                    "source": "PROV_GROUP_TIN",
                    "mode": "copy_clean",
                    "cleaner": "identifier",
                },
                "_npi": {
                    "source": "PROV_GROUP_NPI",
                    "mode": "copy_clean",
                    "cleaner": "identifier",
                },
                "_prov_name": {
                    "source": "PROV_GROUP_NAME_FULL_cleaned",
                    "mode": "copy_clean",
                    "cleaner": "upper_clean",
                },
                "_payer_clean": {
                    "source": "payer_name_cleaned",
                    "mode": "copy_clean",
                    "cleaner": "lower_clean",
                },
                "_state_clean": {
                    "source": "PAYER_STATE",
                    "mode": "copy_clean",
                    "cleaner": "upper_clean",
                },
            },
            cleaning_rules={
                "fields": {
                    "title": {
                        "remove_stopwords": True,
                        "stopwords": DEFAULT_STOPWORDS,
                    },
                }
            },
            derive_fields=[
                {
                    "target": "_prov_name_clean",
                    "source": "_prov_name",
                    "mode": "copy_clean",
                    "cleaner": "upper_clean",
                },
            ],
            grouping={
                "strategy_type": "multi_identifier",
                "identifiers": ["_tin", "_npi", "_prov_name_clean"],
            },
            parent_identification={
                "strategy_type": "keyword",
                "field": "title_clean",
                "fallback_field": "FILE_NAME",
                "exclude_keywords": [
                    "exhibit",
                    "amendment",
                    "amend",
                    "amd",
                    "addendum",
                    "adden",
                    "renewal",
                    "extension",
                    "modification",
                    "attachment",
                    "agenda",
                    "letter",
                    "rate letter",
                    "notice",
                    "notification",
                    "settlement",
                    "release",
                ],
            },
            candidate_index_fields=["_tin", "_npi", "_prov_name_clean"],
            assignment_constraints={
                "state": "_state_clean",
                "payer": "_payer_clean",
                "date": "eff_date",
                "title": "title_clean",
            },
            assignment_rules={
                "pass1": {"state": True, "payer": True, "date": True, "title": True},
                "pass2": {"title": True, "date": True, "title_similarity_min": 0.50},
                "pass3": {"unique_parent": True, "unique_payer": True},
            },
            validation={"source_label_col": "parent_child_flag"},
            rank_output={"preserve_orphan_combined_rank": True},
            qa_flags={
                "flag_mapping_differences": True,
                "flag_rank_mismatch": True,
                "flag_parent_title_keywords": True,
                "flag_payer_suffix": True,
                "flag_similar_group_names": False,
                "flag_low_confidence": False,
                "flag_low_similarity": False,
                "similarity_threshold": 0.60,
            },
            keep_work_cols=False,
        )

    @staticmethod
    def create_caresource() -> ClientConfig:
        """Create CareSource configuration."""
        return ClientConfig(
            name="caresource",
            engine_type="caresource",
            columns={
                "file_name": ["File Name", "Filename", "FILE_NAME"],
                "supplier_group": ["Supplier Group", "SupplierGroup", "SUPPLIER_GROUP"],
                "hierarchical_type": [
                    "Hierarchical Type",
                    "hierarchical_type",
                    "HIERARCHICAL_TYPE",
                ],
                "effective_date": [
                    "Fixed_Effective_Date",
                    "EffectiveDate",
                    "Effective Date",
                ],
                "commodity": ["Commodity", "COMMODITY"],
                "rank_source": ["Rank", "RANK"],
            },
            keep_work_cols=False,
        )

    @staticmethod
    def create_bcbs() -> ClientConfig:
        """Create BCBS configuration matching original bcbs_parent_child_mapping.py logic."""
        return ClientConfig(
            name="bcbs",
            engine_type="bcbs_custom",
            columns={
                "id": ["File Name", "FILE_NAME", "Contract Name"],
                "title": [
                    "contract_title_clean",
                    "fixed_contract_name",
                    "Contract Name",
                    "contract_title_cleaned",
                ],
                "effective_date": [
                    "final_effective_date",
                    "Contract Effective Date",
                    "fixed_effective_date",
                ],
            },
            prep_fields={
                # Grouping fields - old Camelot columns primary, generic pipeline as fallback
                "_folder": {
                    "source": "Folder",
                    "mode": "copy_clean",
                    "cleaner": "upper_clean",
                    "fallback_sources": ["PROV_GROUP_NAME_FULL_cleaned"],
                },
                "_irs_group": {
                    "source": "IRS_Group",
                    "mode": "copy_clean",
                    "cleaner": "lower_clean",
                    "fallback_sources": ["PROV_GROUP_TIN"],
                },
                # Additional fields for matching
                "_provider_type": {
                    "source": "consolidated_provider_type",
                    "mode": "copy_clean",
                    "cleaner": "lower_clean",
                },
                "_lob": {
                    "source": "extracted_lob",
                    "mode": "copy_clean",
                    "cleaner": "lower_clean",
                    "fallback_sources": ["consolidated_aarete_derived_lob"],
                },
            },
            # Match original grouping: Folder + IRS_Group (now mapped to provider name + TIN)
            grouping={
                "strategy_type": "multi_identifier",
                "identifiers": ["_folder", "_irs_group"],
            },
            parent_identification={
                # Use BCBS-specific parent identification logic
                "strategy_type": "bcbs_custom",
                "field": "title_clean",
                "filename_field": "id_clean",
                "lob_field": "_lob",
            },
            # Match original candidate fields
            candidate_index_fields=["_folder", "_irs_group"],
            assignment_constraints={
                "date": "eff_date",
                "title": "title_clean",
                "lob": "_lob",
                "provider_type": "_provider_type",
            },
            assignment_rules={
                # Pass 1: LOB + Provider Type match (Rule 1 from original)
                "pass1": {
                    "date": True,
                    "title": True,
                    "lob": True,
                    "provider_type": True,
                },
                # Pass 2: Title substring match (Rules 0, 2 from original)
                "pass2": {"title": True, "date": True, "title_similarity_min": 0.50},
                # Pass 3: Fallback to latest commercial/parent (Rules 4, 5 from original)
                "pass3": {"unique_parent": True, "commercial_fallback": True},
            },
            validation={"source_label_col": "parent_child_flag"},
            # Parent ranking: Commercial LOB parents ranked first (like original)
            rank_output={
                "preserve_orphan_combined_rank": True,
                "commercial_lob_priority": True,  # Commercial LOB parents get priority
                "lob_field": "_lob",
            },
            qa_flags={
                "flag_mapping_differences": True,
                "flag_rank_mismatch": True,
                "flag_parent_title_keywords": True,
            },
            keep_work_cols=False,
        )

    @staticmethod
    def create_clover() -> ClientConfig:
        """Create Clover Health configuration (same as Molina with source labels)."""
        molina = ConfigFactory.create_molina()
        molina.name = "clover_health"
        molina.validation = {"source_label_col": "parent_child_flag"}
        molina.rank_output = {"preserve_orphan_combined_rank": False}
        return molina

    @staticmethod
    def get_config(client: str) -> ClientConfig:
        """Get configuration for specified client."""
        client = (client or "").strip().lower()

        configs = {
            "molina": ConfigFactory.create_molina,
            "cnc": ConfigFactory.create_molina,  # CNC uses same config as Molina
            "caresource": ConfigFactory.create_caresource,
            "bcbs": ConfigFactory.create_bcbs,
            "bcbsnc": ConfigFactory.create_bcbs,
            "clover_health": ConfigFactory.create_clover,
        }

        if client not in configs:
            logger.warning(
                f"Unknown client '{client}'. Available: {list(configs.keys())}. Defaulting to molina."
            )
            return ConfigFactory.create_molina()

        return configs[client]()


# ============================================================
# Main Engine (continued in next section)
# ============================================================


class ParentChildEngine:
    """Main orchestration pipeline for parent-child mapping."""

    def __init__(self, config: Dict):
        self.cfg = config
        self.cols = config.get("columns", {}) or {}
        self.keep_work_cols = bool(config.get("keep_work_cols", False))
        self.engine_type = config.get("engine_type", "parent_child")

        self.text_cleaner = TextCleaner()
        self.field_deriver = FieldDeriver()
        self.group_builder = GroupKeyBuilder()
        self.parent_identifier = ParentIdentifier()
        self.child_assigner = ChildAssigner()
        self.bcbs_child_assigner = BCBSChildAssigner()  # BCBS-specific assigner

    def run(self, df_in: pd.DataFrame) -> pd.DataFrame:
        """Execute full pipeline."""
        t0 = time.time()
        df = df_in.copy()
        df["input_row_order"] = np.arange(len(df), dtype=np.int32)

        # Initialize assignment columns
        df["assigned_parent_idx"] = pd.Series(
            [pd.NA] * len(df), index=df.index, dtype="Int64"
        )
        df["assignment_pass"] = pd.Series(
            [pd.NA] * len(df), index=df.index, dtype="Int8"
        )
        df["assignment_reasoning"] = pd.Series(
            [""] * len(df), index=df.index, dtype="string"
        )

        df = self.prepare_columns(df)
        df = self.field_deriver.build_fields(
            df,
            self.cfg.get("derive_fields", []) or [],
            self.text_cleaner,
            self.cfg.get("cleaning_rules", {}) or {},
        )
        df = self.build_grouping(df)
        df = self.identify_parents(df)
        df = self.assign_children(df)
        df = self.prepare_output(df)
        df = self.compute_ranks(df)
        self.apply_qa_flags(df)
        self.show_metrics(df)

        elapsed = time.time() - t0
        logger.info("DONE in %.2fs | rows=%s", elapsed, len(df))

        if not self.keep_work_cols:
            df = self._remove_work_columns(df)

        return df

    def prepare_columns(self, df: pd.DataFrame) -> pd.DataFrame:
        """Prepare and clean source columns."""
        rules = self.cfg.get("cleaning_rules", {}) or {}

        id_col = resolve_column(df, self.cols.get("id"), required=True, label="id")
        title_col = resolve_column(
            df, self.cols.get("title"), required=False, label="title"
        )
        eff_col = resolve_column(
            df, self.cols.get("effective_date"), required=False, label="effective_date"
        )

        df["id_raw"] = df[id_col].astype("string")
        df["id_clean"] = TextProcessor.clean_text_lowercase(df["id_raw"])

        if title_col:
            df["title_raw"] = df[title_col].astype("string")
            df["title_clean"] = self.text_cleaner.clean_field(
                "title", df["title_raw"], rules
            )
        else:
            df["title_clean"] = pd.Series([""] * len(df), dtype="string")

        if eff_col:
            df["eff_date"] = to_datetime(df[eff_col])
        else:
            df["eff_date"] = pd.Series([pd.NaT] * len(df))

        # Build prep fields
        prep = self.cfg.get("prep_fields", {}) or {}
        for internal_col, spec in prep.items():
            src = spec.get("source")
            mode = spec.get("mode", "copy_clean")
            cleaner = spec.get("cleaner", "raw")
            fallbacks = spec.get("fallback_sources", [])

            # Try primary source, then fallbacks
            if not src or src not in df.columns:
                resolved = None
                for fb in fallbacks:
                    if fb in df.columns:
                        resolved = fb
                        break
                if resolved:
                    src = resolved
                else:
                    df[internal_col] = pd.Series(
                        [pd.NA] * len(df), index=df.index, dtype="string"
                    )
                    continue

            if mode == "copy_clean":
                if cleaner == "upper_clean":
                    df[internal_col] = TextProcessor.clean_text(
                        to_string(df[src])
                    ).str.upper()
                elif cleaner == "lower_clean":
                    df[internal_col] = TextProcessor.clean_text_lowercase(
                        to_string(df[src])
                    )
                elif cleaner == "identifier":
                    df[internal_col] = TextProcessor.normalize_identifier_series(
                        df[src]
                    )
                else:
                    df[internal_col] = to_string(df[src]).str.strip()

        df["processable"] = True
        return df

    def build_grouping(self, df: pd.DataFrame) -> pd.DataFrame:
        """Build grouping keys."""
        gk, tier = self.group_builder.build_grouping_key(
            df, self.cfg.get("grouping", {}) or {}
        )
        df["grouping_key"] = gk
        df["tier_level"] = tier.astype(np.int8)
        df["processable"] = (
            df["tier_level"].ne(0) & df["grouping_key"].notna()
        ).astype(bool)
        return df

    def identify_parents(self, df: pd.DataFrame) -> pd.DataFrame:
        """Identify parents and compute rank."""
        parent_cfg = self.cfg.get("parent_identification", {}) or {}
        parent_mask = self.parent_identifier.find_parents(df, parent_cfg)
        df["is_parent"] = (parent_mask & df["processable"]).astype(bool)

        df["parent_rank"] = pd.Series(0, index=df.index, dtype=np.int32)

        pm = df["is_parent"]
        if pm.any():
            rank_output = self.cfg.get("rank_output", {}) or {}
            commercial_priority = rank_output.get("commercial_lob_priority", False)
            lob_field = rank_output.get("lob_field", "_lob")

            cols_needed = ["grouping_key", "eff_date", "id_clean", "input_row_order"]
            if commercial_priority and lob_field in df.columns:
                cols_needed.append(lob_field)

            tmp = df.loc[pm, cols_needed].copy()
            tmp["eff_sort"] = tmp["eff_date"].fillna(pd.Timestamp.max)

            # Commercial LOB priority: Commercial parents ranked first within each group
            if commercial_priority and lob_field in tmp.columns:
                # has_commercial = True if 'comm' is in LOB value
                tmp["_has_commercial"] = (
                    tmp[lob_field]
                    .fillna("")
                    .astype(str)
                    .str.contains("comm", case=False, na=False)
                )
                # Sort: group, commercial first (True=0, False=1), then by date
                tmp["_comm_sort"] = (~tmp["_has_commercial"]).astype(int)
                tmp = tmp.sort_values(
                    [
                        "grouping_key",
                        "_comm_sort",
                        "eff_sort",
                        "input_row_order",
                        "id_clean",
                    ],
                    kind="mergesort",
                )
                tmp = tmp.drop(columns=["_has_commercial", "_comm_sort"])
            else:
                tmp = tmp.sort_values(
                    ["grouping_key", "eff_sort", "input_row_order", "id_clean"],
                    kind="mergesort",
                )

            tmp["parent_rank"] = tmp.groupby("grouping_key", sort=False).cumcount() + 1
            df.loc[tmp.index, "parent_rank"] = (
                tmp["parent_rank"].astype(np.int32).values
            )

        return df

    def assign_children(self, df: pd.DataFrame) -> pd.DataFrame:
        """Execute child assignment."""
        # Use BCBS-specific 5-rule assigner for bcbs_custom engine type
        if self.engine_type == "bcbs_custom":
            return self.bcbs_child_assigner.assign_children_bcbs(df, self.cfg)

        # Default: use generic multi-pass assigner
        constraints = self.cfg.get("assignment_constraints", {}) or {}
        cols_needed = ["parent_rank"]
        # Include all constraint fields: state, payer, commodity, date, title, lob, provider_type
        for k in [
            "state",
            "payer",
            "commodity",
            "date",
            "title",
            "lob",
            "provider_type",
        ]:
            col = constraints.get(k)
            if col and col in df.columns:
                cols_needed.append(col)
        cols_needed = list(dict.fromkeys(cols_needed))
        parent_cache = df.loc[df["is_parent"], cols_needed].to_dict("index")
        return self.child_assigner.assign_children(df, self.cfg, parent_cache)

    @staticmethod
    def _normalize_amendment_val(x):
        """Convert amendment value to string for rank construction.

        Handles both numeric values (e.g. 3.0 -> '3') and character
        values (e.g. 'A' -> 'A').  Returns '0' for NaN / empty.
        """
        if pd.notna(x) and str(x).strip() != "":
            try:
                return str(int(float(x)))
            except (ValueError, TypeError):
                raw_str = str(x).strip()
                if len(raw_str) == 1 and raw_str.isalpha():
                    return str(ord(raw_str.upper()) - ord("A") + 1)
                return raw_str
        return "0"

    def prepare_output(self, df: pd.DataFrame) -> pd.DataFrame:
        """Initialize output columns."""
        df["parent_child_flag_dest"] = pd.Series([pd.NA] * len(df), dtype="string")
        df["parent_rank_dest"] = pd.Series([pd.NA] * len(df), dtype="string")
        df["child_rank_dest"] = pd.Series([pd.NA] * len(df), dtype="string")
        df["orphan_rank_dest"] = pd.Series([pd.NA] * len(df), dtype="string")
        df["combined_rank_dest"] = pd.Series([pd.NA] * len(df), dtype="string")

        pm = df["is_parent"]
        cm = df["assigned_parent_idx"].notna()
        om = (~pm) & (~cm)

        df.loc[pm, "parent_child_flag_dest"] = "Parent"
        df.loc[cm, "parent_child_flag_dest"] = "Child"
        df.loc[om, "parent_child_flag_dest"] = "Orphan"

        df.loc[pm, "parent_rank_dest"] = df.loc[pm, "parent_rank"].astype("string")
        df.loc[pm, "combined_rank_dest"] = df.loc[pm, "parent_rank_dest"]

        return df

    def compute_ranks(self, df: pd.DataFrame) -> pd.DataFrame:
        """Compute child and orphan ranks."""
        # Child ranks
        child_mask = df["assigned_parent_idx"].notna()
        if child_mask.any():
            tmp = df.loc[
                child_mask, ["assigned_parent_idx", "eff_date", "input_row_order"]
            ].copy()
            tmp["eff_sort"] = tmp["eff_date"].fillna(pd.Timestamp.max)
            tmp = tmp.sort_values(
                ["assigned_parent_idx", "eff_sort", "input_row_order"], kind="mergesort"
            )

            seq = tmp.groupby("assigned_parent_idx", sort=False).cumcount() + 1
            seq_str = seq.astype("string").reset_index(drop=True).values

            parent_idx = tmp["assigned_parent_idx"].astype("int64").values
            parent_rank_str = (
                df.loc[parent_idx, "parent_rank"]
                .apply(lambda x: str(int(x)) if pd.notna(x) else "0")
                .reset_index(drop=True)
                .values
            )

            if "AARETE_DERIVED_AMENDMENT_NUM" in df.columns:
                amendment_str = (
                    df.loc[tmp.index, "AARETE_DERIVED_AMENDMENT_NUM"]
                    .apply(self._normalize_amendment_val)
                    .reset_index(drop=True)
                    .values
                )
            else:
                amendment_str = "0"
            df.loc[tmp.index, "child_rank_dest"] = (
                parent_rank_str + "." + amendment_str + "." + seq_str
            )
            df.loc[tmp.index, "combined_rank_dest"] = df.loc[
                tmp.index, "child_rank_dest"
            ]

        # Orphan ranks — bucketed by the visible group (provider_group) so two
        # orphans that share a Group label but live in different grouping_keys
        # still get distinct rank strings. Previously bucketing on grouping_key
        # let unrelated singleton orphans collide as "0.<amendment>.1" — see
        # the Inspira Health Network case. Fall back to grouping_key when
        # provider_group is absent (isolated test harnesses).
        pm = df["is_parent"]
        cm = df["assigned_parent_idx"].notna()
        orphan_mask = (~pm) & (~cm)

        if orphan_mask.any():
            sort_cols = ["eff_date", "input_row_order"]
            tmp = df.loc[orphan_mask, sort_cols].copy()
            if "provider_group" in df.columns:
                tmp["_gk"] = (
                    df.loc[orphan_mask, "provider_group"].fillna("__none__").astype(str)
                )
            elif "grouping_key" in df.columns:
                tmp["_gk"] = (
                    df.loc[orphan_mask, "grouping_key"].fillna("__none__").astype(str)
                )
            else:
                # No grouping info → every orphan is its own bucket.
                tmp["_gk"] = [f"__row_{i}__" for i in range(len(tmp))]
            tmp["eff_sort"] = tmp["eff_date"].fillna(pd.Timestamp.max)
            tmp = tmp.sort_values(
                ["_gk", "eff_sort", "input_row_order"], kind="mergesort"
            )

            seq_int = tmp.groupby("_gk", sort=False).cumcount() + 1

            if "AARETE_DERIVED_AMENDMENT_NUM" in df.columns:
                amendment_str = df.loc[tmp.index, "AARETE_DERIVED_AMENDMENT_NUM"].apply(
                    self._normalize_amendment_val
                )
            else:
                amendment_str = pd.Series(["0"] * len(tmp), index=tmp.index)

            seq_str = seq_int.astype("string")
            ranks = ("0." + amendment_str + "." + seq_str).values

            df.loc[tmp.index, "orphan_rank_dest"] = ranks
            df.loc[tmp.index, "combined_rank_dest"] = ranks

        return df

    def apply_qa_flags(self, df: pd.DataFrame) -> None:
        """Apply QA flags for manual review."""
        qa_config = self.cfg.get("qa_flags", {}) or {}

        # Flag mapping differences
        if qa_config.get("flag_mapping_differences", True):
            if (
                "parent_child_flag" in df.columns
                and "parent_child_flag_dest" in df.columns
            ):
                src = TextProcessor.normalize_label(df["parent_child_flag"])
                dest = TextProcessor.normalize_label(df["parent_child_flag_dest"])
                mask = (
                    src.notna()
                    & dest.notna()
                    & (src != "")
                    & (dest != "")
                    & (src != dest)
                )
                self._add_flag(df, mask, "Incorrect Mapping")

        # Flag rank mismatches
        if qa_config.get("flag_rank_mismatch", True):
            if "combined_rank_dest" in df.columns and "combined_rank" in df.columns:
                mask = self._find_significant_rank_differences(df)
                self._add_flag(df, mask, "Rank Mismatch")

        # Flag parent titles with keywords
        if qa_config.get("flag_parent_title_keywords", True):
            title_col = self._find_title_column(df)
            if title_col:
                has_excluded = (
                    df[title_col].astype("string").str.contains(TITLE_REGEX, na=False)
                )
                is_parent = df["parent_child_flag_dest"] == "Parent"
                self._add_flag(df, has_excluded & is_parent, "Title Contains Keywords")

        # Flag payer suffixes
        if qa_config.get("flag_payer_suffix", True):
            if "payer_name_cleaned" in df.columns:
                has_suffix = (
                    df["payer_name_cleaned"]
                    .astype("string")
                    .str.contains(SUFFIX_REGEX, na=False)
                )
                self._add_flag(df, has_suffix, "Payer Suffix Present")

    def _find_significant_rank_differences(self, df: pd.DataFrame) -> pd.Series:
        """Identify significant rank differences."""
        src = df["combined_rank"].astype("string").fillna("")
        dest = df["combined_rank_dest"].astype("string").fillna("")

        def is_significant(src_rank, dest_rank):
            if src_rank == "" or dest_rank == "" or src_rank == dest_rank:
                return False

            src_parts = str(src_rank).split(".")
            dest_parts = str(dest_rank).split(".")

            if len(src_parts) == 2 and src_parts[1] == "0":
                src_parts = [src_parts[0]]
            if len(dest_parts) == 2 and dest_parts[1] == "0":
                dest_parts = [dest_parts[0]]

            if len(src_parts) != len(dest_parts):
                return True

            if len(src_parts) >= 3:
                if src_parts[0] != dest_parts[0]:
                    return True

            return False

        return pd.Series(
            [is_significant(src.iloc[i], dest.iloc[i]) for i in range(len(df))],
            index=df.index,
        )

    @staticmethod
    def _find_title_column(df: pd.DataFrame) -> Optional[str]:
        """Find title column in dataframe."""
        candidates = ["contract_title_cleaned", "contract_title_clean", "_title_clean"]
        for col in candidates:
            if col in df.columns:
                return col
        return None

    @staticmethod
    def _add_flag(df: pd.DataFrame, row_mask: pd.Series, flag: str) -> None:
        """Add manual review flag."""
        if "Manual_Review_Flag" not in df.columns:
            df["Manual_Review_Flag"] = pd.Series([pd.NA] * len(df), dtype="string")

        existing = df.loc[row_mask, "Manual_Review_Flag"]
        df.loc[row_mask, "Manual_Review_Flag"] = existing.mask(
            existing.notna(), existing + " | " + flag
        ).fillna(flag)

    def show_metrics(self, df: pd.DataFrame) -> None:
        """Log accuracy metrics."""
        vcfg = self.cfg.get("validation", {}) or {}
        src_col = vcfg.get("source_label_col")
        if not src_col or src_col not in df.columns:
            return
        src = TextProcessor.normalize_label(df[src_col])
        dest = TextProcessor.normalize_label(df["parent_child_flag_dest"])
        valid = src.notna() & dest.notna()
        match = (src == dest) & valid
        total_valid = int(valid.sum())
        match_cnt = int(match.sum())
        acc = (match_cnt / total_valid * 100.0) if total_valid else 0.0
        logger.info("Classification accuracy: %.2f%% (valid=%s)", acc, total_valid)

    @staticmethod
    def _remove_work_columns(df: pd.DataFrame) -> pd.DataFrame:
        """Remove internal working columns."""
        work_cols = [
            c for c in df.columns if c.startswith("_") or c.startswith("temp_")
        ]
        df.drop(columns=work_cols, inplace=True, errors="ignore")
        return df


# ============================================================
# Main Entry Point
# ============================================================


def qc_main(df_in, client):
    """Main entry point."""
    # output_file = f"output_{client}.csv"

    logger.info("Client: %s", client)

    # Get configuration
    config = ConfigFactory.get_config(client)
    cfg_dict = config.to_dict()

    logger.info("Loaded: %s rows, %s cols", len(df_in), len(df_in.columns))

    # Process
    engine = ParentChildEngine(cfg_dict)
    df_out = engine.run(df_in)

    # Reviewer-facing column order (user-specified).
    out_cols = df_out.columns.tolist()
    preferred_order = [
        "FILE_NAME",
        "fixed_effective_date",
        "contract_title_cleaned",
        "PROV_GROUP_NAME_FULL_cleaned",
        "PROV_GROUP_TIN",
        "PROV_GROUP_NPI",
        "payer_name_cleaned",
        "PAYER_STATE",
        "AARETE_DERIVED_AMENDMENT_NUM",
        "grouping_key",
        "provider_group",
        "parent",
        "combined_rank",
        "parent_child_flag",
        "assignment_reasoning",
        "parent_name",
        "parent_child_flag_dest",
        "combined_rank_dest",
        "Manual_Review_Flag",
    ]

    keep_cols = [c for c in preferred_order if c in out_cols]

    df_final = df_out.loc[:, keep_cols].copy()

    # Reviewer-friendly names, then uppercase every header so the sheet
    # reads consistently. Data values keep their original case. The three
    # AARETE_DERIVED_* renames relabel the header to match the source of
    # truth — the values in these columns are already populated from the
    # derived fields upstream in preprocessing.
    df_final = df_final.rename(
        columns={
            "fixed_effective_date": "AARETE_DERIVED_EFFECTIVE_DT",
            "PROV_GROUP_NAME_FULL_cleaned": "AARETE_DERIVED_PROVIDER_NAME",
            "payer_name_cleaned": "AARETE_DERIVED_PAYER_NAME",
            "provider_group": "Group",
            "parent_child_flag": "Flag",
            "combined_rank": "Rank",
            "parent_name": "Parent_Child_Name",
            "assignment_reasoning": "Reasoning",
            "parent_child_flag_dest": "QC_Flag",
            "combined_rank_dest": "QC_Rank",
        }
    )
    df_final.columns = [str(c).upper() for c in df_final.columns]
    return df_final