Jak uzyskać harmonogram podobny do Crona w Pythonie? [Zamknięte]

Szukam biblioteki w Pythonie, który zapewni ati cronjak funkcjonalność.

Wolałbym mieć czyste rozwiązanie Python, zamiast polegać na narzędziach zainstalowanych w pudełku; w ten sposób uruchamiam na maszynach bez crona.

Dla tych, którzy nie są zaznajomieni z cron: możesz zaplanować zadania w oparciu o wyrażenie takie jak:

 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.

Składnia wyrażenia czasu cron jest mniej ważna, ale chciałbym mieć coś z taką elastycznością.

Jeśli nie ma czegoś, co zrobiłoby to dla mnie natychmiast po wyjęciu z pudełka, wszelkie sugestie dotyczące elementów konstrukcyjnych, aby stworzyć coś takiego, byłyby z wdzięcznością przyjęte.

Edycja Nie jestem zainteresowany uruchamianiem procesów, tylko „zadania” napisane również w funkcjach Pythona. Z konieczności uważam, że byłby to inny wątek, ale nie w innym procesie.

W tym celu szukam ekspresji wyrażenia czasu cron, ale w Pythonie.

Cron istnieje od lat, ale staram się być jak najbardziej przenośny. Nie mogę polegać na jego obecności.

Jeśli szukasz lekkiego harmonogramu realizacji transakcji :

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)

Ujawnienie : Jestem autorem tej biblioteki.

Możesz po prostu użyć normalnej składni przekazywania argumentów w języku Python, aby określić tabelę crontab. Załóżmy na przykład, że definiujemy klasę Event jak poniżej:

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)

(Uwaga: Nie gruntownie przetestowane)

Następnie tablicę CronTab można określić w normalnej składni Pythona jako:

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

W ten sposób zyskujesz pełną moc mechaniki argumentów Pythona (mieszanie argumentów pozycyjnych i słów kluczowych, i możesz używać nazw symbolicznych dla nazw tygodni i miesięcy)

Klasa CronTab zostałaby zdefiniowana jako po prostu spanie w przyrostach minutowych i wywoływanie check () dla każdego zdarzenia. (Prawdopodobnie są pewne subtelności z czasem letnim / strefami czasowymi, o których należy pamiętać). Oto szybka implementacja:

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)

Należy zwrócić uwagę na kilka rzeczy: dni tygodnia / miesiące Pythona są indeksowane przez zero (w przeciwieństwie do crona), a zakres ten wyklucza ostatni element, stąd składnia taka jak „1-5” staje się zakresem (0,5) - tj. [0,1,2, 3,4]. Jeśli wolisz składnię crona, parsowanie nie powinno być jednak zbyt trudne.

może pojawiło się to dopiero po zadaniu pytania; Pomyślałem, że wspomnę o tym wyłącznie ze względu na kompletność: https://apscheduler.readthedocs.org/en/latest/

Jedną z rzeczy, które widziałem podczas moich poszukiwań, jest schedmoduł Pythona, który może być tym, czego szukasz.

„... Moduł Crontab do odczytu i zapisu plików crontab oraz automatycznego dostępu do systemowego crona za pomocą bezpośredniego interfejsu API.”

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

a także APScheduler, pakiet python. Już napisane i debugowane.

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

Mniej więcej tak samo jak powyżej, ale jednocześnie używa 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()

Żadne z wymienionych rozwiązań nie próbuje nawet przeanalizować złożonego ciągu harmonogramu cron. Oto moja wersja, używając cronitera . Podstawowa treść:

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()

Procedury pomocnicze:

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)

Zmodyfikowałem skrypt.

1. Łatwy w użyciu:

   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()

2. Spróbuj rozpocząć zadanie w pierwszej sekundzie minuty.

Kod na Github

Mam drobną poprawkę do metody uruchamiania klasy CronTab zasugerowanej przez Briana .

Czas upłynął o jedną sekundę, co doprowadziło do jednosekundowej, twardej pętli pod koniec każdej minuty.

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()

Nie ma na to sposobu „czystego Pythona”, ponieważ jakiś inny proces musiałby uruchomić Pythona, aby uruchomić twoje rozwiązanie. Każda platforma będzie miała jeden lub dwadzieścia różnych sposobów uruchamiania procesów i monitorowania ich postępów. Na platformach unix cron jest starym standardem. W systemie Mac OS X uruchomiono również aplikację, która łączy uruchamianie typu cron z funkcją watchdog, która może utrzymać proces przy życiu, jeśli tego chcesz. Po uruchomieniu Pythona możesz użyć modułu harmonogramu do planowania zadań.

Wiem, że jest wiele odpowiedzi, ale innym rozwiązaniem może być dekorator . To jest przykład powtarzania funkcji codziennie o określonej godzinie. Fajnym pomysłem na używanie w ten sposób jest to, że wystarczy dodać cukier syntaktyczny do funkcji, którą chcesz zaplanować:

@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

A dekorator będzie wyglądał następująco:

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

Rozwiązanie Briana działa całkiem dobrze. Jednak, jak zauważyli inni, w kodzie uruchamiania występuje subtelny błąd. Również uznałem, że jest to zbyt skomplikowane dla potrzeb.

Oto moja prostsza i funkcjonalna alternatywa dla kodu uruchamiania na wypadek, gdyby ktoś go potrzebował:

def run(self):
    while 1:
        t = datetime.now()
        for e in self.events:
            e.check(t)

        time.sleep(60 - t.second - t.microsecond / 1000000.0)

Innym trywialnym rozwiązaniem byłoby:

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)

I klasa aqcron. Jest to:

# 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

Jeśli szukasz rozproszonego harmonogramu, możesz sprawdzić https://github.com/sherinkurian/mani - to wymaga redis, ale może nie być tym, czego szukasz. (zauważ, że jestem autorem) został on zbudowany w celu zapewnienia odporności na awarie, ponieważ zegar działa na więcej niż jednym węźle.

Nie wiem, czy coś takiego już istnieje. Łatwo byłoby napisać własny moduł czasu, godziny i / lub kalendarza, patrz http://docs.python.org/library/time.html

Jedynym problemem dla Pythona rozwiązania jest to, że potrzebuje pracy, aby być zawsze działa i może być automatycznie „wskrzeszony” po restarcie, coś dla którego zrobić konieczność powoływania się na rozwiązaniach systemowych zależnych.

Możesz sprawdzić [1] Crons PiClouda [2], ale pamiętaj, że twoje zadania nie będą działały na twoim komputerze. Jest to również usługa, za którą musisz zapłacić, jeśli korzystasz z ponad 20 godzin obliczeniowych miesięcznie.

[1] http://www.picloud.com

[2] http://docs.picloud.com/cron.html

Metoda Crontab na serwerze.

Nazwa pliku Python hello.py

Krok 1: Utwórz plik sh, podając nazwę s.sh

python3 /home/ubuntu/Shaurya/Folder/hello.py> /home/ubuntu/Shaurya/Folder/log.txt 2> i 1

Krok 2: Otwórz Edytor Crontab

crontab -e

Krok 3: Dodaj harmonogram

Użyj formatowania Crontab

2 * * * * sudo sh /home/ubuntu/Shaurya/Folder/s.sh

Ten cron będzie działał „W minucie 2.”

Podoba mi się, jak pakiet Pycron rozwiązuje ten problem.

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)