import os import subprocess import sys import threading import time import urllib.request import json import webbrowser import flet as ft # Tray-Icon: pystray + PIL fuer das Icon-Bild try: import pystray from PIL import Image as PILImage TRAY_VERFUEGBAR = True except ImportError: TRAY_VERFUEGBAR = False # Environment configuration BASIS = os.path.dirname(os.path.abspath(__file__)) RIPPY_HOST = os.getenv("RIPPY_TRAY_HOST", "") or os.getenv("RIPPY_HOST", "") WORKER_NAME = os.getenv("WORKER_NAME", "windows-worker") SLOTS = max(1, int(os.getenv("RIPPY_SLOTS", "1") or "1")) ABFRAGE_TAKT_SEKUNDEN = 4 # Log file path def _log_pfad() -> str: basis_daten = os.getenv("LOCALAPPDATA") or os.getenv("APPDATA") or "" if basis_daten: ordner = os.path.join(basis_daten, "Rippy Worker") os.makedirs(ordner, exist_ok=True) return os.path.join(ordner, "worker.log") return os.path.join(BASIS, "worker.log") LOG_PFAD = _log_pfad() # State variables prozess = None aktuell = {"job": None, "online": False} # Windows Standby blocker _wach = {"an": False} def wach_halten(an: bool) -> None: if _wach["an"] == an: return try: import ctypes ES_CONTINUOUS = 0x80000000 ES_SYSTEM_REQUIRED = 0x00000001 flags = (ES_CONTINUOUS | ES_SYSTEM_REQUIRED) if an else ES_CONTINUOUS ctypes.windll.kernel32.SetThreadExecutionState(flags) _wach["an"] = an except Exception: pass # Celery Subprocess Management def _leser(prozess_ref, datei, log_callback) -> None: try: for zeile in prozess_ref.stdout: try: datei.write(zeile) datei.flush() except OSError: pass if log_callback: log_callback(zeile) except Exception: pass finally: try: datei.close() except OSError: pass def worker_laeuft() -> bool: return prozess is not None and prozess.poll() is None def worker_starten(log_callback=None): global prozess if worker_laeuft(): return celery = os.path.join(BASIS, "venv", "Scripts", "celery.exe") if not os.path.exists(celery): celery = "celery" # fallback to global if testing outside venv 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( 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, log, log_callback), daemon=True, name="log-leser", ).start() def worker_stoppen(): global prozess if worker_laeuft(): prozess.terminate() try: prozess.wait(15) except subprocess.TimeoutExpired: prozess.kill() prozess = None wach_halten(False) aktuell["job"] = None # API Polling def hole(pfad: str, host: str = None, timeout: int = 5): ziel = host if host is not None else RIPPY_HOST if not ziel: return None try: with urllib.request.urlopen(f"http://{ziel}/api{pfad}", timeout=timeout) as a: return json.load(a) except Exception: return None def hole_job_und_zustand(): caps = hole("/capabilities") jobs = hole("/jobs") or [] online = False w_info = {} for w in ((caps or {}).get("workers") or []): if w.get("name") == WORKER_NAME: online = bool(w.get("online")) w_info = w break laufend = None for job in jobs if isinstance(jobs, list) else []: if job.get("status") in ("transcoding", "processing", "running"): laufend = job break aktuell["online"] = online aktuell["job"] = laufend wach_halten(bool(laufend)) return online, laufend, caps, jobs, w_info def deinstallieren() -> None: skript = os.path.join(BASIS, "uninstall.ps1") if not os.path.isfile(skript): return subprocess.Popen( ["powershell.exe", "-NoProfile", "-ExecutionPolicy", "Bypass", "-File", skript], creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) def _bruecke_bauen(): 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 _schluessel_wache(bruecke=None, log_callback=None): try: import schluessel if not schluessel.makemkvcon_pfad(): return def melden(level, text): msg = f"[schluessel] {level.upper()} {text}" if bruecke: bruecke.zeile(msg) if log_callback: log_callback(msg + "\n") stand = None while True: try: stand, _was = schluessel.runde(RIPPY_HOST, stand, melden=melden) except Exception as e: if log_callback: log_callback(f"[schluessel] ERROR Schlüssel-Automatik fehlgeschlagen: {e}\n") time.sleep(schluessel.TAKT_SEKUNDEN) except ImportError: pass # GUI Application def main(page: ft.Page): page.title = f"Rippy Worker — {WORKER_NAME}" page.window.width = 900 page.window.height = 700 page.bgcolor = "#0f172a" # Rippy Slate-900 background page.padding = 30 page.theme_mode = ft.ThemeMode.DARK page.fonts = { "Inter": "https://fonts.googleapis.com/css2?family=Inter:wght@400;500;600;700&display=swap" } page.theme = ft.Theme(font_family="Inter") # --- Tray-Icon ----------------------------------------------------------- tray_icon_ref = {"icon": None} def _tray_icon_bild(): """Lade rippy.ico oder erzeuge ein Fallback-Bild.""" ico_pfad = os.path.join(BASIS, "rippy.ico") if os.path.isfile(ico_pfad): try: return PILImage.open(ico_pfad) except Exception: pass # Fallback: kleines lila Quadrat img = PILImage.new("RGB", (64, 64), (99, 102, 241)) return img def tray_zeigen(icon, item): """Tray-Menue: Fenster wieder anzeigen.""" page.window.visible = True page.window.minimized = False page.update() def tray_beenden(icon, item): """Tray-Menue: App komplett beenden.""" if tray_icon_ref["icon"]: tray_icon_ref["icon"].stop() worker_stoppen() page.window.destroy() def tray_starten(): """Erstellt und startet das System-Tray-Icon.""" if not TRAY_VERFUEGBAR: return menu = pystray.Menu( pystray.MenuItem("Rippy Worker anzeigen", tray_zeigen, default=True), pystray.Menu.SEPARATOR, pystray.MenuItem("Beenden", tray_beenden), ) icon = pystray.Icon("rippy_worker", _tray_icon_bild(), f"Rippy Worker — {WORKER_NAME}", menu) tray_icon_ref["icon"] = icon threading.Thread(target=icon.run, daemon=True, name="tray-icon").start() def on_window_event(e): """Fenster-Schliessen minimiert ins Tray statt die App zu beenden.""" if e.data == "close": if TRAY_VERFUEGBAR and tray_icon_ref["icon"]: page.window.visible = False page.update() else: # Kein Tray verfuegbar — wirklich beenden worker_stoppen() page.window.destroy() page.window.prevent_close = True page.window.on_event = on_window_event tray_starten() # Colors c_bg = "#0f172a" c_panel = "#1e293b" c_text = "#e2e8f0" c_muted = "#94a3b8" c_accent = "#6366f1" # Indigo c_success = "#10b981" # Emerald c_warning = "#f59e0b" # Amber c_danger = "#f43f5e" # Rose # Formatter for job status def job_text(job: dict) -> str: titel = (job.get("title") or (job.get("id") or "")[:8]) or "?" status = { "transcoding": "komprimiert", "processing": "rippt", "running": "rippt", "pending": "wartet", "completed": "fertig", "failed": "Fehler", "canceling": "bricht ab", }.get(job.get("status") or "", job.get("status") or "?") return f"{titel} — {status}" # UI Components status_text = ft.Text("Verbinde...", color=c_muted, size=18, weight=ft.FontWeight.BOLD) title_text = ft.Text(f"Rippy Worker: {WORKER_NAME}", size=24, weight=ft.FontWeight.BOLD, color=c_text) machine_info = ft.Text("System-Infos werden geladen...", color=c_muted, size=13, font_family="Consolas") progress_ring = ft.ProgressRing(width=120, height=120, stroke_width=10, value=0, color=c_accent) progress_percentage = ft.Text("0%", size=28, weight=ft.FontWeight.BOLD, color=c_text) job_title = ft.Text("Nichts in Arbeit", size=20, color=c_muted) eta_text = ft.Text("Bereit", size=14, color=c_muted) progress_stack = ft.Stack( [ progress_ring, ft.Container( content=progress_percentage, alignment=ft.alignment.center, width=120, height=120, ) ], width=120, height=120, ) log_view = ft.ListView(expand=1, spacing=2, auto_scroll=True) queue_view = ft.ListView(height=120, spacing=4, auto_scroll=False) def add_log(zeile): # Zeilen anfügen und limitieren log_view.controls.append(ft.Text(zeile.strip(), color=c_muted, size=12, font_family="Consolas")) if len(log_view.controls) > 200: log_view.controls.pop(0) try: page.update() except Exception: pass def toggle_worker(e): if worker_laeuft(): worker_stoppen() btn_toggle.text = "Worker Starten" btn_toggle.icon = ft.icons.PLAY_ARROW btn_toggle.bgcolor = c_success else: worker_starten(add_log) btn_toggle.text = "Worker Stoppen" btn_toggle.icon = ft.icons.STOP btn_toggle.bgcolor = c_danger page.update() btn_toggle = ft.ElevatedButton( text="Worker Stoppen" if worker_laeuft() else "Worker Starten", icon=ft.icons.STOP if worker_laeuft() else ft.icons.PLAY_ARROW, bgcolor=c_danger if worker_laeuft() else c_success, color="white", on_click=toggle_worker ) def uninstall_click(e): def close_dlg(e): dlg.open = False page.update() def do_uninstall(e): deinstallieren() page.window.close() dlg = ft.AlertDialog( title=ft.Text("Deinstallieren?"), content=ft.Text(f"Worker '{WORKER_NAME}' wirklich entfernen? Das beendet den Prozess und entfernt ihn aus dem Autostart."), actions=[ ft.TextButton("Abbrechen", on_click=close_dlg), ft.TextButton("Ja, Deinstallieren", on_click=do_uninstall, style=ft.ButtonStyle(color=c_danger)), ], actions_alignment=ft.MainAxisAlignment.END, ) page.overlay.append(dlg) dlg.open = True page.update() btn_uninstall = ft.TextButton( text="Deinstallieren", icon=ft.icons.DELETE, icon_color=c_danger, on_click=uninstall_click ) btn_dashboard = ft.TextButton( text="Rippy Dashboard", icon=ft.icons.OPEN_IN_BROWSER, on_click=lambda _: webbrowser.open(f"http://{RIPPY_HOST}") if RIPPY_HOST else None ) btn_logs = ft.TextButton( text="Log in Rippy", icon=ft.icons.FORMAT_ALIGN_LEFT, on_click=lambda _: webbrowser.open(f"http://{RIPPY_HOST}/logs") if RIPPY_HOST else None ) btn_local_log = ft.TextButton( text="Lokales Log", icon=ft.icons.INSERT_DRIVE_FILE, on_click=lambda _: os.startfile(LOG_PFAD) if os.path.exists(LOG_PFAD) else None ) # Layout header = ft.Row( [title_text, ft.Container(expand=True), btn_dashboard, btn_logs, btn_local_log, btn_uninstall], alignment=ft.MainAxisAlignment.SPACE_BETWEEN ) dashboard_card = ft.Container( content=ft.Column([ status_text, machine_info, ft.Container(height=5), ft.Row([ progress_stack, ft.Column([job_title, eta_text], alignment=ft.MainAxisAlignment.CENTER, spacing=10) ], alignment=ft.MainAxisAlignment.START, spacing=30), ft.Container(height=10), btn_toggle ]), bgcolor=c_panel, border_radius=15, padding=30, border=ft.border.all(1, "#334155") ) queue_card = ft.Container( content=ft.Column([ ft.Text("Aufgaben (Warteschlange)", size=16, weight=ft.FontWeight.BOLD, color=c_text), queue_view ]), bgcolor="#020617", border_radius=10, padding=15, border=ft.border.all(1, "#1e293b") ) terminal_card = ft.Container( content=log_view, bgcolor="#020617", border_radius=10, padding=15, expand=True, border=ft.border.all(1, "#1e293b") ) page.add( header, ft.Container(height=10), dashboard_card, ft.Container(height=10), queue_card, ft.Container(height=10), ft.Text("Live Logs", size=16, weight=ft.FontWeight.BOLD, color=c_text), terminal_card ) # Background polling loop def poll_loop(): while True: try: online, job, caps, jobs, w_info = hole_job_und_zustand() # Update System Info info = w_info.get("info") or {} teile = [t for t in ( info.get("cpu_modell"), f"{info.get('cpu_kerne')} Kerne" if info.get("cpu_kerne") else "", info.get("cpu_simd") if info.get("cpu_simd") not in ("", "unbekannt") else "", ", ".join(w_info.get("encoders") or []), ) if t] machine_info.value = " · ".join(teile) if teile else f"{SLOTS} Aufträge gleichzeitig möglich" # Update Queue queue_view.controls.clear() if not jobs: queue_view.controls.append(ft.Text("Keine Aufgaben in Rippy", color=c_muted, font_family="Consolas", size=13)) else: for j in jobs[:8]: queue_view.controls.append(ft.Text(" " + job_text(j), color=c_muted, font_family="Consolas", size=13)) # Update status if not worker_laeuft(): status_text.value = "Worker-Prozess gestoppt" status_text.color = c_danger progress_ring.value = 0 progress_percentage.value = "0%" job_title.value = "Nichts in Arbeit" eta_text.value = "Bereit" elif not online: status_text.value = "Nicht mit Rippy verbunden (offline)" status_text.color = c_warning progress_ring.value = 0 progress_percentage.value = "0%" elif job: status_text.value = "Aktiv: Komprimiert" status_text.color = c_success prog = float(job.get("progress") or 0) progress_ring.value = prog / 100.0 progress_percentage.value = f"{int(prog)}%" titel = job.get("title") or (job.get("id") or "")[:8] job_title.value = titel eta = job.get("eta_text") or "Restzeit wird berechnet..." eta_text.value = f"Verbleibend: {eta}" else: status_text.value = "Verbunden & Bereit" status_text.color = c_accent progress_ring.value = 0 progress_percentage.value = "0%" job_title.value = "Warte auf Aufträge..." eta_text.value = "Nichts in Arbeit" page.update() except Exception as e: pass time.sleep(ABFRAGE_TAKT_SEKUNDEN) # Start Celery automatically worker_starten(add_log) # Init initial logs if os.path.exists(LOG_PFAD): try: with open(LOG_PFAD, "r", encoding="utf-8", errors="replace") as f: lines = f.readlines()[-50:] for line in lines: add_log(line) except OSError: pass threading.Thread(target=poll_loop, daemon=True).start() threading.Thread(target=_schluessel_wache, args=(_bruecke_bauen(), add_log), daemon=True, name="schluessel-wache").start() def worker_stoppen(): """Celery-Prozess sauber beenden.""" global prozess if prozess and prozess.poll() is None: try: prozess.terminate() prozess.wait(timeout=5) except Exception: try: prozess.kill() except Exception: pass if __name__ == "__main__": ft.app(target=main)