#!/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()