Spaces:
Sleeping
Sleeping
| 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 | |
| def home(): | |
| return "SuperKart Sales Prediction API is live." | |
| 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) | |