Added accounting and more example laptops

app.py

@@ -5,16 +5,23 @@ import tempfile
 import uuid
 import zipfile
 from datetime import datetime, timezone
+from functools import wraps
 from pathlib import Path
 
-from flask import Flask, jsonify, render_template, request, send_file, send_from_directory
+from flask import Flask, jsonify, redirect, render_template, request, send_file, send_from_directory, session, url_for
+from werkzeug.security import check_password_hash, generate_password_hash
 from werkzeug.utils import secure_filename
 
 app = Flask(__name__)
+app.secret_key = os.environ.get("SECRET_KEY", "change-this-secret-key")
 DATABASE_PATH = os.environ.get("DATABASE_PATH", str(Path(__file__).with_name("inventory.db")))
 UPLOAD_FOLDER = Path(os.environ.get("UPLOAD_FOLDER", str(Path(DATABASE_PATH).parent / "uploads")))
 ALLOWED_IMAGE_EXTENSIONS = {"jpg", "jpeg", "png", "gif", "webp"}
 app.config["MAX_CONTENT_LENGTH"] = 250 * 1024 * 1024
+ROLE_READ_ONLY = "read_only"
+ROLE_READ_WRITE = "read_write"
+ROLE_ADMIN = "admin"
+VALID_ROLES = {ROLE_READ_ONLY, ROLE_READ_WRITE, ROLE_ADMIN}
 
 
 def get_db():
@@ -64,6 +71,14 @@ def initialize_db():
                 created_at TEXT NOT NULL
             );
             CREATE INDEX IF NOT EXISTS idx_inventory_images_item ON inventory_images(item_id);
+            CREATE TABLE IF NOT EXISTS users (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                username TEXT NOT NULL UNIQUE COLLATE NOCASE,
+                password_hash TEXT NOT NULL,
+                role TEXT NOT NULL DEFAULT 'read_only',
+                can_data_tools INTEGER NOT NULL DEFAULT 0,
+                created_at TEXT NOT NULL
+            );
             """
         )
         columns = {row[1] for row in connection.execute("PRAGMA table_info(inventory_items)")}
@@ -86,6 +101,119 @@ def initialize_db():
             WHERE image_filename IS NOT NULL
             AND NOT EXISTS (SELECT 1 FROM inventory_images WHERE inventory_images.item_id = inventory_items.id)"""
         )
+        user_count = connection.execute("SELECT COUNT(*) FROM users").fetchone()[0]
+        if user_count == 0:
+            now = datetime.now(timezone.utc).isoformat()
+            connection.execute(
+                "INSERT INTO users (username, password_hash, role, can_data_tools, created_at) VALUES (?, ?, ?, ?, ?)",
+                (os.environ.get("ADMIN_USERNAME", "admin"), generate_password_hash(os.environ.get("ADMIN_PASSWORD", "admin")), ROLE_ADMIN, 1, now),
+            )
+            connection.execute(
+                "INSERT INTO users (username, password_hash, role, can_data_tools, created_at) VALUES (?, ?, ?, ?, ?)",
+                ("Amin", generate_password_hash(os.environ.get("AMIN_PASSWORD", "amin")), ROLE_READ_ONLY, 1, now),
+            )
+
+
+def current_user():
+    user_id = session.get("user_id")
+    if not user_id:
+        return None
+    with get_db() as connection:
+        return connection.execute("SELECT id, username, role, can_data_tools FROM users WHERE id = ?", (user_id,)).fetchone()
+
+
+def json_auth_error(message, status):
+    return jsonify({"error": message}), status
+
+
+def login_required(view):
+    @wraps(view)
+    def wrapped(*args, **kwargs):
+        if not current_user():
+            if request.path.startswith("/api/"):
+                return json_auth_error("Login required.", 401)
+            return redirect(url_for("login"))
+        return view(*args, **kwargs)
+    return wrapped
+
+
+def permission_required(permission):
+    def decorator(view):
+        @wraps(view)
+        def wrapped(*args, **kwargs):
+            user = current_user()
+            if not user:
+                return json_auth_error("Login required.", 401)
+            allowed = {
+                "write": user["role"] in {ROLE_READ_WRITE, ROLE_ADMIN},
+                "delete": user["role"] == ROLE_ADMIN,
+                "admin": user["role"] == ROLE_ADMIN,
+                "data_tools": user["role"] == ROLE_ADMIN or bool(user["can_data_tools"]),
+            }.get(permission, False)
+            if not allowed:
+                return json_auth_error("You do not have permission for this action.", 403)
+            return view(*args, **kwargs)
+        return wrapped
+    return decorator
+
+
+@app.get("/login")
+def login():
+    if current_user():
+        return redirect(url_for("index"))
+    return render_template("login.html")
+
+
+@app.post("/login")
+def login_submit():
+    username = clean(request.form.get("username"))
+    password = request.form.get("password", "")
+    with get_db() as connection:
+        user = connection.execute("SELECT * FROM users WHERE username = ? COLLATE NOCASE", (username,)).fetchone()
+    if not user or not check_password_hash(user["password_hash"], password):
+        return render_template("login.html", error="Invalid username or password."), 401
+    session.clear()
+    session["user_id"] = user["id"]
+    return redirect(url_for("index"))
+
+
+@app.post("/logout")
+def logout():
+    session.clear()
+    return redirect(url_for("login"))
+
+
+@app.get("/api/me")
+@login_required
+def me():
+    user = current_user()
+    return jsonify({"id": user["id"], "username": user["username"], "role": user["role"], "can_data_tools": bool(user["can_data_tools"])})
+
+
+@app.get("/api/users")
+@permission_required("admin")
+def list_users():
+    with get_db() as connection:
+        users = connection.execute("SELECT id, username, role, can_data_tools, created_at FROM users ORDER BY username COLLATE NOCASE").fetchall()
+    return jsonify([{**dict(user), "can_data_tools": bool(user["can_data_tools"])} for user in users])
+
+
+@app.post("/api/users")
+@permission_required("admin")
+def create_user():
+    data = request.get_json(silent=True) or {}
+    username = clean(data.get("username"))
+    password = data.get("password", "")
+    role = clean(data.get("role")) or ROLE_READ_ONLY
+    can_data_tools = bool(data.get("can_data_tools"))
+    if not username or len(password) < 6 or role not in VALID_ROLES:
+        return jsonify({"error": "Username, a password of at least 6 characters, and a valid level are required."}), 400
+    try:
+        with get_db() as connection:
+            cursor = connection.execute("INSERT INTO users (username, password_hash, role, can_data_tools, created_at) VALUES (?, ?, ?, ?, ?)", (username, generate_password_hash(password), role, int(can_data_tools), datetime.now(timezone.utc).isoformat()))
+        return jsonify({"id": cursor.lastrowid, "username": username, "role": role, "can_data_tools": can_data_tools}), 201
+    except sqlite3.IntegrityError:
+        return jsonify({"error": "That username already exists."}), 409
 
 
 def clean(value):
@@ -209,16 +337,26 @@ def restore_backup(backup):
 
 
 @app.route("/")
+@login_required
 def index():
-    return render_template("index.html")
+    user = current_user()
+    return render_template("index.html", user=dict(user))
+
+
+@app.get("/users")
+@permission_required("admin")
+def users_page():
+    return render_template("users.html", user=dict(current_user()))
 
 
 @app.get("/media/<path:filename>")
+@login_required
 def media(filename):
     return send_from_directory(UPLOAD_FOLDER, filename)
 
 
 @app.get("/api/items")
+@login_required
 def list_items():
     query = clean(request.args.get("q"))
     item_type = clean(request.args.get("type"))
@@ -241,6 +379,7 @@ def list_items():
 
 
 @app.get("/api/summary")
+@login_required
 def summary():
     with get_db() as connection:
         total = connection.execute("SELECT COUNT(*) FROM inventory_items").fetchone()[0]
@@ -251,6 +390,7 @@ def summary():
 
 
 @app.get("/api/backup")
+@permission_required("data_tools")
 def backup():
     with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as temporary_database:
         temporary_database_path = Path(temporary_database.name)
@@ -271,6 +411,7 @@ def backup():
 
 
 @app.post("/api/restore")
+@permission_required("data_tools")
 def restore():
     backup_file = request.files.get("backup")
     if not backup_file or not backup_file.filename:
@@ -283,6 +424,7 @@ def restore():
 
 
 @app.post("/api/items")
+@permission_required("write")
 def create_item():
     data = item_payload(request_data())
     if not data["name"] or not data["item_type"]:
@@ -315,6 +457,7 @@ def create_item():
 
 
 @app.put("/api/items/<int:item_id>")
+@permission_required("write")
 def update_item(item_id):
     data = item_payload(request_data())
     if not data["name"] or not data["item_type"]:
@@ -349,6 +492,7 @@ def update_item(item_id):
 
 
 @app.delete("/api/items/<int:item_id>")
+@permission_required("delete")
 def delete_item(item_id):
     with get_db() as connection:
         images = connection.execute("SELECT filename FROM inventory_images WHERE item_id = ?", (item_id,)).fetchall()
@@ -361,6 +505,7 @@ def delete_item(item_id):
 
 
 @app.post("/api/items/<int:item_id>/images")
+@permission_required("write")
 def add_images(item_id):
     files = files_from_request()
     with get_db() as connection:
@@ -383,6 +528,7 @@ def add_images(item_id):
 
 
 @app.delete("/api/items/<int:item_id>/images/<int:image_id>")
+@permission_required("write")
 def remove_image(item_id, image_id):
     with get_db() as connection:
         image = connection.execute("SELECT filename FROM inventory_images WHERE id = ? AND item_id = ?", (image_id, item_id)).fetchone()