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
|
import sys
import json
import email.utils
from json import JSONDecodeError
from typing import Optional, Any
from base import USER_AGENT, Search, Module, Method, ParsedJson
from post import Url, PostType, PostRef, Date, DateType, DateMeta, Post, User, Media, Image, Video, PostEncoder
from modules.common import get_current_utc_time
class TwitterParserState():
def __init__(self, response_hash: str, mtime: Optional[float], module: Module):
self.response_hash = response_hash;
self.mtime: float = mtime if mtime else get_current_utc_time().timestamp()
self.module: Module = module
self.cursor: Optional[dict] = None
self.page: list[str] = []
def parse_media_url_https(url: str) -> (Url, Url):
question = url.rfind('?')
if question >= 0:
url = url[0:question]
ext = url.rsplit('.')[-1]
return Url(f'{url}?format={ext}&name=orig', ext), Url(f'{url}?format={ext}&name=small', ext)
def create_media_graphql(m: ParsedJson) -> Media:
original, thumbnail = parse_media_url_https(m['media_url_https'])
if m['type'] == 'photo':
return Image(url=original, thumbnail_url=thumbnail)
elif m['type'] == 'video' or m['type'] == 'animated_gif':
variants = sorted(m['video_info']['variants'], key=lambda x: x['bitrate'] if 'bitrate' in x else 0, reverse=True)
return Video(url=Url(variants[0]['url']), thumbnail_url=thumbnail)
return None
def create_post_graphql(data: ParsedJson, tps: TwitterParserState) -> str:
kwargs = {}
kwargs['module'] = 'twitter'
tweet_id = data['rest_id']
unique_id = f'twitter:t:{tweet_id}'
kwargs['unique_id'] = unique_id
kwargs['raw_responses'] = {}
kwargs['raw_responses']['graphql'] = tps.response_hash
kwargs['dates'] = [Date(DateType.RETRIEVED, (tps.mtime, tps.mtime), DateMeta.NONE)]
if 'core' in data and 'user_results' in data['core']:
user_data = data['core']['user_results']['result']
if user_data['__typename'] != 'User':
print(f'Unexpected type {user_data['__typename']} when expecting User')
user_id = user_data['rest_id']
user = User(unique_id=f'twitter:u:{user_id}')
if 'core' in user_data:
user.username = user_data['core']['screen_name']
user.name = user_data['core']['name']
else:
user.username = user_data['legacy']['screen_name']
user.name = user_data['legacy']['name']
if 'avatar' in user_data:
user.profile_picture_url = Url(user_data['avatar']['image_url'])
else:
user.profile_picture_url = Url(user_data['legacy']['profile_image_url_https'])
kwargs['url'] = f'https://twitter.com/{user.username}/status/{tweet_id}'
kwargs['author'] = user
if 'legacy' not in data or ('__typename' in data and (data['__typename'] == 'TweetTombstone' or data['__typename'] == 'TweetUnavailable')):
kwargs['type'] = PostType.TOMBSTONE
post = Post(**kwargs)
tps.module.add_to_map(unique_id, post)
return unique_id
legacy = data['legacy']
create_date = email.utils.parsedate_to_datetime(legacy['created_at']).timestamp()
kwargs['dates'].append(Date(DateType.CREATED, (create_date, create_date), DateMeta.NONE))
# @TODO: Find example of retweeted_status_id.
is_retweet = 'retweeted_status_result' in legacy or 'retweeted_status_id' in legacy
if is_retweet:
kwargs['type'] = PostType.REPOST
if 'retweeted_status_result' in legacy and 'result' in legacy['retweeted_status_result']:
retweet_id = PostRef(create_post_graphql(legacy['retweeted_status_result']['result'], tps))
elif 'retweeted_status_id' in legacy:
retweet_id = PostRef(f'twitter:t:{legacy['retweeted_status_id']}')
else:
print('Unknown ID for supposed retweet.')
retweet_id = PostRef()
post = Post(**kwargs)
tps.module.add_to_map(unique_id, post)
return unique_id
kwargs['type'] = PostType.POST
kwargs['text'] = legacy.get('full_text', '')
kwargs['likes'] = legacy.get('favorite_count', None)
kwargs['bookmarks'] = legacy.get('bookmark_count', None)
kwargs['reposts'] = legacy.get('retweet_count', None)
kwargs['quotes'] = legacy.get('quote_count', None)
kwargs['comments'] = legacy.get('reply_count', None)
if 'views' in data:
kwargs['views'] = data['views'].get('count', None)
merged_entities = []
if 'extended_entities' in legacy and 'media' in legacy['extended_entities']:
merged_entities.extend(legacy['extended_entities']['media'])
merged_entities.extend(legacy['entities'].get('media', []))
seen_media_ids = []
kwargs['media'] = {}
for m in merged_entities:
if m['id_str'] in seen_media_ids:
continue
seen_media_ids.append(m['id_str'])
kwargs['media'][m.get('media_key', m['id_str'])] = create_media_graphql(m)
# quotedRefResult references the tweet that quoted this tweet, not the tweet this tweet is quoting.
has_quoted = 'quoted_status_result' in legacy or 'quoted_status_id_str' in legacy
if has_quoted:
if 'quoted_status_result' in legacy and 'result' in legacy['quoted_status_result']:
quoted_id = PostRef(create_post_graphql(legacy['quoted_status_result']['result'], tps))
elif 'quoted_status_id_str' in legacy:
quoted_id = PostRef(f'twitter:t:{legacy['quoted_status_id_str']}')
else:
print('Unknown ID for supposed quoted tweet.')
quoted_id = PostRef()
if 'in_reply_to_status_id_str' in legacy:
kwargs['in_reply_to'] = PostRef(f'twitter:t:{legacy['in_reply_to_status_id_str']}')
post = Post(**kwargs)
tps.module.add_to_map(unique_id, post)
return unique_id
def parse_graphql_tweet(tweet: ParsedJson, tps: TwitterParserState):
entry_id = tweet['entryId']
if 'item' not in tweet or 'itemContent' not in tweet['item']:
print(f'Unhandled Timeline entry {entry_id}')
return
content = tweet['item']['itemContent']
if content['__typename'] != 'TimelineTweet':
print(f'Unhandled Timeline content type {content['__typename']}')
return
result = content['tweet_results']['result']
if result['__typename'] != 'Tweet':
print(f'Unhandled Tweet type {result['__typename']}')
return
tps.page.append(create_post_graphql(result, tps))
def parse_graphql_timeline_v2(timeline: ParsedJson, tps: TwitterParserState):
for ins in timeline['timeline']['instructions']:
ins_type = ins['type']
nops = ['TimelineClearCache', 'TimelineTerminateTimeline', 'TimelinePinEntry']
if ins_type in nops:
print(f'no-op: {ins_type}')
elif ins_type == 'TimelineAddEntries':
for entry in ins['entries']:
entry_id = entry['entryId']
if entry_id.startswith('cursor-top-'):
pass
elif entry_id.startswith('cursor-bottom-'):
tps.cursor = entry['content']['value']
elif 'content' in entry and entry['content']['entryType'] == 'TimelineTimelineModule':
for item in entry['content']['items']:
parse_graphql_tweet(item, tps)
elif ins_type == 'TimelineAddToModule':
for item in ins['moduleItems']:
parse_graphql_tweet(item, tps)
else:
print(f'Unknown instruction: {ins_type}')
def parse_graphql_raw_response(obj: ParsedJson, response_hash: str, mtime: Optional[float], module: Module) -> TwitterParserState:
tps = TwitterParserState(response_hash, mtime, module)
if 'data' in obj:
if 'user' in obj['data']:
user = obj['data']['user']
if 'result' in user:
parse_graphql_timeline_v2(
user['result']['timeline_v2'] if 'timeline_v2' in user['result'] else user['result']['timeline'], tps)
if len(tps.page) == 0:
print(f'Response with 0 tweets:\n{json.dumps(obj)}')
else:
print('Invalid graphql user.')
else:
print(f'Unexpected response:\n{json.dumps(obj)}')
return tps
def parse_v1_raw_response(obj: ParsedJson, reponse_hash: str, mtime: Optional[float]):
pass
#target = sys.argv[1]
#with open(target, 'r') as f:
# parse_graphql_raw_response(json.loads(f.read())[0], 'AAAAAAAA', None, Module())
|