""" Diese Modul ist für die Logik der Anwendung zuständig. Funktionen: login(): Einloggen von User index(): Rendern der Hauptseite users(): Aufbereitung von der Userdaten von DB in Json logout(): Ausloggen der Users. Löscht alle Sessions insert_form(): Rendert das Formular für das Einfügen neuer Benutzerdaten. insert_data(): Verarbeitet POST-Anfragen zum Einfügen neuer Benutzerdaten in die Datenbank. delete(row_id): Löscht Fahrer indem er bei der Spalte "deleted" auf 1 setzt edit_form(row_id): Rendert das Formular für editieren bestimmter Benutzerdaten. edit_data(): Verarbeitet POST-Anfragen zum editieren bestimmter Benutzerdaten in die Datenbank. Autor: Janakan Selvathas """ from datetime import timedelta import traceback import os import requests import asyncio from flask import Flask, render_template, request, session, flash, redirect, jsonify, url_for, send_file, Response, current_app from flask_session import Session import time as t import pandas as pd import bcrypt import time import ast from datetime import datetime from zoneinfo import ZoneInfo from setup import onduty_it, lara_bot_id, translation,onduty_recruitment from db import init_db from standalone import ( send_msg_telegram, generate_uid, create_google_folder, create_tookan_acc, get_country_dialing_codes, create_planday_acc, get_fk_data_from_dict, get_last_month_number, compare_detailed_changes_users, get_current_year, extract_children_data, update_childrens, inform_jan_bc_user_new_planday, convert_to_log_structure, convert_raw_rows_to_structure,compare_detailed_changes_administration, process_fk_data_for_list, add_user_id, update_partner, replace_empty_with_none, get_fk_data_from_dict, get_last_month_number,remove_files_by_file_id, extract_google_drive_file_id, get_current_year, upload_files_to_google_drive, list_files_in_folder, save_files_tmp, remove_files, insert_users_doc, rename_file, convert_files, append_date_suffix_to_files, check_document_requirements, add_current_files, str_to_list, parse_file_data, put_driver_in_11KA_in_planday, get_file_by_id_in_folder, create_list_with_specific, update_tookan_user_status, deactivate_planday_acc, reactivate_planday_acc, check_if_docs_are_valid_and_update_planday_name,move_planday_acc_DNF,address_changed, apply_planday_address_update, reset_planday_name, _normalize, value_changed, update_planday_employee, get_user_planday_employeeGroups, move_planday_acc_from_DNF_to_employeeGroups ) from models import ( get_login_user, get_users_view_data, get_users_details_data, get_data_from_fk_table, get_insert_description, check_duplicate, insert_data_db,get_users_external_view_data, get_inactive_users_view_data, get_user_data_for_edit,get_user_data_for_edit_as_string, update_user, get_users_tech_data, log, log_view, get_administration_not_comp_data, get_needed_info, get_insert_administration_description, get_all_data_from_fk_table, update_user_administration, get_userid_username,insert_data_db_administration, get_administration_row_data_for_edit,get_administration_comp_data, get_administration_data_for_edit_as_string, log_administration_view, get_expired_approval_employee, get_users_paycheck_data, get_users_addons_details_data, get_users_paycheck_data_uber,insert_childerns_data,get_childrens_data, get_partner_data, insert_partner_db, get_childrens_details_data, get_users_addons_details_data, get_users_paycheck_data_uber, get_needed_docs_from, get_required_docs, get_docs_formate_from_administration_table, insert_external_acc_status, update_external_acc_status, get_external_account_status, get_administration_data_for_edit_with_full_address,insert_last_planday_department_users ,select_last_planday_department_users ) app = Flask(__name__) app.secret_key = "asndijnsau" app.config['SESSION_COOKIE_SECURE'] = True #app.config['SESSION_COOKIE_HTTPONLY'] = True app.config['REMEMBER_COOKIE_DURATION'] = timedelta(days=6) # Adjust based on your requirements app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(minutes=120) init_db(app) LOGIN_EXEMPT_ENDPOINTS = { "login", "static", } LOGIN_EXEMPT_PATH_PREFIXES = ( "/static/", "/favicon.ico", ) LOCK_SECONDS = 60 MAX_ATTEMPTS = 3 def format_iso_date_numeric(iso_date): try: # Parse ISO date and convert to UTC utc_time = datetime.fromisoformat(iso_date.replace("Z", "+00:00")) # Convert to Zürich timezone zurich_time = utc_time.astimezone(ZoneInfo("Europe/Zurich")) # Format the date as dd-mm-yyyy HH:MM:SS return zurich_time.strftime("%d-%m-%Y %H:%M:%S") except Exception as e: return iso_date # Return the original string if parsing fails app.jinja_env.filters['format_iso_date'] = format_iso_date_numeric @app.after_request def apply_security_headers(response): csp = { "default-src": ["'self'"], "style-src": [ "'self'", "https://cdn.jsdelivr.net", "https://cdnjs.cloudflare.com", "https://fonts.googleapis.com", "https://cdn.jsdelivr.net/npm/select2/dist/css", "https://www.gstatic.com", # Allow Google Translate stylesheets "'unsafe-inline'" # Consider removing this in production ], "script-src": [ "'self'", "https://code.jquery.com", "https://stackpath.bootstrapcdn.com/bootstrap/4.5.2/js/bootstrap.min.js", "https://cdn.jsdelivr.net", "https://cdnjs.cloudflare.com", "https://cdn.jsdelivr.net/npm/popper.js", "https://cdn.jsdelivr.net/npm/bootstrap", "https://cdn.jsdelivr.net/npm/select2/dist/js", "https://translate.google.com", "https://translate.googleapis.com", # Allow Google Translate scripts "https://translate-pa.googleapis.com", # Allow additional Google Translate API "'unsafe-inline'" # Consider removing this in production ], "img-src": [ "'self'", "data:", # Allow data URIs for images "https://fonts.gstatic.com", # Allow Google fonts images "https://www.google.com", # Allow Google images "https://www.gstatic.com", # Allow Google static images "https://translate.googleapis.com" # Allow Google Translate images ], "font-src": [ "'self'", "https://cdnjs.cloudflare.com", "https://fonts.gstatic.com" ], "connect-src": [ "'self'", "https://translate.googleapis.com", # Allow connections to Google Translate API "https://translate-pa.googleapis.com" # Allow connections to Google Translate API ] } # Convert CSP dictionary into a header-compatible format csp_directives = [] for directive, sources in csp.items(): csp_directives.append(f"{directive} {' '.join(sources)}") # Set CSP in the response headers response.headers["Content-Security-Policy"] = "; ".join(csp_directives) # Additional security headers (recommended) response.headers["X-Frame-Options"] = "SAMEORIGIN" response.headers["X-Content-Type-Options"] = "nosniff" response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" response.headers["Permissions-Policy"] = ( "geolocation=(), microphone=(), camera=(), payment=(), usb=()" ) ep = request.endpoint or "" if ep != "static": response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0, private" response.headers["Pragma"] = "no-cache" response.headers["Expires"] = "0" return response @app.before_request def ensure_logged_in(): # Exempts ep = request.endpoint or "" if ep in LOGIN_EXEMPT_ENDPOINTS: return path = request.path or "" if any(path.startswith(p) for p in LOGIN_EXEMPT_PATH_PREFIXES): return # Schon eingeloggt? OK if session.get("valid") is True: return # POSTs auf /login erlauben (falls Form submit) if path == url_for("login"): return # Nicht eingeloggt → redirect zum Login (mit next-Parameter bei GET/HEAD) if request.method in ("GET", "HEAD"): return redirect(url_for("login", next=request.url)) return redirect(url_for("login")) @app.route('/login', methods=['GET', 'POST']) def login(): # Session-Zähler initialisieren session.setdefault('failed_attempts', 0) session.setdefault('login_lock_until', 0) now = int(time.time()) # Falls gesperrt: Restzeit berechnen und sofort rendern if request.method == 'GET': remaining = max(0, session.get('login_lock_until', 0) - now) return render_template('login.html', lock_seconds=remaining) # POST: Daten normalisieren login_email = (request.form.get('login', '') or '').strip().lower() login_password = (request.form.get('password', '') or '').strip() if not login_email or not login_password: flash('Bitte alle Felder ausfüllen!', 'warning') remaining = max(0, session.get('login_lock_until', 0) - now) return render_template('login.html', lock_seconds=remaining) # Sperre aktiv? if now < session.get('login_lock_until', 0): remaining = session['login_lock_until'] - now flash(f"Zu viele Versuche. Bitte warten Sie {remaining} Sekunden.", 'danger') return render_template('login.html', lock_seconds=remaining) try: result = get_login_user(login_email) if result and bcrypt.checkpw(login_password.encode('utf-8'), result[1].encode('utf-8')): # Erfolg -> Session neu aufsetzen session.clear() session['username'] = result[0] session['admin'] = result[2] session['valid'] = True return redirect('/') else: # Fehlversuch zählen session['failed_attempts'] = session.get('failed_attempts', 0) + 1 if session['failed_attempts'] >= MAX_ATTEMPTS: # Lock setzen und Zähler zurück session['login_lock_until'] = now + LOCK_SECONDS session['failed_attempts'] = 0 flash(f"Zu viele Versuche. Anmelden für {LOCK_SECONDS} Sekunden gesperrt.", 'danger') return render_template('login.html', lock_seconds=LOCK_SECONDS) flash('Falscher Benutzername oder Passwort!', 'warning') return render_template('login.html', lock_seconds=0) except Exception as e: asyncio.run(send_msg_telegram(lara_bot_id, f"Login Fail: {e}", onduty_it)) flash('Etwas ist schief gelaufen. Bitte IT Team konaktieren', 'danger') remaining = max(0, session.get('login_lock_until', 0) - now) return render_template('login.html', lock_seconds=remaining) @app.route('/', methods = ['POST', 'GET']) def index(): """ Serves as the home page of the application. This endpoint checks if a user session is valid. If valid, it displays the home page which may include a table view or other user-specific information. If the session is not valid, it redirects the user to the login page. Returns:TranslateElementher display the home page if the user is logged in, or redirect to the login page if not. """ if 'valid' in session and session['valid'] is True: status = request.form.get('status') table = 1 if status == '1': table = 1 if status == '0': table = 0 return render_template('index.html',table = table, username = session['username'], site_name = "usersView") return render_template('login.html') @app.route('/tech_view', methods = ['POST', 'GET']) def tech_view(): """ Serves as the technical view page of the application. This route checks if a user session is valid. If the session is valid, it displays the technical view page, which may include a table view or other technical-specific information. If the session is not valid, it redirects the user to the login page. Returns: - If the session is valid: Renders 'techview.html' with the username passed to the template to provide a personalized user experience. - If the session is not valid: Renders 'login.html' to prompt the user for login credentials. Example Usage: Accessing the '/tech_view' URL will either display the technical view page if the user is logged in, or redirect to the login page if not. """ if 'valid' in session and session['valid'] is True: return render_template('techview.html', username = session['username'], site_name = "TechView") return render_template('login.html') @app.route('/administration_view', methods = ['POST', 'GET']) def administration_view(): """ Serves as the technical view page of the application. This route checks if a user session is valid. If the session is valid, it displays the technical view page, which may include a table view or other technical-specific information. If the session is not valid, it redirects the user to the login page. Returns: - If the session is valid: Renders 'techview.html' with the username passed to the template to provide a personalized user experience. - If the session is not valid: Renders 'login.html' to prompt the user for login credentials. Example Usage: Accessing the '/tech_view' URL will either display the technical view page if the user is logged in, or redirect to the login page if not. """ if 'valid' in session and session['valid'] is True: status = request.form.get('status') table = 1 if status == '1': table = 1 if status == '0': table = 0 return render_template('administrationview.html', table=table ,username = session['username'], site_name = "administrationView") return render_template('login.html') @app.route('/external_view', methods = ['POST', 'GET']) def external_view(): """ Serves as the technical view page of the application. This route checks if a user session is valid. If the session is valid, it displays the technical view page, which may include a table view or other technical-specific information. If the session is not valid, it redirects the user to the login page. Returns: - If the session is valid: Renders 'techview.html' with the username passed to the template to provide a personalized user experience. - If the session is not valid: Renders 'login.html' to prompt the user for login credentials. Example Usage: Accessing the '/tech_view' URL will either display the technical view page if the user is logged in, or redirect to the login page if not. """ if 'valid' in session and session['valid'] is True: return render_template('externalview.html', username = session['username'], site_name = "ExternalView") return render_template('login.html') @app.route('/approval_expired_view', methods = ['POST', 'GET']) def approval_expired_view(): """ Serves as the technical view page of the application. This route checks if a user session is valid. If the session is valid, it displays the technical view page, which may include a table view or other technical-specific information. If the session is not valid, it redirects the user to the login page. Returns: - If the session is valid: Renders 'techview.html' with the username passed to the template to provide a personalized user experience. - If the session is not valid: Renders 'login.html' to prompt the user for login credentials. Example Usage: Accessing the '/tech_view' URL will either display the technical view page if the user is logged in, or redirect to the login page if not. """ if 'valid' in session and session['valid'] is True: return render_template('expiredapprovalview.html', username = session['username'], site_name = "approvalExpiredView") return render_template('login.html') @app.route('/needed_data_view', methods = ['POST', 'GET']) def needed_data_view(): """ Serves as the technical view page of the application. This route checks if a user session is valid. If the session is valid, it displays the technical view page, which may include a table view or other technical-specific information. If the session is not valid, it redirects the user to the login page. Returns: - If the session is valid: Renders 'techview.html' with the username passed to the template to provide a personalized user experience. - If the session is not valid: Renders 'login.html' to prompt the user for login credentials. Example Usage: Accessing the '/tech_view' URL will either display the technical view page if the user is logged in, or redirect to the login page if not. """ if 'valid' in session and session['valid'] is True: return render_template('neededinfoview.html', username = session['username'], site_name = "NeededView") return render_template('login.html') @app.route('/active_users_data', methods = ['POST', 'GET']) def active_users(): """ Handles requests to fetch and return user data as JSON. This endpoint supports both POST and GET methods. Processes: - Validates if the user session is valid. If not, prompts the user to log in. - If valid, attempts to retrieve user data from the database. - Returns the data as JSON with column names and corresponding data rows. Returns: - JSON object containing user data and column names if the session is valid and data retrieval is successful. - Redirects to the 'login.html' template with a warning message if the session is not valid. - Returns to the 'index.html' template with an error message if there is a problem fetching data. Raises: - Exception: Logs detailed exceptions in the system logs and notifies the IT team via Telegram if there is a failure in data retrieval. The user is shown a generic database error message. Example Usage: Accessing '/users_data' with a valid session will return JSON formatted user data. Without a valid session, the user will be redirected to the login page. Notes: - This function uses `get_users_view_data()` to fetch user data, which should handle database interactions. - Errors during data retrieval are handled gracefully, providing feedback to the user and notifying support staff. """ if 'valid' in session and session['valid'] is True: try: column_names, rows = get_users_view_data() return jsonify({'columns': column_names, 'data': rows}) except Exception: full_traceback = traceback.format_exc() asyncio.run(send_msg_telegram(lara_bot_id,f"Database query failed: {full_traceback}", onduty_it)) flash('Datenbankfehler!', 'danger') return render_template('index.html') flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/users_needed_data', methods=['POST', 'GET']) def users_needed_data(): """ Endpoint to retrieve specific user data needed for display or processing. Checks for a valid session: - If the session is valid, it fetches required information from the database. - If successful, it returns a JSON response containing column names and data rows. - If an error occurs during the data retrieval process, it handles the exception by notifying IT support and displaying an error message to the user. Returns: - JSON response with 'columns' (list of column names) and 'data' (list of rows) if successful. - Renders 'index.html' with an error message if there is an exception. - Redirects to 'login.html' with a warning message if the user is not logged in. Raises: - Exception: If a database query error occurs, logs the full traceback, notifies IT via Telegram, and shows an error message to the user. """ # Check if the user session is valid if 'valid' in session and session['valid'] is True: try: # Fetch column names and rows of data needed from the database column_names, rows = get_needed_info() # Return data as JSON for easy handling by the front end return jsonify({'columns': column_names, 'data': rows}) except Exception as e: # Capture full traceback in case of an error full_traceback = traceback.format_exc() # Notify IT support about the database query failure asynchronously asyncio.run(send_msg_telegram( lara_bot_id, f"Database query failed: {full_traceback}", onduty_it )) # Flash a database error message to the user and render the main page flash('Datenbankfehler!', 'danger') return render_template('index.html') # If session is not valid, prompt the user to log in and redirect to login page flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/external_users_data', methods=['POST', 'GET']) def external_users_data(): """ Endpoint to retrieve external view data for users. This function is primarily used to provide a specific view of user data that may be displayed to external systems or users. - Checks for a valid session: - If the session is valid, it fetches the external view data from the database. - If successful, it returns the data in JSON format, including column names and data rows. - If an error occurs during the data retrieval process, it logs the error, notifies IT support, and flashes an error message. Returns: - JSON response with 'columns' (list of column names) and 'data' (list of rows) if successful. - Renders 'index.html' with a database error message if an exception occurs. - Redirects to 'login.html' with a warning message if the user is not logged in. Raises: - Exception: If an error occurs during the database query, it logs the full traceback, sends a notification to IT support, and flashes an error message to the user. """ # Check if the user session is valid if 'valid' in session and session['valid'] is True: try: # Fetch column names and rows of external view data from the database column_names, rows = get_users_external_view_data() # Return the data as JSON for easy processing by external or frontend systems return jsonify({'columns': column_names, 'data': rows}) except Exception as e: # Capture full traceback for debugging in case of an error full_traceback = traceback.format_exc() # Notify IT support asynchronously about the database query failure asyncio.run(send_msg_telegram( lara_bot_id, f"Database query failed: {full_traceback}", onduty_it )) # Flash a database error message to the user and render the main page flash('Datenbankfehler!', 'danger') return render_template('index.html') # If session is not valid, prompt the user to log in and redirect to login page flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/docs_from_data', methods=['POST', 'GET']) def docs_from_data(): """ Endpoint to retrieve required documents data. Checks for a valid session: - If the session is valid, it attempts to fetch document data from the database. - If successful, returns the data directly to the client. - If an error occurs, handles the exception by logging it, notifying IT support, and flashing an error message to the user. Returns: - Data directly from `get_needed_docs_from()` if successful (could be JSON or other format based on function). - Renders 'index.html' with a database error message if an exception occurs. - Redirects to 'login.html' with a warning message if the user is not logged in. Raises: - Exception: If there’s an error during the data query, it logs the full traceback, notifies IT support via Telegram, and shows an error message to the user. """ # Check if the user session is valid if 'valid' in session and session['valid'] is True: try: # Retrieve necessary document data from the database data = get_needed_docs_from() # Return the data directly; format will depend on the structure of `get_needed_docs_from()` return data except Exception as e: # Capture full traceback for debugging purposes in case of an error full_traceback = traceback.format_exc() # Notify IT support asynchronously about the database query failure asyncio.run(send_msg_telegram( lara_bot_id, f"Database query failed: {full_traceback}", onduty_it )) # Flash a database error message to the user and render the main page flash('Datenbankfehler!', 'danger') return render_template('index.html') # If session is not valid, prompt the user to log in and redirect to login page flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/approval_expired_data', methods=['POST', 'GET']) def approval_expired_data(): """ Endpoint to retrieve data on employees with expired approvals. Checks for a valid session: - If the session is valid, it attempts to retrieve expired approval data from the database. - If successful, it returns the data as a JSON response, including column names and data rows. - If an error occurs during the data retrieval process, it logs the error, notifies IT support, and displays an error message. Returns: - JSON response with 'columns' (list of column names) and 'data' (list of rows) if data retrieval is successful. - Renders 'index.html' with a database error message if an exception occurs. - Redirects to 'login.html' with a warning message if the user is not logged in. Raises: - Exception: If there is an error during the database query, it logs the full traceback, notifies IT via Telegram, and shows an error message to the user. """ # Check if the user session is valid if 'valid' in session and session['valid'] is True: try: # Fetch column names and rows of employees with expired approvals from the database column_names, rows = get_expired_approval_employee() # Return the data as JSON to facilitate frontend processing return jsonify({'columns': column_names, 'data': rows}) except Exception as e: # Capture the full traceback to understand and debug the issue full_traceback = traceback.format_exc() # Notify IT support asynchronously about the database query failure asyncio.run(send_msg_telegram( lara_bot_id, f"Database query failed: {full_traceback}", onduty_it )) # Flash a database error message to the user and render the main page flash('Datenbankfehler!', 'danger') return render_template('index.html') # If session is not valid, prompt the user to log in and redirect to the login page flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/inactive_users_data', methods = ['POST', 'GET']) def inactive_users(): """ Handles requests to fetch and return user data as JSON. This endpoint supports both POST and GET methods. Processes: - Validates if the user session is valid. If not, prompts the user to log in. - If valid, attempts to retrieve user data from the database. - Returns the data as JSON with column names and corresponding data rows. Returns: - JSON object containing user data and column names if the session is valid and data retrieval is successful. - Redirects to the 'login.html' template with a warning message if the session is not valid. - Returns to the 'index.html' template with an error message if there is a problem fetching data. Raises: - Exception: Logs detailed exceptions in the system logs and notifies the IT team via Telegram if there is a failure in data retrieval. The user is shown a generic database error message. Example Usage: Accessing '/users_data' with a valid session will return JSON formatted user data. Without a valid session, the user will be redirected to the login page. Notes: - This function uses `get_users_view_data()` to fetch user data, which should handle database interactions. - Errors during data retrieval are handled gracefully, providing feedback to the user and notifying support staff. """ if 'valid' in session and session['valid'] is True: try: column_names, rows = get_inactive_users_view_data() return jsonify({'columns': column_names, 'data': rows}) except Exception: full_traceback = traceback.format_exc() asyncio.run(send_msg_telegram(lara_bot_id,f"Database query failed: {full_traceback}", onduty_it)) flash('Datenbankfehler!', 'danger') return render_template('index.html') flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/users_tech_data', methods = ['POST', 'GET']) def tech_users(): """ Handles requests to fetch and return technical user data as JSON. This endpoint supports both POST and GET methods. Processes: - Validates if the user session is valid. If not, the user is prompted to log in. - If valid, attempts to retrieve technical user data from the database. - Returns the data as a JSON object with column names and corresponding data rows. Returns: - JSON object: Contains technical user data and column names if the session is valid and data retrieval is successful. - Rendered 'login.html' template with a warning message if the session is not valid. - Rendered 'index.html' template with a danger message if there is a failure in fetching data from the database. Raises: - Exception: Logs detailed exceptions and optionally sends notifications if there is a failure in data retrieval. The user is shown a generic database error message. Example Usage: Accessing '/users_tech_data' with a valid session will return JSON formatted technical user data. Without a valid session, the user will be redirected to the login page. Notes: - This function uses `get_users_tech_data()` to fetch data, which should manage database interactions. - Errors during data retrieval are logged and the user is informed through a user-friendly message. - Telecommunication alerts are commented out but can be enabled for IT notification in case of errors. """ if 'valid' in session and session['valid'] is True: try: column_names, rows = get_users_tech_data() return jsonify({'columns': column_names, 'data': rows}) except: full_traceback = traceback.format_exc() asyncio.run(send_msg_telegram(lara_bot_id,f"Database query failed: {full_traceback} | Talentsphere",onduty_it)) flash('Datenbankfehler!', 'danger') return render_template('index.html') flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/administration_not_comp_data', methods=['POST', 'GET']) def administration_not_comp_data(): """ Endpoint to retrieve data on administrative tasks that are not complete. Checks for a valid session: - If the session is valid, it attempts to retrieve incomplete administrative task data from the database. - If successful, it returns the data in JSON format, including column names and data rows. - If an error occurs during the data retrieval, it logs the error, notifies IT support, and displays an error message to the user. Returns: - JSON response with 'columns' (list of column names) and 'data' (list of rows) if data retrieval is successful. - Renders 'index.html' with a database error message if an exception occurs. - Redirects to 'login.html' with a warning message if the user is not logged in. Raises: - Exception: If there is an error during the database query, logs the full traceback, notifies IT support, and shows an error message to the user. """ # Check if the user session is valid if 'valid' in session and session['valid'] is True: try: # Fetch column names and rows for incomplete administrative tasks from the database column_names, rows = get_administration_not_comp_data() # Return the data as JSON for frontend processing return jsonify({'columns': column_names, 'data': rows}) except Exception: # Capture full traceback for debugging in case of an error full_traceback = traceback.format_exc() # Notify IT support asynchronously about the database query failure asyncio.run(send_msg_telegram( lara_bot_id, f"Database query failed: {full_traceback}", onduty_it )) # Flash a database error message to the user and render the main page flash('Datenbankfehler!', 'danger') return render_template('index.html') # If session is not valid, prompt the user to log in and redirect to login page flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/administration_comp_data', methods=['POST', 'GET']) def administration_comp_data(): """ Endpoint to retrieve data on completed administrative tasks. Checks for a valid session: - If the session is valid, it attempts to retrieve completed administrative task data from the database. - If successful, it returns the data in JSON format, including column names and data rows. - If an error occurs during data retrieval, it logs the error, notifies IT support, and displays an error message to the user. Returns: - JSON response with 'columns' (list of column names) and 'data' (list of rows) if data retrieval is successful. - Renders 'index.html' with a database error message if an exception occurs. - Redirects to 'login.html' with a warning message if the user is not logged in. Raises: - Exception: If there is an error during the database query, logs the full traceback, notifies IT support, and shows an error message to the user. """ # Check if the user session is valid if 'valid' in session and session['valid'] is True: try: # Fetch column names and rows for completed administrative tasks from the database column_names, rows = get_administration_comp_data() # Return the data as JSON for easy processing by the frontend return jsonify({'columns': column_names, 'data': rows}) except Exception: # Capture full traceback to help with debugging in case of an error full_traceback = traceback.format_exc() # Notify IT support asynchronously about the database query failure asyncio.run(send_msg_telegram( lara_bot_id, f"Database query failed: {full_traceback}", onduty_it )) # Flash a database error message to the user and render the main page flash('Datenbankfehler!', 'danger') return render_template('index.html') # If session is not valid, prompt the user to log in and redirect to login page flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/details/', methods=['GET', 'POST']) def get_details(user_id): """ Handles requests to display detailed information about a user based on the specified user ID. The function retrieves details about the user, including basic information, children’s details, and partner data. The information is then rendered on the 'details.html' template. The function checks if the user session is valid. If valid, it: - Retrieves translations for column headers. - Fetches user details, children’s data, and partner data from the database using helper functions. - Passes the retrieved data to the 'details.html' template for display. Args: user_id (str): The unique identifier of the user for whom details are being fetched. Returns: - Renders 'details.html' with user details, including basic data, children's data, partner data, and translations. - Redirects to 'login.html' with a warning message if the session is not valid. - If an error occurs, it logs the error, optionally sends a notification to IT via Telegram, and renders 'index.html' with an error message. Raises: - Exception: Captures any errors during data retrieval, logs the full traceback, sends a notification via Telegram, and flashes an error message to the user. Example Usage: - Accessing '/details/12345' with a valid session will render 'details.html' showing detailed user information. Notes: - This endpoint requires an active session to access user details. - Error handling includes detailed logging for diagnostics, Telegram notifications for IT, and user-friendly error messages. """ # Check if the session is valid if 'valid' in session and session['valid'] is True: try: # Prepare translation mappings for display labels translations = {item['db']: item['de'] for item in translation} # Fetch basic user details, children's data, and partner data for the given user ID data = get_users_details_data(user_id) childrens_data = get_childrens_details_data(user_id) partner_data = get_partner_data(user_id) files_data = list_files_in_folder(user_id) #equipment_data = get_equipment_data(user_id) administration_docs_formate = get_docs_formate_from_administration_table() list_administration_docs_formate = create_list_with_specific(administration_docs_formate, 'name') # Extract user type from the user data for template display typ = data[0]['typ'] # Render 'details.html' with all retrieved data return render_template( 'details.html', data=data, site_name="employeeDetails", user_id=user_id, typ=typ, childrens_data=childrens_data, partner_data=partner_data, translations=translations, files_data=files_data, list_administration_docs_formate = list_administration_docs_formate, #equipment_data = equipment_data[0], username=session['username'] ) except Exception as e: # Capture the full traceback in case of an error full_traceback = traceback.format_exc() # Notify IT support asynchronously via Telegram with detailed error information asyncio.run(send_msg_telegram( lara_bot_id, f"Database query failed: {full_traceback} | Talentsphere", onduty_it )) # Flash an error message and render the main page in case of a database error flash('Datenbankfehler! Bitte IT Team melden.', 'danger') return render_template('index.html', username=session['username']) # If session is not valid, prompt the user to log in and redirect to login page flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/paycheck//', methods=['GET', 'POST']) def get_paycheck(typ, user_id): """ Handles requests to display paycheck details for a specified user and paycheck type. The function supports both GET and POST methods, allowing the user to specify a particular month and year. If not specified, it defaults to the current year and last month. The function checks if the user session is valid. If valid, it retrieves paycheck details based on the type (`typ`) and user ID (`user_id`). It then renders these details on 'paycheckview.html'. Args: typ (str): Type of paycheck data, e.g., 'Uber' for Uber-specific data. user_id (str): The unique identifier of the user for whom the paycheck details are being fetched. Returns: - Renders 'paycheckview.html' with paycheck details, including base data, any add-ons, and translations. - Redirects to 'login.html' with a warning if the session is not valid. - If an exception occurs, logs the error, optionally notifies IT via Telegram, and renders 'index.html' with an error message. Raises: - Exception: Captures any errors during data retrieval, logs the full traceback, sends a notification via Telegram, and flashes an error message to the user. Example Usage: - Accessing '/paycheck/Uber/12345' with a valid session and no POST data defaults to the last month's paycheck data. - Submitting a POST request with specific 'year' and 'month' parameters fetches paycheck data for the specified period. Notes: - Error handling includes full traceback logging for diagnostics, Telegram notifications for IT support, and user-friendly error messages. """ # Check if the session is valid if 'valid' in session and session['valid'] is True: try: # Default to the current year and last month if no specific month/year is requested year = get_current_year() month = str(get_last_month_number()) # If request method is POST, use the submitted month and year if request.method == 'POST': year = request.form.get('year') month = request.form.get('month') # Determine paycheck data based on the type ('Uber' or other) if typ == 'Uber': data = get_users_paycheck_data_uber(user_id, year, month) addons = None # No add-ons for 'Uber' type paycheck data else: data = get_users_paycheck_data(user_id, year, month) addons = get_users_addons_details_data(user_id, year, month) # Fetch any additional payment details else: # Handle GET request (defaults to the last month and current year) if typ == 'Uber': data = get_users_paycheck_data_uber(user_id, year, month) addons = None else: data = get_users_paycheck_data(user_id, year, month) addons = get_users_addons_details_data(user_id, year, month) # Prepare translations for displaying column headers in the template translations = {item['db']: item['de'] for item in translation} # Render the paycheck view template with all required data return render_template( 'paycheckview.html', data=data, user_id=user_id, typ=typ, addons=addons, year=year, month=month, translations=translations, site_name='paycheckView', username=session['username'] ) except Exception as e: # Capture full traceback for debugging in case of an error full_traceback = traceback.format_exc() # Notify IT support asynchronously about the database query failure asyncio.run(send_msg_telegram( lara_bot_id, f"Database query failed: {full_traceback} | Talentsphere", onduty_it )) # Flash an error message and render the main page in case of a database error flash('Datenbankfehler! Bitte IT Team melden.', 'danger') return render_template('index.html', username=session['username']) # If session is not valid, prompt the user to log in and redirect to login page flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/logs//', methods=['GET', 'POST']) def get_logs(table, user_id): """ Handles requests to display log details for a specific user based on the provided `table` and `user_id`. This function retrieves and processes log data, highlighting changes and resolving foreign key values for readability. The details are displayed on 'logs.html'. Processes: - Verifies if the user session is valid. If not, prompts the user to log in. - If the session is valid, fetches and processes log data associated with the user. - Processes the log data to highlight changes between records, making it easier for users to understand modifications. - Resolves foreign key data into readable values using helper functions for better context. Args: - table (str): Specifies the database table to query (e.g., "employee" or "administration"). - user_id (str): The unique identifier of the user for whom log details are being fetched. Returns: - Renders 'logs.html' with processed log data, translations, and additional context for a valid session. - Redirects to 'login.html' with a warning message if the session is not valid. - Renders 'index.html' with an error message if there is an exception during data retrieval or processing. Raises: - Exception: Catches exceptions related to data fetching and processing, logs the traceback, and optionally notifies the IT team via Telegram. Displays a generic database error message to the user. Notes: - This endpoint requires active session validation to ensure that only authenticated users can view log details. - Error handling includes full traceback logging for diagnostics and user-friendly error messaging. - External function `log_view` is used to fetch initial log data. - The `compare_detailed_changes_*` functions are used to identify and process changes between logs for display. - Foreign key data is resolved using `get_fk_data_from_dict` for readability. """ # Check if the session is valid if 'valid' in session and session['valid'] is True: try: # Initialize variables for site name and data final_data = None translations = {item['db']: item['de'] for item in translation} # Translation mapping for column headers # Fetch and process log data based on the specified table if table == "employee": site_name = "usersView" # Fetch raw log data for the specified user ID pre_data, columns = log_view(user_id) # Process raw rows into structured data for change comparison processed_data = convert_raw_rows_to_structure(pre_data) # Compare processed data to highlight detailed changes between records compared_data = compare_detailed_changes_users(processed_data) # Resolve foreign keys in the log data for readability final_data = process_fk_data_for_list(compared_data) elif table == "administration": site_name = "administrationView" # Fetch log data specific to the administration table for the specified user ID data = log_administration_view(user_id) # Compare log data for detailed changes in the administration logs final_data = compare_detailed_changes_administration(data) # Resolve foreign key values in the administration data for readability for dat in final_data['data']: dat[2] = get_fk_data_from_dict(dat[2]) # Resolves old value foreign keys dat[3] = get_fk_data_from_dict(dat[3]) # Resolves new value foreign keys # Render the logs view template with the processed data, translations, and additional context return render_template( 'logs.html', table=table, data=final_data, translations=translations, site_name=site_name, username=session['username'] ) except Exception: # Capture full traceback for debugging in case of an error full_traceback = traceback.format_exc() # Notify IT support asynchronously about the database query failure asyncio.run(send_msg_telegram( lara_bot_id, f"Database query failed: {full_traceback}", onduty_it )) # Flash a database error message to the user and render the main page flash('Datenbankfehler! Bitte IT Team melden.', 'danger') return render_template('index.html', username=session['username']) # If session is not valid, prompt the user to log in and redirect to login page flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/logout', methods = ['POST']) def logout(): """ Handles the user logout process by clearing the session and redirecting to the login page. This function is triggered through both GET and POST requests. It clears all data stored in the user's session, effectively logging the user out. After the session is cleared, the user is redirected to the login page to potentially log in again or continue as a guest. Returns: - Renders and displays the 'login.html' template, showing the login page where users can re-authenticate or proceed otherwise. Example Usage: - Accessing '/logout' will clear the current session and navigate the user to the login page, ensuring they must re-authenticate to access user-specific resources. Notes: - This endpoint should be secured to prevent unintended logouts via CSRF attacks, ensuring that logout requests are intentional. """ session.clear() resp = redirect(url_for('login'), code=303) # See Other, gut nach POST # Session-Cookie explizit löschen cookie_name = current_app.config.get('SESSION_COOKIE_NAME', 'session') resp.delete_cookie( cookie_name, path='/', domain=current_app.config.get('SESSION_COOKIE_DOMAIN', None), secure=current_app.config.get('SESSION_COOKIE_SECURE', False), httponly=current_app.config.get('SESSION_COOKIE_HTTPONLY', True), samesite=current_app.config.get('SESSION_COOKIE_SAMESITE', 'Lax'), ) # Zusätzliche Sicherheit gegen Cache resp.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate, max-age=0' resp.headers['Pragma'] = 'no-cache' resp.headers['Expires'] = '0' return resp @app.route('/insert_form', methods=['GET', 'POST']) def insert_form(): """ Renders a form for inserting new user data. Access to this form is restricted to logged-in users. The form includes dropdowns populated from foreign tables and fields for user data entry based on database schema. Processes: - Verifies if the user session is valid. If not, the user is redirected to the login page. - Retrieves column names and dropdown data from various database tables (foreign key references). - Generates a unique user ID for new user entries. - Each dropdown data set is fetched using `get_data_from_fk_table` function and displayed on the form. Returns: - Renders 'insert.html' with necessary data such as column names, unique user ID, and dropdown data for fields like approval status, civil status, fleet, etc., if the session is valid. - Redirects to the 'login.html' template with a warning message if the session is not valid. - Redirects to the 'index.html' template with an error message if there is an exception during data fetching. Raises: - Exception: If there is a failure in database queries, logs the exception and optionally sends a notification to the IT team via Telegram. The user is then informed of a database error. Example Usage: - Accessing '/insert_form' with a valid session will render 'insert.html' for new user data entry. - Without a valid session, the user will be redirected to the login page. Notes: - This function uses `get_country_dialing_codes` to fetch country codes and prefixes for phone number fields. - The function assumes all dropdown data is essential and that `get_data_from_fk_table` successfully retrieves valid lists. - Security should be considered to protect against unauthorized data submissions and ensure that only authenticated users can submit the form. """ if 'valid' in session and session['valid'] is True: try: #Nimmt die Columnnames von DB translations = {item['db']: item['de'] for item in translation} colum_names = get_insert_description() #uid wird erstellt user_id = generate_uid() #Liste wird erstellt mit FK Tabellen prefix, country = get_country_dialing_codes() approval = get_data_from_fk_table('users_approval',1, True) civil_status = get_data_from_fk_table('users_civil_status',1, True) fleet = get_data_from_fk_table('users_fleet',1, True) region = get_data_from_fk_table('users_region',1, True) gender = get_data_from_fk_table('users_gender',1, True) status_application = get_data_from_fk_table('users_status_application',1, True) status_work = get_data_from_fk_table('users_status_work',1, True) typ = get_data_from_fk_table('users_typ',1,True) return render_template('insert.html', colum_names = colum_names,user_id = user_id, approval = approval, civil_status = civil_status, fleet = fleet, region = region, gender = gender, status_application = status_application, status_work = status_work, typ = typ, prefix = prefix, country = country, translations = translations,admin = session['admin'], username = session['username']) except: full_traceback = traceback.format_exc() asyncio.run(send_msg_telegram(lara_bot_id,f"Database query failed: {full_traceback}",onduty_it)) flash('Datenbankfehler!', 'danger') return render_template('index.html') flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/insert_data', methods=['GET', 'POST']) def insert_data(): """ Processes POST requests to insert new user data into the database. This endpoint is designed to handle various conditions such as checking for duplicate entries and creating accounts in related systems. Processes: - Validates if the user session is valid. If not, redirects to the login page. - Checks for duplicate email or phone number entries. - Collects form data and constructs a dictionary for database insertion. - Conditionally handles the creation of related accounts (e.g., Google folders, Tookan, and Planday accounts). - Inserts data into the database and logs the action. - Flashes messages based on the success of operations and redirects or renders templates based on user actions. Returns: - Redirects to different routes or renders templates based on the user's choice of action after form submission. - Redirects to the 'index.html' template for successful insertions or actions. - Redirects to the 'login.html' template with a warning if the session is not valid. - Returns to the 'index.html' template with an error message if there is an exception during data processing. Raises: - Exception: Captures any exceptions during data processing, logs the error, and optionally notifies the IT team via Telegram. Flashes a generic database error message to the user. Example Usage: - Submitting the form on '/insert_data' with valid data will insert the data into the database and may create related accounts based on the form inputs. - If a user is not logged in and tries to access this route, they will be redirected to the login page. Notes: - The function ensures that all form submissions are authenticated to protect against unauthorized data manipulation. - The detailed implementation for handling account creations and database interactions should be secured and optimized to handle potential data inconsistencies and to ensure data integrity. """ if 'valid' in session and session['valid'] is True: if request.method == 'POST': try: planday_acc = False google_fold = False tookan_acc = False #Wird geschaut, ob Email oder Telefon schon existiert if check_duplicate(request.form.get('email', None),request.form.get('phone', None)): flash('Email oder Telefon existiert schon!', 'warning') return redirect(url_for('insert_form')) # Daten werden in eine Dict gespeichert # exakt auszuschließen (Case-sensitiv) EXCLUDED_KEYS = {'submitAction'} # Teilstrings zum Ausschließen (Case-insensitiv) EXCLUDED_CONTAINS = {'child', 'partner'} insert_data = { k: _normalize(v) for k, v in request.form.items() if k not in EXCLUDED_KEYS and not any(sub in k.lower() for sub in EXCLUDED_CONTAINS) } insert_partner_data = {key: request.form[key] for key in request.form if key != 'submitAction' and 'partner' in key} if all(not value.strip() for key, value in insert_partner_data.items() if key != 'partner_birth_date'): insert_partner_data = None else: insert_partner_data = add_user_id(insert_partner_data) insert_child_pre_data = {key: request.form[key] for key in request.form if key != 'submitAction' and 'child' in key} insert_child_data = extract_children_data(insert_child_pre_data) insert_data['user_name'] = request.form.get('first_name', None) + " " + request.form.get('last_name', None) insert_data['doc_url'] = handle_account_creation(create_google_folder, request.form.get('user_id')) google_fold = True if request.form.get('tookan_id', False) == "on": insert_data['tookan_id'] = handle_account_creation(create_tookan_acc, insert_data['phone'], insert_data['user_name'], insert_data['first_name'], insert_data['last_name'], insert_data['email']) tookan_acc = True if request.form.get('planday_id', False) == "on": insert_data['planday_id'] = handle_account_creation(create_planday_acc, insert_data['first_name'], insert_data['last_name'], insert_data['email'], insert_data['region'], insert_data['phone'], insert_data['prefix'], insert_data['address'], insert_data['postal_code'], insert_data['location'], insert_data['salutation'], insert_data['country'], insert_data['birth_date']) if insert_data['planday_id']: if insert_data['region'] != 102: asyncio.run(send_msg_telegram(lara_bot_id,inform_jan_bc_user_new_planday(insert_data['user_name'], insert_data['approval']),onduty_recruitment)) planday_acc = True # Ein String wird erstellt mit den Keys und werden mit ',' getrennt keys = ", ".join(insert_data.keys()) # Erstellt platzhalter für die gleiche Anzahl der Keys placeholders = ", ".join(["%s" for _ in insert_data]) insert_values = list(insert_data.values()) # None oder Leere felder mit Python None ersetzen converted_values = [None if item in ('None', '') else item for item in insert_values] # insert_data_db aufgerufen um in der DB zu speichern insert_data_db(keys,placeholders,converted_values) if insert_partner_data: partner_id = generate_uid() insert_partner_data = replace_empty_with_none(insert_partner_data) insert_partner_db(insert_partner_data, insert_data['user_id'],partner_id) insert_child_data = replace_empty_with_none(insert_child_data) insert_childerns_data(insert_data['user_id'], insert_child_data) log(insert_data['user_id'],insert_data,insert_data, "insert", session['username'], 'employee') flash('Erfolgreich editiert!', 'success') if google_fold: flash_message(insert_data['doc_url'] is not None, 'Google Ordner erstellt!', 'Fehler bei Erstellung Google Ordner!') if tookan_acc: flash_message(insert_data['tookan_id'] is not None, 'Tookan Account erstellt!', 'Fehler bei Erstellung Tookan Account!') if planday_acc: flash_message(insert_data['planday_id'] is not None, 'Planday Account erstellt!', 'Fehler bei Erstellung Planday Account!') if request.form['submitAction'] == 'close': return render_template('index.html') if request.form['submitAction'] == 'edit': user_id = request.form.get('user_id') return redirect(url_for('edit_form', user_id=user_id)) except: full_traceback = traceback.format_exc() asyncio.run(send_msg_telegram(lara_bot_id,f"Database query failed: {full_traceback} | Talentsphere",onduty_it)) flash('Datenbankfehler!', 'danger') return render_template('index.html', username = session['username']) flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/insert_form_administration', methods=['GET', 'POST']) def insert_form_administration(): """ Renders a form for inserting new user data. Access to this form is restricted to logged-in users. The form includes dropdowns populated from foreign tables and fields for user data entry based on database schema. Processes: - Verifies if the user session is valid. If not, the user is redirected to the login page. - Retrieves column names and dropdown data from various database tables (foreign key references). - Generates a unique user ID for new user entries. - Each dropdown data set is fetched using `get_data_from_fk_table` function and displayed on the form. Returns: - Renders 'insert.html' with necessary data such as column names, unique user ID, and dropdown data for fields like approval status, civil status, fleet, etc., if the session is valid. - Redirects to the 'login.html' template with a warning message if the session is not valid. - Redirects to the 'index.html' template with an error message if there is an exception during data fetching. Raises: - Exception: If there is a failure in database queries, logs the exception and optionally sends a notification to the IT team via Telegram. The user is then informed of a database error. Example Usage: - Accessing '/insert_form' with a valid session will render 'insert.html' for new user data entry. - Without a valid session, the user will be redirected to the login page. Notes: - This function uses `get_country_dialing_codes` to fetch country codes and prefixes for phone number fields. - The function assumes all dropdown data is essential and that `get_data_from_fk_table` successfully retrieves valid lists. - Security should be considered to protect against unauthorized data submissions and ensure that only authenticated users can submit the form. """ if 'valid' in session and session['valid'] is True: try: #Takes the Columns from db translations = {item['db']: item['de'] for item in translation} case_id = generate_uid() user_data = get_userid_username() colum_names = get_insert_administration_description() docs_formate = get_docs_formate_from_administration_table() talentsphere_concerns = get_all_data_from_fk_table('talentsphere_concerns') talentsphere_transmission_method = get_all_data_from_fk_table('talentsphere_transmission_method') return render_template('insertadministration.html', colum_names = colum_names, talentsphere_concerns = talentsphere_concerns,docs_formate = docs_formate, talentsphere_transmission_method = talentsphere_transmission_method, case_id = case_id, translations = translations, user_data = user_data, site_name = "administrationView",username = session['username']) except: full_traceback = traceback.format_exc() asyncio.run(send_msg_telegram(lara_bot_id,f"Database query failed: {full_traceback}",onduty_it)) flash('Datenbankfehler!', 'danger') return render_template('insertadministration.html') flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/insert_data_administration', methods=['GET', 'POST']) def insert_data_administration(): if 'valid' in session and session['valid'] is True: if request.method == 'POST': try: file_paths, files_new = save_files_tmp(request.files, request.form) file_paths_with_dates = append_date_suffix_to_files(file_paths, request.form.get('validFrom', None), request.form.get('validUntil', None)) url_name = upload_files_to_google_drive(file_paths_with_dates, str(request.form.get('user_id', None))) excluded = {'submitAction', 'file_input', 'file', 'validUntil', 'validFrom'} insert_data = { k: _normalize(v) for k, v in request.form.items() if k not in excluded # z.B. {'submitAction', 'file', ...} } # Ein String wird erstellt mit den Keys und werden mit ',' getrennt keys = ", ".join(insert_data.keys()) # Erstellt platzhalter für die gleiche Anzahl der Keys placeholders = ", ".join(["%s" for _ in insert_data]) insert_values = list(insert_data.values()) if url_name: keys += ", file_url" placeholders += ", %s" insert_values.append(url_name[0]['url']) # None oder Leere felder mit Python None ersetzen converted_values = [None if item in ('None', '') else item for item in insert_values] # insert_data_db aufgerufen um in der DB zu speichern insert_data_db_administration(keys,placeholders,converted_values) log(insert_data['case_id'],insert_data,insert_data, "insert", session['username'],'administration') flash('Erfolgreich eingefügt!', 'success') if request.form['submitAction'] == 'close': return render_template('administrationview.html', table = 1, site_name = "administrationView") if request.form['submitAction'] == 'edit': case_id = request.form.get('case_id') return redirect(url_for('edit_form_administration', case_id=case_id)) except: full_traceback = traceback.format_exc() asyncio.run(send_msg_telegram(lara_bot_id,f"Database query failed: {full_traceback} | Talentsphere",onduty_it)) flash('Datenbankfehler!', 'danger') return render_template('index.html', username = session['username']) flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/edit_form_administration/', methods=['GET', 'POST']) def edit_form_administration(case_id): """ Renders the administration edit form for a specified case, retrieving the necessary data for editing. Process: - Checks if the user session is valid. If not, prompts for login. - If the session is valid, retrieves the data needed for editing the specified administration case: - Column names and values for the selected `case_id`. - Foreign key data for dropdowns, including user data, concerns, and transmission methods. - Renders the 'editadministration.html' template with the retrieved data. - If an error occurs during data retrieval, logs the error, sends a notification to IT, and displays an error message. Args: - case_id (str): The unique identifier of the case to be edited. Returns: - Renders 'editadministration.html' with data for the specified case if the session is valid. - Renders 'login.html' if the session is not valid. - Renders 'index.html' with an error message if an exception occurs. Example Usage: - Accessing '/edit_form_administration/' with a valid session will render 'editadministration.html' for the specified case. Notes: - Assumes that helper functions like `get_administration_row_data_for_edit`, `get_userid_username`, and `get_all_data_from_fk_table` are defined to fetch necessary data from the database. - Includes error handling to capture and report database issues. """ # Check if the session is valid if 'valid' in session and session['valid'] is True: try: # Retrieve column translations from the database translations = {item['db']: item['de'] for item in translation} # Retrieve current values for the specified case_id column_name_and_value = get_administration_data_for_edit_with_full_address(case_id) # Retrieve supporting data for dropdowns and foreign key tables user_data = get_userid_username() # Get user ID and username pairs colum_names = get_insert_administration_description() # Retrieve descriptions for column names docs_formate = get_docs_formate_from_administration_table() talentsphere_concerns = get_all_data_from_fk_table('talentsphere_concerns') # Concerns table data talentsphere_transmission_method = get_all_data_from_fk_table('talentsphere_transmission_method') # Transmission methods # Render the edit form template with the retrieved data return render_template('editadministration.html', data=column_name_and_value, user_data=user_data, colum_names=colum_names, translations=translations, talentsphere_concerns=talentsphere_concerns, talentsphere_transmission_method=talentsphere_transmission_method, docs_formate = docs_formate, admin=session['admin'], site_name="administrationView", username=session['username']) except: # Capture the full traceback in case of an error full_traceback = traceback.format_exc() # Send an error notification to IT via Telegram asyncio.run(send_msg_telegram(lara_bot_id, f"Database query failed: {full_traceback} | Talentsphere", onduty_it)) # Flash an error message to the user and redirect to the index flash('Datenbankfehler! Bitte IT Team melden.', 'danger') return render_template('index.html') # If session is not valid, prompt user to log in flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/edit_data_administration', methods=['GET', 'POST']) def edit_data_administration(): """ Handles requests to edit existing data in the administration table. The function supports both GET and POST methods, with primary data processing on a POST request. Process: - Checks if the user session is valid. If not, prompts the user to log in. - If the session is valid and the request is a POST: - Collects form data and constructs a dictionary for updates, replacing empty values with None. - Dynamically constructs an SQL update query based on the form data keys. - Calls a helper function to update the database. - Logs the update action for auditing and tracking purposes. Returns: - Flashes a success message on successful update and reloads the administration view or redirects to the edit form. - Redirects to 'login.html' with a warning if the session is invalid. - Renders 'index.html' with an error message if there is an exception during data processing. Raises: - Exception: Catches errors during data processing, logs the full traceback, and optionally notifies IT support. A generic database error message is flashed to the user. Example Usage: - Submitting the form on '/edit_data_administration' with modified data updates the relevant record in the database. - If the user is not logged in, they are redirected to the login page. """ # Check if the session is valid if 'valid' in session and session['valid'] is True: # Process form data on POST request if request.method == 'POST': try: # Collect form data into a dictionary, replacing empty strings with None # Exclude 'id' and 'submitAction' as they should not be updated delete_file = request.form.get('delete_file') # Returns None if the key doesn't exist # exakt auszuschließen EXCLUDED_KEYS = { 'id', 'submitAction', 'user_name', 'user_id', 'file_input', 'validFrom', 'validUntil', 'delete_file', 'file', 'full_address' } # Teilstrings zum Ausschließen (case-insensitive) EXCLUDED_CONTAINS = {'child', 'partner'} update_data = { k: _normalize(v) for k, v in request.form.items() if k not in EXCLUDED_KEYS and not any(sub in k.lower() for sub in EXCLUDED_CONTAINS) } update_set = ", ".join([f"{key} = %s" for key in update_data.keys() if not any(ignore in key for ignore in ('user_id', 'file_input','validFrom', 'validUntil')) and 'front' not in key]) if request.files: file_paths, files_new = save_files_tmp(request.files, request.form) file_paths_with_dates = append_date_suffix_to_files(file_paths, request.form.get('validFrom', None), request.form.get('validUntil', None)) url_name = upload_files_to_google_drive(file_paths_with_dates, str(request.form.get('user_id', None))) update_set += f", file_url = %s" update_data['file_url'] = url_name[0]['url'] update_data.pop('file', None) if delete_file: file_id = extract_google_drive_file_id(update_data['file_url']) file_data = get_file_by_id_in_folder(request.form.get('user_id'), file_id) new_name = f"{file_data['name']}_DELETED" status = rename_file(file_id, new_name) if status: update_data['file_url'] = None flash('Datei gelöscht', 'success') else: flash('Datei konnte nicht gelöscht werden', 'warning') # Prepare values for each key in the update_data dictionary values = list(update_data.values()) # Retrieve the unique case_id for this entry (used as the primary key in the update query) update_data['case_id'] = request.form.get('case_id', None) # Fetch existing data to compare with the new data for logging purposes update_data_old = get_administration_data_for_edit_as_string(update_data['case_id']) # Perform the database update using the constructed query and values update_user_administration(update_set, values, update_data['case_id']) # Log the update action, storing both the old and new data for audit tracking log(update_data['case_id'], update_data_old[0], update_data, "update", session['username'], "administration") # Flash success message to inform the user of a successful edit flash('Erfolgreich editiert!', 'success') # Determine the next action based on the submitAction value if request.form['submitAction'] == 'close': # Redirect to administration view after successful editing return render_template('administrationview.html', table=1, username=session['username']) elif request.form['submitAction'] == 'edit': # Redirect to the edit form for further modifications if requested case_id = request.form.get('case_id') return redirect(url_for('edit_form_administration', case_id=case_id, username=session['username'])) except Exception: # Capture the full traceback in case of an error full_traceback = traceback.format_exc() # Notify IT support asynchronously about the database query failure asyncio.run(send_msg_telegram( lara_bot_id, f"Database query failed: {full_traceback} | Talentsphere", onduty_it )) # Flash a database error message to the user and render the main page flash('Datenbankfehler! Bitte IT Team melden.', 'danger') return render_template('index.html', username=session['username']) # If session is not valid, prompt the user to log in and redirect to login page flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/edit_form/', methods=['GET', 'POST']) def edit_form(user_id): """ Renders a form for editing specific user data. Access to this form is restricted to logged-in users. The form includes dropdowns populated from foreign key tables and fields for user data entry based on database schema. The function fetches existing user data from the database and prepares dropdown lists with foreign key table data to facilitate user input corrections or updates. Args: user_id (str): The unique identifier for the user whose data needs to be edited. Returns: - Renders 'edit.html' with user data and dropdown options if the session is valid. - Redirects to 'login.html' with a warning message if the session is not valid. - Returns to 'index.html' with an error message if there is an exception during data fetching. Raises: - Exception: Captures any exceptions during data fetching or processing, logs the error, and optionally notifies the IT team via Telegram. The user is then informed of a database error. Example Usage: - Accessing '/edit_form/12345' with a valid session will render 'edit.html' for editing the data of user 12345. - If a user is not logged in and tries to access this route, they will be redirected to the login page. Notes: - This endpoint ensures all form submissions and edits are authenticated to protect against unauthorized data manipulation. - It is vital to handle foreign key data carefully to ensure data integrity and maintain consistency in dropdowns. - The function makes extensive use of database interactions to fetch currently relevant data user data and related foreign key information. """ if 'valid' in session and session['valid'] is True: try: #Nimmt die Columnnames von DB translations = {item['db']: item['de'] for item in translation} column_name_and_value = get_user_data_for_edit(user_id) childrens_data = get_childrens_data(user_id) partner_data = get_partner_data(user_id) external_acc_status = get_external_account_status(user_id) #Liste wird erstellt mit FK Tabellen prefix, country = get_country_dialing_codes() approval = get_data_from_fk_table('users_approval',1, True) civil_status = get_data_from_fk_table('users_civil_status',1, True) fleet = get_data_from_fk_table('users_fleet',1, True) region = get_data_from_fk_table('users_region',1, True) salutation = get_data_from_fk_table('users_salution',1, True) status_application = get_data_from_fk_table('users_status_application',1, True) status_work = get_data_from_fk_table('users_status_work',1, True) typ = get_data_from_fk_table('users_typ',1, True) gender = get_data_from_fk_table('users_gender',1, True) files = list_files_in_folder(user_id) needed_docs = get_required_docs(column_name_and_value[0]['approval'], column_name_and_value[0]['typ']) needed_docs_list = check_document_requirements(needed_docs,files) administration_docs_formate = get_docs_formate_from_administration_table() list_administration_docs_formate = create_list_with_specific(administration_docs_formate, 'name') #required_table = compare_existing_vs_neended(needed_docs,files) return render_template('edit.html', data = column_name_and_value, approval = approval, external_acc_status = external_acc_status, civil_status = civil_status, fleet = fleet, region = region, childrens_data = childrens_data, salutation = salutation,status_application = status_application, partner_data = partner_data[0], status_work = status_work, typ = typ ,user_id = user_id, prefix = prefix, gender = gender, list_administration_docs_formate = list_administration_docs_formate, country = country, translations = translations, admin = session['admin'], files=files,needed_docs = needed_docs_list,username = session['username']) except: full_traceback = traceback.format_exc() asyncio.run(send_msg_telegram(lara_bot_id,f"Database query failed: {full_traceback} | Talentsphere",onduty_it)) flash('Datenbankfehler! Bitte IT Team melden.', 'danger') return render_template('index.html') flash('Bitte einloggen!', 'warning') return render_template('login.html') """@app.route('/edit_data', methods=['GET', 'POST']) def edit_data(): Processes POST requests to edit existing user data in the database. This function checks for uniqueness of the email or phone number among existing records before committing updates. The function: - Verifies user session validity. - Checks for duplicate email or phone numbers to avoid conflicts. - Gathers form data, excluding specific fields like 'row_id' and 'user_id' which should not be updated. - Optionally creates related accounts (e.g., Tookan or Planday) if indicated by the user. - Updates user data in the database and logs the changes. - Flashes success or error messages based on the outcome of the update process. Returns: - Redirects to 'edit_form' with the current user_id if a duplicate is found. - Redirects to 'index.html' with a success message if the update is successful. - Redirects to 'login.html' with a warning message if the session is not valid. - Returns to 'index.html' with an error message if there is an exception during data processing. # Check if the user is logged in and the session is valid if 'valid' in session and session['valid'] is True: # Check if the request is a POST and if there's an action to submit if request.method == 'POST' and request.form['submitAction']: try: # Initialize flags for conditional account creation planday_acc = False tookan_acc = False partner_operations = None log_metadata = {} tookan_id = request.form.get('tookan_id', None) # This is the actual ID, or '' tookan_checked = request.form.get('tookan_id_checkbox') == '1' planday_id = request.form.get('planday_id', None) planday_checked = request.form.get('planday_id_checkbox') == '1' user_id = request.form.get('user_id') acc_status = get_external_account_status(user_id) # Save temporary files uploaded in the form file_paths, files_new = save_files_tmp(request.files, request.form) # Retrieve file list from the form and parse it file_list = request.form.getlist('files') file_list = parse_file_data(file_list) # Check for duplicate email or phone to prevent conflicts if check_duplicate(request.form.get('email', None), request.form.get('phone', None), request.form.get('user_id', None)): flash('Email oder Telefon existiert schon!', 'warning') return redirect(url_for('edit_form', user_id=request.form.get('user_id', None))) update_data = { k: _normalize(v) for k, v in request.form.items() if k not in {'row_id', 'user_id', 'submitAction'} and not any(sub in k.lower() for sub in {'child', 'partner', 'delete', 'enddate', 'startdate', 'files'}) } # Prepare child and partner data for updates update_child_predata = { k: _normalize(v) for k, v in request.form.items() if k not in ('row_id', 'user_id', 'submitAction') and 'child' in k } update_partner_predata = { k: _normalize(v) for k, v in request.form.items() if k not in ('row_id', 'user_id', 'submitAction') and 'partner' in k } # If no partner data is provided, set update_partner_predata to None if all(value is None for value in update_partner_predata.values()): update_partner_predata = None # Construct the full user name and adjust if needed update_data['user_name'] = request.form.get('first_name', None) + " " + request.form.get('last_name', None) # Add "(DNF)" suffix if user is marked as unable to drive or has a specific work status if int(update_data.get('able_to_drive')) == 0: acc_status = get_external_account_status(user_id) tookan_status = next((item["status"] for item in acc_status if item["name"] == "tookan"), None) planday_status = next((item["status"] for item in acc_status if item["name"] == "planday"), None) update_data['user_name'] += " (DNF)" if request.form.get('tookan_id') and tookan_status is not None and int(tookan_status) == 1: status = update_tookan_user_status(update_data['tookan_id'], 0, "Deaktiviert") if status: update_external_acc_status(request.form.get('user_id', None), 'tookan', 0) log_metadata['tookan_acc'] = "Deaktiviert" flash('Tookan Account deaktiviert', 'info') if int(update_data.get('status_work')) == 10: if request.form.get('planday_id', None) and planday_status is not None and int(planday_status) == 1: status = deactivate_planday_acc(update_data['planday_id']) if status: employeeGroups = get_user_planday_employeeGroups(update_data['planday_id']) insert_last_planday_department_users(update_data['planday_id'], employeeGroups) update_external_acc_status(request.form.get('user_id', None), 'planday', 0) log_metadata['planday_acc'] = "Deaktiviert" flash('Planday Account deaktiviert', 'info') else: if request.form.get('planday_id', None) and planday_status is not None and int(planday_status) == 1: employeeGroups = get_user_planday_employeeGroups(update_data['planday_id']) #print(update_data['planday_id'],employeeGroups) insert_last_planday_department_users(update_data['planday_id'], employeeGroups) #move_planday_acc_from_DNF_to_employeeGroups(update_data['planday_id'], employeeGroups) status = move_planday_acc_DNF(update_data['planday_id']) if status: update_external_acc_status(request.form.get('user_id', None), 'planday', 2) log_metadata['planday_acc'] = "zur DNF department verschoben" flash('Planday Account zur DNF department verschoben', 'info') # Conditionally create Tookan account if requested if tookan_checked: if not tookan_id: update_data['tookan_id'] = handle_account_creation( create_tookan_acc, update_data['phone'], update_data['user_name'], update_data['first_name'], update_data['last_name'], update_data['email'] ) if update_data['tookan_id'] is not None: insert_external_acc_status(user_id, 'tookan', 1) log_metadata['tookan_acc'] = "Erstellt" tookan_acc = True else: flash('Fehler beim Erstellen des Tookan-Kontos!', 'warning') else: # Tookan-Konto existiert – reaktivieren status = update_tookan_user_status(tookan_id, 1, "Reaktiviert") if status: # Prüfen, ob Tookan in den externen Accounts existiert tookan_exists = any(acc['name'] == 'tookan' for acc in acc_status) if tookan_exists: update_external_acc_status(user_id, 'tookan', 1) log_metadata['tookan_acc'] = "Reaktiviert" else: insert_external_acc_status(user_id, 'tookan', 1) log_metadata['tookan_acc'] = "Reaktivierungsstatus neu eingefügt" flash('Kein bestehender Tookan-Eintrag – neuer Status wurde eingefügt.', 'info') else: flash('Fehler beim Aktualisieren des Tookan-Kontos!', 'warning') # Conditionally create Planday account if requested if planday_checked: if not planday_id: update_data['planday_id'] = handle_account_creation( create_planday_acc, update_data['first_name'], update_data['last_name'], update_data['email'], update_data['region'], update_data['phone'], update_data['prefix'], update_data['address'], update_data['postal_code'], update_data['location'], update_data['gender'], update_data['country'], update_data['birth_date'] ) # If account creation succeeded, notify the team via Telegram if update_data['planday_id']: insert_external_acc_status( user_id, 'planday', 1 ) log_metadata['planday_acc'] = "Erstellt" if update_data['planday_id']: asyncio.run( send_msg_telegram( lara_bot_id, inform_jan_bc_user_new_planday(update_data['user_name'], update_data['approval']), onduty_recruitment ) ) planday_acc = True else: flash('Fehler beim Erstellen des Planday Accounts!', 'warning') else: acc_status = get_external_account_status(user_id) planday_status = next((item["status"] for item in acc_status if item["name"] == "planday"), None) # If Planday account already exists, update its status if int(planday_status) == 0: status = reactivate_planday_acc(planday_id, update_data['region']) if status: update_external_acc_status(user_id, 'planday', 1) log_metadata['planday_acc'] = "Reaktiviert" else: flash('Fehler beim Aktualisieren des Planday Accounts!', 'warning') elif int(planday_status) == 2: departments = select_last_planday_department_users(planday_id) print(departments) status = move_planday_acc_from_DNF_to_employeeGroups(planday_id, departments) if status: update_external_acc_status(user_id, 'planday', 1) log_metadata['planday_acc'] = "Von DNF zu alte Personalgruppe verschoben" flash('Planday Account von DNF zu alte Personalgruppe verschoben', 'info') else: flash('Fehler beim Aktualisieren des Planday Accounts!', 'warning') else: flash('Planday Account ist bereits aktiv!', 'info') if update_data['status_application'] == "11" and update_data.get('planday_id') is not None and update_data['able_to_drive'] == "1": update_data['able_to_drive'] = 0 move_to_11KA = put_driver_in_11KA_in_planday(update_data['planday_id']) flash_message(move_to_11KA, 'Mitarbeiter auf 11KA verschoben', 'Fehler bei 11KA verschiebung') for field in ['planday_id_checkbox', 'tookan_id_checkbox']: update_data.pop(field, None) # Prepare SQL update string and values for execution update_set = ", ".join([f"{key} = %s" for key, val in update_data.items()]) values = [val for val in update_data.values()] update_data['user_id'] = request.form.get('user_id', None) update_data_old = get_user_data_for_edit_as_string(update_data['user_id']) old_user = update_data_old['user_data'][0] # first row = old user record if update_data.get("planday_id"): if value_changed(old_user, update_data, "address", "postal_code", "location", "country"): ok, error = update_planday_employee(update_data.get("planday_id"), address=update_data.get("address"), postal_code=update_data.get("postal_code"), location=update_data.get("location"), country=update_data.get("country")) flash_message(ok, 'Adresse in Planday aktualisiert', 'Fehler beim Aktualisieren der Adresse in Planday, bitte manuell anpassen') if error: asyncio.run(send_msg_telegram(lara_bot_id, f"Fehler beim Aktualisieren des Planday Accounts: {error} | Talentsphere", onduty_it)) if value_changed(old_user, update_data, "first_name", "last_name"): ok, error = update_planday_employee(update_data.get("planday_id"), first_name=update_data.get("first_name"), last_name=update_data.get("last_name")) flash_message(ok, 'Name in Planday aktualisiert', 'Fehler beim Aktualisieren der Name in Planday, bitte manuell anpassen') if error: asyncio.run(send_msg_telegram(lara_bot_id, f"Fehler beim Aktualisieren des Planday Accounts: {error} | Talentsphere", onduty_it)) if value_changed(old_user, update_data, "prefix", "phone"): ok, error = update_planday_employee(update_data.get("planday_id"), prefix=update_data.get("prefix"), phone=update_data.get("phone")) flash_message(ok, 'Telefonnummer in Planday aktualisiert', 'Fehler beim Aktualisieren der Telefonnummer in Planday, bitte manuell anpassen') if error: asyncio.run(send_msg_telegram(lara_bot_id, f"Fehler beim Aktualisieren des Planday Accounts: {error} | Talentsphere", onduty_it)) if value_changed(old_user, update_data, "birth_day"): ok, error = update_planday_employee(update_data.get("planday_id"), birth_day=update_data.get("birth_day")) flash_message(ok, 'Geburtstag in Planday aktualisiert', 'Fehler beim Aktualisieren der Telefonnummer in Planday, bitte manuell anpassen') if error: asyncio.run(send_msg_telegram(lara_bot_id, f"Fehler beim Aktualisieren des Planday Accounts: {error} | Talentsphere", onduty_it)) if old_user['able_to_drive'] == 0 and update_data['able_to_drive'] == 1: if "planday_id" in update_data and update_data['planday_id'] is not None: ok = reset_planday_name(update_data['planday_id'], update_data['first_name'], update_data['last_name']) flash_message(ok, '(DNF) in Planday wurde entfernt', 'Fehler beim entfernen von (DNF) in Planday') final_update_old_data = add_current_files(update_data_old) # Extract child data and update child records update_child_data = extract_children_data(update_child_predata) childres_operations = update_childrens(update_child_data, update_data['user_id']) # Update partner records if partner data exists if update_partner_predata: partner_operations = update_partner(update_partner_predata, update_data['user_id']) # Execute the user data update in the database update_user(update_set, values, request.form.get('user_id', None)) # Upload files to Google Drive and save document paths url_name = upload_files_to_google_drive(file_paths, str(request.form.get('user_id', None))) if "planday_id" in update_data and update_data['planday_id'] is not None: check_if_docs_are_valid_and_update_planday_name(update_data['planday_id'], url_name, update_data['first_name'], update_data['last_name']) insert_users_doc(url_name, str(request.form.get('user_id', None))) # Update file list with newly uploaded file URLs file_list = str_to_list(file_list) file_list.extend(url_name) # Combine update_data + log_metadata for logging log_data_combined = {**update_data, **log_metadata} # Log the final update data structure for future reference final_update_data = convert_to_log_structure( log_data_combined, update_child_data, childres_operations, update_partner_predata, partner_operations, file_list ) log( request.form.get('user_id', None), final_update_old_data, final_update_data, "update", session['username'], 'employee' ) # Clean up temporary files after uploading sel = remove_files(file_paths) # Flash success message flash('Erfolgreich editiert!', 'success') # Provide feedback on account creation results if tookan_acc: flash_message(update_data['tookan_id'] is not None, 'Tookan Account erstellt!', 'Fehler bei Erstellung Tookan Account!') if planday_acc: flash_message(update_data['planday_id'] is not None, 'Planday Account erstellt!', 'Fehler bei Erstellung Planday Account!') # Handle different submit actions if request.form['submitAction'] == 'close': return render_template('index.html', table=1, username=session['username']) elif request.form['submitAction'] == 'edit': user_id = request.form.get('user_id') return redirect(url_for('edit_form', user_id=user_id, username=session['username'])) except Exception as e: # Capture full traceback for error reporting full_traceback = traceback.format_exc() # Notify IT via Telegram about the exception asyncio.run(send_msg_telegram(lara_bot_id, f"Database query failed: {full_traceback} | Talentsphere", onduty_it)) # Flash error message to the user flash('Datenbankfehler! Bitte IT Team melden.', 'danger') # Redirect to main page return render_template('index.html', username=session['username']) elif request.form['submitAction']: # If there's an action but no POST data, redirect to edit form user_id = request.form.get('user_id') return redirect(url_for('edit_form', user_id=user_id, username=session['username'])) # If user is not logged in, redirect to login flash('Bitte einloggen!', 'warning') return render_template('login.html')""" @app.route('/edit_data', methods=['GET', 'POST']) def edit_data(): """ Processes POST requests to edit existing user data in the database. This function checks for uniqueness of the email or phone number among existing records before committing updates. The function: - Verifies user session validity. - Checks for duplicate email or phone numbers to avoid conflicts. - Gathers form data, excluding specific fields like 'row_id' and 'user_id' which should not be updated. - Optionally creates related accounts (e.g., Tookan or Planday) if indicated by the user. - Updates user data in the database and logs the changes. - Flashes success or error messages based on the outcome of the update process. Returns: - Redirects to 'edit_form' with the current user_id if a duplicate is found. - Redirects to 'index.html' with a success message if the update is successful. - Redirects to 'login.html' with a warning message if the session is not valid. - Returns to 'index.html' with an error message if there is an exception during data processing. """ # Check if the user is logged in and the session is valid if 'valid' in session and session['valid'] is True: # Check if the request is a POST and if there's an action to submit if request.method == 'POST' and request.form['submitAction']: try: # Initialize flags for conditional account creation planday_acc = False tookan_acc = False partner_operations = None log_metadata = {} tookan_id = request.form.get('tookan_id', None) # This is the actual ID, or '' tookan_checked = request.form.get('tookan_id_checkbox') == '1' planday_id = request.form.get('planday_id', None) planday_checked = request.form.get('planday_id_checkbox') == '1' user_id = request.form.get('user_id') acc_status = get_external_account_status(user_id) # Save temporary files uploaded in the form file_paths, files_new = save_files_tmp(request.files, request.form) # Retrieve file list from the form and parse it file_list = request.form.getlist('files') file_list = parse_file_data(file_list) # Check for duplicate email or phone to prevent conflicts if check_duplicate(request.form.get('email', None), request.form.get('phone', None), request.form.get('user_id', None)): flash('Email oder Telefon existiert schon!', 'warning') return redirect(url_for('edit_form', user_id=request.form.get('user_id', None))) # Gather form data to update, excluding fields that shouldn't be updated update_data = { key: (None if val in ['None', ''] else val) for key, val in request.form.items() if key not in ('row_id', 'user_id', 'submitAction') and 'child' not in key and 'partner' not in key and 'delete' not in key and 'endDate' not in key and 'startDate' not in key and 'files' not in key } # Prepare child and partner data for updates update_child_predata = { key: (None if val in ['None', ''] else val) for key, val in request.form.items() if key not in ('row_id', 'user_id', 'submitAction') and 'child' in key } update_partner_predata = { key: (None if val in ['None', ''] else val) for key, val in request.form.items() if key not in ('row_id', 'user_id', 'submitAction') and 'partner' in key } # If no partner data is provided, set update_partner_predata to None if all(value is None for value in update_partner_predata.values()): update_partner_predata = None # Construct the full user name and adjust if needed update_data['user_name'] = request.form.get('first_name', None) + " " + request.form.get('last_name', None) # Add "(DNF)" suffix if user is marked as unable to drive or has a specific work status if int(update_data.get('able_to_drive')) == 0 or int(update_data.get('status_work')) == 10: acc_status = get_external_account_status(user_id) tookan_status = next((item["status"] for item in acc_status if item["name"] == "tookan"), None) planday_status = next((item["status"] for item in acc_status if item["name"] == "planday"), None) update_data['user_name'] += " (DNF)" if request.form.get('tookan_id') and tookan_status is not None and int(tookan_status) == 1: status = update_tookan_user_status(update_data['tookan_id'], 0, "Gekündigt") if status: update_external_acc_status(request.form.get('user_id', None), 'tookan', 0) log_metadata['tookan_acc'] = "Deaktiviert" if request.form.get('planday_id', None) and planday_status is not None and int(planday_status) == 1: status = deactivate_planday_acc(update_data['planday_id']) if status: update_external_acc_status(request.form.get('user_id', None), 'planday', 0) log_metadata['planday_acc'] = "Deaktiviert" # Conditionally create Tookan account if requested if tookan_checked: if not tookan_id: update_data['tookan_id'] = handle_account_creation( create_tookan_acc, update_data['phone'], update_data['user_name'], update_data['first_name'], update_data['last_name'], update_data['email'] ) if update_data['tookan_id'] is not None: insert_external_acc_status(user_id, 'tookan', 1) log_metadata['tookan_acc'] = "Erstellt" tookan_acc = True else: flash('Fehler beim Erstellen des Tookan-Kontos!', 'warning') else: # Tookan-Konto existiert – reaktivieren status = update_tookan_user_status(tookan_id, 1, "Reaktiviert") if status: # Prüfen, ob Tookan in den externen Accounts existiert tookan_exists = any(acc['name'] == 'tookan' for acc in acc_status) if tookan_exists: update_external_acc_status(user_id, 'tookan', 1) log_metadata['tookan_acc'] = "Reaktiviert" else: insert_external_acc_status(user_id, 'tookan', 1) log_metadata['tookan_acc'] = "Reaktivierungsstatus neu eingefügt" flash('Kein bestehender Tookan-Eintrag – neuer Status wurde eingefügt.', 'info') else: flash('Fehler beim Aktualisieren des Tookan-Kontos!', 'warning') # Conditionally create Planday account if requested if planday_checked: if not planday_id: update_data['planday_id'] = handle_account_creation( create_planday_acc, update_data['first_name'], update_data['last_name'], update_data['email'], update_data['region'], update_data['phone'], update_data['prefix'], update_data['address'], update_data['postal_code'], update_data['location'], update_data['gender'], update_data['country'], update_data['birth_date'] ) # If account creation succeeded, notify the team via Telegram if update_data['planday_id']: insert_external_acc_status( user_id, 'planday', 1 ) log_metadata['planday_acc'] = "Erstellt" if update_data['planday_id']: asyncio.run( send_msg_telegram( lara_bot_id, inform_jan_bc_user_new_planday(update_data['user_name'], update_data['approval']), onduty_recruitment ) ) planday_acc = True else: flash('Fehler beim Erstellen des Planday Accounts!', 'warning') else: # If Planday account already exists, update its status status = reactivate_planday_acc(planday_id, update_data['region']) if status: update_external_acc_status(user_id, 'planday', 1) log_metadata['planday_acc'] = "Reaktiviert" else: flash('Fehler beim Aktualisieren des Planday Accounts!', 'warning') if update_data['status_application'] == "11" and update_data.get('planday_id') is not None and update_data['able_to_drive'] == "1": update_data['able_to_drive'] = 0 move_to_11KA = put_driver_in_11KA_in_planday(update_data['planday_id']) flash_message(move_to_11KA, 'Mitarbeiter auf 11KA verschoben', 'Fehler bei 11KA verschiebung') for field in ['planday_id_checkbox', 'tookan_id_checkbox']: update_data.pop(field, None) # Prepare SQL update string and values for execution update_set = ", ".join([f"{key} = %s" for key, val in update_data.items()]) values = [val for val in update_data.values()] update_data['user_id'] = request.form.get('user_id', None) update_data_old = get_user_data_for_edit_as_string(update_data['user_id']) old_user = update_data_old['user_data'][0] # first row = old user record # check if address differs if "planday_id" in update_data and update_data['planday_id'] is not None and address_changed(old_user, update_data): ok = apply_planday_address_update(update_data.get('planday_id'), update_data) flash_message(ok, 'Adresse in Planday aktualisiert', 'Fehler beim Aktualisieren der Adresse in Planday') if old_user['able_to_drive'] == 0 and update_data['able_to_drive'] == 1: if "planday_id" in update_data and update_data['planday_id'] is not None: ok = reset_planday_name(update_data['planday_id'], update_data['first_name'], update_data['last_name']) flash_message(ok, '(DNF) in Planday wurde entfernt', 'Fehler beim entfernen von (DNF) in Planday') final_update_old_data = add_current_files(update_data_old) # Extract child data and update child records update_child_data = extract_children_data(update_child_predata) childres_operations = update_childrens(update_child_data, update_data['user_id']) # Update partner records if partner data exists if update_partner_predata: partner_operations = update_partner(update_partner_predata, update_data['user_id']) # Execute the user data update in the database update_user(update_set, values, request.form.get('user_id', None)) # Upload files to Google Drive and save document paths url_name = upload_files_to_google_drive(file_paths, str(request.form.get('user_id', None))) if "planday_id" in update_data and update_data['planday_id'] is not None: check_if_docs_are_valid_and_update_planday_name(update_data['planday_id'], url_name, update_data['first_name'], update_data['last_name']) insert_users_doc(url_name, str(request.form.get('user_id', None))) # Update file list with newly uploaded file URLs file_list = str_to_list(file_list) file_list.extend(url_name) # Combine update_data + log_metadata for logging log_data_combined = {**update_data, **log_metadata} # Log the final update data structure for future reference final_update_data = convert_to_log_structure( log_data_combined, update_child_data, childres_operations, update_partner_predata, partner_operations, file_list ) log( request.form.get('user_id', None), final_update_old_data, final_update_data, "update", session['username'], 'employee' ) # Clean up temporary files after uploading sel = remove_files(file_paths) # Flash success message flash('Erfolgreich editiert!', 'success') # Provide feedback on account creation results if tookan_acc: flash_message(update_data['tookan_id'] is not None, 'Tookan Account erstellt!', 'Fehler bei Erstellung Tookan Account!') if planday_acc: flash_message(update_data['planday_id'] is not None, 'Planday Account erstellt!', 'Fehler bei Erstellung Planday Account!') # Handle different submit actions if request.form['submitAction'] == 'close': return render_template('index.html', table=1, username=session['username']) elif request.form['submitAction'] == 'edit': user_id = request.form.get('user_id') return redirect(url_for('edit_form', user_id=user_id, username=session['username'])) except Exception as e: # Capture full traceback for error reporting full_traceback = traceback.format_exc() # Notify IT via Telegram about the exception asyncio.run(send_msg_telegram(lara_bot_id, f"Database query failed: {full_traceback} | Talentsphere", onduty_it)) # Flash error message to the user flash('Datenbankfehler! Bitte IT Team melden.', 'danger') # Redirect to main page return render_template('index.html', username=session['username']) elif request.form['submitAction']: # If there's an action but no POST data, redirect to edit form user_id = request.form.get('user_id') return redirect(url_for('edit_form', user_id=user_id, username=session['username'])) # If user is not logged in, redirect to login flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/delete_gdrive_file', methods=['POST']) def delete_gdrive_file(): """ Handles requests to delete or mark a Google Drive file as deleted. This function renames the specified file to indicate deletion and logs the change. Process: - Retrieves file details (file ID, file name, user ID) from the form data. - Checks if the file ID is provided; if not, returns an error response. - Creates a new name for the file, appending "_DELETED" to the original name. - Calls a helper function to rename the file on Google Drive. - Updates the log to reflect the file deletion and maintains a record of all prior files. Args: - None (relies on POST request data) Returns: - JSON response indicating success or failure. - If the file ID is missing, returns a 400 status with an error message. - Logs the deletion action for auditing and tracking purposes. Raises: - None (handled via return statements with JSON error responses). Example Usage: - Submitting a POST request to '/delete_gdrive_file' with valid file data will mark the specified Google Drive file as deleted, log the action, and return a success response. Notes: - This function assumes access to helper functions for file renaming, logging, and data retrieval. - Error handling includes clear user feedback for missing file IDs or failed rename actions. """ # Retrieve form data: file ID, file name, and user ID file_id = request.form.get('file_id', None) file_name = request.form.get('file_name', None) user_id = request.form.get('user_id', None) # Separate file base name and extension base_name, extension = os.path.splitext(file_name) # Fetch current user data for logging the previous state update_data_old = get_user_data_for_edit_as_string(user_id) final_update_old_data = add_current_files(update_data_old) # Check if file ID is provided; if missing, return error response if not file_id: return jsonify({'status': 'error', 'message': 'File ID missing'}), 400 # Construct a new file name with "_DELETED" appended before the file extension if extension: new_name = f"{base_name}_DELETED{extension}" else: new_name = f"{file_name}_DELETED" # Attempt to rename the file on Google Drive, marking it as deleted status = rename_file(file_id, new_name) # Update the log to reflect the deletion and remove the file reference from the current data final_update_data = remove_files_by_file_id(final_update_old_data, file_id) log(user_id, final_update_old_data, final_update_data, "update", session['username'], 'employee') # Return a JSON response indicating success or failure of the file rename operation if status: return jsonify({'status': 'success', 'message': 'File deleted'}) else: return jsonify({'status': 'error', 'message': 'File could not be deleted'}) def flash_message(condition, success_message, error_message): """ Flashes a success or warning message based on the evaluation of a condition. This helper function simplifies the process of conditionally displaying flash messages in a web application. If the condition is true, a success message is flashed. If the condition is false, a warning message is flashed instead. Args: condition (bool): The condition to evaluate. If True, the success_message is flashed; if False, the error_message is flashed. success_message (str): The message to display if the condition is True. This message will be flashed with a 'success' category, typically displayed in green. error_message (str): The message to display if the condition is False. This message will be flashed with a 'warning' category, typically displayed in yellow or orange. Returns: None: This function returns None. It only triggers a side effect of calling `flash`. Example Usage: - flash_message(user.is_active, "User activated successfully!", "User activation failed!") This will flash "User activated successfully!" if user.is_active is True, and "User activation failed!" if it is False. Note: - The function assumes that the `flash` function from Flask is already imported and available in the scope where `flash_message` is used. """ if condition: flash(success_message, 'success') else: flash(error_message, 'warning') def get_form_flag(flag_name): """ Checks if a specific flag in a submitted form is set to 'on'. This utility function is commonly used to verify the state of checkboxes or switches in a web form, where such inputs might either be marked as 'on' or not included in the form submission if unchecked. Args: flag_name (str): The name of the form element (typically a checkbox) to check. Returns: bool: Returns True if the form element is found in the request and is set to 'on', otherwise False. Example Usage: - if get_form_flag('newsletter_subscribe'): # Execute some action if the 'newsletter_subscribe' checkbox is checked. Note: - This function is designed to be used within a Flask route where `request` is already imported and contains form data from a client POST request. """ return request.form.get(flag_name, False) == "on" def handle_account_creation(create_account_func, *args): """ Executes a given account creation function with the specified arguments. This function is designed to abstract the account creation process, allowing for flexibility in the account creation logic by passing different functions as arguments. It directly passes any received arguments to the account creation function and returns the result. Args: create_account_func (callable): A function intended to handle the creation of an account. It should accept any number of arguments as required for the account creation. *args: Variable length argument list to be passed to the create_account_func. Returns: The return value from the create_account_func, which could be any data type based on the implementation of the function (e.g., boolean indicating success/failure, a new account object, etc.). Example Usage: - result = handle_account_creation(create_user_account, 'john.doe@example.com', 'password123') This would attempt to create a user account with the provided email and password. Note: - This function does not handle exceptions that may be raised by create_account_func. It is recommended to implement error handling within the passed function or at the point where handle_account_creation is called. """ val = create_account_func(*args) return val # Account creation was attempted if __name__ == '__main__': app.run()