import os import sys import time import logging import subprocess import threading from datetime import datetime, timedelta from typing import Optional, Callable from .models import AutoRestartConfig, RestartInterval, RestartTarget logger = logging.getLogger(__name__) class AutoRestartScheduler: def __init__(self, config: AutoRestartConfig, app_path: Optional[str] = None): self.config = config self.app_path = app_path or sys.executable self._running = False self._thread: Optional[threading.Thread] = None self._last_trigger_minute = None self.on_restart_triggered: Optional[Callable[[], None]] = None def start(self): if not self.config.enabled: logger.info("AutoRestart disabled") return if self._running: return self._running = True self._thread = threading.Thread(target=self._loop, daemon=True) self._thread.start() logger.info(f"AutoRestart started: interval={self.config.interval.value}, " f"time={self.config.hour:02d}:{self.config.minute:02d}, " f"target={self.config.target.value}") def stop(self): self._running = False def _loop(self): while self._running: now = datetime.now() if self._should_trigger(now): minute_key = now.strftime("%Y-%m-%d %H:%M") if self._last_trigger_minute != minute_key: self._last_trigger_minute = minute_key logger.info(f"AutoRestart triggered at {now}") if self.on_restart_triggered: self.on_restart_triggered() else: self.perform_restart() time.sleep(30) def _should_trigger(self, now: datetime) -> bool: if now.hour != self.config.hour or now.minute != self.config.minute: return False interval = self.config.interval if interval == RestartInterval.DAY: return True elif interval == RestartInterval.WEEK: return now.weekday() == self.config.weekdays elif interval == RestartInterval.MONTH: return now.day == self.config.day_of_month return False def perform_restart(self): logger.info(f"Performing restart: target={self.config.target.value}") if self.config.target == RestartTarget.DEVICE: self._restart_device() else: self._restart_app() def _restart_app(self): try: args = sys.argv[:] if getattr(sys, 'frozen', False): subprocess.Popen([self.app_path] + args[1:]) else: subprocess.Popen([sys.executable] + args) logger.info("App restart command sent") os._exit(0) except Exception as e: logger.error(f"Failed to restart app: {e}") def _restart_device(self): try: if os.name == 'nt': subprocess.run(['shutdown', '/r', '/t', '5', '/c', 'AutoRestart scheduled'], check=False) else: subprocess.run(['sudo', 'reboot'], check=False) logger.info("Device restart command sent") except Exception as e: logger.error(f"Failed to restart device: {e}") def next_restart_time(self) -> Optional[datetime]: now = datetime.now() interval = self.config.interval target_time = now.replace(hour=self.config.hour, minute=self.config.minute, second=0, microsecond=0) if target_time <= now: if interval == RestartInterval.DAY: target_time += timedelta(days=1) elif interval == RestartInterval.WEEK: days_ahead = self.config.weekdays - now.weekday() if days_ahead <= 0: days_ahead += 7 target_time += timedelta(days=days_ahead) elif interval == RestartInterval.MONTH: if now.month == 12: next_month = 1 next_year = now.year + 1 else: next_month = now.month + 1 next_year = now.year try: target_time = target_time.replace(year=next_year, month=next_month, day=self.config.day_of_month) except ValueError: target_time += timedelta(days=1) return target_time