基类：BasePydanticReader
Twitter 推文阅读器。
读取用户 Twitter 账号的推文。
关于如何获取 Twitter API 访问权限，请查看“https://developer.twitter.com/en/docs/twitter-api/ getting-started/getting-access-to-the-twitter-api”。
参数
| 名称 |
类型 |
描述 |
默认值 |
bearer_token
|
str
|
从 Twitter API 获取的 bearer_token。
|
必需
|
num_tweets
|
可选[整型]
|
每个用户 Twitter 账号的推文数量。默认是 100 条推文。
|
100
|
源代码位于 llama-index-integrations/readers/llama-index-readers-twitter/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
70
71
72
73
74
75
76 | class TwitterTweetReader(BasePydanticReader):
"""
Twitter tweets reader.
Read tweets of user twitter handle.
Check 'https://developer.twitter.com/en/docs/twitter-api/\
getting-started/getting-access-to-the-twitter-api' \
on how to get access to twitter API.
Args:
bearer_token (str): bearer_token that you get from twitter API.
num_tweets (Optional[int]): Number of tweets for each user twitter handle.\
Default is 100 tweets.
"""
is_remote: bool = True
bearer_token: str
num_tweets: Optional[int]
def __init__(
self,
bearer_token: str,
num_tweets: Optional[int] = 100,
) -> None:
"""Initialize with parameters."""
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]:
"""
Load tweets of twitter handles.
Args:
twitterhandles (List[str]): List of user twitter handles to read tweets.
"""
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(twitterhandles: List[str], num_tweets: Optional[int] = None, **load_kwargs: Any) -> List[Document]
加载指定 Twitter 账号的推文。
参数
| 名称 |
类型 |
描述 |
默认值 |
twitterhandles
|
列表[字符串]
|
|
必需
|
源代码位于 llama-index-integrations/readers/llama-index-readers-twitter/llama_index/readers/twitter/base.py
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
70
71
72
73
74
75
76 | def load_data(
self,
twitterhandles: List[str],
num_tweets: Optional[int] = None,
**load_kwargs: Any
) -> List[Document]:
"""
Load tweets of twitter handles.
Args:
twitterhandles (List[str]): List of user twitter handles to read tweets.
"""
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
|
