tweets_producer.py 文件源码

def fetch_twitter_status(producer, symbols):
    """
    Retrieve English tweet associated with stock symbols (e.g. SNAP, AAPL, GOOG, etc)
    :param producer: Kafka producer
    :param symbols: stock symbol list
    :return: None
    """

    try:
        global stream
        if stream is not None and stream.running is True:
            logger.debug("Tweet streamming is running")
            stream.disconnect()
            del stream

        logger.info("Fetching tweets")
        stream_listener = Listener(api, producer, tweet_topic, symbols)
        stream = tweepy.Stream(auth=api.auth, listener=stream_listener)
        stream.filter(track=symbols, async=True, languages=["en"])
    except TweepError as te:
        logger.debug("TweepyExeption: Failed to get tweet for stocks caused by: %s" % te.message)
    except Exception as e:
        logger.warn("Eception: Failed to get tweet for stocks caused by: %s" % e.message)