logo标签列表

Tweepy 流媒体 API：全文

twittertext-miningtweepy
6
我正在使用tweepy流API获取包含特定标签的推文。我遇到的问题是从流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)

有人有解决方案吗？
8个回答
8

tweet_mode=extended在此代码中无效，因为流媒体API不支持该参数。如果推文包含更长的文本，则JSON响应中将包含一个名为extended_tweet的附加对象，该对象将反过来包含一个名为full_text的字段。

在这种情况下，您需要使用类似于print(status.extended_tweet.full_text)的内容来提取更长的文本。

仍然无法工作。 出现错误：AttributeError: 'Status'对象没有'extended_tweet'属性。 - Varad Bhatnagar
这只会出现在超过140个字符的推文中。您尝试过跟踪从API返回的完整JSON对象，以检查确切的结构吗？ - Andy Piper
1
我使用 tweepy 的 on_data() 功能完成了我想要做的任务，该功能返回一个类似 JSON 的对象。 - Varad Bhatnagar
7

在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)

这个full_text字段包含了所有推文的文字，无论是截断还是完整。

太棒了。谢谢你。已经获取到 full_text 和 entities。 - statosdotcom
2

您需要这样启用扩展推文模式：

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

对我有用。

1

基于 @AndyPiper 的 答案，你可以通过 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”实际上无法正常工作并会引发错误。
1
除了之前的回答之外：在我的情况下，它只能作为status.extended_tweet['full_text']工作，因为status.extended_tweet只是一个字典。
0
我使用以下函数：
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。这是一个旧问题，已经有了被接受的答案。你可能想要解释一下你的答案与其他答案的不同之处，以及为什么有人应该使用它。 - Xerillio
0

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

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)

访问链接，它将向您展示如何实现扩展推文

0

这是对我有效的解决方案：

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 - 从这里实现，需要改变他们建议的内容，但是思路仍然相同

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文，
原文链接