647 lines
28 KiB
Python
647 lines
28 KiB
Python
import os
|
|
import json
|
|
import shutil
|
|
import sqlite3
|
|
import tempfile
|
|
import uuid
|
|
import zipfile
|
|
from datetime import datetime, timezone
|
|
from functools import wraps
|
|
from pathlib import Path
|
|
|
|
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}
|
|
CONFIGURABLE_FIELDS = {
|
|
"brand", "model", "serial_number", "asset_tag", "location", "destination",
|
|
"checkout_date", "purchase_date", "notes", "cpu", "ram", "gpu", "storage", "media_format",
|
|
}
|
|
DEFAULT_TYPES = {
|
|
"Computer": ["brand", "model", "serial_number", "asset_tag", "location", "purchase_date", "cpu", "ram", "gpu", "storage", "notes"],
|
|
"Audio": ["brand", "model", "serial_number", "asset_tag", "location", "purchase_date", "media_format", "notes"],
|
|
"Graphics": ["brand", "model", "serial_number", "asset_tag", "location", "purchase_date", "gpu", "storage", "notes"],
|
|
"Long play disk": ["brand", "model", "asset_tag", "location", "purchase_date", "media_format", "notes"],
|
|
"CD": ["brand", "model", "asset_tag", "location", "purchase_date", "media_format", "notes"],
|
|
"Display": ["brand", "model", "serial_number", "asset_tag", "location", "purchase_date", "notes"],
|
|
"Peripheral": ["brand", "model", "serial_number", "asset_tag", "location", "purchase_date", "notes"],
|
|
"Other": ["brand", "model", "serial_number", "asset_tag", "location", "purchase_date", "notes"],
|
|
}
|
|
|
|
|
|
def get_db():
|
|
connection = sqlite3.connect(DATABASE_PATH)
|
|
connection.row_factory = sqlite3.Row
|
|
connection.execute("PRAGMA foreign_keys = ON")
|
|
return connection
|
|
|
|
|
|
def initialize_db():
|
|
Path(DATABASE_PATH).parent.mkdir(parents=True, exist_ok=True)
|
|
UPLOAD_FOLDER.mkdir(parents=True, exist_ok=True)
|
|
with get_db() as connection:
|
|
connection.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS inventory_items (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL,
|
|
item_type TEXT NOT NULL,
|
|
brand TEXT,
|
|
model TEXT,
|
|
serial_number TEXT UNIQUE,
|
|
asset_tag TEXT UNIQUE,
|
|
status TEXT NOT NULL DEFAULT 'Available',
|
|
location TEXT,
|
|
destination TEXT,
|
|
checkout_date TEXT,
|
|
purchase_date TEXT,
|
|
notes TEXT,
|
|
cpu TEXT,
|
|
ram TEXT,
|
|
gpu TEXT,
|
|
storage TEXT,
|
|
media_format TEXT,
|
|
image_filename TEXT,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_inventory_search ON inventory_items(name, brand, model, serial_number, asset_tag);
|
|
CREATE INDEX IF NOT EXISTS idx_inventory_type ON inventory_items(item_type);
|
|
CREATE INDEX IF NOT EXISTS idx_inventory_status ON inventory_items(status);
|
|
CREATE TABLE IF NOT EXISTS inventory_images (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
item_id INTEGER NOT NULL REFERENCES inventory_items(id) ON DELETE CASCADE,
|
|
filename TEXT NOT NULL,
|
|
original_name TEXT,
|
|
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
|
|
);
|
|
CREATE TABLE IF NOT EXISTS inventory_types (
|
|
name TEXT PRIMARY KEY COLLATE NOCASE,
|
|
fields TEXT NOT NULL
|
|
);
|
|
"""
|
|
)
|
|
columns = {row[1] for row in connection.execute("PRAGMA table_info(inventory_items)")}
|
|
migrations = {
|
|
"image_filename": "ALTER TABLE inventory_items ADD COLUMN image_filename TEXT",
|
|
"destination": "ALTER TABLE inventory_items ADD COLUMN destination TEXT",
|
|
"checkout_date": "ALTER TABLE inventory_items ADD COLUMN checkout_date TEXT",
|
|
}
|
|
for column, statement in migrations.items():
|
|
if column not in columns:
|
|
try:
|
|
connection.execute(statement)
|
|
except sqlite3.OperationalError as error:
|
|
if "duplicate column name" not in str(error):
|
|
raise
|
|
for name, fields in DEFAULT_TYPES.items():
|
|
connection.execute(
|
|
"INSERT OR IGNORE INTO inventory_types (name, fields) VALUES (?, ?)",
|
|
(name, json.dumps(fields)),
|
|
)
|
|
connection.execute(
|
|
"""INSERT INTO inventory_images (item_id, filename, original_name, created_at)
|
|
SELECT id, image_filename, image_filename, COALESCE(updated_at, datetime('now'))
|
|
FROM inventory_items
|
|
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):
|
|
return (value or "").strip()
|
|
|
|
|
|
def item_payload(data):
|
|
payload = {
|
|
"name": clean(data.get("name")),
|
|
"item_type": clean(data.get("item_type")),
|
|
"brand": clean(data.get("brand")),
|
|
"model": clean(data.get("model")),
|
|
"serial_number": clean(data.get("serial_number")) or None,
|
|
"asset_tag": clean(data.get("asset_tag")) or None,
|
|
"status": clean(data.get("status")) or "Available",
|
|
"location": clean(data.get("location")),
|
|
"destination": clean(data.get("destination")),
|
|
"checkout_date": clean(data.get("checkout_date")) or None,
|
|
"purchase_date": clean(data.get("purchase_date")) or None,
|
|
"notes": clean(data.get("notes")),
|
|
"cpu": clean(data.get("cpu")),
|
|
"ram": clean(data.get("ram")),
|
|
"gpu": clean(data.get("gpu")),
|
|
"storage": clean(data.get("storage")),
|
|
"media_format": clean(data.get("media_format")),
|
|
}
|
|
return payload
|
|
|
|
|
|
def type_payload(data):
|
|
name = clean(data.get("name"))
|
|
fields = data.get("fields", [])
|
|
if not isinstance(fields, list):
|
|
raise ValueError("Choose the fields to show for this inventory type.")
|
|
fields = [field for field in fields if field in CONFIGURABLE_FIELDS]
|
|
if not name or len(name) > 80:
|
|
raise ValueError("A type name of up to 80 characters is required.")
|
|
return name, fields
|
|
|
|
|
|
def serialize_type(row):
|
|
try:
|
|
fields = json.loads(row["fields"])
|
|
except json.JSONDecodeError:
|
|
fields = []
|
|
return {"name": row["name"], "fields": [field for field in fields if field in CONFIGURABLE_FIELDS]}
|
|
|
|
|
|
def serialize(row, connection):
|
|
item = dict(row)
|
|
item["images"] = [
|
|
{"id": image["id"], "filename": image["filename"], "original_name": image["original_name"], "url": f"/media/{image['filename']}"}
|
|
for image in connection.execute("SELECT id, filename, original_name FROM inventory_images WHERE item_id = ? ORDER BY id", (row["id"],)).fetchall()
|
|
]
|
|
item["image_url"] = item["images"][0]["url"] if item["images"] else None
|
|
return item
|
|
|
|
|
|
def request_data():
|
|
return request.form if request.form else (request.get_json(silent=True) or {})
|
|
|
|
|
|
def save_image(image):
|
|
if not image or not image.filename:
|
|
return None
|
|
original_name = secure_filename(image.filename)
|
|
extension = Path(original_name).suffix.lower().lstrip(".")
|
|
if extension not in ALLOWED_IMAGE_EXTENSIONS:
|
|
raise ValueError("Images must be JPG, PNG, GIF, or WebP files.")
|
|
filename = f"{uuid.uuid4().hex}.{extension}"
|
|
image.save(UPLOAD_FOLDER / filename)
|
|
return filename, original_name
|
|
|
|
|
|
def save_images(files):
|
|
saved = []
|
|
try:
|
|
for image in files:
|
|
if image and image.filename:
|
|
saved.append(save_image(image))
|
|
return saved
|
|
except ValueError:
|
|
for filename, _ in saved:
|
|
delete_image(filename)
|
|
raise
|
|
|
|
|
|
def delete_image(filename):
|
|
if filename:
|
|
(UPLOAD_FOLDER / filename).unlink(missing_ok=True)
|
|
|
|
|
|
def files_from_request():
|
|
return request.files.getlist("images") or request.files.getlist("image")
|
|
|
|
|
|
def add_directory_to_zip(archive, directory, archive_prefix):
|
|
if not directory.exists():
|
|
return
|
|
for path in directory.rglob("*"):
|
|
if path.is_file():
|
|
archive.write(path, f"{archive_prefix}/{path.relative_to(directory).as_posix()}")
|
|
|
|
|
|
def valid_backup_member(member):
|
|
path = Path(member.filename)
|
|
return not path.is_absolute() and ".." not in path.parts and (member.filename == "inventory.db" or member.filename.startswith("uploads/"))
|
|
|
|
|
|
def restore_backup(backup):
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
temporary_path = Path(temporary_directory)
|
|
archive_path = temporary_path / "backup.zip"
|
|
backup.save(archive_path)
|
|
with zipfile.ZipFile(archive_path) as archive:
|
|
members = archive.infolist()
|
|
if not members or not all(valid_backup_member(member) for member in members):
|
|
raise ValueError("The backup file has an invalid format.")
|
|
if "inventory.db" not in archive.namelist():
|
|
raise ValueError("The backup does not contain inventory.db.")
|
|
database_path = temporary_path / "inventory.db"
|
|
uploads_path = temporary_path / "uploads"
|
|
archive.extract("inventory.db", temporary_path)
|
|
archive.extractall(temporary_path, [member for member in members if member.filename.startswith("uploads/")])
|
|
|
|
with sqlite3.connect(database_path) as connection:
|
|
if connection.execute("PRAGMA integrity_check").fetchone()[0] != "ok":
|
|
raise ValueError("The backup database failed its integrity check.")
|
|
Path(DATABASE_PATH).parent.mkdir(parents=True, exist_ok=True)
|
|
staged_database_path = Path(DATABASE_PATH).with_name(f".{Path(DATABASE_PATH).name}.restore")
|
|
shutil.copy2(database_path, staged_database_path)
|
|
os.replace(staged_database_path, DATABASE_PATH)
|
|
if UPLOAD_FOLDER.exists():
|
|
shutil.rmtree(UPLOAD_FOLDER)
|
|
if uploads_path.exists():
|
|
staged_uploads_path = UPLOAD_FOLDER.parent / f".{UPLOAD_FOLDER.name}.restore"
|
|
shutil.copytree(uploads_path, staged_uploads_path, dirs_exist_ok=True)
|
|
os.replace(staged_uploads_path, UPLOAD_FOLDER)
|
|
else:
|
|
UPLOAD_FOLDER.mkdir(parents=True, exist_ok=True)
|
|
initialize_db()
|
|
|
|
|
|
@app.route("/")
|
|
@login_required
|
|
def index():
|
|
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"))
|
|
status = clean(request.args.get("status"))
|
|
clauses = []
|
|
params = []
|
|
if query:
|
|
clauses.append("(name LIKE ? OR item_type LIKE ? OR brand LIKE ? OR model LIKE ? OR serial_number LIKE ? OR asset_tag LIKE ? OR location LIKE ? OR destination LIKE ?)")
|
|
params.extend([f"%{query}%"] * 8)
|
|
if item_type:
|
|
clauses.append("item_type = ?")
|
|
params.append(item_type)
|
|
if status:
|
|
clauses.append("status = ?")
|
|
params.append(status)
|
|
where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
|
|
with get_db() as connection:
|
|
rows = connection.execute(f"SELECT * FROM inventory_items {where} ORDER BY updated_at DESC, id DESC", params).fetchall()
|
|
return jsonify([serialize(row, connection) for row in rows])
|
|
|
|
|
|
@app.get("/api/types")
|
|
@login_required
|
|
def list_types():
|
|
with get_db() as connection:
|
|
rows = connection.execute("SELECT name, fields FROM inventory_types ORDER BY name COLLATE NOCASE").fetchall()
|
|
return jsonify([serialize_type(row) for row in rows])
|
|
|
|
|
|
@app.post("/api/types")
|
|
@permission_required("admin")
|
|
def create_type():
|
|
try:
|
|
name, fields = type_payload(request.get_json(silent=True) or {})
|
|
except ValueError as error:
|
|
return jsonify({"error": str(error)}), 400
|
|
try:
|
|
with get_db() as connection:
|
|
connection.execute("INSERT INTO inventory_types (name, fields) VALUES (?, ?)", (name, json.dumps(fields)))
|
|
return jsonify({"name": name, "fields": fields}), 201
|
|
except sqlite3.IntegrityError:
|
|
return jsonify({"error": "That inventory type already exists."}), 409
|
|
|
|
|
|
@app.put("/api/types/<path:type_name>")
|
|
@permission_required("admin")
|
|
def update_type(type_name):
|
|
try:
|
|
name, fields = type_payload(request.get_json(silent=True) or {})
|
|
except ValueError as error:
|
|
return jsonify({"error": str(error)}), 400
|
|
with get_db() as connection:
|
|
if name.casefold() != type_name.casefold() and connection.execute("SELECT 1 FROM inventory_types WHERE name = ? COLLATE NOCASE", (name,)).fetchone():
|
|
return jsonify({"error": "That inventory type already exists."}), 409
|
|
result = connection.execute("UPDATE inventory_types SET name = ?, fields = ? WHERE name = ? COLLATE NOCASE", (name, json.dumps(fields), type_name))
|
|
if result.rowcount:
|
|
connection.execute("UPDATE inventory_items SET item_type = ? WHERE item_type = ? COLLATE NOCASE", (name, type_name))
|
|
if result.rowcount == 0:
|
|
return jsonify({"error": "Inventory type not found."}), 404
|
|
return jsonify({"name": name, "fields": fields})
|
|
|
|
|
|
@app.delete("/api/types/<path:type_name>")
|
|
@permission_required("admin")
|
|
def delete_type(type_name):
|
|
with get_db() as connection:
|
|
count = connection.execute("SELECT COUNT(*) FROM inventory_items WHERE item_type = ? COLLATE NOCASE", (type_name,)).fetchone()[0]
|
|
if count:
|
|
return jsonify({"error": "This type is used by existing inventory and cannot be removed."}), 400
|
|
result = connection.execute("DELETE FROM inventory_types WHERE name = ? COLLATE NOCASE", (type_name,))
|
|
if result.rowcount == 0:
|
|
return jsonify({"error": "Inventory type not found."}), 404
|
|
return jsonify({"deleted": True})
|
|
|
|
|
|
@app.get("/api/summary")
|
|
@login_required
|
|
def summary():
|
|
with get_db() as connection:
|
|
total = connection.execute("SELECT COUNT(*) FROM inventory_items").fetchone()[0]
|
|
available = connection.execute("SELECT COUNT(*) FROM inventory_items WHERE status = 'Available'").fetchone()[0]
|
|
checked_out = connection.execute("SELECT COUNT(*) FROM inventory_items WHERE status = 'Checked out'").fetchone()[0]
|
|
maintenance = connection.execute("SELECT COUNT(*) FROM inventory_items WHERE status = 'Maintenance'").fetchone()[0]
|
|
return jsonify({"total": total, "available": available, "checked_out": checked_out, "maintenance": maintenance})
|
|
|
|
|
|
@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)
|
|
try:
|
|
with get_db() as source, sqlite3.connect(temporary_database_path) as destination:
|
|
source.backup(destination)
|
|
archive = tempfile.NamedTemporaryFile(suffix=".zip", delete=False)
|
|
archive_path = Path(archive.name)
|
|
archive.close()
|
|
with zipfile.ZipFile(archive_path, "w", zipfile.ZIP_DEFLATED) as zip_archive:
|
|
zip_archive.write(temporary_database_path, "inventory.db")
|
|
add_directory_to_zip(zip_archive, UPLOAD_FOLDER, "uploads")
|
|
response = send_file(archive_path, as_attachment=True, download_name=f"stockroom-backup-{datetime.now().strftime('%Y%m%d-%H%M%S')}.zip", mimetype="application/zip")
|
|
response.call_on_close(lambda: archive_path.unlink(missing_ok=True))
|
|
return response
|
|
finally:
|
|
temporary_database_path.unlink(missing_ok=True)
|
|
|
|
|
|
@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:
|
|
return jsonify({"error": "Choose a backup ZIP file to restore."}), 400
|
|
try:
|
|
restore_backup(backup_file)
|
|
except (ValueError, zipfile.BadZipFile, sqlite3.DatabaseError) as error:
|
|
return jsonify({"error": str(error) or "The backup could not be restored."}), 400
|
|
return jsonify({"restored": True})
|
|
|
|
|
|
@app.post("/api/items")
|
|
@permission_required("write")
|
|
def create_item():
|
|
data = item_payload(request_data())
|
|
if not data["name"] or not data["item_type"]:
|
|
return jsonify({"error": "Name and type are required."}), 400
|
|
selected_files = [image for image in files_from_request() if image and image.filename]
|
|
if len(selected_files) > 5:
|
|
return jsonify({"error": "Each inventory item can have up to 5 pictures."}), 400
|
|
try:
|
|
images = save_images(selected_files)
|
|
except ValueError as error:
|
|
return jsonify({"error": str(error)}), 400
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
try:
|
|
with get_db() as connection:
|
|
cursor = connection.execute(
|
|
"""INSERT INTO inventory_items
|
|
(name, item_type, brand, model, serial_number, asset_tag, status, location, destination, checkout_date, purchase_date, notes, cpu, ram, gpu, storage, media_format, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(*data.values(), now, now),
|
|
)
|
|
for filename, original_name in images:
|
|
connection.execute("INSERT INTO inventory_images (item_id, filename, original_name, created_at) VALUES (?, ?, ?, ?)", (cursor.lastrowid, filename, original_name, now))
|
|
row = connection.execute("SELECT * FROM inventory_items WHERE id = ?", (cursor.lastrowid,)).fetchone()
|
|
result = serialize(row, connection)
|
|
return jsonify(result), 201
|
|
except sqlite3.IntegrityError:
|
|
for filename, _ in images:
|
|
delete_image(filename)
|
|
return jsonify({"error": "Serial number or asset tag already exists."}), 409
|
|
|
|
|
|
@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"]:
|
|
return jsonify({"error": "Name and type are required."}), 400
|
|
data["updated_at"] = datetime.now(timezone.utc).isoformat()
|
|
new_images = files_from_request()
|
|
saved_images = []
|
|
try:
|
|
with get_db() as connection:
|
|
current = connection.execute("SELECT image_filename FROM inventory_items WHERE id = ?", (item_id,)).fetchone()
|
|
if not current:
|
|
return jsonify({"error": "Item not found."}), 404
|
|
existing_count = connection.execute("SELECT COUNT(*) FROM inventory_images WHERE item_id = ?", (item_id,)).fetchone()[0]
|
|
if existing_count + len([image for image in new_images if image and image.filename]) > 5:
|
|
return jsonify({"error": "Each inventory item can have up to 5 pictures."}), 400
|
|
saved_images = save_images(new_images)
|
|
result = connection.execute(
|
|
"""UPDATE inventory_items SET name=?, item_type=?, brand=?, model=?, serial_number=?, asset_tag=?, status=?, location=?, destination=?, checkout_date=?, purchase_date=?, notes=?, cpu=?, ram=?, gpu=?, storage=?, media_format=?, updated_at=? WHERE id=?""",
|
|
(*data.values(), item_id),
|
|
)
|
|
for filename, original_name in saved_images:
|
|
connection.execute("INSERT INTO inventory_images (item_id, filename, original_name, created_at) VALUES (?, ?, ?, ?)", (item_id, filename, original_name, data["updated_at"]))
|
|
row = connection.execute("SELECT * FROM inventory_items WHERE id = ?", (item_id,)).fetchone()
|
|
result = serialize(row, connection)
|
|
return jsonify(result)
|
|
except ValueError as error:
|
|
return jsonify({"error": str(error)}), 400
|
|
except sqlite3.IntegrityError:
|
|
for filename, _ in saved_images:
|
|
delete_image(filename)
|
|
return jsonify({"error": "Serial number or asset tag already exists."}), 409
|
|
|
|
|
|
@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()
|
|
result = connection.execute("DELETE FROM inventory_items WHERE id = ?", (item_id,))
|
|
if result.rowcount == 0:
|
|
return jsonify({"error": "Item not found."}), 404
|
|
for image in images:
|
|
delete_image(image["filename"])
|
|
return jsonify({"deleted": True})
|
|
|
|
|
|
@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:
|
|
if not connection.execute("SELECT id FROM inventory_items WHERE id = ?", (item_id,)).fetchone():
|
|
return jsonify({"error": "Item not found."}), 404
|
|
current_count = connection.execute("SELECT COUNT(*) FROM inventory_images WHERE item_id = ?", (item_id,)).fetchone()[0]
|
|
selected_files = [image for image in files if image and image.filename]
|
|
if current_count + len(selected_files) > 5:
|
|
return jsonify({"error": "Each inventory item can have up to 5 pictures."}), 400
|
|
try:
|
|
saved = save_images(selected_files)
|
|
except ValueError as error:
|
|
return jsonify({"error": str(error)}), 400
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
with get_db() as connection:
|
|
for filename, original_name in saved:
|
|
connection.execute("INSERT INTO inventory_images (item_id, filename, original_name, created_at) VALUES (?, ?, ?, ?)", (item_id, filename, original_name, now))
|
|
row = connection.execute("SELECT * FROM inventory_items WHERE id = ?", (item_id,)).fetchone()
|
|
return jsonify(serialize(row, connection))
|
|
|
|
|
|
@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()
|
|
if not image:
|
|
return jsonify({"error": "Picture not found."}), 404
|
|
connection.execute("DELETE FROM inventory_images WHERE id = ?", (image_id,))
|
|
item = connection.execute("SELECT * FROM inventory_items WHERE id = ?", (item_id,)).fetchone()
|
|
result = serialize(item, connection)
|
|
delete_image(image["filename"])
|
|
return jsonify(result)
|
|
|
|
|
|
initialize_db()
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=5000, debug=False)
|