314 lines
11 KiB
Python
314 lines
11 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:
|
||
# Versuche SQLite-Datenbank zu öffnen
|
||
db_files = ['database.db', 'app.db', 'myp.db']
|
||
|
||
for db_file in db_files:
|
||
if os.path.exists(db_file):
|
||
log_message(f"Gefundene Datenbankdatei: {db_file}")
|
||
|
||
conn = sqlite3.connect(db_file)
|
||
cursor = conn.cursor()
|
||
|
||
# Prüfe ob Printer-Tabelle existiert
|
||
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='printer';")
|
||
if cursor.fetchone():
|
||
log_message("✅ Printer-Tabelle gefunden")
|
||
|
||
# Zähle Drucker
|
||
cursor.execute("SELECT COUNT(*) FROM printer;")
|
||
count = cursor.fetchone()[0]
|
||
log_message(f"📊 Anzahl Drucker in Datenbank: {count}")
|
||
|
||
# Zeige Drucker-Details
|
||
cursor.execute("SELECT id, name, plug_ip, status FROM printer;")
|
||
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("❌ Printer-Tabelle nicht gefunden")
|
||
conn.close()
|
||
|
||
log_message("❌ Keine gültige Datenbank gefunden")
|
||
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:
|
||
db_files = ['database.db', 'app.db', 'myp.db']
|
||
printers = []
|
||
|
||
for db_file in db_files:
|
||
if os.path.exists(db_file):
|
||
conn = sqlite3.connect(db_file)
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT name, plug_ip FROM printer WHERE plug_ip IS NOT NULL;")
|
||
printers = cursor.fetchall()
|
||
conn.close()
|
||
break
|
||
|
||
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)
|
||
else:
|
||
result = subprocess.run(['ping', '-c', '1', '-W', '3', ip],
|
||
capture_output=True, text=True, timeout=5)
|
||
|
||
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_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()
|
||
|
||
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() |