Merge pull request #3 from mshqq/fix/p0-changes

fix: debug/host/port в .env, авторизация админки, порядок проверок и unsubscribe/order-status

Author: mshqq

.envexample

@@ -1,4 +1,8 @@
 SECRET_KEY=
 app_password=
 MAX_TOKEN=
-USER_ID=
+USER_ID=
+
+FLASK_DEBUG=1
+FLASK_HOST=0.0.0.0
+FLASK_PORT=5000

app.py

@@ -1,6 +1,22 @@
 from app import create_app
+from dotenv import load_dotenv
+from os import getenv
+from sys import exit
+
+load_dotenv()
 
 app = create_app()
 
 if __name__ == "__main__":
-    app.run(debug=True, host="0.0.0.0", port=5000)
+    debug = getenv("FLASK_DEBUG", "0") == "1"
+    host = getenv("FLASK_HOST", "127.0.0.1")
+    port = int(getenv("FLASK_PORT", 5000))
+
+    if not (1 <= port <= 65535):
+        exit(f"FLASK_PORT={port} вне диапазона 1-65535")
+
+    app.run(
+        debug=debug,
+        host=host,
+        port=port,
+    )

app/__init__.py

@@ -1,28 +1,36 @@
 from flask import Flask
 from dotenv import load_dotenv
-from app.extensions import db, csrf
+from app.extensions import db, csrf, login_manager
 from app.routes.locations import locations_bp
 from app.routes.main import main_bp
 from app.routes.orders import orders_bp
 from app.routes.products import products_bp
 from app.routes.subscribers import subscribers_bp
-import os
+from os import getenv
+from app.routes.auth import auth_bp
 
 
 def get_password():
-    return os.getenv("app_password")
+    return getenv("app_password")
 
 
 def create_app():
     app = Flask(__name__)
     load_dotenv()
-    app.config["SECRET_KEY"] = os.getenv("SECRET_KEY")
+    app.config["SECRET_KEY"] = getenv("SECRET_KEY")
     app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///shaurmania.db"
     app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
     app.json.sort_keys = False
 
     db.init_app(app)
     csrf.init_app(app)
+    login_manager.init_app(app)
+
+    from app.models import User
+
+    @login_manager.user_loader
+    def load_user(user_id):
+        return User.query.get(int(user_id))
 
     with app.app_context():
         from app import models  # noqa
@@ -36,6 +44,7 @@ def create_app():
     app.register_blueprint(orders_bp)
     app.register_blueprint(products_bp)
     app.register_blueprint(subscribers_bp)
+    app.register_blueprint(auth_bp)
 
     from app.routes.orders import scheduler
 

app/extensions.py

@@ -1,5 +1,8 @@
 from flask_sqlalchemy import SQLAlchemy
 from flask_wtf.csrf import CSRFProtect
+from flask_login import LoginManager
 
 csrf = CSRFProtect()
 db = SQLAlchemy()
+login_manager = LoginManager()
+login_manager.login_view = "auth.login"

app/models.py

@@ -1,6 +1,8 @@
 from app.extensions import db
 from datetime import datetime
 from uuid import uuid4
+from werkzeug.security import generate_password_hash, check_password_hash
+from flask_login import UserMixin
 
 
 class Location(db.Model):
@@ -94,3 +96,21 @@ class Subscriber(db.Model):
 
     def __repr__(self):
         return f"<Subscribers {self.email}>"
+
+
+class User(db.Model, UserMixin):
+    __tablename__ = "users"
+    id = db.Column(db.Integer(), primary_key=True)
+    username = db.Column(db.String(100), nullable=False, unique=True)
+    email = db.Column(db.String(100), nullable=False, unique=True)
+    password_hash = db.Column(db.String(255), nullable=False)
+    created_on = db.Column(db.DateTime(), default=datetime.utcnow)
+
+    def __repr__(self):
+        return f"<{self.id}:{self.username}>"
+
+    def set_password(self, password):
+        self.password_hash = generate_password_hash(password)
+
+    def check_password(self, password):
+        return check_password_hash(self.password_hash, password)

app/routes/auth.py

@@ -0,0 +1,33 @@
+from flask_login import login_user, logout_user, login_required
+from flask import Blueprint, jsonify, request
+from app.models import User
+
+auth_bp = Blueprint("auth", __name__)
+
+
+@auth_bp.route("/login", methods=["POST"])
+def login():
+    data = request.get_json()
+    if not data:
+        return jsonify({"error": "Запрос не содержит JSON"}), 400
+
+    username = data.get("username")
+    password = data.get("password")
+
+    if not username or not password:
+        return jsonify({"error": "Не указан логин или пароль"}), 400
+
+    user = User.query.filter_by(username=username).first()
+
+    if not user or not user.check_password(password):
+        return jsonify({"error": "Неверный логин или пароль"}), 401
+
+    login_user(user)
+    return jsonify({"message": "Успешный вход"}), 200
+
+
+@auth_bp.route("/logout", methods=["POST"])
+@login_required
+def logout():
+    logout_user()
+    return jsonify({"message": "Вы вышли"}), 200

app/routes/orders.py

@@ -30,6 +30,8 @@ def setStatus(app, id, status):
 @orders_bp.route("/api/order/status/<string:token>", methods=["GET"])
 def get_order_status(token):
     order = Order.query.filter(Order.public_token == token).first()
+    if not order:
+        return jsonify({"error": "Не найден заказ"}), 404
     status = order.status
 
     return jsonify({"status": status}), 200

app/routes/subscribers.py

@@ -1,3 +1,4 @@
+from flask_login import login_required
 from flask import Blueprint, jsonify, request
 from app.models import Subscriber
 from app.extensions import db, csrf
@@ -16,12 +17,12 @@
 def unsubscribe(token):
     sub = Subscriber.query.filter(Subscriber.unsubscribe_token == token).first()
 
-    if sub.is_active is False:
-        return jsonify({"error": "Вы уже отписались от рассылки!"}), 400
-
     if not sub:
         return jsonify({"error": "Токен не найден или устарел!"}), 404
 
+    if sub.is_active is False:
+        return jsonify({"error": "Вы уже отписались от рассылки!"}), 400
+
     try:
         sub.is_active = False
         db.session.commit()
@@ -32,8 +33,8 @@ def unsubscribe(token):
         return jsonify({"error": "Произошла ошибка при отписке!"}), 500
 
 
-@csrf.exempt
 @subscribers_bp.route("/admin/subscribers")
+@login_required
 def view_subscribers():
     subs = Subscriber.query.filter(Subscriber.is_active == True).all()  # noqa
     sub_list = [

scripts/create_admin.py

@@ -0,0 +1,38 @@
+import sys
+import os
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from getpass import getpass
+from app import create_app
+from app.extensions import db
+from app.models import User
+from sys import exit
+
+
+def create_admin():
+    app = create_app()
+    with app.app_context():
+        username = input("Юзернейм: ").strip()
+        email = input("Email: ").strip()
+        password = getpass("Пароль: ").strip()
+
+        if not username or not email or not password:
+            exit("Все поля обязательны")
+
+        if User.query.filter_by(username=username).first():
+            exit(f"Пользователь {username} уже существует")
+
+        if User.query.filter_by(email=email).first():
+            exit(f"Email {email} уже занят")
+
+        user = User(username=username, email=email)  # pyrefly: ignore
+        user.set_password(password)
+        db.session.add(user)
+        db.session.commit()
+
+        print(f"Создан пользователь {username} (id={user.id})")
+
+
+if __name__ == "__main__":
+    create_admin()