import argparse import time import requests import subprocess import os import re import shutil import threading import sys import platform import json import signal import zipfile import tarfile import gzip import traceback import ctypes import hashlib from datetime import datetime, timedelta # Textual TUI (optional — install with: pip install textual) # Minimum required version: 0.20 (RichLog, ModalScreen, Binding, DataTable etc.) _TEXTUAL_MIN_VERSION = (0, 20) _VENV_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), '.worker_venv') def _bootstrap_textual(): """Auto-install/upgrade textual if missing or too old, then re-exec. Strategy: 1. Try plain pip install into the current interpreter (works on most systems). 2. If pip rejects it (externally-managed-environment / PEP 668), create a local venv at _VENV_DIR and install there, then re-exec with that Python. """ try: import textual as _t try: from importlib.metadata import version as _pkg_version _tv_str = _pkg_version('textual') except Exception: _tv_str = getattr(_t, '__version__', '0.0') if tuple(int(x) for x in _tv_str.split('.')[:2]) >= _TEXTUAL_MIN_VERSION: return # already fine reason = f"textual {_tv_str} is too old (>= {'.'.join(str(x) for x in _TEXTUAL_MIN_VERSION)} required)" except ImportError: reason = "textual not found" print(f"[*] {reason} — attempting automatic install...") # Attempt 1: plain pip upgrade into current interpreter result = subprocess.run( [sys.executable, '-m', 'pip', 'install', '--upgrade', 'textual'], capture_output=True, text=True, ) if result.returncode == 0: print("[*] textual installed — restarting...") os.execv(sys.executable, [sys.executable] + sys.argv) # Attempt 2: create/reuse a local venv and install there print("[*] pip install failed (system-managed env?) — trying local venv...") try: import venv as _venv if not os.path.isdir(_VENV_DIR): print(f"[*] Creating venv at {_VENV_DIR} ...") _venv.create(_VENV_DIR, with_pip=True) venv_python = ( os.path.join(_VENV_DIR, 'Scripts', 'python.exe') # Windows if sys.platform == 'win32' else os.path.join(_VENV_DIR, 'bin', 'python') ) result2 = subprocess.run( [venv_python, '-m', 'pip', 'install', '--upgrade', 'textual', 'requests'], capture_output=True, text=True, ) if result2.returncode == 0: print(f"[*] textual installed in venv — restarting under {venv_python} ...") os.execv(venv_python, [venv_python] + sys.argv) else: print(f"[!] venv pip install failed:\n{result2.stderr.strip()}") except Exception as _ve: print(f"[!] venv setup failed: {_ve}") print("[!] Could not install textual automatically — continuing without TUI.") _bootstrap_textual() try: import textual as _textual_mod try: from importlib.metadata import version as _pkg_version _tv_str = _pkg_version('textual') except Exception: _tv_str = getattr(_textual_mod, '__version__', '0.0') _tv = tuple(int(x) for x in _tv_str.split('.')[:2]) if _tv < _TEXTUAL_MIN_VERSION: raise ImportError( f"textual {_tv_str} is too old; " f">= {'.'.join(str(x) for x in _TEXTUAL_MIN_VERSION)} required" ) from textual.app import App, ComposeResult from textual.widgets import Header, Footer, RichLog, DataTable, Button, Label from textual.screen import ModalScreen from textual.containers import Horizontal, Vertical from textual.binding import Binding HAS_TEXTUAL = True _TEXTUAL_UNAVAIL_REASON = "" except ImportError as _e: HAS_TEXTUAL = False _TEXTUAL_UNAVAIL_REASON = str(_e) # ============================================================================== # CONFIGURATION # ============================================================================== DEFAULT_MANAGER_URL = "https://encode.fractumseraph.net/" DEFAULT_USERNAME = "Anonymous" DEFAULT_WORKERNAME = f"Node-{int(time.time())}" WORKER_VERSION = "3.0.26" WORKER_SECRET = os.environ.get("WORKER_SECRET", "DefaultInsecureSecret") SHUTDOWN_EVENT = threading.Event() UPDATE_AVAILABLE = False LAST_UPDATE_CHECK = 0 CHECK_LOCK = threading.Lock() CONSOLE_LOCK = threading.Lock() PROGRESS_LOCK = threading.Lock() MONITOR_PAUSED = threading.Event() WORKER_PROGRESS = {} PAUSE_REQUESTED = False ACTIVE_PROCS = {} PROC_LOCK = threading.Lock() TUI_APP = None # Set to WorkerApp instance when running in TUI mode _TUI_WAS_RUNNING = False # Set True before app.run(), never cleared; guards safe_print post-TUI _TUI_SIGNAL = None # Set to 'pause' or 'quit' by signal handler, polled by app timer # Per-worker rich state, polled by the TUI every 0.5s # {"file": str, "phase": str, "pct": int, "job_start": float, "jobs_done": int} WORKER_DETAILS = {} SESSION_STATS = {"jobs_done": 0, "bytes_uploaded": 0, "start": 0.0} STATS_LOCK = threading.Lock() # Global paths for executables FFMPEG_CMD = "ffmpeg" FFPROBE_CMD = "ffprobe" _FFMPEG_META_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), '.ffmpeg_meta.json') # Detect OS to handle Fonts _script_dir = os.path.dirname(os.path.abspath(__file__)) ENCODING_CONFIG = { "VIDEO_CODEC": "libsvtav1", "VIDEO_PRESET": "2", "VIDEO_CRF": "63", "VIDEO_PIX_FMT": "yuv420p", "VIDEO_SCALE": "scale=-2:480", "AUDIO_CODEC": "libopus", "AUDIO_BITRATE": "24k", # Perfect for Mono Speech "AUDIO_CHANNELS": "1", # CHANGED: 2 -> 1 (Mono) for better quality "SUBTITLE_CODEC": "mov_text", "OUTPUT_EXT": ".mp4" } class QuotaTracker: def __init__(self, limit_gb, worker_name): self.limit_bytes = int(limit_gb * 1024**3) if limit_gb > 0 else 0 self.filename = f"usage_{re.sub(r'[^a-zA-Z0-9]', '', worker_name)}.json" self.lock = threading.Lock() self.current_usage = 0 self.last_save = 0 self._load() def _load(self): today = datetime.now().strftime("%Y-%m-%d") if os.path.exists(self.filename): try: with open(self.filename, 'r') as f: data = json.load(f) if data.get('date') == today: self.current_usage = data.get('bytes', 0) else: self.current_usage = 0 self._save() except: self.current_usage = 0 else: self.current_usage = 0 def _save(self): today = datetime.now().strftime("%Y-%m-%d") try: with open(self.filename, 'w') as f: json.dump({"date": today, "bytes": self.current_usage}, f) except: pass def check_cap(self): if self.limit_bytes <= 0: return False today = datetime.now().strftime("%Y-%m-%d") if os.path.exists(self.filename): try: with open(self.filename, 'r') as f: data = json.load(f) if data.get('date') != today: with self.lock: self.current_usage = 0 self._save() return False except: pass with self.lock: return self.current_usage >= self.limit_bytes def add_usage(self, num_bytes): if self.limit_bytes <= 0: return with self.lock: self.current_usage += num_bytes if time.time() - self.last_save > 30: self._save() self.last_save = time.time() def force_save(self): with self.lock: self._save() def get_remaining_str(self): if self.limit_bytes <= 0: return "Unlimited" rem = self.limit_bytes - self.current_usage if rem < 0: rem = 0 return f"{rem / 1024**3:.2f} GB" def get_wait_time(self): now = datetime.now() tomorrow = now + timedelta(days=1) midnight = datetime(year=tomorrow.year, month=tomorrow.month, day=tomorrow.day, hour=0, minute=0, second=1) return (midnight - now).total_seconds() # ============================================================================== # TEXTUAL TUI # ============================================================================== if HAS_TEXTUAL: from rich.text import Text # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _phase_style(phase: str) -> str: return { "Idle": "dim", "Starting": "dim", "Probe": "cyan", "DL": "bold cyan", "Encoding": "bold yellow", "Uploading": "bold green", "Done": "bright_green", "Failed": "bold red", "Paused": "bold orange3", "Quota": "orange_red1", "Retrying": "bold orange3", }.get(phase, "white") def _make_bar(pct: int, width: int = 16) -> str: pct = max(0, min(100, pct)) filled = int(width * pct / 100) return "▓" * filled + "░" * (width - filled) + f" {pct:3d}%" def _fmt_elapsed(seconds: float) -> str: if seconds <= 0: return "-" s = int(seconds) h, r = divmod(s, 3600) m, sec = divmod(r, 60) return f"{h}:{m:02d}:{sec:02d}" if h else f"{m}:{sec:02d}" # --------------------------------------------------------------------------- # Pause modal # --------------------------------------------------------------------------- class PauseModal(ModalScreen): """Modal dialog shown when the worker is paused.""" BINDINGS = [ # type: ignore[assignment] Binding("c", "choose_continue", show=False), Binding("f", "choose_finish", show=False), Binding("s", "choose_stop", show=False), Binding("escape", "choose_continue", show=False), ] def compose(self) -> ComposeResult: on_windows = platform.system() == 'Windows' subtitle = "FFmpeg has been suspended." if not on_windows else "FFmpeg has been suspended (Windows)." with Vertical(id="pause-dialog"): yield Label("\u23f8 WORKER PAUSED", id="pause-title") yield Label(subtitle, id="pause-subtitle") with Horizontal(id="pause-buttons"): yield Button("[C] Continue", id="btn-continue", variant="success") yield Button("[F] Finish", id="btn-finish", variant="warning") yield Button("[S] Stop", id="btn-stop", variant="error") def action_choose_continue(self) -> None: self.dismiss("continue") def action_choose_finish(self) -> None: self.dismiss("finish") def action_choose_stop(self) -> None: self.dismiss("stop") def on_key(self, event) -> None: """Explicit key handler — belt-and-suspenders in case BINDINGS don't fire (known issue in some Textual versions over SSH/tmux).""" key = event.key if key in ('c', 'escape'): self.dismiss("continue") elif key == 'f': self.dismiss("finish") elif key == 's': self.dismiss("stop") def on_button_pressed(self, event: Button.Pressed) -> None: choices = {"btn-continue": "continue", "btn-finish": "finish", "btn-stop": "stop"} if event.button.id in choices: self.dismiss(choices[event.button.id]) # --------------------------------------------------------------------------- # Main application # --------------------------------------------------------------------------- class WorkerApp(App): """Fractum Distributed Worker \u2014 Textual TUI.""" CSS = """ Screen { background: $background; } #workers-table { height: auto; max-height: 16; margin: 1 1 0 1; border: solid $primary; } #stats-bar { height: 1; margin: 0 2; background: $surface-darken-1; color: $text-muted; padding: 0 1; } #log-panel { height: 1fr; margin: 0 1 1 1; border: solid $primary-darken-2; } PauseModal { align: center middle; } #pause-dialog { background: $surface; border: thick $warning; padding: 1 4; width: 56; height: auto; } #pause-title { text-align: center; color: $warning; text-style: bold; padding-bottom: 1; } #pause-subtitle { text-align: center; color: $text-disabled; padding-bottom: 1; } #pause-buttons { height: auto; align: center middle; padding-top: 1; } #btn-continue { margin: 0 1; } #btn-finish { margin: 0 1; } #btn-stop { margin: 0 1; } """ BINDINGS = [ # type: ignore[assignment] Binding("p", "request_pause", "Pause [P]", show=True, priority=True), Binding("q", "request_quit", "Quit [Q]", show=True, priority=True), ] def __init__(self, worker_ids: list, threads: list, quota_tracker=None, manager_url: str = "", **kwargs): super().__init__(**kwargs) self._worker_ids = worker_ids self._threads = threads self._quota_tracker = quota_tracker self._manager_url = manager_url # Column keys assigned in on_mount self._col_file = self._col_phase = self._col_bar = None self._col_elapsed = self._col_done = self._col_eta = None # Monotonic timestamp recorded when the pause modal is first shown. # Used by _check_signals to distinguish a genuine second press (force-stop) # from the race-condition echo of the *same* keypress handled by both # Textual's binding and the _tty_key_reader fallback simultaneously. self._pause_requested_at = 0.0 def compose(self) -> ComposeResult: yield Header() # can_focus=False here (not just in on_mount) prevents the DataTable from # receiving keyboard focus before on_mount runs. In recent Textual versions # DataTable captures unrecognised keys for its internal row-search feature # instead of bubbling them, so focus on startup caused the first 'p' press # to be swallowed and the second press to actually trigger the App binding. yield DataTable(id="workers-table", show_cursor=False) yield Label("", id="stats-bar", markup=False) yield RichLog(id="log-panel", highlight=True, markup=False, max_lines=1000) yield Footer() def on_mount(self) -> None: table = self.query_one("#workers-table", DataTable) cols = table.add_columns("Worker", "Current File", "Phase", "Progress", "Elapsed", "Done", "ETA") _cw, self._col_file, self._col_phase, self._col_bar, self._col_elapsed, self._col_done, self._col_eta = cols for wid in self._worker_ids: table.add_row(wid, "-", "Starting", _make_bar(0), "-", "0", "-", key=wid) self.title = "Fractum Distributed Worker" url_display = self._manager_url or "no manager" self.sub_title = f"v{WORKER_VERSION} \u2502 {len(self._worker_ids)} worker(s) \u2502 {url_display}" self.set_interval(0.5, self._tick) self.set_interval(0.25, self._check_signals) self.set_interval(1.5, self._check_all_done) # Prevent these widgets from stealing focus and swallowing key events. # Belt-and-suspenders: can_focus=False is also set in the constructor, # but re-set here for safety in case Textual ever resets it internally. self.query_one("#workers-table").can_focus = False self.query_one("#log-panel").can_focus = False # Drop focus from any widget so the App (and its BINDINGS) receives # key events directly from the first keypress, not the second. self.set_focus(None) # NOTE: do NOT write raw escape sequences to sys.stdout here. # Textual owns the terminal once on_mount fires; writing outside its # rendering pipeline corrupts its cursor-position tracking and causes # subsequent renders to land at wrong positions (e.g. on the stats bar). # Mouse-tracking cleanup is handled exclusively by the atexit handler # and the reset block that runs after app.run() returns. # Start a fallback key reader on /dev/tty. On some SSH/tmux setups # Textual's own input driver receives no key events; reading /dev/tty # directly bypasses that entirely. We only read p/q here so we # don't conflict with Textual's normal key handling in healthy envs. threading.Thread(target=self._tty_key_reader, daemon=True).start() def _tty_key_reader(self) -> None: """Read single chars from /dev/tty to set _TUI_SIGNAL as a fallback for environments where Textual's input driver is broken (SSH/tmux).""" global _TUI_SIGNAL try: import tty, termios, atexit tty_fd = open('/dev/tty', 'rb', buffering=0) old = termios.tcgetattr(tty_fd) # Register atexit BEFORE entering cbreak mode so the terminal is # always restored even if the daemon thread is killed mid-read on # program exit (blocking read(1) won't unblock for the finally). def _restore_tty(): try: termios.tcsetattr(tty_fd, termios.TCSADRAIN, old) tty_fd.close() except Exception: pass atexit.register(_restore_tty) tty.setcbreak(tty_fd) try: while not SHUTDOWN_EVENT.is_set(): ch = tty_fd.read(1) if not ch: break c = ch.decode('utf-8', errors='ignore').lower() if PAUSE_REQUESTED: # While paused, route c/f/s as pause-choice signals if c == 'c': _TUI_SIGNAL = 'choice:continue' elif c == 'f': _TUI_SIGNAL = 'choice:finish' elif c == 's': _TUI_SIGNAL = 'choice:stop' else: if c == 'p' and _TUI_SIGNAL is None: _TUI_SIGNAL = 'pause' elif c == 'q' and _TUI_SIGNAL is None: _TUI_SIGNAL = 'quit' finally: # Normal clean exit — unregister the atexit handler and # restore immediately so it doesn't run a second time. atexit.unregister(_restore_tty) _restore_tty() except Exception: pass # Not available on Windows or if /dev/tty is inaccessible # ------------------------------------------------------------------ # Periodic refresh # ------------------------------------------------------------------ def _tick(self) -> None: """Refresh the workers table and stats bar from shared state.""" table = self.query_one("#workers-table", DataTable) now = time.time() for wid in self._worker_ids: d = WORKER_DETAILS.get(wid, {}) phase = d.get("phase", "Starting") pct = d.get("pct", 0) filename = d.get("file", "-") job_start = d.get("job_start", 0.0) jobs_done = d.get("jobs_done", 0) idle_phases = {"Idle", "Starting", "Quota", "Done", "Failed"} elapsed = (now - job_start) if job_start > 0 and phase not in idle_phases else 0.0 # ETA: estimate remaining time from elapsed and progress if pct > 5 and elapsed > 0 and phase == "Encoding": eta_sec = elapsed * (100 - pct) / pct else: eta_sec = 0.0 style = _phase_style(phase) phase_text = Text(phase, style=style) bar_text = Text(_make_bar(pct) if phase not in {"Idle", "Starting"} else " " * 20, style=style) max_fn = 32 fn_display = filename if len(filename) <= max_fn else "\u2026" + filename[-(max_fn - 1):] try: table.update_cell(wid, self._col_file, fn_display, update_width=False) table.update_cell(wid, self._col_phase, phase_text, update_width=False) table.update_cell(wid, self._col_bar, bar_text, update_width=False) table.update_cell(wid, self._col_elapsed, _fmt_elapsed(elapsed), update_width=False) table.update_cell(wid, self._col_done, str(jobs_done), update_width=False) table.update_cell(wid, self._col_eta, _fmt_elapsed(eta_sec), update_width=False) except Exception: pass # Stats bar with STATS_LOCK: jd = SESSION_STATS.get("jobs_done", 0) bu = SESSION_STATS.get("bytes_uploaded", 0) st = SESSION_STATS.get("start", now) uptime_str = _fmt_elapsed(now - st) gb_str = f"{bu / 1024 ** 3:.2f} GB" quota_str = "" if self._quota_tracker: quota_str = f" \u2502 Quota Remaining: {self._quota_tracker.get_remaining_str()}" stats = ( f" Jobs Completed: {jd}" f" \u2502 Uploaded: {gb_str}" f" \u2502 Uptime: {uptime_str}" f"{quota_str}" ) try: self.query_one("#stats-bar", Label).update(stats) except Exception: pass async def _check_signals(self) -> None: """Poll the signal flag set by the OS signal handler or tty key reader.""" global _TUI_SIGNAL sig = _TUI_SIGNAL if sig is None: return _TUI_SIGNAL = None if sig == 'pause': if not PAUSE_REQUESTED: await self.action_request_pause() else: # Second Ctrl+C / P while already paused = force stop. # Debounce: if the modal was shown less than 0.5 s ago, this # signal is almost certainly the same keypress being handled by # both Textual's binding and the _tty_key_reader fallback at the # same time (race condition). Ignore it; the modal stays open. if time.monotonic() - self._pause_requested_at < 0.5: return if isinstance(self.screen, PauseModal): self.screen.dismiss("stop") else: self._handle_pause_result('stop') elif sig == 'quit': self.action_request_quit() elif sig.startswith('choice:'): choice = sig[len('choice:'):] # SSH/tmux keyboard fallback: dismiss the modal so the callback # fires automatically, rather than calling _handle_pause_result # directly (which would leave the modal open on screen). if isinstance(self.screen, PauseModal): self.screen.dismiss(choice) else: self._handle_pause_result(choice) def _check_all_done(self) -> None: if SHUTDOWN_EVENT.is_set() and all(not t.is_alive() for t in self._threads): self.exit() # ------------------------------------------------------------------ # Thread-safe log write (called via call_from_thread) # ------------------------------------------------------------------ def write_log(self, message: str) -> None: try: self.query_one("#log-panel", RichLog).write(message) except Exception: pass def update_worker_status(self, worker_id: str, status: str) -> None: # Status is now driven by WORKER_DETAILS polling in _tick; this is a no-op. pass # ------------------------------------------------------------------ # Pause / quit actions # ------------------------------------------------------------------ async def action_request_pause(self) -> None: global PAUSE_REQUESTED if PAUSE_REQUESTED: return PAUSE_REQUESTED = True self._pause_requested_at = time.monotonic() threading.Thread(target=lambda: toggle_processes(suspend=True), daemon=True).start() # Show the popup modal — buttons are mouse-clickable on Windows; # keyboard C/F/S work via Textual bindings or the _tty_key_reader # fallback for SSH/tmux (which dismisses the modal via _check_signals). self.push_screen(PauseModal(), callback=self._handle_pause_result) def _handle_pause_result(self, choice) -> None: global PAUSE_REQUESTED if choice == "continue": PAUSE_REQUESTED = False threading.Thread(target=lambda: toggle_processes(suspend=False), daemon=True).start() self.write_log("[*] Encoding resumed.") elif choice == "finish": PAUSE_REQUESTED = False threading.Thread(target=lambda: toggle_processes(suspend=False), daemon=True).start() SHUTDOWN_EVENT.set() self.write_log("[*] Finishing active jobs, then stopping...") elif choice == "stop": threading.Thread(target=lambda: toggle_processes(suspend=False), daemon=True).start() kill_processes() SHUTDOWN_EVENT.set() PAUSE_REQUESTED = False self.write_log("[*] Stopping immediately...") self.set_timer(1.5, self.exit) def action_request_quit(self) -> None: SHUTDOWN_EVENT.set() kill_processes() self.exit() def _run_text_pause_menu() -> str: """Synchronous text pause menu shown when the TUI is suspended. Reads directly from /dev/tty so it works even when stdin is piped. Returns 'continue', 'finish', or 'stop'. """ src = None if not sys.stdin.isatty(): try: src = open('/dev/tty', 'r') except Exception: pass try: sys.stdout.write("\n" + "="*40 + "\n") sys.stdout.write(" [!] WORKER PAUSED\n") sys.stdout.write("="*40 + "\n") sys.stdout.write(" [C]ontinue - Resume encoding\n") sys.stdout.write(" [F]inish - Finish active, then stop\n") sys.stdout.write(" [S]top - Abort immediately\n") sys.stdout.write(" Ctrl+C - Force abort\n") sys.stdout.flush() while True: try: prompt_src = src if src else sys.stdin sys.stdout.write("Select [c/f/s]: ") sys.stdout.flush() line = prompt_src.readline() if not line: return 'stop' ch = line.strip().lower() if ch in ('c', ''): return 'continue' if ch == 'f': return 'finish' if ch == 's': return 'stop' except KeyboardInterrupt: sys.stdout.write("\n[*] Force abort.\n") sys.stdout.flush() return 'stop' except EOFError: return 'stop' finally: if src: try: src.close() except Exception: pass def get_auth_headers(): headers = {'User-Agent': f'FractumWorker/{WORKER_VERSION}'} if WORKER_SECRET: headers['X-Worker-Token'] = WORKER_SECRET return headers def get_term_width(): try: return shutil.get_terminal_size((80, 20)).columns except: return 80 def safe_print(message): # During interpreter shutdown daemon threads can't safely acquire the # internal C-level lock on BufferedWriter, causing a Fatal Python error. # Bail out early — nobody is reading stdout at that point anyway. if sys.is_finalizing(): return if TUI_APP is not None: try: TUI_APP.call_from_thread(TUI_APP.write_log, message) except Exception: pass # Never fall through to direct stdout writes while the TUI owns the # terminal — writing raw bytes into Textual's alternate screen places # stray characters wherever the cursor happens to be (often top-left # during a render cycle). return # After the TUI exits (TUI_APP set back to None) but before Python fully # finalizes, daemon threads can briefly reach sys.stdout.write(). The # C-level BufferedWriter lock may already be destroyed by then, causing # "Fatal Python error: _enter_buffered_busy" which cannot be caught. # sys.is_finalizing() has a race window — SHUTDOWN_EVENT closes it. if _TUI_WAS_RUNNING and SHUTDOWN_EVENT.is_set(): return if not CONSOLE_LOCK.acquire(timeout=2): return try: width = get_term_width() # Truncate long messages to avoid wrapping onto a second line if len(message) > width - 1: message = message[:width-1] # \033[K (Erase to End of Line) clears any leftover chars from a # previously longer line drawn at the same position — avoids the # padding-width-mismatch flicker that occurred on terminal resize. sys.stdout.write(f'\r{message}\033[K\n') sys.stdout.flush() except Exception: # Absolute fallback try: print(message) except Exception: pass finally: CONSOLE_LOCK.release() def log(worker_id, message, level="INFO"): timestamp = datetime.now().strftime("%H:%M:%S") safe_print(f"[{timestamp}] [{worker_id}] [{level}] {message}") def signal_handler(sig, frame): global PAUSE_REQUESTED, _TUI_SIGNAL if platform.system() == 'Windows': SHUTDOWN_EVENT.set() try: kill_processes() except: pass if TUI_APP is not None: # Route through the TUI's quit action so Textual can restore the # Windows console mode (echo, VT processing, cursor visibility) # before the process exits. sys.exit() here skips that cleanup. _TUI_SIGNAL = 'quit' else: try: sys.stdout.write('\n[!] Shutdown initiated...\n') sys.stdout.flush() except Exception: pass return # Let app.run() / join loop exit naturally else: if TUI_APP is not None: # In TUI mode, Ctrl+C opens the pause menu so the user can choose # Continue / Finish / Stop. SIGTERM still shuts down cleanly. if sig == signal.SIGTERM: SHUTDOWN_EVENT.set() try: kill_processes() except: pass _TUI_SIGNAL = 'quit' else: # SIGINT (Ctrl+C) → request pause via flag; the app's polling # timer picks this up on the event loop thread (call_from_thread # cannot be used from a signal handler — same thread as the loop). _TUI_SIGNAL = 'pause' elif not PAUSE_REQUESTED: PAUSE_REQUESTED = True try: sys.stdout.write('\n\n[!] PAUSE REQUESTED (Stopping gracefully...)\n') sys.stdout.flush() except: pass def toggle_processes(suspend=True): if platform.system() == 'Windows': # Windows has no SIGSTOP/SIGCONT, but NtSuspendProcess / NtResumeProcess # (ntdll.dll) achieve the identical effect: all threads in the target # process are frozen atomically. ctypes is stdlib — no extra deps. try: ntdll = ctypes.windll.ntdll kernel32 = ctypes.windll.kernel32 PROCESS_ALL_ACCESS = 0x1F0FFF with PROC_LOCK: for wid, proc in ACTIVE_PROCS.items(): if proc.poll() is None: try: h = kernel32.OpenProcess(PROCESS_ALL_ACCESS, False, proc.pid) if h: if suspend: ntdll.NtSuspendProcess(h) else: ntdll.NtResumeProcess(h) kernel32.CloseHandle(h) except Exception as _e: safe_print(f"[!] Could not {'suspend' if suspend else 'resume'} PID {proc.pid}: {_e}") except Exception as _e: if suspend: safe_print(f"[!] WARNING: Windows process suspension unavailable: {_e}") return with PROC_LOCK: for wid, proc in ACTIVE_PROCS.items(): if proc.poll() is None: try: sig = signal.SIGSTOP if suspend else signal.SIGCONT # type: ignore[attr-defined] os.kill(proc.pid, sig) except: pass def kill_processes(): with PROC_LOCK: for wid, proc in ACTIVE_PROCS.items(): try: if proc.poll() is None: proc.kill() except: pass def check_version(manager_url): global LAST_UPDATE_CHECK with CHECK_LOCK: if time.time() - LAST_UPDATE_CHECK < 600: return False LAST_UPDATE_CHECK = time.time() try: url = f"{manager_url}/dl/worker" r = requests.get(url, headers=get_auth_headers(), timeout=10) if r.status_code == 200: match = re.search(r'WORKER_VERSION\s*=\s*"([^"]+)"', r.text) if match and match.group(1) != WORKER_VERSION: safe_print(f"[!] Update found: {WORKER_VERSION} -> {match.group(1)}") return True except: pass return False def apply_update(manager_url): safe_print("[*] Downloading and applying update...") tmp_path = None try: url = f"{manager_url}/dl/worker" r = requests.get(url, headers=get_auth_headers(), timeout=30) if r.status_code == 200: target_path = os.path.abspath(sys.argv[0]) tmp_path = target_path + '.tmp' with open(tmp_path, 'w', encoding='utf-8') as f: f.write(r.text) os.replace(tmp_path, target_path) safe_print("[*] Restarting worker...") os.execv(sys.executable, [sys.executable] + sys.argv) except Exception as e: safe_print(f"[!] Failed to apply update: {e}") if tmp_path: try: if os.path.exists(tmp_path): os.remove(tmp_path) except: pass def print_progress(worker_id, current, total, prefix='', suffix=''): if total <= 0: return percent = min(100, int(100 * current / float(total))) if TUI_APP is not None: # Update WORKER_DETAILS so the polling _tick() picks it up d = WORKER_DETAILS.get(worker_id) if d is not None: phase_map = {'DL': 'DL', 'Enc': 'Encoding', 'Up': 'Uploading'} d["phase"] = phase_map.get(prefix, prefix) d["pct"] = percent return if sys.is_finalizing(): return width = get_term_width() overhead = 12 + len(worker_id) + len(prefix) + 10 + len(suffix) bar_length = width - overhead - 5 if bar_length < 10: bar_length = 10 filled_length = int(bar_length * current // total) block_char = '█' fill_char = '-' try: bar = block_char * filled_length + fill_char * (bar_length - filled_length) line = f'[{datetime.now().strftime("%H:%M:%S")}] [{worker_id}] {prefix} |{bar}| {percent:.1f}% {suffix}' if len(line) > width - 1: line = line[:width - 1] with CONSOLE_LOCK: sys.stdout.write(f'\r{line}\033[K') sys.stdout.flush() except UnicodeEncodeError: block_char = '=' fill_char = '-' try: bar = block_char * filled_length + fill_char * (bar_length - filled_length) line = f'[{datetime.now().strftime("%H:%M:%S")}] [{worker_id}] {prefix} |{bar}| {percent:.1f}% {suffix}' if len(line) > width - 1: line = line[:width - 1] with CONSOLE_LOCK: sys.stdout.write(f'\r{line}\033[K') sys.stdout.flush() except: pass if current >= total: try: sys.stdout.write('\n') except: pass def monitor_status_loop(worker_ids): while not SHUTDOWN_EVENT.is_set(): # Daemon threads writing to stdout after interpreter shutdown started # will crash Python at the C level — bail out early. if sys.is_finalizing(): return if TUI_APP is not None: time.sleep(1); continue # TUI handles all status display if PAUSE_REQUESTED or MONITOR_PAUSED.is_set(): time.sleep(0.5); continue parts = [] with PROGRESS_LOCK: for wid in sorted(worker_ids, key=lambda x: x.split('-')[-1]): try: short_id = wid.split('-')[-1] except: short_id = wid state = WORKER_PROGRESS.get(wid, "Idle") parts.append(f"[{short_id}: {state}]") if parts: line = " ".join(parts) width = get_term_width() if len(line) > width - 4: line = line[:width-4] + "..." with CONSOLE_LOCK: try: sys.stdout.write(f'\r{line}\033[K') sys.stdout.flush() except: pass time.sleep(0.5) def get_seconds(t): try: parts = t.split(':') h = int(parts[0]); m = int(parts[1]); s = float(parts[2]) return h*3600 + m*60 + s except: return 0 # ============================================================================== # RESUME / PARTIAL ENCODE HELPERS # ============================================================================== _RESUME_CHK_PREFIX = "resume_chk_" def _probe_local_duration(file_path): """Return duration in seconds of a local media file, or 0.0 on failure.""" try: res = subprocess.run( [FFPROBE_CMD, '-v', 'quiet', '-print_format', 'json', '-show_format', file_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8', errors='replace', timeout=60) dur = json.loads(res.stdout).get('format', {}).get('duration') return float(dur) if dur else 0.0 except Exception: return 0.0 def _save_resume_checkpoint(path, payload): """Atomically write a resume checkpoint JSON.""" tmp = path + '.tmp' try: with open(tmp, 'w', encoding='utf-8') as f: json.dump(payload, f) os.replace(tmp, path) except Exception: try: if os.path.exists(tmp): os.remove(tmp) except Exception: pass def _load_resume_checkpoints(temp_dir): """Return list of (path, data) for all resume checkpoints found in temp_dir.""" results = [] try: for fname in os.listdir(temp_dir): if fname.startswith(_RESUME_CHK_PREFIX) and fname.endswith('.json'): fpath = os.path.join(temp_dir, fname) try: with open(fpath, 'r', encoding='utf-8') as f: data = json.load(f) results.append((fpath, data)) except Exception: pass except Exception: pass return results # Number of bytes hashed for source-integrity check (must match server constant) _HASH_BYTES = 4 * 1024 * 1024 def _compute_source_hash(dl_url, is_local): """Return MD5 hex digest of the first _HASH_BYTES of the source file. For local files the path is read directly; for remote URLs an HTTP Range request is used so we never need to download the whole file first. Returns None on any error so callers can skip the check gracefully.""" h = hashlib.md5() try: if is_local: with open(dl_url, 'rb') as f: remaining = _HASH_BYTES while remaining > 0: chunk = f.read(min(65536, remaining)) if not chunk: break h.update(chunk) remaining -= len(chunk) else: headers = dict(get_auth_headers()) headers['Range'] = f'bytes=0-{_HASH_BYTES - 1}' r = requests.get(dl_url, headers=headers, stream=True, timeout=30) # Accept both 206 Partial Content and 200 OK (some servers ignore Range) if r.status_code not in (200, 206): return None read = 0 for chunk in r.iter_content(65536): if read >= _HASH_BYTES: break if read + len(chunk) > _HASH_BYTES: chunk = chunk[:_HASH_BYTES - read] h.update(chunk) read += len(chunk) except Exception: return None return h.hexdigest() # ============================================================================== # FFMPEG MANAGEMENT # ============================================================================== def _ffmpeg_primary_url(): """Return the primary upstream URL for the current platform/arch.""" if platform.system() == "Windows": return "https://github.com/BtbN/FFmpeg-Builds/releases/download/latest/ffmpeg-master-latest-win64-gpl.zip" arch = platform.machine().lower() if arch in ['x86_64', 'amd64']: return "https://github.com/BtbN/FFmpeg-Builds/releases/download/latest/ffmpeg-master-latest-linux64-gpl.tar.xz" elif arch in ['aarch64', 'arm64']: return "https://johnvansickle.com/ffmpeg/releases/ffmpeg-release-arm64-static.tar.xz" return None def _save_ffmpeg_meta(url): """Store the ETag/Last-Modified of url so we can detect upstream updates.""" try: r = requests.head(url, allow_redirects=True, timeout=15) meta = { "url": url, "etag": r.headers.get('ETag', ''), "last_modified": r.headers.get('Last-Modified', ''), } with open(_FFMPEG_META_FILE, 'w') as f: json.dump(meta, f) except Exception: pass def update_ffmpeg_if_stale(): """Re-download the local ffmpeg binary if the upstream build is newer. Uses a HEAD request to compare the ETag/Last-Modified stored after the last download. A full download (~100-200 MB) only happens when the upstream file has actually changed. """ local_bin = os.path.abspath("ffmpeg.exe" if platform.system() == "Windows" else "./ffmpeg") if not os.path.exists(local_bin): return # nothing local to update; check_ffmpeg() handles first install url = _ffmpeg_primary_url() if url is None: return # Load stored meta stored_etag = stored_lm = "" if os.path.exists(_FFMPEG_META_FILE): try: with open(_FFMPEG_META_FILE) as f: m = json.load(f) if m.get('url') == url: stored_etag = m.get('etag', '') stored_lm = m.get('last_modified', '') except Exception: pass # HEAD request to check remote freshness try: r = requests.head(url, allow_redirects=True, timeout=15) remote_etag = r.headers.get('ETag', '') remote_lm = r.headers.get('Last-Modified', '') except Exception: return # network unavailable — skip silently # If we have nothing stored yet, just save and return (no spurious download) if not stored_etag and not stored_lm: _save_ffmpeg_meta(url) return changed = False if remote_etag and remote_etag != stored_etag: changed = True if remote_lm and remote_lm != stored_lm: changed = True if not changed: return print("[*] Newer FFmpeg build detected upstream — updating...") if platform.system() == "Windows": ok = download_ffmpeg_windows() else: ok = download_ffmpeg_linux() if ok: _save_ffmpeg_meta(url) # Re-point FFMPEG_CMD to the freshly extracted binary global FFMPEG_CMD, FFPROBE_CMD local_ffprobe = os.path.abspath("ffprobe.exe" if platform.system() == "Windows" else "./ffprobe") FFMPEG_CMD = local_bin if os.path.exists(local_ffprobe): FFPROBE_CMD = local_ffprobe print("[*] FFmpeg updated successfully.") def download_ffmpeg_windows(): print("[*] FFmpeg not found. Attempting download (FULL Version ~128MB)...") urls = [ "https://github.com/BtbN/FFmpeg-Builds/releases/download/latest/ffmpeg-master-latest-win64-gpl.zip", "https://vsv.fractumseraph.net/ffmpeg-master-latest-win64-gpl.zip" ] temp_zip = "ffmpeg_temp.zip" for url in urls: print(f"[*] Trying mirror: {url}") try: with requests.get(url, stream=True, timeout=180) as r: r.raise_for_status() total_size = int(r.headers.get('content-length', 0)) downloaded = 0 with open(temp_zip, 'wb') as f: for chunk in r.iter_content(chunk_size=8192): f.write(chunk) downloaded += len(chunk) if total_size > 0: pct = int((downloaded / total_size) * 100) try: msg = f" Downloading... {pct}%" sys.stdout.write(f"\r{msg}") sys.stdout.flush() except: pass print("\n[*] Extracting FFmpeg...") with zipfile.ZipFile(temp_zip) as z: ffmpeg_path = None ffprobe_path = None for file in z.namelist(): if file.endswith("bin/ffmpeg.exe"): ffmpeg_path = file if file.endswith("bin/ffprobe.exe"): ffprobe_path = file if not ffmpeg_path or not ffprobe_path: print("\n[!] Binaries not found in zip.") continue with open("ffmpeg.exe", "wb") as f: f.write(z.read(ffmpeg_path)) with open("ffprobe.exe", "wb") as f: f.write(z.read(ffprobe_path)) os.remove(temp_zip) print("[*] FFmpeg installed locally!") _save_ffmpeg_meta(_ffmpeg_primary_url() or url) return True except Exception as e: print(f"\n[!] Mirror failed: {e}") if os.path.exists(temp_zip): os.remove(temp_zip) continue return False def download_ffmpeg_linux(): print("[*] Downloading static FFmpeg build (BtbN)...") arch = platform.machine().lower() if arch in ['x86_64', 'amd64']: url = "https://github.com/BtbN/FFmpeg-Builds/releases/download/latest/ffmpeg-master-latest-linux64-gpl.tar.xz" elif arch in ['aarch64', 'arm64']: url = "https://johnvansickle.com/ffmpeg/releases/ffmpeg-release-arm64-static.tar.xz" else: print(f"[!] Unsupported architecture for auto-download: {arch}") return False try: r = requests.get(url, stream=True, allow_redirects=True, timeout=180) r.raise_for_status() tar_name = f"ffmpeg_static_{int(time.time())}.tar.xz" total_size = int(r.headers.get('content-length', 0)) downloaded = 0 with open(tar_name, 'wb') as f: for chunk in r.iter_content(chunk_size=8192): f.write(chunk) downloaded += len(chunk) if total_size > 0: pct = int((downloaded / total_size) * 100) try: sys.stdout.write(f"\r Downloading... {pct}%") sys.stdout.flush() except: pass print("\n[*] Extracting FFmpeg...") ext_dir = f"temp_ffmpeg_ext_{int(time.time())}" os.makedirs(ext_dir, exist_ok=True) with tarfile.open(tar_name, "r:xz") as tar: tar.extractall(path=ext_dir) found_ffmpeg = False for root, dirs, files in os.walk(ext_dir): for file in files: if file == "ffmpeg": shutil.move(os.path.join(root, file), "ffmpeg") found_ffmpeg = True elif file == "ffprobe": shutil.move(os.path.join(root, file), "ffprobe") if os.path.exists(tar_name): os.remove(tar_name) if os.path.exists(ext_dir): shutil.rmtree(ext_dir) if found_ffmpeg: os.chmod("ffmpeg", 0o755) if os.path.exists("ffprobe"): os.chmod("ffprobe", 0o755) print("[*] FFmpeg installed locally!") _save_ffmpeg_meta(url) return True else: print("[!] Could not find 'ffmpeg' binary in extracted archive.") return False except Exception as e: print(f"\n[!] Linux Download failed: {e}") return False def has_svtav1(cmd): """Checks if the given ffmpeg command supports libsvtav1""" try: res = subprocess.run([cmd, "-hide_banner", "-encoders"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, encoding='utf-8', errors='replace') return "libsvtav1" in res.stdout except: return False def check_ffmpeg(): global FFMPEG_CMD, FFPROBE_CMD local_ffmpeg = os.path.abspath("ffmpeg.exe" if platform.system() == "Windows" else "./ffmpeg") local_ffprobe = os.path.abspath("ffprobe.exe" if platform.system() == "Windows" else "./ffprobe") if os.path.exists(local_ffmpeg) and has_svtav1(local_ffmpeg): FFMPEG_CMD = local_ffmpeg if os.path.exists(local_ffprobe): FFPROBE_CMD = local_ffprobe return if shutil.which("ffmpeg") and has_svtav1("ffmpeg"): FFMPEG_CMD = "ffmpeg" FFPROBE_CMD = "ffprobe" return print("[!] Valid FFmpeg with libsvtav1 not found.") download_success = False if platform.system() == "Windows": download_success = download_ffmpeg_windows() else: download_success = download_ffmpeg_linux() if download_success: if os.path.exists(local_ffmpeg) and has_svtav1(local_ffmpeg): FFMPEG_CMD = local_ffmpeg if os.path.exists(local_ffprobe): FFPROBE_CMD = local_ffprobe return print("\n[CRITICAL ERROR] Could not find or download a version of FFmpeg with 'libsvtav1' support.") print("Please install FFmpeg with SVT-AV1 support manually.") sys.exit(1) def verify_connection(manager_url): try: if requests.get(manager_url, timeout=10).status_code < 400: return True except: pass print(f"[!] Could not connect to {manager_url}"); return False def worker_task(worker_id, manager_url, temp_dir, quota_tracker, single_mode=False, series_id=None, watermark=False, max_size_mb=0, local_source_dir=None): global UPDATE_AVAILABLE log(worker_id, "Thread active.") os.makedirs(temp_dir, exist_ok=True) def update_status(msg): with PROGRESS_LOCK: WORKER_PROGRESS[worker_id] = msg d = WORKER_DETAILS.get(worker_id) if d is not None: if msg.startswith("DL "): d["phase"] = "DL" try: d["pct"] = int(msg[3:].rstrip('%')) except: pass elif msg.startswith("Enc "): d["phase"] = "Encoding" try: d["pct"] = int(msg[4:].rstrip('%')) except: pass elif msg.startswith("Up "): d["phase"] = "Uploading" try: d["pct"] = int(msg[3:].rstrip('%')) except: pass elif msg == "Idle": d.update({"phase": "Idle", "pct": 0, "file": "-", "job_start": 0.0}) elif msg == "Probing": d["phase"] = "Probe" elif msg == "Quota Limit": d.update({"phase": "Quota", "pct": 0, "file": "-", "job_start": 0.0}) else: d["phase"] = msg def post_status(status, progress=0, duration=0, error_msg=None): try: payload = { "worker_id": worker_id, "job_id": job_id, "status": status, "progress": progress, "version": WORKER_VERSION } if duration > 0: payload["duration"] = duration if error_msg: payload["error"] = error_msg requests.post(f"{manager_url}/report_status", json=payload, headers=get_auth_headers(), timeout=10) except: pass def report_error(rpt_job_id, error_type, message, details=""): """Send a structured error report to the manager for admin visibility.""" try: requests.post( f"{manager_url}/report_error", json={ "job_id": rpt_job_id, "worker_id": worker_id, "error_type": error_type, "message": str(message)[:2048], "details": str(details)[:32768], }, headers=get_auth_headers(), timeout=10 ) except: pass while not SHUTDOWN_EVENT.is_set(): if PAUSE_REQUESTED: time.sleep(1); continue try: if quota_tracker and quota_tracker.check_cap(): wait_sec = quota_tracker.get_wait_time() update_status("Quota Limit") log(worker_id, f"Daily Quota Reached. Reset in {wait_sec/3600:.1f} hours.") while wait_sec > 0 and not SHUTDOWN_EVENT.is_set(): time.sleep(min(60, wait_sec)) wait_sec -= 60 if not quota_tracker.check_cap(): break continue update_status("Idle") # ========================================================= # RESUME: Finish any partial encodes from a previous run # ========================================================= for _r_chk_path, _r_chk in _load_resume_checkpoints(temp_dir): if SHUTDOWN_EVENT.is_set(): break _r_partial = _r_chk.get('partial_path', '') _r_job_id = _r_chk.get('job_id', '') if not _r_job_id or not os.path.exists(_r_partial): log(worker_id, f"Stale resume checkpoint ({_r_job_id}). Removing.") try: os.remove(_r_chk_path) except: pass continue # ---------------------------------------------------------- # Check with the server: is this job still ours to resume? # ---------------------------------------------------------- _r_server_status = None _r_server_owner = None _r_check_failed = False try: _r_chk_resp = requests.get( f"{manager_url}/job_status", params={"job_id": _r_job_id}, headers=get_auth_headers(), timeout=10) if _r_chk_resp.status_code == 200: _r_chk_data = _r_chk_resp.json() _r_server_status = _r_chk_data.get("job_status") _r_server_owner = _r_chk_data.get("worker_id") elif _r_chk_resp.status_code == 404: # Job no longer exists on the server; discard checkpoint log(worker_id, f"Checkpoint job {_r_job_id} not found on server. Discarding.") for _fp in [_r_chk_path, _r_partial]: try: if os.path.exists(_fp): os.remove(_fp) except: pass continue else: _r_check_failed = True except Exception as _r_ce: log(worker_id, f"Could not reach server to validate checkpoint ({_r_job_id}): {_r_ce}. Skipping resume this cycle.", "WARN") _r_check_failed = True if _r_check_failed: # Can't confirm ownership — skip this iteration, try again next cycle continue if _r_server_status == "completed": log(worker_id, f"Job {_r_job_id} already completed on server. Discarding checkpoint.") for _fp in [_r_chk_path, _r_partial]: try: if os.path.exists(_fp): os.remove(_fp) except: pass continue if _r_server_status == "permanently_failed": log(worker_id, f"Job {_r_job_id} is permanently failed. Discarding checkpoint.") for _fp in [_r_chk_path, _r_partial]: try: if os.path.exists(_fp): os.remove(_fp) except: pass continue if _r_server_status == "processing" and _r_server_owner and _r_server_owner != worker_id: # A different worker has been assigned this job — leave it alone log(worker_id, f"Job {_r_job_id} is now owned by {_r_server_owner}. Abandoning checkpoint.") for _fp in [_r_chk_path, _r_partial]: try: if os.path.exists(_fp): os.remove(_fp) except: pass continue # Job is queued (timed out) or still ours — reclaim it before resuming if _r_server_status in ("queued", "failed"): try: _r_reclaim = requests.post( f"{manager_url}/reclaim_job", json={"job_id": _r_job_id, "worker_id": worker_id, "version": WORKER_VERSION}, headers=get_auth_headers(), timeout=10) if _r_reclaim.status_code == 409: # Race: another worker claimed it first log(worker_id, f"Job {_r_job_id} reclaim failed (already taken). Abandoning checkpoint.") for _fp in [_r_chk_path, _r_partial]: try: if os.path.exists(_fp): os.remove(_fp) except: pass continue if _r_reclaim.status_code != 200: log(worker_id, f"Job {_r_job_id} reclaim returned {_r_reclaim.status_code}. Skipping.", "WARN") continue log(worker_id, f"Reclaimed job {_r_job_id} from the queue.") except Exception as _r_re: log(worker_id, f"Reclaim request failed for {_r_job_id}: {_r_re}. Skipping.", "WARN") continue # ---------------------------------------------------------- _r_partial_dur = _probe_local_duration(_r_partial) if _r_partial_dur < 5.0: log(worker_id, f"Partial encode for {_r_job_id} unreadable/too short ({_r_partial_dur:.1f}s). Discarding.") for _fp in [_r_chk_path, _r_partial]: try: if os.path.exists(_fp): os.remove(_fp) except: pass continue _r_total_sec = _r_chk.get('total_sec', 0) _r_total_min = _r_chk.get('total_min', 0) _r_pct = int(_r_partial_dur / _r_total_sec * 100) if _r_total_sec > 0 else 0 log(worker_id, f"Resuming job {_r_job_id}: was {_r_pct}% done ({_r_partial_dur:.1f}s/{_r_total_sec:.1f}s).") job_id = _r_job_id d = WORKER_DETAILS.get(worker_id) if d is not None: d.update({"file": _r_job_id, "phase": "Encoding", "pct": _r_pct, "job_start": time.time()}) post_status("processing", _r_pct, _r_total_min) _r_p1 = _r_partial + '.p1.mp4' _r_p2 = _r_partial + '.p2.mp4' _r_clist = _r_partial + '.concat.txt' _r_merged = _r_partial + '.merged.mp4' _r_tmps = [_r_p1, _r_p2, _r_clist, _r_merged] _resume_ok = False try: # Trim part1 to a clean keyframe boundary using stream-copy (no re-encode) _r_trim_pt = max(0.0, _r_partial_dur - 5.0) _r_trim_rc = subprocess.run( [FFMPEG_CMD, '-y', '-i', _r_partial, '-t', str(_r_trim_pt), '-c', 'copy', _r_p1], stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8', errors='replace', timeout=300) if _r_trim_rc.returncode != 0 or not os.path.exists(_r_p1): raise RuntimeError(f"Part-1 trim failed (rc={_r_trim_rc.returncode})") _r_p1_dur = _probe_local_duration(_r_p1) if _r_p1_dur <= 0: raise RuntimeError("Trimmed part-1 not readable") # Encode the remainder starting from _r_p1_dur _r_hdrs = f"X-Worker-Token: {WORKER_SECRET}\r\n" _r_dl = _r_chk.get('dl_url', '') _r_ai = _r_chk.get('audio_index', 0) _r_si = _r_chk.get('subtitle_indices', []) _r_vf = _r_chk.get('video_filter', ENCODING_CONFIG['VIDEO_SCALE']) _r_af = _r_chk.get('audio_filter', 'aresample=async=1') _r_crf = _r_chk.get('target_crf', int(ENCODING_CONFIG['VIDEO_CRF'])) # Determine if the source was a local file (--local-source) or HTTP. # Local paths don't start with http(s)://, so skip reconnect flags. _r_is_local_src = not (_r_dl.startswith('http://') or _r_dl.startswith('https://')) log(worker_id, f"Encoding remainder from {_r_p1_dur:.1f}s...") _r_input_args = ( ['-ss', str(_r_p1_dur), '-y', '-i', _r_dl] if _r_is_local_src else ['-headers', _r_hdrs, '-reconnect', '1', '-reconnect_streamed', '1', '-reconnect_delay_max', '60', '-reconnect_on_network_error', '1', '-ss', str(_r_p1_dur), '-y', '-i', _r_dl] ) _r_rem_cmd = ( [FFMPEG_CMD] + _r_input_args + ['-map', '0:v:0', '-map', f'0:{_r_ai}'] + [x for idx in _r_si for x in ['-map', f'0:{idx}']] + ['-fps_mode', 'passthrough', '-avoid_negative_ts', 'make_zero', '-c:v', ENCODING_CONFIG["VIDEO_CODEC"], '-preset', ENCODING_CONFIG["VIDEO_PRESET"], '-crf', str(_r_crf), '-pix_fmt', ENCODING_CONFIG["VIDEO_PIX_FMT"], '-vf', _r_vf, '-c:a', ENCODING_CONFIG["AUDIO_CODEC"], '-b:a', ENCODING_CONFIG["AUDIO_BITRATE"], '-ac', ENCODING_CONFIG["AUDIO_CHANNELS"], '-af', _r_af, '-c:s', ENCODING_CONFIG["SUBTITLE_CODEC"], '-movflags', 'frag_keyframe+empty_moov+default_base_moof', '-progress', 'pipe:1', _r_p2]) _r_popen_kw = {} if platform.system() == 'Windows': _r_popen_kw['creationflags'] = subprocess.CREATE_NEW_PROCESS_GROUP else: _r_popen_kw['start_new_session'] = True _r_rem = subprocess.Popen( _r_rem_cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, encoding='utf-8', errors='replace', **_r_popen_kw) assert _r_rem.stdout is not None with PROC_LOCK: ACTIVE_PROCS[worker_id] = _r_rem _r_last_rep = 0 while True: if SHUTDOWN_EVENT.is_set(): _r_rem.terminate(); break _r_ln = _r_rem.stdout.readline() if not _r_ln and _r_rem.poll() is not None: break if "out_time=" in _r_ln and "N/A" not in _r_ln and _r_total_sec > 0: try: _r_cur = get_seconds(_r_ln.split('=')[1].strip()) _r_ov = min(100, int((_r_p1_dur + _r_cur) / _r_total_sec * 100)) update_status(f"Enc {_r_ov}%") if time.time() - _r_last_rep > 10: post_status("processing", _r_ov, _r_total_min) _r_last_rep = time.time() except: pass with PROC_LOCK: if worker_id in ACTIVE_PROCS: del ACTIVE_PROCS[worker_id] if SHUTDOWN_EVENT.is_set(): break # keep partial + checkpoint for the next run if _r_rem.returncode != 0 or not os.path.exists(_r_p2): raise RuntimeError(f"Remainder encode failed (rc={_r_rem.returncode})") # Concatenate part1 + part2 into the final merged file log(worker_id, "Concatenating resume parts...") with open(_r_clist, 'w', encoding='utf-8') as _cf: _cf.write(f"file '{_r_p1.replace(os.sep, '/')}'\n") _cf.write(f"file '{_r_p2.replace(os.sep, '/')}'\n") _r_cat_rc = subprocess.run( [FFMPEG_CMD, '-y', '-f', 'concat', '-safe', '0', '-i', _r_clist, '-c', 'copy', '-movflags', '+faststart', _r_merged], stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8', errors='replace', timeout=600) if _r_cat_rc.returncode != 0 or not os.path.exists(_r_merged): raise RuntimeError(f"Concat failed (rc={_r_cat_rc.returncode})") os.replace(_r_merged, _r_partial) _r_tmps.remove(_r_merged) # already moved, skip final-cleanup _resume_ok = True except Exception as _r_err: log(worker_id, f"Resume failed for {_r_job_id}: {_r_err}. Discarding.", "ERROR") finally: for _fp in _r_tmps: try: if os.path.exists(_fp): os.remove(_fp) except: pass if not _resume_ok: for _fp in [_r_chk_path, _r_partial]: try: if os.path.exists(_fp): os.remove(_fp) except: pass continue # Upload the resumed, fully merged file _r_file_bytes = os.path.getsize(_r_partial) log(worker_id, f"Resume complete ({_r_file_bytes/1024/1024:.2f} MB). Uploading...") post_status("uploading", 0) update_status("Up 0%") _r_up_ok = False for _r_up_att in range(3): if SHUTDOWN_EVENT.is_set(): break try: with open(_r_partial, 'rb') as _r_f: _r_resp = requests.post( f"{manager_url}/upload_result", files={'file': (_r_job_id, _r_f)}, data={'job_id': _r_job_id, 'worker_id': worker_id, 'duration': _r_total_min}, headers=get_auth_headers(), timeout=300) if _r_resp.status_code == 200: _r_up_ok = True; break else: log(worker_id, f"Resume upload rejected: {_r_resp.status_code}", "WARN") except Exception as _r_ue: log(worker_id, f"Resume upload attempt {_r_up_att+1} failed: {_r_ue}", "WARN") time.sleep(10) if _r_up_ok: log(worker_id, f"Resumed job {_r_job_id} complete.") d = WORKER_DETAILS.get(worker_id) if d is not None: d.update({"phase": "Done", "pct": 100}) d["jobs_done"] += 1 with STATS_LOCK: SESSION_STATS["jobs_done"] += 1 SESSION_STATS["bytes_uploaded"] += _r_file_bytes else: log(worker_id, f"Resume upload failed for {_r_job_id}.", "ERROR") post_status("failed", error_msg="Resume upload failed after 3 attempts") # Always clean up the checkpoint and partial file after an upload attempt for _fp in [_r_chk_path, _r_partial]: try: if os.path.exists(_fp): os.remove(_fp) except: pass # ======================================================== if check_version(manager_url): UPDATE_AVAILABLE = True; SHUTDOWN_EVENT.set(); break try: params = {'worker_id': worker_id, 'version': WORKER_VERSION} if series_id: params['series_id'] = series_id if max_size_mb and max_size_mb > 0: params['max_size_mb'] = max_size_mb r = requests.get(f"{manager_url}/get_job", params=params, headers=get_auth_headers(), timeout=10) except: time.sleep(5); continue data = r.json() if r.status_code == 200 else None if r.status_code == 401: log(worker_id, "AUTH FAILED: Worker Secret is invalid or missing.", "CRITICAL") SHUTDOWN_EVENT.set(); break if data and data.get("status") == "ok": job = data["job"]; job_id = job['id']; dl_url = job['download_url'] # If this worker has direct access to the source directory, prefer the # local file path — avoids streaming the file through the HTTP server. _local_src = None if local_source_dir: _candidate = os.path.join(local_source_dir, job['id'].replace('/', os.sep)) if os.path.isfile(_candidate): _local_src = os.path.abspath(_candidate) dl_url = _local_src log(worker_id, f"Job (local): {job['filename']}") else: log(worker_id, f"Job (local path not found, falling back to HTTP): {job['filename']}", "WARN") if _local_src is None: log(worker_id, f"Job: {job['filename']}") d = WORKER_DETAILS.get(worker_id) if d is not None: d.update({"file": job['filename'], "phase": "Probe", "pct": 0, "job_start": time.time()}) local_dst = os.path.join(temp_dir, f"encoded{ENCODING_CONFIG['OUTPUT_EXT']}") # FFmpeg auth header string (format required by libavformat HTTP demuxer) ffmpeg_http_headers = f"X-Worker-Token: {WORKER_SECRET}\r\n" # When the source is a local file the auth header and reconnect flags # are unnecessary; flag this so the probe/encode commands omit them. _is_local = (_local_src is not None) # Quota accounting: a single HEAD request gives us the source size instantly, # letting encoding start immediately rather than waiting for a full download. if quota_tracker: try: if _is_local: src_size = os.path.getsize(dl_url) else: head_r = requests.head(dl_url, headers=get_auth_headers(), timeout=10) src_size = int(head_r.headers.get('content-length', 0)) if quota_tracker.check_cap(): wait_sec = quota_tracker.get_wait_time() update_status("Quota Limit") log(worker_id, f"Daily Quota Reached. Reset in {wait_sec/3600:.1f} hours.") while wait_sec > 0 and not SHUTDOWN_EVENT.is_set(): time.sleep(min(60, wait_sec)) wait_sec -= 60 if not quota_tracker.check_cap(): break continue if src_size > 0: quota_tracker.add_usage(src_size) quota_tracker.force_save() except Exception as e: log(worker_id, f"Quota size check failed: {e}", "WARN") post_status("downloading", 0) # ------------------------------------------------------- # SOURCE INTEGRITY CHECK # Only meaningful for local-source workers: the worker reads # the first 4 MB of the local file and submits its hash so the # server can detect file corruption/replacement on disk. # For remote HTTP sources the server hosts the file itself, so # there is nothing to verify from the worker side — and naively # streaming 4 MB (or the full file when Range isn't supported) # before every encode wastes bandwidth and causes the worker to # appear stuck. # ------------------------------------------------------- if _is_local: log(worker_id, "Verifying source file integrity...") update_status("Verifying") _our_hash = _compute_source_hash(dl_url, _is_local) if _our_hash is None: log(worker_id, "Could not compute source hash. Skipping integrity check.", "WARN") else: try: _vr = requests.post( f"{manager_url}/verify_source_hash", json={"job_id": job_id, "worker_id": worker_id, "source_hash": _our_hash}, headers=get_auth_headers(), timeout=10) _vstatus = _vr.json().get("status") if _vr.status_code == 200 else "error" except Exception as _ve: log(worker_id, f"Hash verify request failed: {_ve}. Skipping check.", "WARN") _vstatus = "error" if _vstatus == "mismatch": log(worker_id, f"SOURCE HASH MISMATCH for {job_id}! " "The file on this worker does not match the server's source. " "Aborting encode.", "ERROR") post_status("failed", error_msg="Source hash mismatch") report_error(job_id, "hash_mismatch", f"Worker hash: {_our_hash}") continue elif _vstatus == "ok": log(worker_id, "Source integrity verified.") # "pending" or "error" → server has no hash yet, proceed normally # ------------------------------------------------------- update_status("Probing") total_sec = 0; total_min = 0; audio_index = 0; subtitle_indices = [] for _probe_attempt in range(3): try: if _is_local: cmd_probe = [FFPROBE_CMD, '-v', 'quiet', '-print_format', 'json', '-show_streams', '-show_format', dl_url] else: cmd_probe = [FFPROBE_CMD, '-headers', ffmpeg_http_headers, '-v', 'quiet', '-print_format', 'json', '-show_streams', '-show_format', dl_url] res = subprocess.run(cmd_probe, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8', errors='replace', timeout=60) probe_data = json.loads(res.stdout) dur = probe_data.get('format', {}).get('duration') if dur: total_sec = float(dur); total_min = int(total_sec / 60) audio_streams = [s for s in probe_data.get('streams', []) if s['codec_type'] == 'audio'] if audio_streams: audio_index = audio_streams[0]['index'] for s in audio_streams: if s.get('tags', {}).get('language', '').lower() in ['eng', 'en', 'english']: audio_index = s['index']; break for s in probe_data.get('streams', []): if s['codec_type'] == 'subtitle': if s.get('codec_name', '').lower() in ['subrip', 'ass', 'webvtt', 'mov_text', 'text', 'srt', 'ssa']: subtitle_indices.append(s['index']) break # probe succeeded except Exception as _probe_err: if _probe_attempt < 2: log(worker_id, f"Probe failed (attempt {_probe_attempt+1}/3): {_probe_err}. Retrying in {5*(_probe_attempt+1)}s...", "WARN") time.sleep(5 * (_probe_attempt + 1)) else: log(worker_id, f"Probe failed after 3 attempts: {_probe_err}. Skipping job.", "ERROR") if 'probe_data' not in locals(): post_status("failed", error_msg="FFprobe failed after 3 attempts") report_error(job_id, "probe_failure", "FFprobe failed after 3 attempts") continue log(worker_id, f"Encoding ({total_min}m)...") post_status("processing", 0, total_min) # Disk space pre-flight: need at least 500 MB free for the encoded output try: _disk_free = shutil.disk_usage(temp_dir).free if _disk_free < 500 * 1024 * 1024: log(worker_id, f"Low disk space: {_disk_free/1024**2:.0f} MB free, need 500 MB. Skipping.", "ERROR") post_status("failed", error_msg="Insufficient disk space") report_error(job_id, "disk_space", f"Only {_disk_free/1024**2:.0f} MB free, need 500 MB") continue except Exception as _disk_e: log(worker_id, f"Disk space check error: {_disk_e}", "WARN") # Font is only needed for the watermark local_font = os.path.join(temp_dir, "arial.ttf") if watermark: src_font = os.path.join(_script_dir, "arial.ttf") if not os.path.exists(src_font): log(worker_id, "arial.ttf missing locally. Downloading from manager...") try: font_req = requests.get(f"{manager_url}/dl/font", headers=get_auth_headers(), timeout=30) if font_req.status_code == 200: with open(src_font, 'wb') as f: f.write(font_req.content) log(worker_id, "Font downloaded successfully.") except Exception as e: log(worker_id, f"Font download failed: {e}", "WARN") try: if os.path.exists(src_font): shutil.copy(src_font, local_font) except: pass # Construct video filter conditionally based on watermark flag and font availability if watermark and os.path.exists(local_font): font_arg = local_font.replace("\\", "/") video_filter = f"{ENCODING_CONFIG['VIDEO_SCALE']},drawtext=text='@FractumSeraph':fontfile='{font_arg}':fontcolor=white@0.2:fontsize=12:x=10:y=h-th-10" else: video_filter = ENCODING_CONFIG['VIDEO_SCALE'] if watermark: log(worker_id, "Warning: arial.ttf could not be sourced. Skipping watermark.", "WARN") # Robust Audio Downmixing (Prevents crashes on corrupt streams claiming 40+ channels) audio_channels = 2 # Default assumption try: for s in probe_data.get('streams', []): if s['index'] == audio_index: audio_channels = int(s.get('channels', 2)) break except: pass # AUDIO FILTER FIX (MONO + DIALOGUE FOCUS) audio_filter = "aresample=async=1" if audio_channels > 2: # 5.1 -> Mono: Mix Center (FC) strongly (50%) to ensure dialogue is clear # c0 = 0.5*FC + 0.25*FL + 0.25*FR + 0.1*Surround audio_filter = "pan=mono|c0=0.5*FC+0.25*FL+0.25*FR+0.1*BL+0.1*BR,aresample=async=1" elif audio_channels == 2: # Stereo -> Mono: Standard Mix audio_filter = "pan=mono|c0=0.5*c0+0.5*c1,aresample=async=1" else: # Mono -> Mono: Passthrough audio_filter = "aresample=async=1" # [ADDED] Dynamic CRF Adjustment for Live Action Profile base_crf = int(ENCODING_CONFIG["VIDEO_CRF"]) profile = job.get('content_profile', 'standard') if profile == 'live_action': target_crf = base_crf - 6 log(worker_id, f"Live Action profile detected! Allocating 2x bitrate (CRF: {target_crf})") else: target_crf = base_crf if _is_local: cmd = [FFMPEG_CMD, '-y', '-i', dl_url, '-map', '0:v:0', '-map', f'0:{audio_index}'] else: cmd = [FFMPEG_CMD, '-headers', ffmpeg_http_headers, '-reconnect', '1', '-reconnect_streamed', '1', '-reconnect_delay_max', '60', '-reconnect_on_network_error', '1', '-y', '-i', dl_url, '-map', '0:v:0', '-map', f'0:{audio_index}'] for idx in subtitle_indices: cmd.extend(['-map', f'0:{idx}']) cmd.extend([ '-fps_mode', 'passthrough', '-avoid_negative_ts', 'make_zero', '-c:v', ENCODING_CONFIG["VIDEO_CODEC"], '-preset', ENCODING_CONFIG["VIDEO_PRESET"], '-crf', str(target_crf), # [CHANGED] Uses dynamically calculated CRF '-pix_fmt', ENCODING_CONFIG["VIDEO_PIX_FMT"], '-vf', video_filter, '-c:a', ENCODING_CONFIG["AUDIO_CODEC"], '-b:a', ENCODING_CONFIG["AUDIO_BITRATE"], '-ac', ENCODING_CONFIG["AUDIO_CHANNELS"], # Enforce 1 channel (Mono) '-af', audio_filter, '-c:s', ENCODING_CONFIG["SUBTITLE_CODEC"], # frag_keyframe writes each GOP as a self-contained fragment so the file # remains readable even if the process is killed mid-encode. '-movflags', 'frag_keyframe+empty_moov+default_base_moof', '-progress', 'pipe:1', local_dst ]) # Save a resume checkpoint so we can recover this encode if interrupted. _chk_safe_id = re.sub(r'[^a-zA-Z0-9_-]', '_', str(job_id)) _chk_path = os.path.join(temp_dir, f"{_RESUME_CHK_PREFIX}{_chk_safe_id}.json") _save_resume_checkpoint(_chk_path, { "job_id": job_id, "dl_url": dl_url, "partial_path": local_dst, "total_sec": total_sec, "total_min": total_min, "audio_index": audio_index, "subtitle_indices": subtitle_indices, "target_crf": target_crf, "video_filter": video_filter, "audio_filter": audio_filter, "created_at": time.time(), }) proc = None; enc_time = 0 for _enc_attempt in range(3): if SHUTDOWN_EVENT.is_set(): break if _enc_attempt > 0: _retry_delay = 30 log(worker_id, f"Encode failed (attempt {_enc_attempt}/3, rc={proc.returncode if proc else '?'}). Retrying in {_retry_delay}s...", "WARN") update_status("Retrying") for _ in range(_retry_delay): if SHUTDOWN_EVENT.is_set(): break time.sleep(1) if SHUTDOWN_EVENT.is_set(): break log(worker_id, f"Re-encoding ({total_min}m)...") post_status("processing", 0, total_min) start_enc = time.time(); last_rep = 0; last_enc_pct = 0; last_hb = 0 log_buffer = [] popen_kwargs = {} if platform.system() == 'Windows': popen_kwargs['creationflags'] = subprocess.CREATE_NEW_PROCESS_GROUP else: popen_kwargs['start_new_session'] = True proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, encoding='utf-8', errors='replace', **popen_kwargs) assert proc.stdout is not None with PROC_LOCK: ACTIVE_PROCS[worker_id] = proc raw_log_path = os.path.join(temp_dir, "encode.log") with open(raw_log_path, 'w', encoding='utf-8') as raw_log: while True: if PAUSE_REQUESTED: _now = time.time() if _now - last_hb > 30: post_status("paused", last_enc_pct) last_hb = _now time.sleep(0.2) continue line = proc.stdout.readline() if line: log_buffer.append(line); log_buffer = log_buffer[-50:] raw_log.write(line) raw_log.flush() if not line and proc.poll() is not None: break if "out_time=" in line and "N/A" not in line and total_sec > 0: try: time_str = line.split('=')[1].strip() curr_sec = get_seconds(time_str) pct = min(100, int((curr_sec/total_sec)*100)) last_enc_pct = pct if single_mode: print_progress(worker_id, curr_sec, total_sec, prefix='Enc') else: update_status(f"Enc {pct}%") if time.time() - last_rep > 10: post_status("processing", pct) last_rep = time.time() last_hb = last_rep except: pass with PROC_LOCK: if worker_id in ACTIVE_PROCS: del ACTIVE_PROCS[worker_id] enc_time = time.time() - start_enc if proc.returncode == 0 or SHUTDOWN_EVENT.is_set(): break gz_log_path = os.path.join(temp_dir, "encode.log.gz") try: with open(raw_log_path, 'rb') as f_in, gzip.open(gz_log_path, 'wb') as f_out: shutil.copyfileobj(f_in, f_out) except Exception as e: log(worker_id, f"Failed to compress log: {e}", "WARN") def upload_encode_log(): if os.path.exists(gz_log_path): try: with open(gz_log_path, 'rb') as lf: requests.post(f"{manager_url}/upload_log", files={'log_file': (f"{job_id.replace('/', '_')}.log.gz", lf)}, data={'job_id': job_id, 'worker_id': worker_id}, headers=get_auth_headers(), timeout=60) except Exception as e: log(worker_id, f"Failed to upload log: {e}", "WARN") if single_mode: print_progress(worker_id, total_sec, total_sec, prefix='Enc', suffix='OK') if proc and proc.returncode == 0 and os.path.exists(local_dst): final_size_bytes = os.path.getsize(local_dst) final_size = final_size_bytes / 1024 / 1024 log(worker_id, f"Encode done ({enc_time:.0f}s, {final_size:.2f}MB). Uploading...") post_status("uploading", 0) class ProgressFileReader: def __init__(self, filename, callback): self._f = open(filename, 'rb'); self._total = os.path.getsize(filename) self._read = 0; self._callback = callback; self._last_time = 0 self._last_pct = 0 def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self._f.close() def read(self, size=-1): if PAUSE_REQUESTED: _up_hb = time.time() while PAUSE_REQUESTED and not SHUTDOWN_EVENT.is_set(): if time.time() - _up_hb > 30: self._callback(self._last_pct) _up_hb = time.time() time.sleep(1) data = self._f.read(size); self._read += len(data) pct = int((self._read / self._total) * 100) self._last_pct = pct if single_mode: print_progress(worker_id, self._read, self._total, prefix='Up') else: update_status(f"Up {pct}%") if time.time() - self._last_time > 30: self._callback(pct); self._last_time = time.time() return data def __getattr__(self, attr): return getattr(self._f, attr) def upload_cb(pct): post_status("uploading", pct) # RETRY LOOP for Uploads upload_success = False for up_attempt in range(3): try: with ProgressFileReader(local_dst, upload_cb) as f: # Timeout = 300s (5 minutes) for socket silence r = requests.post(f"{manager_url}/upload_result", files={'file': (job_id, f)}, data={'job_id': job_id, 'worker_id': worker_id, 'duration': total_min}, headers=get_auth_headers(), timeout=300) if r.status_code == 200: upload_success = True break else: log(worker_id, f"Upload rejected by server: {r.status_code}", "WARN") except Exception as e: log(worker_id, f"Upload attempt {up_attempt+1} failed/timed out: {e}", "WARN") time.sleep(10) # Wait 10s before retry if upload_success: if single_mode: print_progress(worker_id, 100, 100, prefix='Up', suffix='OK') log(worker_id, "Job complete.") d = WORKER_DETAILS.get(worker_id) if d is not None: d.update({"phase": "Done", "pct": 100}) d["jobs_done"] += 1 with STATS_LOCK: SESSION_STATS["jobs_done"] += 1 SESSION_STATS["bytes_uploaded"] += final_size_bytes # Remove the resume checkpoint now that the job is finished try: _done_chk = os.path.join(temp_dir, f"{_RESUME_CHK_PREFIX}{re.sub(r'[^a-zA-Z0-9_-]', '_', str(job_id))}.json") if os.path.exists(_done_chk): os.remove(_done_chk) except: pass upload_encode_log() else: d = WORKER_DETAILS.get(worker_id) if d is not None: d["phase"] = "Failed" err_msg = "Upload failed after 3 attempts" log(worker_id, err_msg, "ERROR") post_status("failed", error_msg=err_msg) upload_encode_log() else: rc = proc.returncode if proc else -1 err_msg = f"FFmpeg exited with code {rc}" if SHUTDOWN_EVENT.is_set(): err_msg = "Aborted by user/update" log(worker_id, err_msg, "ERROR") log(worker_id, "--- FFmpeg Output Dump ---", "ERROR") for l in log_buffer: safe_print(f" {l.strip()}") log(worker_id, "--------------------------", "ERROR") post_status("failed", error_msg=err_msg) if not SHUTDOWN_EVENT.is_set(): report_error(job_id, "encode_failure", err_msg, "\n".join(l.rstrip() for l in log_buffer)) upload_encode_log() if os.path.exists(local_dst): os.remove(local_dst) if os.path.exists(raw_log_path): os.remove(raw_log_path) if os.path.exists(gz_log_path): os.remove(gz_log_path) else: if single_mode: with CONSOLE_LOCK: sys.stdout.write(f"\033[2K\r[{datetime.now().strftime('%H:%M:%S')}] [{worker_id}] Idle. Waiting...") sys.stdout.flush() time.sleep(10) except Exception as e: err_str = str(e) err_tb = traceback.format_exc() log(worker_id, f"Error: {err_str}", "CRITICAL") try: if 'job_id' in locals(): post_status("failed", error_msg=err_str) report_error(job_id, "exception", err_str, err_tb) except: pass time.sleep(10) def run_worker(args): print("==================================================") print(" FRACTUM DISTRIBUTED WORKER") print("==================================================") config_file = "worker_config.json" saved_config = {} if os.path.exists(config_file): try: with open(config_file, 'r') as f: content = f.read().strip() if content: saved_config = json.loads(content) else: print("[!] Config file is empty. Resetting.") f.close() os.remove(config_file) if args.username == DEFAULT_USERNAME and 'username' in saved_config: args.username = saved_config['username'] if args.workername == DEFAULT_WORKERNAME and 'workername' in saved_config: args.workername = saved_config['workername'] except json.JSONDecodeError: print("[!] Config file corrupted. Resetting.") os.remove(config_file) except Exception as e: print(f"[!] Warning: Could not read config file: {e}") if sys.stdin.isatty(): config_changed = False if args.username == DEFAULT_USERNAME: print("\n[*] First Time Setup detected.") print(" Please enter the USERNAME of the person running the program.") print(" (e.g., 'FractumSeraph', 'John Smith')") u_input = input(f" Enter Username (Default: {DEFAULT_USERNAME}): ").strip() if u_input: args.username = u_input config_changed = True if args.workername == DEFAULT_WORKERNAME: w_default = f"Node-{int(time.time())}" print("\n Please enter a name for THIS COMPUTER.") print(" (e.g., 'Fractums Laptop', 'Johns Gaming PC')") w_input = input(f" Enter Worker Name (Default: {w_default}): ").strip() if w_input: args.workername = w_input else: args.workername = w_default config_changed = True if config_changed: try: with open(config_file, 'w') as f: json.dump({"username": args.username, "workername": args.workername}, f, indent=4) print(f"[*] Configuration saved to {config_file}") except: print("[!] Failed to save configuration file.") check_ffmpeg() update_ffmpeg_if_stale() manager_url = (args.manager or DEFAULT_MANAGER_URL).rstrip('/') username = args.username or DEFAULT_USERNAME base_workername = args.workername or DEFAULT_WORKERNAME global WORKER_SECRET if args.secret: WORKER_SECRET = args.secret if WORKER_SECRET == "DefaultInsecureSecret": print("[*] INFO: Using default WORKER_SECRET. Compatible with public manager defaults.") if not verify_connection(manager_url): sys.exit(1) if check_version(manager_url): apply_update(manager_url) quota_tracker = None if args.daily_quota > 0: print(f"[*] Daily Quota Active: {args.daily_quota} GB") quota_tracker = QuotaTracker(args.daily_quota, base_workername) if quota_tracker.check_cap(): print(f"[!] Quota already exceeded for today. Waiting until tomorrow.") num_jobs = args.jobs if args.jobs > 0 else 1 if num_jobs > 32: num_jobs = 32 signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) threads = [] worker_ids = [] _force_tui = getattr(args, 'force_tui', False) use_tui = HAS_TEXTUAL and (not getattr(args, 'no_tui', False) or _force_tui) single_mode = (num_jobs == 1) and not use_tui # Initialize per-worker detail state and session stats global WORKER_DETAILS, SESSION_STATS WORKER_DETAILS = {} with STATS_LOCK: SESSION_STATS["jobs_done"] = 0 SESSION_STATS["bytes_uploaded"] = 0 SESSION_STATS["start"] = time.time() if args.series_id: print(f"[*] SERIES ID ACTIVE: Processing Series #{args.series_id}") for i in range(num_jobs): worker_id = f"{username}-{base_workername}-{i+1}" worker_ids.append(worker_id) WORKER_DETAILS[worker_id] = {"file": "-", "phase": "Starting", "pct": 0, "job_start": 0.0, "jobs_done": 0} temp_dir = f"./temp_encode_{base_workername}_{i+1}" t = threading.Thread(target=worker_task, args=(worker_id, manager_url, temp_dir, quota_tracker, single_mode, args.series_id, args.watermark, args.max_size_mb, getattr(args, 'local_source', None))) t.daemon = True t.start() threads.append(t) if not single_mode and not use_tui: monitor_t = threading.Thread(target=monitor_status_loop, args=(worker_ids,)) monitor_t.daemon = True monitor_t.start() global PAUSE_REQUESTED, TUI_APP if use_tui: app = WorkerApp( worker_ids=worker_ids, threads=threads, quota_tracker=quota_tracker, manager_url=manager_url, ) TUI_APP = app # Save Windows console modes now (before Textual modifies them) so # we can unconditionally restore them on any exit path — including # crashes and abnormal shutdowns that skip Textual's own cleanup. # Without this, echo is left disabled, causing the "can't see typing" # symptom and ANSI mouse-tracking sequences leaking into the shell. if platform.system() == 'Windows': try: import ctypes, ctypes.wintypes, atexit as _atexit _k32 = ctypes.windll.kernel32 _hIn = _k32.GetStdHandle(-10) # STD_INPUT_HANDLE _hOut = _k32.GetStdHandle(-11) # STD_OUTPUT_HANDLE _m_in = ctypes.wintypes.DWORD() _m_out = ctypes.wintypes.DWORD() _k32.GetConsoleMode(_hIn, ctypes.byref(_m_in)) _k32.GetConsoleMode(_hOut, ctypes.byref(_m_out)) def _restore_win_console(): try: # Full terminal reset sequence — covers every mode Textual # may have enabled that survives an abnormal exit. sys.stdout.write( '\033[?1049l' # exit alternate screen buffer (main symptom: # cursor stuck at top-left after exit) '\033[?2004l' # disable bracketed-paste mode '\033[?1000l' # disable basic mouse tracking '\033[?1002l' # disable button-event tracking '\033[?1003l' # disable any-event tracking '\033[?1006l' # disable SGR extended mouse mode '\033[?1015l' # disable URXVT extended mouse mode '\033[?25h' # ensure cursor is visible '\033[0m' # reset all SGR attributes (color, bold…) '\r\n' # move to a clean line ) sys.stdout.flush() except Exception: pass try: _k32.SetConsoleMode(_hIn, _m_in) _k32.SetConsoleMode(_hOut, _m_out) except Exception: pass _atexit.register(_restore_win_console) except Exception: pass # Run the TUI. Mouse support is left enabled so RichLog and DataTable # remain scrollable with the mouse wheel. The atexit handler and the # reset block below send the disable sequences after Textual exits, # which is the only safe time to write raw escapes to the terminal. global _TUI_WAS_RUNNING _TUI_WAS_RUNNING = True app.run() TUI_APP = None # Explicit terminal reset after a clean exit — Textual should have # done this itself, but belt-and-suspenders for any missed sequences. try: sys.stdout.write( '\033[?1049l' # exit alternate screen buffer '\033[?2004l' # disable bracketed-paste mode '\033[?1000l\033[?1002l\033[?1003l\033[?1006l\033[?1015l' '\033[?25h' # show cursor '\033[0m' # reset SGR attributes '\r\n' ) sys.stdout.flush() except Exception: pass else: if HAS_TEXTUAL and getattr(args, 'no_tui', False): _tui_reason = "--no-tui flag passed" else: _tui_reason = _TEXTUAL_UNAVAIL_REASON or "textual not installed (pip install textual)" print(f"Fractum Worker v{WORKER_VERSION} — TUI disabled ({_tui_reason})") while True: if not PAUSE_REQUESTED: all_dead = True for t in threads: if t.is_alive(): all_dead = False; break if all_dead: break if SHUTDOWN_EVENT.is_set() and not PAUSE_REQUESTED: break time.sleep(0.5) continue MONITOR_PAUSED.set() # stop monitor from clobbering stdout time.sleep(0.6) # give it a moment to stop writing sys.stdout.write('\n') # move to a clean line sys.stdout.flush() toggle_processes(suspend=True) print("\n" + "="*40) print(" [!] WORKER PAUSED") print("="*40) print(" [C]ontinue - Resume encoding") print(" [F]inish - Finish active, then stop") print(" [S]top - Abort immediately") print(" Ctrl+C again - Force abort") # Open /dev/tty directly so input works even when stdin is a pipe # (e.g. launched via curl | bash or inside tmux with redirected stdin). _input_src = None if not sys.stdin.isatty(): try: _input_src = open('/dev/tty', 'r') except Exception: pass def _read_choice(): src = _input_src if _input_src else sys.stdin sys.stdout.write("Select [c/f/s]: ") sys.stdout.flush() line = src.readline() if not line: raise EOFError return line.strip().lower() while PAUSE_REQUESTED: try: choice = _read_choice() if choice == 'c': print("[*] Resuming...") MONITOR_PAUSED.clear() PAUSE_REQUESTED = False toggle_processes(suspend=False) elif choice == 'f': print("[*] Draining jobs...") PAUSE_REQUESTED = False MONITOR_PAUSED.clear() toggle_processes(suspend=False) SHUTDOWN_EVENT.set() elif choice == 's': print("[*] Aborting...") toggle_processes(suspend=False) kill_processes() SHUTDOWN_EVENT.set() PAUSE_REQUESTED = False if _input_src: try: _input_src.close() except: pass sys.exit(0) except KeyboardInterrupt: # Second Ctrl+C while paused = force stop print("\n[*] Force abort.") toggle_processes(suspend=False) kill_processes() SHUTDOWN_EVENT.set() PAUSE_REQUESTED = False if _input_src: try: _input_src.close() except: pass sys.exit(0) except EOFError: # stdin fully exhausted and /dev/tty unavailable — can't prompt print("\n[!] No input available — aborting.") toggle_processes(suspend=False) kill_processes() SHUTDOWN_EVENT.set() PAUSE_REQUESTED = False sys.exit(0) except Exception: time.sleep(0.5) if _input_src: try: _input_src.close() except: pass if UPDATE_AVAILABLE: apply_update(manager_url) if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--manager", default=DEFAULT_MANAGER_URL) parser.add_argument("--username", default=DEFAULT_USERNAME) parser.add_argument("--workername", default=DEFAULT_WORKERNAME) parser.add_argument("--jobs", type=int, default=1) parser.add_argument("--series-id", default=None, help="Process only specific Series ID") parser.add_argument("--secret", default=None, help="Manually set worker secret token") parser.add_argument("--daily-quota", type=float, default=0, help="Daily download limit in GB (0 = unlimited)") parser.add_argument("--watermark", action="store_true", default=False, help="Burn the @FractumSeraph watermark into encoded video") parser.add_argument("--no-tui", action="store_true", default=False, help="Disable Textual TUI and use plain terminal output") parser.add_argument("--force-tui", action="store_true", default=False, help="Force Textual TUI even in SSH/tmux environments (may have input issues)") parser.add_argument("--max-size-mb", type=int, default=0, help="Skip source files larger than this size in MB (0 = no limit)") parser.add_argument("--local-source", default=None, metavar="DIR", help="Path to the source media directory on this machine. When set and the " "source file exists locally, it is read directly instead of streaming " "over HTTP (useful when this worker runs on the server itself).") args = parser.parse_args() run_worker(args)