Estoy tratando de tener un servidor Flask que me permita iniciar una transmisión tweepy y en cada mensaje recibido en el oyente de transmisión, envía ese mensaje a un cliente socketio. Al mismo tiempo, se supone que el servidor Flask permite que Twilio publique en él y enrute ese mensaje al cliente, de modo que el cliente reciba mensajes tanto de Twilio como de Twitter.
He estado tratando de hacer que el servidor envíe mensajes al cliente para los datos entrantes de Twitter, el código para Twilio funciona bien. Envía datos al cliente al recibir el mensaje. El bucle principal en tweepy tampoco bloquea el programa: puedo probar declaraciones de impresión y ver tweets y los sms entrantes que se imprimen en la handle_message(msg) de forma asíncrona. Siento que debe haber algo realmente simple que me estoy perdiendo aquí, ya que los SMS se emiten al cliente, pero los tweets entrantes no, a pesar de que se propagan a través de la handle_message(msg) . ¿Lo que da?
servidor.py
from flask import Flask, json, request from twilio.twiml.messaging_response import Message, MessagingResponse from flask_socketio import SocketIO import tweepy import json PATH = '/path/to/credentials/' with open(PATH, "r") as file: credentials = json.load(file) app = Flask(__name__) app.debug = True app.config['SECRET_KEY'] = 'abc123' sio = SocketIO(app, cors_allowed_origins="*") auth = tweepy.OAuthHandler(credentials['CONSUMER_KEY'], credentials['CONSUMER_SECRET']) auth.set_access_token(credentials['ACCESS_TOKEN'], credentials['ACCESS_SECRET']) api = tweepy.API(auth) class MyListener(tweepy.StreamListener): def on_status(self, status): print('status') def on_data(self, data): handle_message(data) def on_error(self, status): print('error') print(status) stream_listener = MyListener() # twilio sms route @app.route('/sms', methods=['POST']) def sms(): number = request.form['From'] message_body = request.form['Body'] message_data = {"number": number, "msg": message_body} resp = MessagingResponse() resp.message('Hello {}, you said: {}'.format(number, message_body)) handle_message(message_data) return str(resp) # flask-socketio stuff @sio.on('connect') def connect(): print('connected') sio.emit('client_connected', "you connected") search_term = "#mysearchterm" stream = tweepy.Stream(auth=api.auth, listener=stream_listener) stream.filter(track=[search_term], is_async=True) sio.emit('client_connected', "the search term is {}".format(search_term)) @sio.on('disconnect') def disconnect(): print('Client Diconnected') @sio.event def handle_message(message): print("This is the message received: ", message) sio.emit('handle_message', message) if __name__ == '__main__': sio.run(app)cliente.py
import socketio client = socketio.Client() @client.on('client_connected') def on_connect(message): print(message) @client.on('handle_message') def message(data): print(data) client.connect('http://localhost:5000/')Desarrollador evangelista de Twilio aquí.
Ha decorado la función handle_message como @sio.event pero , por lo que puedo ver en los documentos , solo debe hacer eso para que el método handle_message responda a los eventos en el socket llamado "handle_message".
Comenzaría eliminando el decorador @sio.event .
No soy un experto en Python, pero también me pregunto si hay un problema de alcance aquí. Usted define su clase MyListener y crea una instancia de ella antes de definir el método handle_message . Solo para probar, ¿puede intentar emitir al socket directamente dentro del método on_data ?
def on_data(self, data): sio.emit('handle_message', data) Si eso funciona, considere mover la definición de handle_message arriba de la definición de MyListener .
¡Resolví mi problema! Como señalé en este comentario , el problema era con los subprocesos múltiples y el paso de información entre los subprocesos. Con tweepy, el parámetro is_async=True , que en 4.1.0 es threading=True , abre un nuevo hilo una vez que se ejecuta la transmisión.
En lugar de tratar de pasar información, aproveché la funcionalidad de matraz-socketio existente usando un servidor redis local como una cola de mensajes (comience desde la sección "Uso de varios trabajadores" si está configurando esto por primera vez, también asegúrese de instalar redis ).
Aquí está el código server.py actualizado. El código client.py se mantuvo esencialmente sin cambios:
import eventlet eventlet.monkey_patch() from flask import Flask, json, request from twilio.twiml.messaging_response import Message, MessagingResponse from flask_socketio import SocketIO import tweepy import json PATH = '/PATH/TO/CREDENTIALS' with open(PATH, "r") as file: credentials = json.load(file) app = Flask(__name__) app.debug = True app.config['SECRET_KEY'] = 'abc123' sio = SocketIO(app, message_queue='redis://', cors_allowed_origins="*") class MyStream(tweepy.Stream): def __init__(self, consumer_key, consumer_secret, access_token, access_secret): super(MyStream, self).__init__(consumer_key, consumer_secret, access_token, access_secret) self.stream_sio = SocketIO(message_queue='redis://') def on_status(self, status): print('status') def on_data(self, data): json_data = json.loads(data) self.stream_sio.emit('handle_message', json_data['text']) # TODO: Send along all necessary information @app.route('/sms', methods=['POST']) def sms(): number = request.form['From'] message_body = request.form['Body'] message_data = {"number": number, "msg": message_body} resp = MessagingResponse() resp.message('Hello {}, you said: {}'.format(number, message_body)) handle_message(message_data) return str(resp) @sio.on('connect') def connect(): print('connected') sio.emit('client_connected', "you connected") search_term = "#testingtesting123" stream = MyStream(credentials['CONSUMER_KEY'], credentials['CONSUMER_SECRET'], credentials['ACCESS_TOKEN'], credentials['ACCESS_SECRET']) stream.filter(track=[search_term], threaded=True) sio.emit('client_connected', "the search term is {}".format(search_term)) @sio.on('disconnect') def disconnect(): print('Client disconnected') def handle_message(message): sio.emit('handle_message', message) if __name__ == '__main__': sio.run(app)