#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Debug-Skript für Druckererkennung Testet die Druckererkennung und identifiziert Probleme """ import sys import os import requests import json import time import threading from datetime import datetime import sqlite3 import subprocess import platform # Füge das Anwendungsverzeichnis zum Python-Pfad hinzu sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) def log_message(message, level="INFO"): """Logge eine Nachricht mit Zeitstempel""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[{timestamp}] [{level}] {message}") def test_database_connection(): """Teste die Datenbankverbindung""" log_message("Teste Datenbankverbindung...") try: # Pfad zur App hinzufügen für korrekten Import sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) try: from utils.settings import DATABASE_PATH db_file = DATABASE_PATH except ImportError: # Fallback für lokale Ausführung db_file = os.path.join('database', 'myp.db') if os.path.exists(db_file): log_message(f"Gefundene Datenbankdatei: {db_file}") conn = sqlite3.connect(db_file) cursor = conn.cursor() # Prüfe ob Printers-Tabelle existiert cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='printers';") if cursor.fetchone(): log_message("✅ Printers-Tabelle gefunden") # Zähle Drucker cursor.execute("SELECT COUNT(*) FROM printers;") count = cursor.fetchone()[0] log_message(f"📊 Anzahl Drucker in Datenbank: {count}") # Zeige Drucker-Details cursor.execute("SELECT id, name, plug_ip, status FROM printers;") printers = cursor.fetchall() for printer in printers: log_message(f" Drucker {printer[0]}: {printer[1]} ({printer[2]}) - Status: {printer[3]}") conn.close() return True else: log_message("❌ Printers-Tabelle nicht gefunden") conn.close() else: log_message(f"❌ Datenbankdatei nicht gefunden: {db_file}") return False except Exception as e: log_message(f"❌ Datenbankfehler: {str(e)}", "ERROR") return False def test_api_endpoints(): """Teste die API-Endpunkte""" log_message("Teste API-Endpunkte...") base_url = "http://localhost:5000" endpoints = [ "/api/printers", "/api/printers/status" ] for endpoint in endpoints: try: log_message(f"Teste {endpoint}...") response = requests.get(f"{base_url}{endpoint}", timeout=10) log_message(f" Status Code: {response.status_code}") if response.status_code == 200: try: data = response.json() if endpoint == "/api/printers": if 'printers' in data: log_message(f" ✅ {len(data['printers'])} Drucker geladen") else: log_message(f" ⚠️ Unerwartete Antwortstruktur: {list(data.keys())}") else: if isinstance(data, list): log_message(f" ✅ {len(data)} Drucker mit Status geladen") else: log_message(f" ⚠️ Unerwartete Antwortstruktur: {type(data)}") except json.JSONDecodeError: log_message(f" ❌ Ungültige JSON-Antwort", "ERROR") else: log_message(f" ❌ HTTP-Fehler: {response.status_code}", "ERROR") try: error_data = response.json() log_message(f" Fehlermeldung: {error_data.get('error', 'Unbekannt')}", "ERROR") except: log_message(f" Antwort: {response.text[:200]}", "ERROR") except requests.exceptions.ConnectionError: log_message(f" ❌ Verbindung zu {base_url} fehlgeschlagen", "ERROR") log_message(" Ist die Flask-Anwendung gestartet?", "ERROR") except requests.exceptions.Timeout: log_message(f" ❌ Timeout bei {endpoint}", "ERROR") except Exception as e: log_message(f" ❌ Fehler: {str(e)}", "ERROR") def test_network_connectivity(): """Teste Netzwerkverbindung zu Druckern""" log_message("Teste Netzwerkverbindung zu Druckern...") # Lade Drucker aus Datenbank try: # Verwende konfigurierten Datenbankpfad try: from utils.settings import DATABASE_PATH db_file = DATABASE_PATH except ImportError: db_file = os.path.join('database', 'myp.db') printers = [] if os.path.exists(db_file): conn = sqlite3.connect(db_file) cursor = conn.cursor() cursor.execute("SELECT name, plug_ip FROM printers WHERE plug_ip IS NOT NULL;") printers = cursor.fetchall() conn.close() if not printers: log_message("❌ Keine Drucker mit IP-Adressen gefunden") return for name, ip in printers: log_message(f"Teste Verbindung zu {name} ({ip})...") # Ping-Test try: if platform.system().lower() == "windows": result = subprocess.run(['ping', '-n', '1', '-w', '3000', ip], capture_output=True, text=True, timeout=5, encoding='utf-8', errors='replace') else: result = subprocess.run(['ping', '-c', '1', '-W', '3', ip], capture_output=True, text=True, timeout=5, encoding='utf-8', errors='replace') if result.returncode == 0: log_message(f" ✅ Ping erfolgreich") else: log_message(f" ❌ Ping fehlgeschlagen") except subprocess.TimeoutExpired: log_message(f" ❌ Ping-Timeout") except Exception as e: log_message(f" ❌ Ping-Fehler: {str(e)}") # HTTP-Test (falls Drucker Webinterface hat) try: response = requests.get(f"http://{ip}", timeout=3) log_message(f" ✅ HTTP-Verbindung erfolgreich (Status: {response.status_code})") except requests.exceptions.Timeout: log_message(f" ⚠️ HTTP-Timeout (normal für Drucker ohne Webinterface)") except requests.exceptions.ConnectionError: log_message(f" ⚠️ HTTP-Verbindung fehlgeschlagen (normal für Drucker ohne Webinterface)") except Exception as e: log_message(f" ⚠️ HTTP-Fehler: {str(e)}") except Exception as e: log_message(f"❌ Fehler beim Testen der Netzwerkverbindung: {str(e)}", "ERROR") def test_tapo_connections(): """Teste TP-Link Tapo P110-Steckdosen-Verbindungen""" log_message("Teste TP-Link Tapo P110-Steckdosen-Verbindungen...") try: # PyP100 importieren from PyP100 import PyP110 log_message("✅ PyP100-Modul erfolgreich importiert") except ImportError: log_message("❌ PyP100-Modul nicht verfügbar", "ERROR") log_message(" Installiere mit: pip install PyP100", "INFO") return # Lade Drucker aus Datenbank try: # Verwende konfigurierten Datenbankpfad try: from utils.settings import DATABASE_PATH db_file = DATABASE_PATH except ImportError: db_file = os.path.join('database', 'myp.db') printers = [] if os.path.exists(db_file): conn = sqlite3.connect(db_file) cursor = conn.cursor() cursor.execute("SELECT id, name, plug_ip, plug_username, plug_password FROM printers WHERE plug_ip IS NOT NULL;") printers = cursor.fetchall() conn.close() if not printers: log_message("❌ Keine Drucker mit Tapo-Konfiguration gefunden") return successful_connections = 0 total_printers = len(printers) for printer_id, name, plug_ip, plug_username, plug_password in printers: log_message(f"Teste Tapo-Verbindung zu {name} ({plug_ip})...") # Konfiguration validieren if not all([plug_ip, plug_username, plug_password]): log_message(f" ❌ Unvollständige Konfiguration") missing = [] if not plug_ip: missing.append("IP-Adresse") if not plug_username: missing.append("Benutzername") if not plug_password: missing.append("Passwort") log_message(f" Fehlend: {', '.join(missing)}") continue try: # Tapo-Verbindung herstellen p110 = PyP110.P110(plug_ip, plug_username, plug_password) p110.handshake() # Authentifizierung p110.login() # Login # Geräteinformationen abrufen device_info = p110.getDeviceInfo() log_message(f" ✅ Tapo-Verbindung erfolgreich") log_message(f" 📛 Gerätename: {device_info.get('nickname', 'Unbekannt')}") log_message(f" ⚡ Status: {'Ein' if device_info.get('device_on', False) else 'Aus'}") if 'on_time' in device_info: on_time = device_info.get('on_time', 0) hours, minutes = divmod(on_time // 60, 60) log_message(f" ⏱️ Betriebszeit: {hours}h {minutes}m") if 'power_usage' in device_info: power_usage = device_info.get('power_usage', {}) current_power = power_usage.get('power_mw', 0) / 1000 # mW zu W log_message(f" 🔋 Aktueller Verbrauch: {current_power:.1f}W") successful_connections += 1 except Exception as e: log_message(f" ❌ Tapo-Verbindung fehlgeschlagen: {str(e)}") # Detaillierte Fehleranalyse if "login" in str(e).lower(): log_message(f" 🔐 Mögliche Ursache: Falsche Anmeldedaten") elif "timeout" in str(e).lower(): log_message(f" ⏱️ Mögliche Ursache: Netzwerk-Timeout") elif "connect" in str(e).lower(): log_message(f" 🌐 Mögliche Ursache: Steckdose nicht erreichbar") elif "handshake" in str(e).lower(): log_message(f" 🤝 Mögliche Ursache: Protokoll-Handshake fehlgeschlagen") # Zusammenfassung success_rate = (successful_connections / total_printers * 100) if total_printers > 0 else 0 log_message(f"📊 Tapo-Verbindungs-Zusammenfassung:") log_message(f" Getestete Drucker: {total_printers}") log_message(f" Erfolgreiche Verbindungen: {successful_connections}") log_message(f" Erfolgsrate: {success_rate:.1f}%") if successful_connections == total_printers: log_message("🎉 Alle Tapo-Verbindungen erfolgreich!") elif successful_connections > 0: log_message("⚠️ Einige Tapo-Verbindungen fehlgeschlagen") else: log_message("❌ Keine Tapo-Verbindungen erfolgreich", "ERROR") except Exception as e: log_message(f"❌ Fehler beim Testen der Tapo-Verbindungen: {str(e)}", "ERROR") def test_flask_app_status(): """Teste den Status der Flask-Anwendung""" log_message("Teste Flask-Anwendung...") try: # Teste Hauptseite response = requests.get("http://localhost:5000", timeout=5) if response.status_code == 200: log_message("✅ Flask-Anwendung läuft") else: log_message(f"⚠️ Flask-Anwendung antwortet mit Status {response.status_code}") except requests.exceptions.ConnectionError: log_message("❌ Flask-Anwendung nicht erreichbar", "ERROR") log_message(" Starte die Anwendung mit: python app.py", "INFO") except Exception as e: log_message(f"❌ Fehler beim Testen der Flask-Anwendung: {str(e)}", "ERROR") def test_threading_timeout(): """Teste die Threading-basierte Timeout-Implementierung""" log_message("Teste Threading-Timeout-Implementierung...") def test_function(): """Simuliere eine langsame Datenbankabfrage""" time.sleep(2) return "Erfolgreich" try: result = None timeout_occurred = False def run_test(): nonlocal result, timeout_occurred try: result = test_function() except Exception as e: log_message(f"Fehler in Test-Thread: {str(e)}", "ERROR") timeout_occurred = True # Starte Test in separatem Thread thread = threading.Thread(target=run_test) thread.daemon = True thread.start() thread.join(timeout=3) # 3 Sekunden Timeout if thread.is_alive() or timeout_occurred or result is None: log_message("❌ Threading-Timeout-Test fehlgeschlagen", "ERROR") else: log_message("✅ Threading-Timeout-Implementierung funktioniert") except Exception as e: log_message(f"❌ Fehler beim Threading-Test: {str(e)}", "ERROR") def check_system_requirements(): """Prüfe Systemanforderungen""" log_message("Prüfe Systemanforderungen...") # Python-Version python_version = sys.version_info log_message(f"Python-Version: {python_version.major}.{python_version.minor}.{python_version.micro}") if python_version.major >= 3 and python_version.minor >= 7: log_message("✅ Python-Version ist kompatibel") else: log_message("❌ Python 3.7+ erforderlich", "ERROR") # Erforderliche Module required_modules = ['flask', 'requests', 'sqlite3', 'threading'] for module in required_modules: try: __import__(module) log_message(f"✅ Modul {module} verfügbar") except ImportError: log_message(f"❌ Modul {module} nicht verfügbar", "ERROR") # Betriebssystem os_name = platform.system() log_message(f"Betriebssystem: {os_name}") if os_name == "Windows": log_message("✅ Windows-spezifische Fixes wurden angewendet") else: log_message("ℹ️ Unix-basiertes System erkannt") def run_comprehensive_test(): """Führe alle Tests aus""" log_message("=== MYP Druckerverwaltung - Diagnose-Tool ===") log_message("Starte umfassende Systemdiagnose...") print() # Systemanforderungen prüfen check_system_requirements() print() # Threading-Test test_threading_timeout() print() # Datenbanktest test_database_connection() print() # Flask-App-Test test_flask_app_status() print() # API-Tests test_api_endpoints() print() # Netzwerk-Tests test_network_connectivity() print() # Tapo-Verbindungen testen test_tapo_connections() print() log_message("=== Diagnose abgeschlossen ===") print() # Empfehlungen log_message("📋 Empfehlungen:") log_message("1. Stelle sicher, dass die Flask-Anwendung läuft: python app.py") log_message("2. Prüfe die Datenbankverbindung und Drucker-Konfiguration") log_message("3. Teste die Netzwerkverbindung zu den Druckern") log_message("4. Bei Windows: Threading-basierte Timeouts wurden implementiert") log_message("5. Überprüfe die Logs in logs/app/ für weitere Details") if __name__ == "__main__": try: run_comprehensive_test() except KeyboardInterrupt: log_message("Diagnose durch Benutzer abgebrochen", "INFO") except Exception as e: log_message(f"Unerwarteter Fehler: {str(e)}", "ERROR") import traceback traceback.print_exc()