CMacD/AIC_PHASE1_POC
0
1"""2Phase 3 quality checks and validation functions.3 4Each function in this module corresponds to a numbered pipeline step.5Functions either modify the DataFrame (returning the updated copy) or6perform read-only QC checks that print results to the log.7 8Step Index (matching pipeline.py)9---------------------------------10 1 update_req_check – Ensure UPDATE_REQUIRED = 1 for all rows.11 2 verify_ao_cat_def – Align ASSORTMENT_CATEGORY_DEFINITION to ModelInfo.12 3 demand_group_check – Audit DEMAND_GROUP and interaction columns.13 6 check_identifier_numeric_format – Fix scientific notation / decimals in identifiers.14 8 flag_invalid_headers – Drop unnamed columns, flag non-standard names.15 9 check_special_chars – Replace '/' with ' OR ', flag special characters.1610 check_duplicate_dimkeys – Deduplicate ITEM_DIM_KEYs (keep highest dollar).1713 check_brand_tool_brand_mismatch – Flag unexpected BRAND vs TOOL_BRAND differences.1814 check_null_modeling_reporting_cols – Report nulls in modeling/reporting columns.19 split_by_raw_assortment_category – Split output by RAW_ASSORTMENT_CATEGORY (Post-QC).20"""21 22from __future__ import annotations23 24import re25from typing import Any, Dict, List, Optional, Tuple26 27import numpy as np28import pandas as pd29from pandas.api.types import (30 is_string_dtype,31 is_object_dtype,32 is_integer_dtype,33 is_float_dtype,34)35 36 37# ═══════════════════════════════════════════════════════════════════════════38# Formatting Constants & Helpers39# ═══════════════════════════════════════════════════════════════════════════40MAJOR_SEP = "=" * 7041MINOR_SEP = "-" * 6042INDENT = " "43 44 45def _indent_block(text: str, indent: str = INDENT) -> str:46 """Indent every line of *text* with the given prefix."""47 return indent + str(text).replace("\n", "\n" + indent)48 49 50def _print_step_header(step: str, title: str) -> None:51 """Print a consistently formatted step header (e.g. ``3) Demand Group Check``)."""52 print(f"\n{step}) {title}")53 print(MINOR_SEP)54 55 56def _print_df(57 df: pd.DataFrame,58 *,59 title: str | None = None,60 indent: str = INDENT,61 max_rows: int | None = None,62 min_col_width: int = 12,63) -> None:64 """Pretty-print a DataFrame for log output (no index, consistent indentation)."""65 if title:66 print(_indent_block(title, indent=indent))67 if df is None:68 print(_indent_block("(None)", indent=indent))69 return70 if df.empty:71 print(_indent_block("(no rows)", indent=indent))72 return73 74 view = df.head(max_rows) if max_rows is not None else df75 with pd.option_context("display.max_colwidth", None, "display.width", None):76 formatted = view.to_string(index=False, col_space=min_col_width)77 print(_indent_block(formatted, indent=indent))78 79 80def _is_text_dtype(series: pd.Series) -> bool:81 """Return True if *series* is string-like (object, string, or categorical)."""82 return isinstance(series.dtype, pd.CategoricalDtype) or is_string_dtype(series) or is_object_dtype(series)83 84 85# ═══════════════════════════════════════════════════════════════════════════86# Step 1: UPDATE_REQUIRED Check87# ═══════════════════════════════════════════════════════════════════════════88 89def update_req_check(df: pd.DataFrame, col: str = "UPDATE_REQUIRED") -> pd.DataFrame:90 """Set UPDATE_REQUIRED to 1 for any rows currently set to 0."""91 col_upper_map = {str(c).upper(): c for c in df.columns}92 col = col_upper_map.get(col.upper(), col)93 94 df[col] = pd.to_numeric(df[col], errors="coerce")95 zero_mask = df[col].eq(0)96 97 _print_step_header("1", "UPDATE_REQUIRED Check")98 if zero_mask.any():99 rows_changed = int(zero_mask.sum())100 df.loc[zero_mask, col] = 1101 print(f"{INDENT}✓ Converted {rows_changed} zero values to 1 in {col}.")102 else:103 print(f"{INDENT}✓ No zero values found in {col}.")104 return df105 106 107# ═══════════════════════════════════════════════════════════════════════════108# Step 2: Assortment Category Definition Check109# ═══════════════════════════════════════════════════════════════════════════110 111def verify_ao_cat_def(112 df: pd.DataFrame,113 model_info: pd.DataFrame,114 col: str = "ASSORTMENT_CATEGORY_DEFINITION",115) -> pd.DataFrame:116 """Align ASSORTMENT_CATEGORY_DEFINITION values to the ModelInfo Category_Name."""117 col_upper_map = {str(c).upper(): c for c in df.columns}118 col = col_upper_map.get(col.upper(), col)119 120 model_col_upper_map = {str(c).upper(): c for c in model_info.columns}121 category_name_col = model_col_upper_map.get("CATEGORY_NAME", "Category_Name")122 123 expected_category = model_info[category_name_col].unique()[0]124 current_values = df[col].unique()125 126 _print_step_header("2", "Assortment Category Definition Check")127 print(f"{INDENT}ModelInfo Category: {expected_category}")128 print(f"{INDENT}Current values in {col}: {list(current_values)}")129 130 # Replace any value that doesn't match the expected category131 for value in current_values:132 if value != expected_category:133 df.loc[df[col] == value, col] = expected_category134 135 print(f"{INDENT}After alignment cleaning : {list(df[col].unique())}")136 return df137 138 139# ═══════════════════════════════════════════════════════════════════════════140# Step 3: Demand Group Check141# ═══════════════════════════════════════════════════════════════════════════142 143def demand_group_check(144 df: pd.DataFrame,145 demand_group_fallback: str = None,146) -> pd.DataFrame:147 """148 Report blanks in DEMAND_GROUP and audit any interaction columns.149 150 If DEMAND_GROUP has blanks and only one distinct non-blank value,151 the blanks are auto-filled. If ALL values are blank but a fallback152 value is available from the FINAL template tab, that value is used153 with a warning. Otherwise the user is advised to check input data.154 """155 col_upper_map = {str(c).upper(): c for c in df.columns}156 demand_group_col = col_upper_map.get("DEMAND_GROUP")157 has_demand_group = demand_group_col is not None158 159 _print_step_header("3", "Demand Group Check")160 print(f"{INDENT}DEMAND_GROUP column present: {has_demand_group}")161 162 if has_demand_group:163 demand_group_series = df[demand_group_col]164 blank_mask = (165 demand_group_series.isna()166 | demand_group_series.astype(str).str.strip().eq("")167 )168 print(f"{INDENT}Has blank values: {blank_mask.any()}")169 170 if blank_mask.any():171 blank_count = int(blank_mask.sum())172 print(f"{INDENT}Blank count: {blank_count} of {len(demand_group_series)}")173 174 non_blank_values = (175 demand_group_series[~blank_mask]176 .dropna()177 .astype(str)178 .str.strip()179 .unique()180 )181 print(f"{INDENT}Unique values (non-blank): {list(non_blank_values)}")182 183 # Safe to auto-fill only when there's exactly one distinct value184 if len(non_blank_values) == 1:185 fill_value = non_blank_values[0]186 df.loc[blank_mask, demand_group_col] = fill_value187 print(f"{INDENT}✓ Replaced {blank_count} blank values with '{fill_value}'")188 elif len(non_blank_values) == 0:189 # All values blank — try FINAL template fallback190 if demand_group_fallback:191 df.loc[blank_mask, demand_group_col] = demand_group_fallback192 print(f"{INDENT}⚠ FLAT_FILE DEMAND_GROUP is entirely blank ({blank_count} rows).")193 print(f"{INDENT} Populated from FINAL template tab with: '{demand_group_fallback}'")194 print(f"{INDENT} Please verify this value matches the project scope form.")195 else:196 print(f"{INDENT}⚠ FLAT_FILE DEMAND_GROUP is entirely blank ({blank_count} rows).")197 print(f"{INDENT} No value found in FINAL template tab either.")198 print(f"{INDENT} Check the project scope form and populate DEMAND_GROUP manually.")199 else:200 print(f"{INDENT}⚠ Multiple non-blank values found — blanks not replaced.")201 202 # Audit interaction columns (if any exist)203 interaction_columns = [204 col_name for col_name in df.columns205 if "interaction" in str(col_name).lower()206 ]207 208 if not interaction_columns:209 print(f"\n{INDENT}No interaction columns found.")210 return df211 212 print(f"{INDENT}Interaction columns ({len(interaction_columns)}):")213 for col_name in interaction_columns:214 series = df[col_name]215 if _is_text_dtype(series):216 as_string = series.astype("string")217 blank_mask = as_string.isna() | as_string.str.strip().eq("")218 else:219 blank_mask = series.isna()220 221 status = f"blanks: {int(blank_mask.sum())}" if blank_mask.any() else "✓ no blanks"222 print(f"{INDENT} • {col_name}: {status}")223 224 return df225 226 227# ═══════════════════════════════════════════════════════════════════════════228# Step 6: Identifier Numeric Format Check229# ═══════════════════════════════════════════════════════════════════════════230 231def check_identifier_numeric_format(232 df: pd.DataFrame,233 cols: tuple = ("UPC10", "SKU", "ITEM_DIM_KEY"),234 show_examples: bool = True,235) -> None:236 """237 Audit identifier columns for scientific notation and decimal formatting.238 239 Auto-fixes whole-number values stored as strings or floats by converting240 them to Int64. Leaves genuinely fractional values untouched with a warning.241 """242 _print_step_header("6", "UPC Column Format Check (scientific notation / decimals)")243 244 col_upper_map = {str(c).upper(): c for c in df.columns}245 246 # Resolve requested columns to their actual DataFrame names247 resolved_columns: List[str] = []248 missing_columns: List[str] = []249 for col_name in cols:250 if col_name.upper() in col_upper_map:251 resolved_columns.append(col_upper_map[col_name.upper()])252 else:253 missing_columns.append(col_name)254 255 if missing_columns:256 print(f"{INDENT}Missing columns (skipped): {missing_columns}")257 if not resolved_columns:258 print(f"{INDENT}No valid columns to check. Exiting.")259 return260 261 print(f"{INDENT}Ensuring identifier values are in correct data format")262 263 for col_name in resolved_columns:264 series = df[col_name]265 print(f"\n{INDENT}{col_name} (dtype: {series.dtype})")266 267 # Already integer — nothing to fix268 if is_integer_dtype(series):269 print(f"{INDENT} ✓ Already integer type")270 continue271 272 series_as_string = series.astype("string")273 series_as_numeric = pd.to_numeric(series, errors="coerce")274 275 # Detection masks276 scientific_notation_mask = series_as_string.str.match(277 r"^[\+\-]?\d+(?:\.\d+)?[eE][\+\-]?\d+$", na=False278 )279 decimal_string_mask = series_as_string.str.contains(r"\.\d+", na=False)280 non_integer_mask = series_as_numeric.notna() & (np.floor(series_as_numeric) != series_as_numeric)281 282 # Report findings based on dtype283 if _is_text_dtype(series):284 print(f"{INDENT} • Strings in scientific notation: {int(scientific_notation_mask.sum())}")285 print(f"{INDENT} • Strings with decimal places: {int(decimal_string_mask.sum())}")286 print(f"{INDENT} • Numeric non-integers: {int(non_integer_mask.sum())}")287 elif is_float_dtype(series):288 not_null = series.notna()289 fractional_mask = not_null & ~np.isclose(series, np.round(series), rtol=0, atol=1e-12)290 print(f"{INDENT} • Float non-integers (fractional): {int(fractional_mask.sum())}")291 else:292 print(f"{INDENT} • Column dtype not string/float/integer; skipped.")293 294 # Auto-fix: convert whole-number values to Int64295 is_whole_number = series_as_numeric.notna() & np.isclose(296 series_as_numeric, np.round(series_as_numeric), rtol=0, atol=1e-12297 )298 299 if _is_text_dtype(series):300 if is_whole_number.any():301 df.loc[is_whole_number, col_name] = (302 pd.to_numeric(series_as_string[is_whole_number], errors="coerce")303 .round()304 .astype("Int64")305 )306 print(f"{INDENT} ✓ Auto-fix: converted {int(is_whole_number.sum())} string values to Int64.")307 308 elif is_float_dtype(series):309 not_null = series.notna()310 float_is_whole = not_null & np.isclose(series, np.round(series), rtol=0, atol=1e-12)311 312 if not_null.any():313 if (float_is_whole | ~not_null).all():314 df[col_name] = pd.Series(np.round(series), index=series.index).astype("Int64")315 print(f"{INDENT} ✓ Auto-fix: entire float column converted to Int64 (all values are whole).")316 else:317 fractional_count = int((not_null & ~float_is_whole).sum())318 print(f"{INDENT} • Auto-fix skipped: {fractional_count} non-integer values present.")319 320 321# ═══════════════════════════════════════════════════════════════════════════322# Step 8: Column Name Validation323# ═══════════════════════════════════════════════════════════════════════════324 325def flag_invalid_headers(326 df: pd.DataFrame,327 allowed_pattern: str = r"^[A-Z0-9_]+$",328) -> pd.DataFrame:329 """330 Drop auto-generated ``Unnamed: N`` columns and flag any remaining331 column names that don't match the A-Z / 0-9 / underscore convention.332 """333 column_names = pd.Index(map(str, df.columns))334 invalid_columns = column_names[~column_names.str.fullmatch(allowed_pattern)].tolist()335 336 unnamed_pattern = re.compile(r"^Unnamed:\s*\d+$", re.IGNORECASE)337 unnamed_columns = [col for col in invalid_columns if unnamed_pattern.match(col)]338 if unnamed_columns:339 df = df.drop(columns=unnamed_columns)340 341 other_invalid = [col for col in invalid_columns if col not in unnamed_columns]342 343 _print_step_header("8", "Column Name Validation")344 if other_invalid:345 print(f"{INDENT}⚠ Invalid column names: {other_invalid}")346 else:347 print(f"{INDENT}✓ All column names are valid")348 if unnamed_columns:349 print(f"{INDENT}✓ Dropped unnamed columns: {unnamed_columns}")350 351 return df352 353 354# ═══════════════════════════════════════════════════════════════════════════355# Step 9: Special Character Check356# ═══════════════════════════════════════════════════════════════════════════357 358def check_special_chars(359 df: pd.DataFrame,360 suffix: Optional[str] = "rptg",361 exclude_prefix: Optional[str] = "raw",362) -> pd.DataFrame:363 """364 Replace ``/`` with ``' OR '`` in selected columns and flag any columns365 that contain special characters (em-dash, en-dash, ampersand, etc.).366 367 DESCRIPTION columns are always excluded from replacement.368 """369 characters_to_flag = "—–&/<>="370 char_regex_class = "[" + re.escape(characters_to_flag) + "]"371 372 def _should_check_column(col_name: str) -> bool:373 """Return True if the column matches the suffix/prefix filter."""374 lower_name = str(col_name).lower()375 if lower_name == "description":376 return False377 has_suffix = True if suffix is None else lower_name.endswith(str(suffix).lower())378 is_excluded = False if exclude_prefix is None else lower_name.startswith(str(exclude_prefix).lower())379 return has_suffix and not is_excluded380 381 target_columns = [col for col in df.columns if _should_check_column(col)]382 383 _print_step_header("9", "Special Character Check")384 385 if not target_columns:386 scope_parts = []387 if suffix is not None:388 scope_parts.append(f"suffix '{suffix}'")389 if exclude_prefix is not None:390 scope_parts.append(f"not starting with prefix '{exclude_prefix}'")391 scope_description = " & ".join(scope_parts) if scope_parts else "all columns"392 print(f"{INDENT}No columns matched the selection ({scope_description}).")393 return df394 395 # --- Flag columns containing special characters ------------------------396 columns_with_specials: Dict[str, List[str]] = {}397 for col_name in target_columns:398 found_chars = df[col_name].astype(str).str.findall(char_regex_class).explode().dropna()399 if not found_chars.empty:400 unique_chars = sorted(set(found_chars.tolist()))401 if unique_chars:402 columns_with_specials[str(col_name)] = unique_chars403 404 if not columns_with_specials:405 print(f"{INDENT}✓ No special characters found.")406 else:407 print(f"{INDENT}Columns with special characters:")408 for col_name, chars in columns_with_specials.items():409 print(f"{INDENT} • {col_name}: {' '.join(chars)}")410 411 # --- Apply replacements (e.g. '/' → ' OR ') ---------------------------412 # Handle all spacing combos (A/B, A / B, A/ B, A /B) by replacing any413 # optional surrounding whitespace + slash with ' OR ', avoiding double spaces.414 columns_replaced: List[str] = []415 for col_name in target_columns:416 original_values = df[col_name].astype(str)417 updated_values = original_values418 419 if updated_values.str.contains("/", regex=False).any():420 updated_values = updated_values.str.replace(421 r" ?/ ?", " OR ", regex=True422 )423 if not original_values.equals(updated_values):424 df[col_name] = updated_values425 columns_replaced.append(col_name)426 427 if columns_replaced:428 print(f"\n{INDENT}Replacements applied ('/' → ' OR '):")429 for col_name in columns_replaced:430 print(f"{INDENT} ✓ {col_name}")431 432 return df433 434 435# ═══════════════════════════════════════════════════════════════════════════436# Step 10: Duplicate DimKey Check437# ═══════════════════════════════════════════════════════════════════════════438 439def check_duplicate_dimkeys(440 df: pd.DataFrame,441 col: str = "ITEM_DIM_KEY",442 dollars_col: str = "RAW_TOTAL_DOLLARS",443 ignore_nulls: bool = True,444 show_examples: bool = False,445 max_examples: int = 5,446) -> Tuple[pd.DataFrame, pd.DataFrame]:447 """448 Find duplicate ITEM_DIM_KEYs and deduplicate by keeping the highest-dollar row.449 450 Returns451 -------452 (deduplicated_df, duplicate_summary_df)453 duplicate_summary_df has columns [col, "Count"] listing which keys454 had duplicates and how many.455 """456 _print_step_header("10", "Duplicate DimKey Check")457 print(458 f"{INDENT}Note: An early duplicate dim key check also runs in Phase 2 "459 f"(before attribute processing) to prevent row explosion during ranking."460 )461 print(f"{INDENT}This is a safety-net check — no duplicates are expected at this stage.")462 463 col_upper_map = {str(c).upper(): c for c in df.columns}464 465 # Resolve column names (case-insensitive)466 if col.upper() not in col_upper_map:467 print(f"{INDENT}Column '{col}' not found.")468 empty_summary = pd.DataFrame(columns=[col, "Count"])469 return df, empty_summary470 col = col_upper_map[col.upper()]471 472 dollars_col_resolved = col_upper_map.get(dollars_col.upper())473 474 # Build masks for null/blank keys475 key_series = df[col]476 key_as_string = key_series.astype("string")477 is_null = key_series.isna()478 is_blank = key_as_string.str.strip().eq("")479 480 if ignore_nulls:481 valid_mask = ~(is_null | is_blank)482 keys_to_check = key_series[valid_mask]483 else:484 keys_to_check = key_series485 486 # Count occurrences and identify duplicates487 value_counts = keys_to_check.value_counts(dropna=False)488 duplicate_counts = value_counts[value_counts >= 2]489 490 if duplicate_counts.empty:491 print(f"{INDENT}✓ No duplicate values found in '{col}'.")492 empty_summary = pd.DataFrame(columns=[col, "Count"])493 return df, empty_summary494 495 # Build QC summary of duplicate keys496 duplicate_summary = (497 duplicate_counts.rename("Count")498 .reset_index()499 .rename(columns={"index": col})500 .sort_values(["Count", col], ascending=[False, True], ignore_index=True)501 )502 503 # If no dollars column, report but don't drop504 if not dollars_col_resolved:505 print(506 f"{INDENT}Found {len(duplicate_summary)} duplicate {col} values; "507 f"no '{dollars_col}' column present—no rows were dropped."508 )509 if show_examples:510 example_rows = df[df[col].isin(duplicate_summary[col])].sort_values([col]).head(max_examples)511 print(f"{INDENT}Example duplicate rows (no dollars col):")512 _print_df(example_rows[[col]], indent=INDENT)513 514 return df, duplicate_summary515 516 # Deduplicate: keep the row with the highest dollar value per key517 dollar_values = pd.to_numeric(df[dollars_col_resolved], errors="coerce")518 dollar_values_filled = dollar_values.fillna(float("-inf"))519 520 duplicate_key_set = set(duplicate_summary[col])521 is_duplicate_row = df[col].isin(duplicate_key_set)522 if ignore_nulls:523 is_duplicate_row &= ~(is_null | is_blank)524 525 # Index of the highest-dollar row for each duplicate key526 rows_to_keep = (527 dollar_values_filled[is_duplicate_row]528 .groupby(df.loc[is_duplicate_row, col], sort=False)529 .idxmax()530 )531 532 rows_to_drop = is_duplicate_row & ~df.index.isin(rows_to_keep.values)533 dropped_count = int(rows_to_drop.sum())534 deduplicated_df = df.loc[~rows_to_drop].copy()535 536 print(f"{INDENT}Found {len(duplicate_summary)} duplicate {col} values; dropped {dropped_count} row(s)")537 print(f"{INDENT}Keeping highest '{dollars_col_resolved}' per {col}.")538 539 if show_examples and dropped_count:540 display_columns = [col, dollars_col_resolved]541 kept_examples = (542 df.loc[rows_to_keep.values, display_columns]543 .sort_values([col, dollars_col_resolved], ascending=[True, False])544 .head(max_examples)545 )546 print(f"{INDENT}Kept rows (sorted by key, dollars desc):")547 _print_df(kept_examples, indent=INDENT)548 549 return deduplicated_df, duplicate_summary550 551 552# ═══════════════════════════════════════════════════════════════════════════553# Step 13: BRAND vs TOOL_BRAND Mismatch Check554# ═══════════════════════════════════════════════════════════════════════════555 556def check_brand_tool_brand_mismatch(557 df: pd.DataFrame,558 brand_col: str = "BRAND",559 tool_brand_col: str = "TOOL_BRAND",560 raw_manufacturer_col: str = "",561 valid_model_suffixes: set = None,562) -> List[Dict[str, Any]]:563 """564 Compare BRAND vs TOOL_BRAND (including model-suffixed variants) and565 report unexpected mismatches.566 567 When *raw_manufacturer_col* is provided and any mismatched BRAND or568 TOOL_BRAND value starts with ``"AO "``, the manufacturer column is569 included in the distinct-pair grouping so that each parent gets its570 own row in the resolution dialog.571 572 Returns a list of per-model mismatch groups. Each group is a dict::573 574 {575 "model_suffix": str, # "" for base, "MULO", "CONV", etc.576 "brand_col": str, # actual column name in df577 "tool_brand_col": str, # actual column name in df578 "mismatch_df": DataFrame, # columns: BRAND, TOOL_BRAND[, PARENT]579 }580 581 Only groups with at least one mismatch are included.582 Returns an empty list when there are no mismatches.583 """584 _print_step_header("13", "BRAND vs TOOL_BRAND Mismatch Check")585 586 col_upper_map = {str(c).upper(): c for c in df.columns}587 588 # --- Resolve manufacturer column for AO-brand parent inclusion ---------589 mfr_col_actual = None590 if raw_manufacturer_col:591 mfr_col_actual = col_upper_map.get(raw_manufacturer_col.upper())592 593 # --- Find all BRAND / TOOL_BRAND column pairs (including suffixed) -----594 suffix_whitelist = (595 {s.upper() for s in valid_model_suffixes} if valid_model_suffixes else None596 )597 598 def _find_brand_tool_pairs() -> List[tuple]:599 # Discover all TOOL_X / X base pairs (any base name, e.g. BRAND, SUBBRAND)600 # using the same two-pass approach as _find_all_tool_base_pairs in transforms.py.601 pairs: List[tuple] = []602 seen: set = set()603 confirmed_bases: List[str] = []604 605 # Pass 1: unambiguous base pairs (TOOL_X where X has no underscores)606 for col_key in sorted(col_upper_map.keys()):607 if not col_key.startswith("TOOL_"):608 continue609 base_upper = col_key[len("TOOL_"):]610 if not base_upper or "_" in base_upper:611 continue612 if base_upper not in col_upper_map:613 continue614 base_name = col_upper_map[base_upper]615 tool_name = col_upper_map[col_key]616 pair = (base_name, tool_name)617 if pair not in seen:618 pairs.append((base_name, tool_name, ""))619 seen.add(pair)620 confirmed_bases.append(base_upper)621 622 # Pass 2: suffixed variants of each confirmed base pair623 for base_upper in confirmed_bases:624 base_prefix = f"{base_upper}_"625 for col_key in col_upper_map:626 if not col_key.startswith(base_prefix):627 continue628 model_suffix = col_key[len(base_prefix):]629 if not model_suffix:630 continue631 if suffix_whitelist is not None and model_suffix.upper() not in suffix_whitelist:632 continue633 tool_candidate = f"TOOL_{base_upper}_{model_suffix}"634 if tool_candidate not in col_upper_map:635 continue636 brand_name = col_upper_map[col_key]637 tool_name = col_upper_map[tool_candidate]638 pair = (brand_name, tool_name)639 if pair not in seen:640 pairs.append((brand_name, tool_name, model_suffix))641 seen.add(pair)642 643 # Pass 3: when a whitelist is provided, also find TOOL_X_SUFFIX / X_SUFFIX pairs644 # that have no base TOOL_X / X column (projects where only suffixed columns exist).645 if suffix_whitelist:646 for col_key in sorted(col_upper_map.keys()):647 if not col_key.startswith("TOOL_"):648 continue649 rest = col_key[len("TOOL_"):] # e.g. "SUBBRAND_MULO"650 for suffix in sorted(suffix_whitelist):651 suffix_tag = f"_{suffix}"652 if rest.endswith(suffix_tag) and len(rest) > len(suffix_tag):653 non_tool = rest # e.g. "SUBBRAND_MULO"654 if non_tool in col_upper_map:655 brand_name = col_upper_map[non_tool]656 tool_name = col_upper_map[col_key]657 pair = (brand_name, tool_name)658 if pair not in seen:659 pairs.append((brand_name, tool_name, suffix))660 seen.add(pair)661 break # each column matches at most one suffix662 663 return pairs664 665 column_pairs = _find_brand_tool_pairs()666 667 if not column_pairs:668 print(f"{INDENT}No base/TOOL_* column pairs found. Skipping check.")669 return []670 671 # Log which pairs will be checked672 if len(column_pairs) == 1 and column_pairs[0][2] == "":673 print(f"{INDENT}Checking: {column_pairs[0][0]} vs {column_pairs[0][1]}")674 else:675 model_suffixes = [pair[2] for pair in column_pairs if pair[2]]676 if model_suffixes:677 print(f"{INDENT}Detected model variants: {', '.join(model_suffixes)}")678 print(f"{INDENT}Checking {len(column_pairs)} base/TOOL_* pair(s)")679 680 total_mismatches = 0681 mismatch_groups: List[Dict[str, Any]] = []682 683 for brand_column, tool_column, model_suffix in column_pairs:684 brand_values = df[brand_column].astype("string").str.strip().fillna("")685 tool_values = df[tool_column].astype("string").str.strip().fillna("")686 tool_values_upper = tool_values.str.upper()687 688 # Basic mismatch: BRAND != TOOL_BRAND (case-insensitive)689 mismatch_mask = brand_values.str.upper() != tool_values_upper690 691 # Exclude rows where both are blank692 both_blank = (brand_values == "") & (tool_values == "")693 mismatch_mask = mismatch_mask & ~both_blank694 695 mismatch_count = int(mismatch_mask.sum())696 total_mismatches += mismatch_count697 698 pair_label = (699 f"{brand_column} / {tool_column}"700 if not model_suffix701 else f"{brand_column} / {tool_column} (model={model_suffix})"702 )703 704 if mismatch_count == 0:705 print(f"{INDENT}{pair_label}: ✓ No mismatches")706 continue707 708 # Distinct mismatched pairs (using actual column names)709 # When AO brands are present and manufacturer column is available,710 # include the manufacturer in the grouping so each parent gets its711 # own row — helps spot incorrectly mapped client suffixes.712 include_parent = False713 if mfr_col_actual:714 brand_vals_upper = df.loc[mismatch_mask, brand_column].astype(str).str.upper()715 tool_vals_upper = df.loc[mismatch_mask, tool_column].astype(str).str.upper()716 include_parent = (717 brand_vals_upper.str.startswith("AO ").any()718 or tool_vals_upper.str.startswith("AO ").any()719 )720 721 if include_parent:722 group_cols = [brand_column, tool_column, mfr_col_actual]723 rename_map = {brand_column: "BRAND", tool_column: "TOOL_BRAND", mfr_col_actual: "PARENT"}724 else:725 group_cols = [brand_column, tool_column]726 rename_map = {brand_column: "BRAND", tool_column: "TOOL_BRAND"}727 728 distinct_pairs = (729 df.loc[mismatch_mask, group_cols]730 .drop_duplicates()731 .sort_values(group_cols)732 .reset_index(drop=True)733 )734 735 parent_note = " (incl. parent for AO brands)" if include_parent else ""736 print(f"{INDENT}{pair_label}: ⚠ {mismatch_count} row(s), {len(distinct_pairs)} distinct pair(s){parent_note} — analyst review required")737 738 # Store group with actual column names for correction targeting739 mismatch_groups.append({740 "model_suffix": model_suffix,741 "brand_col": brand_column,742 "tool_brand_col": tool_column,743 "mismatch_df": distinct_pairs.rename(columns=rename_map),744 "parent_col": mfr_col_actual if include_parent else None,745 })746 747 if total_mismatches == 0:748 print(f"{INDENT}✓ All base/TOOL_* pairs match.")749 else:750 print(f"\n{INDENT}⚠ {total_mismatches} total mismatch row(s) across {len(mismatch_groups)} pair(s) — awaiting user review")751 752 return mismatch_groups753 754 755# ═══════════════════════════════════════════════════════════════════════════756# Step 15: Null Check for Modeling / Reporting Columns757# ═══════════════════════════════════════════════════════════════════════════758 759def check_null_modeling_reporting_cols(760 df: pd.DataFrame,761 meta_df: pd.DataFrame = None,762 show_step_header: bool = True,763) -> int:764 """765 Flag modeling and reporting columns that contain null or blank values.766 767 Column classification comes from two sources:768 1. The META sheet (Attribute_Type = MODELING or REPORTING).769 2. Any column ending in ``_RPTG`` is treated as reporting.770 771 Columns with exactly one distinct non-blank value are auto-filled.772 All other flagged columns are reported with their null count and773 percentage for manual review.774 775 Returns the number of columns that were auto-filled.776 """777 if show_step_header:778 _print_step_header("15", "Null Check for Modeling/Reporting Columns")779 780 col_upper_map = {str(c).upper(): c for c in df.columns}781 782 modeling_columns: set[str] = set()783 reporting_columns: set[str] = set()784 785 # --- Classify columns from META sheet ----------------------------------786 if meta_df is not None and not meta_df.empty:787 meta_upper_map = {str(c).upper(): c for c in meta_df.columns}788 attr_type_col = meta_upper_map.get("ATTRIBUTE_TYPE")789 attr_group_col = meta_upper_map.get("ATTRIBUTE GROUP NAME")790 791 if attr_type_col and attr_group_col:792 for _, meta_row in meta_df.iterrows():793 attribute_type = str(meta_row[attr_type_col]).strip().upper()794 attribute_group = str(meta_row[attr_group_col]).strip()795 796 actual_col = col_upper_map.get(attribute_group.upper())797 if actual_col:798 if attribute_type == "MODELING":799 modeling_columns.add(actual_col)800 elif attribute_type == "REPORTING":801 reporting_columns.add(actual_col)802 else:803 print(f"{INDENT}META tab missing required columns (Attribute_Type, Attribute Group name)")804 805 # --- Also treat *_RPTG columns as reporting ----------------------------806 for col_key, actual_col in col_upper_map.items():807 if col_key.endswith("RPTG") or col_key.endswith("_RPTG"):808 reporting_columns.add(actual_col)809 810 all_target_columns = modeling_columns | reporting_columns811 if not all_target_columns:812 print(f"{INDENT}No modeling/reporting columns identified. Skipping check.")813 return 0814 815 print(f"{INDENT}Found {len(modeling_columns)} modeling column(s), {len(reporting_columns)} reporting column(s)")816 print(f"{INDENT}Total columns to check: {len(all_target_columns)}")817 818 # --- Check each column for nulls / blanks / NaN text -------------------819 total_rows = len(df)820 null_report_rows: List[Dict[str, Any]] = []821 auto_filled_count = 0822 823 _NAN_TEXT_VALUES = {"NAN", "NONE", "NULL", "N/A", "NA"}824 825 for col_name in sorted(all_target_columns):826 # Determine column classification label827 if col_name in modeling_columns and col_name in reporting_columns:828 column_type = "MODELING/RPTG"829 elif col_name in modeling_columns:830 column_type = "MODELING"831 else:832 column_type = "REPORTING"833 834 column_series = df[col_name]835 836 # Numeric columns: only check for NaN/null837 if is_integer_dtype(column_series) or is_float_dtype(column_series):838 null_mask = column_series.isna()839 null_count = int(null_mask.sum())840 nan_text_count = 0841 non_blank_series = column_series.dropna()842 else:843 # String columns: check nulls, blanks, and literal "NaN" text844 stripped_series = column_series.astype("string").str.strip()845 null_blank_mask = stripped_series.isna() | (stripped_series == "")846 nan_text_mask = stripped_series.str.upper().isin(_NAN_TEXT_VALUES)847 null_mask = null_blank_mask | nan_text_mask848 null_count = int(null_blank_mask.sum())849 nan_text_count = int((nan_text_mask & ~null_blank_mask).sum())850 non_blank_series = stripped_series[~null_mask].dropna()851 852 total_null = null_count + nan_text_count853 if total_null == 0:854 continue855 856 percent_missing = (total_null / total_rows * 100) if total_rows > 0 else 0.0857 858 # Auto-fill columns with exactly one distinct non-blank value859 action = ""860 if len(non_blank_series) > 0 and non_blank_series.nunique() == 1:861 fill_value = non_blank_series.iloc[0]862 df.loc[null_mask, col_name] = fill_value863 action = f"Auto-filled → {fill_value}"864 auto_filled_count += 1865 print(f"{INDENT} {col_name}: auto-filled {total_null} null(s) with '{fill_value}' (only value in column)")866 867 row_entry: Dict[str, Any] = {868 "Column": col_name,869 "Type": column_type,870 "Nulls": total_null,871 "% Missing": f"{percent_missing:.1f}%",872 }873 if nan_text_count > 0:874 row_entry["NaN Text"] = nan_text_count875 if action:876 row_entry["Action"] = action877 null_report_rows.append(row_entry)878 879 if not null_report_rows:880 print(f"{INDENT}✓ All modeling/reporting columns have complete data (no nulls).")881 return 0882 883 null_report_df = pd.DataFrame(null_report_rows).sort_values(884 ["Nulls", "Column"], ascending=[False, True], ignore_index=True885 )886 887 print(f"\n{INDENT}⚠ {len(null_report_df)} column(s) contain null/blank values:")888 _print_df(null_report_df, indent=INDENT + " ")889 890 if auto_filled_count:891 print(f"\n{INDENT}✓ Auto-filled {auto_filled_count} column(s) where only a single value existed.")892 893 return auto_filled_count894 895 896# ═══════════════════════════════════════════════════════════════════════════897# Step 17: RAW_ASSORTMENT_CATEGORY Split898# ═══════════════════════════════════════════════════════════════════════════899 900def split_by_raw_assortment_category(df: pd.DataFrame) -> dict[str, pd.DataFrame]:901 """902 Split the final DataFrame by RAW_ASSORTMENT_CATEGORY into separate903 DataFrames (one per category), used as individual Excel sheets.904 """905 col_upper_map = {str(c).upper(): c for c in df.columns}906 category_col = col_upper_map.get("RAW_ASSORTMENT_CATEGORY")907 908 if not category_col:909 print(f"{INDENT}Column 'RAW_ASSORTMENT_CATEGORY' is MISSING — treating all rows as a single category.")910 return {"_NO_CATEGORY_": df}911 912 category_series = df[category_col]913 has_blanks = (914 category_series.isna().any()915 or (category_series.astype(str).str.strip() == "").any()916 )917 918 # Get sorted list of unique non-blank category values919 unique_categories = (920 category_series.dropna()921 .astype(str)922 .str.strip()923 .loc[lambda s: s != ""]924 .unique()925 .tolist()926 )927 unique_categories = sorted(unique_categories)928 929 print(f"{INDENT}Blank values present: {has_blanks}")930 print(f"{INDENT}Unique categories ({len(unique_categories)}):")931 932 # Build category → DataFrame mapping933 category_splits: dict[str, pd.DataFrame] = {}934 for category_name in unique_categories:935 category_subset = df[category_series.astype(str).str.strip() == category_name].copy()936 # Excel sheet names are limited to 31 chars; also sanitize path separators937 safe_sheet_name = category_name[:30].replace("/", "_").replace("\\", "_")938 category_splits[safe_sheet_name] = category_subset939 print(f"{INDENT} {category_name}: {len(category_subset)} row(s)")940 941 # Capture blank-category rows separately942 if has_blanks:943 blank_rows = df[category_series.astype(str).str.strip() == ""]944 if not blank_rows.empty:945 category_splits["_BLANK_CATEGORY_"] = blank_rows946 print(f"{INDENT} _BLANK_CATEGORY_: {len(blank_rows)} row(s)")947 948 print(f"\n{INDENT}Split into {len(category_splits)} category group(s) for export.")949 950 return category_splits951