360 lines
12 KiB
Python
360 lines
12 KiB
Python
#!/usr/bin/env python3.11
|
|
"""
|
|
Printer Utilities - Konsolidierte Drucker-Management-Hilfsfunktionen
|
|
Zusammenfassung von Drucker-Aktivierung und Standort-Updates
|
|
"""
|
|
|
|
from models import get_db_session, Printer
|
|
from utils.logging_config import get_logger
|
|
from datetime import datetime
|
|
|
|
# Logger initialisieren
|
|
logger = get_logger("printer_utilities")
|
|
|
|
# ===== DRUCKER AKTIVIERUNG =====
|
|
|
|
def aktiviere_alle_drucker():
|
|
"""
|
|
Aktiviert alle Drucker in der Datenbank.
|
|
|
|
Returns:
|
|
dict: Ergebnis der Aktivierung mit Statistiken
|
|
"""
|
|
try:
|
|
session = get_db_session()
|
|
drucker = session.query(Printer).all()
|
|
|
|
if not drucker:
|
|
logger.warning("Keine Drucker in der Datenbank gefunden.")
|
|
session.close()
|
|
return {
|
|
'success': False,
|
|
'message': 'Keine Drucker gefunden',
|
|
'activated_count': 0
|
|
}
|
|
|
|
logger.info(f"Anzahl Drucker: {len(drucker)}")
|
|
logger.info("Aktiviere alle Drucker...")
|
|
|
|
activated_count = 0
|
|
for d in drucker:
|
|
if not d.active:
|
|
d.active = True
|
|
activated_count += 1
|
|
logger.info(f"Drucker {d.id}: {d.name} - IP: {d.plug_ip} - Aktiviert")
|
|
else:
|
|
logger.debug(f"Drucker {d.id}: {d.name} - Bereits aktiv")
|
|
|
|
session.commit()
|
|
session.close()
|
|
|
|
logger.info(f"✅ {activated_count} Drucker wurden erfolgreich aktiviert!")
|
|
|
|
return {
|
|
'success': True,
|
|
'message': f'{activated_count} Drucker aktiviert',
|
|
'activated_count': activated_count,
|
|
'total_count': len(drucker)
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Fehler bei Drucker-Aktivierung: {str(e)}")
|
|
try:
|
|
session.rollback()
|
|
session.close()
|
|
except:
|
|
pass
|
|
|
|
return {
|
|
'success': False,
|
|
'message': f'Fehler: {str(e)}',
|
|
'activated_count': 0
|
|
}
|
|
|
|
def deaktiviere_alle_drucker():
|
|
"""
|
|
Deaktiviert alle Drucker in der Datenbank.
|
|
|
|
Returns:
|
|
dict: Ergebnis der Deaktivierung mit Statistiken
|
|
"""
|
|
try:
|
|
session = get_db_session()
|
|
drucker = session.query(Printer).all()
|
|
|
|
if not drucker:
|
|
logger.warning("Keine Drucker in der Datenbank gefunden.")
|
|
session.close()
|
|
return {
|
|
'success': False,
|
|
'message': 'Keine Drucker gefunden',
|
|
'deactivated_count': 0
|
|
}
|
|
|
|
logger.info(f"Anzahl Drucker: {len(drucker)}")
|
|
logger.info("Deaktiviere alle Drucker...")
|
|
|
|
deactivated_count = 0
|
|
for d in drucker:
|
|
if d.active:
|
|
d.active = False
|
|
deactivated_count += 1
|
|
logger.info(f"Drucker {d.id}: {d.name} - IP: {d.plug_ip} - Deaktiviert")
|
|
else:
|
|
logger.debug(f"Drucker {d.id}: {d.name} - Bereits inaktiv")
|
|
|
|
session.commit()
|
|
session.close()
|
|
|
|
logger.info(f"✅ {deactivated_count} Drucker wurden erfolgreich deaktiviert!")
|
|
|
|
return {
|
|
'success': True,
|
|
'message': f'{deactivated_count} Drucker deaktiviert',
|
|
'deactivated_count': deactivated_count,
|
|
'total_count': len(drucker)
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Fehler bei Drucker-Deaktivierung: {str(e)}")
|
|
try:
|
|
session.rollback()
|
|
session.close()
|
|
except:
|
|
pass
|
|
|
|
return {
|
|
'success': False,
|
|
'message': f'Fehler: {str(e)}',
|
|
'deactivated_count': 0
|
|
}
|
|
|
|
# ===== STANDORT-MANAGEMENT =====
|
|
|
|
def update_printer_locations(new_location="Werk 040 - Berlin - TBA"):
|
|
"""
|
|
Aktualisiert alle Drucker-Standorte zu einem neuen Standort.
|
|
|
|
Args:
|
|
new_location (str): Neuer Standort für alle Drucker
|
|
|
|
Returns:
|
|
dict: Ergebnis der Standort-Aktualisierung mit Statistiken
|
|
"""
|
|
try:
|
|
session = get_db_session()
|
|
|
|
# Alle Drucker abrufen
|
|
all_printers = session.query(Printer).all()
|
|
logger.info(f"Gefundene Drucker: {len(all_printers)}")
|
|
|
|
if not all_printers:
|
|
logger.warning("Keine Drucker in der Datenbank gefunden.")
|
|
session.close()
|
|
return {
|
|
'success': False,
|
|
'message': 'Keine Drucker gefunden',
|
|
'updated_count': 0
|
|
}
|
|
|
|
updated_count = 0
|
|
location_changes = []
|
|
|
|
# Alle Drucker durchgehen und Standort aktualisieren
|
|
for printer in all_printers:
|
|
old_location = printer.location
|
|
if old_location != new_location:
|
|
printer.location = new_location
|
|
location_changes.append({
|
|
'printer_id': printer.id,
|
|
'printer_name': printer.name,
|
|
'old_location': old_location,
|
|
'new_location': new_location
|
|
})
|
|
logger.info(f"✅ {printer.name}: '{old_location}' → '{new_location}'")
|
|
updated_count += 1
|
|
else:
|
|
logger.debug(f"Drucker {printer.name}: Standort bereits korrekt")
|
|
|
|
# Änderungen speichern
|
|
session.commit()
|
|
session.close()
|
|
|
|
logger.info(f"✅ {updated_count} Drucker-Standorte erfolgreich aktualisiert")
|
|
logger.info(f"Neuer Standort: {new_location}")
|
|
|
|
return {
|
|
'success': True,
|
|
'message': f'{updated_count} Standorte aktualisiert',
|
|
'updated_count': updated_count,
|
|
'total_count': len(all_printers),
|
|
'new_location': new_location,
|
|
'changes': location_changes
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Fehler bei der Standort-Aktualisierung: {e}")
|
|
try:
|
|
session.rollback()
|
|
session.close()
|
|
except:
|
|
pass
|
|
|
|
return {
|
|
'success': False,
|
|
'message': f'Fehler: {str(e)}',
|
|
'updated_count': 0
|
|
}
|
|
|
|
def get_printer_locations():
|
|
"""
|
|
Gibt eine Übersicht aller Drucker-Standorte zurück.
|
|
|
|
Returns:
|
|
dict: Standort-Statistiken
|
|
"""
|
|
try:
|
|
session = get_db_session()
|
|
all_printers = session.query(Printer).all()
|
|
session.close()
|
|
|
|
if not all_printers:
|
|
return {
|
|
'success': False,
|
|
'message': 'Keine Drucker gefunden',
|
|
'locations': {}
|
|
}
|
|
|
|
# Standorte gruppieren
|
|
locations = {}
|
|
for printer in all_printers:
|
|
location = printer.location or 'Unbekannt'
|
|
if location not in locations:
|
|
locations[location] = []
|
|
locations[location].append({
|
|
'id': printer.id,
|
|
'name': printer.name,
|
|
'active': printer.active,
|
|
'plug_ip': printer.plug_ip
|
|
})
|
|
|
|
return {
|
|
'success': True,
|
|
'total_printers': len(all_printers),
|
|
'locations': locations,
|
|
'location_count': len(locations)
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Abrufen der Standorte: {str(e)}")
|
|
return {
|
|
'success': False,
|
|
'message': f'Fehler: {str(e)}',
|
|
'locations': {}
|
|
}
|
|
|
|
# ===== STATUS UND STATISTIKEN =====
|
|
|
|
def get_printer_status_summary():
|
|
"""
|
|
Gibt eine Zusammenfassung des Drucker-Status zurück.
|
|
|
|
Returns:
|
|
dict: Status-Zusammenfassung
|
|
"""
|
|
try:
|
|
session = get_db_session()
|
|
all_printers = session.query(Printer).all()
|
|
session.close()
|
|
|
|
if not all_printers:
|
|
return {
|
|
'success': False,
|
|
'message': 'Keine Drucker gefunden',
|
|
'summary': {}
|
|
}
|
|
|
|
active_count = sum(1 for p in all_printers if p.active)
|
|
inactive_count = len(all_printers) - active_count
|
|
|
|
# Standort-Verteilung
|
|
location_distribution = {}
|
|
for printer in all_printers:
|
|
location = printer.location or 'Unbekannt'
|
|
location_distribution[location] = location_distribution.get(location, 0) + 1
|
|
|
|
return {
|
|
'success': True,
|
|
'summary': {
|
|
'total_printers': len(all_printers),
|
|
'active_printers': active_count,
|
|
'inactive_printers': inactive_count,
|
|
'locations': location_distribution,
|
|
'last_updated': datetime.now().isoformat()
|
|
}
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Abrufen der Status-Zusammenfassung: {str(e)}")
|
|
return {
|
|
'success': False,
|
|
'message': f'Fehler: {str(e)}',
|
|
'summary': {}
|
|
}
|
|
|
|
# ===== CLI INTERFACE =====
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
|
|
if len(sys.argv) > 1:
|
|
command = sys.argv[1]
|
|
|
|
if command == "activate-all":
|
|
result = aktiviere_alle_drucker()
|
|
print(f"✅ {result['message']}")
|
|
|
|
elif command == "deactivate-all":
|
|
result = deaktiviere_alle_drucker()
|
|
print(f"✅ {result['message']}")
|
|
|
|
elif command == "update-locations":
|
|
new_location = sys.argv[2] if len(sys.argv) > 2 else "Werk 040 - Berlin - TBA"
|
|
result = update_printer_locations(new_location)
|
|
print(f"✅ {result['message']}")
|
|
|
|
elif command == "locations":
|
|
result = get_printer_locations()
|
|
if result['success']:
|
|
print("=== Drucker-Standorte ===")
|
|
for location, printers in result['locations'].items():
|
|
print(f"\n📍 {location} ({len(printers)} Drucker):")
|
|
for printer in printers:
|
|
status = "🟢" if printer['active'] else "🔴"
|
|
print(f" {status} {printer['name']} (ID: {printer['id']}, IP: {printer['plug_ip']})")
|
|
else:
|
|
print(f"❌ {result['message']}")
|
|
|
|
elif command == "status":
|
|
result = get_printer_status_summary()
|
|
if result['success']:
|
|
summary = result['summary']
|
|
print("=== Drucker-Status ===")
|
|
print(f"Gesamt: {summary['total_printers']}")
|
|
print(f"Aktiv: {summary['active_printers']} 🟢")
|
|
print(f"Inaktiv: {summary['inactive_printers']} 🔴")
|
|
print(f"Standorte: {len(summary['locations'])}")
|
|
print(f"Letzte Aktualisierung: {summary['last_updated']}")
|
|
else:
|
|
print(f"❌ {result['message']}")
|
|
|
|
else:
|
|
print("Verfügbare Kommandos:")
|
|
print(" activate-all - Aktiviert alle Drucker")
|
|
print(" deactivate-all - Deaktiviert alle Drucker")
|
|
print(" update-locations [STANDORT] - Aktualisiert alle Standorte")
|
|
print(" locations - Zeigt Standort-Übersicht")
|
|
print(" status - Zeigt Status-Zusammenfassung")
|
|
else:
|
|
print("Verwendung: python3.11 printer_utilities.py <command>")
|
|
print("Verfügbare Kommandos: activate-all, deactivate-all, update-locations, locations, status") |