python - KeyError During Twitter Filtered Stream Using Tweepy

Question

I am currently running into a KeyError issue that I cannot figure out in the below code. The aim of the programme is to stream a particular users tweets using Tweepy and then send an email containing that tweet whenever those users post about a specified keyword. It has been working perfectly and has been receiving a number of tweets and sending the emails however every now and then it throw up this error: KeyError: 'user' relating to the line: 'Identity': data['user']['screen_name']

The error is listed below the code for reference. I'm sure I can use the .get() function somewhere or even use an except to try and handle the KeyError but I cannot figure out how to do this in my codes context?

import json
import yaml


from tweepy import API, OAuthHandler, Stream, StreamListener

from emailService import Email

consumer_key = ''
consumer_secret = ''
access_token = ''
access_token_secret = ''
recipient_email = ''

with open("config.yaml", 'r') as stream:
    config = yaml.safe_load(stream)


class StdOutListener(StreamListener):
    """ A listener handles tweets that are received from the stream.
    This is a basic listener that just prints received tweets to stdout.
    """
    def __init__(self, config):
        super().__init__()
        self.config = config

    def on_status(self, status):

        data = json.loads(status)


        try:
            extended = data['extended_tweet']

            tweetData = {
                'Identity': data['user']['screen_name'],
                'message': extended['full_text'],
                'urls': list(map(lambda url: url['expanded_url'], data['entities']['urls']))
            }

        except:

            tweetData = {
                'Identity': data['user']['screen_name'],
                'message': data['text'],
                'urls': list(map(lambda url: url['expanded_url'], data['entities']['urls']))
            }

        return tweetData

    def _sendEmail(self, tweetData):
        subject = f'Twitter Keyword Match: {tweetData["Identity"]}'
        message = self._composeEmailMessage(tweetData)

        Email().send(recipient_email, subject, message)


    def _composeEmailMessage(self, tweetData):
        message = f'''
New Twitter Keyword Match:

@{tweetData['Identity']}: {tweetData["message"]}

'''

        for url in tweetData['urls']:
            message += url + '
'

        return message

    def on_data(self, data):
        tweetData = self.on_status(data)
        print(f'@{tweetData["Identity"]} tweeted: {tweetData["message"]}
')

        if tweetData['Identity'] in self.config:
            matchedWords = [keyword for keyword in self.config[tweetData['Identity']] if keyword in tweetData['message']]

            if len(matchedWords):
                print(matchedWords, 'Keyword match found. Sending email.
')
                self._sendEmail(tweetData)

        return True

    def on_error(self, status):
        print('STREAM ERROR:', status)


class TwitterAlerts:
    def __init__(self, config):
        self.config = config
        self.auth = self._createAuth()
        self.api = API(self.auth)
        self.stream = self._createStream(config)

        self.generateInitMsg()

    def generateInitMsg(self):
        maxUsernameLen = len(max(list(self.config.keys()) + ['USERNAME'], key = len))
        print('Twitter keyword alert service initialized
')
        print('USERNAME', ' '*(maxUsernameLen - 8), ':', "[KEYWORDS]")
        for username, keywords in self.config.items():
            print(username, ' '*(maxUsernameLen - len(username)), ':', keywords)
        print('')

    def _createAuth(self):
        auth = OAuthHandler(consumer_key, consumer_secret)
        auth.set_access_token(access_token, access_token_secret)
        return auth

    def getUser(self, username):
        return self.api.get_user(username)

    def _createStream(self, config):
        listener = StdOutListener(config)
        return Stream(self.auth, listener)

    def startFilter(self):
        # convert usernames to ids
        userIds = list(map(lambda username: str(self.getUser(username).id), self.config.keys()))
        # create listener for all userids in config
        self.stream.filter(follow = userIds)


TwitterAlerts(config['usernames']).startFilter()

---

Error: Traceback (most recent call last): File "/Users/CraftyBadger/Desktop/PycharmProjects/TwitterUserKeywordAlertMonitor/index.py", line 33, in on_status extended = data['extended_tweet'] KeyError: 'extended_tweet'

During handling of the above exception, another exception occurred:

Traceback (most recent call last): File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/six.py", line 703, in reraise raise value File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/tweepy/streaming.py", line 289, in _run self._read_loop(resp) File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/tweepy/streaming.py", line 351, in _read_loop self._data(next_status_obj) File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/tweepy/streaming.py", line 323, in _data if self.listener.on_data(data) is False: File "/Users/CraftyBadger/Desktop/PycharmProjects/TwitterUserKeywordAlertMonitor/index.py", line 70, in on_data tweetData = self.on_status(data) File "/Users/CraftyBadger/Desktop/PycharmProjects/TwitterUserKeywordAlertMonitor/index.py", line 44, in on_status 'Identity': data['user']['screen_name'], KeyError: 'user'

Answer 1

等待大神解答