CMacD/AIC_PHASE1_POC
0
1#!/usr/bin/env python2# coding: utf-83"""4MappingLookup — fuzzy historical match engine (Phase 1, Step 1).5 6This is the first stage of the pipeline and the most straightforward one.7For every new product in the flat file it asks: have we seen something like8this before in the historical FINAL data? If yes, carry the historical label9across. If no, find the closest thing and flag it for analyst review.10 11What it does12------------13 1. Builds a single lookup key per product by concatenating the relevant14 attribute columns — for example, RAW_BRAND + RAW_SUB_BRAND becomes one15 string that represents the product's identity for that attribute.16 17 2. Does a direct join against the historical data first. Anything that18 matches exactly gets score=100 and is done.19 20 3. For products that do not match exactly, runs vectorised rapidfuzz21 token_set_ratio scoring against all known historical keys. The minimum22 of token_set_ratio and token_sort_ratio is used as the final score so23 that partial string matches like "KRAFT" vs "KRAFT HEINZ" are not24 artificially inflated. Suggestions below 50 are dropped as noise.25 26 4. If a product still has no suggestion after fuzzy scoring, and the27 attribute has a recognised catch-all label in its vocabulary (e.g.28 "ALL OTHER BRANDS", "AO COFFEE"), that product is assigned the fallback29 label with score=0. Score=0 guarantees HIGH QC Priority in Ensemble so30 an analyst always sees it.31 32 5. Returns one ranked lookup table per attribute plus the flat-file output33 and a flag table that tells Ensemble whether a key maps to one historical34 label or several.35 36Abbreviation expansion37----------------------38A small set of safe product-name abbreviations are expanded before matching39(CHOC → CHOCOLATE, ORG → ORGANIC etc). Only unambiguous ones are included40here. Category-specific synonyms should go in a synonyms.json at project41level, not in this file.42 43Code style44----------45Functions are written to be read straight through. Steps are broken into named46variables rather than chained. If something is not immediately obvious from the47code it has a comment. Keep it that way.48"""49 50import re51import sqlite352import traceback53 54import numpy as np55import pandas as pd56from rapidfuzz import process as rfprocess, fuzz as rffuzz57 58 59# ── Key normalisation ─────────────────────────────────────────────────────────60# Joining attribute tokens with a space (not '') lets token_set_ratio correctly61# tokenise "DARK CHOCOLATE" rather than matching against "DARKCHOCOLATE".62_ABBREV = {63 'CHOC': 'CHOCOLATE',64 'NAT': 'NATURAL',65 'ORG': 'ORGANIC',66 'WHL': 'WHOLE',67 'LF': 'LOW FAT',68 'FF': 'FAT FREE',69 'RF': 'REDUCED FAT',70}71_ABBREV_RE = {re.compile(r'\b' + k + r'\b'): v for k, v in _ABBREV.items()}72 73# ── "All Other" bucket detection ──────────────────────────────────────────────74# Matches standard catch-all label conventions used across Circana categories:75# "ALL OTHER" "ALL OTHERS" "ALL OTHER BRANDS"76# "AO BRANDS" "AO COFFEE" etc.77# Deliberately excludes:78# "AO" alone — too ambiguous; could be a real brand abbreviation.79# "OTHER" alone — too generic; matches many non-catch-all labels.80# If no label in the attribute's value space matches, the fallback is skipped81# entirely and the analyst handles the gap — no invented values are written.82_AO_RE = re.compile(r'^(ALL\s+OTHERS?(?:\s|$)|AO\s+\S)', re.IGNORECASE)83 84 85def _normalise_key(text: str) -> str:86 """Uppercase, collapse whitespace, expand safe product-name abbreviations."""87 text = str(text).upper().strip()88 text = re.sub(r'\s+', ' ', text)89 for pat, rep in _ABBREV_RE.items():90 text = pat.sub(rep, text)91 return text92 93 94def _build_key(row) -> str:95 """Concatenate all values in a DataFrame row into a normalised lookup key."""96 return _normalise_key(' '.join(str(v) for v in row.values))97 98 99# ── Internal helpers ──────────────────────────────────────────────────────────100 101def _score_unmapped_keys(history_df: pd.DataFrame, attr_key_cols: list,102 mdm_col: str) -> pd.DataFrame:103 """104 Score every unmapped product key against every mapped historical key using105 rapidfuzz token_set_ratio (vectorised).106 107 Parameters108 ----------109 history_df : DataFrame110 Aggregated historical rows with a 'combinedkey' column and the target111 MDM column. Rows where the MDM value equals 'missing' are unmapped.112 attr_key_cols : list of str113 Raw key columns for this attribute (carried through to the output).114 mdm_col : str115 Name of the MDM column being matched.116 117 Returns118 -------119 DataFrame with columns: unmapped, mapped, {mdm_col}, score, attr_key_cols.120 Pre-mapped rows (score=100) are included alongside the scored rows.121 """122 history_df.fillna('missing', inplace=True)123 124 unmapped = history_df[history_df[mdm_col] == 'missing'][['combinedkey'] + attr_key_cols].drop_duplicates()125 mapped = history_df[history_df[mdm_col] != 'missing'][['combinedkey', mdm_col]].drop_duplicates()126 pre_mapped = (127 history_df[history_df[mdm_col] != 'missing'][['combinedkey', mdm_col] + attr_key_cols]128 .drop_duplicates()129 .rename(columns={'combinedkey': 'mapped'})130 .assign(score=100)131 )132 133 if unmapped.empty:134 mapped['score'] = 100135 mapped['mapped'] = mapped['combinedkey']136 return mapped137 138 # Vectorised pairwise scoring — replaces cross-join + row-by-row apply.139 unmapped_reset = unmapped.reset_index(drop=True)140 mapped_reset = mapped.reset_index(drop=True)141 scores_mat = rfprocess.cdist(142 unmapped_reset['combinedkey'].tolist(),143 mapped_reset['combinedkey'].tolist(),144 scorer=rffuzz.token_set_ratio,145 workers=-1,146 )147 unmapped_grid, mapped_grid = np.meshgrid(148 np.arange(len(unmapped_reset)),149 np.arange(len(mapped_reset)),150 indexing='ij',151 )152 unmapped_indices = unmapped_grid.ravel()153 mapped_indices = mapped_grid.ravel()154 scored = pd.DataFrame({155 'unmapped': unmapped_reset['combinedkey'].values[unmapped_indices],156 'mapped': mapped_reset['combinedkey'].values[mapped_indices],157 mdm_col: mapped_reset[mdm_col].values[mapped_indices],158 'score': scores_mat.ravel(),159 })160 for col in attr_key_cols:161 if col in unmapped_reset.columns:162 scored[col] = unmapped_reset[col].values[unmapped_indices]163 164 return pd.concat([scored, pre_mapped], ignore_index=True)165 166 167def _size_bin_lookup(history_df: pd.DataFrame, results: pd.DataFrame,168 attr_key_cols: list, mdm_col: str):169 """170 Map raw numeric SIZE values onto labelled size bins via an in-memory SQL171 range join — the most readable and efficient approach for range lookups.172 173 Returns (results_with_size_col, aggregated_size_sales).174 """175 size_limits = history_df[[mdm_col]].drop_duplicates().dropna()176 size_limits['lower'] = np.where(177 size_limits['SIZE'].str.contains('LESS'), 0,178 size_limits['SIZE'].str.findall(r'(\d+(?:\.\d+)?)').str[0],179 ).astype('float')180 size_limits['higher'] = np.where(181 size_limits['SIZE'].str.contains('PLUS'), 9999,182 size_limits['SIZE'].str.findall(r'(\d+(?:\.\d+)?)').str[1],183 )184 size_limits['higher'] = np.where(185 size_limits['SIZE'].str.contains('LESS'),186 size_limits['SIZE'].str.findall(r'(\d+(?:\.\d+)?)').str[0],187 size_limits['higher'],188 ).astype('float')189 190 results['combinedkey'] = results['combinedkey'].astype('float')191 history_df[attr_key_cols[0]] = history_df[attr_key_cols[0]].astype('float')192 193 conn = sqlite3.connect(':memory:')194 history_df.to_sql('history_df', conn, index=False)195 size_limits.to_sql('size_limits', conn, index=False)196 results.to_sql('results', conn, index=False)197 198 results = pd.read_sql_query(199 "SELECT DISTINCT A.*, B.SIZE FROM results A "200 "LEFT JOIN size_limits B ON A.combinedkey >= B.lower AND A.combinedkey < B.higher",201 conn,202 )203 agg = pd.read_sql_query(204 f"SELECT {attr_key_cols[0]}, B.SIZE, SUM(TOTAL_UNIT_SALES) AS TOTAL_UNIT_SALES "205 f"FROM history_df A LEFT JOIN size_limits B "206 f"ON {attr_key_cols[0]} BETWEEN B.lower AND B.higher "207 f"GROUP BY {attr_key_cols[0]}, B.SIZE",208 conn,209 )210 agg['combinedkey'] = agg[attr_key_cols[0]]211 return results, agg212 213 214def _build_attribute_table(flat_file_df: pd.DataFrame, meta_df: pd.DataFrame,215 history_df: pd.DataFrame):216 """217 Build the base result table by joining flat-file products against historical218 attribute mappings for every MDM attribute group.219 220 For each attribute:221 - Exact key matches carry the historical MDM value directly (score=100).222 - Unmatched rows are collected for fuzzy scoring in a later step.223 - A 'flag' column marks whether a key combo maps to one (0) or multiple (1)224 distinct historical labels, used by Ensemble to weight confidence.225 226 Returns227 -------228 tuple of:229 results – base result DataFrame (one row per flat-file product)230 historical_agg – aggregated historical records per attribute231 fuzzy_matches – scored unmatched rows per attribute232 flat_file_combos – unique key combos from the flat file per attribute233 attr_key_map – {mdm_col: [key_col, ...]}234 flag_map – {mdm_col: flag DataFrame}235 """236 flat_file_df = flat_file_df.copy()237 flat_file_df['index'] = flat_file_df.index238 239 # Start results with every column present in the flat file.240 # Each attribute loop reads key columns directly from results — no merging needed.241 results = flat_file_df.copy()242 243 historical_agg = {}244 fuzzy_matches = {}245 flat_file_combos = {}246 attr_key_map = {}247 flag_map = {}248 249 for mdm_col in meta_df['Attribute Group name'].unique():250 attr_key_cols = list(meta_df.loc[meta_df['Attribute Group name'] == mdm_col,251 'Attribute Name in MDM'])252 253 # Key columns are already present in results (flat_file_df.copy()),254 # so no merge is needed here.255 256 if mdm_col in history_df.columns or mdm_col.replace(' ', '_') in history_df.columns:257 # Build normalised lookup keys from attribute columns258 results['key'] = flat_file_df[attr_key_cols].fillna('').apply(_build_key, axis=1)259 history_df['combinedkey'] = history_df[attr_key_cols].fillna('').apply(_build_key, axis=1)260 261 # Aggregate historical records: best label per key by sales volume262 historical = history_df.loc[263 :, history_df.columns.isin(264 attr_key_cols + ['combinedkey', mdm_col, mdm_col.replace(' ', '_'), 'TOTAL_UNIT_SALES']265 )266 ].fillna('')267 agg_cols = [c for c in historical.columns if c != 'TOTAL_UNIT_SALES']268 historical = (269 historical.groupby(agg_cols)270 .agg(TOTAL_UNIT_SALES=('TOTAL_UNIT_SALES', sum), Rec=('TOTAL_UNIT_SALES', 'count'))271 .reset_index()272 .sort_values(agg_cols + ['TOTAL_UNIT_SALES', 'Rec'])273 )274 historical = (275 historical[historical[mdm_col] != '']276 .sort_values(attr_key_cols + ['Rec'], ascending=False)277 )278 historical['rank'] = historical.groupby(attr_key_cols).cumcount(ascending=True) + 1279 historical = historical[historical['rank'] == 1]280 281 # Unique key combos from the flat file (for coverage reporting)282 ff_combos = flat_file_df.loc[283 :, flat_file_df.columns.isin(284 attr_key_cols + ['UPDATE_REQUIRED', mdm_col, mdm_col.replace(' ', '_')]285 )286 ].fillna('')287 ff_agg_cols = [c for c in ff_combos.columns if c != 'UPDATE_REQUIRED']288 ff_combos = ff_combos.groupby(ff_agg_cols).agg(Rec=('UPDATE_REQUIRED', 'count')).reset_index()289 290 # Resolve actual column name — historical FINAL sheets sometimes use291 # underscores where META uses spaces (e.g. 'PACK TYPE' vs 'PACK_TYPE').292 # Taking the first match normalises to whatever variant exists in history.293 mdm_col = list(history_df.columns[294 history_df.columns.isin([mdm_col, mdm_col.replace(' ', '_')])295 ])[0]296 attr_key_map[mdm_col] = attr_key_cols297 298 # Direct join: flat-file keys → best historical mapping by sales299 best_mapping = (300 historical[['combinedkey', mdm_col, 'TOTAL_UNIT_SALES', 'Rec']]301 .groupby(['combinedkey', mdm_col])302 .agg(TOTAL_UNIT_SALES=('TOTAL_UNIT_SALES', sum), Rec=('Rec', sum))303 .reset_index()304 .sort_values('TOTAL_UNIT_SALES', ascending=False)305 )306 best_mapping['rank'] = (307 best_mapping.groupby(['combinedkey'])['TOTAL_UNIT_SALES']308 .rank(method='dense', ascending=False)309 )310 results = pd.merge(311 results,312 best_mapping.loc[best_mapping['rank'] == 1, ['combinedkey', mdm_col]],313 left_on='key', right_on='combinedkey', how='left', suffixes=('', '_remove'),314 )315 316 # Flag: does this key combo map to one label (0) or multiple (1)?317 flag_df = history_df[attr_key_cols + [mdm_col]].drop_duplicates()318 flag_df['count'] = flag_df.groupby(attr_key_cols)[mdm_col].transform('count')319 flag_df['flag'] = (flag_df['count'] > 1).astype(int)320 flag_map[mdm_col] = flag_df[attr_key_cols + ['flag']].drop_duplicates()321 322 results = results[[c for c in results.columns if 'key' not in c and '_remove' not in c]]323 324 elif mdm_col == 'SIZE':325 results['combinedkey'] = flat_file_df[attr_key_cols].fillna('').apply(_build_key, axis=1)326 results, historical = _size_bin_lookup(history_df, results, attr_key_cols, mdm_col)327 historical['UNIT_SHARE'] = historical['TOTAL_UNIT_SALES'] / historical['TOTAL_UNIT_SALES'].sum()328 ff_combos = historical329 attr_key_map[mdm_col] = attr_key_cols330 331 else:332 history_df['combinedkey'] = history_df[attr_key_cols]333 results[mdm_col] = history_df[attr_key_cols]334 historical = history_df.loc[335 :, history_df.columns.isin(336 attr_key_cols + ['combinedkey', mdm_col, mdm_col.replace(' ', '_'), 'TOTAL_UNIT_SALES']337 )338 ].fillna('')339 agg_cols = [c for c in historical.columns if c != 'TOTAL_UNIT_SALES']340 historical = (341 historical.groupby(agg_cols)342 .agg(TOTAL_UNIT_SALES=('TOTAL_UNIT_SALES', sum), Rec=('TOTAL_UNIT_SALES', 'count'))343 .reset_index()344 .sort_values(agg_cols + ['TOTAL_UNIT_SALES', 'Rec'])345 )346 historical['rank'] = historical.groupby(agg_cols).cumcount(ascending=False) + 1347 historical = historical[historical['rank'] == 1]348 ff_combos = flat_file_df.loc[349 :, flat_file_df.columns.isin(350 attr_key_cols + ['UPDATE_REQUIRED', mdm_col, mdm_col.replace(' ', '_')]351 )352 ].fillna('missing')353 ff_agg_cols = [c for c in ff_combos.columns if c != 'UPDATE_REQUIRED']354 ff_combos = ff_combos.groupby(ff_agg_cols).agg(Rec=('UPDATE_REQUIRED', 'count')).reset_index()355 attr_key_map[mdm_col] = attr_key_cols356 357 historical_agg[mdm_col] = historical358 flat_file_combos[mdm_col] = ff_combos359 fuzzy_matches[mdm_col] = _score_unmapped_keys(historical, attr_key_cols, mdm_col)360 print(f" Lookup {mdm_col}: {len(ff_combos)} products to map | {len(historical)} matched to history")361 362 return results.drop(columns=['index']), historical_agg, fuzzy_matches, flat_file_combos, attr_key_map, flag_map363 364 365def _select_top_matches(fuzzy_matches: dict) -> dict:366 """367 Retain only the top-ranked fuzzy match per unmapped key.368 369 For SIZE the threshold is score=100 (exact bin match required).370 For all other attributes the top-scoring candidate per key is kept371 regardless of score — low-confidence suggestions are filtered in372 _build_lookup_table (threshold ≥50).373 """374 for mdm_col, match_df in fuzzy_matches.items():375 try:376 if 'unmapped' not in match_df.columns:377 # All items matched directly — no fuzzy ranking needed.378 continue379 match_df['rank'] = match_df.groupby('unmapped')['score'].rank(380 method='dense', ascending=False381 )382 if mdm_col == 'SIZE':383 # SIZE requires an exact bin match — partial matches are meaningless384 fuzzy_matches[mdm_col] = match_df[match_df['score'] == 100]385 else:386 fuzzy_matches[mdm_col] = match_df[match_df['rank'] == 1]387 except Exception as exc:388 print(f" WARNING: fuzzy rank failed for '{mdm_col}': {exc} — no lookup suggestions")389 return fuzzy_matches390 391 392def _build_flat_file_output(flat_file_df: pd.DataFrame, meta_df: pd.DataFrame):393 """394 Build the FLAT_FILE output sheet from the raw flat-file data.395 396 Adds IS_NEW_UPC and RAW_IS_ACTIVE sentinel columns, and appends the397 MDM attribute columns so the sheet carries the full template structure.398 Returns (output_df, column_list).399 """400 output = flat_file_df.copy()401 output['IS_NEW_UPC'] = 0402 output['RAW_IS_ACTIVE'] = 1403 flat_file_columns = list(output.columns) + list(meta_df['Attribute Name in MDM'])404 return output, flat_file_columns405 406 407def _build_lookup_table(mapped_df: pd.DataFrame, new_products_df: pd.DataFrame,408 attr_key_cols: list, mdm_col: str):409 """410 Build the ranked lookup table for a single attribute.411 412 Combines:413 - Directly matched products (score=100 from historical join).414 - Fuzzy-matched new products (vectorised min of token_set_ratio and415 token_sort_ratio; threshold ≥50 to suppress noise).416 417 The min-of-two-scorers approach prevents subset-match inflation —418 e.g. "KRAFT" vs "KRAFT HEINZ" scores 100 on token_set but 72 on419 token_sort, so the combined score is 72, which is more accurate.420 421 Returns (matched_df, lookup_table_df).422 lookup_table_df columns: attr_key_cols + [mdm_col, score, Rank, Record].423 'missing' values are replaced with blank before returning.424 """425 new_products_df = new_products_df.copy()426 new_products_df['mapped'] = new_products_df[attr_key_cols].fillna('').apply(_build_key, axis=1)427 directly_matched = pd.merge(new_products_df, mapped_df[['mapped', mdm_col, 'score']], how='inner')428 remaining = new_products_df[~new_products_df['mapped'].isin(directly_matched['mapped'])]429 430 if remaining.empty:431 lookup_table = directly_matched[attr_key_cols + [mdm_col]].fillna(100).drop_duplicates()432 lookup_table['score'] = 100433 lookup_table[mdm_col] = lookup_table[mdm_col].astype(str)434 else:435 remaining_r = remaining.drop(columns=mdm_col, errors='ignore').reset_index(drop=True)436 mapped_lkp = mapped_df[['mapped', mdm_col]].drop_duplicates('mapped').reset_index(drop=True)437 438 u_keys = remaining_r['mapped'].tolist()439 m_keys = mapped_lkp['mapped'].tolist()440 441 mat_set = rfprocess.cdist(u_keys, m_keys, scorer=rffuzz.token_set_ratio, workers=-1)442 mat_sort = rfprocess.cdist(u_keys, m_keys, scorer=rffuzz.token_sort_ratio, workers=-1)443 scores_mat = np.minimum(mat_set, mat_sort)444 445 # Suppress numeric↔non-numeric cross-pairings (e.g. "16OZ" vs "CHEDDAR").446 # Broadcasting [N,1] != [1,M] produces an [N,M] boolean mask — True where447 # the unmapped key has digits and the mapped key does not (or vice versa).448 u_has_digits = np.array([bool(re.search(r'\d+', str(k))) for k in u_keys])449 m_has_digits = np.array([bool(re.search(r'\d+', str(k))) for k in m_keys])450 numeric_type_mismatch = u_has_digits[:, None] != m_has_digits[None, :]451 scores_mat[numeric_type_mismatch] = 0452 453 # Exact string match always → 100 (overrides scorer floating-point rounding).454 # Same [N,M] broadcasting: True wherever the unmapped key equals the mapped key.455 u_arr, m_arr = np.array(u_keys), np.array(m_keys)456 exact_match_mask = u_arr[:, None] == m_arr[None, :]457 scores_mat[exact_match_mask] = 100458 459 best_idx = np.argmax(scores_mat, axis=1)460 best_scores = scores_mat[np.arange(len(u_keys)), best_idx]461 462 fuzzy_matched = remaining_r.copy()463 fuzzy_matched['newmapping'] = fuzzy_matched['mapped']464 fuzzy_matched['mapped'] = mapped_lkp['mapped'].iloc[best_idx].values465 fuzzy_matched[mdm_col] = mapped_lkp[mdm_col].iloc[best_idx].values466 fuzzy_matched['score'] = best_scores467 468 # Below 50 is noise — suppress to avoid polluting the analyst view.469 fuzzy_matched = fuzzy_matched[fuzzy_matched['score'] >= 50]470 471 # ── "All Other" fallback ──────────────────────────────────────────472 # If a standard catch-all bucket exists in the label space, any flat-473 # file key that still has no match (score < 50 for every candidate) is474 # assigned it with score=0. This ensures every product appears in the475 # lookup table so Phase 2 can fill the cell rather than leaving it blank.476 # Score=0 guarantees QC Priority=HIGH in Ensemble so the analyst sees it.477 # Deduplicate to unique AO-style labels only — the same label will478 # appear once per historical key that uses it (e.g. 646× "AO BRAND"),479 # so iterating the raw column would give a list of length 646, not 1,480 # causing the single-AO-bucket guard to incorrectly suppress the fallback.481 ao_label_candidates = list({482 lbl for lbl in mapped_lkp[mdm_col].dropna().str.strip()483 if _AO_RE.match(lbl)484 })485 # Only apply the fallback when exactly one AO bucket exists.486 # Multiple distinct AO labels (e.g. "AO MAINSTREAM BRANDS" + "AO VALUE BRANDS")487 # are ambiguous — assigning the wrong bucket is worse than leaving488 # the cell blank for the analyst to handle.489 ao_label = ao_label_candidates[0] if len(ao_label_candidates) == 1 else None490 if ao_label is not None:491 already_matched_keys = set(fuzzy_matched['newmapping'].tolist()) if not fuzzy_matched.empty else set()492 fallback_rows = remaining_r[~remaining_r['mapped'].isin(already_matched_keys)].copy()493 if not fallback_rows.empty:494 fallback_rows[mdm_col] = ao_label495 fallback_rows['score'] = 0496 print(f" AO fallback: {len(fallback_rows)} unmatched key(s) → '{ao_label}'")497 fuzzy_matched = pd.concat([fuzzy_matched, fallback_rows], ignore_index=True)498 499 if not fuzzy_matched.empty:500 lookup_table = pd.concat([501 directly_matched[attr_key_cols + [mdm_col, 'score']],502 fuzzy_matched[attr_key_cols + [mdm_col, 'score']],503 ]).fillna(100).drop_duplicates()504 else:505 lookup_table = directly_matched[attr_key_cols + [mdm_col, 'score']]506 507 lookup_table[mdm_col] = lookup_table[mdm_col].astype(str)508 directly_matched = pd.concat([directly_matched, fuzzy_matched]).drop_duplicates()509 510 # ── Full-coverage guarantee ───────────────────────────────────────────────511 # Every key combo present in the flat file must have at least one lkp row,512 # even if the algorithm couldn't assign a value. Without this, Phase 1513 # silently drops unresolvable key combos and Phase 2's left-join returns514 # NaN with no analyst visibility. Blank-value rows (score=0) make the gap515 # explicit: the analyst sees them in the lkp, fills them in, and Phase 2516 # then picks up the corrected value — satisfying the design contract that517 # all MODELING attributes are filled before the pipeline continues.518 _ff_keys = new_products_df[attr_key_cols].drop_duplicates()519 _lkp_keys = lookup_table[attr_key_cols].drop_duplicates()520 _missing = pd.merge(_ff_keys, _lkp_keys, on=attr_key_cols, how='left', indicator=True)521 _missing = _missing[_missing['_merge'] == 'left_only'].drop(columns=['_merge'])522 if not _missing.empty:523 _missing = _missing.copy()524 _missing[mdm_col] = ''525 _missing['score'] = 0526 # A coverage gap means key-column VALUE(S) in the new data were not present527 # in history, so Lookup can't carry a label across. It is NOT (usually) a528 # column name/alignment problem -- that would leave ~the whole attribute529 # unmatched, which the near-total note below calls out separately. Modeling530 # values aren't expected to change between refreshes, so an unseen value is531 # typically a data-entry slip (a placeholder/typo IN THE VALUE, e.g. "NA ...")532 # or a genuinely new value for ML + analyst QC. List the offending values533 # (repr() shows spacing/casing); cap the list so a large gap doesn't flood.534 _total = len(_ff_keys)535 _frac = len(_missing) / _total if _total else 1.0536 _combos = [537 ", ".join(f"{c}={v!r}" for c, v in zip(attr_key_cols, _row))538 for _row in _missing[attr_key_cols].astype(str).itertuples(index=False, name=None)539 ]540 _preview = _combos[:10]541 print(f" WARNING - Coverage gap in attribute '{mdm_col}': {len(_missing)} of "542 f"{_total} key combination(s) in the new data are not in the historical data, "543 f"so they were written blank (flagged HIGH for QC). These are unseen values in "544 f"({', '.join(attr_key_cols)}); modeling values are not expected to change "545 f"between refreshes, so review each as a data-entry slip in the value or a "546 f"genuinely new value for ML/analyst assignment:")547 for _c in _preview:548 print(f" - {_c}")549 if len(_combos) > len(_preview):550 print(f" ... and {len(_combos) - len(_preview)} more "551 f"(see the '{mdm_col}' lookup sheet -- blank rows with score 0)")552 if _frac >= 0.99:553 print(f" NOTE: nearly ALL key combos for '{mdm_col}' are unmatched -- also "554 f"check the key columns line up with history (names/structure), not just values.")555 lookup_table = pd.concat(556 [lookup_table, _missing[attr_key_cols + [mdm_col, 'score']]],557 ignore_index=True,558 )559 560 lookup_table['flag'] = 1561 lookup_table['Rank'] = (562 lookup_table[attr_key_cols + ['flag']]563 .groupby(attr_key_cols)['flag']564 .rank(method='dense', ascending=False)565 )566 lookup_table.drop(columns=['flag'], inplace=True)567 lookup_table = pd.merge(568 lookup_table,569 lookup_table[attr_key_cols + [mdm_col]]570 .groupby(attr_key_cols).agg(Record=(mdm_col, 'count')).reset_index(),571 on=attr_key_cols,572 )573 lookup_table.replace({'missing': ''}, regex=True, inplace=True)574 return directly_matched, lookup_table575 576 577# ── Public entry point ────────────────────────────────────────────────────────578 579def runLookup(flat_file_df: pd.DataFrame, meta_df: pd.DataFrame,580 history_df: pd.DataFrame, recom_dict: dict) -> tuple:581 """582 Run the full Lookup stage for all MODELING attributes defined in META.583 584 Parameters585 ----------586 flat_file_df : DataFrame587 New products to classify (the flat-file CSV, string-coerced).588 meta_df : DataFrame589 META sheet — defines attribute groups and their key columns.590 history_df : DataFrame591 Historical FINAL data — the labelled training reference.592 recom_dict : dict593 Accumulator for all predictor outputs. Lookup results are added594 as "Lookup_{attrG}" keys.595 596 Returns597 -------598 (recom_dict, history_df, flat_file_output, flag_map)599 """600 base_results, historical_agg, fuzzy_matches, flat_file_combos, attr_key_map, flag_map = (601 _build_attribute_table(flat_file_df, meta_df, history_df)602 )603 top_matches = _select_top_matches(fuzzy_matches)604 flat_file_output, _ = _build_flat_file_output(flat_file_df, meta_df)605 606 for mdm_col in historical_agg:607 try:608 _, recom_dict[f'Lookup_{mdm_col}'] = _build_lookup_table(609 top_matches[mdm_col], flat_file_combos[mdm_col], attr_key_map[mdm_col], mdm_col610 )611 except Exception as exc:612 print(f" ERROR: Lookup failed for '{mdm_col}': {exc}")613 print(traceback.format_exc())614 615 return recom_dict, history_df, flat_file_output, flag_map616