IntelSight/Hetzner-Backup
3
0
Fork 0
Code Issues Pull-Requests Actions Pakete Projekte Releases Wiki Aktivität

bug fix - route probleme

Quellcode durchsuchen
Dieser Commit ist enthalten in:
UserIsMH
2025-06-18 19:40:14 +02:00
Ursprung 4bfe1983a3
Commit 55e8a66f9f
21 geänderte Dateien mit 3847 neuen und 97 gelöschten Zeilen
Patch-Datei herunterladen Vergleichs-Datei herunterladen Alle Dateien ausklappen Alle Dateien einklappen

70
v2_adminpanel/routes/api_routes.py
Datei anzeigen

@@ -55,7 +55,7 @@ def api_customers():
@api_bp.route("/license/<int:license_id>/toggle", methods=["POST"])
@login_required
def toggle_license(license_id):
"""Toggle license active status"""
"""Toggle license is_active status"""
conn = get_connection()
cur = conn.cursor()
@@ -192,32 +192,38 @@ def get_license_devices(license_id):
cur.execute("""
SELECT
dr.id,
dr.device_id,
dr.hardware_id,
dr.device_name,
dr.device_type,
dr.registration_date,
dr.first_seen as registration_date,
dr.last_seen,
dr.is_active,
dr.operating_system,
dr.ip_address,
(SELECT COUNT(*) FROM sessions s
WHERE s.license_key = dr.license_key
AND s.device_id = dr.device_id
AND s.active = true) as active_sessions
WHERE s.license_key = l.license_key
AND s.hardware_id = dr.hardware_id
AND s.is_active = true) as active_sessions
FROM device_registrations dr
WHERE dr.license_key = %s
ORDER BY dr.registration_date DESC
JOIN licenses l ON dr.license_id = l.id
WHERE l.license_key = %s
ORDER BY dr.first_seen DESC
""", (license_data['license_key'],))
devices = []
for row in cur.fetchall():
devices.append({
'id': row[0],
'device_id': row[1],
'hardware_id': row[1],
'device_name': row[2],
'device_type': row[3],
'registration_date': row[4].isoformat() if row[4] else None,
'last_seen': row[5].isoformat() if row[5] else None,
'is_active': row[6],
'active_sessions': row[7]
'operating_system': row[7] or 'Unknown',
'ip_address': row[8] or 'Unknown',
'active_sessions': row[9],
'first_seen': row[4].isoformat() if row[4] else None
})
return jsonify({
@@ -241,11 +247,11 @@ def register_device(license_id):
"""Registriere ein neues Gerät für eine Lizenz"""
data = request.get_json()
device_id = data.get('device_id')
hardware_id = data.get('hardware_id')
device_name = data.get('device_name')
device_type = data.get('device_type', 'unknown')
if not device_id or not device_name:
if not hardware_id or not device_name:
return jsonify({'error': 'Geräte-ID und Name erforderlich'}), 400
conn = get_connection()
@@ -259,8 +265,9 @@ def register_device(license_id):
# Prüfe Gerätelimit
cur.execute("""
SELECT COUNT(*) FROM device_registrations
WHERE license_key = %s AND is_active = true
SELECT COUNT(*) FROM device_registrations dr
JOIN licenses l ON dr.license_id = l.id
WHERE l.license_key = %s AND dr.is_active = true
""", (license_data['license_key'],))
active_device_count = cur.fetchone()[0]
@@ -270,9 +277,10 @@ def register_device(license_id):
# Prüfe ob Gerät bereits registriert
cur.execute("""
SELECT id, is_active FROM device_registrations
WHERE license_key = %s AND device_id = %s
""", (license_data['license_key'], device_id))
SELECT dr.id, dr.is_active FROM device_registrations dr
JOIN licenses l ON dr.license_id = l.id
WHERE l.license_key = %s AND dr.hardware_id = %s
""", (license_data['license_key'], hardware_id))
existing = cur.fetchone()
@@ -290,15 +298,15 @@ def register_device(license_id):
# Registriere neues Gerät
cur.execute("""
INSERT INTO device_registrations
(license_key, device_id, device_name, device_type, is_active)
(license_id, hardware_id, device_name, device_type, is_active)
VALUES (%s, %s, %s, %s, true)
""", (license_data['license_key'], device_id, device_name, device_type))
""", (license_id, hardware_id, device_name, device_type))
conn.commit()
# Audit-Log
log_audit('DEVICE_REGISTER', 'license', license_id,
additional_info=f"Gerät {device_name} ({device_id}) registriert")
additional_info=f"Gerät {device_name} ({hardware_id}) registriert")
return jsonify({'success': True})
@@ -321,7 +329,7 @@ def deactivate_device(license_id, device_id):
try:
# Prüfe ob Gerät zur Lizenz gehört
cur.execute("""
SELECT dr.device_name, dr.device_id, l.license_key
SELECT dr.device_name, dr.hardware_id, l.license_key
FROM device_registrations dr
JOIN licenses l ON dr.license_key = l.license_key
WHERE dr.id = %s AND l.id = %s
@@ -341,8 +349,8 @@ def deactivate_device(license_id, device_id):
# Beende aktive Sessions
cur.execute("""
UPDATE sessions
SET active = false, logout_time = CURRENT_TIMESTAMP
WHERE license_key = %s AND device_id = %s AND active = true
SET is_active = false, ended_at = CURRENT_TIMESTAMP
WHERE license_key = %s AND hardware_id = %s AND is_active = true
""", (device[2], device[1]))
conn.commit()
@@ -450,11 +458,11 @@ def quick_edit_license(license_id):
old_values['valid_until'] = str(current_license['valid_until'])
new_values['valid_until'] = data['valid_until']
if 'active' in data:
if 'is_active' in data:
updates.append("is_active = %s")
params.append(bool(data['active']))
params.append(bool(data['is_active']))
old_values['is_active'] = current_license['is_active']
new_values['is_active'] = bool(data['active'])
new_values['is_active'] = bool(data['is_active'])
if not updates:
return jsonify({'error': 'Keine Änderungen angegeben'}), 400
@@ -846,10 +854,10 @@ def global_search():
# Suche in Sessions
cur.execute("""
SELECT id, license_key, username, device_id, active
SELECT id, license_key, username, hardware_id, is_active
FROM sessions
WHERE username ILIKE %s OR device_id ILIKE %s
ORDER BY login_time DESC
WHERE username ILIKE %s OR hardware_id ILIKE %s
ORDER BY started_at DESC
LIMIT 10
""", (f'%{query}%', f'%{query}%'))
@@ -858,8 +866,8 @@ def global_search():
'id': row[0],
'license_key': row[1],
'username': row[2],
'device_id': row[3],
'active': row[4]
'hardware_id': row[3],
'is_active': row[4]
})
return jsonify(results)

921
v2_adminpanel/routes/api_routes.py.backup Normale Datei
Datei anzeigen

@@ -0,0 +1,921 @@
import logging
from datetime import datetime
from zoneinfo import ZoneInfo
from flask import Blueprint, request, jsonify, session
import config
from auth.decorators import login_required
from utils.audit import log_audit
from utils.network import get_client_ip
from utils.license import generate_license_key
from db import get_connection, get_db_connection, get_db_cursor
from models import get_license_by_id, get_customers
# Create Blueprint
api_bp = Blueprint('api', __name__, url_prefix='/api')
@api_bp.route("/customers", methods=["GET"])
@login_required
def api_customers():
"""API endpoint for customer search (used by Select2)"""
search = request.args.get('q', '').strip()
page = int(request.args.get('page', 1))
per_page = 20
try:
# Get all customers (with optional search)
customers = get_customers(show_test=True, search=search)
# Pagination
start = (page - 1) * per_page
end = start + per_page
paginated_customers = customers[start:end]
# Format for Select2
results = []
for customer in paginated_customers:
results.append({
'id': customer['id'],
'text': f"{customer['name']} ({customer['email'] or 'keine E-Mail'})"
})
return jsonify({
'results': results,
'pagination': {
'more': len(customers) > end
}
})
except Exception as e:
logging.error(f"Error in api_customers: {str(e)}")
return jsonify({'error': 'Fehler beim Laden der Kunden'}), 500
@api_bp.route("/license/<int:license_id>/toggle", methods=["POST"])
@login_required
def toggle_license(license_id):
"""Toggle license active status"""
conn = get_connection()
cur = conn.cursor()
try:
# Get current status
license_data = get_license_by_id(license_id)
if not license_data:
return jsonify({'error': 'Lizenz nicht gefunden'}), 404
new_status = not license_data['is_active']
# Update status
cur.execute("UPDATE licenses SET is_active = %s WHERE id = %s", (new_status, license_id))
conn.commit()
# Log change
log_audit('TOGGLE', 'license', license_id,
old_values={'is_active': license_data['is_active']},
new_values={'is_active': new_status})
return jsonify({'success': True, 'is_active': new_status})
except Exception as e:
conn.rollback()
logging.error(f"Fehler beim Umschalten der Lizenz: {str(e)}", exc_info=True)
return jsonify({'error': 'Fehler beim Umschalten der Lizenz'}), 500
finally:
cur.close()
conn.close()
@api_bp.route("/licenses/bulk-activate", methods=["POST"])
@login_required
def bulk_activate_licenses():
"""Aktiviere mehrere Lizenzen gleichzeitig"""
data = request.get_json()
license_ids = data.get('license_ids', [])
if not license_ids:
return jsonify({'error': 'Keine Lizenzen ausgewählt'}), 400
conn = get_connection()
cur = conn.cursor()
try:
# Update all selected licenses
cur.execute("""
UPDATE licenses
SET is_active = true
WHERE id = ANY(%s) AND is_active = false
RETURNING id
""", (license_ids,))
updated_ids = [row[0] for row in cur.fetchall()]
conn.commit()
# Log changes
for license_id in updated_ids:
log_audit('BULK_ACTIVATE', 'license', license_id,
new_values={'is_active': True})
return jsonify({
'success': True,
'updated_count': len(updated_ids)
})
except Exception as e:
conn.rollback()
logging.error(f"Fehler beim Bulk-Aktivieren: {str(e)}")
return jsonify({'error': 'Fehler beim Aktivieren der Lizenzen'}), 500
finally:
cur.close()
conn.close()
@api_bp.route("/licenses/bulk-deactivate", methods=["POST"])
@login_required
def bulk_deactivate_licenses():
"""Deaktiviere mehrere Lizenzen gleichzeitig"""
data = request.get_json()
license_ids = data.get('license_ids', [])
if not license_ids:
return jsonify({'error': 'Keine Lizenzen ausgewählt'}), 400
conn = get_connection()
cur = conn.cursor()
try:
# Update all selected licenses
cur.execute("""
UPDATE licenses
SET is_active = false
WHERE id = ANY(%s) AND is_active = true
RETURNING id
""", (license_ids,))
updated_ids = [row[0] for row in cur.fetchall()]
conn.commit()
# Log changes
for license_id in updated_ids:
log_audit('BULK_DEACTIVATE', 'license', license_id,
new_values={'is_active': False})
return jsonify({
'success': True,
'updated_count': len(updated_ids)
})
except Exception as e:
conn.rollback()
logging.error(f"Fehler beim Bulk-Deaktivieren: {str(e)}")
return jsonify({'error': 'Fehler beim Deaktivieren der Lizenzen'}), 500
finally:
cur.close()
conn.close()
@api_bp.route("/license/<int:license_id>/devices")
@login_required
def get_license_devices(license_id):
"""Hole alle Geräte einer Lizenz"""
conn = get_connection()
cur = conn.cursor()
try:
# Hole Lizenz-Info
license_data = get_license_by_id(license_id)
if not license_data:
return jsonify({'error': 'Lizenz nicht gefunden'}), 404
# Hole registrierte Geräte
cur.execute("""
SELECT
dr.id,
dr.device_id,
dr.device_name,
dr.device_type,
dr.registration_date,
dr.last_seen,
dr.is_active,
(SELECT COUNT(*) FROM sessions s
WHERE s.license_key = dr.license_key
AND s.device_id = dr.device_id
AND s.active = true) as active_sessions
FROM device_registrations dr
WHERE dr.license_key = %s
ORDER BY dr.registration_date DESC
""", (license_data['license_key'],))
devices = []
for row in cur.fetchall():
devices.append({
'id': row[0],
'device_id': row[1],
'device_name': row[2],
'device_type': row[3],
'registration_date': row[4].isoformat() if row[4] else None,
'last_seen': row[5].isoformat() if row[5] else None,
'is_active': row[6],
'active_sessions': row[7]
})
return jsonify({
'license_key': license_data['license_key'],
'device_limit': license_data['device_limit'],
'devices': devices,
'device_count': len(devices)
})
except Exception as e:
logging.error(f"Fehler beim Abrufen der Geräte: {str(e)}")
return jsonify({'error': 'Fehler beim Abrufen der Geräte'}), 500
finally:
cur.close()
conn.close()
@api_bp.route("/license/<int:license_id>/register-device", methods=["POST"])
@login_required
def register_device(license_id):
"""Registriere ein neues Gerät für eine Lizenz"""
data = request.get_json()
device_id = data.get('device_id')
device_name = data.get('device_name')
device_type = data.get('device_type', 'unknown')
if not device_id or not device_name:
return jsonify({'error': 'Geräte-ID und Name erforderlich'}), 400
conn = get_connection()
cur = conn.cursor()
try:
# Hole Lizenz-Info
license_data = get_license_by_id(license_id)
if not license_data:
return jsonify({'error': 'Lizenz nicht gefunden'}), 404
# Prüfe Gerätelimit
cur.execute("""
SELECT COUNT(*) FROM device_registrations
WHERE license_key = %s AND is_active = true
""", (license_data['license_key'],))
active_device_count = cur.fetchone()[0]
if active_device_count >= license_data['device_limit']:
return jsonify({'error': 'Gerätelimit erreicht'}), 400
# Prüfe ob Gerät bereits registriert
cur.execute("""
SELECT id, is_active FROM device_registrations
WHERE license_key = %s AND device_id = %s
""", (license_data['license_key'], device_id))
existing = cur.fetchone()
if existing:
if existing[1]: # is_active
return jsonify({'error': 'Gerät bereits registriert'}), 400
else:
# Reaktiviere Gerät
cur.execute("""
UPDATE device_registrations
SET is_active = true, last_seen = CURRENT_TIMESTAMP
WHERE id = %s
""", (existing[0],))
else:
# Registriere neues Gerät
cur.execute("""
INSERT INTO device_registrations
(license_key, device_id, device_name, device_type, is_active)
VALUES (%s, %s, %s, %s, true)
""", (license_data['license_key'], device_id, device_name, device_type))
conn.commit()
# Audit-Log
log_audit('DEVICE_REGISTER', 'license', license_id,
additional_info=f"Gerät {device_name} ({device_id}) registriert")
return jsonify({'success': True})
except Exception as e:
conn.rollback()
logging.error(f"Fehler beim Registrieren des Geräts: {str(e)}")
return jsonify({'error': 'Fehler beim Registrieren des Geräts'}), 500
finally:
cur.close()
conn.close()
@api_bp.route("/license/<int:license_id>/deactivate-device/<int:device_id>", methods=["POST"])
@login_required
def deactivate_device(license_id, device_id):
"""Deaktiviere ein Gerät einer Lizenz"""
conn = get_connection()
cur = conn.cursor()
try:
# Prüfe ob Gerät zur Lizenz gehört
cur.execute("""
SELECT dr.device_name, dr.device_id, l.license_key
FROM device_registrations dr
JOIN licenses l ON dr.license_key = l.license_key
WHERE dr.id = %s AND l.id = %s
""", (device_id, license_id))
device = cur.fetchone()
if not device:
return jsonify({'error': 'Gerät nicht gefunden'}), 404
# Deaktiviere Gerät
cur.execute("""
UPDATE device_registrations
SET is_active = false
WHERE id = %s
""", (device_id,))
# Beende aktive Sessions
cur.execute("""
UPDATE sessions
SET active = false, logout_time = CURRENT_TIMESTAMP
WHERE license_key = %s AND device_id = %s AND active = true
""", (device[2], device[1]))
conn.commit()
# Audit-Log
log_audit('DEVICE_DEACTIVATE', 'license', license_id,
additional_info=f"Gerät {device[0]} ({device[1]}) deaktiviert")
return jsonify({'success': True})
except Exception as e:
conn.rollback()
logging.error(f"Fehler beim Deaktivieren des Geräts: {str(e)}")
return jsonify({'error': 'Fehler beim Deaktivieren des Geräts'}), 500
finally:
cur.close()
conn.close()
@api_bp.route("/licenses/bulk-delete", methods=["POST"])
@login_required
def bulk_delete_licenses():
"""Lösche mehrere Lizenzen gleichzeitig"""
data = request.get_json()
license_ids = data.get('license_ids', [])
if not license_ids:
return jsonify({'error': 'Keine Lizenzen ausgewählt'}), 400
conn = get_connection()
cur = conn.cursor()
try:
deleted_count = 0
for license_id in license_ids:
# Hole Lizenz-Info für Audit
cur.execute("SELECT license_key FROM licenses WHERE id = %s", (license_id,))
result = cur.fetchone()
if result:
license_key = result[0]
# Lösche Sessions
cur.execute("DELETE FROM sessions WHERE license_key = %s", (license_key,))
# Lösche Geräte-Registrierungen
cur.execute("DELETE FROM device_registrations WHERE license_key = %s", (license_key,))
# Lösche Lizenz
cur.execute("DELETE FROM licenses WHERE id = %s", (license_id,))
# Audit-Log
log_audit('BULK_DELETE', 'license', license_id,
old_values={'license_key': license_key})
deleted_count += 1
conn.commit()
return jsonify({
'success': True,
'deleted_count': deleted_count
})
except Exception as e:
conn.rollback()
logging.error(f"Fehler beim Bulk-Löschen: {str(e)}")
return jsonify({'error': 'Fehler beim Löschen der Lizenzen'}), 500
finally:
cur.close()
conn.close()
@api_bp.route("/license/<int:license_id>/quick-edit", methods=['POST'])
@login_required
def quick_edit_license(license_id):
"""Schnellbearbeitung einer Lizenz"""
data = request.get_json()
conn = get_connection()
cur = conn.cursor()
try:
# Hole aktuelle Lizenz für Vergleich
current_license = get_license_by_id(license_id)
if not current_license:
return jsonify({'error': 'Lizenz nicht gefunden'}), 404
# Update nur die übergebenen Felder
updates = []
params = []
old_values = {}
new_values = {}
if 'device_limit' in data:
updates.append("device_limit = %s")
params.append(int(data['device_limit']))
old_values['device_limit'] = current_license['device_limit']
new_values['device_limit'] = int(data['device_limit'])
if 'valid_until' in data:
updates.append("valid_until = %s")
params.append(data['valid_until'])
old_values['valid_until'] = str(current_license['valid_until'])
new_values['valid_until'] = data['valid_until']
if 'active' in data:
updates.append("is_active = %s")
params.append(bool(data['active']))
old_values['is_active'] = current_license['is_active']
new_values['is_active'] = bool(data['active'])
if not updates:
return jsonify({'error': 'Keine Änderungen angegeben'}), 400
# Führe Update aus
params.append(license_id)
cur.execute(f"""
UPDATE licenses
SET {', '.join(updates)}
WHERE id = %s
""", params)
conn.commit()
# Audit-Log
log_audit('QUICK_EDIT', 'license', license_id,
old_values=old_values,
new_values=new_values)
return jsonify({'success': True})
except Exception as e:
conn.rollback()
logging.error(f"Fehler bei Schnellbearbeitung: {str(e)}")
return jsonify({'error': 'Fehler bei der Bearbeitung'}), 500
finally:
cur.close()
conn.close()
@api_bp.route("/license/<int:license_id>/resources")
@login_required
def get_license_resources(license_id):
"""Hole alle Ressourcen einer Lizenz"""
conn = get_connection()
cur = conn.cursor()
try:
# Hole Lizenz-Info
license_data = get_license_by_id(license_id)
if not license_data:
return jsonify({'error': 'Lizenz nicht gefunden'}), 404
# Hole zugewiesene Ressourcen
cur.execute("""
SELECT
rp.id,
rp.resource_type,
rp.resource_value,
rp.is_test,
rp.status_changed_at,
lr.assigned_at,
lr.assigned_by
FROM resource_pools rp
JOIN license_resources lr ON rp.id = lr.resource_id
WHERE lr.license_id = %s
ORDER BY rp.resource_type, rp.resource_value
""", (license_id,))
resources = []
for row in cur.fetchall():
resources.append({
'id': row[0],
'type': row[1],
'value': row[2],
'is_test': row[3],
'status_changed_at': row[4].isoformat() if row[4] else None,
'assigned_at': row[5].isoformat() if row[5] else None,
'assigned_by': row[6]
})
# Gruppiere nach Typ
grouped = {}
for resource in resources:
res_type = resource['type']
if res_type not in grouped:
grouped[res_type] = []
grouped[res_type].append(resource)
return jsonify({
'license_key': license_data['license_key'],
'resources': resources,
'grouped': grouped,
'total_count': len(resources)
})
except Exception as e:
logging.error(f"Fehler beim Abrufen der Ressourcen: {str(e)}")
return jsonify({'error': 'Fehler beim Abrufen der Ressourcen'}), 500
finally:
cur.close()
conn.close()
@api_bp.route("/resources/allocate", methods=['POST'])
@login_required
def allocate_resources():
"""Weise Ressourcen einer Lizenz zu"""
data = request.get_json()
license_id = data.get('license_id')
resource_ids = data.get('resource_ids', [])
if not license_id or not resource_ids:
return jsonify({'error': 'Lizenz-ID und Ressourcen erforderlich'}), 400
conn = get_connection()
cur = conn.cursor()
try:
# Prüfe Lizenz
license_data = get_license_by_id(license_id)
if not license_data:
return jsonify({'error': 'Lizenz nicht gefunden'}), 404
allocated_count = 0
errors = []
for resource_id in resource_ids:
try:
# Prüfe ob Ressource verfügbar ist
cur.execute("""
SELECT resource_value, status, is_test
FROM resource_pools
WHERE id = %s
""", (resource_id,))
resource = cur.fetchone()
if not resource:
errors.append(f"Ressource {resource_id} nicht gefunden")
continue
if resource[1] != 'available':
errors.append(f"Ressource {resource[0]} ist nicht verfügbar")
continue
# Prüfe Test/Produktion Kompatibilität
if resource[2] != license_data['is_test']:
errors.append(f"Ressource {resource[0]} ist {'Test' if resource[2] else 'Produktion'}, Lizenz ist {'Test' if license_data['is_test'] else 'Produktion'}")
continue
# Weise Ressource zu
cur.execute("""
UPDATE resource_pools
SET status = 'allocated',
allocated_to_license = %s,
status_changed_at = CURRENT_TIMESTAMP,
status_changed_by = %s
WHERE id = %s
""", (license_id, session['username'], resource_id))
# Erstelle Verknüpfung
cur.execute("""
INSERT INTO license_resources (license_id, resource_id, assigned_by)
VALUES (%s, %s, %s)
""", (license_id, resource_id, session['username']))
# History-Eintrag
cur.execute("""
INSERT INTO resource_history (resource_id, license_id, action, action_by, ip_address)
VALUES (%s, %s, 'allocated', %s, %s)
""", (resource_id, license_id, session['username'], get_client_ip()))
allocated_count += 1
except Exception as e:
errors.append(f"Fehler bei Ressource {resource_id}: {str(e)}")
conn.commit()
# Audit-Log
if allocated_count > 0:
log_audit('RESOURCE_ALLOCATE', 'license', license_id,
additional_info=f"{allocated_count} Ressourcen zugewiesen")
return jsonify({
'success': True,
'allocated_count': allocated_count,
'errors': errors
})
except Exception as e:
conn.rollback()
logging.error(f"Fehler beim Zuweisen der Ressourcen: {str(e)}")
return jsonify({'error': 'Fehler beim Zuweisen der Ressourcen'}), 500
finally:
cur.close()
conn.close()
@api_bp.route("/resources/check-availability", methods=['GET'])
@login_required
def check_resource_availability():
"""Prüfe Verfügbarkeit von Ressourcen"""
# Einzelne Ressource prüfen (alte API)
resource_type = request.args.get('type')
if resource_type:
count = int(request.args.get('count', 1))
is_test = request.args.get('is_test', 'false') == 'true'
show_test = request.args.get('show_test', 'false') == 'true'
conn = get_connection()
cur = conn.cursor()
try:
# Hole verfügbare Ressourcen mit Details
if show_test:
# Zeige alle verfügbaren Ressourcen (Test und Produktion)
cur.execute("""
SELECT id, resource_value, is_test
FROM resource_pools
WHERE resource_type = %s
AND status = 'available'
ORDER BY is_test, resource_value
LIMIT %s
""", (resource_type, count))
else:
# Zeige nur Produktions-Ressourcen
cur.execute("""
SELECT id, resource_value, is_test
FROM resource_pools
WHERE resource_type = %s
AND status = 'available'
AND is_test = false
ORDER BY resource_value
LIMIT %s
""", (resource_type, count))
available_resources = []
for row in cur.fetchall():
available_resources.append({
'id': row[0],
'value': row[1],
'is_test': row[2]
})
return jsonify({
'resource_type': resource_type,
'requested': count,
'available': available_resources,
'sufficient': len(available_resources) >= count,
'show_test': show_test
})
except Exception as e:
logging.error(f"Fehler beim Prüfen der Verfügbarkeit: {str(e)}")
return jsonify({'error': 'Fehler beim Prüfen der Verfügbarkeit'}), 500
finally:
cur.close()
conn.close()
# Mehrere Ressourcen gleichzeitig prüfen (für Batch)
domain_count = int(request.args.get('domain', 0))
ipv4_count = int(request.args.get('ipv4', 0))
phone_count = int(request.args.get('phone', 0))
is_test = request.args.get('is_test', 'false') == 'true'
conn = get_connection()
cur = conn.cursor()
try:
# Zähle verfügbare Ressourcen für jeden Typ
result = {}
# Domains
cur.execute("""
SELECT COUNT(*)
FROM resource_pools
WHERE resource_type = 'domain'
AND status = 'available'
AND is_test = %s
""", (is_test,))
domain_available = cur.fetchone()[0]
# IPv4
cur.execute("""
SELECT COUNT(*)
FROM resource_pools
WHERE resource_type = 'ipv4'
AND status = 'available'
AND is_test = %s
""", (is_test,))
ipv4_available = cur.fetchone()[0]
# Phones
cur.execute("""
SELECT COUNT(*)
FROM resource_pools
WHERE resource_type = 'phone'
AND status = 'available'
AND is_test = %s
""", (is_test,))
phone_available = cur.fetchone()[0]
return jsonify({
'domain_requested': domain_count,
'domain_available': domain_available,
'domain_sufficient': domain_available >= domain_count,
'ipv4_requested': ipv4_count,
'ipv4_available': ipv4_available,
'ipv4_sufficient': ipv4_available >= ipv4_count,
'phone_requested': phone_count,
'phone_available': phone_available,
'phone_sufficient': phone_available >= phone_count,
'all_sufficient': (
domain_available >= domain_count and
ipv4_available >= ipv4_count and
phone_available >= phone_count
),
'is_test': is_test
})
except Exception as e:
logging.error(f"Fehler beim Prüfen der Verfügbarkeit: {str(e)}")
return jsonify({'error': 'Fehler beim Prüfen der Verfügbarkeit'}), 500
finally:
cur.close()
conn.close()
@api_bp.route("/global-search", methods=['GET'])
@login_required
def global_search():
"""Globale Suche über alle Entitäten"""
query = request.args.get('q', '').strip()
if not query or len(query) < 3:
return jsonify({'error': 'Suchbegriff muss mindestens 3 Zeichen haben'}), 400
conn = get_connection()
cur = conn.cursor()
results = {
'licenses': [],
'customers': [],
'resources': [],
'sessions': []
}
try:
# Suche in Lizenzen
cur.execute("""
SELECT id, license_key, customer_name, is_active
FROM licenses
WHERE license_key ILIKE %s
OR customer_name ILIKE %s
OR customer_email ILIKE %s
LIMIT 10
""", (f'%{query}%', f'%{query}%', f'%{query}%'))
for row in cur.fetchall():
results['licenses'].append({
'id': row[0],
'license_key': row[1],
'customer_name': row[2],
'is_active': row[3]
})
# Suche in Kunden
cur.execute("""
SELECT id, name, email
FROM customers
WHERE name ILIKE %s OR email ILIKE %s
LIMIT 10
""", (f'%{query}%', f'%{query}%'))
for row in cur.fetchall():
results['customers'].append({
'id': row[0],
'name': row[1],
'email': row[2]
})
# Suche in Ressourcen
cur.execute("""
SELECT id, resource_type, resource_value, status
FROM resource_pools
WHERE resource_value ILIKE %s
LIMIT 10
""", (f'%{query}%',))
for row in cur.fetchall():
results['resources'].append({
'id': row[0],
'type': row[1],
'value': row[2],
'status': row[3]
})
# Suche in Sessions
cur.execute("""
SELECT id, license_key, username, device_id, active
FROM sessions
WHERE username ILIKE %s OR device_id ILIKE %s
ORDER BY login_time DESC
LIMIT 10
""", (f'%{query}%', f'%{query}%'))
for row in cur.fetchall():
results['sessions'].append({
'id': row[0],
'license_key': row[1],
'username': row[2],
'device_id': row[3],
'active': row[4]
})
return jsonify(results)
except Exception as e:
logging.error(f"Fehler bei der globalen Suche: {str(e)}")
return jsonify({'error': 'Fehler bei der Suche'}), 500
finally:
cur.close()
conn.close()
@api_bp.route("/generate-license-key", methods=['POST'])
@login_required
def api_generate_key():
"""API Endpoint zur Generierung eines neuen Lizenzschlüssels"""
try:
# Lizenztyp aus Request holen (default: full)
data = request.get_json() or {}
license_type = data.get('type', 'full')
# Key generieren
key = generate_license_key(license_type)
# Prüfen ob Key bereits existiert (sehr unwahrscheinlich aber sicher ist sicher)
conn = get_connection()
cur = conn.cursor()
# Wiederhole bis eindeutiger Key gefunden
attempts = 0
while attempts < 10: # Max 10 Versuche
cur.execute("SELECT 1 FROM licenses WHERE license_key = %s", (key,))
if not cur.fetchone():
break # Key ist eindeutig
key = generate_license_key(license_type)
attempts += 1
cur.close()
conn.close()
# Log für Audit
log_audit('GENERATE_KEY', 'license',
additional_info={'type': license_type, 'key': key})
return jsonify({
'success': True,
'key': key,
'type': license_type
})
except Exception as e:
logging.error(f"Fehler bei Key-Generierung: {str(e)}")
return jsonify({
'success': False,
'error': 'Fehler bei der Key-Generierung'
}), 500

4
v2_adminpanel/routes/batch_routes.py
Datei anzeigen

@@ -209,8 +209,8 @@ def batch_update():
params.append(int(request.form['device_limit']))
if 'update_active' in request.form:
updates.append("active = %s")
params.append('active' in request.form)
updates.append("is_active = %s")
params.append('is_active' in request.form)
if not updates:
flash('Keine Änderungen angegeben!', 'error')

373
v2_adminpanel/routes/batch_routes.py.backup Normale Datei
Datei anzeigen

@@ -0,0 +1,373 @@
import os
import logging
import secrets
import string
from datetime import datetime, timedelta
from pathlib import Path
from flask import Blueprint, render_template, request, redirect, session, url_for, flash, send_file
import config
from auth.decorators import login_required
from utils.audit import log_audit
from utils.network import get_client_ip
from utils.export import create_batch_export
from db import get_connection, get_db_connection, get_db_cursor
from models import get_customers
# Create Blueprint
batch_bp = Blueprint('batch', __name__)
def generate_license_key():
"""Generiert einen zufälligen Lizenzschlüssel"""
chars = string.ascii_uppercase + string.digits
return '-'.join([''.join(secrets.choice(chars) for _ in range(4)) for _ in range(4)])
@batch_bp.route("/batch", methods=["GET", "POST"])
@login_required
def batch_create():
"""Batch-Erstellung von Lizenzen"""
customers = get_customers()
if request.method == "POST":
conn = get_connection()
cur = conn.cursor()
try:
# Form data
customer_id = int(request.form['customer_id'])
license_type = request.form['license_type']
count = int(request.form['quantity']) # Korrigiert von 'count' zu 'quantity'
valid_from = request.form['valid_from']
valid_until = request.form['valid_until']
device_limit = int(request.form['device_limit'])
is_test = 'is_test' in request.form
# Validierung
if count < 1 or count > 100:
flash('Anzahl muss zwischen 1 und 100 liegen!', 'error')
return redirect(url_for('batch.batch_create'))
# Hole Kundendaten
cur.execute("SELECT name, email FROM customers WHERE id = %s", (customer_id,))
customer = cur.fetchone()
if not customer:
flash('Kunde nicht gefunden!', 'error')
return redirect(url_for('batch.batch_create'))
created_licenses = []
# Erstelle Lizenzen
for i in range(count):
license_key = generate_license_key()
# Prüfe ob Schlüssel bereits existiert
while True:
cur.execute("SELECT id FROM licenses WHERE license_key = %s", (license_key,))
if not cur.fetchone():
break
license_key = generate_license_key()
# Erstelle Lizenz
cur.execute("""
INSERT INTO licenses (
license_key, customer_id,
license_type, valid_from, valid_until, device_limit,
is_test, created_at
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
RETURNING id
""", (
license_key, customer_id,
license_type, valid_from, valid_until, device_limit,
is_test, datetime.now()
))
license_id = cur.fetchone()[0]
created_licenses.append({
'id': license_id,
'license_key': license_key
})
# Audit-Log
log_audit('CREATE', 'license', license_id,
new_values={
'license_key': license_key,
'customer_name': customer[0],
'batch_creation': True
})
conn.commit()
# Speichere erstellte Lizenzen in Session für Export
session['batch_created_licenses'] = created_licenses
session['batch_customer_name'] = customer[0]
session['batch_customer_email'] = customer[1]
flash(f'{count} Lizenzen erfolgreich erstellt!', 'success')
# Weiterleitung zum Export
return redirect(url_for('batch.batch_export'))
except Exception as e:
conn.rollback()
logging.error(f"Fehler bei Batch-Erstellung: {str(e)}")
flash('Fehler bei der Batch-Erstellung!', 'error')
finally:
cur.close()
conn.close()
return render_template("batch_form.html", customers=customers)
@batch_bp.route("/batch/export")
@login_required
def batch_export():
"""Exportiert die zuletzt erstellten Batch-Lizenzen"""
created_licenses = session.get('batch_created_licenses', [])
if not created_licenses:
flash('Keine Lizenzen zum Exportieren gefunden!', 'error')
return redirect(url_for('batch.batch_create'))
conn = get_connection()
cur = conn.cursor()
try:
# Hole vollständige Lizenzdaten
license_ids = [l['id'] for l in created_licenses]
cur.execute("""
SELECT
l.license_key, c.name, c.email,
l.license_type, l.valid_from, l.valid_until,
l.device_limit, l.is_test, l.created_at
FROM licenses l
JOIN customers c ON l.customer_id = c.id
WHERE l.id = ANY(%s)
ORDER BY l.id
""", (license_ids,))
licenses = []
for row in cur.fetchall():
licenses.append({
'license_key': row[0],
'customer_name': row[1],
'customer_email': row[2],
'license_type': row[3],
'valid_from': row[4],
'valid_until': row[5],
'device_limit': row[6],
'is_test': row[7],
'created_at': row[8]
})
# Lösche aus Session
session.pop('batch_created_licenses', None)
session.pop('batch_customer_name', None)
session.pop('batch_customer_email', None)
# Erstelle und sende Excel-Export
return create_batch_export(licenses)
except Exception as e:
logging.error(f"Fehler beim Export: {str(e)}")
flash('Fehler beim Exportieren der Lizenzen!', 'error')
return redirect(url_for('batch.batch_create'))
finally:
cur.close()
conn.close()
@batch_bp.route("/batch/update", methods=["GET", "POST"])
@login_required
def batch_update():
"""Batch-Update von Lizenzen"""
if request.method == "POST":
conn = get_connection()
cur = conn.cursor()
try:
# Form data
license_keys = request.form.get('license_keys', '').strip().split('\n')
license_keys = [key.strip() for key in license_keys if key.strip()]
if not license_keys:
flash('Keine Lizenzschlüssel angegeben!', 'error')
return redirect(url_for('batch.batch_update'))
# Update-Parameter
updates = []
params = []
if 'update_valid_until' in request.form and request.form['valid_until']:
updates.append("valid_until = %s")
params.append(request.form['valid_until'])
if 'update_device_limit' in request.form and request.form['device_limit']:
updates.append("device_limit = %s")
params.append(int(request.form['device_limit']))
if 'update_active' in request.form:
updates.append("active = %s")
params.append('active' in request.form)
if not updates:
flash('Keine Änderungen angegeben!', 'error')
return redirect(url_for('batch.batch_update'))
# Führe Updates aus
updated_count = 0
not_found = []
for license_key in license_keys:
# Prüfe ob Lizenz existiert
cur.execute("SELECT id FROM licenses WHERE license_key = %s", (license_key,))
result = cur.fetchone()
if not result:
not_found.append(license_key)
continue
license_id = result[0]
# Update ausführen
update_params = params + [license_id]
cur.execute(f"""
UPDATE licenses
SET {', '.join(updates)}
WHERE id = %s
""", update_params)
# Audit-Log
log_audit('BATCH_UPDATE', 'license', license_id,
additional_info=f"Batch-Update: {', '.join(updates)}")
updated_count += 1
conn.commit()
# Feedback
flash(f'{updated_count} Lizenzen erfolgreich aktualisiert!', 'success')
if not_found:
flash(f'{len(not_found)} Lizenzen nicht gefunden: {", ".join(not_found[:5])}{"..." if len(not_found) > 5 else ""}', 'warning')
except Exception as e:
conn.rollback()
logging.error(f"Fehler bei Batch-Update: {str(e)}")
flash('Fehler beim Batch-Update!', 'error')
finally:
cur.close()
conn.close()
return render_template("batch_update.html")
@batch_bp.route("/batch/import", methods=["GET", "POST"])
@login_required
def batch_import():
"""Import von Lizenzen aus CSV/Excel"""
if request.method == "POST":
if 'file' not in request.files:
flash('Keine Datei ausgewählt!', 'error')
return redirect(url_for('batch.batch_import'))
file = request.files['file']
if file.filename == '':
flash('Keine Datei ausgewählt!', 'error')
return redirect(url_for('batch.batch_import'))
# Verarbeite Datei
try:
import pandas as pd
# Lese Datei
if file.filename.endswith('.csv'):
df = pd.read_csv(file)
elif file.filename.endswith(('.xlsx', '.xls')):
df = pd.read_excel(file)
else:
flash('Ungültiges Dateiformat! Nur CSV und Excel erlaubt.', 'error')
return redirect(url_for('batch.batch_import'))
# Validiere Spalten
required_columns = ['customer_email', 'license_type', 'valid_from', 'valid_until', 'device_limit']
missing_columns = [col for col in required_columns if col not in df.columns]
if missing_columns:
flash(f'Fehlende Spalten: {", ".join(missing_columns)}', 'error')
return redirect(url_for('batch.batch_import'))
conn = get_connection()
cur = conn.cursor()
imported_count = 0
errors = []
for index, row in df.iterrows():
try:
# Finde oder erstelle Kunde
email = row['customer_email']
cur.execute("SELECT id, name FROM customers WHERE email = %s", (email,))
customer = cur.fetchone()
if not customer:
# Erstelle neuen Kunden
name = row.get('customer_name', email.split('@')[0])
cur.execute("""
INSERT INTO customers (name, email, created_at)
VALUES (%s, %s, %s)
RETURNING id
""", (name, email, datetime.now()))
customer_id = cur.fetchone()[0]
customer_name = name
else:
customer_id = customer[0]
customer_name = customer[1]
# Generiere Lizenzschlüssel
license_key = row.get('license_key', generate_license_key())
# Erstelle Lizenz
cur.execute("""
INSERT INTO licenses (
license_key, customer_id,
license_type, valid_from, valid_until, device_limit,
is_test, created_at
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
RETURNING id
""", (
license_key, customer_id,
row['license_type'], row['valid_from'], row['valid_until'],
int(row['device_limit']), row.get('is_test', False),
datetime.now()
))
license_id = cur.fetchone()[0]
imported_count += 1
# Audit-Log
log_audit('IMPORT', 'license', license_id,
additional_info=f"Importiert aus {file.filename}")
except Exception as e:
errors.append(f"Zeile {index + 2}: {str(e)}")
conn.commit()
# Feedback
flash(f'{imported_count} Lizenzen erfolgreich importiert!', 'success')
if errors:
flash(f'{len(errors)} Fehler aufgetreten. Erste Fehler: {"; ".join(errors[:3])}', 'warning')
except Exception as e:
logging.error(f"Fehler beim Import: {str(e)}")
flash(f'Fehler beim Import: {str(e)}', 'error')
finally:
if 'conn' in locals():
cur.close()
conn.close()
return render_template("batch_import.html")

2
v2_adminpanel/routes/customer_routes.py
Datei anzeigen

@@ -389,7 +389,7 @@ def api_customer_licenses(customer_id):
'id': row[0],
'license_key': row[1],
'license_type': row[2],
'is_active': row[3], # Korrigiert von 'active' zu 'is_active'
'is_active': row[3], # Korrigiert von 'is_active' zu 'is_active'
'is_test': row[4],
'valid_from': row[5].strftime('%Y-%m-%d') if row[5] else None,
'valid_until': row[6].strftime('%Y-%m-%d') if row[6] else None,

461
v2_adminpanel/routes/customer_routes.py.backup Normale Datei
Datei anzeigen

@@ -0,0 +1,461 @@
import os
import logging
from datetime import datetime
from zoneinfo import ZoneInfo
from flask import Blueprint, render_template, request, redirect, session, url_for, flash, jsonify
import config
from auth.decorators import login_required
from utils.audit import log_audit
from db import get_connection, get_db_connection, get_db_cursor
from models import get_customers, get_customer_by_id
# Create Blueprint
customer_bp = Blueprint('customers', __name__)
# Test route
@customer_bp.route("/test-customers")
def test_customers():
return "Customer blueprint is working!"
@customer_bp.route("/customers")
@login_required
def customers():
show_test = request.args.get('show_test', 'false').lower() == 'true'
search = request.args.get('search', '').strip()
page = request.args.get('page', 1, type=int)
per_page = 20
sort = request.args.get('sort', 'name')
order = request.args.get('order', 'asc')
customers_list = get_customers(show_test=show_test, search=search)
# Sortierung
if sort == 'name':
customers_list.sort(key=lambda x: x['name'].lower(), reverse=(order == 'desc'))
elif sort == 'email':
customers_list.sort(key=lambda x: x['email'].lower(), reverse=(order == 'desc'))
elif sort == 'created_at':
customers_list.sort(key=lambda x: x['created_at'], reverse=(order == 'desc'))
# Paginierung
total_customers = len(customers_list)
total_pages = (total_customers + per_page - 1) // per_page
start = (page - 1) * per_page
end = start + per_page
paginated_customers = customers_list[start:end]
return render_template("customers.html",
customers=paginated_customers,
show_test=show_test,
search=search,
page=page,
per_page=per_page,
total_pages=total_pages,
total_customers=total_customers,
sort=sort,
order=order,
current_order=order)
@customer_bp.route("/customer/edit/<int:customer_id>", methods=["GET", "POST"])
@login_required
def edit_customer(customer_id):
if request.method == "POST":
try:
# Get current customer data for comparison
current_customer = get_customer_by_id(customer_id)
if not current_customer:
flash('Kunde nicht gefunden!', 'error')
return redirect(url_for('customers.customers'))
with get_db_connection() as conn:
cur = conn.cursor()
try:
# Update customer data
new_values = {
'name': request.form['name'],
'email': request.form['email'],
'is_test': 'is_test' in request.form
}
cur.execute("""
UPDATE customers
SET name = %s, email = %s, is_test = %s
WHERE id = %s
""", (
new_values['name'],
new_values['email'],
new_values['is_test'],
customer_id
))
conn.commit()
# Log changes
log_audit('UPDATE', 'customer', customer_id,
old_values={
'name': current_customer['name'],
'email': current_customer['email'],
'is_test': current_customer.get('is_test', False)
},
new_values=new_values)
flash('Kunde erfolgreich aktualisiert!', 'success')
# Redirect mit show_test Parameter wenn nötig
redirect_url = url_for('customers.customers')
if request.form.get('show_test') == 'true':
redirect_url += '?show_test=true'
return redirect(redirect_url)
finally:
cur.close()
except Exception as e:
logging.error(f"Fehler beim Aktualisieren des Kunden: {str(e)}")
flash('Fehler beim Aktualisieren des Kunden!', 'error')
return redirect(url_for('customers.customers'))
# GET request
customer_data = get_customer_by_id(customer_id)
if not customer_data:
flash('Kunde nicht gefunden!', 'error')
return redirect(url_for('customers.customers'))
return render_template("edit_customer.html", customer=customer_data)
@customer_bp.route("/customer/create", methods=["GET", "POST"])
@login_required
def create_customer():
if request.method == "POST":
conn = get_connection()
cur = conn.cursor()
try:
# Insert new customer
name = request.form['name']
email = request.form['email']
cur.execute("""
INSERT INTO customers (name, email, created_at)
VALUES (%s, %s, %s)
RETURNING id
""", (name, email, datetime.now()))
customer_id = cur.fetchone()[0]
conn.commit()
# Log creation
log_audit('CREATE', 'customer', customer_id,
new_values={
'name': name,
'email': email
})
flash(f'Kunde {name} erfolgreich erstellt!', 'success')
return redirect(url_for('customers.customers'))
except Exception as e:
conn.rollback()
logging.error(f"Fehler beim Erstellen des Kunden: {str(e)}")
flash('Fehler beim Erstellen des Kunden!', 'error')
finally:
cur.close()
conn.close()
return render_template("create_customer.html")
@customer_bp.route("/customer/delete/<int:customer_id>", methods=["POST"])
@login_required
def delete_customer(customer_id):
conn = get_connection()
cur = conn.cursor()
try:
# Get customer data before deletion
customer_data = get_customer_by_id(customer_id)
if not customer_data:
flash('Kunde nicht gefunden!', 'error')
return redirect(url_for('customers.customers'))
# Check if customer has licenses
cur.execute("SELECT COUNT(*) FROM licenses WHERE customer_id = %s", (customer_id,))
license_count = cur.fetchone()[0]
if license_count > 0:
flash(f'Kunde kann nicht gelöscht werden - hat noch {license_count} Lizenz(en)!', 'error')
return redirect(url_for('customers.customers'))
# Delete the customer
cur.execute("DELETE FROM customers WHERE id = %s", (customer_id,))
conn.commit()
# Log deletion
log_audit('DELETE', 'customer', customer_id,
old_values={
'name': customer_data['name'],
'email': customer_data['email']
})
flash(f'Kunde {customer_data["name"]} erfolgreich gelöscht!', 'success')
except Exception as e:
conn.rollback()
logging.error(f"Fehler beim Löschen des Kunden: {str(e)}")
flash('Fehler beim Löschen des Kunden!', 'error')
finally:
cur.close()
conn.close()
return redirect(url_for('customers.customers'))
@customer_bp.route("/customers-licenses")
@login_required
def customers_licenses():
"""Zeigt die Übersicht von Kunden und deren Lizenzen"""
import logging
import psycopg2
logging.info("=== CUSTOMERS-LICENSES ROUTE CALLED ===")
# Get show_test parameter from URL
show_test = request.args.get('show_test', 'false').lower() == 'true'
logging.info(f"show_test parameter: {show_test}")
try:
# Direkte Verbindung ohne Helper-Funktionen
conn = psycopg2.connect(
host=os.getenv("POSTGRES_HOST", "postgres"),
port=os.getenv("POSTGRES_PORT", "5432"),
dbname=os.getenv("POSTGRES_DB"),
user=os.getenv("POSTGRES_USER"),
password=os.getenv("POSTGRES_PASSWORD")
)
conn.set_client_encoding('UTF8')
cur = conn.cursor()
try:
# Hole alle Kunden mit ihren Lizenzen
# Wenn show_test=false, zeige nur Nicht-Test-Kunden
query = """
SELECT
c.id,
c.name,
c.email,
c.created_at,
COUNT(l.id),
COUNT(CASE WHEN l.is_active = true THEN 1 END),
COUNT(CASE WHEN l.is_test = true THEN 1 END),
MAX(l.created_at),
c.is_test
FROM customers c
LEFT JOIN licenses l ON c.id = l.customer_id
WHERE (%s OR c.is_test = false)
GROUP BY c.id, c.name, c.email, c.created_at, c.is_test
ORDER BY c.name
"""
cur.execute(query, (show_test,))
customers = []
results = cur.fetchall()
logging.info(f"=== QUERY RETURNED {len(results)} ROWS ===")
for idx, row in enumerate(results):
logging.info(f"Row {idx}: Type={type(row)}, Length={len(row) if hasattr(row, '__len__') else 'N/A'}")
customers.append({
'id': row[0],
'name': row[1],
'email': row[2],
'created_at': row[3],
'license_count': row[4],
'active_licenses': row[5],
'test_licenses': row[6],
'last_license_created': row[7],
'is_test': row[8]
})
return render_template("customers_licenses.html",
customers=customers,
show_test=show_test)
finally:
cur.close()
conn.close()
except Exception as e:
import traceback
error_details = f"Fehler beim Laden der Kunden-Lizenz-Übersicht: {str(e)}\nType: {type(e)}\nTraceback: {traceback.format_exc()}"
logging.error(error_details)
flash(f'Datenbankfehler: {str(e)}', 'error')
return redirect(url_for('admin.dashboard'))
@customer_bp.route("/api/customer/<int:customer_id>/licenses")
@login_required
def api_customer_licenses(customer_id):
"""API-Endpunkt für die Lizenzen eines Kunden"""
conn = get_connection()
cur = conn.cursor()
try:
# Hole Kundeninformationen
customer = get_customer_by_id(customer_id)
if not customer:
return jsonify({'error': 'Kunde nicht gefunden'}), 404
# Hole alle Lizenzen des Kunden
cur.execute("""
SELECT
l.id,
l.license_key,
l.license_type,
l.is_active,
l.is_test,
l.valid_from,
l.valid_until,
l.device_limit,
l.created_at,
(SELECT COUNT(*) FROM sessions s WHERE s.license_id = l.id AND s.is_active = true) as active_sessions,
(SELECT COUNT(DISTINCT hardware_id) FROM device_registrations dr WHERE dr.license_id = l.id) as registered_devices,
CASE
WHEN l.valid_until < CURRENT_DATE THEN 'abgelaufen'
WHEN l.valid_until < CURRENT_DATE + INTERVAL '30 days' THEN 'läuft bald ab'
WHEN l.is_active = false THEN 'inaktiv'
ELSE 'aktiv'
END as status,
l.domain_count,
l.ipv4_count,
l.phone_count,
(SELECT COUNT(*) FROM device_registrations WHERE license_id = l.id AND is_active = TRUE) as active_devices,
-- Actual resource counts
(SELECT COUNT(*) FROM license_resources lr
JOIN resource_pools rp ON lr.resource_id = rp.id
WHERE lr.license_id = l.id AND lr.is_active = true AND rp.resource_type = 'domain') as actual_domain_count,
(SELECT COUNT(*) FROM license_resources lr
JOIN resource_pools rp ON lr.resource_id = rp.id
WHERE lr.license_id = l.id AND lr.is_active = true AND rp.resource_type = 'ipv4') as actual_ipv4_count,
(SELECT COUNT(*) FROM license_resources lr
JOIN resource_pools rp ON lr.resource_id = rp.id
WHERE lr.license_id = l.id AND lr.is_active = true AND rp.resource_type = 'phone') as actual_phone_count
FROM licenses l
WHERE l.customer_id = %s
ORDER BY l.created_at DESC
""", (customer_id,))
licenses = []
for row in cur.fetchall():
license_id = row[0]
# Hole die konkreten zugewiesenen Ressourcen für diese Lizenz
conn2 = get_connection()
cur2 = conn2.cursor()
cur2.execute("""
SELECT rp.id, rp.resource_type, rp.resource_value, lr.assigned_at
FROM resource_pools rp
JOIN license_resources lr ON rp.id = lr.resource_id
WHERE lr.license_id = %s AND lr.is_active = true
ORDER BY rp.resource_type, rp.resource_value
""", (license_id,))
resources = {
'domains': [],
'ipv4s': [],
'phones': []
}
for res_row in cur2.fetchall():
resource_data = {
'id': res_row[0],
'value': res_row[2],
'assigned_at': res_row[3].strftime('%Y-%m-%d %H:%M:%S') if res_row[3] else None
}
if res_row[1] == 'domain':
resources['domains'].append(resource_data)
elif res_row[1] == 'ipv4':
resources['ipv4s'].append(resource_data)
elif res_row[1] == 'phone':
resources['phones'].append(resource_data)
cur2.close()
conn2.close()
licenses.append({
'id': row[0],
'license_key': row[1],
'license_type': row[2],
'is_active': row[3], # Korrigiert von 'active' zu 'is_active'
'is_test': row[4],
'valid_from': row[5].strftime('%Y-%m-%d') if row[5] else None,
'valid_until': row[6].strftime('%Y-%m-%d') if row[6] else None,
'device_limit': row[7],
'created_at': row[8].strftime('%Y-%m-%d %H:%M:%S') if row[8] else None,
'active_sessions': row[9],
'registered_devices': row[10],
'status': row[11],
'domain_count': row[12],
'ipv4_count': row[13],
'phone_count': row[14],
'active_devices': row[15],
'actual_domain_count': row[16],
'actual_ipv4_count': row[17],
'actual_phone_count': row[18],
'resources': resources
})
return jsonify({
'success': True, # Wichtig: Frontend erwartet dieses Feld
'customer': {
'id': customer['id'],
'name': customer['name'],
'email': customer['email']
},
'licenses': licenses
})
except Exception as e:
logging.error(f"Fehler beim Laden der Kundenlizenzen: {str(e)}")
return jsonify({'error': 'Fehler beim Laden der Daten'}), 500
finally:
cur.close()
conn.close()
@customer_bp.route("/api/customer/<int:customer_id>/quick-stats")
@login_required
def api_customer_quick_stats(customer_id):
"""Schnelle Statistiken für einen Kunden"""
conn = get_connection()
cur = conn.cursor()
try:
cur.execute("""
SELECT
COUNT(l.id) as total_licenses,
COUNT(CASE WHEN l.is_active = true THEN 1 END) as active_licenses,
COUNT(CASE WHEN l.is_test = true THEN 1 END) as test_licenses,
SUM(l.device_limit) as total_device_limit
FROM licenses l
WHERE l.customer_id = %s
""", (customer_id,))
row = cur.fetchone()
return jsonify({
'total_licenses': row[0] or 0,
'active_licenses': row[1] or 0,
'test_licenses': row[2] or 0,
'total_device_limit': row[3] or 0
})
except Exception as e:
logging.error(f"Fehler beim Laden der Kundenstatistiken: {str(e)}")
return jsonify({'error': 'Fehler beim Laden der Daten'}), 500
finally:
cur.close()
conn.close()

46
v2_adminpanel/routes/export_routes.py
Datei anzeigen

@@ -34,17 +34,17 @@ def export_licenses():
l.license_type,
l.valid_from,
l.valid_until,
l.active,
l.is_active,
l.device_limit,
l.created_at,
l.is_test,
CASE
WHEN l.valid_until < CURRENT_DATE THEN 'Abgelaufen'
WHEN l.active = false THEN 'Deaktiviert'
WHEN l.is_active = false THEN 'Deaktiviert'
ELSE 'Aktiv'
END as status,
(SELECT COUNT(*) FROM sessions s WHERE s.license_key = l.license_key AND s.active = true) as active_sessions,
(SELECT COUNT(DISTINCT device_id) FROM sessions s WHERE s.license_key = l.license_key) as registered_devices
(SELECT COUNT(*) FROM sessions s WHERE s.license_key = l.license_key AND s.is_active = true) as active_sessions,
(SELECT COUNT(DISTINCT hardware_id) FROM sessions s WHERE s.license_key = l.license_key) as registered_devices
FROM licenses l
LEFT JOIN customers c ON l.customer_id = c.id
ORDER BY l.created_at DESC
@@ -59,17 +59,17 @@ def export_licenses():
l.license_type,
l.valid_from,
l.valid_until,
l.active,
l.is_active,
l.device_limit,
l.created_at,
l.is_test,
CASE
WHEN l.valid_until < CURRENT_DATE THEN 'Abgelaufen'
WHEN l.active = false THEN 'Deaktiviert'
WHEN l.is_active = false THEN 'Deaktiviert'
ELSE 'Aktiv'
END as status,
(SELECT COUNT(*) FROM sessions s WHERE s.license_key = l.license_key AND s.active = true) as active_sessions,
(SELECT COUNT(DISTINCT device_id) FROM sessions s WHERE s.license_key = l.license_key) as registered_devices
(SELECT COUNT(*) FROM sessions s WHERE s.license_key = l.license_key AND s.is_active = true) as active_sessions,
(SELECT COUNT(DISTINCT hardware_id) FROM sessions s WHERE s.license_key = l.license_key) as registered_devices
FROM licenses l
LEFT JOIN customers c ON l.customer_id = c.id
WHERE l.is_test = false
@@ -165,7 +165,7 @@ def export_customers():
c.created_at,
c.is_test,
COUNT(l.id) as license_count,
COUNT(CASE WHEN l.active = true THEN 1 END) as active_licenses,
COUNT(CASE WHEN l.is_active = true THEN 1 END) as active_licenses,
COUNT(CASE WHEN l.valid_until < CURRENT_DATE THEN 1 END) as expired_licenses
FROM customers c
LEFT JOIN licenses l ON c.id = l.customer_id
@@ -221,17 +221,17 @@ def export_sessions():
s.license_key,
l.customer_name,
s.username,
s.device_id,
s.login_time,
s.logout_time,
s.last_activity,
s.active,
s.hardware_id,
s.started_at,
s.ended_at,
s.last_heartbeat,
s.is_active,
l.license_type,
l.is_test
FROM sessions s
LEFT JOIN licenses l ON s.license_key = l.license_key
WHERE s.active = true
ORDER BY s.login_time DESC
WHERE s.is_active = true
ORDER BY s.started_at DESC
"""
cur.execute(query)
else:
@@ -241,17 +241,17 @@ def export_sessions():
s.license_key,
l.customer_name,
s.username,
s.device_id,
s.login_time,
s.logout_time,
s.last_activity,
s.active,
s.hardware_id,
s.started_at,
s.ended_at,
s.last_heartbeat,
s.is_active,
l.license_type,
l.is_test
FROM sessions s
LEFT JOIN licenses l ON s.license_key = l.license_key
WHERE s.login_time >= CURRENT_TIMESTAMP - INTERVAL '%s days'
ORDER BY s.login_time DESC
WHERE s.started_at >= CURRENT_TIMESTAMP - INTERVAL '%s days'
ORDER BY s.started_at DESC
"""
cur.execute(query, (days,))

364
v2_adminpanel/routes/export_routes.py.backup Normale Datei
Datei anzeigen

@@ -0,0 +1,364 @@
import logging
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from flask import Blueprint, request, send_file
import config
from auth.decorators import login_required
from utils.export import create_excel_export, prepare_audit_export_data
from db import get_connection
# Create Blueprint
export_bp = Blueprint('export', __name__, url_prefix='/export')
@export_bp.route("/licenses")
@login_required
def export_licenses():
"""Exportiert Lizenzen als Excel-Datei"""
conn = get_connection()
cur = conn.cursor()
try:
# Filter aus Request
show_test = request.args.get('show_test', 'false') == 'true'
# SQL Query mit optionalem Test-Filter
if show_test:
query = """
SELECT
l.id,
l.license_key,
c.name as customer_name,
c.email as customer_email,
l.license_type,
l.valid_from,
l.valid_until,
l.active,
l.device_limit,
l.created_at,
l.is_test,
CASE
WHEN l.valid_until < CURRENT_DATE THEN 'Abgelaufen'
WHEN l.active = false THEN 'Deaktiviert'
ELSE 'Aktiv'
END as status,
(SELECT COUNT(*) FROM sessions s WHERE s.license_key = l.license_key AND s.active = true) as active_sessions,
(SELECT COUNT(DISTINCT device_id) FROM sessions s WHERE s.license_key = l.license_key) as registered_devices
FROM licenses l
LEFT JOIN customers c ON l.customer_id = c.id
ORDER BY l.created_at DESC
"""
else:
query = """
SELECT
l.id,
l.license_key,
c.name as customer_name,
c.email as customer_email,
l.license_type,
l.valid_from,
l.valid_until,
l.active,
l.device_limit,
l.created_at,
l.is_test,
CASE
WHEN l.valid_until < CURRENT_DATE THEN 'Abgelaufen'
WHEN l.active = false THEN 'Deaktiviert'
ELSE 'Aktiv'
END as status,
(SELECT COUNT(*) FROM sessions s WHERE s.license_key = l.license_key AND s.active = true) as active_sessions,
(SELECT COUNT(DISTINCT device_id) FROM sessions s WHERE s.license_key = l.license_key) as registered_devices
FROM licenses l
LEFT JOIN customers c ON l.customer_id = c.id
WHERE l.is_test = false
ORDER BY l.created_at DESC
"""
cur.execute(query)
# Daten für Export vorbereiten
data = []
columns = ['ID', 'Lizenzschlüssel', 'Kunde', 'E-Mail', 'Typ', 'Gültig von',
'Gültig bis', 'Aktiv', 'Gerätelimit', 'Erstellt am', 'Test-Lizenz',
'Status', 'Aktive Sessions', 'Registrierte Geräte']
for row in cur.fetchall():
data.append(list(row))
# Excel-Datei erstellen
excel_file = create_excel_export(data, columns, 'Lizenzen')
# Datei senden
filename = f"lizenzen_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
return send_file(
excel_file,
mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
as_attachment=True,
download_name=filename
)
except Exception as e:
logging.error(f"Fehler beim Export: {str(e)}")
return "Fehler beim Exportieren der Lizenzen", 500
finally:
cur.close()
conn.close()
@export_bp.route("/audit")
@login_required
def export_audit():
"""Exportiert Audit-Logs als Excel-Datei"""
conn = get_connection()
cur = conn.cursor()
try:
# Filter aus Request
days = int(request.args.get('days', 30))
action_filter = request.args.get('action', '')
entity_type_filter = request.args.get('entity_type', '')
# Daten für Export vorbereiten
data = prepare_audit_export_data(days, action_filter, entity_type_filter)
# Excel-Datei erstellen
columns = ['Zeitstempel', 'Benutzer', 'Aktion', 'Entität', 'Entität ID',
'IP-Adresse', 'Alte Werte', 'Neue Werte', 'Zusatzinfo']
excel_file = create_excel_export(data, columns, 'Audit-Log')
# Datei senden
filename = f"audit_log_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
return send_file(
excel_file,
mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
as_attachment=True,
download_name=filename
)
except Exception as e:
logging.error(f"Fehler beim Export: {str(e)}")
return "Fehler beim Exportieren der Audit-Logs", 500
finally:
cur.close()
conn.close()
@export_bp.route("/customers")
@login_required
def export_customers():
"""Exportiert Kunden als Excel-Datei"""
conn = get_connection()
cur = conn.cursor()
try:
# SQL Query
cur.execute("""
SELECT
c.id,
c.name,
c.email,
c.phone,
c.address,
c.created_at,
c.is_test,
COUNT(l.id) as license_count,
COUNT(CASE WHEN l.active = true THEN 1 END) as active_licenses,
COUNT(CASE WHEN l.valid_until < CURRENT_DATE THEN 1 END) as expired_licenses
FROM customers c
LEFT JOIN licenses l ON c.id = l.customer_id
GROUP BY c.id, c.name, c.email, c.phone, c.address, c.created_at, c.is_test
ORDER BY c.name
""")
# Daten für Export vorbereiten
data = []
columns = ['ID', 'Name', 'E-Mail', 'Telefon', 'Adresse', 'Erstellt am',
'Test-Kunde', 'Anzahl Lizenzen', 'Aktive Lizenzen', 'Abgelaufene Lizenzen']
for row in cur.fetchall():
data.append(list(row))
# Excel-Datei erstellen
excel_file = create_excel_export(data, columns, 'Kunden')
# Datei senden
filename = f"kunden_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
return send_file(
excel_file,
mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
as_attachment=True,
download_name=filename
)
except Exception as e:
logging.error(f"Fehler beim Export: {str(e)}")
return "Fehler beim Exportieren der Kunden", 500
finally:
cur.close()
conn.close()
@export_bp.route("/sessions")
@login_required
def export_sessions():
"""Exportiert Sessions als Excel-Datei"""
conn = get_connection()
cur = conn.cursor()
try:
# Filter aus Request
days = int(request.args.get('days', 7))
active_only = request.args.get('active_only', 'false') == 'true'
# SQL Query
if active_only:
query = """
SELECT
s.id,
s.license_key,
l.customer_name,
s.username,
s.device_id,
s.login_time,
s.logout_time,
s.last_activity,
s.active,
l.license_type,
l.is_test
FROM sessions s
LEFT JOIN licenses l ON s.license_key = l.license_key
WHERE s.active = true
ORDER BY s.login_time DESC
"""
cur.execute(query)
else:
query = """
SELECT
s.id,
s.license_key,
l.customer_name,
s.username,
s.device_id,
s.login_time,
s.logout_time,
s.last_activity,
s.active,
l.license_type,
l.is_test
FROM sessions s
LEFT JOIN licenses l ON s.license_key = l.license_key
WHERE s.login_time >= CURRENT_TIMESTAMP - INTERVAL '%s days'
ORDER BY s.login_time DESC
"""
cur.execute(query, (days,))
# Daten für Export vorbereiten
data = []
columns = ['ID', 'Lizenzschlüssel', 'Kunde', 'Benutzer', 'Geräte-ID',
'Login-Zeit', 'Logout-Zeit', 'Letzte Aktivität', 'Aktiv',
'Lizenztyp', 'Test-Lizenz']
for row in cur.fetchall():
data.append(list(row))
# Excel-Datei erstellen
excel_file = create_excel_export(data, columns, 'Sessions')
# Datei senden
filename = f"sessions_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
return send_file(
excel_file,
mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
as_attachment=True,
download_name=filename
)
except Exception as e:
logging.error(f"Fehler beim Export: {str(e)}")
return "Fehler beim Exportieren der Sessions", 500
finally:
cur.close()
conn.close()
@export_bp.route("/resources")
@login_required
def export_resources():
"""Exportiert Ressourcen als Excel-Datei"""
conn = get_connection()
cur = conn.cursor()
try:
# Filter aus Request
resource_type = request.args.get('type', 'all')
status_filter = request.args.get('status', 'all')
show_test = request.args.get('show_test', 'false') == 'true'
# SQL Query aufbauen
query = """
SELECT
rp.id,
rp.resource_type,
rp.resource_value,
rp.status,
rp.is_test,
l.license_key,
c.name as customer_name,
rp.created_at,
rp.created_by,
rp.status_changed_at,
rp.status_changed_by,
rp.quarantine_reason
FROM resource_pools rp
LEFT JOIN licenses l ON rp.allocated_to_license = l.id
LEFT JOIN customers c ON l.customer_id = c.id
WHERE 1=1
"""
params = []
if resource_type != 'all':
query += " AND rp.resource_type = %s"
params.append(resource_type)
if status_filter != 'all':
query += " AND rp.status = %s"
params.append(status_filter)
if not show_test:
query += " AND rp.is_test = false"
query += " ORDER BY rp.resource_type, rp.resource_value"
cur.execute(query, params)
# Daten für Export vorbereiten
data = []
columns = ['ID', 'Typ', 'Wert', 'Status', 'Test-Ressource', 'Lizenzschlüssel',
'Kunde', 'Erstellt am', 'Erstellt von', 'Status geändert am',
'Status geändert von', 'Quarantäne-Grund']
for row in cur.fetchall():
data.append(list(row))
# Excel-Datei erstellen
excel_file = create_excel_export(data, columns, 'Ressourcen')
# Datei senden
filename = f"ressourcen_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
return send_file(
excel_file,
mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
as_attachment=True,
download_name=filename
)
except Exception as e:
logging.error(f"Fehler beim Export: {str(e)}")
return "Fehler beim Exportieren der Ressourcen", 500
finally:
cur.close()
conn.close()

74
v2_adminpanel/routes/session_routes.py
Datei anzeigen

@@ -21,7 +21,7 @@ def sessions():
cur = conn.cursor()
try:
# Get active sessions with calculated inactive time
# Get is_active sessions with calculated inactive time
cur.execute("""
SELECT s.id, s.session_id, l.license_key, c.name, s.ip_address,
s.user_agent, s.started_at, s.last_heartbeat,
@@ -81,11 +81,11 @@ def session_history():
s.id,
s.license_key,
s.username,
s.device_id,
s.login_time,
s.logout_time,
s.last_activity,
s.active,
s.hardware_id,
s.started_at,
s.ended_at,
s.last_heartbeat,
s.is_active,
l.customer_name,
l.license_type,
l.is_test
@@ -106,22 +106,22 @@ def session_history():
params.append(f'%{username}%')
# Time filter
query += " AND s.login_time >= CURRENT_TIMESTAMP - INTERVAL '%s days'"
query += " AND s.started_at >= CURRENT_TIMESTAMP - INTERVAL '%s days'"
params.append(days)
query += " ORDER BY s.login_time DESC LIMIT 1000"
query += " ORDER BY s.started_at DESC LIMIT 1000"
cur.execute(query, params)
sessions_list = []
for row in cur.fetchall():
session_duration = None
if row[4] and row[5]: # login_time and logout_time
if row[4] and row[5]: # started_at and ended_at
duration = row[5] - row[4]
hours = int(duration.total_seconds() // 3600)
minutes = int((duration.total_seconds() % 3600) // 60)
session_duration = f"{hours}h {minutes}m"
elif row[4] and row[7]: # login_time and active
elif row[4] and row[7]: # started_at and is_active
duration = datetime.now(ZoneInfo("UTC")) - row[4]
hours = int(duration.total_seconds() // 3600)
minutes = int((duration.total_seconds() % 3600) // 60)
@@ -131,11 +131,11 @@ def session_history():
'id': row[0],
'license_key': row[1],
'username': row[2],
'device_id': row[3],
'login_time': row[4],
'logout_time': row[5],
'last_activity': row[6],
'active': row[7],
'hardware_id': row[3],
'started_at': row[4],
'ended_at': row[5],
'last_heartbeat': row[6],
'is_active': row[7],
'customer_name': row[8],
'license_type': row[9],
'is_test': row[10],
@@ -147,7 +147,7 @@ def session_history():
SELECT DISTINCT s.license_key, l.customer_name
FROM sessions s
LEFT JOIN licenses l ON s.license_key = l.license_key
WHERE s.login_time >= CURRENT_TIMESTAMP - INTERVAL '30 days'
WHERE s.started_at >= CURRENT_TIMESTAMP - INTERVAL '30 days'
ORDER BY l.customer_name, s.license_key
""")
@@ -186,9 +186,9 @@ def terminate_session(session_id):
try:
# Get session info
cur.execute("""
SELECT license_key, username, device_id
SELECT license_key, username, hardware_id
FROM sessions
WHERE id = %s AND active = true
WHERE id = %s AND is_active = true
""", (session_id,))
session_info = cur.fetchone()
@@ -199,7 +199,7 @@ def terminate_session(session_id):
# Terminate session
cur.execute("""
UPDATE sessions
SET active = false, logout_time = CURRENT_TIMESTAMP
SET is_active = false, ended_at = CURRENT_TIMESTAMP
WHERE id = %s
""", (session_id,))
@@ -230,10 +230,10 @@ def terminate_all_sessions(license_key):
cur = conn.cursor()
try:
# Count active sessions
# Count is_active sessions
cur.execute("""
SELECT COUNT(*) FROM sessions
WHERE license_key = %s AND active = true
WHERE license_key = %s AND is_active = true
""", (license_key,))
active_count = cur.fetchone()[0]
@@ -245,8 +245,8 @@ def terminate_all_sessions(license_key):
# Terminate all sessions
cur.execute("""
UPDATE sessions
SET active = false, logout_time = CURRENT_TIMESTAMP
WHERE license_key = %s AND active = true
SET is_active = false, ended_at = CURRENT_TIMESTAMP
WHERE license_key = %s AND is_active = true
""", (license_key,))
conn.commit()
@@ -281,8 +281,8 @@ def cleanup_sessions():
# Delete old inactive sessions
cur.execute("""
DELETE FROM sessions
WHERE active = false
AND logout_time < CURRENT_TIMESTAMP - INTERVAL '%s days'
WHERE is_active = false
AND ended_at < CURRENT_TIMESTAMP - INTERVAL '%s days'
RETURNING id
""", (days,))
@@ -322,10 +322,10 @@ def session_statistics():
SELECT
COUNT(DISTINCT s.license_key) as active_licenses,
COUNT(DISTINCT s.username) as unique_users,
COUNT(DISTINCT s.device_id) as unique_devices,
COUNT(DISTINCT s.hardware_id) as unique_devices,
COUNT(*) as total_active_sessions
FROM sessions s
WHERE s.active = true
WHERE s.is_active = true
""")
current_stats = cur.fetchone()
@@ -337,7 +337,7 @@ def session_statistics():
COUNT(*) as session_count
FROM sessions s
JOIN licenses l ON s.license_key = l.license_key
WHERE s.active = true
WHERE s.is_active = true
GROUP BY l.license_type
ORDER BY session_count DESC
""")
@@ -358,7 +358,7 @@ def session_statistics():
l.device_limit
FROM sessions s
JOIN licenses l ON s.license_key = l.license_key
WHERE s.active = true
WHERE s.is_active = true
GROUP BY s.license_key, l.customer_name, l.device_limit
ORDER BY session_count DESC
LIMIT 10
@@ -376,13 +376,13 @@ def session_statistics():
# Session-Verlauf (letzte 7 Tage)
cur.execute("""
SELECT
DATE(login_time) as date,
DATE(started_at) as date,
COUNT(*) as login_count,
COUNT(DISTINCT license_key) as unique_licenses,
COUNT(DISTINCT username) as unique_users
FROM sessions
WHERE login_time >= CURRENT_DATE - INTERVAL '7 days'
GROUP BY DATE(login_time)
WHERE started_at >= CURRENT_DATE - INTERVAL '7 days'
GROUP BY DATE(started_at)
ORDER BY date
""")
@@ -398,12 +398,12 @@ def session_statistics():
# Durchschnittliche Session-Dauer
cur.execute("""
SELECT
AVG(EXTRACT(EPOCH FROM (logout_time - login_time))/3600) as avg_duration_hours
AVG(EXTRACT(EPOCH FROM (ended_at - started_at))/3600) as avg_duration_hours
FROM sessions
WHERE active = false
AND logout_time IS NOT NULL
AND logout_time - login_time < INTERVAL '24 hours'
AND login_time >= CURRENT_DATE - INTERVAL '30 days'
WHERE is_active = false
AND ended_at IS NOT NULL
AND ended_at - started_at < INTERVAL '24 hours'
AND started_at >= CURRENT_DATE - INTERVAL '30 days'
""")
avg_duration = cur.fetchone()[0] or 0

429
v2_adminpanel/routes/session_routes.py.backup Normale Datei
Datei anzeigen

@@ -0,0 +1,429 @@
import logging
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from flask import Blueprint, render_template, request, redirect, session, url_for, flash
import config
from auth.decorators import login_required
from utils.audit import log_audit
from utils.network import get_client_ip
from db import get_connection, get_db_connection, get_db_cursor
from models import get_active_sessions
# Create Blueprint
session_bp = Blueprint('sessions', __name__)
@session_bp.route("/sessions")
@login_required
def sessions():
conn = get_connection()
cur = conn.cursor()
try:
# Get active sessions with calculated inactive time
cur.execute("""
SELECT s.id, s.session_id, l.license_key, c.name, s.ip_address,
s.user_agent, s.started_at, s.last_heartbeat,
EXTRACT(EPOCH FROM (NOW() - s.last_heartbeat))/60 as minutes_inactive
FROM sessions s
JOIN licenses l ON s.license_id = l.id
JOIN customers c ON l.customer_id = c.id
WHERE s.is_active = TRUE
ORDER BY s.last_heartbeat DESC
""")
active_sessions = cur.fetchall()
# Get recent ended sessions
cur.execute("""
SELECT s.id, s.session_id, l.license_key, c.name, s.ip_address,
s.started_at, s.ended_at,
EXTRACT(EPOCH FROM (s.ended_at - s.started_at))/60 as duration_minutes
FROM sessions s
JOIN licenses l ON s.license_id = l.id
JOIN customers c ON l.customer_id = c.id
WHERE s.is_active = FALSE
AND s.ended_at > NOW() - INTERVAL '24 hours'
ORDER BY s.ended_at DESC
LIMIT 50
""")
recent_sessions = cur.fetchall()
return render_template("sessions.html",
active_sessions=active_sessions,
recent_sessions=recent_sessions)
except Exception as e:
logging.error(f"Error loading sessions: {str(e)}")
flash('Fehler beim Laden der Sessions!', 'error')
return redirect(url_for('admin.dashboard'))
finally:
cur.close()
conn.close()
@session_bp.route("/sessions/history")
@login_required
def session_history():
"""Zeigt die Session-Historie"""
conn = get_connection()
cur = conn.cursor()
try:
# Query parameters
license_key = request.args.get('license_key', '')
username = request.args.get('username', '')
days = int(request.args.get('days', 7))
# Base query
query = """
SELECT
s.id,
s.license_key,
s.username,
s.device_id,
s.login_time,
s.logout_time,
s.last_activity,
s.active,
l.customer_name,
l.license_type,
l.is_test
FROM sessions s
LEFT JOIN licenses l ON s.license_key = l.license_key
WHERE 1=1
"""
params = []
# Apply filters
if license_key:
query += " AND s.license_key = %s"
params.append(license_key)
if username:
query += " AND s.username ILIKE %s"
params.append(f'%{username}%')
# Time filter
query += " AND s.login_time >= CURRENT_TIMESTAMP - INTERVAL '%s days'"
params.append(days)
query += " ORDER BY s.login_time DESC LIMIT 1000"
cur.execute(query, params)
sessions_list = []
for row in cur.fetchall():
session_duration = None
if row[4] and row[5]: # login_time and logout_time
duration = row[5] - row[4]
hours = int(duration.total_seconds() // 3600)
minutes = int((duration.total_seconds() % 3600) // 60)
session_duration = f"{hours}h {minutes}m"
elif row[4] and row[7]: # login_time and active
duration = datetime.now(ZoneInfo("UTC")) - row[4]
hours = int(duration.total_seconds() // 3600)
minutes = int((duration.total_seconds() % 3600) // 60)
session_duration = f"{hours}h {minutes}m (aktiv)"
sessions_list.append({
'id': row[0],
'license_key': row[1],
'username': row[2],
'device_id': row[3],
'login_time': row[4],
'logout_time': row[5],
'last_activity': row[6],
'active': row[7],
'customer_name': row[8],
'license_type': row[9],
'is_test': row[10],
'duration': session_duration
})
# Get unique license keys for filter dropdown
cur.execute("""
SELECT DISTINCT s.license_key, l.customer_name
FROM sessions s
LEFT JOIN licenses l ON s.license_key = l.license_key
WHERE s.login_time >= CURRENT_TIMESTAMP - INTERVAL '30 days'
ORDER BY l.customer_name, s.license_key
""")
available_licenses = []
for row in cur.fetchall():
available_licenses.append({
'license_key': row[0],
'customer_name': row[1] or 'Unbekannt'
})
return render_template("session_history.html",
sessions=sessions_list,
available_licenses=available_licenses,
filters={
'license_key': license_key,
'username': username,
'days': days
})
except Exception as e:
logging.error(f"Fehler beim Laden der Session-Historie: {str(e)}")
flash('Fehler beim Laden der Session-Historie!', 'error')
return redirect(url_for('sessions.sessions'))
finally:
cur.close()
conn.close()
@session_bp.route("/session/end/<int:session_id>", methods=["POST"])
@login_required
def terminate_session(session_id):
"""Beendet eine aktive Session"""
conn = get_connection()
cur = conn.cursor()
try:
# Get session info
cur.execute("""
SELECT license_key, username, device_id
FROM sessions
WHERE id = %s AND active = true
""", (session_id,))
session_info = cur.fetchone()
if not session_info:
flash('Session nicht gefunden oder bereits beendet!', 'error')
return redirect(url_for('sessions.sessions'))
# Terminate session
cur.execute("""
UPDATE sessions
SET active = false, logout_time = CURRENT_TIMESTAMP
WHERE id = %s
""", (session_id,))
conn.commit()
# Audit log
log_audit('SESSION_TERMINATE', 'session', session_id,
additional_info=f"Session beendet für {session_info[1]} auf Lizenz {session_info[0]}")
flash('Session erfolgreich beendet!', 'success')
except Exception as e:
conn.rollback()
logging.error(f"Fehler beim Beenden der Session: {str(e)}")
flash('Fehler beim Beenden der Session!', 'error')
finally:
cur.close()
conn.close()
return redirect(url_for('sessions.sessions'))
@session_bp.route("/sessions/terminate-all/<license_key>", methods=["POST"])
@login_required
def terminate_all_sessions(license_key):
"""Beendet alle aktiven Sessions einer Lizenz"""
conn = get_connection()
cur = conn.cursor()
try:
# Count active sessions
cur.execute("""
SELECT COUNT(*) FROM sessions
WHERE license_key = %s AND active = true
""", (license_key,))
active_count = cur.fetchone()[0]
if active_count == 0:
flash('Keine aktiven Sessions gefunden!', 'info')
return redirect(url_for('sessions.sessions'))
# Terminate all sessions
cur.execute("""
UPDATE sessions
SET active = false, logout_time = CURRENT_TIMESTAMP
WHERE license_key = %s AND active = true
""", (license_key,))
conn.commit()
# Audit log
log_audit('SESSION_TERMINATE_ALL', 'license', None,
additional_info=f"{active_count} Sessions beendet für Lizenz {license_key}")
flash(f'{active_count} Sessions erfolgreich beendet!', 'success')
except Exception as e:
conn.rollback()
logging.error(f"Fehler beim Beenden der Sessions: {str(e)}")
flash('Fehler beim Beenden der Sessions!', 'error')
finally:
cur.close()
conn.close()
return redirect(url_for('sessions.sessions'))
@session_bp.route("/sessions/cleanup", methods=["POST"])
@login_required
def cleanup_sessions():
"""Bereinigt alte inaktive Sessions"""
conn = get_connection()
cur = conn.cursor()
try:
days = int(request.form.get('days', 30))
# Delete old inactive sessions
cur.execute("""
DELETE FROM sessions
WHERE active = false
AND logout_time < CURRENT_TIMESTAMP - INTERVAL '%s days'
RETURNING id
""", (days,))
deleted_ids = [row[0] for row in cur.fetchall()]
deleted_count = len(deleted_ids)
conn.commit()
# Audit log
if deleted_count > 0:
log_audit('SESSION_CLEANUP', 'system', None,
additional_info=f"{deleted_count} Sessions älter als {days} Tage gelöscht")
flash(f'{deleted_count} alte Sessions bereinigt!', 'success')
except Exception as e:
conn.rollback()
logging.error(f"Fehler beim Bereinigen der Sessions: {str(e)}")
flash('Fehler beim Bereinigen der Sessions!', 'error')
finally:
cur.close()
conn.close()
return redirect(url_for('sessions.session_history'))
@session_bp.route("/sessions/statistics")
@login_required
def session_statistics():
"""Zeigt Session-Statistiken"""
conn = get_connection()
cur = conn.cursor()
try:
# Aktuelle Statistiken
cur.execute("""
SELECT
COUNT(DISTINCT s.license_key) as active_licenses,
COUNT(DISTINCT s.username) as unique_users,
COUNT(DISTINCT s.device_id) as unique_devices,
COUNT(*) as total_active_sessions
FROM sessions s
WHERE s.active = true
""")
current_stats = cur.fetchone()
# Sessions nach Lizenztyp
cur.execute("""
SELECT
l.license_type,
COUNT(*) as session_count
FROM sessions s
JOIN licenses l ON s.license_key = l.license_key
WHERE s.active = true
GROUP BY l.license_type
ORDER BY session_count DESC
""")
sessions_by_type = []
for row in cur.fetchall():
sessions_by_type.append({
'license_type': row[0],
'count': row[1]
})
# Top 10 Lizenzen nach aktiven Sessions
cur.execute("""
SELECT
s.license_key,
l.customer_name,
COUNT(*) as session_count,
l.device_limit
FROM sessions s
JOIN licenses l ON s.license_key = l.license_key
WHERE s.active = true
GROUP BY s.license_key, l.customer_name, l.device_limit
ORDER BY session_count DESC
LIMIT 10
""")
top_licenses = []
for row in cur.fetchall():
top_licenses.append({
'license_key': row[0],
'customer_name': row[1],
'session_count': row[2],
'device_limit': row[3]
})
# Session-Verlauf (letzte 7 Tage)
cur.execute("""
SELECT
DATE(login_time) as date,
COUNT(*) as login_count,
COUNT(DISTINCT license_key) as unique_licenses,
COUNT(DISTINCT username) as unique_users
FROM sessions
WHERE login_time >= CURRENT_DATE - INTERVAL '7 days'
GROUP BY DATE(login_time)
ORDER BY date
""")
session_history = []
for row in cur.fetchall():
session_history.append({
'date': row[0].strftime('%Y-%m-%d'),
'login_count': row[1],
'unique_licenses': row[2],
'unique_users': row[3]
})
# Durchschnittliche Session-Dauer
cur.execute("""
SELECT
AVG(EXTRACT(EPOCH FROM (logout_time - login_time))/3600) as avg_duration_hours
FROM sessions
WHERE active = false
AND logout_time IS NOT NULL
AND logout_time - login_time < INTERVAL '24 hours'
AND login_time >= CURRENT_DATE - INTERVAL '30 days'
""")
avg_duration = cur.fetchone()[0] or 0
return render_template("session_statistics.html",
current_stats={
'active_licenses': current_stats[0],
'unique_users': current_stats[1],
'unique_devices': current_stats[2],
'total_sessions': current_stats[3]
},
sessions_by_type=sessions_by_type,
top_licenses=top_licenses,
session_history=session_history,
avg_duration=round(avg_duration, 1))
except Exception as e:
logging.error(f"Fehler beim Laden der Session-Statistiken: {str(e)}")
flash('Fehler beim Laden der Statistiken!', 'error')
return redirect(url_for('sessions.sessions'))
finally:
cur.close()
conn.close()