我正在使用 Flask 和 Tweepy 搜索实时推文。在前端，我有一个用户文本输入和一个名为“搜索”的按钮。理想情况下，当用户在输入中输入搜索词并单击“搜索”按钮时，Tweepy 应该监听新的搜索词并停止先前的搜索词流。单击“搜索”按钮时，它会执行此功能:

@app.route('/search', methods=['POST'])
# gets search-keyword and starts stream
def streamTweets():
    search_term = request.form['tweet']
    search_term_hashtag = '#' + search_term
    # instantiate listener
    listener = StdOutListener()
    # stream object uses listener we instantiated above to listen for data
    stream = tweepy.Stream(auth, listener)

    if stream is not None:
        print "Stream disconnected..."
        stream.disconnect()

    stream.filter(track=[search_term or search_term_hashtag], async=True)
    redirect('/stream') # execute '/stream' sse
    return render_template('index.html')

上述代码倒数第二行执行的/stream路由如下:

@app.route('/stream')
def stream():
    # we will use Pub/Sub process to send real-time tweets to client
    def event_stream():
        # instantiate pubsub
        pubsub = red.pubsub()
        # subscribe to tweet_stream channel
        pubsub.subscribe('tweet_stream')
        # initiate server-sent events on messages pushed to channel
        for message in pubsub.listen():
            yield 'data: %s\n\n' % message['data']
    return Response(stream_with_context(event_stream()), mimetype="text/event-stream")

我的代码运行良好，因为只要单击“搜索”按钮，它就会启动一个新的流并搜索给定的术语，但它不会停止之前的搜索。例如，如果我的第一个搜索词是“NYC”，然后我想搜索另一个词，比如“Los Angeles”，它会给我“NYC”和“Los Angeles”的结果，这不是我想要的想。我只想搜索“洛杉矶”。我该如何解决？换句话说，我如何停止上一个流？我查看了其他以前的线程，我知道我必须使用 stream.disconnect()，但我不确定如何在我的代码中实现它。任何帮助或输入将不胜感激。非常感谢!!

最佳答案

下面是一些在创建新流时会取消旧流的代码。它的工作原理是将新流添加到全局列表，然后在创建新流时对列表中的所有流调用 stream.disconnect()。

diff --git a/app.py b/app.py
index 1e3ed10..f416ddc 100755
--- a/app.py
+++ b/app.py
@@ -23,6 +23,8 @@ auth.set_access_token(access_token, access_token_secret)
 app = Flask(__name__)
 red = redis.StrictRedis()

+# Add a place to keep track of current streams
+streams = []

 @app.route('/')
 def index():
@@ -32,12 +34,18 @@ def index():
 @app.route('/search', methods=['POST'])
 # gets search-keyword and starts stream
 def streamTweets():
+        # cancel old streams
+        for stream in streams:
+            stream.disconnect()
+
        search_term = request.form['tweet']
        search_term_hashtag = '#' + search_term
        # instantiate listener
        listener = StdOutListener()
        # stream object uses listener we instantiated above to listen for data
        stream = tweepy.Stream(auth, listener)
+        # add this stream to the global list
+        streams.append(stream)
        stream.filter(track=[search_term or search_term_hashtag],
                async=True) # make sure stream is non-blocking
        redirect('/stream') # execute '/stream' sse

这没有解决的是 session 管理问题。在您当前的设置下，一个用户的搜索将影响所有用户的搜索。这可以通过为您的用户提供一些标识符并将他们的流与他们的标识符一起存储来避免。最简单的方法可能是使用 Flask 的 session支持。您也可以按照 Pierre 的建议使用 requestId 来执行此操作。在任何一种情况下，您还需要代码来通知用户何时关闭页面并关闭他们的流。

关于python - 使用 Tweepy 收听流媒体和搜索推文。如何停止以前的搜索并只收听新的流？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/27430754/