Как получить функцию выхода через 1 час в Python

Question

Я делаю простое приложение, чтобы транслировать публичную временную шкалу твиттера, и я хочу, чтобы потоковая передача автоматически прекращалась через час, и я понятия не имею, как это сделать. Я читал datetime и timeit docs, но не могу их понять. Вот мой код, и это потоковое время, которое я хочу прекрасно, но бесконечно.

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() для получения текущего объекта datetime, затем используйте класс 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 компонентов.

NB: Я не проверял это. Я оставлю это до вас, поскольку я особо не использовал библиотеку клиентов Twtiter и ни один из ее API (ов). Удачи!

Также обратите внимание, что я предполагаю, что вы не хотите постоянно проверять поток, поскольку API Twitter (ы), вероятно, имеет какие-то ограничения для trhottling. Таким образом, второй таймер будет запускать событие Check каждые 10 секунд и будет сохраняться в течение всего срока действия запущенного приложения.

Answer 3

я рекомендую datetime.datetime.now() модуля
datetime.datetime.now() + datetime.deltatime (секунды = 3600), как ваш один час времени stopage.

Answer 4

Я не мог использовать код Apoorv для репликации с модификацией pztrick. Запись:

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

будет генерировать сообщение об ошибке:

TypeError: __init__() takes 1 positional argument but 5 were given 

Следующая не работал:

класса 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') 

Я новичок в занятиях, поэтому не понимаю, почему это работает, но я рад, что это так.