here is my progress_tracker.py from tqdm import tqdm from flask_socketio import SocketIO # The Flask app should inject its SocketIO instance here socketio: SocketIO = None class ProgressTracker: def __init__(self, total_steps, description="Processing", namespace=None): """ Args: total_steps (int): total number of steps for the progress. description (str): name of the phase (constant key) and initial description. namespace (str|None): socketio namespace to emit to. If None, no socket emission. """ # Store the fixed phase name separately self.phase = description # pbar.desc will update as description changes, but self.phase remains the key self.pbar = tqdm(total=total_steps, desc=description, unit="step") self.current_step = 0 self.namespace = namespace def update(self, increment=1, description=None): # If user provided a new description, update the bar's caption if description: self.pbar.set_description(description) self.pbar.update(increment) self.current_step += increment # Emit a progress event with the constant phase key if socketio and self.namespace: socketio.emit( 'progress', { 'phase': self.phase, # constant key 'current': self.current_step, 'total': self.pbar.total, 'description': self.pbar.desc # dynamic user text }, namespace=self.namespace ) def close(self): self.pbar.close() def __enter__(self): # Emit initial 0% event if socketio and self.namespace: socketio.emit( 'progress', { 'phase': self.phase, 'current': 0, 'total': self.pbar.total, 'description': self.pbar.desc }, namespace=self.namespace ) return self def __exit__(self, exc_type, exc_val, exc_tb): # Emit completed with the constant phase key if socketio and self.namespace: socketio.emit( 'completed', {'phase': self.phase}, namespace=self.namespace ) self.close() \ here is my db_operation.py # ─────────────────────────────────────────────────────────────────────────────── # LIBRARY IMPORTS # ─────────────────────────────────────────────────────────────────────────────── import pandas as pd import pyodbc import os import logging import re from datetime import datetime import tempfile import uuid from .progress_tracker import ProgressTracker # ─────────────────────────────────────────────────────────────────────────────── # CONFIGURATION / LOGGING SETUP # ─────────────────────────────────────────────────────────────────────────────── LOG_FORMAT = "%(asctime)s %(levelname)-8s %(message)s" logging.basicConfig(level=logging.INFO, format=LOG_FORMAT) logger = logging.getLogger(__name__) # ─────────────────────────────────────────────────────────────────────────────── # USER CONFIGURATION # ─────────────────────────────────────────────────────────────────────────────── EXCEL_PATH = r"D:\DEV PATEL\2025\SQL_DB\Data\Optionaldata-v1.xlsx" ODBC_DRIVER = "ODBC Driver 17 for SQL Server" SQL_SERVER_INSTANCE = r"DESKTOP-EQL9MBT\SQLEXPRESS" DATABASE_NAME = "MyData Warehouse" # ← replace with your existing database name TRUSTED_CONNECTION = True # If using SQL authentication, set False and supply below SQL_USERNAME = None SQL_PASSWORD = None # Fixed table names RAW_TABLE_NAME = "raw_diamond_data" PROCESSED_TABLE_NAME = "processed_diamond_data" # ─────────────────────────────────────────────────────────────────────────────── # SOURCE (CLOUD) & TARGET (SQL SERVER) CONFIG (FOR SQL-TO-SQL) # ─────────────────────────────────────────────────────────────────────────────── SRC_ODBC_DRIVER = "ODBC Driver 17 for SQL Server" SRC_SERVER = r"DESKTOP-EQL9MBT\SQLEXPRESS" # e.g. your-rds-endpoint SRC_DATABASE = "source_DB" SRC_USE_TRUSTED = True SRC_USERNAME = None SRC_PASSWORD = None # **Static** source table name SRC_TABLE_NAME = "dbo.diamond_data" TGT_ODBC_DRIVER = "ODBC Driver 17 for SQL Server" TGT_SERVER = r"DESKTOP-EQL9MBT\SQLEXPRESS" TGT_DATABASE = "MyData Warehouse" TGT_USE_TRUSTED = True TGT_USERNAME = None TGT_PASSWORD = None RAW_TABLE_NAME = "raw_diamond_data" PROCESSED_TABLE_NAME = "processed_diamond_data" # ─────────────────────────────────────────────────────────────────────────────── # VERSIONING & SQL # ─────────────────────────────────────────────────────────────────────────────── # VERSION string (you can replace this logic with whatever version logic you want) VERSION_STRING = datetime.now().strftime("%Y%m%d_%H%M%S") # Build a simple SELECT * from your static table SRC_SQL_QUERY = f"SELECT * FROM {SRC_TABLE_NAME};" # ─────────────────────────────────────────────────────────────────────────────── # STEP 0: CONNECT TO DATABASE (single DB, no CREATE DATABASE logic) # ─────────────────────────────────────────────────────────────────────────────── def get_db_connection(driver: str, server: str, database: str, use_trusted: bool, username: str = None, password: str = None) -> pyodbc.Connection: """ Open a pyodbc connection to the specified database. """ if use_trusted: conn_str = ( f"DRIVER={{{driver}}};" f"SERVER={server};" f"DATABASE={database};" "Trusted_Connection=yes;" ) else: conn_str = ( f"DRIVER={{{driver}}};" f"SERVER={server};" f"DATABASE={database};" f"UID={username};" f"PWD={password};" ) return pyodbc.connect(conn_str) # ─────────────────────────────────────────────────────────────────────────────── # STEP 1: LOAD ALL SHEET NAMES & FIND “MAX‐COLUMN” SHEET # ─────────────────────────────────────────────────────────────────────────────── def load_sheet_names(excel_path: str) -> list[str]: """ Return a list of all sheet names in the Excel file. """ xls = pd.ExcelFile(excel_path, engine="calamine") return xls.sheet_names def find_sheet_with_most_columns(excel_path: str, sheet_names: list[str]) -> tuple[str, pd.DataFrame]: """ Among all sheets, pick the one with the greatest number of columns (by reading nrows=0 for each). Then fully load that sheet into a DataFrame and return (sheet_name, full DataFrame). """ max_cols = -1 best_sheet = None best_df = None for name in sheet_names: try: temp_df = pd.read_excel(excel_path, sheet_name=name, engine="calamine", nrows=0) col_count = len(temp_df.columns) logger.debug(f"Sheet '{name}' has {col_count} columns.") if col_count > max_cols: max_cols = col_count best_sheet = name best_df = pd.read_excel(excel_path, sheet_name=name, engine="calamine", dtype=str) logger.info(f"→ '{name}' now has max columns ({col_count}).") except Exception as e: logger.warning(f"Could not read header of sheet '{name}': {e}. Skipping.") if best_sheet is None or best_df is None: raise RuntimeError("Failed to find or load any sheet with columns.") return best_sheet, best_df # ─────────────────────────────────────────────────────────────────────────────── # STEP 2: YOUR PROVIDED PREPROCESS FUNCTION (unpasted, verbatim) # ─────────────────────────────────────────────────────────────────────────────── def preprocess_max_dataframe(data: pd.DataFrame) -> pd.DataFrame: """ Applies preprocessing to the sheet with the most columns, but uses a case-insensitive + space/underscore-insensitive lookup so that 'GIA Cts', 'gia_cts', 'GiaCts', etc. all match. """ # 1) Build a normalization map: normalized → actual name. # Normalization: lowercase, remove spaces and underscores. norm_to_actual = {} for col in data.columns: key = col.lower().replace(" ", "").replace("_", "") norm_to_actual[key] = col # 2) Helper to “find” a required column by its “normalized” lookup: def find_col(required_name: str) -> str | None: """Return the actual column name in data whose normalized version matches required_name.lower().""" key = required_name.lower().replace(" ", "").replace("_", "") return norm_to_actual.get(key) # 3) List of “required” column *keys* (these are your original names, but we will look them up) required_columns = [ "EngGraphCts", "EngCts", "EngShp", "EngQua", "EngCol", "EngCut", "EngPol", "EngSym", "EngFlo", "EngNts", "EngMikly", "EngLab", "EngAmt", "EngDate", "EngCurAmt", "EngCurDate", "EngBlk", "EngWht", "EngOpen", "EngPav", "MkblCts", "MkblShp", "MkblQua", "MkblCol", "MkblCut", "MkblPol", "MkblSym", "MkblFlo", "MkblNts", "MkblMikly", "MkblLab", "MkblAmt", "MkblDate", "MkblCurAmt", "MkblCurrDate", "MkblBlk", "MkblWht", "MkblOpen", "MkblPav", "GrdCts", "GrdShp", "GrdQua", "GrdCol", "GrdCut", "GrdPol", "GrdSym", "GrdFlo", "GrdNts", "GrdMikly", "GrdBlk", "GrdWht", "GrdOpen", "GrdPav", "GrdAmt", "GrdDate", "GrdCurDate", "ByGrdCts", "ByGrdShp", "ByGrdQua", "ByGrdCol", "ByGrdCut", "ByGrdPol", "ByGrdSym", "ByGrdFlo", "ByGrdNts", "ByGrdMikly", "ByGrdBlk", "ByGrdWht", "ByGrdOpen", "ByGrdPav", "ByGrdAmt", "ByGrdDate", "ByGrdCurDate", "GiaCts", "GiaShp", "GiaQua", "GiaCol", "GiaCut", "GiaPol", "GiaSym", "GiaFlo", "GiaNts", "GiaMikly", "GiaBlk", "GiaWht", "GiaOpen", "GiaPav", "GiaAmt", "GiaDate", "GiaCurDate", ] # 4) Determine which of those required columns actually exist in data (via normalized lookup) present = [] missing = [] for req in required_columns: actual = find_col(req) if actual: present.append(actual) else: missing.append(req) if missing: logger.warning(f"Missing columns in max_df (after normalization): {missing}. They will be added as NA.") for req in missing: # Add an NA‐filled column under the “normalized” name we expect (use the exact string) # But to keep everything consistent in SQL, give it the exact required name (no spaces). data[req] = pd.NA present.append(req) # now it exists under “req” # 5) Create classification_data with just the present columns classification_data = data[present].copy() # 6) Fill NA for block/white/open/pav columns (but normalize those names too) fill_na_cols_1 = [ "EngBlk", "EngWht", "EngOpen", "EngPav", "MkblBlk", "MkblWht", "MkblOpen", "MkblPav", "GrdBlk", "GrdWht", "GrdOpen", "GrdPav", "ByGrdBlk", "ByGrdWht", "ByGrdOpen", "ByGrdPav", "GiaBlk", "GiaWht", "GiaOpen", "GiaPav" ] for req in fill_na_cols_1: actual = find_col(req) or req if actual in classification_data.columns: classification_data[actual] = classification_data[actual].fillna("NA") fill_na_cols_2 = [ 'EngShp', 'EngQua', 'EngCol', 'EngCut', 'MkblShp', 'MkblQua', 'MkblCol', 'MkblCut', 'GrdShp', 'GrdQua', 'GrdCol', 'GrdCut', 'ByGrdShp', 'ByGrdQua', 'ByGrdCol', 'ByGrdCut', 'GiaShp', 'GiaQua', 'GiaCol', 'GiaCut' ] for req in fill_na_cols_2: actual = find_col(req) or req if actual in classification_data.columns: classification_data[actual] = classification_data[actual].fillna("NA") logger.info(f"[preprocess] Rows before EngCts filter: {len(classification_data)}") # 7) Filter rows by EngCts ≠ 0 (using normalized lookup again) engcts_col = find_col("EngCts") or "EngCts" if engcts_col in classification_data.columns: classification_data[engcts_col] = pd.to_numeric(classification_data[engcts_col], errors='coerce').fillna(0) classification_data = classification_data[classification_data[engcts_col] != 0].copy() else: logger.warning("[preprocess] 'EngCts' not found; skipping EngCts filter.") logger.info(f"[preprocess] Rows after EngCts filter: {len(classification_data)}") # 8) Helper to create “change” columns, but also normalized def create_change_column(df_in: pd.DataFrame, prefix: str, col1: str, col2: str) -> pd.DataFrame: # Normalize keys actual1 = find_col(col1) or col1 actual2 = find_col(col2) or col2 if "Mkbl" in col2: out_col = f"Change_{prefix}_Eng_to_Mkbl_value" else: out_col = f"Change_{prefix}_Eng_to_Gia_value" if actual1 in df_in.columns and actual2 in df_in.columns: df_in[out_col] = df_in.apply( lambda row: f"{row[actual1]} to {row[actual2]} {prefix} change" if str(row[actual1]) != str(row[actual2]) else f"{prefix} not change", axis=1 ) logger.debug(f"[preprocess] Created '{out_col}' from '{actual1}' → '{actual2}'.") else: missing_cols = [] if actual1 not in df_in.columns: missing_cols.append(actual1) if actual2 not in df_in.columns: missing_cols.append(actual2) logger.warning(f"[preprocess] Skipping '{out_col}' because missing: {missing_cols}") return df_in # 9) Build change columns classification_data = create_change_column(classification_data, 'shape', 'EngShp', 'MkblShp') classification_data = create_change_column(classification_data, 'shape', 'EngShp', 'GiaShp') classification_data = create_change_column(classification_data, 'quality', 'EngQua', 'MkblQua') classification_data = create_change_column(classification_data, 'quality', 'EngQua', 'GiaQua') classification_data = create_change_column(classification_data, 'color', 'EngCol', 'MkblCol') classification_data = create_change_column(classification_data, 'color', 'EngCol', 'GiaCol') classification_data = create_change_column(classification_data, 'cut', 'EngCut', 'MkblCut') classification_data = create_change_column(classification_data, 'cut', 'EngCut', 'GiaCut') classification_data = create_change_column(classification_data, 'Blk', 'EngBlk', 'MkblBlk') classification_data = create_change_column(classification_data, 'Blk', 'EngBlk', 'GiaBlk') classification_data = create_change_column(classification_data, 'Wht', 'EngWht', 'MkblWht') classification_data = create_change_column(classification_data, 'Wht', 'EngWht', 'GiaWht') classification_data = create_change_column(classification_data, 'Open', 'EngOpen', 'MkblOpen') classification_data = create_change_column(classification_data, 'Open', 'EngOpen', 'GiaOpen') classification_data = create_change_column(classification_data, 'Pav', 'EngPav', 'MkblPav') classification_data = create_change_column(classification_data, 'Pav', 'EngPav', 'GiaPav') # 10) Final filter on amount/cts columns amount_cols_to_check = ["EngCts", "EngAmt"] # But we need to look up their actual names again: actual_check_cols = [] for req in amount_cols_to_check: col = find_col(req) or req if col in classification_data.columns: actual_check_cols.append(col) if actual_check_cols: for col in actual_check_cols: classification_data[col] = pd.to_numeric(classification_data[col], errors='coerce').fillna(0) initial_rows = len(classification_data) filter_condition = (classification_data[actual_check_cols] != 0).all(axis=1) classification_data = classification_data[filter_condition].copy() classification_data["StoneId"] = [str(uuid.uuid4()) for _ in range(len(classification_data))] logger.info(f"[preprocess] Rows after final amount/cts filter: {len(classification_data)} (dropped {initial_rows - len(classification_data)})") else: logger.warning(f"[preprocess] None of {amount_cols_to_check} found. Skipping final filter.") return classification_data # ─────────────────────────────────────────────────────────────────────────────── # STEP 3: DETERMINE SCHEMA FOR processed_diamond_data (ALL POSSIBLE COLUMNS) # ─────────────────────────────────────────────────────────────────────────────── # We need: # 1) every “required_columns” item (some may be missing in max_df, but we still create columns) # 2) all “change” column names that preprocess_max_dataframe might generate # 3) plus “version” and “created_date” # ─────────────────────────────────────────────────────────────────────────────── REQUIRED_COLUMNS = [ "EngGraphCts", "EngCts", "EngShp", "EngQua", "EngCol", "EngCut", "EngPol", "EngSym", "EngFlo", "EngNts", "EngMikly", "EngLab", "EngAmt", "EngDate", "EngCurAmt", "EngCurDate", "EngBlk", "EngWht", "EngOpen", "EngPav", "MkblCts", "MkblShp", "MkblQua", "MkblCol", "MkblCut", "MkblPol", "MkblSym", "MkblFlo", "MkblNts", "MkblMikly", "MkblLab", "MkblAmt", "MkblDate", "MkblCurAmt", "MkblCurrDate", "MkblBlk", "MkblWht", "MkblOpen", "MkblPav", "GrdCts", "GrdShp", "GrdQua", "GrdCol", "GrdCut", "GrdPol", "GrdSym", "GrdFlo", "GrdNts", "GrdMikly", "GrdBlk", "GrdWht", "GrdOpen", "GrdPav", "GrdAmt", "GrdDate", "GrdCurDate", "ByGrdCts", "ByGrdShp", "ByGrdQua", "ByGrdCol", "ByGrdCut", "ByGrdPol", "ByGrdSym", "ByGrdFlo", "ByGrdNts", "ByGrdMikly", "ByGrdBlk", "ByGrdWht", "ByGrdOpen", "ByGrdPav", "ByGrdAmt", "ByGrdDate", "ByGrdCurDate", "GiaCts", "GiaShp", "GiaQua", "GiaCol", "GiaCut", "GiaPol", "GiaSym", "GiaFlo", "GiaNts", "GiaMikly", "GiaBlk", "GiaWht", "GiaOpen", "GiaPav", "GiaAmt", "GiaDate", "GiaCurDate", ] CHANGE_COLUMNS = [ "Change_shape_Eng_to_Mkbl_value", "Change_shape_Eng_to_Gia_value", "Change_quality_Eng_to_Mkbl_value", "Change_quality_Eng_to_Gia_value", "Change_color_Eng_to_Mkbl_value", "Change_color_Eng_to_Gia_value", "Change_cut_Eng_to_Mkbl_value", "Change_cut_Eng_to_Gia_value", "Change_Blk_Eng_to_Mkbl_value", "Change_Blk_Eng_to_Gia_value", "Change_Wht_Eng_to_Mkbl_value", "Change_Wht_Eng_to_Gia_value", "Change_Open_Eng_to_Mkbl_value", "Change_Open_Eng_to_Gia_value", "Change_Pav_Eng_to_Mkbl_value", "Change_Pav_Eng_to_Gia_value" ] PROCESSED_COLUMNS = ["StoneId"] + REQUIRED_COLUMNS + CHANGE_COLUMNS + ["version", "created_date","model_id"] def ensure_processed_schema(cursor: pyodbc.Cursor) -> None: """ 1) If processed_diamond_data does NOT exist, CREATE it with all columns in PROCESSED_COLUMNS: - StoneId NVARCHAR(50) NULL (was NOT NULL before) - Every required‐column & change‐column as NVARCHAR(MAX) NULL - version VARCHAR(50) NULL - created_date DATETIME NOT NULL DEFAULT GETDATE() 2) If it DOES exist, ALTER TABLE to add any PROCESSED_COLUMNS that are missing. In particular, StoneId will now be added as NULLABLE so that ALTER succeeds on a non‐empty table. """ # a) Check if the table exists table_exists = cursor.execute( "SELECT 1 FROM sys.objects WHERE object_id = OBJECT_ID(?) AND type = 'U';", PROCESSED_TABLE_NAME ).fetchone() is not None if not table_exists: # Build CREATE DDL using PROCESSED_COLUMNS ddl_cols = [] for col in PROCESSED_COLUMNS: if col == "StoneId": # NOTE: create StoneId as NULLABLE here ddl_cols.append("[StoneId] NVARCHAR(50) NULL") elif col == "version": ddl_cols.append("[version] VARCHAR(50) NULL") elif col == "model_id": ddl_cols.append("[model_id] NVARCHAR(MAX) NOT NULL") elif col == "created_date": ddl_cols.append("[created_date] DATETIME NOT NULL DEFAULT GETDATE()") else: ddl_cols.append(f"[{col}] NVARCHAR(MAX) NULL") full_ddl = ",\n ".join(ddl_cols) create_sql = f""" CREATE TABLE [{PROCESSED_TABLE_NAME}] ( {full_ddl} ); """ cursor.execute(create_sql) cursor.connection.commit() logger.info(f"➕ Created new table: {PROCESSED_TABLE_NAME}") return # b) Table exists—find existing columns existing = cursor.execute( """ SELECT name FROM sys.columns WHERE object_id = OBJECT_ID(?) """, PROCESSED_TABLE_NAME ).fetchall() existing_cols = {row[0] for row in existing} # c) Figure out which PROCESSED_COLUMNS are missing needed = set(PROCESSED_COLUMNS) missing = [col for col in needed if col not in existing_cols] if not missing: logger.info(f"→ Table '{PROCESSED_TABLE_NAME}' already has all {len(needed)} columns. No ALTER needed.") return # d) ALTER TABLE ADD each missing column for col in missing: if col == "StoneId": # Change here: make StoneId NULL to allow ALTER on a non‐empty table alter_sql = f"ALTER TABLE [{PROCESSED_TABLE_NAME}] ADD [StoneId] NVARCHAR(50) NULL;" elif col == "version": alter_sql = f"ALTER TABLE [{PROCESSED_TABLE_NAME}] ADD [version] VARCHAR(50) NULL;" elif col == "model_id": alter_sql = f"ALTER TABLE [{PROCESSED_TABLE_NAME}] ADD [model_id] NVARCHAR(MAX) NOT NULL;" elif col == "created_date": alter_sql = f"ALTER TABLE [{PROCESSED_TABLE_NAME}] ADD [created_date] DATETIME NOT NULL DEFAULT GETDATE();" else: alter_sql = f"ALTER TABLE [{PROCESSED_TABLE_NAME}] ADD [{col}] NVARCHAR(MAX) NULL;" cursor.execute(alter_sql) logger.info(f" ↪ ALTER TABLE ADD COLUMN: [{col}]") cursor.connection.commit() logger.info(f"✅ Added {len(missing)} new columns to '{PROCESSED_TABLE_NAME}'.") def insert_processed_dataframe(df: pd.DataFrame, cursor: pyodbc.Cursor, conn: pyodbc.Connection) -> None: """ Insert every row of processed_max_df into processed_diamond_data in chunks of 5 000 rows. """ if df.empty: logger.warning("Processed DataFrame is empty; no rows to insert.") return # 1) Add any missing PROCESSED_COLUMNS as None for col in PROCESSED_COLUMNS: if col not in df.columns: df[col] = None # 2) Reorder so columns match exactly PROCESSED_COLUMNS df = df[PROCESSED_COLUMNS] # 3) Add version & created_date unique_id = uuid.uuid1() # <-- DO NOT use `uuid = ...` df["version"] = VERSION_STRING df["created_date"] = datetime.now() df["model_id"] = "Model_id" + str(unique_id) # 4) Build parameterized INSERT once placeholders = ", ".join("?" for _ in PROCESSED_COLUMNS) cols_escaped = ", ".join(f"[{c}]" for c in PROCESSED_COLUMNS) insert_sql = f"INSERT INTO [{PROCESSED_TABLE_NAME}] ({cols_escaped}) VALUES ({placeholders});" # 5) Convert DataFrame → list of tuples records = df.fillna(pd.NA).replace({pd.NA: None}).itertuples(index=False, name=None) records = list(records) # materialize so we can slice # 6) Do chunked executemany chunk_size = 5000 cursor.fast_executemany = True total_rows = len(records) for start in range(0, total_rows, chunk_size): end = min(start + chunk_size, total_rows) batch = records[start:end] cursor.executemany(insert_sql, batch) conn.commit() logger.info(f" → Inserted rows {start+1} through {end} of {total_rows} into [{PROCESSED_TABLE_NAME}].") logger.info(f"✅ Finished inserting all {total_rows} rows into [{PROCESSED_TABLE_NAME}].") # ─────────────────────────────────────────────────────────────────────────────── # STEP 4: “RAW” TABLE LOGIC (unchanged from previous version) # ─────────────────────────────────────────────────────────────────────────────── def gather_all_columns(excel_path: str, sheet_names: list[str]) -> list[str]: """ Read only headers from every sheet, collect union of all column names. """ superset = set() for name in sheet_names: try: header_df = pd.read_excel(excel_path, sheet_name=name, engine="calamine", nrows=0) superset.update(header_df.columns.tolist()) logger.debug(f"Sheet '{name}' columns: {header_df.columns.tolist()}") except Exception as e: logger.warning(f"Could not read header of '{name}': {e}") return sorted(superset) def ensure_raw_schema(superset_cols: list[str], cursor: pyodbc.Cursor) -> None: """ 1) If raw_diamond_data does NOT exist, CREATE it using all superset_cols, plus [version] VARCHAR(50) NULL and [created_date] DATETIME NOT NULL DEFAULT GETDATE(). 2) If it DOES exist, query sys.columns to see which columns are missing, and ALTER TABLE ADD each one. """ # a) Check if table exists: table_exists = cursor.execute( "SELECT 1 FROM sys.objects WHERE object_id = OBJECT_ID(?) AND type = 'U';", RAW_TABLE_NAME ).fetchone() is not None if not table_exists: # Build DDL using all superset_cols + version + created_date ddl_cols = [] for col in superset_cols: # All incoming columns as NVARCHAR(MAX) NULL ddl_cols.append(f"[{col}] NVARCHAR(MAX) NULL") # Add version + created_date ddl_cols.append("[version] VARCHAR(50) NULL") ddl_cols.append("[created_date] DATETIME NOT NULL DEFAULT GETDATE()") full_ddl = ",\n ".join(ddl_cols) create_sql = f""" CREATE TABLE [{RAW_TABLE_NAME}] ( {full_ddl} ); """ cursor.execute(create_sql) cursor.connection.commit() logger.info(f"➕ Created new table: {RAW_TABLE_NAME}") return # b) If table already exists, find existing columns: existing = cursor.execute( """ SELECT name FROM sys.columns WHERE object_id = OBJECT_ID(?) """, RAW_TABLE_NAME ).fetchall() existing_cols = {row[0] for row in existing} # c) Determine which of superset_cols + {version, created_date} are missing needed = set(superset_cols + ["version", "created_date"]) missing = [col for col in needed if col not in existing_cols] if not missing: logger.info(f"→ Table '{RAW_TABLE_NAME}' already has all {len(needed)} columns. No ALTER needed.") return # d) Alter the table to add each missing column for col in missing: if col == "version": alter_sql = f"ALTER TABLE [{RAW_TABLE_NAME}] ADD [version] VARCHAR(50) NULL;" elif col == "created_date": alter_sql = f"ALTER TABLE [{RAW_TABLE_NAME}] ADD [created_date] DATETIME NOT NULL DEFAULT GETDATE();" else: alter_sql = f"ALTER TABLE [{RAW_TABLE_NAME}] ADD [{col}] NVARCHAR(MAX) NULL;" cursor.execute(alter_sql) logger.info(f" ↪ ALTER TABLE ADD COLUMN: [{col}]") cursor.connection.commit() logger.info(f"✅ Completed ALTER TABLE for {len(missing)} new columns in '{RAW_TABLE_NAME}'.") def clean_dataframe_strings(df: pd.DataFrame) -> pd.DataFrame: """ Remove non‐printable ASCII (0x00‐0x1F, 0x7F‐0x9F, U+FFFD) from string columns, strip whitespace, then convert empty strings to None. """ for col in df.columns: if df[col].dtype == "object": df[col] = df[col].apply( lambda x: ( re.sub(r"[\x00-\x1F\x7F-\x9F\uFFFD]", "", str(x)).strip() if pd.notna(x) else "" ) ) df[col] = df[col].replace("", None) return df def append_sheet_to_raw(excel_path: str, sheet_name: str, superset_cols: list[str], cursor: pyodbc.Cursor, conn: pyodbc.Connection) -> None: """ 1) Read the sheet into a pandas DataFrame. 2) Clean string columns. 3) Align to superset schema via reindex(). 4) Write out a temp CSV (UTF-16 with BOM, tab-delimited). 5) Run T-SQL BULK INSERT to push all rows at once. 6) Clean up the temporary CSV. """ # 1) Read full sheet df = pd.read_excel(excel_path, sheet_name=sheet_name, engine="openpyxl", dtype=str) total_rows = len(df) if total_rows == 0: logger.warning(f"⚠️ Sheet '{sheet_name}' is empty. Skipping.") return logger.info(f"→ [RAW-BULK] Loaded '{sheet_name}' with {total_rows} rows.") # 2) Clean string columns for col in df.columns: if df[col].dtype == "object": df[col] = df[col].apply( lambda x: ( re.sub(r"[\x00-\x1F\x7F-\x9F\uFFFD]", "", str(x)).strip() if pd.notna(x) else "" ) ) df[col] = df[col].replace("", None) logger.info(f"→ [RAW-BULK] Cleaned strings in '{sheet_name}'.") # 3) Align to superset schema in one shot df = df.reindex(columns=superset_cols) # 4) Add version & created_date via concat (to avoid fragmentation) meta = pd.DataFrame({ "version": [VERSION_STRING] * total_rows, "created_date": [datetime.now()] * total_rows }) df = pd.concat([df, meta], axis=1) # 5) Write to a temporary CSV in UTF-16 (with BOM) and tab delimiters with tempfile.NamedTemporaryFile( mode="w", encoding="utf-16", # ← use "utf-16" so pandas writes a BOM delete=False, suffix=".csv", prefix=f"{sheet_name}_" ) as tmp: temp_path = tmp.name # Use sep='\t' so we can BULK INSERT with FIELDTERMINATOR = '\t' # Use lineterminator='\r\n' so SQL Server sees Windows CRLF df.to_csv(tmp, index=False, sep="\t", lineterminator="\r\n") logger.info(f"→ [RAW-BULK] Temp CSV created at: {temp_path}") # 6) Build & run T-SQL BULK INSERT # Because the file has a UTF-16 BOM, DATAFILETYPE='widechar' is correct. bulk_sql = f""" BULK INSERT [{RAW_TABLE_NAME}] FROM '{temp_path}' WITH ( FIRSTROW = 2, -- Skip header row DATAFILETYPE = 'widechar', -- Treat file as UTF-16 FIELDTERMINATOR = '\\t', -- Tab ROWTERMINATOR = '\\r\\n', -- CRLF TABLOCK, MAXERRORS = 0, BATCHSIZE = 10000 -- Push in bigger batches ); """ try: cursor.execute(bulk_sql) conn.commit() logger.info(f"→ [RAW-BULK] BULK INSERTed {total_rows} rows from '{sheet_name}' into [{RAW_TABLE_NAME}].") except Exception as ex: logger.error(f"❌ [RAW-BULK] Error during BULK INSERT of '{sheet_name}': {ex}", exc_info=True) finally: # 7) Delete the temp CSV file if os.path.exists(temp_path): os.remove(temp_path) logger.debug(f"→ Deleted temp CSV: {temp_path}") # ─────────────────────────────────────────────────────────────────────────────── # ENSURE DB EXISTS # ─────────────────────────────────────────────────────────────────────────────── def ensure_database_exists(driver, server, use_trusted, username=None, password=None, db_name="MyDataWarehouse"): # 1) connect to master conn_str = ( f"DRIVER={{{driver}}};" f"SERVER={server};" f"DATABASE=master;" f"{'Trusted_Connection=yes;' if use_trusted else f'UID={username};PWD={password};'}" ) with pyodbc.connect(conn_str, autocommit=True) as conn: cur = conn.cursor() cur.execute(f"IF DB_ID(N'{db_name}') IS NULL BEGIN CREATE DATABASE [{db_name}]; END;") conn.commit() cur.close() # In your main(), before you do get_db_connection(...) for the target: ensure_database_exists( TGT_ODBC_DRIVER, TGT_SERVER, TGT_USE_TRUSTED, username=TGT_USERNAME, password=TGT_PASSWORD, db_name=TGT_DATABASE ) # ─────────────────────────────────────────────────────────────────────────────── # MAIN EXCEL TO SQL FLOW # ─────────────────────────────────────────────────────────────────────────────── def main_process_excel_to_sql( excel_path: str, odbc_driver: str, sql_server_instance: str, database_name: str, trusted_connection: str = "yes", sql_username: str = "", sql_password: str = "", ): total_steps = 6 namespace = "/upload-excel" with ProgressTracker(total_steps, "Excel to SQL ETL", namespace=namespace) as p: try: db_conn = get_db_connection( odbc_driver, sql_server_instance, database_name, trusted_connection, sql_username, sql_password ) p.update(1, "Connected to target DB") except Exception as e: logger.critical(f"❌ Cannot connect to database '{database_name}': {e}", exc_info=True) raise cursor = db_conn.cursor() try: sheets = load_sheet_names(excel_path) logger.info(f"Found sheets: {sheets}") max_sheet, max_df = find_sheet_with_most_columns(excel_path, sheets) p.update(1, "Loaded and selected Excel sheet") logger.info(f"Max-column sheet: '{max_sheet}' with {max_df.shape[1]} columns.") #logger.info(f"All columns in max_df (raw): {list(max_df.columns)}") processed_max_df = preprocess_max_dataframe(max_df) p.update(1, "Preprocessed Excel sheet") logger.info(f"Processed max_df: {len(processed_max_df)} rows after filtering.") ensure_processed_schema(cursor) p.update(1, "Ensured schema") insert_processed_dataframe(processed_max_df, cursor, db_conn) p.update(1, "Inserted processed data") # raw_columns = gather_all_columns(excel_path, sheets) # if not raw_columns: # logger.critical("No columns found in any sheet. Exiting.") # raise RuntimeError("Excel has no sheets or no columns to import.") # logger.info(f"Superset of all raw columns (before dedupe): {raw_columns}") # clean_superset = [] # seen_lower = set() # for col in raw_columns: # low = col.lower() # if low not in seen_lower: # seen_lower.add(low) # clean_superset.append(col) # else: # logger.warning(f"Duplicate (case-insensitive) column skipped: '{col}'") # logger.info(f"Superset of all raw columns (after dedupe): {clean_superset}") # ensure_raw_schema(clean_superset, cursor) # append_sheet_to_raw(excel_path, max_sheet, clean_superset, cursor, db_conn) # ───── STEP E: EXECUTE STORED PROCEDURE ───────────────────────────── try: cursor.execute("{CALL dbo.sp_ExecuteUpdateTables}") # Ensure you consume any resultsets/messages while cursor.nextset(): pass db_conn.commit() p.update(1, "Executed SP & committed") logger.info("✅ Stored procedure executed and committed successfully.") except Exception as e: logger.error(f"❌ SP execution failed: {e}", exc_info=True) finally: cursor.close() db_conn.close() logger.info("🎉 All done. Raw and processed tables are up to date.") logger.info("EXCEL-to-SQL ETL complete.") # ─────────────────────────────────────────────────────────────────────────────── # MAIN SQL-to-SQL FLOW # ─────────────────────────────────────────────────────────────────────────────── def main_process_sql_to_sql( SRC_ODBC_DRIVER: str, SRC_SERVER: str, SRC_DATABASE: str, SRC_TABLE_NAME: str, TGT_ODBC_DRIVER: str, TGT_SERVER: str, TGT_DATABASE: str, SRC_USE_TRUSTED: str = "yes", SRC_USERNAME: str = "", SRC_PASSWORD: str = "", TGT_USE_TRUSTED: str = "yes", TGT_USERNAME: str = "", TGT_PASSWORD: str = "", ): total_steps = 6 namespace = "/connect-database" # Build a simple SELECT * from your static table SRC_SQL_QUERY = f"SELECT * FROM {SRC_TABLE_NAME};" with ProgressTracker(total_steps, "SQL to SQL ETL", namespace=namespace) as p: ensure_database_exists( TGT_ODBC_DRIVER, TGT_SERVER, TGT_USE_TRUSTED, username=TGT_USERNAME, password=TGT_PASSWORD, db_name=TGT_DATABASE ) p.update(1, "Ensured target DB exists") # 1) Extract from source src_conn = get_db_connection( SRC_ODBC_DRIVER, SRC_SERVER, SRC_DATABASE, SRC_USE_TRUSTED, SRC_USERNAME, SRC_PASSWORD ) df_raw = pd.read_sql(SRC_SQL_QUERY, src_conn) src_conn.close() p.update(1, "Extracted from source") # 2) Preprocess processed_df = preprocess_max_dataframe(df_raw) p.update(1, "Preprocessed source data") # 3) Load into target tgt_conn = get_db_connection( TGT_ODBC_DRIVER, TGT_SERVER, TGT_DATABASE, TGT_USE_TRUSTED, TGT_USERNAME, TGT_PASSWORD ) cursor = tgt_conn.cursor() ensure_processed_schema(cursor) p.update(1, "Ensured schema") insert_processed_dataframe(processed_df, cursor, tgt_conn) p.update(1, "Inserted into target") try: cursor.execute("{CALL dbo.sp_ExecuteUpdateTables}") # Ensure you consume any resultsets/messages while cursor.nextset(): pass tgt_conn.commit() p.update(1, "Executed SP & committed") logger.info("✅ Stored procedure executed and committed successfully.") except Exception as e: logger.error(f"❌ SP execution failed: {e}", exc_info=True) cursor.close() tgt_conn.close() logger.info("🎉 All done. Raw and processed tables are up to date.") logger.info("SQL-to-SQL ETL complete.") # ─────────────────────────────────────────────────────────────────────────────── # MAIN EXECUTION # ─────────────────────────────────────────────────────────────────────────────── if __name__ == "__main__": main_process_excel_to_sql( excel_path=EXCEL_PATH, odbc_driver=ODBC_DRIVER, sql_server_instance=SQL_SERVER_INSTANCE, database_name=DATABASE_NAME, trusted_connection=TRUSTED_CONNECTION, sql_username=SQL_USERNAME, sql_password=SQL_PASSWORD ) main_process_sql_to_sql( SRC_ODBC_DRIVER=SRC_ODBC_DRIVER, SRC_SERVER=SRC_SERVER, SRC_DATABASE=SRC_DATABASE, SRC_TABLE_NAME=SRC_TABLE_NAME, TGT_ODBC_DRIVER=TGT_ODBC_DRIVER, TGT_SERVER=TGT_SERVER, TGT_DATABASE=TGT_DATABASE, SRC_USE_TRUSTED=SRC_USE_TRUSTED, SRC_USERNAME=SRC_USERNAME, SRC_PASSWORD=SRC_PASSWORD, TGT_USE_TRUSTED=TGT_USE_TRUSTED, TGT_USERNAME=TGT_USERNAME, TGT_PASSWORD=TGT_PASSWORD ) here is my app.py from flask import Flask, render_template, request, redirect, url_for, flash, jsonify from flask_socketio import SocketIO import os from threading import Thread import os from werkzeug.utils import secure_filename import pandas as pd import pyodbc from datetime import datetime from utils.db_operations import main_process_excel_to_sql, main_process_sql_to_sql, get_db_connection from utils.model_training import train_all_models, train_prediction_model, train_class_models from utils.data_loader import load_and_prepare_data from utils.config import * import utils.progress_tracker as progress_tracker app = Flask(__name__) app.secret_key = 'your_secret_key_here' app.config['UPLOAD_FOLDER'] = 'uploads' app.config['ALLOWED_EXTENSIONS'] = {'xlsx', 'xls'} # Ensure upload folder exists os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True) # Initialize SocketIO and inject into progress_tracker socketio = SocketIO(app) progress_tracker.socketio = socketio def allowed_file(filename): return '.' in filename and \ filename.rsplit('.', 1)[1].lower() in app.config['ALLOWED_EXTENSIONS'] def process_excel(filepath): try: #df = pd.read_excel(filepath) # Add your processing logic here main_process_excel_to_sql( excel_path=filepath, odbc_driver=TGT_ODBC_DRIVER, sql_server_instance=TGT_SERVER, database_name=TGT_DATABASE, trusted_connection=TGT_USE_TRUSTED, sql_username=TGT_USERNAME, sql_password=TGT_PASSWORD ) return True, "File processed successfully" except Exception as e: return False, str(e) def connect_to_database(connection_details): try: main_process_sql_to_sql( SRC_ODBC_DRIVER=connection_details['SRC_ODBC_DRIVER'], SRC_SERVER=connection_details['SRC_SERVER'], SRC_DATABASE=connection_details['SRC_DATABASE'], SRC_TABLE_NAME=connection_details['SRC_TABLE_NAME'], TGT_ODBC_DRIVER=TGT_ODBC_DRIVER, TGT_SERVER=TGT_SERVER, TGT_DATABASE=TGT_DATABASE, SRC_USE_TRUSTED=connection_details['SRC_USE_TRUSTED'], SRC_USERNAME=connection_details['SRC_USERNAME'], SRC_PASSWORD=connection_details['SRC_PASSWORD'], TGT_USE_TRUSTED=TGT_USE_TRUSTED, TGT_USERNAME=TGT_USERNAME, TGT_PASSWORD=TGT_PASSWORD ) return True, "Data fetched and processed successfully" except Exception as e: return False, str(e), None def run_training(mode): global _preprocessed_df namespace = '/model_training' try: # 1) Only preprocess once if mode in ('preprocess', 'all'): _preprocessed_df = load_and_prepare_data() # 2) For prediction or all: use cached df if mode in ('prediction', 'all'): if _preprocessed_df is None: _preprocessed_df = load_and_prepare_data() train_prediction_model(_preprocessed_df) # 3) For classification or all: use cached df if mode in ('classification', 'all'): if _preprocessed_df is None: _preprocessed_df = load_and_prepare_data() # call your loop wrapper: train_class_models(_preprocessed_df) except Exception as e: socketio.emit('error', {'message': str(e)}, namespace=namespace) # Upon exit, ProgressTracker emits 'completed' @app.route('/') def home(): return render_template('index.html') @app.route('/insert-data') def insert_data(): return render_template('insert_data.html') @app.route('/view-reports', methods=['GET','POST']) def view_reports(): # 1) your lists all_pav = ['TO0-CO0', 'NA', 'TO0-CO1', 'TO1-CO0', 'TO0-CO2', 'TO1-CO1', 'TO1-CO2', 'TO2-CO0', 'TO2-CO1', 'TO2-CO2', 'TO0-CO3', 'TO3-CO0', 'TO3-CO1', 'TO3-CO2', 'TO3-CO3', 'TO1-CO3', 'TO0-PO1', 'TO0-PO0', 'TO1-PO1', 'TO1-PO0', 'TO0-PO2', 'NV-PO1', 'TO2-PO2', 'TO2-PO0', 'TO1-PO2', 'TO3-PO2', 'TO2-PO1', 'TO3-PO0', 'TO0-PO3', 'TO1-PO3', 'TO2-PO3', 'TO3-PO1', 'TO3-PO3', 'NV-CO0', 'NV-CO1', 'NV-CO2', 'NV-CO3', 'NV-NV', 'TO0-NV', 'TO1-NV', 'TO2-CO3', 'TO2-NV', 'TO3-NV','NA'] all_open = all_pav all_black = [" NV-NV", "BT0-BC0", "BT0-BC1", "BT0-BC2", "BT0-BC3", "BT0-NV", "BT1-BC0", "BT1-BC1", "BT1-BC2", "BT1-BC3", "BT1-NV", "BT2-BC0", "BT2-BC1", "BT2-BC2", "BT2-BC3", "BT2-NV", "BT3-BC0", "BT3-BC1", "BT3-BC2", "BT3-BC3", "BT3-NV", "NV-BC0", "NV-BC1", "NV-BC2", "NV-BC3",'NA'] all_white = ['WT0-NV','WT0-WC0','WT0-WC1','WT0-WC2','WT0-WC3','WT1-NV','WT1-WC0','WT1-WC1','WT1-WC2','WT1-WC3','WT2-NV','WT2-WC0', 'WT2-WC1','WT2-WC2','WT2-WC3','WT3-NV','WT3-WC0','WT3-WC1','WT3-WC2','WT3-WC3','NV-NV','NV-WC0','NV-WC1','NV-WC2','NV-WC3','NA'] all_shape = ['NA','AS','CB','CU','EM','HR','LR','MQ','OV','PE','PR','R','RD','RQ','TR'] all_cut =['NA','EX1','EX2','EX3','EX4','FG1','GD1','GD2','VG1','VG2'] all_quality = ['NA','FL','I1','I1_','I2','I2_','I3','I3_','I4','I4_','I5','I5_','I6','I7','I8','IF','SI1','SI2','SI2_','SI2M','SI3','VS1','VS2','VVS1','VVS2',] all_color = ['D','E','F','G','H','I','J','K','L','M','N','O','P','Q','R','S','T','U','V','W','X','Y','Z','NA'] # 2) map category → (table_name, column_name, value_list) CATEGORY_MAP = { 'Blk': ('Eng_Mkbl_change_Blk_tbl', 'original_Blk', all_black), 'Wht': ('Eng_Mkbl_change_Wht_tbl', 'original_Wht', all_white), 'Pav': ('Eng_Mkblchange_Pav_tbl', 'original_Pav', all_pav), 'Open': ('Eng_Mkbl_change_Open_tbl', 'original_Open', all_open), 'Shp': ('Eng_Mkbl_change_Shp_tbl', 'original_Shp', all_shape), 'Cut': ('Eng_Mkbl_change_Cut_tbl', 'original_Cut', all_cut), 'Col': ('Eng_Mkbl_change_Col_tbl', 'original_Col', all_color), 'Qua': ('Eng_Mkbl_change_Qua_tbl', 'original_Qua', all_quality), } # 3) pretty labels CATEGORY_LABELS = { 'Blk':'Black', 'Wht':'White', 'Pav':'Pav', 'Open':'Open', 'Shp':'Shape','Cut':'Cut','Col':'Color','Qua':'Quality' } #----------------------------------------------------------------------------- selected_category = request.form.get('category') # e.g. 'blk' selected_value = request.form.get('value') # e.g. 'BT2-BC0' results = None if request.method == 'POST' and selected_category and selected_value: table, column, _ = CATEGORY_MAP[selected_category] # build & run parameterized query conn = get_db_connection( TGT_ODBC_DRIVER, TGT_SERVER, TGT_DATABASE, TGT_USE_TRUSTED, TGT_USERNAME, TGT_PASSWORD ) sql = f""" SELECT *, ROUND([change %],2) AS [change %] FROM dbo.{table} WHERE {column} = ? ORDER BY [count] DESC """ cursor = conn.execute(sql, selected_value) cols = [c[0] for c in cursor.description] rows = cursor.fetchall() # convert each row to dict {col: value} results = [dict(zip(cols, r)) for r in rows] conn.close() return render_template( 'view_reports.html', category_labels=CATEGORY_LABELS, all_values={k: v for (k, (_,_,v)) in CATEGORY_MAP.items()}, selected_category=selected_category, selected_value=selected_value, results=results ) @app.route('/model-training') def model_training(): # Render the progress page; actual training is triggered via WebSocket connect return render_template('model_training.html') @socketio.on('connect', namespace='/model_training') def on_connect(): app.logger.info("Client connected to /model_training; waiting for start signal.") @socketio.on('start_training', namespace='/model_training') def on_start_training(data): """Only once client explicitly says to start, we fire the background job.""" mode = data.get('mode', 'all') Thread(target=run_training, args=(mode,)).start() @app.route('/model-prediction') def model_prediction(): return render_template('model_prediction.html') @app.route('/upload-excel', methods=['POST']) def upload_excel(): if 'file' not in request.files: flash('No file part', 'error') return redirect(url_for('insert_data')) file = request.files['file'] if file.filename == '': flash('No selected file', 'error') return redirect(url_for('insert_data')) if file and allowed_file(file.filename): filename = secure_filename(file.filename) filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename) file.save(filepath) success, message = process_excel(filepath) if success: flash('File uploaded and processed successfully!', 'success') else: flash(f'Error processing file: {message}', 'error') return redirect(url_for('insert_data')) flash('Invalid file type', 'error') return redirect(url_for('insert_data')) @app.route('/connect-database', methods=['POST']) def connect_database(): connection_details = request.form.to_dict() success, message = connect_to_database(connection_details) if success: flash(message, 'success') # Here you would typically save the processed data return redirect(url_for('insert_data')) else: flash(f'Error: {message}', 'error') return redirect(url_for('insert_data')) if __name__ == '__main__': socketio.run(app, debug=True) here is my insert_data.html {% extends "base.html" %} {% block content %}

Insert New Data

Database Type

Server Name/IP

Database Name

Port

Use Trusted Connection (Windows Authentication)

Username

Password

SQL Query

{% endblock %} {% block scripts %} {% endblock %} error is that the socket is not updateing the progress bar in html please look for issue or missing logic