tweepy Streaming API：全文

Posted 2023-02-19

【中文标题】tweepy Streaming API：全文

【英文标题】：tweepy Streaming API : full text

【发布时间】：2018-06-27 09:42:27

【问题描述】：

我正在使用 tweepy 流 API 来获取包含特定主题标签的推文。我面临的问题是我无法从 Streaming API 中提取推文的全文。只有 140 个字符可用，之后会被截断。

代码如下：

auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
api = tweepy.API(auth)


def analyze_status(text):

    if 'RT' in text[0:3]:
        return True
    else:
        return False

    class MyStreamListener(tweepy.StreamListener):

    def on_status(self, status):

    if not analyze_status(status.text):

        with open('fetched_tweets.txt', 'a') as tf:
            tf.write(status.text.encode('utf-8') + '\n\n')

        print(status.text)

    def on_error(self, status):
    print("Error Code : " + status)

    def test_rate_limit(api, wait=True, buffer=.1):
        """
        Tests whether the rate limit of the last request has been reached.
        :param api: The `tweepy` api instance.
        :param wait: A flag indicating whether to wait for the rate limit reset
                 if the rate limit has been reached.
        :param buffer: A buffer time in seconds that is added on to the waiting
                   time as an extra safety margin.
        :return: True if it is ok to proceed with the next request. False otherwise.
        """
        # Get the number of remaining requests
        remaining = int(api.last_response.getheader('x-rate-limit-remaining'))
        # Check if we have reached the limit
        if remaining == 0:
        limit = int(api.last_response.getheader('x-rate-limit-limit'))
        reset = int(api.last_response.getheader('x-rate-limit-reset'))
        # Parse the UTC time
        reset = datetime.fromtimestamp(reset)
        # Let the user know we have reached the rate limit
        print "0 of  requests remaining until .".format(limit, reset)

        if wait:
            # Determine the delay and sleep
            delay = (reset - datetime.now()).total_seconds() + buffer
            print "Sleeping for s...".format(delay)
            sleep(delay)
            # We have waited for the rate limit reset. OK to proceed.
            return True
        else:
            # We have reached the rate limit. The user needs to handle the rate limit manually.
            return False

        # We have not reached the rate limit
        return True

    myStreamListener = MyStreamListener()
    myStream = tweepy.Stream(auth=api.auth, listener=myStreamListener,
                             tweet_mode='extended')

    myStream.filter(track=['#bitcoin'], async=True)

有人有解决办法吗？

【参考方案1】：

tweet_mode=extended 在此代码中无效，因为 Streaming API 不支持该参数。如果推文包含较长的文本，它将在 JSON 响应中包含一个名为 extended_tweet 的附加对象，该对象又将包含一个名为 full_text 的字段。

在这种情况下，您需要像 print(status.extended_tweet.full_text) 这样的东西来提取较长的文本。

【讨论】：

还是不行。出现错误：AttributeError：'Status' 对象没有属性'extended_tweet'

只有超过 140 个字符的推文才会出现。您是否尝试过追踪从 API 返回的完整 JSON 对象，以检查确切的结构？

我使用 tweepy 的 on_data() 功能完成了我想做的任务，它返回了一个类似 JSON 的对象。

【参考方案2】：

Twitter 流中有可用的布尔值。当消息包含超过 140 个字符时，'status.truncated' 为 True。只有这样，“extended_tweet”对象才可用：

        if not status.truncated:
            text = status.text
        else:
            text = status.extended_tweet['full_text']

这仅在您流式传输推文时有效。当您使用 API 方法收集较旧的推文时，您可以使用以下方法：

tweets = api.user_timeline(screen_name='whoever', count=5, tweet_mode='extended')
for tweet in tweets:
    print(tweet.full_text)

此全文字段包含所有推文的文本，无论是否截断。

【讨论】：

太棒了。感谢那。得到全文和实体。

【参考方案3】：

您必须像这样启用扩展推文模式：

s = tweepy.Stream(auth, l, tweet_mode='extended')

然后您可以打印扩展推文，但请记住，由于 Twitter API，您必须确保扩展推文存在，否则会引发错误

l = listener()

class listener(StreamListener):
    def on_status(self, status):
        try:
            print(status.extended_tweet['full_text'])
        except Exception as e:
            raise
        else:
            print(status.text)
        return True
    def on_error(self, status_code):
        if status_code == 420:
            return False

为我工作。

【参考方案4】：

基于@AndyPiper 的answer，您可以通过try/except 来检查推文是否存在：

  def get_tweet_text(tweet):
    try:
      return tweet.extended_tweet['full_text']
    except AttributeError as e:
      return tweet.text

或检查内部 json：

  def get_tweet_text(tweet):
    if 'extended_tweet' in tweet._json:
      return tweet.extended_tweet['full_text']
    else:
      return tweet.text

请注意，extended_tweet 是一个字典对象，因此“tweet.extended_tweet.full_text”实际上不起作用，会引发错误。

【参考方案5】：

除了上一个答案：在我的情况下，它只能用作 status.extended_tweet['full_text']，因为 status.extended_tweet 只不过是一本字典。

【参考方案6】：

这对我有用：

status = tweet if 'extended_tweet' in status._json: status_json = status._json['extended_tweet']['full_text'] elif 'retweeted_status' in status._json and 'extended_tweet' in status._json['retweeted_status']: status_json = status._json['retweeted_status']['extended_tweet']['full_text'] elif 'retweeted_status' in status._json: status_json = status._json['retweeted_status']['full_text'] else: status_json = status._json['full_text'] print(status_json)'

https://github.com/tweepy/tweepy/issues/935 - 从这里实现，需要改变他们的建议，但想法保持不变

【参考方案7】：

我使用以下功能：

def full_text_tweeet(id_):
    status = api.get_status(id_, tweet_mode="extended")
    try:
        return status.retweeted_status.full_text
    except AttributeError:  
        return status.full_text

然后在我的列表中调用它

 tweets_list = []
    # foreach through all tweets pulled
    for tweet in tweets:
        # printing the text stored inside the tweet object
        tweet_list = [str(tweet.id),str(full_text_tweeet(tweet.id))]
        tweets_list.append(tweet_list)

【讨论】：

欢迎来到 SO。这是一个已经被接受的答案的老问题。您可能想多解释一下您的答案与其他答案有何不同以及为什么有人应该使用它。

【参考方案8】：

试试这个，这是最简单最快的方法。

def on_status(self, status):
if hasattr(status, "retweeted_status"):  # Check if Retweet
    try:
        print(status.retweeted_status.extended_tweet["full_text"])
    except AttributeError:
        print(status.retweeted_status.text)
else:
    try:
        print(status.extended_tweet["full_text"])
    except AttributeError:
        print(status.text)

Visit the link it will give you the how extended tweet can be achieve