from __future__ import annotations import sys from pathlib import Path import servicemanager import win32event import win32service import win32serviceutil if __package__ in {None, ""}: sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from bakrest.config import load_config from bakrest.logging_setup import configure_logging from bakrest.watcher import WatchdogRuntime class BakRestWatchdogService(win32serviceutil.ServiceFramework): _svc_name_ = "BakRestWatchdog" _svc_display_name_ = "Bak&Rest Watchdog Service" _svc_description_ = "Monitors configured folders and records file changes for Bak&Rest backups." def __init__(self, args: list[str]) -> None: super().__init__(args) self.stop_event_handle = win32event.CreateEvent(None, 0, 0, None) self.runtime: WatchdogRuntime | None = None def SvcStop(self) -> None: self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING) if self.runtime: self.runtime.stop() win32event.SetEvent(self.stop_event_handle) def SvcDoRun(self) -> None: servicemanager.LogMsg( servicemanager.EVENTLOG_INFORMATION_TYPE, servicemanager.PYS_SERVICE_STARTED, (self._svc_name_, ""), ) config = load_config() logger = configure_logging(config.storage.logs_dir, "service") logger.info("Windows service started with configuration: %s", config.path) self.runtime = WatchdogRuntime(config, logger) try: self.runtime.run() except Exception: logger.exception("BakRestWatchdog failed") raise def main() -> None: _apply_configured_startup() win32serviceutil.HandleCommandLine(BakRestWatchdogService) def _apply_configured_startup() -> None: if len(sys.argv) < 2 or sys.argv[1].lower() != "install": return if "--startup" in sys.argv: return config = load_config() startup = "auto" if config.service_autostart else "manual" sys.argv.extend(["--startup", startup]) if __name__ == "__main__": main()