346 lines
13 KiB
Python
346 lines
13 KiB
Python
"""
|
|
Routing-Modul für die neue Frontend V2-Implementierung.
|
|
Stellt die notwendigen Endpunkte bereit, während die Original-API-Endpunkte intakt bleiben.
|
|
"""
|
|
|
|
from flask import Blueprint, render_template, redirect, url_for, request, flash, jsonify, g, session
|
|
import datetime
|
|
import os
|
|
from functools import wraps
|
|
import logging
|
|
import jwt
|
|
|
|
# Blueprint für Frontend V2 erstellen
|
|
frontend_v2 = Blueprint('frontend_v2', __name__, template_folder='frontend_v2/templates')
|
|
|
|
# Logger konfigurieren
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Importiere Funktionen aus dem Hauptmodul
|
|
# Diese werden während der Registrierung des Blueprints übergeben,
|
|
# um zirkuläre Importe zu vermeiden
|
|
get_current_user = None
|
|
get_user_by_id = None
|
|
get_socket_by_id = None
|
|
get_job_by_id = None
|
|
get_all_sockets = None
|
|
get_all_users = None
|
|
get_all_jobs = None
|
|
get_jobs_by_user = None
|
|
login_required = None
|
|
admin_required = None
|
|
delete_session = None
|
|
socket_to_dict = None
|
|
job_to_dict = None
|
|
user_to_dict = None
|
|
|
|
def set_app_functions(app_functions):
|
|
"""
|
|
Setzt die importierten Funktionen aus dem Hauptmodul.
|
|
|
|
Args:
|
|
app_functions: Ein Dictionary mit Funktionen aus dem Hauptmodul
|
|
"""
|
|
global get_current_user, get_user_by_id, get_socket_by_id, get_job_by_id
|
|
global get_all_sockets, get_all_users, get_all_jobs, get_jobs_by_user
|
|
global login_required, admin_required, delete_session
|
|
global socket_to_dict, job_to_dict, user_to_dict
|
|
|
|
get_current_user = app_functions.get('get_current_user')
|
|
get_user_by_id = app_functions.get('get_user_by_id')
|
|
get_socket_by_id = app_functions.get('get_socket_by_id')
|
|
get_job_by_id = app_functions.get('get_job_by_id')
|
|
get_all_sockets = app_functions.get('get_all_sockets')
|
|
get_all_users = app_functions.get('get_all_users')
|
|
get_all_jobs = app_functions.get('get_all_jobs')
|
|
get_jobs_by_user = app_functions.get('get_jobs_by_user')
|
|
login_required = app_functions.get('login_required')
|
|
admin_required = app_functions.get('admin_required')
|
|
delete_session = app_functions.get('delete_session')
|
|
socket_to_dict = app_functions.get('socket_to_dict')
|
|
job_to_dict = app_functions.get('job_to_dict')
|
|
user_to_dict = app_functions.get('user_to_dict')
|
|
|
|
# Wrapper für Login-Erfordernis im Frontend
|
|
def frontend_login_required(f):
|
|
@wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
user = get_current_user()
|
|
if not user:
|
|
flash('Bitte melden Sie sich an, um diese Seite zu besuchen.', 'error')
|
|
return redirect(url_for('frontend_v2.login'))
|
|
|
|
g.current_user = user
|
|
return f(*args, **kwargs)
|
|
|
|
return decorated
|
|
|
|
# Wrapper für Admin-Erfordernis im Frontend
|
|
def frontend_admin_required(f):
|
|
@wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
if not g.get('current_user') or g.current_user['role'] != 'admin':
|
|
flash('Sie benötigen Administrator-Rechte, um diese Seite zu besuchen.', 'error')
|
|
return redirect(url_for('frontend_v2.dashboard'))
|
|
return f(*args, **kwargs)
|
|
|
|
return decorated
|
|
|
|
# Öffentliche Routen
|
|
@frontend_v2.route('/')
|
|
def index():
|
|
current_user = get_current_user()
|
|
if current_user:
|
|
return redirect(url_for('frontend_v2.dashboard'))
|
|
return redirect(url_for('frontend_v2.login'))
|
|
|
|
@frontend_v2.route('/login')
|
|
def login():
|
|
current_user = get_current_user()
|
|
if current_user:
|
|
return redirect(url_for('frontend_v2.dashboard'))
|
|
return render_template('login.html', current_user=None, active_page='login')
|
|
|
|
@frontend_v2.route('/register')
|
|
def register():
|
|
current_user = get_current_user()
|
|
if current_user:
|
|
return redirect(url_for('frontend_v2.dashboard'))
|
|
return render_template('register.html', current_user=None, active_page='register')
|
|
|
|
@frontend_v2.route('/logout')
|
|
def logout():
|
|
session_id = session.get('session_id')
|
|
if session_id:
|
|
delete_session(session_id)
|
|
session.pop('session_id', None)
|
|
|
|
flash('Sie wurden erfolgreich abgemeldet.', 'success')
|
|
return redirect(url_for('frontend_v2.login'))
|
|
|
|
# Geschützte Routen
|
|
@frontend_v2.route('/dashboard')
|
|
@frontend_login_required
|
|
def dashboard():
|
|
current_user = g.current_user
|
|
current_year = datetime.datetime.now().year
|
|
return render_template('dashboard.html',
|
|
current_user=user_to_dict(current_user),
|
|
current_year=current_year,
|
|
active_page='dashboard')
|
|
|
|
@frontend_v2.route('/jobs')
|
|
@frontend_login_required
|
|
def jobs():
|
|
current_user = g.current_user
|
|
|
|
# Admins sehen alle Jobs, normale Benutzer nur ihre eigenen
|
|
if current_user['role'] == 'admin':
|
|
all_jobs = get_all_jobs()
|
|
else:
|
|
all_jobs = get_jobs_by_user(current_user['id'])
|
|
|
|
jobs_list = [job_to_dict(job) for job in all_jobs]
|
|
|
|
# Sortiere Jobs nach Startzeit (neueste zuerst)
|
|
jobs_list.sort(key=lambda x: x['startAt'], reverse=True)
|
|
|
|
# Gruppiere Jobs nach Status (aktiv, abgebrochen, abgeschlossen)
|
|
active_jobs = [job for job in jobs_list if job['remainingMinutes'] > 0 and not job['aborted']]
|
|
completed_jobs = [job for job in jobs_list if job['remainingMinutes'] == 0 and not job['aborted']]
|
|
aborted_jobs = [job for job in jobs_list if job['aborted']]
|
|
|
|
return render_template('jobs.html',
|
|
current_user=user_to_dict(current_user),
|
|
active_jobs=active_jobs,
|
|
completed_jobs=completed_jobs,
|
|
aborted_jobs=aborted_jobs,
|
|
active_page='jobs')
|
|
|
|
@frontend_v2.route('/job/<job_id>')
|
|
@frontend_login_required
|
|
def job_details(job_id):
|
|
current_user = g.current_user
|
|
|
|
job = get_job_by_id(job_id)
|
|
if not job:
|
|
flash('Der angeforderte Job wurde nicht gefunden.', 'error')
|
|
return redirect(url_for('frontend_v2.jobs'))
|
|
|
|
# Benutzer können nur ihre eigenen Jobs sehen, es sei denn, sie sind Admins
|
|
if current_user['role'] != 'admin' and job['user_id'] != current_user['id']:
|
|
flash('Sie haben keine Berechtigung, diesen Job anzusehen.', 'error')
|
|
return redirect(url_for('frontend_v2.jobs'))
|
|
|
|
job_data = job_to_dict(job)
|
|
|
|
# Holen Sie sich die Drucker-Informationen
|
|
printer = get_socket_by_id(job['socket_id'])
|
|
printer_data = socket_to_dict(printer) if printer else None
|
|
|
|
# Benutzerinformationen abrufen
|
|
job_user = get_user_by_id(job['user_id'])
|
|
job_user_data = user_to_dict(job_user) if job_user else None
|
|
|
|
return render_template('job_details.html',
|
|
current_user=user_to_dict(current_user),
|
|
job=job_data,
|
|
printer=printer_data,
|
|
job_user=job_user_data,
|
|
active_page='jobs')
|
|
|
|
@frontend_v2.route('/profile')
|
|
@frontend_login_required
|
|
def profile():
|
|
current_user = g.current_user
|
|
|
|
# Benutzer-Jobs abrufen
|
|
user_jobs = get_jobs_by_user(current_user['id'])
|
|
jobs_list = [job_to_dict(job) for job in user_jobs]
|
|
|
|
# Jobs nach Startzeit sortieren (neueste zuerst)
|
|
jobs_list.sort(key=lambda x: x['startAt'], reverse=True)
|
|
|
|
# Gruppiere Jobs nach Status (aktiv, abgebrochen, abgeschlossen)
|
|
active_jobs = [job for job in jobs_list if job['remainingMinutes'] > 0 and not job['aborted']]
|
|
recent_jobs = jobs_list[:5] # Die 5 neuesten Jobs
|
|
|
|
# Nutzungsstatistiken berechnen
|
|
total_jobs = len(jobs_list)
|
|
total_minutes_used = sum(job['durationInMinutes'] for job in jobs_list if not job['aborted'])
|
|
avg_duration = total_minutes_used // total_jobs if total_jobs > 0 else 0
|
|
|
|
return render_template('profile.html',
|
|
current_user=user_to_dict(current_user),
|
|
active_jobs=active_jobs,
|
|
recent_jobs=recent_jobs,
|
|
total_jobs=total_jobs,
|
|
total_minutes_used=total_minutes_used,
|
|
avg_duration=avg_duration,
|
|
active_page='profile')
|
|
|
|
# Admin-Routen
|
|
@frontend_v2.route('/printers')
|
|
@frontend_login_required
|
|
@frontend_admin_required
|
|
def printers():
|
|
current_user = g.current_user
|
|
|
|
# Alle Drucker abrufen
|
|
all_sockets = get_all_sockets()
|
|
printers_list = [socket_to_dict(socket) for socket in all_sockets]
|
|
|
|
return render_template('printers.html',
|
|
current_user=user_to_dict(current_user),
|
|
printers=printers_list,
|
|
active_page='printers')
|
|
|
|
@frontend_v2.route('/printer/<printer_id>')
|
|
@frontend_login_required
|
|
@frontend_admin_required
|
|
def printer_details(printer_id):
|
|
current_user = g.current_user
|
|
|
|
printer = get_socket_by_id(printer_id)
|
|
if not printer:
|
|
flash('Der angeforderte Drucker wurde nicht gefunden.', 'error')
|
|
return redirect(url_for('frontend_v2.printers'))
|
|
|
|
printer_data = socket_to_dict(printer)
|
|
|
|
return render_template('printer_details.html',
|
|
current_user=user_to_dict(current_user),
|
|
printer=printer_data,
|
|
active_page='printers')
|
|
|
|
@frontend_v2.route('/users')
|
|
@frontend_login_required
|
|
@frontend_admin_required
|
|
def users():
|
|
current_user = g.current_user
|
|
|
|
# Alle Benutzer abrufen
|
|
all_users = get_all_users()
|
|
users_list = [user_to_dict(user) for user in all_users]
|
|
|
|
return render_template('users.html',
|
|
current_user=user_to_dict(current_user),
|
|
users=users_list,
|
|
active_page='users')
|
|
|
|
@frontend_v2.route('/user/<user_id>')
|
|
@frontend_login_required
|
|
@frontend_admin_required
|
|
def user_details(user_id):
|
|
current_user = g.current_user
|
|
|
|
user = get_user_by_id(user_id)
|
|
if not user:
|
|
flash('Der angeforderte Benutzer wurde nicht gefunden.', 'error')
|
|
return redirect(url_for('frontend_v2.users'))
|
|
|
|
user_data = user_to_dict(user)
|
|
|
|
# Benutzer-Jobs abrufen
|
|
user_jobs = get_jobs_by_user(user_id)
|
|
jobs_list = [job_to_dict(job) for job in user_jobs]
|
|
|
|
# Jobs nach Startzeit sortieren (neueste zuerst)
|
|
jobs_list.sort(key=lambda x: x['startAt'], reverse=True)
|
|
|
|
# Gruppiere Jobs nach Status
|
|
active_jobs = [job for job in jobs_list if job['remainingMinutes'] > 0 and not job['aborted']]
|
|
completed_jobs = [job for job in jobs_list if job['remainingMinutes'] == 0 and not job['aborted']]
|
|
aborted_jobs = [job for job in jobs_list if job['aborted']]
|
|
|
|
# Nutzungsstatistiken berechnen
|
|
total_jobs = len(jobs_list)
|
|
total_minutes_used = sum(job['durationInMinutes'] for job in jobs_list if not job['aborted'])
|
|
avg_duration = total_minutes_used // total_jobs if total_jobs > 0 else 0
|
|
|
|
return render_template('user_details.html',
|
|
current_user=user_to_dict(current_user),
|
|
user=user_data,
|
|
active_jobs=active_jobs,
|
|
completed_jobs=completed_jobs,
|
|
aborted_jobs=aborted_jobs,
|
|
total_jobs=total_jobs,
|
|
total_minutes_used=total_minutes_used,
|
|
avg_duration=avg_duration,
|
|
active_page='users')
|
|
|
|
@frontend_v2.route('/statistics')
|
|
@frontend_login_required
|
|
@frontend_admin_required
|
|
def statistics():
|
|
current_user = g.current_user
|
|
|
|
return render_template('statistics.html',
|
|
current_user=user_to_dict(current_user),
|
|
active_page='statistics')
|
|
|
|
# Fehlerbehandlung
|
|
@frontend_v2.errorhandler(404)
|
|
def page_not_found(e):
|
|
current_user = get_current_user()
|
|
return render_template('error.html',
|
|
current_user=user_to_dict(current_user) if current_user else None,
|
|
error_code=404,
|
|
error_message='Die angeforderte Seite wurde nicht gefunden.'), 404
|
|
|
|
@frontend_v2.errorhandler(403)
|
|
def forbidden(e):
|
|
current_user = get_current_user()
|
|
return render_template('error.html',
|
|
current_user=user_to_dict(current_user) if current_user else None,
|
|
error_code=403,
|
|
error_message='Sie haben keine Berechtigung, auf diese Seite zuzugreifen.'), 403
|
|
|
|
@frontend_v2.errorhandler(500)
|
|
def server_error(e):
|
|
current_user = get_current_user()
|
|
logger.error(f'Serverfehler: {e}')
|
|
return render_template('error.html',
|
|
current_user=user_to_dict(current_user) if current_user else None,
|
|
error_code=500,
|
|
error_message='Ein interner Serverfehler ist aufgetreten.'), 500 |