How to map CSI MasterFormat to custom WBS codes in Python

import re
import logging
from typing import Dict, Optional, Tuple, List
from dataclasses import dataclass

# Configure structured logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger(__name__)

@dataclass(frozen=True)
class WBSMappingResult:
    original_input: str
    normalized_csi: str
    wbs_code: str
    match_type: str  # 'exact', 'division', 'regex', 'fallback'
    confidence: float

class CSIToWBSMapper:
    """Deterministic CSI MasterFormat to WBS code resolver."""

    def __init__(
        self,
        exact_map: Dict[str, str],
        division_map: Dict[str, str],
        default_wbs: str = "00-00-00-00"
    ):
        self.exact_map = {k.upper().strip(): v for k, v in exact_map.items()}
        self.division_map = {k.upper().strip(): v for k, v in division_map.items()}
        self.default_wbs = default_wbs
        # Matches 5 or 6 digits, optionally separated by hyphens/spaces
        self.csi_pattern = re.compile(r"^(\d{2})[-\s]?(\d{2,3})[-\s]?(\d{2,3})?$")

    def normalize_csi(self, raw_code: str) -> str:
        """Strips suffixes, extracts digits, and pads to canonical DD-SS-NN."""
        clean = re.sub(r"[-_]\w+", "", raw_code)
        digits_only = re.sub(r"[^\d]", "", clean)

        if len(digits_only) not in (5, 6):
            raise ValueError(f"Invalid CSI digit count ({len(digits_only)}): {raw_code}")

        if len(digits_only) == 5:
            # Legacy 2004: DD-SSS -> Pad to DD-SS-NN (append trailing '0' for NarrowScope)
            digits_only = f"{digits_only}0"

        return f"{digits_only[:2]}-{digits_only[2:4]}-{digits_only[4:]}"

# Extends the CSIToWBSMapper class defined above.
class CSIToWBSMapper(CSIToWBSMapper):
    def resolve(self, raw_code: str) -> WBSMappingResult:
        """Resolves a raw CSI string to a WBS code with confidence scoring."""
        try:
            normalized = self.normalize_csi(raw_code)
        except ValueError as e:
            logger.error("Normalization failed: %s", e)
            return WBSMappingResult(
                original_input=raw_code,
                normalized_csi="INVALID",
                wbs_code=self.default_wbs,
                match_type="fallback",
                confidence=0.0,
            )

        # 1. Exact match (highest confidence)
        if normalized in self.exact_map:
            return WBSMappingResult(raw_code, normalized, self.exact_map[normalized], "exact", 1.0)

        # 2. Division-level match (BroadScope fallback)
        division_key = f"{normalized[:2]}-00-00"
        if division_key in self.division_map:
            return WBSMappingResult(raw_code, normalized, self.division_map[division_key], "division", 0.85)

        # 3. Cross-divisional routing (00/01 General Requirements)
        if normalized.startswith(("00-", "01-")):
            fallback_div = self.division_map.get("00-00-00", self.default_wbs)
            return WBSMappingResult(raw_code, normalized, fallback_div, "regex", 0.7)

        # 4. Hard fallback with audit warning
        logger.warning("No mapping found for %s, routing to default WBS", normalized)
        return WBSMappingResult(raw_code, normalized, self.default_wbs, "fallback", 0.1)

def process_cost_codes_batch(
    mapper: CSIToWBSMapper,
    raw_codes: List[str]
) -> List[WBSMappingResult]:
    """Batch processes raw CSI strings with error isolation and audit logging."""
    results = []
    for code in raw_codes:
        try:
            result = mapper.resolve(code.strip())
            results.append(result)
            if result.match_type == "fallback":
                logger.info("AUDIT | Fallback routed | Input: %s | WBS: %s", code, result.wbs_code)
        except Exception as e:
            logger.critical("UNHANDLED EXCEPTION | Input: %s | Error: %s", code, str(e))
            results.append(WBSMappingResult(
                original_input=code,
                normalized_csi="ERROR",
                wbs_code=mapper.default_wbs,
                match_type="fallback",
                confidence=0.0
            ))
    return results

# Example usage
if __name__ == "__main__":
    # Enterprise mapping dictionaries (typically loaded from DB/JSON)
    EXACT_MAPPINGS = {
        "03-30-00": "WBS-CON-0330",
        "09-21-16": "WBS-FIN-0921",
        "26-05-19": "WBS-ELC-2605"
    }
    DIVISION_MAPPINGS = {
        "03-00-00": "WBS-CON-0300",
        "09-00-00": "WBS-FIN-0900",
        "00-00-00": "WBS-ADM-0000"
    }

    mapper = CSIToWBSMapper(
        exact_map=EXACT_MAPPINGS,
        division_map=DIVISION_MAPPINGS,
        default_wbs="WBS-ADM-UNMAPPED"
    )

    test_inputs = [
        "03-30-00",      # Exact match
        "09-21-16",      # Exact match
        "03-30-00-ALT1", # Suffix handling
        "03-30-12",      # Division fallback
        "01-00-00",      # Cross-divisional regex
        "123",           # Invalid length
        "26-05-19"       # Exact match
    ]

    resolved = process_cost_codes_batch(mapper, test_inputs)
    for r in resolved:
        print(f"{r.original_input:<15} -> {r.wbs_code:<18} | Type: {r.match_type:<10} | Conf: {r.confidence}")