Coverage for merco/scheduler/cron.py: 34%

1"""Cron 调度器""" 

2 

3import asyncio 

4from datetime import datetime 

5from typing import Callable, Awaitable 

6 

7 

8class CronJob: 

9 """定时任务""" 

10 

11 def __init__(self, name: str, schedule: str, handler: Callable): 

12 self.name = name 

13 self.schedule = schedule # cron 表达式 

14 self.handler = handler 

15 self.enabled = True 

16 self.last_run = None 

17 self.run_count = 0 

18 

19 

20class CronScheduler: 

21 """Cron 表达式调度器""" 

22 

23 def __init__(self): 

24 self._jobs: dict[str, CronJob] = {} 

25 self._running = False 

26 

27 def add_job(self, name: str, schedule: str, handler: Callable): 

28 """添加定时任务""" 

29 self._jobs[name] = CronJob(name, schedule, handler) 

30 

31 def remove_job(self, name: str): 

32 """移除定时任务""" 

33 self._jobs.pop(name, None) 

34 

35 def list_jobs(self) -> list[dict]: 

36 """列出所有任务""" 

37 return [ 

38 { 

39 "name": job.name, 

40 "schedule": job.schedule, 

41 "enabled": job.enabled, 

42 "last_run": job.last_run, 

43 "run_count": job.run_count, 

44 } 

45 for job in self._jobs.values() 

46 ] 

47 

48 async def start(self): 

49 """启动调度器""" 

50 self._running = True 

51 while self._running: 

52 await self._check_jobs() 

53 await asyncio.sleep(60) # 每分钟检查 

54 

55 async def stop(self): 

56 """停止调度器""" 

57 self._running = False 

58 

59 async def _check_jobs(self): 

60 """检查并执行到期任务""" 

61 now = datetime.now() 

62 for job in self._jobs.values(): 

63 if job.enabled and self._is_due(job.schedule, now): 

64 await self._run_job(job) 

65 

66 async def _run_job(self, job: CronJob): 

67 """执行单个任务""" 

68 try: 

69 if asyncio.iscoroutinefunction(job.handler): 

70 await job.handler() 

71 else: 

72 job.handler() 

73 job.last_run = datetime.now() 

74 job.run_count += 1 

75 except Exception as e: 

76 pass # TODO: 添加错误处理与通知 

77 

78 @staticmethod 

79 def _is_due(schedule: str, now: datetime) -> bool: 

80 """检查 cron 表达式是否匹配当前时间""" 

81 # TODO: 实现完整的 cron 解析 

82 # 简化实现：支持 "* * * * *" (每分钟) 和具体数字 

83 parts = schedule.split() 

84 if len(parts) != 5: 

85 return False 

86 

87 minute, hour, day, month, weekday = parts 

88 

89 def match(part, value): 

90 if part == "*": 

91 return True 

92 return str(value) == part 

93 

94 return ( 

95 match(minute, now.minute) 

96 and match(hour, now.hour) 

97 and match(day, now.day) 

98 and match(month, now.month) 

99 and match(weekday, now.weekday()) 

100 )