from flask import Flask, request, jsonify import joblib, pandas as pd, numpy as np, os from huggingface_hub import hf_hub_download # ---- Config ---- MODEL_REPO_ID = os.getenv("MODEL_REPO_ID", "johnny-five-c/superkart-rf-pipeline") MODEL_FILENAME = os.getenv("MODEL_FILENAME", "rf_tuned_pipeline.joblib") # Create Flask app (gunicorn target: app:superkart_api) superkart_api = Flask(__name__) app = superkart_api # alias so app:app also works # Download model from HF model repo into this folder (public repo = no token needed) LOCAL_PATH = os.path.join(os.path.dirname(__file__), MODEL_FILENAME) if not os.path.exists(LOCAL_PATH): LOCAL_PATH = hf_hub_download(repo_id=MODEL_REPO_ID, filename=MODEL_FILENAME, repo_type="model", local_dir=os.path.dirname(__file__)) model = joblib.load(LOCAL_PATH) # Original LC list; pipeline also expects these two extra columns: FEATURE_ORDER = [ "Product_Weight","Product_Sugar_Content","Product_Allocated_Area","Product_MRP", "Store_Size","Store_Location_City_Type","Store_Type","Product_Id_char", "Store_Age_Years","Product_Type_Category" ] REQUIRED_EXTRA = ["Store_Id", "Product_Type_Clean"] REQUIRED_COLS = FEATURE_ORDER + REQUIRED_EXTRA def _get_preprocessor(model_obj): """Find the fitted ColumnTransformer inside the pipeline (best-effort).""" try: # common: Pipeline([... ('preprocess', ColumnTransformer(...)) ...]) for name, step in getattr(model_obj, "named_steps", {}).items(): # step could itself be a pipeline; search recursively once if step.__class__.__name__ == "ColumnTransformer": return step if hasattr(step, "named_steps"): for _, inner in step.named_steps.items(): if inner.__class__.__name__ == "ColumnTransformer": return inner except Exception: pass return None def _columns_by_role(preproc): """Infer which columns the model treats as numeric vs categorical based on the transformers it fitted (looks for typical numeric/cat pipeline components).""" numeric, categorical, other = set(), set(), set() if preproc is None: return numeric, categorical, other def _has_num_clues(obj): # mean/median imputer, scaler are strong signals for numeric try: import sklearn from sklearn.impute import SimpleImputer from sklearn.preprocessing import StandardScaler, MinMaxScaler if isinstance(obj, SimpleImputer) and obj.strategy in ("mean","median"): return True if isinstance(obj, (StandardScaler, MinMaxScaler)): return True except Exception: pass # name heuristic as fallback name = getattr(obj, "__class__", type("X",(object,),{})).__name__.lower() return any(tok in name for tok in ["standardscaler","minmax","powertransformer"]) def _flatten_cols(cols): # expected to be list-like of column names try: return list(cols) except Exception: return [] for name, transformer, cols in getattr(preproc, "transformers_", []): col_list = _flatten_cols(cols) # unwrap Pipeline if present inner = transformer try: if hasattr(transformer, "steps"): inner = [s for _, s in transformer.steps] except Exception: pass # Decide role is_num = False try: if isinstance(inner, list): is_num = any(_has_num_clues(s) for s in inner) else: is_num = _has_num_clues(inner) except Exception: is_num = False if is_num: numeric.update(col_list) else: # Heuristic: if not numeric but clearly encoding-like, mark categorical # (OneHotEncoder, OrdinalEncoder, etc.) or anything else default to cat. categorical.update(col_list) # Anything not covered but expected by model: remaining = set(REQUIRED_COLS) - numeric - categorical other.update(remaining) return numeric, categorical, other @superkart_api.get("/") def home(): return "SuperKart Sales Prediction API is live." @superkart_api.post("/v1/predict") def predict_sales(): try: data = request.get_json(force=True) or {} df = pd.DataFrame([data]) # Ensure required columns exist (backfill minimal) if "Store_Id" not in df.columns: df["Store_Id"] = 1 if "Product_Type_Clean" not in df.columns: df["Product_Type_Clean"] = df.get("Product_Type_Category", "") # Light normalization if "Product_Id_char" in df.columns: df["Product_Id_char"] = df["Product_Id_char"].astype(str).str.strip().str.upper().str[:2] # Discover the model's own numeric vs categorical assignment preproc = _get_preprocessor(model) num_cols, cat_cols, other_cols = _columns_by_role(preproc) # Fallback if discovery failed: treat typical schema if not num_cols and not cat_cols: num_cols = {"Product_Weight","Product_Allocated_Area","Product_MRP","Store_Age_Years","Store_Id"} cat_cols = set(REQUIRED_COLS) - num_cols # Cast by the model's expectation for c in num_cols: if c in df.columns: df[c] = pd.to_numeric(df[c], errors="coerce") else: df[c] = np.nan for c in cat_cols: if c in df.columns: df[c] = df[c].astype(str) else: df[c] = "" # For any remaining expected columns, be conservative: for c in other_cols: # default to string so encoders won't crash; if numeric transformer grabs them, they will be NaN if c in df.columns: df[c] = df[c].astype(str) else: df[c] = "" # Reindex to the union the model expects (order is harmless) expected_cols = list(dict.fromkeys(list(num_cols) + list(cat_cols) + list(other_cols))) # ensure all REQUIRED_COLS are present too for c in REQUIRED_COLS: if c not in expected_cols: expected_cols.append(c) df = df.reindex(columns=expected_cols) pred = float(model.predict(df)[0]) return jsonify({"result": pred}) except Exception as e: # Return compact debug to finish alignment if needed try: dtypes = {col: str(dt) for col, dt in df.dtypes.items()} sample = df.to_dict(orient="records")[0] roles = { "numeric": list(_columns_by_role(_get_preprocessor(model))[0]), "categorical": list(_columns_by_role(_get_preprocessor(model))[1]), "other": list(_columns_by_role(_get_preprocessor(model))[2]), } except Exception: dtypes, sample, roles = None, None, None return jsonify({ "error": str(e), "dtypes": dtypes, "row": sample, "roles": roles }), 500 if __name__ == "__main__": superkart_api.run(host="0.0.0.0", port=7860, debug=True)