From 87484d68632136b22f497302f55fbce82d218a6d Mon Sep 17 00:00:00 2001 From: Hitonabi Date: Sun, 26 Jul 2026 14:17:14 +0200 Subject: [PATCH] feat(worker): der externe Worker wird erwachsen - Log in Rippy, Anzeige, Slots Commander 26.07.2026: "der externe Encoder Worker ist ein bisschen duenn - der koennte noch viel mehr." Drei Punkte, alle am Tray. 1. LOG IN RIPPY STATT TXT-DATEI (ausdruecklich gewuenscht). Das Tray schrieb sein Log nach %LOCALAPPDATA% und oeffnete es im Editor - wer wissen wollte, warum der Worker nichts tut, musste sich an den PC setzen. Neue Bruecke (logbruecke.py) meldet die wichtigen Zeilen nach Rippy, Quelle "w:", und die Logs-Seite hat jetzt Knoepfe je Quelle: "was macht mein PC" ist ein Klick. Der Filter konnte Quellen schon immer, es gab nur keinen Knopf. Durchgelassen wird WENIG und mit Grund: Die Job-Meldungen stehen laengst in Rippy (tasks.py schreibt sie selbst). Es fehlte, was DANEBEN passiert und den Worker unbrauchbar macht, ohne dass ein Job existiert - hochgefahren oder nicht, Verbindung zu Redis/Postgres, Abstuerze. Alles andere fliegt weg: Celery ist bei --loglevel=info gespraechig, die logs-Tabelle hat keine Aufraeumung, und ein zugemuelltes Log ist so unbrauchbar wie keins. Dazu eine Drossel (30 Zeilen/Minute), die MELDET, wieviel sie verschluckt hat. Zeilenformat woertlich aus dem laufenden Container abgenommen (Celery 5.4.0). Die lokale Datei bleibt - sie ist genau dann die einzige Auskunft, wenn Rippy nicht erreichbar ist. 2. DAS TRAY ZEIGT, WAS LAEUFT. Vorher stand dort "laeuft" oder "gestoppt" - auf einer Maschine, die stundenlang an einem Film rechnet, ist das keine Auskunft. Jetzt Titel, Prozent und Restzeit, geholt von Rippys /jobs. Bewusst dieselbe Quelle wie das Dashboard, damit im Tray nicht eine zweite, abweichende Schaetzung steht. Dazu: Windows schlaeft nicht mehr mitten im Encode ein (SetThreadExecutionState, ohne ES_DISPLAY_REQUIRED - der Bildschirm darf ausgehen). Die Sperre wird zurueckgenommen, sobald nichts laeuft, und auch bei einem harten Ende des Trays - sonst schlaeft der PC nie wieder ein und niemand weiss warum. 3. MEHRERE ENCODES GLEICHZEITIG. Der Worker lief fest mit --pool=solo und nahm genau EINEN Auftrag an. Der Installer fragt die Zahl jetzt (GUI: Feld neben dem Namen, mit der erkannten Kernzahl daneben), Vorbelegung ab 12 Kernen zwei, sonst einer: HandBrake nutzt schon alle Kerne, aber x265 skaliert nicht linear. Auf Windows gibt es keinen prefork-Pool (kein fork) - deshalb --pool=threads, was hier passt, weil die Arbeit ein Kind-Prozess ist und der Thread nur wartet. GUI-Layout headless gerendert und angesehen (nichts ueberlappt, 16 Kerne korrekt erkannt), beide .ps1 mit echtem PowerShell 5.1 geprueft, BOM und CRLF erhalten, .exe neu gebaut. Co-Authored-By: Claude Opus 5 --- deploy/worker-windows/RippyWorkerSetup.exe | Bin 66048 -> 68096 bytes deploy/worker-windows/install-gui.ps1 | 44 +++- deploy/worker-windows/install.ps1 | 18 +- docker/ui/src/pages/Logs.tsx | 41 +++- docker/worker/logbruecke.py | 157 +++++++++++++++ docker/worker/test_logbruecke.py | 127 ++++++++++++ docker/worker/tray.py | 221 +++++++++++++++++++-- 7 files changed, 587 insertions(+), 21 deletions(-) create mode 100644 docker/worker/logbruecke.py create mode 100644 docker/worker/test_logbruecke.py diff --git a/deploy/worker-windows/RippyWorkerSetup.exe b/deploy/worker-windows/RippyWorkerSetup.exe index 1be880a24d704d1c0bcdd6f5bfa50c9d8fa41ad3..8b986be13e4dab8065525af2273c795d8458f686 100644 GIT binary patch delta 1712 zcmaJ?&u<$=6yCK0wNf>LScHm5)r(^ho7yYymO8;w5Fs&5h@DuCo2WlfJ+{a8lHFNr zcDD(xQs=;-;*@APao~W25F8N5g$w)*+(81BmO~X)LR=$+c(dNPn**%0+L<@=zW3(6 z@0;EKaAg1Ek%Rm3FOTW_Qt&*KBJx@JtRx+w;ljZ9b@cG7&9Q1!mR5xUWa-J|;Z)3< zl5dYbcp@zL9(fvNY6>jKT6X1q9h3`m!#*jIDWa~ zO+9|CUyeMy7>RZInld^W(O&o7T=}AR#h`k<-KDytwVxJYt*1Em$h(jy$!3Bvo2{Hc_1}VHGfVS zba$wp2HRp7i0X`T1+H)mkcVYF$=jT4Q;{EYovs6Ry=^h*5yPMZ|1^SOn!8%=GT7S$ z)8`H_ERKUmVU%#l0CAfGWT~p8gWLb^*&}hZ&%ryrg7v`tWuem1Wdi`2w$=bm)K3g2dc*Bn|bg~LF zy;+Q*oiZyWo$01)V%};9MJbWpz&sZ=ZxX|$xHpm1n~=O&V0UzGF_SXqy?nmRtva=B zi!WL(LlPlJRV@W~L4!S@hE2zDtiql+<1nqi-kFFFCu$`$DOI$WPF4m5z?utQdFu5E z@2y*7q*w8y3w=O?a3P`{ z!X4_MT22m>u!UrhKR0z`%#Z|%Ppi;2ar-Rr7H$+Y3%oLT4XA=n@zC`Y52yuqw}$qh z40wyxEmn6qr>y?L`~U^_KZ{eeTBp?-YFZuw2k*?@KNB`};Oa`beEs%DalNt;fc52j z`|^LRdGfua_j&K#Xd*aF|G0PK#tTnB+bYig@b~&B&d*E6)poMq_~lk~?0)cv5PykJ mqux)CUWpt;`$vxwvX^~a?T`I&JQ?|B?zxGIeA00Sd~H%LH%k>Q2( zL`C7rij47GA`d|#K>TQO7o#@glgZl{)fpKk-(yr#`vBAc0vc+Q&zOAhmDc6}rU%+g#TlE8^thy%j14B+c`n~v;;&~sxvAyTWR-T6&6`>y z7=esDZe87i{QMl-;{2Ta&9mC0q6By>=N!NO|C;yOYrh#Uzgx9!Gs~kG4!##qyFvE< kW0;)xN{Z log.level === filter || log.source === filter) + /* + * Quellen zum Anklicken (Commander-Wunsch 26.07.2026: „bessere Log-Ansichten + * — kein txt file, direkt von Rippy Logs"). + * + * Der Filter konnte Quellen schon immer, es gab nur keinen Knopf dafür — man + * musste die Zeilen mit dem Auge durchsuchen. Seit dieser Runde meldet auch + * der externe Windows-Worker hierher (worker/logbruecke.py, Quelle „w:"), + * und damit wird die Auswahl erst richtig nützlich: „was macht mein PC?" ist + * ein Klick. + * + * Die Liste wird aus den WIRKLICH vorhandenen Zeilen gebaut, nicht aus einer + * festen Aufzählung — sonst fehlt jede neue Maschine. + */ + const quellen = [...new Set(logs.map(l => l.source).filter(Boolean))].sort() + + // „w:tobisnicerpc" ist als Knopfbeschriftung unschön. + const quelleLabel = (q: string) => q.startsWith('w:') ? `Worker ${q.slice(2)}` : q + const stats = { info: logs.filter(l => l.level === 'info').length, warning: logs.filter(l => l.level === 'warning').length, @@ -162,6 +180,25 @@ export default function LogsPage() { {f === 'all' ? 'Alle' : getLevelLabel(f)} ))} + + {quellen.length > 1 && ( + <> + Quelle: + {quellen.map(q => ( + + ))} + + )}
diff --git a/docker/worker/logbruecke.py b/docker/worker/logbruecke.py new file mode 100644 index 0000000..2abb5af --- /dev/null +++ b/docker/worker/logbruecke.py @@ -0,0 +1,157 @@ +"""Celery-Ausgabe des Workers nach Rippy melden — statt in eine txt-Datei. + +Commander-Wunsch 26.07.2026: *„bessere Log-Ansichten (kein txt file → direkt von +Rippy Logs)"*. Der native Windows-Worker schrieb sein Log bisher nach +`%LOCALAPPDATA%\\Rippy Worker\\worker.log`, und das Tray öffnete die Datei im +Editor. Wer wissen wollte, warum der Worker nichts tut, musste sich also an den +PC setzen — während Rippy im Browser schon eine Log-Seite hat. + +## Was hier durchgeht — und warum so wenig + +Die JOB-Meldungen stehen längst in Rippy: `tasks.py` schreibt „Kompression +gestartet", Fehlschläge, Abbrüche selbst über `db.add_log`. Was fehlte, ist +alles, was DANEBEN passiert und den Worker unbrauchbar macht, ohne dass ein Job +existiert: + +- Der Worker ist hochgefahren (oder eben nicht). +- Die Verbindung zu Redis/Postgres steht nicht. +- Er ist abgestürzt (Traceback). + +Alles andere wird verworfen. Celery ist bei `--loglevel=info` gesprächig, die +`logs`-Tabelle hat keine automatische Aufräumung, und ein zugemülltes Log ist +genauso unbrauchbar wie keins. + +## Das Format ist gemessen, nicht geraten + +Wörtlich aus dem laufenden Worker-Container (26.07.2026, Celery 5.4.0) — +AGENTS Regel D: + + [2026-07-26 12:04:18,060: INFO/MainProcess] Connected to redis://redis:6379/0 + [2026-07-26 12:04:19,063: INFO/MainProcess] mingle: searching for neighbors + [2026-07-26 12:04:19,082: INFO/MainProcess] celery@ced31798928d ready. + [2026-07-26 12:06:19,094: WARNING/MainProcess] Zombie-Erkennung: {...} + +Dazu der Startbanner (Zeilen ohne diese Klammer) und Warnungen aus Pythons +warnings-Modul. +""" + +import re + +# `[: /] ` +CELERY_ZEILE = re.compile(r"^\[[^\]]*?:\s*(\w+)/([^\]]+)\]\s*(.*)$") + +# Celery-Level → Rippy-Level. DEBUG fällt bewusst weg. +LEVEL_MAP = { + "CRITICAL": "error", + "ERROR": "error", + "WARNING": "warning", + "INFO": "info", +} + +# INFO-Zeilen sind nur interessant, wenn sie einen ZUSTAND melden. Diese Stücke +# sind alle oben im echten Log nachweisbar. +INFO_DURCHLASSEN = ("ready.", "Connected to", "Trying again", "connection") + +# ... und diese sind reines Grundrauschen. +INFO_VERWERFEN = ("mingle:", "task events", "searching for neighbors", "all alone") + +# Zeilen OHNE Celery-Klammer (Startbanner, Python-warnings) kommen nur durch, +# wenn sie nach einem echten Problem aussehen. +ROHE_FEHLER = ("Traceback", "Error:", "error(", "Errno", "refused", "Cannot") + +# Schutz für die logs-Tabelle: mehr als das pro Minute ist kein Log, das ist ein +# Wasserfall. +MAX_ZEILEN_JE_MINUTE = 30 + + +def einordnen(zeile: str): + """(level, text) für eine Ausgabezeile — oder None, wenn sie nichts wert ist. + + Reine Funktion, damit das Urteil ohne laufenden Worker prüfbar ist. + """ + text = (zeile or "").strip() + if not text: + return None + treffer = CELERY_ZEILE.match(text) + if not treffer: + # Startbanner und warnings-Ausgaben: nur echte Probleme. + if any(s in text for s in ROHE_FEHLER): + return ("error", text[:500]) + return None + celery_level, _prozess, inhalt = treffer.groups() + level = LEVEL_MAP.get(celery_level.upper()) + if not level or not inhalt: + return None + if level == "info": + if any(s in inhalt for s in INFO_VERWERFEN): + return None + if not any(s in inhalt for s in INFO_DURCHLASSEN): + return None + # „ready." heißt: der Worker nimmt jetzt Aufträge an. Das ist die + # Meldung, auf die man wartet, wenn nichts passiert. + if "ready." in inhalt: + return ("success", inhalt[:500]) + return (level, inhalt[:500]) + + +def quelle_fuer(worker_name: str) -> str: + """Log-Quelle für diesen Worker. + + Gekappt auf 32 Zeichen, weil `logs.source` genau so breit ist (db.py). Ohne + das Kappen scheitert das Insert still bei einem langen Worker-Namen — und + dann fehlt genau das Log, das man sucht. + """ + name = (worker_name or "worker").strip() or "worker" + return f"w:{name}"[:32] + + +class Bruecke: + """Nimmt Ausgabezeilen und schreibt die wichtigen nach Rippy. + + `schreiber(level, quelle, text)` und `jetzt()` sind einspritzbar — so ist + die Drosselung ohne Datenbank und ohne echte Uhr prüfbar. + """ + + def __init__(self, worker_name: str, schreiber, jetzt): + self.quelle = quelle_fuer(worker_name) + self.schreiber = schreiber + self.jetzt = jetzt + self._fenster_start = None + self._im_fenster = 0 + self._unterdrueckt = 0 + + def zeile(self, text: str) -> bool: + """Eine Ausgabezeile verarbeiten. True, wenn sie nach Rippy ging.""" + eingeordnet = einordnen(text) + if not eingeordnet: + return False + level, inhalt = eingeordnet + + jetzt = self.jetzt() + if self._fenster_start is None or jetzt - self._fenster_start >= 60: + # Neues Fenster: erst melden, wieviel im alten verschluckt wurde. + if self._unterdrueckt: + self._melden( + "warning", + f"{self._unterdrueckt} weitere Log-Zeilen unterdrückt " + "(mehr als 30 pro Minute) — Details im Worker-Log auf der " + "Maschine selbst.", + ) + self._fenster_start = jetzt + self._im_fenster = 0 + self._unterdrueckt = 0 + + if self._im_fenster >= MAX_ZEILEN_JE_MINUTE: + self._unterdrueckt += 1 + return False + self._im_fenster += 1 + return self._melden(level, inhalt) + + def _melden(self, level: str, text: str) -> bool: + try: + self.schreiber(level, self.quelle, text) + return True + except Exception: + # Rippy nicht erreichbar → die lokale Datei bleibt der Rückfall. + # Hier NICHTS werfen: der Worker soll weiterarbeiten. + return False diff --git a/docker/worker/test_logbruecke.py b/docker/worker/test_logbruecke.py new file mode 100644 index 0000000..ed03aaf --- /dev/null +++ b/docker/worker/test_logbruecke.py @@ -0,0 +1,127 @@ +"""Tests der Log-Brücke — reine Funktionen, keine Datenbank, keine Uhr. + +Die Beispielzeilen sind WÖRTLICH aus dem laufenden Worker-Container abgenommen +(26.07.2026, Celery 5.4.0) — AGENTS Regel D. Erfundene Formate hätten hier +keinen Wert: Genau am geratenen Format scheitert so eine Brücke. +""" + +import logbruecke + +# Aus `docker compose logs worker`, unverändert. +ECHT_CONNECTED = "[2026-07-26 12:04:18,060: INFO/MainProcess] Connected to redis://redis:6379/0" +ECHT_MINGLE = "[2026-07-26 12:04:18,063: INFO/MainProcess] mingle: searching for neighbors" +ECHT_ALLEIN = "[2026-07-26 12:04:19,071: INFO/MainProcess] mingle: all alone" +ECHT_READY = "[2026-07-26 12:04:19,082: INFO/MainProcess] celery@ced31798928d ready." +ECHT_WARNUNG = ( + "[2026-07-26 12:06:19,094: WARNING/MainProcess] Zombie-Erkennung: " + "{'geprueft': 0, 'aufgeraeumt': [], 'uebersprungen': ''}" +) +ECHT_BANNER = " -------------- celery@ced31798928d v5.4.0 (opalescent)" + + +def test_ready_ist_die_meldung_auf_die_man_wartet(): + """Wenn der Worker nichts tut, ist „ready." die Zeile, die man sehen will — + deshalb als Erfolg, nicht als Info im Grundrauschen.""" + assert logbruecke.einordnen(ECHT_READY) == ("success", "celery@ced31798928d ready.") + + +def test_verbindung_kommt_durch(): + level, text = logbruecke.einordnen(ECHT_CONNECTED) + assert level == "info" + assert "redis" in text + + +def test_warnungen_und_fehler_kommen_immer_durch(): + level, text = logbruecke.einordnen(ECHT_WARNUNG) + assert level == "warning" + assert "Zombie-Erkennung" in text + assert logbruecke.einordnen( + "[2026-07-26 12:00:00,000: ERROR/MainProcess] kaputt")[0] == "error" + assert logbruecke.einordnen( + "[2026-07-26 12:00:00,000: CRITICAL/MainProcess] ganz kaputt")[0] == "error" + + +def test_grundrauschen_wird_verworfen(): + """Celery ist bei --loglevel=info gesprächig, und die logs-Tabelle hat keine + automatische Aufräumung. Ein zugemülltes Log ist so unbrauchbar wie keins.""" + assert logbruecke.einordnen(ECHT_MINGLE) is None + assert logbruecke.einordnen(ECHT_ALLEIN) is None + assert logbruecke.einordnen(ECHT_BANNER) is None + assert logbruecke.einordnen("") is None + assert logbruecke.einordnen(None) is None + assert logbruecke.einordnen( + "[2026-07-26 12:00:00,000: DEBUG/MainProcess] kleinteiliges Zeug") is None + + +def test_absturz_ohne_celery_klammer_kommt_durch(): + """Ein Traceback trägt kein Celery-Präfix — und ist das Wichtigste, was + passieren kann.""" + assert logbruecke.einordnen("Traceback (most recent call last):")[0] == "error" + assert logbruecke.einordnen( + "ConnectionRefusedError: [Errno 111] Connection refused")[0] == "error" + + +def test_quelle_wird_auf_32_zeichen_gekappt(): + """`logs.source` ist String(32). Ohne Kappen scheitert das Insert STILL bei + einem langen Worker-Namen — und dann fehlt genau das Log, das man sucht.""" + assert logbruecke.quelle_fuer("tobisnicerpc") == "w:tobisnicerpc" + lang = logbruecke.quelle_fuer("x" * 100) + assert len(lang) == 32 + assert lang.startswith("w:") + assert logbruecke.quelle_fuer("") == "w:worker" + assert logbruecke.quelle_fuer(None) == "w:worker" + + +# --- Drosselung ------------------------------------------------------------- + + +def _bruecke(uhr): + geschrieben = [] + b = logbruecke.Bruecke( + "testpc", + schreiber=lambda level, quelle, text: geschrieben.append((level, text)), + jetzt=lambda: uhr[0], + ) + return b, geschrieben + + +def test_drosselung_haelt_bei_30_zeilen_die_minute(): + uhr = [0.0] + b, geschrieben = _bruecke(uhr) + for i in range(40): + b.zeile(f"[2026-07-26 12:00:00,000: ERROR/MainProcess] Fehler {i}") + assert len(geschrieben) == logbruecke.MAX_ZEILEN_JE_MINUTE + + +def test_naechste_minute_meldet_wieviel_verschluckt_wurde(): + """Stilles Verschlucken wäre genau der Fehler, den diese Sitzung schon + einmal eine Stunde gekostet hat.""" + uhr = [0.0] + b, geschrieben = _bruecke(uhr) + for i in range(40): + b.zeile(f"[2026-07-26 12:00:00,000: ERROR/MainProcess] Fehler {i}") + uhr[0] = 61.0 + b.zeile("[2026-07-26 12:01:01,000: ERROR/MainProcess] naechster Fehler") + + unterdrueckt = [t for lvl, t in geschrieben if "unterdrückt" in t] + assert len(unterdrueckt) == 1 + assert "10 weitere" in unterdrueckt[0] + + +def test_kaputter_schreiber_bringt_den_worker_nicht_um(): + """Rippy nicht erreichbar → die lokale Log-Datei bleibt der Rückfall. Der + Worker muss weiterarbeiten.""" + def kaputt(level, quelle, text): + raise RuntimeError("Datenbank weg") + + b = logbruecke.Bruecke("testpc", schreiber=kaputt, jetzt=lambda: 0.0) + assert b.zeile("[2026-07-26 12:00:00,000: ERROR/MainProcess] Fehler") is False + + +def test_verworfene_zeilen_zaehlen_nicht_gegen_die_drossel(): + uhr = [0.0] + b, geschrieben = _bruecke(uhr) + for _ in range(100): + b.zeile(ECHT_MINGLE) + b.zeile(ECHT_READY) + assert geschrieben == [("success", "celery@ced31798928d ready.")] diff --git a/docker/worker/tray.py b/docker/worker/tray.py index d565fa5..7c530a5 100644 --- a/docker/worker/tray.py +++ b/docker/worker/tray.py @@ -1,10 +1,25 @@ """Tray-Symbol für den NATIVEN Windows-Worker (nicht im Docker-Image genutzt). -Zeigt den Worker-Status unten rechts neben der Uhr und bietet: -Start/Stopp, Rippy im Browser öffnen, Log anzeigen, Beenden. Gestartet -wird es über start-tray.bat (setzt die Verbindungs-Umgebung und ruft -pythonw.exe auf — kein Konsolenfenster). Der Worker-Prozess (celery, -Queue transcode) läuft als Kind-Prozess und stoppt mit dem Tray. +Zeigt neben der Uhr, was diese Maschine gerade tut, und bietet Start/Stopp, +Rippy im Browser, Log und Beenden. Gestartet wird es über start-tray.bat (setzt +die Verbindungs-Umgebung und ruft pythonw.exe auf — kein Konsolenfenster). Der +Worker-Prozess (celery, Queue transcode) läuft als Kind-Prozess und stoppt mit +dem Tray. + +## Was diese Runde dazugekommen ist (26.07.2026, Commander: „der Worker ist ein +## bisschen dünn") + +1. **Es zeigt, WAS läuft.** Vorher stand im Menü nur „läuft" oder „gestoppt" — + auf einer Maschine, die stundenlang an einem Film rechnet, ist das keine + Auskunft. Jetzt: Titel, Prozent und Restzeit, geholt von Rippys `/jobs` + (die Restzeit rechnet Rippy, siehe api/eta.py — sie ist damit dieselbe Zahl + wie im Dashboard und nicht eine zweite, abweichende Schätzung). +2. **Windows schläft nicht mehr mitten im Encode ein.** `SetThreadExecutionState` + hält die Maschine wach, solange wirklich etwas läuft — und lässt sie danach + sofort wieder einschlafen. +3. **Das Log geht nach Rippy** (logbruecke.py). Die lokale Datei bleibt als + Rückfall, denn sie ist genau dann die einzige Auskunft, wenn Rippy nicht + erreichbar ist. Abhängigkeiten pystray + Pillow installiert NUR der Windows-Installer (install.ps1) — das Docker-Image bleibt davon unberührt. @@ -13,6 +28,8 @@ Abhängigkeiten pystray + Pillow installiert NUR der Windows-Installer import os import subprocess import sys +import threading +import time import webbrowser try: @@ -25,6 +42,14 @@ except ImportError: BASIS = os.path.dirname(os.path.abspath(__file__)) RIPPY_HOST = os.getenv("RIPPY_TRAY_HOST", "") WORKER_NAME = os.getenv("WORKER_NAME", "windows-worker") +# Wie viele Aufträge dieser Worker gleichzeitig annimmt. Siehe Kommentar bei +# worker_starten() — auf Windows ist der Pool `threads`, nicht `prefork`. +SLOTS = max(1, int(os.getenv("RIPPY_SLOTS", "1") or "1")) + +# Wie oft beim Dashboard nachgefragt wird, was läuft. Fünf Sekunden sind für ein +# Tray-Menü reichlich und belasten die API nicht (der Endpunkt ist ohnehin der, +# den das Dashboard alle vier Sekunden abfragt). +ABFRAGE_TAKT_SEKUNDEN = 5 def _log_pfad() -> str: @@ -55,6 +80,9 @@ def _log_pfad() -> str: LOG_PFAD = _log_pfad() prozess = None +# Was Rippy gerade meldet: None = nichts in Arbeit, sonst +# {"titel": str, "progress": int, "eta": str}. +aktuell = {"job": None} def disc_bild(laeuft: bool) -> Image.Image: @@ -71,18 +99,104 @@ def worker_laeuft() -> bool: return prozess is not None and prozess.poll() is None +# --- Windows am Einschlafen hindern ---------------------------------------- +# +# Ein 4K-Encode läuft Stunden. Schläft der PC dabei ein, ist der Job weg — und +# im UI steht bis zum nächsten Worker-Start eine Job-Leiche. Die Zustandsflags +# stehen in winbase.h; ES_CONTINUOUS hält den Zustand, bis er zurückgenommen +# wird. Bewusst OHNE ES_DISPLAY_REQUIRED: der Bildschirm darf ausgehen. +ES_CONTINUOUS = 0x80000000 +ES_SYSTEM_REQUIRED = 0x00000001 + +_wach = {"an": False} + + +def wach_halten(an: bool) -> None: + """Verhindert (an=True) bzw. erlaubt wieder (an=False) den Standby.""" + if _wach["an"] == an: + return + try: + import ctypes + + flags = (ES_CONTINUOUS | ES_SYSTEM_REQUIRED) if an else ES_CONTINUOUS + ctypes.windll.kernel32.SetThreadExecutionState(flags) + _wach["an"] = an + except (ImportError, AttributeError, OSError): + pass # kein Windows oder Aufruf verweigert — dann eben ohne + + +def _log_weiterleiten(zeile: str, bruecke, datei) -> None: + """Eine Ausgabezeile in die Datei UND (wenn wichtig) nach Rippy.""" + try: + datei.write(zeile) + datei.flush() + except OSError: + pass + if bruecke: + bruecke.zeile(zeile) + + +def _leser(prozess_ref, bruecke, datei) -> None: + """Liest die Ausgabe des Worker-Prozesses, solange er lebt.""" + try: + for zeile in prozess_ref.stdout: + _log_weiterleiten(zeile, bruecke, datei) + except Exception: + pass + finally: + try: + datei.close() + except OSError: + pass + + +def _bruecke_bauen(): + """Log-Brücke nach Rippy — oder None, wenn die Datenbank nicht erreichbar ist. + + Scheitert der Aufbau, läuft der Worker ohne Brücke weiter und das lokale Log + bleibt die Auskunft. Ein Tray, das wegen der Log-Anbindung nicht startet, + wäre die schlechtere Wahl. + """ + try: + import db + import logbruecke + + db.init_db() + return logbruecke.Bruecke( + WORKER_NAME, + schreiber=lambda level, quelle, text: db.add_log(level, quelle, text), + jetzt=time.monotonic, + ) + except Exception: + return None + + def worker_starten(icon=None, item=None): global prozess if worker_laeuft(): return celery = os.path.join(BASIS, "venv", "Scripts", "celery.exe") + # Auf Windows gibt es KEINEN prefork-Pool (kein fork). `solo` bediente + # genau einen Auftrag; mit `threads` sind mehrere möglich, und das passt + # hier, weil die eigentliche Arbeit ein Kind-Prozess ist (HandBrake) und der + # Thread nur darauf wartet. + befehl = [ + celery, "-A", "celery_app", "worker", "--loglevel=info", + "-Q", "transcode", "-n", f"{WORKER_NAME}@%h", + ] + befehl += (["--pool=solo"] if SLOTS == 1 + else ["--pool=threads", f"--concurrency={SLOTS}"]) log = open(LOG_PFAD, "a", encoding="utf-8", errors="replace") prozess = subprocess.Popen( - [celery, "-A", "celery_app", "worker", "--loglevel=info", - "-Q", "transcode", "--pool=solo", "-n", f"{WORKER_NAME}@%h"], - cwd=BASIS, stdout=log, stderr=subprocess.STDOUT, + befehl, cwd=BASIS, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + text=True, bufsize=1, errors="replace", creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) + threading.Thread( + target=_leser, args=(prozess, _bruecke_bauen(), log), + daemon=True, name="log-leser", + ).start() if icon: icon.icon = disc_bild(True) icon.update_menu() @@ -97,6 +211,8 @@ def worker_stoppen(icon=None, item=None): except subprocess.TimeoutExpired: prozess.kill() prozess = None + wach_halten(False) + aktuell["job"] = None if icon: icon.icon = disc_bild(False) icon.update_menu() @@ -109,12 +225,80 @@ def umschalten(icon, item): worker_starten(icon) +def _job_holen(): + """Was komprimiert Rippy gerade? (dict oder None) + + Gefragt wird Rippys `/jobs` — dieselbe Quelle wie das Dashboard, damit im + Tray nicht eine zweite, abweichende Zahl steht. Die Restzeit rechnet die API + (api/eta.py), hier wird sie nur angezeigt. + """ + if not RIPPY_HOST: + return None + try: + import json + import urllib.request + + with urllib.request.urlopen( + f"http://{RIPPY_HOST}/api/jobs", timeout=8 + ) as antwort: + jobs = json.load(antwort) + except Exception: + return None + for job in jobs if isinstance(jobs, list) else []: + if job.get("status") == "transcoding": + return { + "titel": job.get("title") or (job.get("id") or "")[:8], + "progress": job.get("progress") or 0, + "eta": job.get("eta_text") or "", + } + return None + + +def _beobachter(icon): + """Fragt im Takt nach, was läuft — für Menütext und Standby-Sperre.""" + while True: + job = _job_holen() if worker_laeuft() else None + aktuell["job"] = job + # Wach bleiben nur, solange wirklich etwas läuft. + wach_halten(bool(job)) + try: + icon.update_menu() + except Exception: + pass + time.sleep(ABFRAGE_TAKT_SEKUNDEN) + + +def status_text(item=None) -> str: + """Die erste Menüzeile — sagt, was die Maschine tut.""" + if not worker_laeuft(): + return f'Rippy-Worker "{WORKER_NAME}": gestoppt' + job = aktuell["job"] + if not job: + return f'Rippy-Worker "{WORKER_NAME}": bereit, nichts in Arbeit' + return f"Komprimiert: {job['titel']} — {job['progress']} %" + + +def restzeit_text(item=None) -> str: + job = aktuell["job"] + if not job: + return f"{SLOTS} Auftrag gleichzeitig" if SLOTS == 1 else f"{SLOTS} Aufträge gleichzeitig" + return job["eta"] or "Restzeit wird gemessen" + + def rippy_oeffnen(icon, item): if RIPPY_HOST: webbrowser.open(f"http://{RIPPY_HOST}") -def log_anzeigen(icon, item): +def log_in_rippy(icon, item): + """Rippys Log-Seite öffnen — Commander-Wunsch: kein txt-File.""" + if RIPPY_HOST: + webbrowser.open(f"http://{RIPPY_HOST}/logs") + + +def log_datei_anzeigen(icon, item): + """Rückfall: das lokale Log. Genau dann die einzige Auskunft, wenn Rippy + nicht erreichbar ist — deshalb bleibt der Eintrag.""" if os.path.exists(LOG_PFAD): os.startfile(LOG_PFAD) # noqa: S606 — bewusst: Standard-Editor des Nutzers @@ -125,17 +309,16 @@ def beenden(icon, item): menue = pystray.Menu( - pystray.MenuItem( - lambda item: f'Rippy-Worker "{WORKER_NAME}": {"läuft" if worker_laeuft() else "gestoppt"}', - None, enabled=False, - ), + pystray.MenuItem(status_text, None, enabled=False), + pystray.MenuItem(restzeit_text, None, enabled=False), pystray.Menu.SEPARATOR, pystray.MenuItem( lambda item: "Worker stoppen" if worker_laeuft() else "Worker starten", umschalten, ), pystray.MenuItem("Rippy öffnen", rippy_oeffnen), - pystray.MenuItem("Log anzeigen", log_anzeigen), + pystray.MenuItem("Log in Rippy anzeigen", log_in_rippy), + pystray.MenuItem("Log-Datei auf diesem PC", log_datei_anzeigen), pystray.Menu.SEPARATOR, pystray.MenuItem("Beenden (stoppt den Worker)", beenden), ) @@ -145,4 +328,12 @@ if __name__ == "__main__": tray = pystray.Icon("rippy-worker", disc_bild(False), "Rippy Worker", menue) worker_starten() tray.icon = disc_bild(worker_laeuft()) - tray.run() + threading.Thread( + target=_beobachter, args=(tray,), daemon=True, name="job-beobachter" + ).start() + try: + tray.run() + finally: + # Auch bei einem harten Ende die Standby-Sperre zurücknehmen — sonst + # schläft der PC nie wieder ein, und niemand weiß warum. + wach_halten(False)