"Update printer blueprint and test for Tapo Sofort integration (feat)"
This commit is contained in:
parent
a642456d09
commit
c7c27452e7
@ -74,7 +74,7 @@ def get_live_printer_status():
|
||||
|
||||
@printers_blueprint.route("/control/<int:printer_id>/power", methods=["POST"])
|
||||
@login_required
|
||||
@require_permission(Permission.MANAGE_PRINTERS)
|
||||
@require_permission(Permission.CONTROL_PRINTER) # Verwende die bereits vorhandene Berechtigung
|
||||
@measure_execution_time(logger=printers_logger, task_name="API-Drucker-Stromversorgung-Steuerung")
|
||||
def control_printer_power(printer_id):
|
||||
"""
|
||||
|
||||
@ -2,206 +2,183 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
"""
|
||||
SOFORT-TEST für TP-Link Tapo P110-Steckdosen
|
||||
Testet die Verbindung mit echten Anmeldedaten und findet alle verfügbaren Steckdosen.
|
||||
Sofort-Test für TP-Link Tapo P110-Steckdosen
|
||||
Testet direkt die Verbindung zu allen konfigurierten Steckdosen
|
||||
"""
|
||||
|
||||
import sys
|
||||
import time
|
||||
import os
|
||||
import subprocess
|
||||
import time
|
||||
from datetime import datetime
|
||||
|
||||
# Tapo-Anmeldedaten (HARDKODIERT)
|
||||
# Anmeldedaten für Tapo-Steckdosen
|
||||
TAPO_USERNAME = "till.tomczak@mercedes-benz.com"
|
||||
TAPO_PASSWORD = "744563017196"
|
||||
|
||||
# IP-Bereiche zum Testen
|
||||
IP_RANGES_TO_SCAN = [
|
||||
"192.168.1.{}",
|
||||
"192.168.0.{}",
|
||||
"10.0.0.{}",
|
||||
"172.16.0.{}"
|
||||
# Standard-IPs für Tapo-Steckdosen
|
||||
# (falls nicht verfügbar, passen Sie diese an die tatsächlichen IPs in Ihrem Netzwerk an)
|
||||
TAPO_IPS = [
|
||||
"192.168.1.100",
|
||||
"192.168.1.101",
|
||||
"192.168.1.102",
|
||||
"192.168.1.103",
|
||||
"192.168.1.104",
|
||||
"192.168.1.105",
|
||||
"192.168.0.100",
|
||||
"192.168.0.101",
|
||||
"192.168.0.102",
|
||||
"192.168.0.103",
|
||||
"192.168.0.104",
|
||||
"192.168.0.105",
|
||||
"192.168.1.200",
|
||||
"192.168.1.201",
|
||||
"192.168.0.200",
|
||||
"192.168.0.201",
|
||||
]
|
||||
|
||||
def log_message(message, level="INFO"):
|
||||
"""Logge eine Nachricht mit Zeitstempel"""
|
||||
timestamp = datetime.now().strftime("%H:%M:%S")
|
||||
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
print(f"[{timestamp}] [{level}] {message}")
|
||||
|
||||
def test_ping(ip_address):
|
||||
"""Teste ob eine IP erreichbar ist"""
|
||||
def ping_address(ip_address, timeout=2):
|
||||
"""Führt einen Ping-Test durch"""
|
||||
try:
|
||||
# Platform-spezifische Ping-Befehle
|
||||
import platform
|
||||
if platform.system().lower() == "windows": # Windows
|
||||
cmd = ['ping', '-n', '1', '-w', str(timeout * 1000), ip_address]
|
||||
else: # Unix/Linux/macOS
|
||||
cmd = ['ping', '-c', '1', '-W', str(timeout), ip_address]
|
||||
|
||||
result = subprocess.run(
|
||||
['ping', '-n', '1', '-w', '1000', ip_address],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=2
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=timeout + 1
|
||||
)
|
||||
|
||||
return result.returncode == 0
|
||||
except:
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def test_tapo_connection(ip_address):
|
||||
"""Teste Tapo-Verbindung zu einer IP"""
|
||||
try:
|
||||
from PyP100 import PyP110
|
||||
|
||||
log_message(f"Teste Tapo-Verbindung zu {ip_address}...")
|
||||
p110 = PyP110.P110(ip_address, TAPO_USERNAME, TAPO_PASSWORD)
|
||||
p110.handshake() # Authentifizierung
|
||||
p110.login() # Login
|
||||
|
||||
# Geräteinformationen abrufen
|
||||
device_info = p110.getDeviceInfo()
|
||||
|
||||
log_message(f"✅ ERFOLG: {ip_address} ist eine Tapo-Steckdose!")
|
||||
log_message(f" 📛 Name: {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")
|
||||
|
||||
return True, device_info
|
||||
|
||||
except Exception as e:
|
||||
error_msg = str(e).lower()
|
||||
if "login" in error_msg or "credentials" in error_msg:
|
||||
log_message(f"❌ {ip_address}: Falsche Anmeldedaten")
|
||||
elif "timeout" in error_msg:
|
||||
log_message(f"❌ {ip_address}: Timeout")
|
||||
elif "connect" in error_msg or "unreachable" in error_msg:
|
||||
log_message(f"❌ {ip_address}: Nicht erreichbar")
|
||||
else:
|
||||
log_message(f"❌ {ip_address}: {str(e)}")
|
||||
return False, None
|
||||
|
||||
def main():
|
||||
"""Hauptfunktion: Scanne Netzwerk nach Tapo-Steckdosen"""
|
||||
log_message("🚀 SOFORT-TEST: TP-Link Tapo P110-Steckdosen")
|
||||
log_message(f"👤 Benutzername: {TAPO_USERNAME}")
|
||||
log_message(f"🔐 Passwort: {'*' * len(TAPO_PASSWORD)}")
|
||||
print()
|
||||
def test_tapo_connection():
|
||||
"""
|
||||
Testet die Verbindung zu TP-Link Tapo P110-Steckdosen.
|
||||
"""
|
||||
log_message("🔄 Überprüfe ob PyP100-Modul installiert ist...")
|
||||
|
||||
# PyP100 Import testen
|
||||
try:
|
||||
from PyP100 import PyP110
|
||||
log_message("✅ PyP100-Modul erfolgreich importiert")
|
||||
except ImportError:
|
||||
log_message("❌ FEHLER: PyP100-Modul nicht gefunden!", "ERROR")
|
||||
log_message(" Installiere mit: pip install PyP100", "ERROR")
|
||||
return
|
||||
|
||||
print()
|
||||
log_message("🔍 Scanne Netzwerk nach erreichbaren Geräten...")
|
||||
|
||||
# Erst bekannte IP-Adressen testen
|
||||
known_ips = [
|
||||
"192.168.1.100", "192.168.1.101", "192.168.1.102",
|
||||
"192.168.1.103", "192.168.1.104", "192.168.1.105",
|
||||
"192.168.0.100", "192.168.0.101", "192.168.0.102"
|
||||
]
|
||||
|
||||
found_steckdosen = []
|
||||
|
||||
for ip in known_ips:
|
||||
if test_ping(ip):
|
||||
log_message(f"🌐 {ip} ist erreichbar - teste Tapo...")
|
||||
success, device_info = test_tapo_connection(ip)
|
||||
if success:
|
||||
found_steckdosen.append((ip, device_info))
|
||||
else:
|
||||
log_message(f"⚫ {ip} nicht erreichbar")
|
||||
|
||||
# Wenn keine gefunden, erweiterte Suche
|
||||
if not found_steckdosen:
|
||||
log_message("🔍 Keine Steckdosen in Standard-IPs gefunden - erweitere Suche...")
|
||||
|
||||
for ip_range in IP_RANGES_TO_SCAN:
|
||||
log_message(f"Scanne Bereich {ip_range.format('1-20')}...")
|
||||
|
||||
for i in range(1, 21): # Erste 20 IPs pro Bereich
|
||||
ip = ip_range.format(i)
|
||||
if test_ping(ip):
|
||||
log_message(f"🌐 {ip} erreichbar - teste Tapo...")
|
||||
success, device_info = test_tapo_connection(ip)
|
||||
if success:
|
||||
found_steckdosen.append((ip, device_info))
|
||||
|
||||
# Ergebnisse
|
||||
print()
|
||||
log_message("📊 ERGEBNISSE:")
|
||||
|
||||
if found_steckdosen:
|
||||
log_message(f"🎉 {len(found_steckdosen)} Tapo-Steckdose(n) gefunden!")
|
||||
print()
|
||||
|
||||
for i, (ip, device_info) in enumerate(found_steckdosen, 1):
|
||||
log_message(f"Steckdose #{i}:")
|
||||
log_message(f" IP: {ip}")
|
||||
log_message(f" Name: {device_info.get('nickname', 'Unbekannt')}")
|
||||
log_message(f" Status: {'EIN' if device_info.get('device_on', False) else 'AUS'}")
|
||||
log_message(f" Modell: {device_info.get('model', 'Unbekannt')}")
|
||||
print()
|
||||
|
||||
# Steckdosen-Test durchführen
|
||||
log_message("🧪 Teste Ein/Aus-Schaltung der ersten Steckdose...")
|
||||
test_ip, _ = found_steckdosen[0]
|
||||
log_message("❌ PyP100-Modul nicht installiert", "ERROR")
|
||||
log_message(" Installiere jetzt mit: pip install PyP100==0.1.2")
|
||||
|
||||
try:
|
||||
subprocess.run([sys.executable, "-m", "pip", "install", "PyP100==0.1.2"], check=True)
|
||||
log_message("✅ PyP100-Modul erfolgreich installiert")
|
||||
|
||||
# Erneut importieren
|
||||
from PyP100 import PyP110
|
||||
p110 = PyP110.P110(test_ip, TAPO_USERNAME, TAPO_PASSWORD)
|
||||
p110.handshake()
|
||||
p110.login()
|
||||
except Exception as e:
|
||||
log_message(f"❌ Fehler bei Installation von PyP100: {str(e)}", "ERROR")
|
||||
log_message(" Bitte installieren Sie manuell: pip install PyP100==0.1.2")
|
||||
return
|
||||
|
||||
log_message("🔍 Starte Test für Tapo-Steckdosen...")
|
||||
log_message(f"🔐 Anmeldedaten: {TAPO_USERNAME} / {TAPO_PASSWORD}")
|
||||
|
||||
successful_connections = 0
|
||||
found_ips = []
|
||||
|
||||
for ip in TAPO_IPS:
|
||||
log_message(f"🔄 Teste IP-Adresse: {ip}")
|
||||
|
||||
# Ping-Test für Grundkonnektivität
|
||||
ping_success = ping_address(ip)
|
||||
if not ping_success:
|
||||
log_message(f" ❌ IP {ip} nicht erreichbar (Ping fehlgeschlagen)")
|
||||
continue
|
||||
|
||||
# Aktuelle Status abrufen
|
||||
log_message(f" ✅ IP {ip} ist erreichbar (Ping erfolgreich)")
|
||||
|
||||
# Tapo-Verbindung testen
|
||||
try:
|
||||
log_message(f" 🔄 Verbinde zu Tapo-Steckdose {ip}...")
|
||||
p110 = PyP110.P110(ip, TAPO_USERNAME, TAPO_PASSWORD)
|
||||
p110.handshake() # Authentifizierung
|
||||
p110.login() # Login
|
||||
|
||||
# Geräteinformationen abrufen
|
||||
device_info = p110.getDeviceInfo()
|
||||
current_state = device_info.get('device_on', False)
|
||||
|
||||
log_message(f"Aktueller Status: {'EIN' if current_state else 'AUS'}")
|
||||
# Status abrufen
|
||||
is_on = device_info.get('device_on', False)
|
||||
nickname = device_info.get('nickname', "Unbekannt")
|
||||
|
||||
# Gegenteil schalten
|
||||
if current_state:
|
||||
log_message("Schalte AUS...")
|
||||
p110.turnOff()
|
||||
time.sleep(2)
|
||||
log_message("✅ Erfolgreich ausgeschaltet!")
|
||||
log_message(f" ✅ Verbindung zu Tapo-Steckdose '{nickname}' ({ip}) erfolgreich")
|
||||
log_message(f" 📱 Gerätename: {nickname}")
|
||||
log_message(f" ⚡ Status: {'Eingeschaltet' if is_on else 'Ausgeschaltet'}")
|
||||
|
||||
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")
|
||||
|
||||
successful_connections += 1
|
||||
found_ips.append(ip)
|
||||
|
||||
# Optionales Ein-/Ausschalten zum Testen
|
||||
if len(sys.argv) > 1 and sys.argv[1] == '--toggle':
|
||||
if is_on:
|
||||
log_message(f" 🔄 Schalte Steckdose {nickname} AUS...")
|
||||
p110.turnOff()
|
||||
log_message(f" ✅ Steckdose ausgeschaltet")
|
||||
else:
|
||||
log_message(f" 🔄 Schalte Steckdose {nickname} EIN...")
|
||||
p110.turnOn()
|
||||
log_message(f" ✅ Steckdose eingeschaltet")
|
||||
|
||||
# Kurze Pause
|
||||
time.sleep(1)
|
||||
|
||||
log_message("Schalte wieder EIN...")
|
||||
p110.turnOn()
|
||||
time.sleep(2)
|
||||
log_message("✅ Erfolgreich eingeschaltet!")
|
||||
else:
|
||||
log_message("Schalte EIN...")
|
||||
p110.turnOn()
|
||||
time.sleep(2)
|
||||
log_message("✅ Erfolgreich eingeschaltet!")
|
||||
|
||||
log_message("Schalte wieder AUS...")
|
||||
p110.turnOff()
|
||||
time.sleep(2)
|
||||
log_message("✅ Erfolgreich ausgeschaltet!")
|
||||
|
||||
log_message("🎉 STECKDOSEN-STEUERUNG FUNKTIONIERT PERFEKT!")
|
||||
# Status erneut abrufen
|
||||
device_info = p110.getDeviceInfo()
|
||||
is_on = device_info.get('device_on', False)
|
||||
log_message(f" ⚡ Neuer Status: {'Eingeschaltet' if is_on else 'Ausgeschaltet'}")
|
||||
|
||||
except Exception as e:
|
||||
log_message(f"❌ Fehler beim Testen der Schaltung: {str(e)}", "ERROR")
|
||||
log_message(f" ❌ Verbindung zu Tapo-Steckdose {ip} fehlgeschlagen: {str(e)}", "ERROR")
|
||||
|
||||
# Zusammenfassung
|
||||
log_message("\n📊 Zusammenfassung:")
|
||||
log_message(f" Getestete IPs: {len(TAPO_IPS)}")
|
||||
log_message(f" Gefundene Tapo-Steckdosen: {successful_connections}")
|
||||
|
||||
if successful_connections > 0:
|
||||
log_message("✅ Tapo-Steckdosen erfolgreich gefunden und getestet!")
|
||||
log_message(f"📝 Gefundene IPs: {found_ips}")
|
||||
|
||||
# Ausgabe für Konfiguration
|
||||
log_message("\n🔧 Konfiguration für settings.py:")
|
||||
log_message(f"""
|
||||
# TP-Link Tapo Standard-Anmeldedaten
|
||||
TAPO_USERNAME = "{TAPO_USERNAME}"
|
||||
TAPO_PASSWORD = "{TAPO_PASSWORD}"
|
||||
|
||||
# Automatische Steckdosen-Erkennung aktivieren
|
||||
TAPO_AUTO_DISCOVERY = True
|
||||
|
||||
# Standard-Steckdosen-IPs
|
||||
DEFAULT_TAPO_IPS = {found_ips}
|
||||
""")
|
||||
else:
|
||||
log_message("❌ KEINE Tapo-Steckdosen gefunden!", "ERROR")
|
||||
log_message("Mögliche Ursachen:", "ERROR")
|
||||
log_message("1. Steckdosen sind in einem anderen IP-Bereich", "ERROR")
|
||||
log_message("2. Anmeldedaten sind falsch", "ERROR")
|
||||
log_message("3. Netzwerkprobleme", "ERROR")
|
||||
log_message("4. Steckdosen sind nicht im Netzwerk", "ERROR")
|
||||
log_message("❌ Keine Tapo-Steckdosen gefunden!", "ERROR")
|
||||
log_message(" Bitte überprüfen Sie die IP-Adressen und Anmeldedaten")
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
main()
|
||||
except KeyboardInterrupt:
|
||||
log_message("Test abgebrochen durch Benutzer", "INFO")
|
||||
except Exception as e:
|
||||
log_message(f"Unerwarteter Fehler: {str(e)}", "ERROR")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
print("\n====== TAPO P110 STECKDOSEN-TEST ======\n")
|
||||
test_tapo_connection()
|
||||
print("\n====== TEST ABGESCHLOSSEN ======\n")
|
||||
@ -17,7 +17,7 @@ import os
|
||||
|
||||
from models import get_db_session, Printer
|
||||
from utils.logging_config import get_logger
|
||||
from config.settings import PRINTERS, TAPO_USERNAME, TAPO_PASSWORD
|
||||
from config.settings import PRINTERS, TAPO_USERNAME, TAPO_PASSWORD, DEFAULT_TAPO_IPS, TAPO_AUTO_DISCOVERY
|
||||
|
||||
# TP-Link Tapo P110 Unterstützung hinzufügen
|
||||
try:
|
||||
@ -42,12 +42,17 @@ class PrinterMonitor:
|
||||
self.monitoring_active = False
|
||||
self.monitor_thread = None
|
||||
self.startup_initialized = False
|
||||
self.auto_discovered_tapo = False
|
||||
|
||||
# Cache-Konfiguration
|
||||
self.session_cache_ttl = 30 # 30 Sekunden für Session-Cache
|
||||
self.db_cache_ttl = 300 # 5 Minuten für DB-Cache
|
||||
|
||||
monitor_logger.info("🖨️ Drucker-Monitor initialisiert")
|
||||
|
||||
# Automatische Steckdosenerkennung starten, falls aktiviert
|
||||
if TAPO_AUTO_DISCOVERY:
|
||||
self.auto_discover_tapo_outlets()
|
||||
|
||||
def initialize_all_outlets_on_startup(self) -> Dict[str, bool]:
|
||||
"""
|
||||
@ -450,6 +455,140 @@ class PrinterMonitor:
|
||||
summary["offline"] += 1
|
||||
|
||||
return summary
|
||||
|
||||
def auto_discover_tapo_outlets(self) -> Dict[str, bool]:
|
||||
"""
|
||||
Automatische Erkennung und Konfiguration von TP-Link Tapo P110-Steckdosen im Netzwerk.
|
||||
|
||||
Returns:
|
||||
Dict[str, bool]: Ergebnis der Steckdosenerkennung mit IP als Schlüssel
|
||||
"""
|
||||
if self.auto_discovered_tapo:
|
||||
monitor_logger.info("🔍 Tapo-Steckdosen wurden bereits erkannt")
|
||||
return {}
|
||||
|
||||
monitor_logger.info("🔍 Starte automatische Tapo-Steckdosenerkennung...")
|
||||
results = {}
|
||||
|
||||
# 1. Zuerst die Standard-IPs aus der Konfiguration testen
|
||||
monitor_logger.info(f"🔄 Teste {len(DEFAULT_TAPO_IPS)} Standard-IPs aus der Konfiguration")
|
||||
|
||||
for ip in DEFAULT_TAPO_IPS:
|
||||
try:
|
||||
# Ping-Test für Grundkonnektivität
|
||||
ping_success = self._ping_address(ip, timeout=2)
|
||||
|
||||
if ping_success:
|
||||
monitor_logger.info(f"✅ Steckdose mit IP {ip} ist pingbar")
|
||||
|
||||
# Tapo-Verbindung testen
|
||||
if TAPO_AVAILABLE:
|
||||
try:
|
||||
p110 = PyP110.P110(ip, TAPO_USERNAME, TAPO_PASSWORD)
|
||||
p110.handshake()
|
||||
p110.login()
|
||||
device_info = p110.getDeviceInfo()
|
||||
|
||||
# Steckdose gefunden und verbunden
|
||||
nickname = device_info.get('nickname', f"Tapo P110 ({ip})")
|
||||
state = "on" if device_info.get('device_on', False) else "off"
|
||||
|
||||
monitor_logger.info(f"✅ Tapo-Steckdose '{nickname}' ({ip}) gefunden - Status: {state}")
|
||||
results[ip] = True
|
||||
|
||||
# Steckdose in Datenbank speichern/aktualisieren
|
||||
self._ensure_tapo_in_database(ip, nickname)
|
||||
|
||||
except Exception as e:
|
||||
monitor_logger.debug(f"❌ IP {ip} ist pingbar, aber keine Tapo-Steckdose: {str(e)}")
|
||||
results[ip] = False
|
||||
else:
|
||||
monitor_logger.warning("⚠️ PyP100-Modul nicht verfügbar - kann Tapo-Verbindung nicht testen")
|
||||
results[ip] = False
|
||||
else:
|
||||
monitor_logger.debug(f"❌ IP {ip} nicht erreichbar")
|
||||
results[ip] = False
|
||||
|
||||
except Exception as e:
|
||||
monitor_logger.error(f"❌ Fehler bei Steckdosen-Erkennung für IP {ip}: {str(e)}")
|
||||
results[ip] = False
|
||||
|
||||
# Erfolgsstatistik berechnen
|
||||
success_count = sum(1 for success in results.values() if success)
|
||||
monitor_logger.info(f"✅ Steckdosen-Erkennung abgeschlossen: {success_count}/{len(results)} Steckdosen gefunden")
|
||||
|
||||
# Markieren, dass automatische Erkennung durchgeführt wurde
|
||||
self.auto_discovered_tapo = True
|
||||
|
||||
return results
|
||||
|
||||
def _ensure_tapo_in_database(self, ip_address: str, nickname: str = None) -> bool:
|
||||
"""
|
||||
Stellt sicher, dass eine erkannte Tapo-Steckdose in der Datenbank existiert.
|
||||
|
||||
Args:
|
||||
ip_address: IP-Adresse der Steckdose
|
||||
nickname: Name der Steckdose (optional)
|
||||
|
||||
Returns:
|
||||
bool: True wenn erfolgreich in Datenbank gespeichert/aktualisiert
|
||||
"""
|
||||
try:
|
||||
db_session = get_db_session()
|
||||
|
||||
# Prüfen, ob Drucker mit dieser IP bereits existiert
|
||||
existing_printer = db_session.query(Printer).filter(Printer.plug_ip == ip_address).first()
|
||||
|
||||
if existing_printer:
|
||||
# Drucker aktualisieren, falls nötig
|
||||
if not existing_printer.plug_username or not existing_printer.plug_password:
|
||||
existing_printer.plug_username = TAPO_USERNAME
|
||||
existing_printer.plug_password = TAPO_PASSWORD
|
||||
monitor_logger.info(f"✅ Drucker {existing_printer.name} mit Tapo-Anmeldedaten aktualisiert")
|
||||
|
||||
if nickname and existing_printer.name != nickname and "Tapo P110" not in existing_printer.name:
|
||||
old_name = existing_printer.name
|
||||
existing_printer.name = nickname
|
||||
monitor_logger.info(f"✅ Drucker {old_name} umbenannt zu {nickname}")
|
||||
|
||||
# Status aktualisieren
|
||||
existing_printer.last_checked = datetime.now()
|
||||
db_session.commit()
|
||||
db_session.close()
|
||||
return True
|
||||
else:
|
||||
# Neuen Drucker erstellen, falls keiner existiert
|
||||
printer_name = nickname or f"Tapo P110 ({ip_address})"
|
||||
mac_address = f"tapo:{ip_address.replace('.', '-')}" # Pseudo-MAC-Adresse
|
||||
|
||||
new_printer = Printer(
|
||||
name=printer_name,
|
||||
model="TP-Link Tapo P110",
|
||||
location="Automatisch erkannt",
|
||||
ip_address=ip_address, # Drucker-IP setzen wir gleich Steckdosen-IP
|
||||
mac_address=mac_address,
|
||||
plug_ip=ip_address,
|
||||
plug_username=TAPO_USERNAME,
|
||||
plug_password=TAPO_PASSWORD,
|
||||
status="offline",
|
||||
active=True,
|
||||
last_checked=datetime.now()
|
||||
)
|
||||
|
||||
db_session.add(new_printer)
|
||||
db_session.commit()
|
||||
monitor_logger.info(f"✅ Neuer Drucker '{printer_name}' mit Tapo-Steckdose {ip_address} erstellt")
|
||||
db_session.close()
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
monitor_logger.error(f"❌ Fehler beim Speichern der Tapo-Steckdose {ip_address}: {str(e)}")
|
||||
try:
|
||||
db_session.rollback()
|
||||
db_session.close()
|
||||
except:
|
||||
pass
|
||||
return False
|
||||
|
||||
# Globale Instanz
|
||||
printer_monitor = PrinterMonitor()
|
||||
Loading…
x
Reference in New Issue
Block a user