388 lines
17 KiB
Python
388 lines
17 KiB
Python
import os
|
|
import shutil
|
|
import sqlite3
|
|
import tempfile
|
|
import uuid
|
|
import zipfile
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from flask import Flask, jsonify, render_template, request, send_file, send_from_directory
|
|
from werkzeug.utils import secure_filename
|
|
|
|
app = Flask(__name__)
|
|
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
|
|
|
|
|
|
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,
|
|
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);
|
|
"""
|
|
)
|
|
columns = {row[1] for row in connection.execute("PRAGMA table_info(inventory_items)")}
|
|
if "image_filename" not in columns:
|
|
connection.execute("ALTER TABLE inventory_items ADD COLUMN image_filename TEXT")
|
|
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)"""
|
|
)
|
|
|
|
|
|
def clean(value):
|
|
return (value or "").strip()
|
|
|
|
|
|
def item_payload(data):
|
|
return {
|
|
"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")),
|
|
"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")),
|
|
}
|
|
|
|
|
|
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)
|
|
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return render_template("index.html")
|
|
|
|
|
|
@app.get("/media/<path:filename>")
|
|
def media(filename):
|
|
return send_from_directory(UPLOAD_FOLDER, filename)
|
|
|
|
|
|
@app.get("/api/items")
|
|
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 ?)")
|
|
params.extend([f"%{query}%"] * 7)
|
|
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/summary")
|
|
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")
|
|
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")
|
|
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")
|
|
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, 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>")
|
|
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=?, 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>")
|
|
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")
|
|
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>")
|
|
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)
|