doczyai-pipelines/src/parent_child/qc.py

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

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__)


# ============================================================
# 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 normalize_identifier(value: Any) -> str:
        """Normalize TIN/NPI identifiers (handles pipe-separated values)."""
        if value is None or pd.isna(value):
            return ""

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

        if "|" in val_str:
            parts = [
                TextProcessor.normalize_single_identifier(p) for p in val_str.split("|")
            ]
            seen = set()
            unique_parts = []
            for p in parts:
                if p and p not in seen:
                    seen.add(p)
                    unique_parts.append(p)
            return "|".join(unique_parts) if unique_parts 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 pipe-separated)."""
        if value is None or pd.isna(value):
            return set()

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

        if "|" in val_str:
            parts = [
                TextProcessor.normalize_single_identifier(p) for p in val_str.split("|")
            ]
            return {p for p in parts 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 = config.get("field")
        exclude_keywords = config.get("exclude_keywords", [])
        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]
        has_keyword = (
            df[field].astype("string").str.contains(compiled, na=False, regex=True)
        )
        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
            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:
                continue

            # 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")

        # Group by Folder + IRS_Group (grouping_key)
        for gk, group in df.groupby("grouping_key", dropna=False):
            if pd.isna(gk):
                continue

            parents = group[group["is_parent"]].copy()
            children = group[~group["is_parent"]].copy()

            if parents.empty or children.empty:
                continue

            # Sort parents by effective date
            parents = parents.sort_values(col_eff, na_position="last")

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

            # Process each child
            for cidx in children.index:
                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 = ""

                # Get date-eligible parents (parent_eff <= child_eff)
                def is_date_eligible(pidx: int) -> bool:
                    p_eff = parent_cache[pidx]["eff_date"]
                    if pd.isna(child_eff):
                        return True  # No child date, all parents eligible
                    if pd.isna(p_eff):
                        return False  # Parent has no date, not eligible
                    return p_eff <= child_eff

                eligible_parents = [
                    pidx for pidx in parent_cache.keys() if is_date_eligible(pidx)
                ]

                # Rule 0: Professional logic
                if "professional" in child_title and candidate_idx is None:
                    # 0a: Parent title fully contained in child title
                    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

                    # 0b: Parent title contains "professional"
                    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:
                        # 3a: Match by LOB + Provider Type
                        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:
                        # 3b: Fallback - find parent with same LOB, prefer 'facility' provider type
                        lob_matches = [
                            pidx
                            for pidx in eligible_parents
                            if parent_cache[pidx]["lob"] == child_lob
                        ]
                        if lob_matches:
                            # Prefer facility
                            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  # All BCBS rules are 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",
                "exclude_keywords": [
                    "exhibit",
                    "amendment",
                    "amend",
                    "addendum",
                    "adden",
                    "renewal",
                    "extension",
                    "modification",
                    "attachment",
                    "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",
                ],
                "effective_date": ["final_effective_date", "Contract Effective Date"],
            },
            prep_fields={
                # Grouping fields - match original: Folder + IRS_Group
                "_folder": {
                    "source": "Folder",
                    "mode": "copy_clean",
                    "cleaner": "upper_clean",
                },
                "_irs_group": {
                    "source": "IRS_Group",
                    "mode": "copy_clean",
                    "cleaner": "lower_clean",
                },
                # 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",
                },
            },
            # Match original grouping: Folder + IRS_Group
            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,
            "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")

            if not src or src not in df.columns:
                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)

    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"]
                .astype("string")
                .reset_index(drop=True)
                .values
            )

            df.loc[tmp.index, "child_rank_dest"] = parent_rank_str + ".0." + seq_str
            df.loc[tmp.index, "combined_rank_dest"] = df.loc[
                tmp.index, "child_rank_dest"
            ]

        # Orphan ranks
        pm = df["is_parent"]
        cm = df["assigned_parent_idx"].notna()
        orphan_mask = (~pm) & (~cm)

        if orphan_mask.any():
            tmp = df.loc[orphan_mask, ["input_row_order"]].copy()
            tmp = tmp.sort_values(["input_row_order"], kind="mergesort")
            seq = pd.Series(np.arange(1, len(tmp) + 1, dtype=np.int32), index=tmp.index)
            df.loc[tmp.index, "orphan_rank_dest"] = (
                "0.0." + seq.astype("string")
            ).values
            df.loc[tmp.index, "combined_rank_dest"] = df.loc[
                tmp.index, "orphan_rank_dest"
            ]

        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)

    # Export
    input_cols = df_in.columns.tolist()
    out_cols = df_out.columns.tolist()
    extra_cols = [
        "parent_child_flag_dest",
        "combined_rank_dest",
        "Manual_Review_Flag",
        "assignment_reasoning",
        "parent_name_dest",
    ]
    keep_cols = [c for c in input_cols if c in out_cols]
    keep_cols.extend([c for c in extra_cols if c in out_cols and c not in keep_cols])

    df_final = df_out.loc[:, keep_cols].copy()
    # df_final.to_csv(output_file, index=False, encoding="utf-8", quoting=1)
    # logger.info("Exported: %s", output_file)
    return df_final