read - twitter bot python github

Usando Tweepy para escuchar la transmisión y buscar tweets. ¿Cómo detener la búsqueda anterior y solo escuchar nuevas transmisiones?

Question

Estoy usando Flask y Tweepy para buscar tweets en vivo. En la parte delantera tengo una entrada de texto de usuario, y un botón llamado "Buscar". Idealmente, cuando un usuario introduce un término de búsqueda en la entrada y hace clic en el botón "Buscar", Tweepy debería escuchar el nuevo término de búsqueda y detener la secuencia de términos de búsqueda anterior. Cuando se hace clic en el botón "Buscar" ejecuta esta función:

@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'')

La ruta /stream que se ejecuta en la segunda a la última línea del código anterior es la siguiente:

@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")

Mi código funciona bien, en el sentido de que comienza una nueva secuencia y busca un término determinado cada vez que se hace clic en el botón "Buscar", pero no detiene la búsqueda anterior. Por ejemplo, si mi primer término de búsqueda fue "NYC" y luego quería buscar un término diferente, diga "Los Angeles", me dará resultados tanto para "NYC" como para "Los Angeles", que no es lo que yo buscaba. querer. Sólo quiero que busquen "Los Ángeles". ¿Cómo puedo solucionar esto? En otras palabras, ¿cómo detengo la transmisión anterior? Miré a través de otros subprocesos anteriores, y sé que tengo que usar stream.disconnect() , pero no estoy seguro de cómo implementar esto en mi código. Cualquier ayuda o entrada sería muy apreciada. ¡¡Muchas gracias!!

Answer 1

A continuación se muestra un código que cancelará las transmisiones antiguas cuando se cree una nueva. Funciona al agregar nuevas secuencias a una lista global, y luego llamar a stream.disconnect() en todas las secuencias en la lista cada vez que se crea una nueva secuencia.

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

Lo que esto no resuelve es el problema de la gestión de sesión. Con su configuración actual, una búsqueda de un usuario afectará las búsquedas de todos los usuarios. Esto puede evitarse dando a sus usuarios algún identificador y almacenando sus flujos junto con su identificador. La forma más fácil de hacer esto es usar el soporte de session de Flask. También puede hacer esto con un requestId como Pierre sugirió. En cualquier caso, también necesitará un código para darse cuenta cuando un usuario haya cerrado la página y cierre su transmisión.

Answer 2

Bueno, lo resolví usando el método del temporizador. Pero todavía estoy buscando una forma pitónica.

from streamer import StreamListener
def stream():
    hashtag = input
    #assign each user an ID ( for pubsub )
    StreamListener.userid = random_user_id
    def handler(signum, frame):
        print("Forever is over")
        raise Exception("end of time")

    def main_stream():
        stream = tweepy.Stream(auth, StreamListener())
        stream.filter(track=track,async=True)
        redirect(url_for(''map_stream''))

    def close_stream():
        # this is for closing client list in redis but don''t know it''s working
        obj = redis.client_list(tweet_stream)
        redis_client_list = obj[0][''addr'']
        redis.client_kill(redis_client_list)
        stream = tweepy.Stream(auth, StreamListener())
        stream.disconnect()

    import signal
    signal.signal(signal.SIGALRM, handler)
    signal.alarm(300)
    try:
        main_stream()
    except Exception:
        close_stream()
        print("function terminate")

Answer 3

Descargo de responsabilidad: no sé nada acerca de Tweepy, pero esto parece ser un problema de diseño.

¿Estás tratando de agregar estado a una API RESTful? Es posible que tenga un problema de diseño. Como respondió JRichardSnape, su API no debería ser la encargada de cancelar una solicitud; Se debe hacer en la parte delantera. Lo que quiero decir aquí es en el javascript / AJAX / etc que llama a esta función, agrega otra llamada a la nueva función

@app.route(''/cancelSearch'', methods=[''POST'']) Con el "POST" que tiene los términos de búsqueda. Mientras no tengas un estado, no puedes hacer esto de manera segura en una llamada asincrónica: imagina que alguien más realiza la misma búsqueda al mismo tiempo y luego cancela una cancelará ambas (recuerda, no tienes un estado así que no sabes a quien estas cancelando). Tal vez usted necesita estado con su diseño.

Si debe seguir usando esto y no le importa romper la regla "sin estado", agregue un "estado" a su solicitud. En este caso, no es tan malo porque podría iniciar un hilo y nombrarlo con el ID de usuario, y luego eliminar el hilo en cada nueva búsqueda

def streamTweets():
    search_term = request.form[''tweet'']
    userId = request.form[''userId''] # If your limit is one request per user at a time. If multiple windows can be opened and you want to follow this limit, store userId in a cookie.
    #Look for any request currently running with this ID, and cancel them

Alternativamente, puede devolver un requestId , que luego mantendría en el front-end puede llamar cancelSearch?requestId=$requestId . En cancelSearch, deberías encontrar la solicitud pendiente (parece que está en Tweepy ya que no estás usando tus propios hilos) y desconectarlo.

Por curiosidad acabo de ver lo que sucede cuando buscas en Google y usa una solicitud GET. Eche un vistazo (herramientas de depuración -> Red; luego ingrese un texto y vea el relleno automático). Google usa un token enviado con cada solicitud (cada vez que escribe algo)). No significa que se use para esto, pero eso es básicamente lo que describí. Si no desea una sesión, utilice un identificador único .