import sys import os import shutil import gzip import json import email.utils sys.path.append('../py') from base import Method from post import PostEncoder, User, Post, PostRef, PostType, DateType, Image, Video, MediaUrl from modules import ALL_MODULES from archive_query import QueryDownloadThread from logger import Logger module = ALL_MODULES['twitter'][0] if not module.init(): sys.exit(1) #origin_base = '/mnt/store/camu_db/twitter' origin_base = '/mnt/store/files/camu_db/data/twitter_orig' #RUNTIME_PATH = '.' RUNTIME_PATH = '/mnt/store/files/tmp/run' LOG_FILE = f'{RUNTIME_PATH}/twitter_reparse3.log' log = Logger(LOG_FILE) compat_thread = QueryDownloadThread(log, None, None, sys.argv[1]) def compat_media_download(obj, post, output_base): for index, key in enumerate(post.media.keys()): media = post.media[key] url = media.url media_path = f'{output_base}_media{key}.{url.ext}' if os.path.isfile(media_path): obj.log.write(f'Skipping media {media_path}.') continue origin_path = f'{origin_base}/{post.author.unique_id.replace(':', '_')}/{post.unique_id.replace(':', '_')}_media{index}.{url.ext}' if os.path.isfile(origin_path): shutil.copyfile(origin_path, media_path) #post.media[key].url.url = url.url.replace('name=orig', 'name=large') obj.log.write(f'Used backup {origin_path}.') continue obj.log.write(f'Attempting to download file {media_path}.') response = module.do_request(Method.GET, url.url, None, 1) if not response: return False if not obj.write_to_file(media_path, 'wb+', response.content, False): return False compat_thread.media_download = compat_media_download def process_post(post): compat_thread.supply_post(post) #print(json.dumps(post, indent=4, cls=PostEncoder)) def create_media_url(url, format=True): question = url.rfind('?') if question >= 0: ext = url[0:question].rsplit('.')[-1] else: ext = url.rsplit('.')[-1] if format: return MediaUrl(f'{url}?format={ext}&name=orig', ext), MediaUrl(f'{url}?format={ext}&name=small', ext) return MediaUrl(url, ext), None def create_media(m): original, thumbnail = create_media_url(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) url = variants[0]['url'] video_url, _ = create_media_url(url, False) return Video(url=video_url, thumbnail_url=thumbnail) return None def create_user(data): unique_id = f'twitter:u:{data['id']}' user = User(unique_id=unique_id, username=data['screen_name'], display_name=data['name']) user.profile_picture_url = data['profile_image_url_https'] return user def create_post_v1(data, user, mtime, hash): kwargs = {} kwargs['unique_id'] = f'twitter:t:{data['id']}' kwargs['raw_responses'] = {} kwargs['raw_responses']['hash'] = hash kwargs['url'] = f'https://twitter.com/{user.username}/status/{data['id']}' kwargs['dates'] = { DateType.CREATED: email.utils.parsedate_to_datetime(data['created_at']).timestamp(), DateType.RETRIEVED: mtime } kwargs['author'] = user if 'retweeted_status_id' in data: kwargs['type'] = PostType.REPOST kwargs['post'] = PostRef(f'twitter:t:{data['retweeted_status_id']}') process_post(Post(**kwargs)) return kwargs['type'] = PostType.POST kwargs['text'] = data['full_text'] kwargs['likes'] = data['favorite_count'] kwargs['reposts'] = data['retweet_count'] kwargs['quotes'] = data['quote_count'] kwargs['comments'] = data['reply_count'] media = {} if 'media' in data['entities']: for m in data['entities']['media']: media_id = m['id_str'] if 'id_str' in m else str(m['id']) media[media_id] = create_media(m) if 'extended_entities' in data and 'media' in data['extended_entities']: for m in data['extended_entities']['media']: media_id = m['id_str'] if 'id_str' in m else str(m['id']) media[media_id] = create_media(m) kwargs['media'] = media if data['in_reply_to_status_id']: kwargs['in_reply_to'] = PostRef(f'twitter:t:{data['in_reply_to_status_id']}') if 'quoted_status_id' in data: kwargs['quoted'] = PostRef(f'twitter:t:{data['quoted_status_id']}') process_post(Post(**kwargs)) def parse_v1_raw_response(obj, mtime, hash): users = {} for id, user in obj['globalObjects']['users'].items(): users[id] = create_user(user) for tweet in obj['globalObjects']['tweets'].values(): create_post_v1(tweet, users[tweet['user_id_str']], mtime, hash) def create_post_graphql(data, users, tweet_id, mtime, hash): if not tweet_id: tweet_id = data.get('rest_id', None) if data['__typename'] == 'TweetWithVisibilityResults': data = data['tweet'] if not tweet_id: tweet_id = data.get('rest_id', None) if not tweet_id: log.write('no id') return PostRef() kwargs = {} unique_id = f'twitter:t:{tweet_id}' kwargs['unique_id'] = unique_id kwargs['raw_responses'] = {} kwargs['raw_responses']['hash'] = hash if 'legacy' not in data or ('__typename' in data and (data['__typename'] == 'TweetTombstone' or data['__typename'] == 'TweetUnavailable')): kwargs['type'] = PostType.TOMBSTONE process_post(Post(**kwargs)) return PostRef(unique_id) tweet = data['legacy'] user_id = tweet['user_id_str'] if user_id not in users: user_data = data['core']['user_results']['result']['legacy'] user = User(unique_id=f'twitter:u:{user_id}', username=user_data['screen_name'], display_name=user_data['name']) user.profile_picture_url, _ = create_media_url(user_data['profile_image_url_https'], False) users[user_id] = user else: user = users[user_id] kwargs['url'] = f'https://twitter.com/{user.username}/status/{tweet_id}' kwargs['dates'] = { DateType.CREATED: email.utils.parsedate_to_datetime(tweet['created_at']).timestamp(), DateType.RETRIEVED: mtime } kwargs['author'] = user if 'retweeted_status_result' in tweet: if 'result' in tweet['retweeted_status_result']: retweet_id = create_post_graphql(tweet['retweeted_status_result']['result'], users, None, mtime, hash) elif 'retweeted_status_id' in tweet: retweet_id = PostRef(f'twitter:t:{tweet['retweeted_status_id']}') else: retweet_id = PostRef() kwargs['type'] = PostType.REPOST kwargs['post'] = retweet_id process_post(Post(**kwargs)) return PostRef(unique_id) kwargs['type'] = PostType.POST kwargs['text'] = tweet['full_text'] kwargs['likes'] = tweet['favorite_count'] kwargs['reposts'] = tweet['retweet_count'] kwargs['quotes'] = tweet['quote_count'] kwargs['comments'] = tweet['reply_count'] if 'views' in data and 'count' in data['views']: kwargs['views'] = int(data['views']['count']) if 'urls' in tweet['entities']: urls = [] for url in tweet['entities']['urls']: urls.append(url['expanded_url']) kwargs['links'] = urls media = {} if 'media' in tweet['entities']: for m in tweet['entities']['media']: media_id = m['id_str'] if 'id_str' in m else str(m['id']) media[media_id] = create_media(m) if 'extended_entities' in tweet and 'media' in tweet['extended_entities']: for m in tweet['extended_entities']['media']: media_id = m['id_str'] if 'id_str' in m else str(m['id']) media[media_id] = create_media(m) kwargs['media'] = media if 'quoted_status_result' in data: if 'result' not in data['quoted_status_result']: quoted_id = PostRef(f'twitter:t:{tweet['quoted_status_id_str']}') else: quoted_id = create_post_graphql(data['quoted_status_result']['result'], users, None, mtime, hash) kwargs['quoted'] = quoted_id elif 'quoted_status_id_str' in tweet: log.write('quote id no data') quoted_id = PostRef(f'twitter:t:{tweet['quoted_status_id_str']}') kwargs['quoted'] = quoted_id elif data.get('quotedRefResult'): log.write('ref quote') quoted = data['quotedRefResult']['result'] if quoted['__typename'] == 'TweetWithVisibilityResults': quoted = quoted['tweet'] quoted_id = PostRef(f'twitter:t:{quoted['rest_id']}') kwargs['quoted'] = quoted_id if 'in_reply_to_status_id_str' in tweet: kwargs['in_reply_to'] = PostRef(f'twitter:t:{tweet['in_reply_to_status_id_str']}') process_post(Post(**kwargs)) return PostRef(unique_id) def parse_graphql_raw_response(obj, mtime, hash): users = {} if 'user' in obj['data']: instructions = obj['data']['user']['result']['timeline_v2']['timeline']['instructions'] elif 'search_by_raw_query' in obj['data']: instructions = obj['data']['search_by_raw_query']['search_timeline']['timeline']['instructions'] else: instructions = [] for inst in instructions: if inst['type'] == 'TimelineAddEntries': for entry in inst['entries']: if entry['entryId'].startswith('tweet-'): tweet_id = int(entry['entryId'].split('-', 1)[1]) if entry['content']['entryType'] == 'TimelineTimelineItem' and entry['content']['itemContent']['itemType'] == 'TimelineTweet': if 'result' not in entry['content']['itemContent']['tweet_results']: continue create_post_graphql(entry['content']['itemContent']['tweet_results']['result'], users, tweet_id, mtime, hash) else: log.write('Got unrecognised timeline tweet item(s)') # snscrape copy and paste elif entry['entryId'].startswith(('homeConversation-', 'profile-conversation-')): if entry['content']['entryType'] == 'TimelineTimelineModule': for item in reversed(entry['content']['items']): if not item['entryId'].startswith(entry['entryId'].split('ion-', 1)[0] + 'ion-') or '-tweet-' not in item['entryId']: log.write(f'Unexpected conversation entry ID: {item['entryId']!r}') # snscrape copy and paste continue if item['item']['itemContent']['itemType'] == 'TimelineTweet': tweet_id = int(item['entryId'].split('-tweet-', 1)[1]) if 'result' in item['item']['itemContent']['tweet_results']: create_post_graphql(item['item']['itemContent']['tweet_results']['result'], users, tweet_id, mtime, hash) else: kwargs = {} unique_id = f'twitter:t:{tweet_id}' kwargs['unique_id'] = unique_id kwargs['raw_responses'] = {} kwargs['raw_responses']['hash'] = hash kwargs['type'] = PostType.TOMBSTONE process_post(Post(**kwargs)) def parse_raw_response(path, hash): mtime = os.path.getmtime(path) with gzip.open(path, 'rb') as f: obj = json.loads(f.read()) if 'globalObjects' not in obj: parse_graphql_raw_response(obj, mtime, hash) else: parse_v1_raw_response(obj, mtime, hash) #target = '/mnt/store/camu_db/twitter/raw_responses' target = '/mnt/store/files/camu_db/data/twitter_orig/raw_responses' for file in os.listdir(target): hash = file.split('.')[0] print(hash) parse_raw_response(f'{target}/{file}', hash)