ethiele/keyVerwaltung
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

um Poolfahrzeuge erweitert

Browse Source
This commit is contained in:
Erik Thiele
2026-05-19 20:22:18 +02:00
parent 60a5dc39b2
commit 0199a21a66
12 changed files with 487 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

138
app.py
View File

@@ -26,6 +26,7 @@ logging.basicConfig(
ASSET_LABELS = {
"chip": "Tuerchip",
"parking_card": "Parkkarte",
"pool_vehicle": "Poolfahrzeug",
}
ACTION_LABELS = {
@@ -38,6 +39,12 @@ PRINT_DESCRIPTIONS = {
"return": "Rueckgabebestaetigung fuer die Ruecknahme eines Mediums",
}
INPUT_PLACEHOLDERS = {
"chip": "z. B. CHIP-1001",
"parking_card": "z. B. PARK-2001",
"pool_vehicle": "z. B. GZ-CC-123",
}
IMPORT_ACTIONS = {"import", "ausgabe", "assign"}
IMPORT_HEADER = ["user", "typ", "kennung", "aktion"]
@@ -79,13 +86,15 @@ def init_db() -> None:
chip_assigned_at TEXT,
parking_card_code TEXT,
parking_card_assigned_at TEXT,
pool_vehicle_code TEXT,
pool_vehicle_assigned_at TEXT,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS transactions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
asset_type TEXT NOT NULL CHECK (asset_type IN ('chip', 'parking_card')),
asset_type TEXT NOT NULL CHECK (asset_type IN ('chip', 'parking_card', 'pool_vehicle')),
asset_code TEXT NOT NULL,
handled_by TEXT,
action TEXT NOT NULL CHECK (action IN ('assign', 'return')),
@@ -112,6 +121,35 @@ def init_db() -> None:
if "handled_by" not in columns:
db.execute("ALTER TABLE transactions ADD COLUMN handled_by TEXT")
transactions_sql = db.execute(
"SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'transactions'"
).fetchone()
if transactions_sql and "pool_vehicle" not in (transactions_sql["sql"] or ""):
db.execute("DROP TABLE transactions")
db.execute(
"""
CREATE TABLE transactions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
asset_type TEXT NOT NULL CHECK (asset_type IN ('chip', 'parking_card', 'pool_vehicle')),
asset_code TEXT NOT NULL,
handled_by TEXT,
action TEXT NOT NULL CHECK (action IN ('assign', 'return')),
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id)
)
"""
)
user_columns = {
row["name"]
for row in db.execute("PRAGMA table_info(users)").fetchall()
}
if "pool_vehicle_code" not in user_columns:
db.execute("ALTER TABLE users ADD COLUMN pool_vehicle_code TEXT")
if "pool_vehicle_assigned_at" not in user_columns:
db.execute("ALTER TABLE users ADD COLUMN pool_vehicle_assigned_at TEXT")
admin_exists = db.execute(
"SELECT id FROM staff_users WHERE role = 'admin' LIMIT 1"
).fetchone()
@@ -194,7 +232,11 @@ def read_recent_logs(limit: int = 20) -> list[str]:
def asset_code_in_use(db: sqlite3.Connection, asset_type: str, asset_code: str) -> bool:
column_name = "chip_code" if asset_type == "chip" else "parking_card_code"
column_name = {
"chip": "chip_code",
"parking_card": "parking_card_code",
"pool_vehicle": "pool_vehicle_code",
}[asset_type]
existing = db.execute(
f"SELECT id FROM users WHERE {column_name} = ?",
(asset_code,),
@@ -210,6 +252,9 @@ def normalize_asset_type(value: str) -> str | None:
"parkkarte": "parking_card",
"parking_card": "parking_card",
"parking card": "parking_card",
"poolfahrzeug": "pool_vehicle",
"pool vehicle": "pool_vehicle",
"pool_vehicle": "pool_vehicle",
}
return aliases.get(normalized)
@@ -461,17 +506,20 @@ def import_data() -> str:
current_chip_code = user["chip_code"] if user else None
current_card_code = user["parking_card_code"] if user else None
current_vehicle_code = user["pool_vehicle_code"] if user else None
user_id = user["id"] if user else None
if pending_user is not None:
current_chip_code = pending_user["chip_code"]
current_card_code = pending_user["parking_card_code"]
current_vehicle_code = pending_user["pool_vehicle_code"]
if user is None and pending_user is None:
pending_user = {
"full_name": full_name,
"chip_code": None,
"parking_card_code": None,
"pool_vehicle_code": None,
}
operations.append(("user", pending_user))
@@ -480,6 +528,8 @@ def import_data() -> str:
asset_type == "chip" and current_chip_code == asset_code
) or (
asset_type == "parking_card" and current_card_code == asset_code
) or (
asset_type == "pool_vehicle" and current_vehicle_code == asset_code
)
if not assigned_to_same_user:
errors.append(f"Zeile {index}: Kennung '{asset_code}' ist bereits vergeben.")
@@ -504,12 +554,18 @@ def import_data() -> str:
continue
if pending_user is not None:
pending_user["chip_code"] = asset_code
else:
elif asset_type == "parking_card":
if current_card_code and current_card_code != asset_code:
errors.append(f"Zeile {index}: User '{full_name}' hat bereits eine andere Parkkarte.")
continue
if pending_user is not None:
pending_user["parking_card_code"] = asset_code
else:
if current_vehicle_code and current_vehicle_code != asset_code:
errors.append(f"Zeile {index}: User '{full_name}' hat bereits ein anderes Poolfahrzeug.")
continue
if pending_user is not None:
pending_user["pool_vehicle_code"] = asset_code
operations.append(
(
@@ -549,11 +605,16 @@ def import_data() -> str:
"UPDATE users SET chip_code = ?, chip_assigned_at = CURRENT_TIMESTAMP WHERE id = ?",
(payload["asset_code"], resolved_user_id),
)
else:
elif payload["asset_type"] == "parking_card":
db.execute(
"UPDATE users SET parking_card_code = ?, parking_card_assigned_at = CURRENT_TIMESTAMP WHERE id = ?",
(payload["asset_code"], resolved_user_id),
)
else:
db.execute(
"UPDATE users SET pool_vehicle_code = ?, pool_vehicle_assigned_at = CURRENT_TIMESTAMP WHERE id = ?",
(payload["asset_code"], resolved_user_id),
)
db.execute(
"INSERT INTO transactions (user_id, asset_type, asset_code, handled_by, action) VALUES (?, ?, ?, ?, 'assign')",
@@ -577,7 +638,8 @@ def index() -> str:
params: tuple[str, ...] = ()
user_query = """
SELECT id, full_name, email, department, chip_code, chip_assigned_at,
parking_card_code, parking_card_assigned_at
parking_card_code, parking_card_assigned_at,
pool_vehicle_code, pool_vehicle_assigned_at
FROM users
"""
@@ -586,10 +648,11 @@ def index() -> str:
user_query += """
WHERE full_name LIKE ?
OR email LIKE ?
OR chip_code LIKE ?
OR parking_card_code LIKE ?
OR chip_code LIKE ?
OR parking_card_code LIKE ?
OR pool_vehicle_code LIKE ?
"""
params = (like_value, like_value, like_value, like_value)
params = (like_value, like_value, like_value, like_value, like_value)
user_query += " ORDER BY full_name COLLATE NOCASE"
users = db.execute(user_query, params).fetchall()
@@ -599,7 +662,8 @@ def index() -> str:
SELECT
COUNT(*) AS users,
SUM(CASE WHEN chip_code IS NOT NULL AND chip_code != '' THEN 1 ELSE 0 END) AS active_chips,
SUM(CASE WHEN parking_card_code IS NOT NULL AND parking_card_code != '' THEN 1 ELSE 0 END) AS active_cards
SUM(CASE WHEN parking_card_code IS NOT NULL AND parking_card_code != '' THEN 1 ELSE 0 END) AS active_cards,
SUM(CASE WHEN pool_vehicle_code IS NOT NULL AND pool_vehicle_code != '' THEN 1 ELSE 0 END) AS active_pool_vehicles
FROM users
"""
).fetchone()
@@ -624,6 +688,7 @@ def index() -> str:
"users": stats["users"],
"active_chips": stats["active_chips"],
"active_cards": stats["active_cards"],
"active_pool_vehicles": stats["active_pool_vehicles"],
},
search_query=search_query,
asset_labels=ASSET_LABELS,
@@ -660,14 +725,19 @@ def create_user() -> str:
@login_required
def assign_asset() -> str:
db = get_db()
asset_type = request.form.get("asset_type", "chip").strip() if request.method == "POST" else request.args.get("asset_type", "chip").strip()
if asset_type not in {"chip", "parking_card", "pool_vehicle"}:
asset_type = "chip"
if request.method == "POST":
user_id = request.form.get("user_id", "").strip()
asset_type = request.form.get("asset_type", "").strip()
asset_code = request.form.get("asset_code", "").strip()
handled_by = g.current_staff["full_name"]
if not user_id or asset_type not in {"chip", "parking_card"} or not asset_code:
if not user_id or asset_type not in {"chip", "parking_card", "pool_vehicle"}:
flash("Bitte alle Felder fuer die Ausgabe ausfuellen.")
return redirect(url_for("assign_asset"))
if not asset_code:
flash("Bitte alle Felder fuer die Ausgabe ausfuellen.")
return redirect(url_for("assign_asset"))
@@ -677,7 +747,7 @@ def assign_asset() -> str:
return redirect(url_for("assign_asset"))
if asset_code_in_use(db, asset_type, asset_code):
flash("Diese Kennung ist bereits vergeben.")
flash("Dieses Kennzeichen ist bereits vergeben." if asset_type == "pool_vehicle" else "Diese Kennung ist bereits vergeben.")
return redirect(url_for("assign_asset"))
if asset_type == "chip":
@@ -688,7 +758,7 @@ def assign_asset() -> str:
"UPDATE users SET chip_code = ?, chip_assigned_at = CURRENT_TIMESTAMP WHERE id = ?",
(asset_code, user_id),
)
else:
elif asset_type == "parking_card":
if user["parking_card_code"]:
flash("Dieser User hat bereits eine aktive Parkkarte.")
return redirect(url_for("assign_asset"))
@@ -696,6 +766,14 @@ def assign_asset() -> str:
"UPDATE users SET parking_card_code = ?, parking_card_assigned_at = CURRENT_TIMESTAMP WHERE id = ?",
(asset_code, user_id),
)
else:
if user["pool_vehicle_code"]:
flash("Dieser User hat bereits ein aktives Poolfahrzeug.")
return redirect(url_for("assign_asset"))
db.execute(
"UPDATE users SET pool_vehicle_code = ?, pool_vehicle_assigned_at = CURRENT_TIMESTAMP WHERE id = ?",
(asset_code, user_id),
)
db.execute(
"INSERT INTO transactions (user_id, asset_type, asset_code, handled_by, action) VALUES (?, ?, ?, ?, 'assign')",
@@ -708,20 +786,27 @@ def assign_asset() -> str:
return redirect(url_for("print_transaction", transaction_id=transaction_id))
users = db.execute("SELECT id, full_name FROM users ORDER BY full_name COLLATE NOCASE").fetchall()
return render_template("assign_asset.html", users=users)
return render_template(
"assign_asset.html",
users=users,
selected_asset_type=asset_type,
input_placeholder=INPUT_PLACEHOLDERS[asset_type],
)
@app.route("/return", methods=["GET", "POST"])
@login_required
def return_asset() -> str:
db = get_db()
asset_type = request.form.get("asset_type", "chip").strip() if request.method == "POST" else request.args.get("asset_type", "chip").strip()
if asset_type not in {"chip", "parking_card", "pool_vehicle"}:
asset_type = "chip"
if request.method == "POST":
user_id = request.form.get("user_id", "").strip()
asset_type = request.form.get("asset_type", "").strip()
handled_by = g.current_staff["full_name"]
if not user_id or asset_type not in {"chip", "parking_card"}:
if not user_id or asset_type not in {"chip", "parking_card", "pool_vehicle"}:
flash("Bitte User und Typ fuer die Rueckgabe waehlen.")
return redirect(url_for("return_asset"))
@@ -730,7 +815,13 @@ def return_asset() -> str:
flash("Ausgewaehlter User wurde nicht gefunden.")
return redirect(url_for("return_asset"))
asset_code = user["chip_code"] if asset_type == "chip" else user["parking_card_code"]
asset_code = (
user["chip_code"]
if asset_type == "chip"
else user["parking_card_code"]
if asset_type == "parking_card"
else user["pool_vehicle_code"]
)
if not asset_code:
flash("Fuer diesen User ist kein entsprechendes Medium aktiv hinterlegt.")
return redirect(url_for("return_asset"))
@@ -740,11 +831,16 @@ def return_asset() -> str:
"UPDATE users SET chip_code = NULL, chip_assigned_at = NULL WHERE id = ?",
(user_id,),
)
else:
elif asset_type == "parking_card":
db.execute(
"UPDATE users SET parking_card_code = NULL, parking_card_assigned_at = NULL WHERE id = ?",
(user_id,),
)
else:
db.execute(
"UPDATE users SET pool_vehicle_code = NULL, pool_vehicle_assigned_at = NULL WHERE id = ?",
(user_id,),
)
db.execute(
"INSERT INTO transactions (user_id, asset_type, asset_code, handled_by, action) VALUES (?, ?, ?, ?, 'return')",
@@ -757,7 +853,11 @@ def return_asset() -> str:
return redirect(url_for("print_transaction", transaction_id=transaction_id))
users = db.execute("SELECT id, full_name FROM users ORDER BY full_name COLLATE NOCASE").fetchall()
return render_template("return_asset.html", users=users)
return render_template(
"return_asset.html",
users=users,
selected_asset_type=asset_type,
)
@app.route("/transactions/<int:transaction_id>/print")