twitter_streamer.py 文件源码

def run_twitter_fetcher():
    sentry = Client(SENTRY_DSN)
    while True:
        try:
            l.info('starting streamer with {} emojis...'.format(len(EMOJIS)))
            sentry.captureMessage('starting `emoji-prediction`')
            streamer = TwitterEmojiStreamer(TwitterAuth.CONSUMER_KEY,
                                TwitterAuth.CONSUMER_SECRET,
                                TwitterAuth.ACCESS_TOKEN,
                                TwitterAuth.ACCESS_TOKEN_SECRET)
            streamer.statuses.filter(track=EMOJIS, language=LANGUAGE)
        # requests.exceptions.ConnectionError
        except ChunkedEncodingError:
            l.debug('chunked_encoding_error happened')
            pass
        except ConnectionError:
            l.debug('connection_error happened')
            pass
        except UnknownTwitterEmojiException as e:
            l.error('unknown exception ocurred')
            l.error(e)
            sentry.captureException()