一分钟后无法停止 tweepy 的流媒体传输

Question

我正在尝试使用Stream.filter()方法，流式传输Twitter数据，持续时间为5分钟。我将获取的推文存储在JSON文件中。问题是，我无法在程序内停止filter()方法的执行。我需要手动停止执行。我尝试使用time包基于系统时间停止数据传输。我能够停止写入推文到JSON文件，但stream方法仍在进行中，它无法继续执行下一行代码。我使用IPython笔记本编写和执行代码。以下是代码：

auth = OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_secret)
api = tweepy.API(auth)

from tweepy import Stream
from tweepy.streaming import StreamListener

class MyListener(StreamListener):

    def __init__(self, start_time, time_limit=60):
        self.time = start_time
        self.limit = time_limit

    def on_data(self, data):
        while (time.time() - self.time) < self.limit:
            try:
                saveFile = open('abcd.json', 'a')
                saveFile.write(data)
                saveFile.write('\n')
                saveFile.close()
                return True
            except BaseException as e:
                print 'failed ondata,', str(e)
                time.sleep(5)
        return True

    def on_status(self, status):
        if (time.time() - self.time) >= self.limit:
            print 'time is over'
            return false

    def on_error(self, status):
        if (time.time() - self.time) >= self.limit:
            print 'time is over'
            return false
        else:
            print(status)
            return True

start_time = time.time()
stream_data = Stream(auth, MyListener(start_time,20))
stream_data.filter(track=['name1','name2',...list ...,'name n'])#list of the strings I want to track

这些链接类似，但都没有直接回答我的问题。

- Tweepy: 如何在 X 分钟内获取流数据？ - 如何在设定的持续时间（行数、秒数、推文数等）后停止 Tweepy 流？ - Tweepy Streaming - 如何在达到一定数量时停止收集推文？ 我参考了这个链接：http://stats.seandolinar.com/collecting-twitter-data-using-a-python-stream-listener/

- Abin

---

你有一个有效的问题，不需要担心。你所说的“问题在于我无法在程序内停止filter()方法”，是什么意思？你是想暂停流吗？还是更改筛选关键词？ - Leb

@Leb 我希望流可以在我需要它运行时运行，比如每小时运行一次。如果我让它永远运行，它将会达到API限制并停止工作。我想在代码中以编程方式打开和关闭流。 - Abin

Answer 1

1. 为了关闭流，你需要从 on_data() 或者 on_status() 返回 False。

2. 因为 tweepy.Stream() 本身已经运行一个 while 循环，所以在 on_data() 中不需要再加入 while 循环。

3. 在初始化 MyListener 时，你没有调用父类的 __init__ 方法，所以它没有被正确地初始化。

所以，针对你想要做的事情，代码应该像这样：

class MyStreamListener(tweepy.StreamListener):
    def __init__(self, time_limit=60):
        self.start_time = time.time()
        self.limit = time_limit
        self.saveFile = open('abcd.json', 'a')
        super(MyStreamListener, self).__init__()

    def on_data(self, data):
        if (time.time() - self.start_time) < self.limit:
            self.saveFile.write(data)
            self.saveFile.write('\n')
            return True
        else:
            self.saveFile.close()
            return False

myStream = tweepy.Stream(auth=api.auth, listener=MyStreamListener(time_limit=20))
myStream.filter(track=['test'])

- yprez

Answer 2

访问变量 myListener.running，但不要直接将 MyListener 传递给 Stream，而应该创建一个变量，如下所示：

myListener = MyListener()
timeout code here... suchas time.sleep(20)
myListener.running = False 

- John

Answer 3

所以，我也遇到了这个问题。幸运的是，Tweepy是开源的，所以很容易深入研究问题。

基本上，重要的部分在这里：

def _data(self, data):
    if self.listener.on_data(data) is False:
        self.running = False

在 streaming.py 文件的 Stream 类中，

这意味着，要关闭连接，只需在监听器的 on_data() 方法中返回 false。

- Eduardo Rocha

Answer 4

对于那些正在尝试使用 Twitter API V2（StreamingClient 类）的人，这里是解决方案：

client.disconnect()

- no-stale-reads