摘要：问题描述：我正在寻找一个能够提供类似功能的 Pythonat库cron。我非常希望有一个纯 Python 解决方案，而不是依赖于机器上安装的工具；这样我就可以在没有 cron 的机器上运行。对于那些不熟悉的人cron：您可以根据以下表达式来安排任务： 0 2 * * 7 /usr/bin/run-backup...

问题描述：

我正在寻找一个能够提供类似功能的 Pythonat库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：

如果您正在寻找一些轻量级的结帐时间表：

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)

披露：我是该图书馆的作者。

解决方案 2：

您可以使用普通的 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)

（注：未经彻底测试）

然后，您可以使用正常的 Python 语法指定您的 CronTab，如下所示：

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

这样，您就可以充分利用 Python 的参数机制（混合位置和关键字参数，并且可以使用符号名称作为周和月的名称）

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：

大致与上面相同，但是使用 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()

解决方案 4：

Croniter：解析 CRON 字符串并获取下一个事件时间序列

Croniter 的文档可以在这里找到

列出的解决方案均未采用解析 CRON 计划字符串的方法。因此，这是我使用croniter的版本。基本要点：

import croniter
import datetime

now = datetime.datetime.now()
iter = croniter.croniter('0 0 * * *', now)

next_datetime = croniter.get_next(datetime.datetime) # datetime not now!

remaining_seconds = next_datetime - now
remaining_seconds = remaining_seconds.total_seconds()
print(f'seconds until next event: {remaining_seconds}')

简单的！

具有更详细逻辑的更复杂的用例：

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)

解决方案 5：

我喜欢pycron包解决这个问题的方式。

import pycron
import time

while True:
    if pycron.is_now('0 2 * * 0'):   # True Every Sunday at 02:00
        print('running backup')
        time.sleep(60)               # The process should take at least 60 sec
                                     # to avoid running twice in one minute
    else:
        time.sleep(15)               # Check again in 15 seconds

解决方案 6：

我知道有很多答案，但另一个解决方案可能是使用装饰器。这是一个在特定时间每天重复执行某个功能的示例。使用这种方法的妙处在于，您只需要在要安排的函数中添加语法糖：

@repeatEveryDay(hour=6, minutes=30)
def sayHello(name):
    print(f"Hello {name}")

sayHello("Bob") # Now this function will be invoked every day at 6.30 a.m

装饰器看起来如下：

def repeatEveryDay(hour, minutes=0, seconds=0):
    """
    Decorator that will run the decorated function everyday at that hour, minutes and seconds.
    :param hour: 0-24
    :param minutes: 0-60 (Optional)
    :param seconds: 0-60 (Optional)
    """
    def decoratorRepeat(func):

        @functools.wraps(func)
        def wrapperRepeat(*args, **kwargs):

            def getLocalTime():
                return datetime.datetime.fromtimestamp(time.mktime(time.localtime()))

            # Get the datetime of the first function call
            td = datetime.timedelta(seconds=15)
            if wrapperRepeat.nextSent == None:
                now = getLocalTime()
                wrapperRepeat.nextSent = datetime.datetime(now.year, now.month, now.day, hour, minutes, seconds)
                if wrapperRepeat.nextSent < now:
                    wrapperRepeat.nextSent += td

            # Waiting till next day
            while getLocalTime() < wrapperRepeat.nextSent:
                time.sleep(1)

            # Call the function
            func(*args, **kwargs)

            # Get the datetime of the next function call
            wrapperRepeat.nextSent += td
            wrapperRepeat(*args, **kwargs)

        wrapperRepeat.nextSent = None
        return wrapperRepeat

    return decoratorRepeat

解决方案 7：

没有“纯 Python”的方式来做到这一点，因为其他进程必须启动 Python 才能运行您的解决方案。每个平台都会有一到二十种不同的方式来启动进程并监控其进度。在 Unix 平台上，cron 是旧标准。在 Mac OS X 上还有 launchd，它将类似 cron 的启动与监视程序功能相结合，如果您需要，它可以让您的进程保持活动状态。一旦 Python 运行，您就可以使用sched 模块来安排任务。

解决方案 8：

另一个简单的解决方案是：

from aqcron import At
from time import sleep
from datetime import datetime

# Event scheduling
event_1 = At( second=5 )
event_2 = At( second=[0,20,40] )

while True:
    now = datetime.now()

    # Event check
    if now in event_1: print "event_1"
    if now in event_2: print "event_2"

    sleep(1)

并且 aqcron.At 类是：

# aqcron.py

class At(object):
    def __init__(self, year=None,    month=None,
                 day=None,     weekday=None,
                 hour=None,    minute=None,
                 second=None):
        loc = locals()
        loc.pop("self")
        self.at = dict((k, v) for k, v in loc.iteritems() if v != None)

    def __contains__(self, now):
        for k in self.at.keys():
            try:
                if not getattr(now, k) in self.at[k]: return False
            except TypeError:
                if self.at[k] != getattr(now, k): return False
        return True

解决方案 9：

我不知道是否已经存在类似的东西。使用时间、日期时间和/或日历模块编写自己的模块很容易，请参阅http://docs.python.org/library/time.html

对于 Python 解决方案，唯一需要担心的是，你的工作需要始终运行，并且可能在重启后自动“复活”，而对于这一点，你需要依赖系统相关的解决方案。

解决方案 10：

如果我需要检查有多少个匹配项并且我不能一直等待在同一点，那么一个简单的离线检查器可能看起来像：

import re
import random
import time
from datetime import datetime, timedelta
from datetime import datetime, timezone, tzinfo

class Crontab:
  ALL = '.*'
  STEP = timedelta(seconds=1)

def __init__(self, **specs):
      self.t0 = self.now()
      self.specs = specs

  def now(self):
      now = datetime.now(tz=timezone.utc).replace(microsecond=0)
      return now

  def check(self):
      "Fires all time that matches from last call"
      now = self.now()
      while self.t0 <= now:
          self.t0 += self.STEP
        for key, pattern in self.specs.items():
            value = getattr(self.t0, key, None)
            if value is not None:
                if not isinstance(value, int):
                    value = value()  # method
                if not re.match(f"{pattern}$", f"{value}"):
                    break
        else:
            # all specs (if any) matches
            return self.t0


if __name__ == "__main__":
cron = Crontab(second='0', minute='0|15|30|45')
while True:
    while not (t := cron.check()):
        time.sleep(random.randint(1, 10))
    print(f"match at: {t}")

输出