437 lines
17 KiB
Python
437 lines
17 KiB
Python
#!/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 config.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 config.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 config.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() |