我希望安排每小时运行一次 python 脚本并将数据保存在 elasticsearch 索引中。所以我使用了我编写的函数 set_interval ,它使用 tweepy 库。但它不起作用,因为我需要它工作。它每分钟运行一次并将数据保存在index.html中。即使在秒数等于 3600 的设置之后,它也会每分钟运行一次。但我想将其配置为每小时运行。
我该如何解决这个问题?这是我的 python 脚本:
def call_at_interval(time, callback, args):
while True:
timer = Timer(time, callback, args=args)
timer.start()
timer.join()
def set_interval(time, callback, *args):
Thread(target=call_at_interval, args=(time, callback, args)).start()
def get_all_tweets(screen_name):
# authorize twitter, initialize tweepy
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_key, access_secret)
api = tweepy.API(auth)
screen_name = ""
# initialize a list to hold all the tweepy Tweets
alltweets = []
# make initial request for most recent tweets (200 is the maximum allowed count)
new_tweets = api.user_timeline(screen_name=screen_name, count=200)
# save most recent tweets
alltweets.extend(new_tweets)
# save the id of the oldest tweet less one
oldest = alltweets[-1].id - 1
# keep grabbing tweets until there are no tweets left to grab
while len(new_tweets) > 0:
#print
#"getting tweets before %s" % (oldest)
# all subsiquent requests use the max_id param to prevent duplicates
new_tweets = api.user_timeline(screen_name=screen_name, count=200, max_id=oldest)
# save most recent tweets
alltweets.extend(new_tweets)
# update the id of the oldest tweet less one
oldest = alltweets[-1].id - 1
#print
#"...%s tweets downloaded so far" % (len(alltweets))
outtweets = [{'ID': tweet.id_str, 'Text': tweet.text, 'Date': tweet.created_at, 'author': tweet.user.screen_name} for tweet in alltweets]
def save_es(outtweets, es): # Peps8 convention
data = [ # Please without s in data
{
"_index": "index name",
"_type": "type name",
"_id": index,
"_source": ID
}
for index, ID in enumerate(outtweets)
]
helpers.bulk(es, data)
save_es(outtweets, es)
print('Run at:')
print(datetime.now())
print("\n")
set_interval(3600, get_all_tweets(screen_name))
最佳答案
为什么每小时完成一些任务需要这么复杂?您可以按照下面的方式每隔一小时运行一次脚本,请注意,它运行了 1 小时 + 工作时间:
import time
def do_some_work():
print("Do some work")
time.sleep(1)
print("Some work is done!")
if __name__ == "__main__":
time.sleep(60) # imagine you would like to start work in 1 minute first time
while True:
do_some_work()
time.sleep(3600) # do work every one hour
如果您想每隔一小时运行一次脚本,请执行以下代码:
import time
import threading
def do_some_work():
print("Do some work")
time.sleep(4)
print("Some work is done!")
if __name__ == "__main__":
time.sleep(60) # imagine you would like to start work in 1 minute first time
while True:
thr = threading.Thread(target=do_some_work)
thr.start()
time.sleep(3600) # do work every one hour
在这种情况下,thr 应该以超过 3600 秒的速度完成工作,尽管事实并非如此,您仍然会得到结果,但结果将来自另一次尝试,请参阅下面的示例:
import time
import threading
class AttemptCount:
def __init__(self, attempt_number):
self.attempt_number = attempt_number
def do_some_work(_attempt_number):
print(f"Do some work {_attempt_number.attempt_number}")
time.sleep(4)
print(f"Some work is done! {_attempt_number.attempt_number}")
_attempt_number.attempt_number += 1
if __name__ == "__main__":
attempt_number = AttemptCount(1)
time.sleep(1) # imagine you would like to start work in 1 minute first time
while True:
thr = threading.Thread(target=do_some_work, args=(attempt_number, ),)
thr.start()
time.sleep(1) # do work every one hour
在这种情况下你会得到的结果是:
做一些工作1 做一些工作 1 做一些工作 1 做一些工作 1 一些工作已经完成! 1 做一些工作 2 一些工作已经完成! 2 做一些工作 3 一些工作已经完成! 3 做一些工作 4 一些工作已经完成! 4 做一些工作 5 一些工作已经完成! 5 做一些工作 6 一些工作已经完成! 6 做一些工作 7 一些工作已经完成! 7 做一些工作 8 一些工作已经完成! 8 做一些工作9
我喜欢使用 subprocess.Popen 来执行此类任务,如果子子进程由于任何原因没有在一小时内完成其工作,您只需终止它并启动一个新的子进程即可。
您还可以使用 CRON 安排某些进程每隔一小时运行一次。
关于python - 每小时运行一次 python 脚本,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/58335344/