app.py - master - muhxa/into-practice

app.py
237 строк7 KB
muhxa
Update: app.py
09 июл 2026, 03:48
09 июл 2026, 03:483cd67f8
from flask import Flask, request, session, jsonify, render_template
import importlib.util
from pathlib import Path

app = Flask(__name__)
app.secret_key = "instacart-demo-secret-key"
BASE_DIR = Path(__file__).parent


def load_business_rules():
    py_path = BASE_DIR / "business-rules.py"
    if not py_path.exists():
        raise FileNotFoundError(
            "Не найден business-rules.py."
        )

    spec = importlib.util.spec_from_file_location("business-rules", py_path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


# ============================================================
# Главная страница / авторизация
# ============================================================
@app.route("/", methods=["GET"])
def index():
    startup_error = None
    if not (BASE_DIR / "business-rules.py").exists():
        startup_error = "Файл business-rules.py не найден."

    return render_template(
        "index.html",
        user_id=session.get("user_id"),
        error=startup_error,
    )


@app.route("/login", methods=["POST"])
def login():
    user_id = request.form.get("user_id", "").strip()
    if not user_id.isdigit():
        return render_template(
            "index.html",
            user_id=None,
            error="user_id должен быть целым числом",
        )
    session["user_id"] = int(user_id)
    return render_template(
        "index.html",
        user_id=session.get("user_id"),
        error=None,
    )


@app.route("/logout", methods=["POST"])
def logout():
    session.clear()
    return render_template("index.html", user_id=None, error=None)


# ============================================================
# /history — «Мои покупки»: топ-10 товаров + рекомендация GigaChat
# ============================================================
@app.route("/history", methods=["GET"])
def history_page():
    return render_template(
        "history.html",
        user_id=session.get("user_id"),
        products=None,
        prompt_text=None,
        recommendation=None,
        from_cache=False,
        error=None,
    )


@app.route("/history", methods=["POST"])
def history():
    user_id = session.get("user_id")
    if not user_id:
        return render_template(
            "history.html",
            user_id=None,
            products=None,
            prompt_text=None,
            recommendation=None,
            from_cache=False,
            error="Сначала авторизуйтесь по user_id",
        )

    try:
        rules = load_business_rules()
        rows = rules.get_user_history(user_id)

        if not rows:
            return render_template(
                "history.html",
                user_id=user_id,
                products=None,
                prompt_text=None,
                recommendation=None,
                from_cache=False,
                error="Loginom вернул пустой список товаров",
            )

        products_text = rules.build_products_text(rows)
        prompt_text, recommendation, from_cache = rules.ask_gigachat(user_id, products_text)

        return render_template(
            "history.html",
            user_id=user_id,
            products=rows,
            prompt_text=prompt_text,
            recommendation=recommendation,
            from_cache=from_cache,
            error=None,
        )
    except Exception as e:
        return render_template(
            "history.html",
            user_id=user_id,
            products=None,
            prompt_text=None,
            recommendation=None,
            from_cache=False,
            error=f"Ошибка: {e}",
        )


@app.route("/api/history", methods=["POST"])
def api_history():
    user_id = session.get("user_id")
    if not user_id:
        return jsonify({"error": "Сначала авторизуйтесь"}), 403

    try:
        rules = load_business_rules()
        rows = rules.get_user_history(user_id)
        products_text = rules.build_products_text(rows)
        prompt_text, recommendation, from_cache = rules.ask_gigachat(user_id, products_text)
        return jsonify({
            "user_id": user_id,
            "products": rows,
            "prompt_text": prompt_text,
            "recommendation": recommendation,
            "from_cache": from_cache,
        })
    except Exception as e:
        return jsonify({"error": str(e)}), 500


# ============================================================
# /insights — «Аналитика»: 3 новых умных сервиса
# ============================================================
@app.route("/insights", methods=["GET"])
def insights_page():
    return render_template(
        "insights.html",
        user_id=session.get("user_id"),
        data=None,
        error=None,
    )

@app.route("/insights", methods=["POST"])
def insights_run():
    user_id = session.get("user_id")
    if not user_id:
        return render_template(
            "insights.html",
            user_id=None,
            data=None,
            error="Сначала авторизуйтесь по user_id",
        )

    try:
        rules = load_business_rules()

        print(f"DEBUG: Запуск аналитики для user_id={user_id}")
        
        rhythm = rules.get_order_rhythm(user_id)
        print(f"DEBUG: rhythm={rhythm}")
        
        if rhythm:
            _, rhythm_text, rhythm_cached = rules.ask_gigachat_rhythm(user_id, rhythm)
        else:
            rhythm_text, rhythm_cached = None, False
        print(f"DEBUG: rhythm_text получен")

        variety_rows = rules.get_basket_variety(user_id)
        print(f"DEBUG: variety_rows={len(variety_rows) if variety_rows else 0} строк")
        
        if variety_rows:
            _, variety_text, variety_cached = rules.ask_gigachat_variety(user_id, variety_rows)
        else:
            variety_text, variety_cached = None, False
        print(f"DEBUG: variety_text получен")

        loyalty_rows = rules.get_loyalty_rate(user_id)
        print(f"DEBUG: loyalty_rows={len(loyalty_rows) if loyalty_rows else 0} строк")
        
        if loyalty_rows:
            _, loyalty_text, loyalty_cached = rules.ask_gigachat_loyalty(user_id, loyalty_rows)
        else:
            loyalty_text, loyalty_cached = None, False
        print(f"DEBUG: loyalty_text получен")

        data = {
            "rhythm": rhythm,
            "rhythm_text": rhythm_text,
            "rhythm_cached": rhythm_cached,
            "variety_rows": variety_rows,
            "variety_text": variety_text,
            "variety_cached": variety_cached,
            "loyalty_rows": loyalty_rows,
            "loyalty_text": loyalty_text,
            "loyalty_cached": loyalty_cached,
        }
        return render_template("insights.html", user_id=user_id, data=data, error=None)
    except Exception as e:
        import traceback
        print(f"DEBUG ERROR: {e}")
        print(traceback.format_exc())
        return render_template(
            "insights.html",
            user_id=user_id,
            data=None,
            error=f"Ошибка: {e}",
        )



if __name__ == "__main__":
    import os

    port = int(os.environ.get("PORT", 8000))
    app.run(host="0.0.0.0", port=port)