Skip to content

Twitter

TwitterTweetReader #

Bases: BasePydanticReader

Twitter推文阅读器。

读取用户Twitter账号的推文。

查看'https://developer.twitter.com/en/docs/twitter-api/ getting-started/getting-access-to-the-twitter-api'上如何获取Twitter API的访问权限。

Parameters:

Name Type Description Default
bearer_token str

从Twitter API获取的bearer_token。

required
num_tweets Optional[int]

每个用户Twitter账号的推文数量。默认为100条推文。

100
Source code in llama_index/readers/twitter/base.py
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class TwitterTweetReader(BasePydanticReader):
    """Twitter推文阅读器。

    读取用户Twitter账号的推文。

    查看'https://developer.twitter.com/en/docs/twitter-api/        getting-started/getting-access-to-the-twitter-api'上如何获取Twitter API的访问权限。

    Args:
        bearer_token (str): 从Twitter API获取的bearer_token。
        num_tweets (Optional[int]): 每个用户Twitter账号的推文数量。默认为100条推文。
"""

    is_remote: bool = True
    bearer_token: str
    num_tweets: Optional[int]

    def __init__(
        self,
        bearer_token: str,
        num_tweets: Optional[int] = 100,
    ) -> None:
        """使用参数进行初始化。"""
        super().__init__(
            num_tweets=num_tweets,
            bearer_token=bearer_token,
        )

    @classmethod
    def class_name(cls) -> str:
        return "TwitterTweetReader"

    def load_data(
        self,
        twitterhandles: List[str],
        num_tweets: Optional[int] = None,
        **load_kwargs: Any
    ) -> List[Document]:
        """加载Twitter用户的推文。

Args:
    twitterhandles(List[str]):要读取推文的用户Twitter句柄列表。
"""
        try:
            import tweepy
        except ImportError:
            raise ImportError(
                "`tweepy` package not found, please run `pip install tweepy`"
            )

        client = tweepy.Client(bearer_token=self.bearer_token)
        results = []
        for username in twitterhandles:
            # tweets = api.user_timeline(screen_name=user, count=self.num_tweets)
            user = client.get_user(username=username)
            tweets = client.get_users_tweets(
                user.data.id, max_results=num_tweets or self.num_tweets
            )
            response = " "
            for tweet in tweets.data:
                response = response + tweet.text + "\n"
            results.append(Document(text=response, id_=username))
        return results

load_data #

load_data(
    twitterhandles: List[str],
    num_tweets: Optional[int] = None,
    **load_kwargs: Any
) -> List[Document]

加载Twitter用户的推文。

Source code in llama_index/readers/twitter/base.py
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
    def load_data(
        self,
        twitterhandles: List[str],
        num_tweets: Optional[int] = None,
        **load_kwargs: Any
    ) -> List[Document]:
        """加载Twitter用户的推文。

Args:
    twitterhandles(List[str]):要读取推文的用户Twitter句柄列表。
"""
        try:
            import tweepy
        except ImportError:
            raise ImportError(
                "`tweepy` package not found, please run `pip install tweepy`"
            )

        client = tweepy.Client(bearer_token=self.bearer_token)
        results = []
        for username in twitterhandles:
            # tweets = api.user_timeline(screen_name=user, count=self.num_tweets)
            user = client.get_user(username=username)
            tweets = client.get_users_tweets(
                user.data.id, max_results=num_tweets or self.num_tweets
            )
            response = " "
            for tweet in tweets.data:
                response = response + tweet.text + "\n"
            results.append(Document(text=response, id_=username))
        return results