如何在Python中使用1小時後退出函數

Question

我正在製作一個簡單的應用程序來流式傳輸twitter的公共時間表，並且我希望流式傳輸在一小時後自動停止，並且我不知道如何執行此操作。我閱讀日期時間和timeit文檔，但無法理解它們。這裏是我的代碼，它正在流暢地完成我想要的時間線，但無限期地。

from twython import TwythonStreamer 
import json 
import os 
import datetime 
from datetime import * 
APP_KEY = 'XX' 
APP_SECRET = 'XX' 
OAUTH_TOKEN = 'XX' 
OAUTH_TOKEN_SECRET = 'XX' 
class MyStreamer(TwythonStreamer): 

    def on_success(self, data): 
     print data['text'] 
     with open('scratch1.json', 'ab') as outfile: 
      json.dump(data, outfile, indent = 4) 
     with open('scratch2.json', 'ab') as xoutfile: 
      json.dump(data, xoutfile, indent = 4) 
     return 


    def on_error(self, status_code, data): 
     print status_code 
     return True # Don't kill the stream 

    def on_timeout(self): 
     print >> sys.stderr, 'Timeout...' 
     return True # Don't kill the stream 

stream = MyStreamer(APP_KEY, APP_SECRET, 
        OAUTH_TOKEN, OAUTH_TOKEN_SECRET) 
stream.statuses.filter(follow = [95995660, 8820362]) 

任何人都可以幫我嗎？

Answer 1

使用datetime.datetime.now()方法獲取當前日期時間對象，然後使用timedelta類爲其添加一小時。

import datetime 
stop_time = datetime.datetime.now() + datetime.timedelta(hours=1) 

# ... 

# in relevant function ... 
if datetime.datetime.now() > stop_time: 
    stop_streaming() 

---

我不熟悉你TwythonStreamer類，但可能是這樣的：

class MyStreamer(TwythonStreamer): 

    # the init function is called when you create instance of class 
    def __init__(self): 
     self.stop_time = datetime.datetime.now() + datetime.timedelta(hours=1) 

    # ... 

    def on_success(self, data): 
     if datetime.datetime.now() > self.stop_time: 
      raise Exception("Time expired") 

     # ... 

Answer 2

因爲我想提出一個大多是完整的解決方案。這裏是我的版本：

#!/usr/bin/env python 


import sys 
import json 
import datetime 


from twython import TwythonStreamer 


from circuits import Component, Event, Debugger, Timer 


APP_KEY = 'XX' 
APP_SECRET = 'XX' 
OAUTH_TOKEN = 'XX' 
OAUTH_TOKEN_SECRET = 'XX' 


class MyStreamer(TwythonStreamer): 

    def on_success(self, data): 
     print data['text'] 
     with open('scratch1.json', 'ab') as outfile: 
      json.dump(data, outfile, indent=4) 
     with open('scratch2.json', 'ab') as xoutfile: 
      json.dump(data, xoutfile, indent=4) 
     return 

    def on_error(self, status_code, data): 
     print status_code 
     return True # Don't kill the stream 

    def on_timeout(self): 
     print >> sys.stderr, 'Timeout...' 
     return True # Don't kill the stream 


class Check(Event): 
    """Check Event""" 


class Terminate(Event): 
    """Terminate Event""" 


class App(Component): 

    def init(self, *args, **kwargs): 
     self.stream = MyStreamer(
      APP_KEY, APP_SECRET, 
      OAUTH_TOKEN, OAUTH_TOKEN_SECRET 
     ) 

     interval = datetime.datetime.now() + datetime.timedelta(hours=1) 
     Timer(interval, Terminate(), self.channel).register(self) 

     Timer(10, Check(), persist=True, channel=self.channel).register(self) 

    def terminate(self): 
     raise SystemExit(0) 

    def check(self): 
     self.stream.statuses.filter(follow=[95995660, 8820362]) 


app = App() 
Debugger().register(app) 
app.run() 

這添加一個額外的庫/框架，以您的解決方案稱爲circuits和它的強大三項賽和內置Timer組件。

注：我還沒有測試過這個。我將留給你，因爲我沒有特別使用0123tTwtiter客戶端庫，也沒有使用它的任何API。祝你好運！

另外請注意，我假設你不想連續檢查流，因爲Twitter API可能有某種限制。因此，第二個計時器將每10秒觸發一次Check事件並堅持運行應用程序的整個生命週期。

Answer 3

我建議使用datetime.datetime.now（）模塊
datetime.datetime.now（）+ datetime.deltatime（秒= 3600）作爲您的一小時停機時間。

Answer 4

我不能用Apoorv的代碼來複制pztrick的修改。寫作：

class MyStreamer(TwythonStreamer): 
    def __init__(self): 
     self.stop_time = dt.datetime.now() + dt.timedelta(minutes=1) 

會產生此錯誤消息：

​​

下不工作：

類MyStreamer（twy.TwythonStreamer）：

def __init__(self): 
    self.stop_time = dt.datetime.now() + dt.timedelta(minutes=1) 
    self.app_key = APP_KEY 
    self.app_secret = APP_SECRET 
    self.oauth_token = OAUTH_TOKEN 
    self.oauth_token_secret = OAUTH_TOKEN_SECRET 

做什麼工作但是，只需定義stop_time而不需要init。我最終的解決方案是這樣的：

class MyStreamer(twy.TwythonStreamer): 

    stop_time = dt.datetime.now() + dt.timedelta(minutes=1) 

    def on_success(self, data): 
     if dt.datetime.now() > self.stop_time: 
      raise Exception('Time expired') 

     fileName = self.fileDirectory + 'Tweets_' + dt.datetime.now().strftime("%Y_%m_%d_%H") + '.txt' # File name includes date out to hour. 
     open(fileName, 'a').write(json.dumps(data) + '\n') 

我新的類，所以不明白爲什麼這樣的作品，但我很高興，它的作用。