CMacD/AIC_PHASE1_POC
0
1#!/usr/bin/env python2# coding: utf-83"""41_Phase_1_Attribute_Mapping.py — AIC Phase 1 Attribute Mapping (Streamlit edition)5 6Run:7 streamlit run 1_Phase_1_Attribute_Mapping.py8 streamlit run 1_Phase_1_Attribute_Mapping.py --server.port 8501 --server.address 0.0.0.09"""10 11import io12import os13import queue14import re as _re15import sys16import tempfile17import threading18import time19import traceback20import zipfile21from pathlib import Path22 23# NLTK corpora download on first run (the HF Space has internet); no bundle shipped.24import pandas as pd25import streamlit as st26from ml_package import mapping_lookup as _ml_mod27from ml_package import text_match as _tm28from ml_package import ensemble as _ens29from ml_package import ml_classifier as _rfx30from ml_package.write_results import write_results as _write_results31 32# ── Global pipeline lock — serialises sys.stdout/sys.stderr redirection ───────33# print() inside _run_pipeline is captured by redirecting the global sys.stdout/34# sys.stderr to a per-session queue stream. Both are process-wide variables, so35# concurrent pipeline threads would clobber each other's redirection. Removing36# this lock requires threading the log queue explicitly through every pipeline37# module call (a larger refactor).38_PIPELINE_LOCK = threading.Lock()39 40# ── Page config (must be first Streamlit call) ────────────────────────────────41st.set_page_config(42 page_title='AIC Phase 1 — Attribute Mapping',43 layout='wide',44 initial_sidebar_state='collapsed',45)46 47# ── Circana brand colours ─────────────────────────────────────────────────────48ACCENT = '#4E106F' # deep purple49FG_OK = '#059669' # green50FG_WARN = '#D97706' # amber51FG_ERR = '#DC2626' # red52FG_DIM = '#6B7280' # muted grey53BG_CODE = '#F9FAFB' # log area background54 55# Stage → progress fraction56_STAGE_PROGRESS = {57 'read inputs': 0.10,58 'mappinglookup': 0.22,59 'textmatch': 0.68,60 'ensemble': 0.78,61 'write output': 0.98,62}63_STAGE_LABELS = {64 'read inputs': 'Reading & validating input files…',65 'mappinglookup': 'Running exact & fuzzy lookup matching…',66 'textmatch': 'Predicting attributes for unresolved products (lookup + ML)…',67 'ensemble': 'Combining lookup + ML results and assigning QC priority…',68 'write output': 'Writing output Excel workbook…',69}70 71SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))72 73 74# ── Custom CSS ────────────────────────────────────────────────────────────────75st.markdown(f"""76<style>77 /* Header accent bar */78 .aic-header {{79 background: {ACCENT};80 color: white;81 padding: 18px 28px 14px 28px;82 border-radius: 8px;83 margin-bottom: 18px;84 }}85 .aic-header h1 {{86 margin: 0 0 4px 0;87 font-size: 1.55rem;88 font-weight: 700;89 letter-spacing: -0.3px;90 }}91 .aic-header p {{92 margin: 0;93 font-size: 0.88rem;94 opacity: 0.82;95 }}96 /* Log box */97 .log-box {{98 background: {BG_CODE};99 border: 1px solid #E5E7EB;100 border-radius: 6px;101 padding: 12px 14px;102 font-family: Consolas, monospace;103 font-size: 0.78rem;104 white-space: pre-wrap;105 max-height: 420px;106 overflow-y: auto;107 line-height: 1.55;108 }}109 /* Stage label */110 .stage-label {{111 color: {FG_DIM};112 font-size: 0.82rem;113 margin-top: 4px;114 }}115 /* Purple progress bar (overrides Streamlit default blue) */116 [data-testid="stProgressBar"] > div {{117 background-color: rgba(78, 16, 111, 0.15) !important;118 }}119 [data-testid="stProgressBar"] > div > div {{120 background-color: {ACCENT} !important;121 }}122 /* Green Run button (enabled only) */123 [data-testid="stBaseButton-primary"]:enabled {{124 background-color: #059669 !important;125 border-color: #059669 !important;126 color: white !important;127 }}128 [data-testid="stBaseButton-primary"]:enabled:hover {{129 background-color: #047857 !important;130 border-color: #047857 !important;131 }}132 /* Purple download button */133 [data-testid="stDownloadButton"] [data-testid="stBaseButton-primary"]:enabled {{134 background-color: {ACCENT} !important;135 border-color: {ACCENT} !important;136 color: white !important;137 }}138 [data-testid="stDownloadButton"] [data-testid="stBaseButton-primary"]:enabled:hover {{139 background-color: #3a0d54 !important;140 border-color: #3a0d54 !important;141 }}142 /* Purple Browse buttons (file uploaders) */143 [data-testid="stFileUploader"] [data-testid="stBaseButton-secondary"]:enabled {{144 background-color: {ACCENT} !important;145 border-color: {ACCENT} !important;146 color: white !important;147 }}148 [data-testid="stFileUploader"] [data-testid="stBaseButton-secondary"]:enabled:hover {{149 background-color: #3a0d54 !important;150 border-color: #3a0d54 !important;151 color: white !important;152 }}153 /* Red Stop button */154 [data-testid="stBaseButton-secondary"]:enabled {{155 border-color: #DC2626 !important;156 color: #DC2626 !important;157 }}158 [data-testid="stBaseButton-secondary"]:enabled:hover {{159 background-color: #DC2626 !important;160 color: white !important;161 border-color: #DC2626 !important;162 }}163</style>164""", unsafe_allow_html=True)165 166 167# ── Header ────────────────────────────────────────────────────────────────────168_logo_path = os.path.join(SCRIPT_DIR, 'Circana_logo.png')169if os.path.exists(_logo_path):170 import base64171 with open(_logo_path, 'rb') as _f:172 _logo_b64 = base64.b64encode(_f.read()).decode()173 st.markdown(174 f'<div style="text-align:right; margin-bottom:8px; margin-right:4px">'175 f'<img src="data:image/png;base64,{_logo_b64}" width="143"></div>',176 unsafe_allow_html=True,177 )178 179st.markdown("""180<div class="aic-header">181 <h1>AIC Phase 1 — Attribute Mapping</h1>182 <p>Upload your files, click <strong>Run</strong>, then download and QC the workbook before proceeding to Phase 2 & 3.</p>183</div>184""", unsafe_allow_html=True)185 186 187# ── Instructions ──────────────────────────────────────────────────────────────188with st.expander('Input file requirements', expanded=False):189 st.markdown("""190**Upload mode — ZIP (recommended)**191 192Upload a single `.zip` of your input folder. The ZIP must contain:193 194| File | Requirement |195|------|-------------|196| **Excel (.xlsx)** | Must contain a **FINAL** sheet (historical labelled data) and a **META** sheet (attribute definitions). |197| **CSV (.csv)** | New product flat file to classify. Column headers must include `ITEM_DIM_KEY` and all *Attribute Name in MDM* values listed in META. |198 199Using ZIP mode automatically pre-loads your folder into Phase 3 — no re-upload needed after Phase 1 completes.200 201---202 203**Upload mode — Individual files**204 205Upload the Excel and CSV separately. After Phase 1 you will need to supply a ZIP containing the Phase 3 tool files (`ModelInfo.txt`, `Attributes.txt`, `AttributeValues.txt`) on the Phase 3 page.206 207---208 209**Before proceeding to Phase 3** — review `File_For_Mapping_QC.xlsx` thoroughly. QC all HIGH and MEDIUM priority rows in each attribute tab before running Phase 3.210""")211 212 213# ── Upload mode ───────────────────────────────────────────────────────────────214upload_mode = st.radio(215 'Upload mode',216 ['📁 ZIP — full pipeline (Phase 1 + 2/3)', '📄 Individual files — Phase 1 only'],217 horizontal=True,218 key='upload_mode',219 label_visibility='collapsed',220)221_zip_mode = upload_mode.startswith('📁')222 223# ── File upload ───────────────────────────────────────────────────────────────224xlsx_file = None225csv_file = None226zip_file = None227 228if _zip_mode:229 zip_file = st.file_uploader(230 'Upload input folder as .zip (must contain an Excel with META+FINAL sheets and a .csv flat file)',231 type=['zip'],232 key='p1_zip',233 )234 if zip_file:235 st.caption(236 'The ZIP will supply files to **both** Phase 1 (Excel + CSV) and Phase 2/3 '237 '(ModelInfo.txt, Attributes.txt etc.). After Phase 1 completes, '238 'navigate to Phase 2/3 — your folder will already be loaded.'239 )240else:241 col_xl, col_csv = st.columns(2)242 with col_xl:243 xlsx_file = st.file_uploader('Excel file (.xlsx)', type=['xlsx', 'xls'], key='xlsx')244 with col_csv:245 csv_file = st.file_uploader('CSV flat file (.csv)', type=['csv'], key='csv')246 247 248def _extract_p1_zip(zip_bytes: bytes, dest: str):249 """250 Extract ZIP, detect single-wrapper-folder, and locate:251 - xl_path : first .xlsx that has META + FINAL sheets (not File_For_Mapping_QC)252 - csv_path: first .csv file253 Returns (effective_root, xl_path, csv_path) or raises RuntimeError.254 """255 dest_path = Path(dest)256 with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:257 zf.extractall(dest_path)258 259 # Unwrap single top-level folder260 entries = [e for e in dest_path.iterdir() if not e.name.startswith('.')]261 root = entries[0] if len(entries) == 1 and entries[0].is_dir() else dest_path262 263 # Find Excel (META + FINAL sheets, exclude File_For_Mapping_QC)264 import openpyxl265 xl_path = None266 for p in sorted(root.rglob('*.xlsx')):267 if 'file_for_mapping_qc' in p.name.lower():268 continue269 try:270 wb = openpyxl.load_workbook(str(p), read_only=True)271 names_upper = [s.upper() for s in wb.sheetnames]272 wb.close()273 if any('META' in n for n in names_upper) and any('FINAL' in n for n in names_upper):274 xl_path = str(p)275 break276 except Exception:277 pass278 279 # Find CSV280 csv_path = None281 for p in sorted(root.rglob('*.csv')):282 csv_path = str(p)283 break284 285 if not xl_path:286 raise RuntimeError(287 'No Excel file with META and FINAL sheets found in the ZIP. '288 'Check the file is included and the sheet names are correct.'289 )290 if not csv_path:291 raise RuntimeError('No .csv flat file found in the ZIP.')292 293 return str(root), xl_path, csv_path294 295 296# ── Pipeline stream (stdout → queue) ─────────────────────────────────────────297class _QueueStream:298 """Captures pipeline print() output and puts each line into a queue."""299 300 def __init__(self, q: queue.Queue):301 self._q = q302 self._buf = ''303 304 def write(self, text: str):305 self._buf += text306 while '\n' in self._buf:307 line, self._buf = self._buf.split('\n', 1)308 if line.strip():309 self._q.put(line)310 return len(text)311 312 def flush(self):313 if self._buf.strip():314 self._q.put(self._buf)315 self._buf = ''316 317 def fileno(self):318 raise OSError('_QueueStream has no file descriptor')319 320 321# ── Sentinel raised when the user clicks Stop ─────────────────────────────────322class _PipelineStopped(Exception):323 pass324 325 326def _run_pipeline(folder: str, excel_path: str, csv_path: str,327 stop_event: threading.Event | None = None):328 """329 Full pipeline. Runs in a worker thread; all print() output goes to _QueueStream.330 stop_event: if set between stages, raises _PipelineStopped.331 """332 def _maybe_stop():333 if stop_event and stop_event.is_set():334 raise _PipelineStopped()335 import numpy as np336 import pandas as pd337 import openpyxl338 from concurrent.futures import ThreadPoolExecutor339 340 # ── Read inputs ───────────────────────────────────────────────────────341 print('START read inputs')342 print(f' Reading Excel: {os.path.basename(excel_path)}')343 wb = openpyxl.load_workbook(excel_path, read_only=True)344 sheet_names = wb.sheetnames345 wb.close()346 print(f' Excel sheets found: {sheet_names}')347 348 meta_sheet = next((s for s in sheet_names if 'META' in s.upper()), None)349 final_sheet = next((s for s in sheet_names if 'FINAL' in s.upper()), None)350 if not meta_sheet:351 raise RuntimeError(f'No META sheet in {excel_path}. Found: {sheet_names}')352 if not final_sheet:353 raise RuntimeError(f'No FINAL sheet in {excel_path}. Found: {sheet_names}')354 355 print(f' Loading META sheet ({meta_sheet}) and FINAL sheet ({final_sheet})…')356 metaGridPDF = pd.read_excel(excel_path, sheet_name=meta_sheet)357 combinedAMPPDF = pd.read_excel(excel_path, sheet_name=final_sheet)358 print(f' Reading CSV: {os.path.basename(csv_path)}')359 attrGridPDF = pd.read_csv(csv_path, low_memory=False)360 print(f' Loaded — META {metaGridPDF.shape} | FINAL {combinedAMPPDF.shape} | CSV {attrGridPDF.shape}')361 362 # Strip '|' (pipe) from DESCRIPTION — some client submission systems treat it363 # as a delimiter and reject the file. Replace with a space so adjacent words364 # don't merge.365 for _df in (combinedAMPPDF, attrGridPDF):366 if 'DESCRIPTION' in _df.columns:367 _df['DESCRIPTION'] = _df['DESCRIPTION'].astype(str).str.replace('|', ' ', regex=False).str.strip()368 369 FINAL = combinedAMPPDF.copy()370 SAL_COL = 'RAW_TOTAL_DOLLARS'371 if SAL_COL not in combinedAMPPDF.columns:372 SAL_COL = next((c for c in combinedAMPPDF.columns if 'dollar' in c.lower()), None)373 print(f' WARNING: using {SAL_COL!r} as sales column')374 print('DONE read inputs')375 _maybe_stop()376 377 # ── Validate + prep META ──────────────────────────────────────────────378 for col in ('Attribute Name in MDM', 'Attribute Group name', 'Attribute_Type', 'Type'):379 if col not in metaGridPDF.columns:380 raise RuntimeError(f"META sheet missing column: '{col}'")381 382 metaGridPDF_old = metaGridPDF.copy()383 # Case-insensitive match: META sheets export Attribute_Type as "Modeling"/384 # "Reporting" (title case), not "MODELING". Strip + upper before comparing385 # (mirrors the V2 api/pipeline.py fix) — otherwise 0 attributes match.386 metaGridPDF = metaGridPDF[387 metaGridPDF['Attribute_Type'].astype(str).str.strip().str.upper() == 'MODELING'388 ]389 metaFields = list(set(390 list(metaGridPDF['Attribute Name in MDM'].unique()) +391 list(metaGridPDF['Attribute Group name'].unique())392 ))393 394 missing_flat = list(set(metaGridPDF['Attribute Name in MDM']) - set(attrGridPDF.columns))395 missing_amp = list(set(metaFields) - set(combinedAMPPDF.columns))396 if missing_flat:397 print(f' WARNING - columns missing from CSV: {missing_flat}')398 if missing_amp:399 print(f' WARNING - columns missing from FINAL: {missing_amp}')400 401 attrGridPDF = attrGridPDF.astype(object).fillna('nan').astype(str)402 combinedAMPPDF = combinedAMPPDF.astype(object).fillna('nan').astype(str)403 metaGridPDF = metaGridPDF.astype(object).fillna('nan').astype(str)404 combinedAMPPDF['TOTAL_UNIT_SALES'] = pd.to_numeric(405 combinedAMPPDF[SAL_COL], errors='coerce'406 ).fillna(0)407 408 dictRecom = {}409 410 # ── Lookup ────────────────────────────────────────────────────────────411 print('START mappinglookup')412 print(' Exact & fuzzy lookup — finds direct matches for new products using values already seen in the historical labelled data.')413 n_attrs = len(metaGridPDF['Attribute Name in MDM'].unique())414 n_new = len(attrGridPDF)415 print(f' Matching {n_new} new products against {n_attrs} attributes…')416 dictRecom, _, FLAT_FILE_OUT, dict_split_parent = _ml_mod.runLookup(417 attrGridPDF, metaGridPDF, combinedAMPPDF, dictRecom418 )419 print('DONE mappinglookup')420 _maybe_stop()421 422 # ── BM25 + XGBoost in parallel ────────────────────────────────────────423 print('START textmatch')424 print(' For products not resolved by lookup, predicting attributes from patterns in your historical data (lookup + ML)…')425 _dictRecom_tm = {}426 _dictRecom_ml = [None]427 428 def _run_tm():429 if stop_event and stop_event.is_set():430 return431 _tm.runTextMatch(metaGridPDF, combinedAMPPDF, attrGridPDF, _dictRecom_tm)432 433 def _run_ml():434 if stop_event and stop_event.is_set():435 return436 _dictRecom_ml[0] = _rfx.runML(attrGridPDF, metaGridPDF, combinedAMPPDF)437 438 with ThreadPoolExecutor(max_workers=4) as pool:439 f_tm = pool.submit(_run_tm)440 f_ml = pool.submit(_run_ml)441 f_tm.result()442 f_ml.result()443 444 dictRecom.update(_dictRecom_tm)445 if _dictRecom_ml[0]:446 dictRecom.update(_dictRecom_ml[0])447 print('DONE textmatch')448 _maybe_stop()449 450 # ── Ensemble ──────────────────────────────────────────────────────────451 print('START ensemble')452 print(' Combining ML predictions with lookup results and assigning QC priority…')453 dictEnsemble = _ens.runEnsemble(dictRecom, metaGridPDF, dict_split_parent)454 print('DONE ensemble')455 _maybe_stop()456 457 # ── Hand off to QC wizard (Excel written after analyst review) ────────458 sheets = ['FINAL', 'FLAT_FILE', 'META'] + list(dictEnsemble.keys())459 print(f'DONE pipeline — {len(dictEnsemble)} lookup sheet(s) ready for QC review: {", ".join(dictEnsemble.keys())}')460 return FINAL, FLAT_FILE_OUT, metaGridPDF_old, dictEnsemble461 462 463# ── Disk-usage helpers ───────────────────────────────────────────────────────464def _dir_size(path: str) -> int:465 """Return total bytes of all files under path (recursive)."""466 total = 0467 try:468 for dirpath, _, filenames in os.walk(path):469 for fn in filenames:470 try:471 total += os.path.getsize(os.path.join(dirpath, fn))472 except OSError:473 pass474 except OSError:475 pass476 return total477 478 479def _fmt_bytes(n: int) -> str:480 """Format byte count as human-readable string."""481 for unit in ('B', 'KB', 'MB', 'GB'):482 if n < 1024:483 return f'{n:.1f} {unit}'484 n /= 1024485 return f'{n:.1f} GB'486 487 488# ── Log filtering ─────────────────────────────────────────────────────────────489# Per-attribute detail lines (e.g. " BM25 BRAND: 100 products") are490# kept in the full download log but hidden from the UI — analysts see only491# high-level stage progress and any error/warning lines.492_DETAIL_LOG_RE = _re.compile(493 r'^\s{2,}(Lookup|BM25|ML|XGB|VOCAB|Done|Ensemble|AO\s)\s{2,}',494)495 496 497def _is_bridge_line(line: str) -> bool:498 """Return True for lines that should be visible in the UI log box."""499 low = line.lower()500 if any(w in low for w in ('error', 'exception', 'traceback', 'runtimeerror', 'warning')):501 return True502 return not _DETAIL_LOG_RE.match(line)503 504 505def _sort_lkp_df(df: pd.DataFrame) -> pd.DataFrame:506 """Pre-sort a lookup DataFrame identically to _write_lkp_sheet:507 HIGH → MEDIUM → LOW priority, No → Yes ML agreement, score ascending."""508 df = df.copy()509 _pri = {'HIGH': 0, 'MEDIUM': 1, 'LOW': 2}510 _ml = {'No': 0, 'Yes': 1}511 sort_cols, sort_asc = [], []512 if 'QC Priority' in df.columns:513 df['_s_pri'] = df['QC Priority'].map(_pri).fillna(3)514 sort_cols.append('_s_pri'); sort_asc.append(True)515 if 'ML Matches Lookup' in df.columns:516 df['_s_ml'] = df['ML Matches Lookup'].map(_ml).fillna(2)517 sort_cols.append('_s_ml'); sort_asc.append(True)518 if 'score' in df.columns:519 sort_cols.append('score'); sort_asc.append(True)520 if sort_cols:521 df = df.sort_values(sort_cols, ascending=sort_asc)522 df = df.drop(columns=[c for c in ('_s_pri', '_s_ml') if c in df.columns])523 return df.reset_index(drop=True)524 525 526def _write_qc_excel(payload: dict, qc_edits: dict, lkp_keys: list) -> None:527 """Write File_For_Mapping_QC.xlsx using analyst-edited lookup DataFrames.528 529 Sheets already reviewed are taken from qc_edits; any sheets skipped use530 the original DataFrame from the pipeline. Called on the main thread after531 the wizard completes (or is skipped), so xlsxwriter runs without GIL532 contention from the pipeline thread.533 """534 final_dict = {535 key: qc_edits[key] if key in qc_edits else payload['dictEnsemble'][key]536 for key in lkp_keys537 }538 out_path = os.path.join(st.session_state.tmpdir, 'File_For_Mapping_QC.xlsx')539 _write_results(out_path, payload['FINAL'], payload['FLAT_FILE_OUT'], payload['meta'], final_dict)540 with open(out_path, 'rb') as fh:541 _out_data = fh.read()542 st.session_state.output_bytes = _out_data543 # Update disk metrics with the final output file size544 if st.session_state.disk_metrics is not None:545 st.session_state.disk_metrics['output_bytes'] = len(_out_data)546 # Re-measure tmpdir now the output xlsx has been written547 st.session_state.disk_metrics['tmpdir_bytes'] = _dir_size(548 os.path.join(st.session_state.tmpdir)549 )550 st.session_state.progress_value = 1.0551 st.session_state.stage_label = '✓ Complete'552 553 554# ── Log line colouring ────────────────────────────────────────────────────────555def _colour_line(line: str) -> str:556 """Wrap a log line in an HTML span with the appropriate colour."""557 low = line.lower()558 if any(w in low for w in ('error', 'exception', 'traceback', 'runtimeerror')):559 colour = FG_ERR560 elif any(w in low for w in ('warn', 'warning', 'missing')):561 colour = FG_WARN562 elif any(w in low for w in ('done', 'complete', 'written', 'filled')):563 colour = FG_OK564 elif any(w in low for w in ('running', 'reading', 'building', 'writing', 'applying', 'start')):565 colour = ACCENT566 else:567 colour = '#374151'568 return f'<span style="color:{colour}">{line}</span>'569 570 571def _stage_from_line(line: str):572 """Return (stage_key, is_done) if line announces a stage transition, else None."""573 low = line.lower()574 if 'done' not in low and 'start' not in low:575 return None576 for key in _STAGE_PROGRESS:577 if key in low:578 return key, 'done' in low579 return None580 581 582# ── Session state ─────────────────────────────────────────────────────────────583for _k, _v in [584 ('running', False),585 ('stop_event', None),586 ('done_event', None),587 ('log_q', None),588 ('log_lines', []),589 ('err_holder', [None]),590 ('stopped_by_user', False),591 ('t_start', None),592 ('tmpdir', None),593 ('output_bytes', None),594 ('run_elapsed', ''),595 ('progress_value', 0.0),596 ('stage_label', ''),597 ('error_count', 0),598 ('run_log', ''),599 ('disk_metrics', None),600 # QC wizard state601 ('result_holder', None),602 ('qc_phase', None), # None | 'wizard'603 ('qc_payload', None),604 ('qc_step', 0),605 ('qc_lkp_keys', []),606 ('qc_edits', {}),607]:608 if _k not in st.session_state:609 st.session_state[_k] = _v610 611# ── Controls row ──────────────────────────────────────────────────────────────612# The stop button must be rendered at the TOP of each rerun so Streamlit613# registers clicks before the polling/rerun logic runs below.614pipeline_busy = _PIPELINE_LOCK.locked() and not st.session_state.running615run_clicked = False616stop_clicked = False617 618if st.session_state.running:619 col_stop, col_status = st.columns([2, 10])620 with col_stop:621 stop_clicked = st.button('⏹ Stop', type='secondary')622 with col_status:623 if st.session_state.t_start:624 # JS timer runs entirely in the browser — unaffected by server625 # rerun frequency or GIL contention from the pipeline threads.626 _start_ms = int(st.session_state.t_start * 1000)627 st.components.v1.html(f"""628<span id="aic-t" style="font-size:0.85rem;color:#6B7280;font-family:sans-serif">629 Pipeline running… 0:00630</span>631<script>632(function(){{633 var s0 = {_start_ms};634 function tick(){{635 var el = Math.floor((Date.now() - s0) / 1000);636 var m = Math.floor(el / 60);637 var s = el % 60;638 var el2 = document.getElementById('aic-t');639 if (el2) el2.textContent = 'Pipeline running\u2026 ' + m + ':' + (s < 10 ? '0' : '') + s;640 }}641 tick();642 setInterval(tick, 1000);643}})();644</script>645""", height=28, scrolling=False)646else:647 if _zip_mode:648 ready = zip_file is not None and not pipeline_busy649 else:650 ready = (xlsx_file is not None and csv_file is not None) and not pipeline_busy651 run_clicked = st.button('▶ Run', disabled=not ready, type='primary')652 if pipeline_busy:653 st.warning('Pipeline is currently running for another user — please wait.')654 elif not ready:655 st.caption('Upload both files to enable Run.')656 657# ── Handle stop click ─────────────────────────────────────────────────────────658if stop_clicked and st.session_state.stop_event and not st.session_state.stop_event.is_set():659 st.session_state.stop_event.set()660 st.session_state.stopped_by_user = True661 st.session_state.stage_label = '⏹ Stopping — waiting for current stage to finish…'662 663# ── Start pipeline on Run click ───────────────────────────────────────────────664if run_clicked and not st.session_state.running:665 stop_ev = threading.Event()666 done_ev = threading.Event()667 log_q = queue.Queue()668 err_hold = [None]669 670 result_holder = [None]671 st.session_state.update({672 'running': True,673 'stop_event': stop_ev,674 'done_event': done_ev,675 'log_q': log_q,676 'log_lines': [],677 'err_holder': err_hold,678 'stopped_by_user': False,679 't_start': time.time(),680 'output_bytes': None,681 'run_elapsed': '',682 'progress_value': 0.0,683 'stage_label': '',684 'error_count': 0,685 'run_log': '',686 'disk_metrics': None,687 # reset wizard state for fresh run688 'result_holder': result_holder,689 'qc_phase': None,690 'qc_payload': None,691 'qc_step': 0,692 'qc_lkp_keys': [],693 'qc_edits': {},694 })695 696 tmpdir = tempfile.mkdtemp(prefix='aic_')697 st.session_state.tmpdir = tmpdir698 699 if _zip_mode:700 # Extract ZIP and locate the Phase 1 inputs701 try:702 p1_root, xl_p, csv_p = _extract_p1_zip(zip_file.getvalue(), tmpdir)703 except RuntimeError as _e:704 st.error(str(_e))705 st.session_state.running = False706 st.stop()707 708 # Pre-populate Phase 2/3 session state so the folder is ready on page 2709 st.session_state['p3_extracted_dir'] = p1_root710 # Reset cols so Phase 2/3 detects them fresh on first render711 st.session_state['p3_raw_cols'] = []712 st.session_state['p3_all_cols'] = []713 st.session_state['p3_default_upc_col'] = ''714 st.session_state['p3_default_mfr_col'] = ''715 st.session_state['p3_mfr_values'] = []716 st.session_state['p3_brand_values'] = []717 st.session_state['p3_tool_brand_values'] = []718 else:719 xl_p = os.path.join(tmpdir, xlsx_file.name)720 csv_p = os.path.join(tmpdir, csv_file.name)721 with open(xl_p, 'wb') as fh: fh.write(xlsx_file.getvalue())722 with open(csv_p, 'wb') as fh: fh.write(csv_file.getvalue())723 724 # Capture references for the worker thread — do NOT access st.session_state725 # from inside the thread (it has no Streamlit context).726 def _worker(727 _tmpdir=tmpdir, _xl=xl_p, _csv=csv_p,728 _log_q=log_q, _stop_ev=stop_ev, _done_ev=done_ev, _err=err_hold,729 _result=result_holder,730 ):731 _stream = _QueueStream(_log_q)732 _old_out = sys.stdout733 _old_err = sys.stderr734 with _PIPELINE_LOCK:735 sys.stdout = _stream736 sys.stderr = _stream737 try:738 _result[0] = _run_pipeline(_tmpdir, _xl, _csv, stop_event=_stop_ev)739 except _PipelineStopped:740 _log_q.put('⏹ Run cancelled by user.')741 except Exception:742 for _ln in traceback.format_exc().splitlines():743 _log_q.put(_ln)744 _err[0] = traceback.format_exc()745 finally:746 _stream.flush()747 sys.stdout = _old_out748 sys.stderr = _old_err749 _done_ev.set()750 751 # Capture input file sizes before the thread starts so we can report them later752 if _zip_mode and zip_file is not None:753 _input_bytes = len(zip_file.getvalue())754 else:755 _input_bytes = (756 (len(xlsx_file.getvalue()) if xlsx_file else 0) +757 (len(csv_file.getvalue()) if csv_file else 0)758 )759 st.session_state['_input_bytes'] = _input_bytes760 761 threading.Thread(target=_worker, daemon=True).start()762 st.rerun()763 764# ── Live progress + log (runs on every rerun while pipeline is active) ─────────765if st.session_state.running or st.session_state.log_lines:766 767 # Drain any queued log lines into session state768 if st.session_state.running and st.session_state.log_q:769 while True:770 try:771 line = st.session_state.log_q.get_nowait()772 st.session_state.log_lines.append(line)773 hit = _stage_from_line(line)774 if hit:775 key, is_done = hit776 if is_done:777 st.session_state.progress_value = _STAGE_PROGRESS[key]778 st.session_state.stage_label = f'✓ {_STAGE_LABELS[key]}'779 else:780 st.session_state.stage_label = f'⟳ {_STAGE_LABELS.get(key, "")}'781 low = line.lower()782 if any(w in low for w in ('error', 'exception', 'traceback', 'runtimeerror')):783 st.session_state.error_count += 1784 except queue.Empty:785 break786 787 # Progress bar — deterministic fill based on pipeline stage reached788 if st.session_state.running or st.session_state.stage_label.startswith('✓'):789 st.progress(st.session_state.progress_value)790 elif st.session_state.stage_label.startswith('✗'):791 st.markdown(792 f'<div style="height:4px;background:{FG_ERR};border-radius:2px;margin:8px 0 2px 0;"></div>',793 unsafe_allow_html=True,794 )795 if st.session_state.stage_label:796 st.markdown(797 f'<p class="stage-label">{st.session_state.stage_label}</p>',798 unsafe_allow_html=True,799 )800 801 # Live error banner802 _nerr = st.session_state.error_count803 if _nerr > 0:804 st.warning(f'⚠ {_nerr} error{"s" if _nerr > 1 else ""} logged — see highlighted lines below')805 806 # Log box — bridge lines only in UI; full detail goes to downloadable log807 bridge = [l for l in st.session_state.log_lines if _is_bridge_line(l)]808 visible = bridge[-60:]809 html_lines = '<br>'.join(_colour_line(l) for l in visible)810 st.markdown(f'<div class="log-box">{html_lines}</div>', unsafe_allow_html=True)811 812 # ── Check for completion ───────────────────────────────────────────────813 if st.session_state.running and st.session_state.done_event and st.session_state.done_event.is_set():814 _el = time.time() - st.session_state.t_start815 _m, _s = divmod(int(_el), 60)816 st.session_state.run_elapsed = f'{_m}:{_s:02d}'817 st.session_state.run_log = '\n'.join(st.session_state.log_lines)818 819 # Measure disk usage now that the pipeline has finished writing to tmpdir820 _tmpdir_now = st.session_state.get('tmpdir', '')821 _tmpdir_size = _dir_size(_tmpdir_now) if _tmpdir_now else 0822 _input_size = st.session_state.get('_input_bytes', 0)823 _log_size = len(st.session_state.run_log.encode('utf-8', errors='replace'))824 st.session_state.disk_metrics = {825 'input_bytes': _input_size,826 'tmpdir_bytes': _tmpdir_size,827 'log_bytes': _log_size,828 }829 830 if st.session_state.stopped_by_user:831 st.session_state.progress_value = 0.0832 st.session_state.stage_label = '⏹ Stopped'833 elif (st.session_state.err_holder[0] is None834 and st.session_state.result_holder is not None835 and st.session_state.result_holder[0] is not None):836 FINAL, FLAT_FILE_OUT, metaGridPDF_old, dictEnsemble = st.session_state.result_holder[0]837 st.session_state.qc_payload = {838 'FINAL': FINAL,839 'FLAT_FILE_OUT': FLAT_FILE_OUT,840 'meta': metaGridPDF_old,841 'dictEnsemble': dictEnsemble,842 }843 st.session_state.qc_phase = 'wizard'844 st.session_state.qc_step = 0845 st.session_state.qc_lkp_keys = list(dictEnsemble.keys())846 st.session_state.qc_edits = {}847 st.session_state.progress_value = 1.0848 st.session_state.stage_label = '✓ Pipeline complete — QC review ready'849 # Persist the QC payload to disk so a browser/Spaces session drop after850 # a successful run can RELOAD the sheets (see the wizard guard below)851 # instead of forcing a full pipeline re-run.852 try:853 import os as _os, pickle as _pkl, tempfile as _tf854 _qc_dir = _os.path.join(_tf.gettempdir(), 'aic_qc_cache')855 _os.makedirs(_qc_dir, exist_ok=True)856 with open(_os.path.join(_qc_dir, 'last_payload.pkl'), 'wb') as _fh:857 _pkl.dump({'payload': st.session_state.qc_payload,858 'lkp_keys': st.session_state.qc_lkp_keys}, _fh)859 except Exception:860 pass861 else:862 st.session_state.progress_value = 0.0863 st.session_state.stage_label = '✗ Failed'864 865 st.session_state.running = False866 st.rerun()867 868 elif st.session_state.running:869 # Still running — poll again shortly.870 # 0.5 s keeps the UI responsive without hammering the server with871 # 12 reruns/second (which caused apparent freezes on HF Spaces).872 time.sleep(0.5)873 st.rerun()874 875# ── QC Wizard ─────────────────────────────────────────────────────────────────876if st.session_state.qc_phase == 'wizard':877 from st_aggrid import AgGrid, GridOptionsBuilder, GridUpdateMode, DataReturnMode878 879 payload = st.session_state.qc_payload880 lkp_keys = st.session_state.qc_lkp_keys881 step = st.session_state.qc_step882 total = len(lkp_keys)883 884 # Guard: distinguish a genuinely-empty result from a lost session — these885 # have completely different causes and the old code conflated them, so an886 # empty pipeline result (0 attributes matched) wrongly read as a timeout.887 _guard_msg = None888 if payload is None:889 # Session state gone — e.g. HF Spaces browser/WebSocket timeout mid-run.890 # First try to recover the last successful run from disk (the completion891 # handler persisted it); only warn if there is nothing to recover.892 try:893 import os as _os, pickle as _pkl, tempfile as _tf894 _cache_f = _os.path.join(_tf.gettempdir(), 'aic_qc_cache', 'last_payload.pkl')895 if _os.path.exists(_cache_f):896 with open(_cache_f, 'rb') as _fh:897 _cached = _pkl.load(_fh)898 st.session_state.qc_payload = _cached['payload']899 st.session_state.qc_lkp_keys = _cached['lkp_keys']900 st.session_state.qc_phase = 'wizard'901 st.rerun()902 except Exception:903 pass904 _guard_msg = (905 'QC session data was lost — this can happen after a browser timeout on '906 'HuggingFace Spaces. Please re-run the pipeline to regenerate the lookup sheets.'907 )908 elif not lkp_keys:909 # Pipeline DID complete (payload present) but produced zero lookup sheets:910 # it matched 0 attributes. Almost always a META/column mismatch, NOT a911 # timeout. Give the analyst the real cause instead of the misleading one.912 _guard_msg = (913 'The pipeline finished but produced **0 lookup sheets** (0 attributes matched). '914 'This is almost always a META/column mismatch — not a session timeout. '915 'Check that each META "Attribute Name in MDM" value exists as a column in '916 'BOTH the FINAL sheet and the flat-file CSV (watch for renamed or missing '917 'RAW_ columns), then re-run.'918 )919 if _guard_msg is not None:920 (st.warning if payload is None else st.error)(_guard_msg)921 if st.button('↩ Re-run pipeline', key='btn_rerun_after_timeout'):922 st.session_state.qc_phase = None923 st.session_state.qc_payload = None924 st.session_state.qc_step = 0925 st.session_state.qc_lkp_keys = []926 st.session_state.run_elapsed = ''927 st.rerun()928 st.stop()929 930 if step >= total:931 # State drift — step advanced past last sheet without completing wizard932 st.info('All lookup sheets reviewed. Writing Excel workbook…')933 with st.spinner('Writing Excel workbook…'):934 _write_qc_excel(payload, st.session_state.qc_edits, lkp_keys)935 st.session_state.qc_phase = None936 st.rerun()937 938 st.markdown('---')939 st.markdown(940 f'<div class="aic-header" style="margin-top:12px">'941 f'<h1 style="font-size:1.2rem">QC Lookup Review</h1>'942 f'<p>Review and correct each lookup sheet before downloading the final workbook.</p>'943 f'</div>',944 unsafe_allow_html=True,945 )946 947 if step < total:948 sheet_key = lkp_keys[step]949 attr = sheet_key.replace('Final_', '').replace('_lkp', '')950 951 # Progress indicator952 st.markdown(953 f'<p class="stage-label">Sheet <strong>{step + 1} of {total}</strong>: '954 f'<code>{attr}</code></p>',955 unsafe_allow_html=True,956 )957 st.progress(step / total)958 959 # Build display DataFrame — sort, drop internal columns, clean nulls960 orig_df = payload['dictEnsemble'][sheet_key]961 display_df = _sort_lkp_df(orig_df)962 # ML Method carries pipeline-internal labels not useful to analysts963 if 'ML Method' in display_df.columns:964 display_df = display_df.drop(columns=['ML Method'])965 ml_col_raw = f'ML{attr}'966 if ml_col_raw in display_df.columns:967 display_df = display_df.rename(columns={ml_col_raw: 'ML Suggestion'})968 # Replace float NaN and the pipeline's 'nan' sentinel with blank so the969 # grid never shows "nan" — mirrors _clean() in _write_results.py970 display_df = display_df.fillna('')971 obj_cols = display_df.select_dtypes(include='object').columns972 display_df[obj_cols] = display_df[obj_cols].replace('nan', '')973 974 # Dropdown values: union of existing attribute values + ML Suggestion values975 attr_vals = (976 sorted(display_df[attr].dropna().astype(str).unique().tolist())977 if attr in display_df.columns else []978 )979 ml_vals = (980 sorted(display_df['ML Suggestion'].dropna().astype(str).unique().tolist())981 if 'ML Suggestion' in display_df.columns else []982 )983 dropdown_vals = [''] + sorted({v for v in set(attr_vals + ml_vals) if v and v != 'nan'})984 985 # ── AgGrid configuration ───────────────────────────────────────────986 gb = GridOptionsBuilder.from_dataframe(display_df)987 gb.configure_default_column(editable=False, sortable=False, filter=True, resizable=True)988 gb.configure_pagination(enabled=True, paginationAutoPageSize=False, paginationPageSize=50)989 990 if attr in display_df.columns:991 gb.configure_column(992 attr,993 editable=True,994 cellEditor='agSelectCellEditor',995 cellEditorParams={'values': dropdown_vals},996 )997 998 grid_options = gb.build()999 1000 response = AgGrid(1001 display_df,1002 gridOptions=grid_options,1003 height=480,1004 update_mode=GridUpdateMode.VALUE_CHANGED,1005 data_return_mode=DataReturnMode.AS_INPUT,1006 fit_columns_on_grid_load=False,1007 theme='streamlit',1008 reload_data=False,1009 key=f'qc_grid_{sheet_key}',1010 )1011 1012 # ── Navigation buttons ─────────────────────────────────────────────1013 btn_label = ('Save & Next →' if step < total - 11014 else 'Save & Generate Excel ✓')1015 col_next, col_skip = st.columns([3, 2])1016 1017 with col_next:1018 if st.button(btn_label, type='primary', key=f'btn_next_{step}'):1019 edited_df = pd.DataFrame(response['data'])1020 # Restore numeric dtypes that AgGrid may have stringified1021 for _nc in ('score', 'Rank'):1022 if _nc in edited_df.columns:1023 edited_df[_nc] = pd.to_numeric(edited_df[_nc], errors='coerce')1024 # Rename ML Suggestion back to original column name for write_results1025 if ml_col_raw in orig_df.columns and 'ML Suggestion' in edited_df.columns:1026 edited_df = edited_df.rename(columns={'ML Suggestion': ml_col_raw})1027 st.session_state.qc_edits[sheet_key] = edited_df1028 # Keep dictEnsemble in sync so any in-memory reads see the edits1029 payload['dictEnsemble'][sheet_key] = edited_df.copy()1030 st.session_state.qc_step += 11031 if st.session_state.qc_step >= total:1032 with st.spinner('Writing Excel workbook…'):1033 _write_qc_excel(payload, st.session_state.qc_edits, lkp_keys)1034 st.session_state.qc_phase = None1035 st.rerun()1036 1037 with col_skip:1038 if st.button('Skip remaining sheets & Download', key='btn_skip'):1039 with st.spinner('Writing Excel workbook…'):1040 _write_qc_excel(payload, st.session_state.qc_edits, lkp_keys)1041 st.session_state.qc_phase = None1042 st.rerun()1043 1044 1045# ── Post-run status + download ────────────────────────────────────────────────1046if (not st.session_state.running1047 and st.session_state.run_elapsed1048 and st.session_state.qc_phase != 'wizard'):1049 if st.session_state.output_bytes is not None:1050 st.success(f'Done in {st.session_state.run_elapsed}. Download your QC workbook below.')1051 st.download_button(1052 '⬇ Download File_For_Mapping_QC.xlsx',1053 data=st.session_state.output_bytes,1054 file_name='File_For_Mapping_QC.xlsx',1055 mime='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',1056 type='primary',1057 use_container_width=True,1058 key='p1_xlsx',1059 )1060 st.markdown('')1061 if st.button('▶▶ Save & proceed to Phase 2/3', type='primary', key='p1_to_p2'):1062 st.switch_page('pages/2_Phase_3_Pipeline_and_QC.py')1063 1064 # ── Disk usage estimate ────────────────────────────────────────────1065 _dm = st.session_state.get('disk_metrics')1066 if _dm:1067 with st.expander('Disk usage estimate (Azure sizing)', expanded=False):1068 _inp = _dm.get('input_bytes', 0)1069 _tmp = _dm.get('tmpdir_bytes', 0)1070 _out = _dm.get('output_bytes', 0)1071 _log = _dm.get('log_bytes', 0)1072 1073 # Peak per-user = everything in tmpdir at its fullest point1074 # (includes extracted inputs, working files, output xlsx)1075 _peak = max(_tmp, _inp + _out)1076 1077 # 3× covers persistent logs, Streamlit session cache,1078 # Python/model artefacts, and a safety buffer.1079 _MULTIPLIER = 31080 1081 st.markdown(1082 '**Per-user breakdown for this run** '1083 '*(actual disk written to the container temp directory)*'1084 )1085 _rows = [1086 ('Uploaded input files', _fmt_bytes(_inp)),1087 ('Peak temp directory (all files)', _fmt_bytes(_tmp)),1088 ('Output workbook (QC xlsx)', _fmt_bytes(_out) if _out else '—'),1089 ('Run log', _fmt_bytes(_log)),1090 ('**Peak per-user total**', f'**{_fmt_bytes(_peak)}**'),1091 ]1092 st.table(pd.DataFrame(_rows, columns=['Component', 'Size']))1093 1094 st.markdown('---')1095 st.markdown(1096 '**Azure storage sizing table** — peak per user × concurrent users × overhead multiplier.'1097 )1098 _az_rows = []1099 for _u in [1, 5, 10, 25, 50]:1100 _raw = _peak * _u1101 _total = _raw * _MULTIPLIER1102 _az_rows.append((str(_u), _fmt_bytes(_raw), _fmt_bytes(_total)))1103 st.table(pd.DataFrame(1104 _az_rows,1105 columns=[1106 'Concurrent users',1107 'Raw peak (users × per-user peak)',1108 f'Recommended ({_MULTIPLIER}× buffer for logs + OS overhead)',1109 ],1110 ))1111 st.caption(1112 f'Per-user peak: **{_fmt_bytes(_peak)}**. '1113 f'The {_MULTIPLIER}× multiplier covers persistent run logs, '1114 'Streamlit session cache, Python/model artefacts, and a safety buffer. '1115 'If you need logs to survive redeploys, add an Azure File Share '1116 'separately (≥ 1 GB per 1 000 runs is a reasonable starting point).'1117 )1118 1119 elif st.session_state.stopped_by_user:1120 st.warning(f'Run stopped after {st.session_state.run_elapsed}.')1121 else:1122 st.error(f'Pipeline failed after {st.session_state.run_elapsed} — check the log above for details.')1123