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