如何在Python中获得类似Cron的调度程序？

我正在寻找一个Python的库,它将提供at和cron功能相似.

我非常喜欢使用纯Python解决方案,而不是依赖于盒子上安装的工具; 这样我就可以在没有cron的机器上运行.

对于那些不熟悉的人cron:您可以根据以下表达式安排任务:

 0 2 * * 7 /usr/bin/run-backup # run the backups at 0200 on Every Sunday
 0 9-17/2 * * 1-5 /usr/bin/purge-temps # run the purge temps command, every 2 hours between 9am and 5pm on Mondays to Fridays.

cron时间表达式语法不那么重要,但我希望能有这种灵活性.

如果没有任何东西可以为我开箱即用,那么任何关于构建模块的建议都会感激不尽.

编辑 我对启动进程不感兴趣,只是用Python编写的"作业" - python函数.必要时我认为这将是一个不同的主题,但不是在一个不同的过程中.

为此,我正在寻找cron时间表达式的表现力,但在Python中.

Cron 已经存在多年了,但我试图尽可能地便携.我不能依赖它的存在.

1> dbader..：

---

如果您正在寻找轻量级结账时间表:

import schedule
import time

def job():
    print("I'm working...")

schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)

while 1:
    schedule.run_pending()
    time.sleep(1)

披露:我是该图书馆的作者.

---

有没有办法将参数传递给作业？我想做这样的事情:schedule.every().hour.do(job(myParam))

---

@darrel-holt和@ zen-skunkworx:`do()`函数将传递给它的额外参数转发给作业函数:https://schedule.readthedocs.io/en/stable/api.html#schedule.Job .do例如,你可以这样做:`schedule.every().hour.do(job,param1,param2)`不需要使用lambda.希望这可以帮助 :)

---

schedule.every().hour.do(job)每小时运行一次吗？喜欢01:00,02:00,03:00等？即使开始时间不是整整一个小时？

---

你应该提到你是`schedule`的维护者.它对我很有用.如果它具有类似cron的语法和支持的装饰器会更好(参见crython但不使用此库因为它不起作用;调度似乎写得不好).

2> Brian..：

---

您可以使用普通的Python参数传递语法来指定您的crontab.例如,假设我们定义一个Event类,如下所示:

from datetime import datetime, timedelta
import time

# Some utility classes / functions first
class AllMatch(set):
    """Universal set - match everything"""
    def __contains__(self, item): return True

allMatch = AllMatch()

def conv_to_set(obj):  # Allow single integer to be provided
    if isinstance(obj, (int,long)):
        return set([obj])  # Single item
    if not isinstance(obj, set):
        obj = set(obj)
    return obj

# The actual Event class
class Event(object):
    def __init__(self, action, min=allMatch, hour=allMatch, 
                       day=allMatch, month=allMatch, dow=allMatch, 
                       args=(), kwargs={}):
        self.mins = conv_to_set(min)
        self.hours= conv_to_set(hour)
        self.days = conv_to_set(day)
        self.months = conv_to_set(month)
        self.dow = conv_to_set(dow)
        self.action = action
        self.args = args
        self.kwargs = kwargs

    def matchtime(self, t):
        """Return True if this event should trigger at the specified datetime"""
        return ((t.minute     in self.mins) and
                (t.hour       in self.hours) and
                (t.day        in self.days) and
                (t.month      in self.months) and
                (t.weekday()  in self.dow))

    def check(self, t):
        if self.matchtime(t):
            self.action(*self.args, **self.kwargs)

(注意:未经过彻底测试)

然后你的CronTab可以用普通的python语法指定为:

c = CronTab(
  Event(perform_backup, 0, 2, dow=6 ),
  Event(purge_temps, 0, range(9,18,2), dow=range(0,5))
)

通过这种方式,您可以获得Python的参数机制的全部功能(混合位置和关键字args,并且可以使用符号名称来表示数周和数月的名称)

CronTab类将被定义为以微小的增量进行简单的睡眠,并在每个事件上调用check().(可能会有一些细微之处,夏令时/时区要警惕).这是一个快速实现:

class CronTab(object):
    def __init__(self, *events):
        self.events = events

    def run(self):
        t=datetime(*datetime.now().timetuple()[:5])
        while 1:
            for e in self.events:
                e.check(t)

            t += timedelta(minutes=1)
            while datetime.now() < t:
                time.sleep((t - datetime.now()).seconds)

需要注意的一点是:Python的工作日/月是零索引(与cron不同),并且该范围排除了最后一个元素,因此像"1-5"这样的语法变为范围(0,5) - 即[0,1,2, 3,4].如果您更喜欢cron语法,那么解析它应该不会太困难.

3> ssc..：

---

也许只有在提出问题之后才会出现这种情况; 我想我只是为了完整起见而提到它:https://apscheduler.readthedocs.org/en/latest/

4> Sean..：

---

在我的搜索中我看到的一件事是python的sched模块,它可能是你正在寻找的那种东西.

---

sched现在有enterabs()进行绝对调度.

---

我希望这是首选的答案，因为sched现在是python2和3 stdlib的一部分，并且可以进行绝对调度。

5> bootload..：

---

"... Crontab模块用于读取和写入crontab文件并自动访问系统cron并简单地使用直接API ...."

http://pypi.python.org/pypi/python-crontab

以及APScheduler,一个python包.已经编写和调试.

http://packages.python.org/APScheduler/cronschedule.html

6> Hackeron..：

---

或多或少与上面相同,但使用gevent并发:)

"""Gevent based crontab implementation"""

from datetime import datetime, timedelta
import gevent

# Some utility classes / functions first
def conv_to_set(obj):
    """Converts to set allowing single integer to be provided"""

    if isinstance(obj, (int, long)):
        return set([obj])  # Single item
    if not isinstance(obj, set):
        obj = set(obj)
    return obj

class AllMatch(set):
    """Universal set - match everything"""
    def __contains__(self, item): 
        return True

allMatch = AllMatch()

class Event(object):
    """The Actual Event Class"""

    def __init__(self, action, minute=allMatch, hour=allMatch, 
                       day=allMatch, month=allMatch, daysofweek=allMatch, 
                       args=(), kwargs={}):
        self.mins = conv_to_set(minute)
        self.hours = conv_to_set(hour)
        self.days = conv_to_set(day)
        self.months = conv_to_set(month)
        self.daysofweek = conv_to_set(daysofweek)
        self.action = action
        self.args = args
        self.kwargs = kwargs

    def matchtime(self, t1):
        """Return True if this event should trigger at the specified datetime"""
        return ((t1.minute     in self.mins) and
                (t1.hour       in self.hours) and
                (t1.day        in self.days) and
                (t1.month      in self.months) and
                (t1.weekday()  in self.daysofweek))

    def check(self, t):
        """Check and run action if needed"""

        if self.matchtime(t):
            self.action(*self.args, **self.kwargs)

class CronTab(object):
    """The crontab implementation"""

    def __init__(self, *events):
        self.events = events

    def _check(self):
        """Check all events in separate greenlets"""

        t1 = datetime(*datetime.now().timetuple()[:5])
        for event in self.events:
            gevent.spawn(event.check, t1)

        t1 += timedelta(minutes=1)
        s1 = (t1 - datetime.now()).seconds + 1
        print "Checking again in %s seconds" % s1
        job = gevent.spawn_later(s1, self._check)

    def run(self):
        """Run the cron forever"""

        self._check()
        while True:
            gevent.sleep(60)

import os 
def test_task():
    """Just an example that sends a bell and asd to all terminals"""

    os.system('echo asd | wall')  

cron = CronTab(
  Event(test_task, 22, 1 ),
  Event(test_task, 0, range(9,18,2), daysofweek=range(0,5)),
)
cron.run()

7> rouble..：

---

列出的解决方案都没有尝试解析复杂的cron调度字符串.所以,这是我的版本,使用croniter.基本要点:

schedule = "*/5 * * * *" # Run every five minutes

nextRunTime = getNextCronRunTime(schedule)
while True:
     roundedDownTime = roundDownTime()
     if (roundedDownTime == nextRunTime):
         ####################################
         ### Do your periodic thing here. ###
         ####################################
         nextRunTime = getNextCronRunTime(schedule)
     elif (roundedDownTime > nextRunTime):
         # We missed an execution. Error. Re initialize.
         nextRunTime = getNextCronRunTime(schedule)
     sleepTillTopOfNextMinute()

帮助程序:

from croniter import croniter
from datetime import datetime, timedelta

# Round time down to the top of the previous minute
def roundDownTime(dt=None, dateDelta=timedelta(minutes=1)):
    roundTo = dateDelta.total_seconds()
    if dt == None : dt = datetime.now()
    seconds = (dt - dt.min).seconds
    rounding = (seconds+roundTo/2) // roundTo * roundTo
    return dt + timedelta(0,rounding-seconds,-dt.microsecond)

# Get next run time from now, based on schedule specified by cron string
def getNextCronRunTime(schedule):
    return croniter(schedule, datetime.now()).get_next(datetime)

# Sleep till the top of the next minute
def sleepTillTopOfNextMinute():
    t = datetime.utcnow()
    sleeptime = 60 - (t.second + t.microsecond/1000000.0)
    time.sleep(sleeptime)

8> 小智..：

---

我修改了脚本.

使用方便:

cron = Cron()
cron.add('* * * * *'   , minute_task) # every minute
cron.add('33 * * * *'  , day_task)    # every hour
cron.add('34 18 * * *' , day_task)    # every day
cron.run()

尝试在一分钟的第一秒开始任务.

Github上的代码

9> 小智..：

---

我对Brian建议的CronTab类运行方法有一个小修复.

时间已经过了一秒钟,在每分钟结束时导致一秒钟的硬循环.

class CronTab(object):
    def __init__(self, *events):
        self.events = events

    def run(self):
        t=datetime(*datetime.now().timetuple()[:5])
        while 1:
            for e in self.events:
                e.check(t)

            t += timedelta(minutes=1)
            n = datetime.now()
            while n < t:
                s = (t - n).seconds + 1
                time.sleep(s)
                n = datetime.now()