파이썬에서 크론 같은 스케줄러를 어떻게 구하지?
나는 파이톤에서 제공할 도서관을 찾고 있다.at그리고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.
크론 시간표현 구문은 덜 중요하지만 이런 융통성이 있는 구문이 있었으면 좋겠다.
만약 나에게 이 일을 즉시 처리할 수 있는 무언가가 없다면, 이런 것을 만들기 위한 빌딩 블록에 대한 어떠한 제안도 정중하게 받아들여질 것이다.
편집 프로세스 시작에는 관심이 없으며 Python - python 함수로도 작성된 "작업"만 사용하십시오.필연적으로 나는 이것이 다른 실이 될 것이라고 생각하지만, 다른 과정에서는 그렇지 않다.
이를 위해 크론 시간표현의 표현력을 찾고 있지만, 파이톤에서는.
크롱은 몇 년 전부터 있었지만, 나는 가능한 한 휴대하기 쉽게 하려고 노력하고 있다.나는 그것의 존재에 의지할 수 없다.
가벼운 체크아웃 일정을 찾고 있는 경우:
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)
공개:나는 그 도서관의 작가다.
당신은 당신의 크론탑을 지정하기 위해 구문을 통과하는 보통의 파이톤 인수를 사용할 수 있다.예를 들어 이벤트 클래스를 다음과 같이 정의한다고 가정합시다.
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의 주장 역학(위치 및 키워드 아그(arg)를 믹싱하고, 몇 주와 몇 달의 이름에 기호 이름을 사용할 수 있음)의 전권을 얻을 수 있다.
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)
while datetime.now() < t:
time.sleep((t - datetime.now()).seconds)
몇 가지 유의할 사항:Python의 평일/개월은 (크론과는 달리) 제로 인덱싱되며, 그 범위는 마지막 요소를 제외하므로 "1-5"와 같은 구문은 범위(0,5)가 된다. 즉 [0,1,2,3,4]하지만 크론 구문을 선호한다면 구문 구문을 분석하는 것은 그리 어렵지 않을 것이다.
위와 거의 동일하지만 동시에 지번트를 사용하는 경우 :)
"""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()
나열된 솔루션 중 어떤 것도 복잡한 크론 스케줄 문자열을 구문 분석하려고 시도하지 않는다.자, 여기 크로니터를 사용하는 내 버전이 있다.기본 요지:
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)
이 방법에는 "순수 비단뱀" 방법이 없다. 왜냐하면 어떤 다른 과정에서는 당신의 솔루션을 실행하기 위해 비단뱀을 실행해야 하기 때문이다.모든 플랫폼에는 프로세스를 시작하고 진행 상황을 모니터링할 수 있는 한 가지 또는 스무 가지 다른 방법이 있을 것이다.유닉스 플랫폼에서 크론은 오래된 표준이다.맥 OS X에는 크론 같은 런칭과 감시 기능을 결합한 launchd도 있다. 이 기능은 만약 당신이 원한다면 당신의 공정을 지속시킬 수 있다.일단 python이 실행되면 스케줄링 모듈을 사용하여 작업을 예약할 수 있다.
많은 해답이 있다는 것을 알지만, 또 다른 해결책은 장식가들과 함께 하는 것일 수도 있다.특정 시간에 매일 기능을 반복하는 예다.이렇게 사용하는 것에 대한 멋진 생각은 당신이 예약하고자 하는 기능에 통사적 슈가만 추가하면 된다는 것이다.
@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
나는 파이크론 패키지가 이 문제를 해결하는 방식이 마음에 든다.
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
또 다른 사소한 해결책은 다음과 같다.
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.위치:
# 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
그런 것이 이미 존재하는지 모르겠다.시간, datetime 및/또는 달력 모듈을 사용하여 직접 작성하는 것이 쉬울 것이다. http://docs.python.org/library/time.html을 참조하십시오.
Python 솔루션의 유일한 우려 사항은 항상 작업을 실행해야 하고 재부팅 후 자동으로 "복구"되어야 한다는 것인데, 이는 시스템 의존적 솔루션에 의존해야 하는 것이다.
참조URL: https://stackoverflow.com/questions/373335/how-do-i-get-a-cron-like-scheduler-in-python
'Programing' 카테고리의 다른 글
| 이번의history.properties/"는 나를 홈페이지로 리디렉션하지 않는다. (0) | 2022.03.25 |
|---|---|
| nuxt 구성 파일의 저장소에 액세스 (0) | 2022.03.25 |
| Vue UI를 사용하는 동안 Vue 라우터를 구현하는 방법 (0) | 2022.03.25 |
| VueJS 라이프사이클 후크 내에서 데이터를 설정하는 방법 (0) | 2022.03.25 |
| JavaScript 코드가 유효한 TypeScript 코드인가? (0) | 2022.03.25 |