43 lines
1.4 KiB
Python
43 lines
1.4 KiB
Python
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()
|