import re import pandas as pd from typing import Dict class QSRuleBasedFallbackEngine: """ Deterministic rule-based classifier utilizing expert QS heuristics. This engine executes ONLY on items rejected by the primary ML model to safely route rare classes, nested composite descriptions, and pricing schedules. It relies on Python's dictionary insertion order to enforce 'Specificity Cascading'. """ def __init__(self, rule_map: Dict[str, str]): """ Initializes the engine and pre-compiles regex patterns for high-speed inference. Parameters ---------- rule_map : Dict[str, str] A dictionary mapping regex patterns to NRM Categories. Keys must be ordered from most specific to least specific. """ # We pre-compile the regexes during initialization so inference is lightning fast. self.rules = { re.compile(pattern): category for pattern, category in rule_map.items() } def predict(self, descriptions: pd.Series) -> pd.Series: """ Vectorized application of expert heuristics. Parameters ---------- descriptions : pd.Series Pandas Series containing the BoQ text descriptions. Returns ------- pd.Series A Series of predicted NRM categories, or 'Unclassified' if no rules match. """ # 1. Initialize everything as Unclassified results = pd.Series("Unclassified", index=descriptions.index) # 2. Safe string conversion to prevent regex failures on NaN or float values clean_desc = descriptions.fillna("").astype(str) # 3. Iterate through rules sequentially (Top-to-Bottom evaluation) for pattern, category in self.rules.items(): # Find rows that are STILL unclassified unclassified_mask = results == "Unclassified" # Performance Optimization: If we've successfully classified everything, # short-circuit and stop running expensive regexes! if not unclassified_mask.any(): break # Pandas complains about regex capture groups in str.contains. # Using apply with our pre-compiled regex is warning-free and faster. match_mask = clean_desc.apply(lambda x: bool(pattern.search(x))) # Update the results only where it was unclassified AND matched the pattern results.loc[unclassified_mask & match_mask] = category return results