479 lines
21 KiB
Python
479 lines
21 KiB
Python
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: <a href=\"{profile_link}\">{display_name}</a>",
|
|
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() |