import logging import sqlite3 import os import yaml from pathlib import Path from telegram import Update from telegram.ext import Application, MessageHandler, filters, ContextTypes, CommandHandler from datetime import datetime # --- KONFIGURATION LADEN --- def load_config(config_path: str = 'config.yaml') -> dict: """Lädt die Konfiguration aus einer YAML-Datei.""" if not os.path.exists(config_path): raise FileNotFoundError( f"Konfigurationsdatei '{config_path}' nicht gefunden.\n" f"Bitte kopiere 'config_example.yaml' zu '{config_path}' und passe die Werte an." ) with open(config_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) return config # Config laden try: config = load_config('config.yaml') except FileNotFoundError as e: print(f"❌ Fehler: {e}") exit(1) # Konfigurationsvariablen extrahieren BOT_TOKEN = config['bot']['token'] ENABLE_DEBUG_MODE = config['bot']['debug_mode'] DATABASE_NAME = config['database']['name'] WRITE_ACTIVITY_TO_FILE = config['logging']['write_to_file'] LOGGING_FILE = config['logging']['file_name'] LOG_LEVEL = getattr(logging, config['logging']['level'], logging.INFO) # Forum Topics: Ausnahlmeliste (nicht zu reopening) NOT_REOPEN_TOPICS = config.get('forum_topics', {}).get('not_reopen_topics', []) # Logging konfigurieren logging.basicConfig(level=LOG_LEVEL, format='%(asctime)s - %(levelname)s - %(message)s') class DatabaseManager: """Kapselt alle Datenbankinteraktionen und verwaltet Benutzer-/Gruppenzustände.""" def __init__(self): # ACHTUNG: Führen Sie vor dem Start einmal manuell die Migration durch, # oder lassen Sie Python das Schema anpassen (siehe unten). self.conn = sqlite3.connect(DATABASE_NAME) self.cursor = self.conn.cursor() self._create_tables() def _create_tables(self): """Stellt sicher, dass alle notwendigen Tabellen und Spalten existieren.""" # 1. activity_log (Bleibt gleich) self.cursor.execute(""" CREATE TABLE IF NOT EXISTS activity_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, group_chat_id TEXT NOT NULL, topic_id TEXT DEFAULT 'MAIN', timestamp TEXT NOT NULL, activity_type TEXT NOT NULL, user_id INTEGER NOT NULL, username TEXT, details TEXT ) """) # 2. user_profile (Refaktoriert: first_name, last_name, is_bot, kein message_count) self.cursor.execute(""" CREATE TABLE IF NOT EXISTS user_profile ( user_id INTEGER PRIMARY KEY, first_name TEXT, last_name TEXT, username TEXT, is_bot INTEGER DEFAULT 0, lastseen_timestamp TEXT NULL ) """) # 3. NEUE TABELLE: Profile History Log (Protokolliert Änderungen) self.cursor.execute(""" CREATE TABLE IF NOT EXISTS profile_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, group_chat_id TEXT NOT NULL, timestamp TEXT NOT NULL, activity_type TEXT NOT NULL, old_full_name TEXT, -- Name vor der Änderung new_full_name TEXT, -- Neuer Name old_username TEXT, -- Username vor der Änderung new_username TEXT, -- Neuer Username details TEXT ); """) print(f"✅ Datenbank '{DATABASE_NAME}' erfolgreich initialisiert/überprüft.") def log_activity(self, group_chat_id: int, topic_id: str, activity_type: str, user_id: int, username: str, details: str = None): """Speichert eine allgemeine Aktivität im Protokoll.""" timestamp = datetime.now().isoformat() try: self.cursor.execute(""" INSERT INTO activity_log (group_chat_id, topic_id, timestamp, activity_type, user_id, username, details) VALUES (?, ?, ?, ?, ?, ?, ?) """, (str(group_chat_id), topic_id, timestamp, activity_type, user_id, username or None, details)) self.conn.commit() except Exception as e: logging.error(f"Fehler beim Logging der Aktivität: {e}") # NEU: Funktion zur Speicherung von Profiländerungen im History Log def log_profile_change(self, group_chat_id: int, user_id: int, topic_id: str, old_name: str, new_name: str, old_user: str, new_user: str, details: str): """Speichert die detaillierte Historie einer Profiländerung.""" timestamp = datetime.now().isoformat() try: self.cursor.execute(""" INSERT INTO profile_history (user_id, group_chat_id, timestamp, activity_type, old_full_name, new_full_name, old_username, new_username, details) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, (user_id, str(group_chat_id), timestamp, "PROFILE_CHANGE", old_name, new_name, old_user, new_user, details)) self.conn.commit() except Exception as e: logging.error(f"Fehler beim Protokollieren der Profiländerung: {e}") def get_user_profile(self, user_id: int) -> dict: """Lädt das gespeicherte Profil eines Benutzers aus der Datenbank.""" try: self.cursor.execute(""" SELECT first_name, last_name, username, is_bot FROM user_profile WHERE user_id = ? """, (user_id,)) result = self.cursor.fetchone() if result: return { 'first_name': result[0], 'last_name': result[1], 'username': result[2], 'is_bot': result[3] } return None except Exception as e: logging.error(f"Fehler beim Laden des Benutzerprofils {user_id}: {e}") return None def update_user_profile(self, user_id: int, first_name: str, last_name: str, username: str, is_bot: bool = False): """Aktualisiert nur die Profilinformationen eines Benutzers (ohne Nachrichtenstatistiken). Erstellt den Benutzer automatisch, falls er noch nicht existiert. """ try: self.cursor.execute(""" INSERT INTO user_profile (user_id, first_name, last_name, username, is_bot, lastseen_timestamp) VALUES (?, ?, ?, ?, ?, NULL) ON CONFLICT(user_id) DO UPDATE SET first_name = excluded.first_name, last_name = excluded.last_name, username = excluded.username, is_bot = excluded.is_bot """, (user_id, first_name, last_name, username, int(is_bot))) self.conn.commit() except Exception as e: logging.error(f"Fehler beim Aktualisieren des Benutzerprofils {user_id}: {e}") def update_user_stats(self, user_id: int, lastseen_timestamp: str, first_name: str = None, last_name: str = None, username: str = None, is_bot: bool = False): """Aktualisiert die Statistiken und das letzte bekannte Profil des Benutzers.""" try: self.cursor.execute(""" INSERT INTO user_profile (user_id, first_name, last_name, username, is_bot, lastseen_timestamp) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(user_id) DO UPDATE SET lastseen_timestamp = excluded.lastseen_timestamp, first_name = COALESCE(excluded.first_name, user_profile.first_name), last_name = COALESCE(excluded.last_name, user_profile.last_name), username = COALESCE(excluded.username, user_profile.username), is_bot = COALESCE(NULLIF(excluded.is_bot, 0), user_profile.is_bot) """, (user_id, first_name, last_name, username, int(is_bot), lastseen_timestamp)) self.conn.commit() except Exception as e: logging.error(f"Fehler beim Aktualisieren der User Stats für {user_id}: {e}") def close(self): """Schließt die Datenbankverbindung.""" self.conn.close() class MonitoringBot: """Der Hauptbot-Klasse, erweitert um Topic- und Profilprotokollierung.""" def __init__(self): self.db = DatabaseManager() # Start-Methode (Muss auf der Hauptebene des Klassenblocks sein) async def start_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Verarbeitet den /start Befehl.""" await update.message.reply_text("Monitoring-Bot ist online und protokolliert Aktivitäten.") async def id_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Verarbeitet den /id Befehl und gibt die User-ID des replied-to Users zurück.""" if not update.message: return reply = update.message.reply_to_message if not reply or not reply.from_user: await update.message.reply_text( "Bitte antworte mit /id auf die Nachricht der Person, deren User-ID du wissen möchtest." ) return target_user = reply.from_user user_id = target_user.id username = getattr(target_user, 'username', None) display_name = target_user.full_name or username or str(user_id) profile_link = None if username: profile_link = f"https://t.me/{username}" else: profile_link = f"tg://user?id={user_id}" await update.message.reply_text( f"User-ID: `{user_id}`\n" f"Profil: {display_name}", parse_mode='HTML' ) # --- HILFSFUNKTIONEN ZUR TOPIC-ERKENNUNG --- def _get_topic_id(self, update: Update) -> str: """Extrahiert die Topic ID aus dem Update (Topic Aware). Versucht verschiedene Attribute je nach Telegram PTB-Version: - message_thread_id: Für Nachrichten in Topics/Threads - forum_topic_created: Wenn ein neues Topic erstellt wurde - reply_to_topic_id: Fallback für explizite Topic-Antworten """ if not update.message: return 'MAIN' # Versuche message_thread_id (Standard für Topics) if hasattr(update.message, 'message_thread_id') and update.message.message_thread_id: return str(update.message.message_thread_id) # Versuche reply_to_topic_id (Fallback) if hasattr(update.message, 'reply_to_topic_id') and update.message.reply_to_topic_id: return str(update.message.reply_to_topic_id) # Fallback: Hauptthema (MAIN) return 'MAIN' # --- EVENT HANDLER 1: BEITRITT/AUSSTIEG --- async def handle_member_update(self, client: Update, context: ContextTypes.DEFAULT_TYPE): """Verarbeitet JOIN- und LEAVE-Ereignisse.""" if not client or not context.message: return new_member = client.effective_user chat_id = client.effective_chat.id topic_id = self._get_topic_id(client.update) self._log_event(client, context, chat_id, topic_id, "JOIN", new_member.id, new_member.full_name, f"Beitritt von {new_member.full_name}") # --- EVENT HANDLER 1B: FORUM TOPIC GESCHLOSSEN --- async def handle_forum_topic_closed(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Überwacht das Schließen von Forum-Topics und öffnet sie wieder (außer in der Ausnahlmeliste).""" if not update or not update.message or not update.effective_chat: return chat_id = update.effective_chat.id # Prüfe, ob ein Topic-Close-Event vorliegt if hasattr(update.message, 'forum_topic_closed') and update.message.forum_topic_closed: topic_id = self._get_topic_id(update) details = f"Forum-Topic wurde geschlossen (Topic ID: {topic_id})" self._log_event( client=update, context=context, chat_id=chat_id, topic_id=topic_id, activity_type="TOPIC_CLOSED", user_id=0, username="system", details=details ) # Prüfe, ob dieses Topic in der Ausnahlmeliste ist should_skip = any( item['group_id'] == chat_id and str(item['topic_id']) == topic_id for item in NOT_REOPEN_TOPICS ) if should_skip: skip_details = f"Topic-Reopening übersprungen (in Ausnahmeliste)" self._log_event( client=update, context=context, chat_id=chat_id, topic_id=topic_id, activity_type="TOPIC_REOPEN_SKIPPED", user_id=0, username="system", details=skip_details ) logging.info(f"⏭️ Topic {topic_id} in Chat {chat_id} wird nicht wiedereröffnet (Ausnahlmeliste)") return # Versuche das Topic wieder zu öffnen try: await context.bot.reopen_forum_topic(chat_id=chat_id, message_thread_id=int(topic_id)) reopen_details = f"Forum-Topic automatisch wiedereröffnet (Topic ID: {topic_id})" self._log_event( client=update, context=context, chat_id=chat_id, topic_id=topic_id, activity_type="TOPIC_REOPENED", user_id=0, username="system", details=reopen_details ) logging.info(f"✅ Topic {topic_id} in Chat {chat_id} wurde wiedereröffnet") except Exception as e: error_details = f"Fehler beim Wiedereröffnen von Topic {topic_id}: {str(e)}" self._log_event( client=update, context=context, chat_id=chat_id, topic_id=topic_id, activity_type="TOPIC_REOPEN_FAILED", user_id=0, username="system", details=error_details ) logging.error(f"❌ {error_details}") # --- EVENT HANDLER 2: PROFILDÄNDERUNG (Profilvergleich mit DB) --- async def handle_profile_update(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Erkennt Profiländerungen durch Vergleich des gespeicherten Profils mit dem aktuellen.""" if not update or not update.effective_user or not update.effective_chat: return user = update.effective_user chat_id = update.effective_chat.id topic_id = self._get_topic_id(update) # Extrahiere current profile current_first_name = user.first_name or "" current_last_name = user.last_name or "" current_username = getattr(user, 'username', None) current_is_bot = user.is_bot # Lade das alte Profil aus der Datenbank old_profile = self.db.get_user_profile(user.id) # Vergleiche mit dem aktuellen Profil if old_profile: old_first_name = old_profile['first_name'] or "" old_last_name = old_profile['last_name'] or "" old_username = old_profile['username'] old_is_bot = old_profile['is_bot'] # Prüfe auf Änderungen has_name_changed = (old_first_name != current_first_name) or (old_last_name != current_last_name) has_username_changed = old_username != current_username has_is_bot_changed = old_is_bot != current_is_bot if has_name_changed or has_username_changed or has_is_bot_changed: details = f"Profiländerung erkannt" if has_name_changed: old_full = f"{old_first_name} {old_last_name}".strip() new_full = f"{current_first_name} {current_last_name}".strip() details += f" | Name: '{old_full}' → '{new_full}'" if has_username_changed: details += f" | Username: '{old_username}' → '{current_username}'" if has_is_bot_changed: details += f" | is_bot: {old_is_bot} → {current_is_bot}" self._log_event( client=update, context=context, chat_id=chat_id, topic_id=topic_id, activity_type="PROFILE_CHANGE", user_id=user.id, username=current_username or user.id, details=details ) # Speichere die Änderung im History Log self.db.log_profile_change( group_chat_id=chat_id, user_id=user.id, topic_id=topic_id, old_name=f"{old_first_name} {old_last_name}".strip() or "N/A", new_name=f"{current_first_name} {current_last_name}".strip(), old_user=old_username or "N/A", new_user=current_username or "N/A", details=details ) # Persistiere die neuen Profilinformationen in user_stats self.db.update_user_profile( user_id=user.id, first_name=current_first_name, last_name=current_last_name, username=current_username, is_bot=current_is_bot ) # --- EVENT HANDLER 3: MESSAGE (Statistik & Topic Aware & Profiländerungserkennung) --- async def handle_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Verarbeitet jede Nachricht und aktualisiert die Statistiken des ABSENDERS. Erkennt auch automatisch Profiländerungen durch Vergleich mit der Datenbank. """ if not update or not update.effective_chat or not update.effective_user: return # -1. Überwache Forum-Topic-Closes await self.handle_forum_topic_closed(update, context) # 0. Erkenne Profiländerungen BEVOR die Stats aktualisiert werden await self.handle_profile_update(update, context) # 1. TOPIC ID EXTRAHIEREN topic_id = self._get_topic_id(update) chat_id = update.effective_chat.id message_text = update.message.text if update.message else None user = update.effective_user timestamp_str = datetime.now().isoformat() # 2. Statistik-Update: Profildaten speichern und Timestamp aktualisieren self._update_user_stats( user_id=user.id, lastseen_timestamp=timestamp_str, first_name=user.first_name, last_name=user.last_name or None, username=getattr(user, 'username', None), is_bot=user.is_bot ) # General Logging (Protokollierung des Ereignisses) if message_text and not message_text.startswith('/'): self._log_event( client=update, context=context, chat_id=chat_id, topic_id=topic_id, activity_type="MESSAGE", user_id=user.id, username=user.username, details=f"Nachricht gesendet: '{message_text[:50]}...'" ) # --- INTERNE HILFSFUNKTIONEN (General Logging bleibt stabil) --- def _log_event(self, client, context, chat_id, topic_id, activity_type, user_id, username, details): """Interne Funktion zur Durchführung des Datenbank- und File-Logging.""" self.db.log_activity(group_chat_id=chat_id, topic_id=topic_id, activity_type=activity_type, user_id=user_id, username=username, details=details) # Debug Output if ENABLE_DEBUG_MODE: logging.info(f"[LOG] Aktivität erfasst ({activity_type}): User {user_id} | Topic '{topic_id}'") if WRITE_ACTIVITY_TO_FILE: print(f"--- DEBUG LOG ENTRY --- Type: {activity_type}, Topic: {topic_id}, User: {username or user_id}") # Optional File Logging if WRITE_ACTIVITY_TO_FILE: log_message = f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] | CHAT_ID={chat_id} | TOPIC={topic_id} | TYPE={activity_type} | USER={user_id} ({username}) | DETAILS={details}\n" with open(LOGGING_FILE, 'a', encoding='utf-8') as f: f.write(log_message) def _update_user_stats(self, user_id: int, lastseen_timestamp: str, first_name: str = None, last_name: str = None, username: str = None, is_bot: bool = False): """Wrapper für DatabaseManager.update_user_stats().""" self.db.update_user_stats(user_id, lastseen_timestamp, first_name, last_name, username, is_bot) # --- HAUPT-RUN FUNKTION (Bleibt gleich) --- def run(self): print("\n=============================================") print(" 🚀 Starting Monitoring Bot...") print("=============================================\n") application = Application.builder().token(BOT_TOKEN).build() # Handlers registrieren application.add_handler(CommandHandler("start", self.start_command)) application.add_handler(CommandHandler("id", self.id_command)) application.add_handler(MessageHandler(filters.ALL & ~filters.COMMAND, self.handle_message)) print("🤖 Bot wird gestartet. Warte auf Updates...") try: application.run_polling() except KeyboardInterrupt: pass finally: self.db.close() logging.info("Bot wurde manuell gestoppt und Datenbankverbindung geschlossen.") if __name__ == '__main__': monitor = MonitoringBot() monitor.run()