from dataclasses import dataclass from datetime import datetime, timedelta from typing import Optional @dataclass class ScheduleConfig: time: str postpone_on_activity: bool = True max_downtime: int = 120 retry_failed: bool = True retry_interval: int = 3600 class Scheduler: def __init__(self, config: ScheduleConfig) -> None: self.config = config self._last_run: Optional[datetime] = None self._last_failed: Optional[datetime] = None def should_run(self, last_activity: Optional[datetime] = None) -> bool: now = datetime.now() if self._last_run and self._last_run.date() == now.date(): return False if self._last_failed and self.config.retry_failed: if now - self._last_failed >= timedelta(seconds=self.config.retry_interval): return True target = datetime.strptime(self.config.time, "%H:%M").time() target_dt = datetime.combine(now.date(), target) if now < target_dt: return False if self.config.postpone_on_activity and last_activity: if (now - last_activity).total_seconds() < self.config.max_downtime: return False return True def mark_run(self) -> None: self._last_run = datetime.now() self._last_failed = None def mark_failed(self) -> None: self._last_failed = datetime.now()