f75957c15e
Directus-Zugang aus eigener config.json, der Agenten-config.json oder Env-Vars. Bei erkanntem Error-Fenster + bei Erreichen der max. Recover-Versuche wird die Fehlermeldung (inkl. Fenstertext) nach agent_logs (source=recover) gesendet. requests ist optional – ohne läuft AutoRecover wie bisher (nur ohne Directus-Log). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
142 lines
4.5 KiB
Python
142 lines
4.5 KiB
Python
"""
|
||
AutoRecover - Drucker Fehler Handler
|
||
Erkennt das "Error"-Fenster (Java AWT) und klickt automatisch "Recover" (max. 3x).
|
||
|
||
Zusätzlich werden die erkannten Fehlermeldungen (best-effort) an Directus gemeldet
|
||
(Collection agent_logs, source="recover"). Die Directus-Zugangsdaten kommen aus
|
||
einer config.json neben diesem Skript ODER aus der des Agenten
|
||
(../skrift-agent/config.json) ODER aus Umgebungsvariablen
|
||
(SKRIFT_DIRECTUS_URL / SKRIFT_DIRECTUS_TOKEN / SKRIFT_MACHINE_ID).
|
||
"""
|
||
|
||
import win32gui
|
||
import win32con
|
||
import ctypes
|
||
import time
|
||
import os
|
||
import json
|
||
|
||
try:
|
||
import requests
|
||
except Exception: # requests optional – ohne läuft AutoRecover trotzdem (nur ohne Directus-Log)
|
||
requests = None
|
||
|
||
MAX_VERSUCHE = 3
|
||
PRUEF_INTERVAL = 2 # Sekunden zwischen den Checks
|
||
|
||
_HIER = os.path.dirname(os.path.abspath(__file__))
|
||
|
||
|
||
def _lade_directus_cfg():
|
||
"""Directus-Zugang aus AutoRecover- oder Agenten-config.json bzw. Env-Variablen."""
|
||
for pfad in (os.path.join(_HIER, "config.json"),
|
||
os.path.join(_HIER, "..", "skrift-agent", "config.json")):
|
||
try:
|
||
with open(pfad, "r", encoding="utf-8") as f:
|
||
c = json.load(f)
|
||
if c.get("directus_url") and c.get("directus_token"):
|
||
return c.get("directus_url"), c.get("directus_token"), c.get("machine_id")
|
||
except Exception:
|
||
pass
|
||
url = os.environ.get("SKRIFT_DIRECTUS_URL")
|
||
token = os.environ.get("SKRIFT_DIRECTUS_TOKEN")
|
||
mid = os.environ.get("SKRIFT_MACHINE_ID")
|
||
return url, token, (int(mid) if mid and str(mid).isdigit() else None)
|
||
|
||
|
||
_DX_URL, _DX_TOKEN, _DX_MID = _lade_directus_cfg()
|
||
|
||
|
||
def dx_log(level, message):
|
||
"""Meldet ein Ereignis an Directus (agent_logs). Fehler nie fatal."""
|
||
if not (requests and _DX_URL and _DX_TOKEN):
|
||
return
|
||
try:
|
||
requests.post(
|
||
_DX_URL.rstrip("/") + "/items/agent_logs",
|
||
headers={"Authorization": f"Bearer {_DX_TOKEN}"},
|
||
json={"machine": _DX_MID, "level": level, "source": "recover",
|
||
"message": str(message)[:2000]},
|
||
timeout=15,
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def fenster_text(hwnd):
|
||
"""Liest die Beschriftungen der Kind-Fenster (die eigentliche Fehlermeldung)."""
|
||
teile = []
|
||
|
||
def _cb(h, _):
|
||
try:
|
||
t = win32gui.GetWindowText(h)
|
||
if t and t.strip():
|
||
teile.append(t.strip())
|
||
except Exception:
|
||
pass
|
||
return True
|
||
|
||
try:
|
||
win32gui.EnumChildWindows(hwnd, _cb, None)
|
||
except Exception:
|
||
pass
|
||
return " / ".join(teile)[:500]
|
||
|
||
def click_recover_button(hwnd):
|
||
"""
|
||
Holt die Fensterposition und klickt auf den Recover-Button.
|
||
Der Button ist zentriert am unteren Rand des Fensters.
|
||
"""
|
||
# Fenster in den Vordergrund holen
|
||
win32gui.SetForegroundWindow(hwnd)
|
||
time.sleep(0.3)
|
||
|
||
# Fensterposition und -größe ermitteln
|
||
left, top, right, bottom = win32gui.GetWindowRect(hwnd)
|
||
width = right - left
|
||
height = bottom - top
|
||
|
||
# Recover-Button liegt ca. bei: horizontal zentriert, ~85% der Höhe
|
||
click_x = left + width // 2
|
||
click_y = top + int(height * 0.85)
|
||
|
||
# Mausklick simulieren
|
||
ctypes.windll.user32.SetCursorPos(click_x, click_y)
|
||
time.sleep(0.1)
|
||
ctypes.windll.user32.mouse_event(0x0002, 0, 0, 0, 0) # linke Taste runter
|
||
ctypes.windll.user32.mouse_event(0x0004, 0, 0, 0, 0) # linke Taste hoch
|
||
|
||
print(f" → Klick auf Position ({click_x}, {click_y})")
|
||
|
||
def main():
|
||
versuche = 0
|
||
print("AutoRecover läuft... (Strg+C zum Beenden)")
|
||
print(f"Maximal {MAX_VERSUCHE} Versuche.\n")
|
||
|
||
while versuche < MAX_VERSUCHE:
|
||
hwnd = win32gui.FindWindow(None, "Error")
|
||
|
||
if hwnd and win32gui.IsWindowVisible(hwnd):
|
||
versuche += 1
|
||
meldung = fenster_text(hwnd) or "Error"
|
||
print(f"[Versuch {versuche}/{MAX_VERSUCHE}] Fehlerfenster gefunden!")
|
||
dx_log("error", f"Plotter-Fehlerfenster erkannt (Versuch {versuche}/{MAX_VERSUCHE}): {meldung}")
|
||
|
||
time.sleep(0.5)
|
||
click_recover_button(hwnd)
|
||
|
||
if versuche >= MAX_VERSUCHE:
|
||
print(f"\nMaximale Versuche ({MAX_VERSUCHE}) erreicht. Programm beendet.")
|
||
dx_log("error", f"Maximale Recover-Versuche ({MAX_VERSUCHE}) erreicht – manueller Eingriff nötig.")
|
||
break
|
||
|
||
time.sleep(3)
|
||
else:
|
||
time.sleep(PRUEF_INTERVAL)
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except KeyboardInterrupt:
|
||
print("\nProgramm manuell beendet.")
|