Como obtenho um Cron como agendador em Python? [fechadas]

Estou procurando uma biblioteca em Python que forneça ate crongere funcionalidade.

Eu gostaria de ter uma solução Python pura, em vez de confiar nas ferramentas instaladas na caixa; desta forma eu corro em máquinas sem cron.

Para aqueles que não estão familiarizados com cron: você pode agendar tarefas com base em uma expressão como:

 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.

A sintaxe da expressão cron time é menos importante, mas eu gostaria de ter algo com esse tipo de flexibilidade.

Se não houver algo que faça isso imediatamente, todas as sugestões para os blocos de construção para fazer algo assim serão recebidas com gratidão.

Editar Não estou interessado em iniciar processos, apenas "trabalhos" também escritos nas funções Python - python. Por necessidade, acho que esse seria um segmento diferente, mas não em um processo diferente.

Para esse fim, procuro a expressividade da expressão cron time, mas em Python.

Cron existe há anos, mas estou tentando ser o mais portátil possível. Não posso confiar na presença dele.

Se você estiver procurando por algo leve na agenda de checkout :

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)

Divulgação : sou o autor dessa biblioteca.

Você pode usar o argumento normal do Python passando a sintaxe para especificar seu crontab. Por exemplo, suponha que definimos uma classe Event como abaixo:

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)

(Nota: Não testado completamente)

Em seguida, seu CronTab pode ser especificado na sintaxe python normal como:

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

Dessa forma, você obtém todo o poder da mecânica de argumentos do Python (misturando argumentos posicionais e de palavras-chave e pode usar nomes simbólicos para nomes de semanas e meses)

A classe CronTab seria definida como simplesmente dormir em incrementos de minutos e chamar check () em cada evento. (Provavelmente existem algumas sutilezas com o horário de verão / fusos horários para se ter cuidado). Aqui está uma implementação rápida:

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)

Algumas coisas a serem observadas: os dias da semana / meses do Python são zero indexados (diferente do cron), e esse intervalo exclui o último elemento; portanto, sintaxe como "1-5" se torna intervalo (0,5) - ou seja, [0,1,2, 3,4] Se você preferir a sintaxe do cron, a análise não deve ser muito difícil.

talvez isso tenha surgido somente depois que a pergunta foi feita; Eu pensei em mencioná-lo por uma questão de integridade: https://apscheduler.readthedocs.org/en/latest/

Uma coisa que nas minhas pesquisas que vi é o schedmódulo python, que pode ser o tipo de coisa que você está procurando.

"... módulo Crontab para ler e gravar arquivos crontab e acessar o cron do sistema automaticamente e simplesmente usando uma API direta. ..."

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

e também o APScheduler, um pacote python. Já está escrito e depurado.

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

Mais ou menos o mesmo que acima, mas simultaneamente usando 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()

Nenhuma das soluções listadas sequer tenta analisar uma sequência complexa de cronograma cron. Então, aqui está a minha versão, usando croniter . Essência básica:

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

Rotinas auxiliares:

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)

Eu modifiquei o script.

1. Fácil de usar:

   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. Tente iniciar a tarefa no primeiro segundo de um minuto.

Código no Github

Eu tenho uma pequena correção para o método de execução de classe CronTab sugerido por Brian .

O tempo terminou em um segundo, levando a um ciclo difícil de um segundo no final de cada minuto.

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

Não existe uma maneira "python pura" de fazer isso, porque algum outro processo teria que iniciar o python para executar sua solução. Cada plataforma terá uma ou vinte maneiras diferentes de iniciar processos e monitorar seu progresso. Em plataformas unix, o cron é o padrão antigo. No Mac OS X, também existe o launchd, que combina o lançamento do tipo cron com a funcionalidade watchdog que pode manter seu processo ativo, se é isso que você deseja. Depois que o python estiver em execução, você poderá usar o módulo de agendamento para agendar tarefas.

Sei que há muitas respostas, mas outra solução poderia ser a de decoradores . Este é um exemplo para repetir uma função todos os dias em um horário específico. O melhor de usar dessa maneira é que você só precisa adicionar o Sugar Syntactic Sugar à função que deseja agendar:

@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

E o decorador terá a seguinte aparência:

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

A solução de Brian está funcionando muito bem. No entanto, como outros já apontaram, há um bug sutil no código de execução. Também achei excessivamente complicado para as necessidades.

Aqui está minha alternativa mais simples e funcional para o código de execução, caso alguém precise:

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)

Outra solução trivial seria:

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)

E a classe 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

Se você estiver procurando por um agendador distribuído, consulte https://github.com/sherinkurian/mani - ele precisa de redis, embora possa não ser o que você está procurando. (observe que eu sou o autor) isso foi criado para garantir a tolerância a falhas, com o relógio sendo executado em mais de um nó.

Não sei se algo assim já existe. Seria fácil escrever seu próprio com módulos de hora, data e hora e / ou calendário, consulte http://docs.python.org/library/time.html

A única preocupação para uma solução python é que as suas necessidades de trabalho para estar sempre em execução e, possivelmente, ser automaticamente "ressuscitado" após uma reinicialização, algo para o qual você fazer necessidade de contar com soluções dependentes do sistema.

Você pode conferir os [1] Crons do PiCloud [2], mas observe que seus trabalhos não serão executados em sua própria máquina. Também é um serviço que você precisará pagar se usar mais de 20 horas de computação por mês.

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

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

Método do Crontab no servidor.

Nome do arquivo Python hello.py

Etapa 1: Crie um arquivo sh, deixe o nome s.sh

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

Etapa 2: Abrir o Crontab Editor

crontab -e

Etapa 3: adicionar horário da programação

Use a formatação Crontab

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

Este cron será executado "No minuto 2."

Eu gosto de como o pacote pycron resolve esse problema.

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)