| from flask import Flask, request, jsonify, Response |
| import os |
| import joblib |
| import pandas as pd |
| from typing import Any, Optional |
|
|
| app = Flask(__name__) |
|
|
| |
| MODEL_PATH = os.getenv("MODEL_PATH", "best_model_random_forest.joblib") |
|
|
| _model: Optional[Any] = None |
| _pipe: Optional[Any] = None |
| _model_error: Optional[str] = None |
|
|
|
|
| def load_model_if_needed(): |
| """Load the model lazily so the app can boot even if the model is missing.""" |
| global _model, _pipe, _model_error |
| if _pipe is not None or _model_error is not None: |
| return |
| try: |
| if not os.path.exists(MODEL_PATH): |
| _model_error = f"Model file not found at '{MODEL_PATH}'. Upload it or set MODEL_PATH." |
| return |
| _model = joblib.load(MODEL_PATH) |
| _pipe = _model["pipeline"] if isinstance(_model, dict) and "pipeline" in _model else _model |
| except Exception as e: |
| _model_error = f"Failed to load model from '{MODEL_PATH}': {e}" |
|
|
|
|
| |
|
|
| @app.route("/", methods=["GET"]) |
| def root_html(): |
| |
| return Response( |
| "<!doctype html><html><head><meta charset='utf-8'><title>Backend</title></head>" |
| "<body><h1>Backend running β
</h1><p>See <code>/health</code> and <code>/predict</code>.</p></body></html>", |
| mimetype="text/html", |
| status=200, |
| ) |
|
|
| @app.route("/__ping__", methods=["GET"]) |
| def ping_plain(): |
| return Response("ok", mimetype="text/plain", status=200) |
|
|
| @app.route("/health", methods=["GET"]) |
| def health(): |
| load_model_if_needed() |
| status = "ok" if _pipe is not None and _model_error is None else "degraded" |
| return jsonify({"status": status, "model_path": MODEL_PATH, "model_error": _model_error}) |
|
|
|
|
| |
|
|
| @app.route("/predict", methods=["POST"]) |
| def predict(): |
| load_model_if_needed() |
| if _pipe is None: |
| return jsonify({"error": f"Model not available. Details: {_model_error}"}), 500 |
|
|
| data = request.get_json(force=True) |
|
|
| |
| if isinstance(data, dict) and "records" in data: |
| df = pd.DataFrame(data["records"]) |
| elif isinstance(data, list): |
| df = pd.DataFrame(data) |
| elif isinstance(data, dict): |
| df = pd.DataFrame([data]) |
| else: |
| return jsonify({"error": "Unsupported payload format"}), 400 |
|
|
| try: |
| preds = _pipe.predict(df) |
| predictions = [float(p) for p in preds] |
| return jsonify({"predictions": predictions}) |
| except Exception as e: |
| return jsonify({"error": str(e)}), 400 |
|
|
|
|
| |
| if __name__ == "__main__": |
| |
| port = int(os.getenv("PORT", 7860)) |
| print(f"β
Starting Flask on port {port} for Hugging Face Spaces") |
| app.run(host="0.0.0.0", port=port, debug=False) |
|
|