使用 Tweepy 在 Twitter 过滤流期间出现 KeyError

问题描述

我目前遇到了一个 KeyError 问题，我无法在下面的代码中弄清楚。该程序的目的是使用 Tweepy 流式传输特定用户的推文，然后在这些用户发布有关指定关键字的信息时发送包含该推文的电子邮件。 它一直工作得很好，并且一直在接收一些推文并发送电子邮件，但是它时不时地抛出这个错误： KeyError: 'user' related to the line: 'Identity': data['user']['屏幕名称']

错误列在代码下方以供参考。我确定我可以在某处使用 .get() 函数，甚至可以使用 except 来尝试处理 KeyError 但我不知道如何在我的代码上下文中执行此操作？

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'],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"]}\n\n'''

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

        return message

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

        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.\n')
                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\n')
        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()

---

错误： 回溯（最近一次调用最后一次）： 文件“/Users/CraftyBadger/Desktop/PycharmProjects/TwitterUserKeywordAlertMonitor/index.py”，第 33 行，在 on_status 扩展 = 数据['extended_tweet'] KeyError: 'extended_tweet'

在处理上述异常的过程中，又发生了一个异常：

回溯（最近一次调用最后一次）： 文件“/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/six.py”，第 703 行，重新提出 提升价值 文件“/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/tweepy/streaming.py”，第 289 行，在 _run self._read_loop(resp) 文件“/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/tweepy/streaming.py”，第 351 行，在 _read_loop self._data(next_status_obj) 文件“/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/tweepy/streaming.py”，第 323 行，在 _data 如果 self.listener.on_data(data) 为 False： 文件“/Users/CraftyBadger/Desktop/PycharmProjects/TwitterUserKeywordAlertMonitor/index.py”，第 70 行，在 on_data tweetData = self.on_status(data) 文件“/Users/CraftyBadger/Desktop/PycharmProjects/TwitterUserKeywordAlertMonitor/index.py”，第 44 行，在 on_status '身份'：数据['用户']['screen_name']， 密钥错误：'用户'

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）