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 from flask_session import Session import time as t 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 werkzeug.utils import secure_filename import json from models import ( get_login_user, get_company_view_data, get_company_data_for_edit, get_data_from_fk_table, update_company, get_insert_description, update_contact_person, insert_contact_person, insert_company_and_get_id, update_google_url, get_company_by_id_for_log, insert_audit_log, get_logs_from_db, get_customer_interactions_view_data, get_customer_interactions_for_edit, update_customer_interactions, get_insert_description_customer_interactions, insert_customer_interactions_and_get_id, get_customer_contact_person_with_company_name, get_customer_interactions_by_id_for_log ) from standalone import ( send_msg_telegram, allowed_file, compare_changes, upload_files_to_google_drive, list_files_in_folder, rename_file, get_country_dialing_codes, create_google_folder, get_company_new_data_from_form, get_change_diff, get_customer_interactions_new_data_from_form, get_change_diff_customer_interaction, parse_int ) app = Flask(__name__, static_url_path='/crm/static') app.secret_key = "asndijnsau" app.config['SESSION_COOKIE_SECURE'] = False #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) 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://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=()" ) return response @app.route('/login', methods=['GET', 'POST']) def login(): """ Handles login requests using GET and POST methods with rate limiting for failed attempts. """ MAX_ATTEMPTS = 3 # Maximum number of allowed attempts TIMEOUT_SECONDS = 60 # Timeout duration in seconds # Initialize session data for failed attempts if not present if 'failed_attempts' not in session: session['failed_attempts'] = 0 session['last_attempt_time'] = None if request.method == 'POST': login_email = request.form.get('login', '').lower() login_password = request.form.get('password') if not login_email or not login_password: flash('Bitte alle Felder ausfüllen!', 'warning') return render_template('login.html') # Check if the user is within timeout period now = time.time() if session['failed_attempts'] >= MAX_ATTEMPTS: if session['last_attempt_time'] and now - session['last_attempt_time'] < TIMEOUT_SECONDS: remaining_time = TIMEOUT_SECONDS - (now - session['last_attempt_time']) flash(f"Zu viele Versuche. Bitte warten Sie {int(remaining_time)} Sekunden.", 'danger') return render_template('login.html') else: # Reset failed attempts after timeout session['failed_attempts'] = 0 session['last_attempt_time'] = None try: result = get_login_user(login_email) if result and bcrypt.checkpw(login_password.encode('utf-8'), result[2].encode('utf-8')): session.clear() session['username'] = result[0] session['admin'] = 1 session['valid'] = True print("xes") return redirect('/crm/') else: # Increment failed attempts and record the time of this attempt session['failed_attempts'] += 1 session['last_attempt_time'] = now flash('Falscher Benutzername oder Passwort!', 'warning') return render_template('login.html') except Exception as e: print(f"Login Fail: {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') return render_template('login.html') else: return render_template('login.html') @app.route('/logout') def logout(): """ Logs out the current user by clearing the session and redirecting to the login page. """ session.clear() flash('Erfolgreich ausgeloggt.', 'success') return redirect('/crm/login') @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. """ print(session) if 'valid' in session and session['valid'] is True: #status = request.form.get('status') status = request.form.get('site_name') or request.args.get('site_name') return render_template('index.html',table = status, username = session['username'], site_name = status) return render_template('login.html') @app.route('/customer_relations_view', methods = ['POST', 'GET']) def customer_interactions_view(): if 'valid' in session and session['valid'] is True: try: column_names, rows = get_customer_interactions_view_data() return jsonify({'columns': column_names, 'data': rows}) except Exception: full_traceback = traceback.format_exc() print(full_traceback) 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('/company_view', methods = ['POST', 'GET']) def company_view(): if 'valid' in session and session['valid'] is True: try: column_names, rows = get_company_view_data() print(column_names, rows) return jsonify({'columns': column_names, 'data': rows}) except Exception: full_traceback = traceback.format_exc() print(full_traceback) 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('/edit_form/company/', methods=['GET', 'POST']) def company_edit_form(row_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} company_column_name_and_value, company_contact_person_column_name_and_value = get_company_data_for_edit(row_id) industry = get_data_from_fk_table('company_industry',1,False) files = list_files_in_folder(row_id) country_dial_codes, country_names = get_country_dialing_codes() company_status = get_data_from_fk_table('company_status', 1, False) return render_template('edit.html', company_status = company_status ,country_names = country_names ,company_data = company_column_name_and_value,company_contact_person_data = company_contact_person_column_name_and_value, files = files, industry = industry, translations = translations, row_id = row_id, prefix = country_dial_codes,prefix_json=json.dumps(country_dial_codes), site_name = "0", username = session['username']) except: full_traceback = traceback.format_exc() print(full_traceback) 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_form/customer_interactions/', methods=['GET', 'POST']) def customer_interactions_edit_form(row_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} customer_interactions_column_name_and_value = get_customer_interactions_for_edit(row_id) communication_method = get_data_from_fk_table('communication_method',1,False) company = get_data_from_fk_table('company',1,False) company = sorted(company, key=lambda x: x['description'].lower()) company_contact_person = get_customer_contact_person_with_company_name(customer_interactions_column_name_and_value[0]['company']) return render_template('interaction_edit.html',company_contact_person = company_contact_person, company = company ,customer_interactions_column_name_and_value = customer_interactions_column_name_and_value, communication_method = communication_method, translations = translations, row_id = row_id, site_name = "1", username = session['username']) except: full_traceback = traceback.format_exc() asyncio.run(send_msg_telegram(lara_bot_id,f"Database query failed: {full_traceback} | CRM",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/customer_interactions', methods=['POST']) def customer_interactions_edit_data(): if 'valid' in session and session['valid']: try: customer_interactions_id = request.form.get('row_id') # Firma updaten update_data = { key: (None if val in ['None', ''] else val) for key, val in request.form.items() if key not in ('row_id', 'id', 'submitAction') and not key.startswith('contact_') and not key.startswith('files') } update_set = ", ".join([f"{key} = %s" for key in update_data]) values = list(update_data.values()) customer_interactions_old_data = get_customer_interactions_by_id_for_log(customer_interactions_id) customer_interactions_new_data = get_customer_interactions_new_data_from_form(request) changes = get_change_diff_customer_interaction(customer_interactions_old_data, customer_interactions_new_data) insert_audit_log("customer_interactions", customer_interactions_id, session['username'], changes) update_customer_interactions(update_set, values, customer_interactions_id) flash("Erfolgreich editiert!", "success") if request.form['submitAction'] == 'close': return render_template("index.html", site_name = "1", username=session['username']) else: return redirect(url_for("customer_interactions_edit_form", row_id=customer_interactions_id)) except Exception: full_traceback = traceback.format_exc() asyncio.run(send_msg_telegram(lara_bot_id,f"Database query failed: {full_traceback} | CRM",onduty_it)) flash("Datenbankfehler! Bitte IT Team melden.", "danger") return render_template("index.html", username=session['username']) flash("Bitte einloggen!", "warning") return render_template("login.html") @app.route('/edit_data/company', methods=['POST']) def company_edit_data(): if 'valid' in session and session['valid']: try: company_id = request.form.get('row_id') # Firma updaten update_data = { key: (None if val in ['None', ''] else val) for key, val in request.form.items() if key not in ('row_id', 'id', 'submitAction') and not key.startswith('contact_') and not key.startswith('files') } old_company = get_company_by_id_for_log(company_id) new_company = get_company_new_data_from_form(request) changes = get_change_diff(old_company, new_company) insert_audit_log("company", company_id, session['username'], changes) update_set = ", ".join([f"{key} = %s" for key in update_data]) values = list(update_data.values()) update_company(update_set, values, company_id) # Kontaktpersonen updaten/hinzufügen contact_persons = [] i = 1 while request.form.get(f'contact_name_{i}'): contact = { 'id': request.form.get(f'contact_id_{i}'), 'name': request.form.get(f'contact_name_{i}'), 'email': request.form.get(f'contact_email_{i}'), 'phone': request.form.get(f'contact_phone_{i}'), 'job_title': request.form.get(f'contact_job_title_{i}'), 'prefix': request.form.get(f'contact_prefix_{i}'), 'company_id': company_id } contact_persons.append(contact) i += 1 for person in contact_persons: if person.get('id') and str(person.get('id')).strip().isdigit(): fields = ['name', 'email', 'prefix' ,'phone', 'job_title'] update = ", ".join([f"{k} = %s" for k in fields]) values = [person[k] for k in fields] update_contact_person(update, values, person['id']) else: insert_contact_person(person) flash("Erfolgreich editiert!", "success") if request.form['submitAction'] == 'close': return render_template("index.html", site_name = "0", username=session['username']) else: return redirect(url_for("company_edit_form", row_id=company_id)) except Exception: full_traceback = traceback.format_exc() asyncio.run(send_msg_telegram(lara_bot_id,f"Database query failed: {full_traceback} | CRM",onduty_it)) flash("Datenbankfehler! Bitte IT Team melden.", "danger") return render_template("index.html", username=session['username']) flash("Bitte einloggen!", "warning") return render_template("login.html") @app.route('/insert_form/company', methods=['GET', 'POST']) def company_insert_form(): """ Zeigt ein Formular zum Erfassen einer neuen Firma mit Kontaktpersonen an. Nur für eingeloggte Benutzer zugänglich. - Lädt Feldübersetzungen. - Holt Branchen (FK-Daten) für Dropdown. - Gibt alle Daten an das HTML-Template weiter. """ if 'valid' in session and session['valid'] is True: try: # Holt Spaltennamen für die Firma (z. B. aus `SHOW COLUMNS`) column_names = get_insert_description() # z. B. [{'name': 'name'}, {'email': 'email'}, ...] # Holt Übersetzungen für Labels (deutsche Bezeichnungen) translations = {item['db']: item['de'] for item in translation} # Holt Branchenliste aus FK-Tabelle industry = get_data_from_fk_table('company_industry', 1, False) status = get_data_from_fk_table('company_status', 1, False) # Holt Prefix-Liste für Telefonnummern country_dial_codes, country_names = get_country_dialing_codes() # z. B. [{'dial_code': '+41'}, {'dial_code': '+49'}, ...] return render_template( 'insert.html', status = status, country_names=country_names, industry=industry, site_name = "0", translations=translations, column_names=column_names, prefix=country_dial_codes, prefix_json=json.dumps(country_dial_codes), username=session['username'] ) except Exception: full_traceback = traceback.format_exc() print(full_traceback) asyncio.run(send_msg_telegram(lara_bot_id,f"Database query failed: {full_traceback} | CRM",onduty_it)) flash('Datenbankfehler! Bitte IT Team melden.', 'danger') return render_template('index.html', username=session['username']) flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/insert_form/costumer_interactioons', methods=['GET', 'POST']) def customer_interactions_insert_form(): """ Zeigt ein Formular zum Erfassen einer neuen Firma mit Kontaktpersonen an. Nur für eingeloggte Benutzer zugänglich. - Lädt Feldübersetzungen. - Holt Branchen (FK-Daten) für Dropdown. - Gibt alle Daten an das HTML-Template weiter. """ if 'valid' in session and session['valid'] is True: try: # Holt Spaltennamen für die Firma (z. B. aus `SHOW COLUMNS`) column_names = get_insert_description_customer_interactions() # z. B. [{'name': 'name'}, {'email': 'email'}, ...] # Holt Übersetzungen für Labels (deutsche Bezeichnungen) translations = {item['db']: item['de'] for item in translation} communication_method = get_data_from_fk_table('communication_method',1,False) company = get_data_from_fk_table('company',1,False) return render_template( 'interaction_insert.html', communication_method = communication_method, company=company, translations=translations, column_names=column_names, username=session['username'] ) except Exception: full_traceback = traceback.format_exc() print(full_traceback) asyncio.run(send_msg_telegram(lara_bot_id,f"Database query failed: {full_traceback} | CRM",onduty_it)) flash('Datenbankfehler! Bitte IT Team melden.', 'danger') return render_template('index.html', username=session['username']) flash('Bitte einloggen!', 'warning') return render_template('login.html') @app.route('/insert_data/customer_interactions', methods=['GET', 'POST']) def customer_interactions_insert_data(): if 'valid' in session and session['valid']: try: customer_interactions_data = { 'date': request.form.get('date'), 'follow_up': request.form.get('follow_up'), 'communicated_over': parse_int(request.form.get('communicated_over')), 'company': parse_int(request.form.get('company')), 'company_user': parse_int(request.form.get('company_user')), 'note': request.form.get('note'), } # Firma einfügen und neue company_id holen (du musst diese Funktion selbst implementieren) company_id = insert_customer_interactions_and_get_id(customer_interactions_data) flash("Kunden Interaktion erfolgreich erstellt!", "success") if request.form['submitAction'] == 'close': return render_template("index.html", site_name = "1", username=session['username']) else: return redirect(url_for("customer_interactions_edit_form", row_id=company_id)) except Exception: full_traceback = traceback.format_exc() asyncio.run(send_msg_telegram(lara_bot_id,f"Database query failed: {full_traceback} | CRM",onduty_it)) flash("Datenbankfehler! Bitte IT Team melden.", "danger") return render_template("index.html", username=session['username']) @app.route('/insert_data/company', methods=['GET', 'POST']) def company_insert_data(): if 'valid' in session and session['valid']: try: industry_value = request.form.get('industry') if not industry_value or industry_value.lower() == 'none': industry_value = None else: industry_value = int(industry_value) company_data = { 'name': request.form.get('name'), 'email': request.form.get('email'), 'prefix': request.form.get('prefix'), 'phone': request.form.get('phone'), 'address': request.form.get('address'), 'postal_code': request.form.get('postal_code'), 'industry': industry_value, } # Firma einfügen und neue company_id holen (du musst diese Funktion selbst implementieren) company_id = insert_company_and_get_id(company_data) # Kontaktpersonen sammeln contact_persons = [] i = 1 while request.form.get(f'contact_name_{i}'): person = { 'name': request.form.get(f'contact_name_{i}'), 'email': request.form.get(f'contact_email_{i}'), 'prefix': request.form.get(f'contact_prefix_{i}'), 'phone': request.form.get(f'contact_phone_{i}'), 'job_title': request.form.get(f'contact_job_title_{i}'), 'company_id': company_id } contact_persons.append(person) i += 1 # In die DB einfügen (du brauchst insert_contact_person) for person in contact_persons: insert_contact_person(person) folder_id = handle_account_creation(create_google_folder, company_id) update_google_url(folder_id,company_id) flash("Firma erfolgreich erstellt!", "success") if request.form['submitAction'] == 'close': return render_template("index.html", site_name = "0", username=session['username']) else: return redirect(url_for("company_edit_form", row_id=company_id)) except Exception: full_traceback = traceback.format_exc() asyncio.run(send_msg_telegram(lara_bot_id,f"Database query failed: {full_traceback} | CRM",onduty_it)) flash("Datenbankfehler! Bitte IT Team melden.", "danger") return render_template("index.html", username=session['username']) flash("Bitte einloggen!", "warning") return render_template("login.html") @app.route('/logs//', methods=['GET', 'POST']) def get_logs(table, row_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: translations = {item['db']: item['de'] for item in translation} # Translation mapping for column headers logs = get_logs_from_db(table, row_id) name = logs[0]['name'] if logs else 'Unbekannt' # Render the logs view template with the processed data, translations, and additional context return render_template( 'logs.html', table=table, name=name, data=logs, translations=translations, username=session['username'] ) except Exception: full_traceback = traceback.format_exc() asyncio.run(send_msg_telegram(lara_bot_id,f"Database query failed: {full_traceback} | CRM",onduty_it)) 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('/upload', methods=['POST']) def upload_files(): files = request.files.getlist('file') row_id = request.form.get('row_id') if not files: return jsonify({"status": "error", "message": "No files uploaded"}), 400 local_paths = [] os.makedirs('tmp', exist_ok=True) # Ensure tmp/ exists try: for file in files: if file and file.filename: filename = secure_filename(file.filename) local_path = os.path.join('tmp', filename) file.save(local_path) local_paths.append(local_path) # Upload all files at once upload_results = upload_files_to_google_drive(local_paths, folder_name=str(row_id)) return jsonify({"status": "success", "results": upload_results}) finally: # Clean up all saved local files for path in local_paths: if os.path.exists(path): os.remove(path) @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) row_id = request.form.get('row_id', None) # Separate file base name and extension base_name, extension = os.path.splitext(file_name) # 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) # 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'}) @app.route('/list_files/') def list_files_ajax(row_id): try: files = list_files_in_folder(row_id) translations = {item['db']: item['de'] for item in translation} return render_template('partials/file_table.html', files=files, translations=translations, row_id=row_id) except Exception as e: print("Fehler beim Abrufen der Dateien:", e) return "Fehler beim Abrufen der Dateien", 500 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()