path: root/src/portal/py/tests/old_twitter_api.py

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)