From 7cab8ee653568995de223f82a887db3b97806145 Mon Sep 17 00:00:00 2001 From: Sebastian Serfling Date: Mon, 24 Mar 2025 10:00:01 +0100 Subject: [PATCH] add mysql_timestamp_check.py & .env --- .env | 10 ++ README.md | 2 +- mysql_timestamp_check.py | 264 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 275 insertions(+), 1 deletion(-) create mode 100644 .env create mode 100644 mysql_timestamp_check.py diff --git a/.env b/.env new file mode 100644 index 0000000..bbb8357 --- /dev/null +++ b/.env @@ -0,0 +1,10 @@ +MYSQL_HOST=172.17.1.21 +MYSQL_USER=root +MYSQL_PASSWORD=N53yBCswuawzBzS445VNAhWVMs3N59Gb9szEsrzXRBzarDqpdETpQeyt5v5CGe +MYSQL_DATABASE=Kunden +EMAIL_SENDER=sebastianserfling@stines.de +EMAIL_RECEIVER=sebastianserfling@stines.de +SMTP_SERVER=mail.stines.de +SMTP_PORT=587 +SMTP_USER=sebastianserfling@stines.de +SMTP_PASSWORD=c6tzJBxDtNEU84ZAWY39eG8RUM5XR4UyfrDesfgCekEbFhkHDkbn \ No newline at end of file diff --git a/README.md b/README.md index 8ba9730..8c77c91 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ Ein Python-Skript zur Überwachung von MySQL-Tabellen, um sicherzustellen, dass ## Funktionsweise -Das Skript prüft alle Tabellen, die mit dem Präfix "RAW." beginnen, und überprüft, ob der letzte Zeitstempel in der Spalte "add_date" vom Vortag (gestern) stammt. Falls eine oder mehrere Tabellen nicht aktuell sind, wird eine Benachrichtigungs-E-Mail gesendet. +Das Skript prüft alle Tabellen, die "RAW." im Namen enthalten, und überprüft, ob irgendein Zeitstempel in der Spalte "add_date" vom Vortag (gestern) existiert. Falls eine oder mehrere Tabellen keine Daten von gestern haben, wird eine Benachrichtigungs-E-Mail gesendet. ## Hauptfunktionen diff --git a/mysql_timestamp_check.py b/mysql_timestamp_check.py new file mode 100644 index 0000000..b843610 --- /dev/null +++ b/mysql_timestamp_check.py @@ -0,0 +1,264 @@ +#!/usr/bin/env python3 +import mysql.connector +import datetime +import smtplib +from email.mime.text import MIMEText +from email.mime.multipart import MIMEMultipart +import argparse +import sys +import os +from dotenv import load_dotenv + + +def get_tables_from_db(connection): + """ + Holt die Liste der zu überprüfenden Tabellen, die 'RAW.' im Namen enthalten + + Returns: + list: Liste der Tabellennamen + """ + try: + cursor = connection.cursor() + + # Holen aller Tabellen in der Kunden-Datenbank + cursor.execute("SHOW TABLES FROM Kunden") + tables = [] + + for (table,) in cursor.fetchall(): + if 'RAW.' in table: + # Tabellennamen mit Backticks versehen + tables.append(f"Kunden.`{table}`") + + cursor.close() + + if not tables: + print("Warnung: Keine Tabellen gefunden, die 'RAW.' enthalten.") + else: + print(f"Gefundene zu überprüfende Tabellen: {', '.join(tables)}") + + return tables + except mysql.connector.Error as err: + print(f"Fehler beim Abrufen der Tabellenliste: {err}") + sys.exit(1) + + +def check_timestamp(connection, tables, timestamp_column, email_config, test_mode=False): + """ + Überprüft, ob der letzte Zeitstempel in jeder angegebenen Tabelle von gestern ist. + Sendet eine E-Mail mit den Namen der Tabellen, die nicht aktuell sind oder + gibt die Nachricht auf der Konsole aus, wenn test_mode=True. + """ + cursor = connection.cursor() + + # Berechne das Datum für gestern + yesterday = datetime.date.today() - datetime.timedelta(days=1) + yesterday_start = datetime.datetime.combine(yesterday, datetime.time.min) + yesterday_end = datetime.datetime.combine(yesterday, datetime.time.max) + + # Liste der Tabellen, die nicht aktuell sind + outdated_tables = [] + database = os.getenv("MYSQL_DATABASE", "unbekannt") + + # Jede Tabelle überprüfen + for table in tables: + try: + # SQL-Abfrage, um zu prüfen, ob es Zeitstempel von gestern gibt + query = f""" + SELECT 1 + FROM {table} + WHERE {timestamp_column} BETWEEN %s AND %s + LIMIT 1 + """ + cursor.execute(query, (yesterday_start, yesterday_end)) + result = cursor.fetchone() + + # Überprüfen, ob ein Ergebnis vorhanden ist (Daten von gestern existieren) + if result is None: + print(f"Tabelle {table}: KEINE Zeitstempel von gestern gefunden.") + outdated_tables.append(f"{table} (keine Daten von gestern)") + else: + print(f"Tabelle {table}: Zeitstempel von gestern gefunden - OK.") + + except mysql.connector.Error as err: + print(f"Fehler beim Überprüfen der Tabelle {table}: {err}") + outdated_tables.append(f"{table} (Fehler: {err}") + + # Cursor schließen + cursor.close() + + # Wenn es nicht aktuelle Tabellen gibt, sende eine E-Mail oder gib Meldung aus + if outdated_tables: + if test_mode: + print("\n--- TEST MODUS: E-Mail würde gesendet werden ---") + print(f"Betreff: Warnung: Nicht aktuelle Tabellen in Datenbank {database}") + print("Inhalt:") + print("Hallo,") + print( + f"\ndie folgenden Tabellen in der Datenbank '{database}' haben keinen aktuellen Zeitstempel von gestern:") + for table in outdated_tables: + print(f"- {table}") + print("\nBitte überprüfen Sie diese Tabellen.") + print("\nDies ist eine automatisch generierte Nachricht.") + print("--- ENDE TEST MODUS ---\n") + else: + send_email(outdated_tables, database, email_config) + return False + else: + print("Alle Tabellen sind aktuell.") + return True + + +def send_email(outdated_tables, database, email_config): + """ + Sendet eine E-Mail mit der Liste der nicht aktuellen Tabellen. + """ + sender_email = email_config["sender"] + receiver_email = email_config["receiver"] + smtp_server = email_config["smtp_server"] + smtp_port = email_config["smtp_port"] + smtp_user = email_config.get("username", sender_email) + smtp_password = email_config.get("password", "") + + # E-Mail erstellen + message = MIMEMultipart() + message["From"] = sender_email + message["To"] = receiver_email + message["Subject"] = f"Warnung: Nicht aktuelle Tabellen in Datenbank {database}" + + # E-Mail-Inhalt + body = f""" + Hallo, + + die folgenden Tabellen in der Datenbank '{database}' haben keinen aktuellen Zeitstempel von gestern: + + {chr(10).join('- ' + table for table in outdated_tables)} + + Bitte überprüfen Sie diese Tabellen. + + Dies ist eine automatisch generierte Nachricht. + """ + + message.attach(MIMEText(body, "plain")) + + try: + # Verbindung zum SMTP-Server herstellen + server = smtplib.SMTP(smtp_server, smtp_port) + server.starttls() # TLS-Verschlüsselung aktivieren + + # Anmelden (falls erforderlich) + if smtp_password: + server.login(smtp_user, smtp_password) + + # E-Mail senden + server.send_message(message) + server.quit() + + print(f"E-Mail an {receiver_email} gesendet.") + except Exception as e: + print(f"Fehler beim Senden der E-Mail: {e}") + + +def main(): + # .env-Datei laden + load_dotenv() + + parser = argparse.ArgumentParser(description="Überprüft MySQL-Tabellen auf aktuelle Zeitstempel.") + + # Optionale DB-Parameter (werden sonst aus .env gelesen) + parser.add_argument("--host", help="MySQL-Hostadresse (Standard: aus .env MYSQL_HOST)") + parser.add_argument("--user", help="MySQL-Benutzername (Standard: aus .env MYSQL_USER)") + parser.add_argument("--password", help="MySQL-Passwort (Standard: aus .env MYSQL_PASSWORD)") + parser.add_argument("--database", help="MySQL-Datenbankname (Standard: aus .env MYSQL_DATABASE)") + + # Parameter für Zeitstempel-Spalte (mit add_date als Standard) + parser.add_argument("--timestamp-column", default="add_date", + help="Name der Zeitstempel-Spalte (Standard: add_date)") + + # Optionale Parameter + parser.add_argument("--test", action="store_true", + help="Testmodus: Keine E-Mail senden, nur Ausgabe auf der Konsole") + + # E-Mail-Parameter + email_group = parser.add_argument_group("E-Mail-Konfiguration", + "Diese Parameter werden nicht benötigt, wenn --test verwendet wird") + email_group.add_argument("--email-sender", help="Absender-E-Mail-Adresse (Standard: aus .env EMAIL_SENDER)") + email_group.add_argument("--email-receiver", help="Empfänger-E-Mail-Adresse (Standard: aus .env EMAIL_RECEIVER)") + email_group.add_argument("--smtp-server", help="SMTP-Server-Adresse (Standard: aus .env SMTP_SERVER)") + email_group.add_argument("--smtp-port", type=int, help="SMTP-Server-Port (Standard: aus .env SMTP_PORT)") + email_group.add_argument("--smtp-user", help="SMTP-Benutzername (Standard: aus .env SMTP_USER)") + email_group.add_argument("--smtp-password", help="SMTP-Passwort (Standard: aus .env SMTP_PASSWORD)") + + args = parser.parse_args() + + # Datenbankverbindungsparameter aus .env oder Kommandozeile + host = args.host or os.getenv("MYSQL_HOST", "localhost") + user = args.user or os.getenv("MYSQL_USER") + password = args.password or os.getenv("MYSQL_PASSWORD") + database = args.database or os.getenv("MYSQL_DATABASE") + + # Prüfe, ob erforderliche DB-Parameter vorhanden sind + if not user or not password or not database: + print( + "Fehler: MySQL-Benutzer, Passwort und Datenbankname müssen entweder in der .env-Datei oder als Parameter angegeben werden!") + sys.exit(1) + + # Verbindung zur Datenbank herstellen + try: + connection = mysql.connector.connect( + host=host, + user=user, + password=password, + database=database + ) + except mysql.connector.Error as err: + print(f"Fehler bei der Datenbankverbindung: {err}") + sys.exit(1) + + # Testmodus aktiviert? + test_mode = args.test + + # E-Mail-Konfiguration + email_config = { + "sender": args.email_sender or os.getenv("EMAIL_SENDER"), + "receiver": args.email_receiver or os.getenv("EMAIL_RECEIVER"), + "smtp_server": args.smtp_server or os.getenv("SMTP_SERVER", "localhost"), + "smtp_port": args.smtp_port or int(os.getenv("SMTP_PORT", "587")), + "username": args.smtp_user or os.getenv("SMTP_USER"), + "password": args.smtp_password or os.getenv("SMTP_PASSWORD", "") + } + + # Wenn nicht im Testmodus, prüfe ob E-Mail-Parameter angegeben wurden + if not test_mode: + if not email_config["sender"] or not email_config["receiver"]: + print( + "Fehler: E-Mail-Absender und -Empfänger müssen entweder in der .env-Datei oder als Parameter angegeben werden!") + connection.close() + sys.exit(1) + + # Tabellen aus der Datenbanktabelle holen + tables = get_tables_from_db(connection) + + if not tables: + print("Keine zu überprüfenden Tabellen gefunden. Programm wird beendet.") + connection.close() + sys.exit(0) + + # Zeitstempel-Spalte (Standard: add_date) + timestamp_column = args.timestamp_column + print(f"Überprüfe Zeitstempel in Spalte: {timestamp_column}") + + # Timestamp-Überprüfung durchführen + check_timestamp( + connection, + tables, + timestamp_column, + email_config, + test_mode + ) + + # Verbindung schließen + connection.close() + + +if __name__ == "__main__": + main() \ No newline at end of file