1
2
3
4
5
6
7
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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
|
import log
import http.cookiejar
from typing import Optional, Any, Iterator
from datetime import timezone
from base import USER_AGENT, Search, Module
from post import Url, PostType, Date, DateType, DateMeta, PostRef, Post, User, Media, Image, Video
from query_parser import QueryParser
from modules.common import get_current_utc_time
# GraphQL API
from snscrape.base import ScraperException
from snscrape.modules import twitter
from snscrape.modules.twitter import (Tweet, TweetRef, Tombstone, UserRef,
TwitterSearchScraper, TwitterSearchScraperMode, TwitterProfileScraper, TwitterUserScraper, TwitterTweetScraper)
class TwitterScrapeBase(Search):
# Posts per emulated page because snscrape returns an iterator.
POSTS_PER_PAGE = 8
# Don't allow requests for a page more than this amount past the current page.
REACH_LIMIT = 4
def __init__(self, userdata: Any) -> None:
super().__init__()
self.module: TwitterScrapeModule = userdata
self.iterator: Iterator[Tweet | TweetRef | Tombstone]
self.page: int = 0
def create_media_url(self, url: str) -> Url:
format = url.find('format=')
ext = None
if format >= 0:
ext = url[format + 7:]
amp = ext.find('&')
if amp >= 0:
ext = ext[:amp]
return Url(url, ext)
def parse_media(self, data: list[twitter.Medium]) -> dict[str, Media]:
media = {}
for index, m in enumerate(data):
if isinstance(m, twitter.Photo):
media[str(index)] = Image(url=self.create_media_url(m.fullUrl), thumbnail_url=self.create_media_url(m.previewUrl))
elif isinstance(m, twitter.Video) or isinstance(m, twitter.Gif):
videos = sorted(m.variants, key=lambda x: x.bitrate if x.bitrate else 0, reverse=True)
media[str(index)] = Video(url=self.create_media_url(videos[0].url), thumbnail_url=self.create_media_url(m.thumbnailUrl))
return media
def create_post(self, tweet: Tweet | TweetRef | Tombstone) -> Post:
kwargs = {}
kwargs['module'] = 'twitter_scrape'
kwargs['unique_id'] = f'twitter:t:{tweet.id}'
if isinstance(tweet, TweetRef) or isinstance(tweet, Tombstone):
kwargs['type'] = PostType.TOMBSTONE
return Post(**kwargs)
kwargs['raw_responses'] = tweet.rawResponses
kwargs['url'] = tweet.url
create_date = tweet.date.replace(tzinfo=timezone.utc).timestamp()
current_date = get_current_utc_time().timestamp()
kwargs['dates'] = [
Date(DateType.CREATED, (create_date, create_date), DateMeta.NONE),
Date(DateType.RETRIEVED, (current_date, current_date), DateMeta.NONE)
]
author = User(unique_id=f'twitter:u:{tweet.user.id}')
if not isinstance(tweet.user, UserRef):
author.username = tweet.user.username
if tweet.user.displayname:
author.display_name = tweet.user.displayname
if tweet.user.profileImageUrl:
author.profile_picture_url = self.create_media_url(tweet.user.profileImageUrl)
kwargs['author'] = author
if tweet.retweetedTweet:
kwargs['type'] = PostType.REPOST
kwargs['post'] = PostRef(f'twitter:t:{tweet.retweetedTweet.id}')
return Post(**kwargs)
kwargs['type'] = PostType.POST
kwargs['text'] = tweet.rawContent
kwargs['likes'] = tweet.likeCount
kwargs['reposts'] = tweet.retweetCount
kwargs['quotes'] = tweet.quoteCount
kwargs['comments'] = tweet.replyCount
kwargs['views'] = tweet.viewCount
if tweet.media:
kwargs['media'] = self.parse_media(tweet.media)
if tweet.inReplyToTweetId:
kwargs['in_reply_to'] = PostRef(f'twitter:t:{tweet.inReplyToTweetId}')
if tweet.quotedTweet:
kwargs['quoted'] = PostRef(f'twitter:t:{tweet.quotedTweet.id}')
return Post(**kwargs)
def add_post_to_page(self, num: int, item: Post | User) -> None:
self.module.add_to_map(item.unique_id, item)
self.pages[num].append(item.unique_id)
def step_iterator_for_page(self, num: int) -> bool:
for _ in range(0, self.POSTS_PER_PAGE):
try:
tweet = next(self.iterator)
except StopIteration:
self.completed = True
break
except ScraperException as e:
log.error(repr(e))
continue
self.add_post_to_page(num, self.create_post(tweet))
if isinstance(tweet, TweetRef) or isinstance(tweet, Tombstone):
continue
if tweet.retweetedTweet:
retweet = self.create_post(tweet.retweetedTweet)
self.module.add_to_map(retweet.unique_id, retweet)
if tweet.quotedTweet:
quote = self.create_post(tweet.quotedTweet)
self.module.add_to_map(quote.unique_id, quote)
return len(self.pages[num]) > 0
def request_page(self, num: int) -> bool:
if num - self.page >= self.REACH_LIMIT:
return False
request_satisfied = False
for index in range(self.page, num + 1):
if self.completed:
break
if index in self.pages:
continue
self.pages[index] = []
if not self.step_iterator_for_page(index):
break
self.page = index
if index == num:
request_satisfied = True
return request_satisfied
class TwitterScrapeSearch(TwitterScrapeBase):
def __init__(self, userdata: Any, arg: str) -> None:
super().__init__(userdata)
self.iterator = TwitterSearchScraper(arg, mode=TwitterSearchScraperMode.TOP, cookies=self.module.cookies).get_items()
class TwitterScrapeUser(TwitterScrapeBase):
def __init__(self, userdata: Any, arg: str) -> None:
super().__init__(userdata)
if arg.startswith('@'):
arg = arg[1:]
self.iterator = TwitterUserScraper(arg, cookies=self.module.cookies).get_items()
class TwitterScrapeProfile(TwitterScrapeBase):
def __init__(self, userdata: Any, arg: str) -> None:
super().__init__(userdata)
if arg.startswith('@'):
arg = arg[1:]
self.iterator = TwitterProfileScraper(arg, cookies=self.module.cookies).get_items()
class TwitterScrapeTweet(TwitterScrapeBase):
def __init__(self, userdata: Any, arg: str) -> None:
super().__init__(userdata)
if arg.startswith('https://'):
arg = arg[arg.rfind('/') + 1:]
question = arg.find('?')
if question >= 0:
arg = arg[:question]
self.iterator = TwitterTweetScraper(arg, cookies=self.module.cookies).get_items()
class TwitterScrapeModule(Module):
def __init__(self, cookies_path: str) -> None:
super().__init__()
self.parser: QueryParser = QueryParser(self, 'search')
self.parser.add_command('search', TwitterScrapeSearch)
self.parser.add_command('user', TwitterScrapeUser)
self.parser.add_command('profile', TwitterScrapeProfile)
self.parser.add_command('tweet', TwitterScrapeTweet)
if cookies_path:
cookie_jar = http.cookiejar.MozillaCookieJar()
cookie_jar.load(filename=cookies_path, ignore_expires=True)
for c in cookie_jar:
if c.value:
self.cookies[c.name] = c.value
def search(self, query: str, *extra_args: Any) -> Optional[Search]:
return self.parser.parse_query(query)
def get_download(self, unique_id: str, key: str) -> Optional[dict[str, Any]]:
if unique_id not in self.unique_id_map:
return None
post = self.unique_id_map[unique_id]
if key not in post.media:
return None
return {
'urls': [post.media[key].url],
'headers': {
'User-Agent': USER_AGENT
}
}
|