Building fallback logic for ambiguous tariff classifications

import logging
import pandas as pd
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from pathlib import Path

# Configure structured logging for audit compliance
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[logging.StreamHandler()],
)
logger = logging.getLogger("hts_fallback_router")


@dataclass
class ClassificationResult:
    hts_code: str
    confidence: float
    routing_path: str
    requires_review: bool
    metadata: Dict[str, str] = field(default_factory=dict)


class FallbackClassificationEngine:
    """Deterministic tie-break + quarantine router for ambiguous HTSUS mappings."""

    def __init__(
        self,
        fuzzy_threshold: float = 0.15,
        min_confidence: float = 0.75,
        tie_margin: float = 0.05,
        chunk_size: int = 50_000,
    ) -> None:
        self.fuzzy_threshold = fuzzy_threshold
        self.min_confidence = min_confidence
        # If the two best candidates are within tie_margin, treat as ambiguous.
        self.tie_margin = tie_margin
        self.chunk_size = chunk_size
        logger.info(
            "Engine ready: min_confidence=%.2f, tie_margin=%.2f, chunk_size=%d",
            self.min_confidence, self.tie_margin, self.chunk_size,
        )

    def _rank_fuzzy_matches(
        self, description: str, historical_map: Dict[str, str]
    ) -> List[Tuple[str, float]]:
        """Tier 1: rank historical codes by similarity (use rapidfuzz in prod)."""
        scored: Dict[str, float] = {}
        for hist_desc, hts_code in historical_map.items():
            overlap = len(set(description.lower()) & set(hist_desc.lower()))
            distance = 1 - (overlap / max(len(description), len(hist_desc), 1))
            if distance <= self.fuzzy_threshold:
                confidence = 1.0 - distance
                # Keep the best score seen per distinct HTS code.
                scored[hts_code] = max(scored.get(hts_code, 0.0), confidence)
        return sorted(scored.items(), key=lambda kv: kv[1], reverse=True)

    def _evaluate_weighted_matrix(
        self, sku_meta: Dict[str, str], rule_weights: Dict[str, float]
    ) -> Optional[Tuple[str, float]]:
        """Tier 2: GRI-anchored material/end-use weighted scoring."""
        score = sum(
            weight for attr, weight in rule_weights.items()
            if sku_meta.get(attr)
        )
        if score >= self.min_confidence:
            code = sku_meta.get("default_hts_candidate", "UNKNOWN")
            return code, min(score, 1.0)
        return None

    def _classify(
        self,
        description: str,
        sku_meta: Dict[str, str],
        historical_map: Dict[str, str],
        rule_weights: Dict[str, float],
        sku_id: str,
    ) -> ClassificationResult:
        ranked = self._rank_fuzzy_matches(description, historical_map)
        if ranked and ranked[0][1] >= self.min_confidence:
            # Ambiguity guard: two near-equal top codes must not be auto-picked.
            if len(ranked) > 1 and (ranked[0][1] - ranked[1][1]) < self.tie_margin:
                logger.warning(
                    "Ambiguous Tier 1 match for %s: %s vs %s within tie_margin",
                    sku_id, ranked[0][0], ranked[1][0],
                )
                return ClassificationResult(
                    "QUARANTINE", 0.0, "TIER3_TIE_BREAK", True,
                    {"sku_id": sku_id, "candidates": f"{ranked[0][0]}|{ranked[1][0]}"},
                )
            return ClassificationResult(
                ranked[0][0], ranked[0][1], "TIER1_FUZZY", False
            )

        tier2 = self._evaluate_weighted_matrix(sku_meta, rule_weights)
        if tier2:
            return ClassificationResult(tier2[0], tier2[1], "TIER2_MATRIX", False)

        return ClassificationResult(
            "QUARANTINE", 0.0, "TIER3_EXCEPTION", True,
            {"sku_id": sku_id, "description": description},
        )

    def process_chunk(
        self,
        chunk: pd.DataFrame,
        historical_map: Dict[str, str],
        rule_weights: Dict[str, float],
    ) -> pd.DataFrame:
        """Execute fallback routing on a memory-bounded DataFrame chunk."""
        results: List[ClassificationResult] = []
        for _, row in chunk.iterrows():
            desc = str(row.get("product_description", ""))
            meta = {
                k: str(v) for k, v in row.items()
                if k not in ("product_description", "sku_id")
            }
            results.append(
                self._classify(
                    desc, meta, historical_map, rule_weights,
                    str(row.get("sku_id", "")),
                )
            )
        return pd.DataFrame([r.__dict__ for r in results])

    def run_pipeline(
        self,
        input_path: Path,
        output_path: Path,
        historical_map: Dict[str, str],
        rule_weights: Dict[str, float],
    ) -> None:
        """Chunked execution with strict memory bounds.

        Note: pandas' chunked CSV iterator (`chunksize`) is not compatible with
        `engine='pyarrow'`; use the default C engine for streamed ingestion.
        """
        logger.info("Starting chunked fallback processing: %s", input_path)
        first_chunk = True
        for chunk in pd.read_csv(input_path, chunksize=self.chunk_size):
            processed = self.process_chunk(chunk, historical_map, rule_weights)
            processed.to_csv(
                output_path,
                mode="w" if first_chunk else "a",
                header=first_chunk,
                index=False,
            )
            logger.info("Wrote %d records to %s", len(processed), output_path)
            first_chunk = False
        logger.info("Pipeline completed.")