app.py - master - lamar/intro-practice

app.py
356 строк11 KB
Ivan
Обновлены сервисы: rhythm, fridge, freshness
06 июл 2026, 17:03
06 июл 2026, 17:03adf0478
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"),
        products=None,
        prompt_text=None,
        recommendation=None,
        error=startup_error,
    )


MAX_USER_ID = 206209


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

    user_id = int(user_id)

    if user_id > MAX_USER_ID:
        return render_template(
            "index.html",
            user_id=None,
            products=None,
            prompt_text=None,
            recommendation=None,
            error=f"user_id должен быть меньше {MAX_USER_ID}",
        )

    session["user_id"] = user_id
    return render_template(
        "index.html",
        user_id=session.get("user_id"),
        products=None,
        prompt_text=None,
        recommendation=None,
        error=None,
    )


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


@app.route("/history", methods=["POST"])
def history():
    user_id = session.get("user_id")
    if not user_id:
        return render_template(
            "index.html",
            user_id=None,
            products=None,
            prompt_text=None,
            recommendation=None,
            error="Сначала авторизуйтесь по user_id",
        )
    try:
        rules = load_business_rules()
        rows = rules.get_user_history(user_id)
        if not rows:
            return render_template(
                "index.html",
                user_id=user_id,
                products=None,
                prompt_text=None,
                recommendation=None,
                error="Loginom вернул пустой список товаров",
            )
        products_text = rules.build_products_text(rows)
        prompt_text, recommendation = rules.ask_gigachat(user_id, products_text)
        return render_template(
            "index.html",
            user_id=user_id,
            products=rows,
            prompt_text=prompt_text,
            recommendation=recommendation,
            error=None,
        )
    except Exception as e:
        return render_template(
            "index.html",
            user_id=user_id,
            products=None,
            prompt_text=None,
            recommendation=None,
            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 = rules.ask_gigachat(user_id, products_text)
        return jsonify({
            "user_id": user_id,
            "products": rows,
            "prompt_text": prompt_text,
            "recommendation": recommendation,
        })
    except Exception as e:
        return jsonify({"error": str(e)}), 500

@app.route("/rhythm", methods=["POST"])
def rhythm():
    user_id = session.get("user_id")
    if not user_id:
        return render_template(
            "index.html",
            user_id=None,
            products=None,
            prompt_text=None,
            recommendation=None,
            error="Сначала авторизуйтесь по user_id",
        )
    try:
        rules = load_business_rules()
        if not rules.check_user_exists(user_id):
            return render_template(
                "index.html",
                user_id=user_id,
                products=None,
                prompt_text=None,
                recommendation=None,
                error=f"Пользователь с ID {user_id} не найден",
            )
        rhythm_data = rules.get_rhythm(user_id)
        if not rhythm_data:
            return render_template(
                "index.html",
                user_id=user_id,
                products=None,
                prompt_text=None,
                recommendation=None,
                error="Недостаточно данных для анализа ритма",
            )
        recommendation = rules.ask_rhythm(
            user_id,
            rhythm_data['avg_days'],
            rhythm_data['min_days'],
            rhythm_data['max_days'],
            rhythm_data['total_orders']
        )
        return render_template(
            "rhythm.html",
            user_id=user_id,
            rhythm_data=rhythm_data,
            recommendation=recommendation,
            error=None,
        )
    except Exception as e:
        return render_template(
            "index.html",
            user_id=user_id,
            products=None,
            prompt_text=None,
            recommendation=None,
            error=f"Ошибка: {e}",
        )


@app.route("/fridge", methods=["POST"])
def fridge():
    user_id = session.get("user_id")
    if not user_id:
        return render_template(
            "index.html",
            user_id=None,
            products=None,
            prompt_text=None,
            recommendation=None,
            error="Сначала авторизуйтесь по user_id",
        )
    try:
        rules = load_business_rules()
        if not rules.check_user_exists(user_id):
            return render_template(
                "index.html",
                user_id=user_id,
                products=None,
                prompt_text=None,
                recommendation=None,
                error=f"Пользователь с ID {user_id} не найден",
            )
        products = rules.get_smart_fridge(user_id)
        recommendation = rules.ask_fridge_recommendation(user_id, products)
        return render_template(
            "fridge.html",
            user_id=user_id,
            products=products,
            recommendation=recommendation,
            error=None,
        )
    except Exception as e:
        return render_template(
            "index.html",
            user_id=user_id,
            products=None,
            prompt_text=None,
            recommendation=None,
            error=f"Ошибка: {e}",
        )


@app.route("/freshness", methods=["POST"])
def freshness():
    user_id = session.get("user_id")
    if not user_id:
        return render_template(
            "index.html",
            user_id=None,
            products=None,
            prompt_text=None,
            recommendation=None,
            error="Сначала авторизуйтесь по user_id",
        )
    try:
        rules = load_business_rules()
        if not rules.check_user_exists(user_id):
            return render_template(
                "index.html",
                user_id=user_id,
                products=None,
                prompt_text=None,
                recommendation=None,
                error=f"Пользователь с ID {user_id} не найден",
            )
        freshness_data = rules.get_freshness_alert(user_id)
        recommendation = rules.ask_freshness_recommendation(user_id, freshness_data)
        expired_products = []
        for row in freshness_data:
            if row.get('orders_since_last_product', 0) > 1.2:
                expired_products.append(row)
        return render_template(
            "freshness.html",
            user_id=user_id,
            products=expired_products,
            all_data=freshness_data,
            recommendation=recommendation,
            error=None,
        )
    except Exception as e:
        return render_template(
            "index.html",
            user_id=user_id,
            products=None,
            prompt_text=None,
            recommendation=None,
            error=f"Ошибка: {e}",
        )


@app.route("/api/rhythm", methods=["POST"])
def api_rhythm():
    user_id = session.get("user_id")
    if not user_id:
        return jsonify({"error": "Сначала авторизуйтесь"}), 403
    try:
        rules = load_business_rules()
        rhythm_data = rules.get_rhythm(user_id)
        recommendation = rules.ask_rhythm(
            user_id,
            rhythm_data['avg_days'],
            rhythm_data['min_days'],
            rhythm_data['max_days'],
            rhythm_data['total_orders']
        )
        return jsonify({
            "user_id": user_id,
            "rhythm": rhythm_data,
            "recommendation": recommendation,
        })
    except Exception as e:
        return jsonify({"error": str(e)}), 500


@app.route("/api/fridge", methods=["POST"])
def api_fridge():
    user_id = session.get("user_id")
    if not user_id:
        return jsonify({"error": "Сначала авторизуйтесь"}), 403
    try:
        rules = load_business_rules()
        products = rules.get_smart_fridge(user_id)
        recommendation = rules.ask_fridge_recommendation(user_id, products)
        return jsonify({
            "user_id": user_id,
            "products": products,
            "recommendation": recommendation,
        })
    except Exception as e:
        return jsonify({"error": str(e)}), 500


@app.route("/api/freshness", methods=["POST"])
def api_freshness():
    user_id = session.get("user_id")
    if not user_id:
        return jsonify({"error": "Сначала авторизуйтесь"}), 403
    try:
        rules = load_business_rules()
        freshness_data = rules.get_freshness_alert(user_id)
        recommendation = rules.ask_freshness_recommendation(user_id, freshness_data)
        return jsonify({
            "user_id": user_id,
            "data": freshness_data,
            "recommendation": recommendation,
        })
    except Exception as e:
        return jsonify({"error": str(e)}), 500


if __name__ == '__main__':
    import os
    port = int(os.environ.get('PORT', 8000))
    app.run(host='0.0.0.0', port=port)