From 60b4ebfbf3be78dba9dc7c65ab2bdaa0b218c0c2 Mon Sep 17 00:00:00 2001 From: Andrew Opalach Date: Mon, 21 Oct 2024 19:22:50 -0400 Subject: Everything before initial synced list Signed-off-by: Andrew Opalach --- src/portal/py/tests/__init__.py | 0 src/portal/py/tests/archive_query.py | 263 ++++++++++++++++++++++++++++++++ src/portal/py/tests/logger.py | 15 ++ src/portal/py/tests/old_twitter_api.py | 271 +++++++++++++++++++++++++++++++++ src/portal/py/tests/rewrite_post.py | 56 +++++++ src/portal/py/tests/test.py | 103 +++++++++++++ src/portal/py/tests/unescape_json.py | 10 ++ 7 files changed, 718 insertions(+) create mode 100644 src/portal/py/tests/__init__.py create mode 100644 src/portal/py/tests/archive_query.py create mode 100644 src/portal/py/tests/logger.py create mode 100644 src/portal/py/tests/old_twitter_api.py create mode 100644 src/portal/py/tests/rewrite_post.py create mode 100644 src/portal/py/tests/test.py create mode 100644 src/portal/py/tests/unescape_json.py (limited to 'src/portal/py/tests') diff --git a/src/portal/py/tests/__init__.py b/src/portal/py/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/portal/py/tests/archive_query.py b/src/portal/py/tests/archive_query.py new file mode 100644 index 0000000..1097634 --- /dev/null +++ b/src/portal/py/tests/archive_query.py @@ -0,0 +1,263 @@ +import os +import sys +import signal +import json +import threading +import gzip +from datetime import datetime, timezone +sys.path.append('../') +from base import Method +from post import PostEncoder, PostType, DateType, User +from modules import ALL_MODULES +from logger import Logger + +mode = 'fanbox' +cmd = 'user' +module = ALL_MODULES[mode][0] +if not module.init(): + sys.exit(1) + +quit_mutex = threading.Lock() +queue_mutex = threading.Lock() +queue_cond = threading.Condition(queue_mutex) +is_running = True + +def default_media_download(obj, post, output_base): + for key in post.media.keys(): + download_params = module.get_download(post.unique_id, key) + if not download_params: + obj.log.write(f'Skipping media{key} for post {post.unique_id}.') + continue + for url in download_params['urls']: + media_path = f'{output_base}_media{key}.{url.ext}' + obj.log.write(f'Attempting to download file {media_path}.') + response = module.do_request(Method.GET, url.url) + if not response: + obj.log.write(f'ERROR: Downloading post media failed ({post.unique_id} #{key}).') + continue + if not obj.write_to_file(media_path, 'wb+', response.content, False): + return False + +class QueryDownloadThread(threading.Thread): + def __init__(self, log, complete, query, output_dir): + super().__init__() + self.log = log + self.complete = complete + self.query = query + self.output_dir = output_dir + self.media_download = default_media_download + + def make_path(self, post): + prefix: str + if post.type != PostType.TOMBSTONE: + prefix = post.author.unique_id.replace(':', '_') + else: + prefix = 'tombstones' + path = f'{self.output_dir}/{prefix}' + if not os.path.isdir(path): + os.mkdir(path) + return path + + def write_to_file(self, path, open_method, content, compress): + try: + if compress: + with gzip.open(path, open_method) as f: + f.write(content.encode('utf-8')) + else: + with open(path, open_method) as f: + f.write(content) + except Exception as e: + self.log.write(f'ERROR: Writing file failed: {path} ({repr(e)}).') + return False + return True + + def write_post_to_disk(self, post): + if post.post.unique_id: + original = module.get_item(post.post.unique_id) + if original: + self.write_post_to_disk(original) + if post.quoted.unique_id: + quoted = module.get_item(post.quoted.unique_id) + if quoted: + self.write_post_to_disk(quoted) + message = f'Downloading post {post.unique_id}' + if 'detached_api' in post.raw_responses: + raw_response_path = f'{self.output_dir}/raw_responses' + if not os.path.isdir(raw_response_path): + os.mkdir(raw_response_path) + raw_response_path += f'/{post.raw_responses['hash']}.json.gz' + if not self.write_to_file(raw_response_path, 'wb+', json.dumps(json.loads(post.raw_responses['detached_api'])), True): + return False + del post.raw_responses['detached_api'] + if post.type != PostType.TOMBSTONE: + message += f':{datetime.fromtimestamp(post.dates[DateType.CREATED], tz=timezone.utc).isoformat()} from {post.author.username}({post.author.display_name}):{post.author.unique_id}.' + else: + message += '.' + output_path = self.make_path(post) + output_base = f'{output_path}/{post.unique_id.replace(':', '_')}' + output = output_base + '.json.gz' + ''' + if not os.path.isfile(output): + print('done') + sys.exit(1) + ''' + ''' + has_mp4 = False + for m in post.media.values(): + if m.url.ext == 'mp4': + self.log.write('Rewriting post with mp4.') + has_mp4 = True + break + if post.quoted.unique_id: + self.log.write('Rewriting post with quote.') + has_mp4 = True + ''' + if os.path.isfile(output): + skip_message = f'Skipping already downloaded post {post.unique_id}' + if post.type != PostType.TOMBSTONE: + skip_message += f' from {post.author.unique_id}.' + else: + skip_message += '.' + self.log.write(skip_message) + return True + self.log.write(message) + self.media_download(self, post, output_base) + if not self.write_to_file(output, 'wb+', json.dumps(post, cls=PostEncoder), True): + return False + return True + + def supply_post(self, post): + if not self.write_post_to_disk(post): + return False + return True + + def run(self): + global quit_mutex + global queue_cond + global is_running + search = module.search(self.query) + if not search: + return + completed = True + arg = self.query.split(':')[-1] + num = 0 + count = 0 + while completed: + self.log.write(f'Starting page {num} of {arg}.') + page = search.get_page(num) + if not page: + break + num += 1 + for unique_id in page: + if unique_id == 'twitter:t:1597637140833529856': + self.log.write('---------Hit target---------') + with quit_mutex: + if not is_running: + self.complete.write(f'{mode}:{cmd}:{arg}:i:{str(count)},{str(num)}') + completed = False + break + post = module.get_item(unique_id) + if not post: + self.log.write(f'Missing post with id {unique_id}.') + continue + if not self.write_post_to_disk(post): + continue + count += 1 + if completed: + self.complete.write(f'{mode}:{cmd}:{arg}:c:{str(count)},{str(num)}') + with queue_cond: + queue_cond.notify() + +#RUNTIME_PATH = "./run" +RUNTIME_PATH = "/mnt/store/files/tmp/run" +LOG_FILE = f'{RUNTIME_PATH}/archive3.log' +ARG_FILE = f'{RUNTIME_PATH}/completed_args3.log' + +def read_completed_args(path): + args = [] + if os.path.isfile(path): + with open(path, 'r') as f: + for line in f.read().splitlines(): + if line[0] == '-': + continue + s = line.split(':') + if s[3] == 'c': + args.append(f'{s[0]}:{s[1]}:{s[2]}') + return args + +def download_args(): + global quit_mutex + global queue_cond + global compat_thread + + completed_args = read_completed_args(ARG_FILE) + + log = Logger(LOG_FILE) + complete = Logger(ARG_FILE) + + header = f'--------{datetime.now().isoformat()}---------' + log.write(header) + complete.write(header) + + output_dir = sys.argv[1] + if not os.path.isdir(output_dir): + os.mkdir(output_dir) + + #args = module.search(f'following:{22781328}') + args = ['anoh223'] + + threads = [] + start = True + + for arg in args: + #if mode == 'fanbox': + # user = module.get_item(arg) + # if not user or not isinstance(user, User): + # log.write(f'Failed to retrive information for user {arg}.') + # continue + # arg = user.username + with quit_mutex: + if not is_running: + break + if mode == 'pixiv_web' and not start: + if arg == '': + start = True + continue + if not start: + continue + if mode != 'instagram': + arg = f'{cmd}:{arg}' + if f'{mode}:{arg}' in completed_args: + log.write(f'Skipping already downloaded arg {arg}.') + continue + if len(threads) >= 1: + with queue_cond: + queue_cond.wait() + threads = [t for t in threads if t.is_alive()] + with quit_mutex: + if not is_running: + break + log.write(f'***********************\nStarting download of {arg}\n***********************\n') + arg_thread = QueryDownloadThread(log, complete, arg, output_dir) + threads.append(arg_thread) + arg_thread.start() + + for t in threads: + t.join() + + log.close() + complete.close() + +def signal_handler(sig, frame): + global quit_mutex + global is_running + print('Attempting to quit...') + with quit_mutex: + if not is_running: + sys.exit(1) + else: + is_running = False + +if __name__ == "__main__": + signal.signal(signal.SIGINT, signal_handler) + download_args() diff --git a/src/portal/py/tests/logger.py b/src/portal/py/tests/logger.py new file mode 100644 index 0000000..80e22da --- /dev/null +++ b/src/portal/py/tests/logger.py @@ -0,0 +1,15 @@ +import threading + +class Logger(): + def __init__(self, path): + self.file = open(path, 'a+') + self.mutex = threading.Lock() + + def write(self, msg): + with self.mutex: + print(msg) + self.file.write(msg + '\n') + self.file.flush() + + def close(self): + self.file.close() diff --git a/src/portal/py/tests/old_twitter_api.py b/src/portal/py/tests/old_twitter_api.py new file mode 100644 index 0000000..40d0346 --- /dev/null +++ b/src/portal/py/tests/old_twitter_api.py @@ -0,0 +1,271 @@ +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 i, 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{i}.{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) diff --git a/src/portal/py/tests/rewrite_post.py b/src/portal/py/tests/rewrite_post.py new file mode 100644 index 0000000..3bb152b --- /dev/null +++ b/src/portal/py/tests/rewrite_post.py @@ -0,0 +1,56 @@ +import sys +import os +import gzip +import json +import hashlib +sys.path.append('../py') +import config +#from post import PostEncoder, DateType +#from modules.pixiv_web import PixivWebBase, PixivWebModule +#from modules.twitter import TwitterScrapeBase, TwitterScrapeModule +from logger import Logger + +#module = PixivWebModule(config.PIXIV_SESSID) +#search = PixivWebBase(module) + +log = Logger('deleted_posts.log') + +''' +def rewrite_post(path): + print(f'Rewriting post @ {path}') + with gzip.open(path, 'rb') as f: + obj = json.loads(f.read()) + post = search.remake_post(obj['raw_responses'], obj['dates'][str(DateType.RETRIEVED.value)]) + if not post: + log.write(path) + return + #print(json.dumps(post, indent=4, cls=PostEncoder)) + with gzip.open(path, 'wb') as f: + f.write(json.dumps(post, cls=PostEncoder).encode('utf-8')) +''' + +def dump_raw_response(path, target): + print(f'Extracting raw_response from {path}') + with gzip.open(path, 'rb') as f: + obj = json.loads(f.read()) + if 'api' in obj['raw_responses']: + hasher = hashlib.new('sha256') + data = json.dumps(json.loads(obj['raw_responses']['api'])).encode('utf-8') + hasher.update(data) + output = f'{target}/{hasher.hexdigest()}.json.gz' + with gzip.open(output, 'wb+') as f: + f.write(data) + +target = sys.argv[1] +raw_responses = f'{target}/raw_responses' + +if not os.path.isdir(raw_responses): + os.mkdir(raw_responses) + +for user in os.listdir(target): + if user == 'raw_responses': + continue + for file in os.listdir(f'{target}/{user}'): + if file.split('.')[-1] == 'gz': + dump_raw_response(f'{target}/{user}/{file}', raw_responses) + #rewrite_post(f'{target}/{user}/{file}') diff --git a/src/portal/py/tests/test.py b/src/portal/py/tests/test.py new file mode 100644 index 0000000..bee7265 --- /dev/null +++ b/src/portal/py/tests/test.py @@ -0,0 +1,103 @@ +import json +import httpx +import os.path +from base import Method +from typing import Optional, Any +from post import DateType, DateMeta, Date, PostEncoder, Post, MediaType +from modules import ALL_MODULES +#from twitter.scraper import Scraper + +#scraper = Scraper(cookies = { +# 'ct0': '5a3c98b7b50a74dc556fa497f497932e8958b5b2ef7ba74440ce3f341bb4753f663d3fe7633222d62a52d4ce121673234b927d97eeef806129d42bbfba7ab4d4cfbd2749c20fe6f8da480d0a1e6629e1', +# 'auth_token': 'ea66b4c2ebd04e22de855dce6f1366221635bf2e' +#}) +# +#media = scraper.media([1557919548904419328], limit=10) +#print(media) + +#output = '/mnt/store/files/ext/patreon_tmp/FPSBlyck' +# +#module = ALL_MODULES['patreon'][0] +#module.init() +#search = module.search('FPSBlyck') +#page_num = 0 +#mark = False +#while not mark: +# page = search.get_page(page_num) +# if not page: +# break +# page_num += 1 +# for unique_id in page: +# if unique_id == 'patreon:p:29707872': +# print('hit mark') +# mark = True +# post = module.get_item(unique_id) +# print(json.dumps(post, indent=4, cls=PostEncoder)) +# for key, m in post.media.items(): +# if m.type == MediaType.FILE: +# path = f'{output}/{post.unique_id.replace(':', '_')}_{m.name}' +# if not os.path.isfile(path): +# dl = module.get_download(post.unique_id, key) +# retries = 3 +# r: Optional[httpx.Response] = None +# while retries >= 0: +# try: +# r = httpx.get(dl['urls'][0].url, headers=dl['headers'], cookies=dl['cookies'], follow_redirects=True) +# except: +# retries -= 1 +# else: +# break +# if r: +# with open(path, 'wb+') as f: +# f.write(r.content) +# else: +# print('Download failed') +# else: +# print('Skipping file') + +ALL_MODULES['youtube'][0].init() +search = ALL_MODULES['youtube'][0].search('yoyoyoy') +page = search.get_page(0) +print(page) +#print(json.dumps(page, indent=4, cls=PostEncoder)) + +#d = Date(DateType.EDITED, (0.0, 0.0), DateMeta.EDITED_AT_UNKNOWN_TIME) +#d = Date(DateType.EDITED, (0.0, 0.0), DateMeta.NONE) +#print(json.dumps(d, indent=4, cls=PostEncoder)) +#sys.exit(1) + +#module = ALL_MODULES['fanbox'][0] +#module = ALL_MODULES['pixiv_web'][0] +#module = ALL_MODULES['youtube'][0] +#module = ALL_MODULES['twitter'][0] +#module.init() + +#search = module.search('opium_00pium') +#search = module.search('search:ddd') +#page = search.get_page(0) +#for unique_id in page: +# post = module.get_item(unique_id) +# print(json.dumps(post, indent=4, cls=PostEncoder)) +# break + +#post = module.get_item(page[0]) +#for key in post.media.keys(): +# download_params = module.get_download(post.unique_id, key) +# for url in download_params['urls']: +# #response = module.do_request(Method.GET, url.url) +# r = httpx.get(url.url, headers=download_params['headers']) +# with open(f'test_{key}.{url.ext}', 'wb+') as f: +# f.write(r.content) + +#print(json.dumps(page, indent=4, cls=PostEncoder)) +#for i in range(0, len(post.media)): +# download_params = module.get_download(post.unique_id, i) +# r = httpx.get(download_params['url'], headers=download_params['headers']) +# if r.status_code != 200: +# print('error{}'.format(r.status_code)) +# else: +# with open('test{}.png'.format(i), 'wb+') as f: +# f.write(r.content) + +#print(json.dumps(page, indent=4, cls=PostEncoder)) +#print(page) diff --git a/src/portal/py/tests/unescape_json.py b/src/portal/py/tests/unescape_json.py new file mode 100644 index 0000000..eaf5194 --- /dev/null +++ b/src/portal/py/tests/unescape_json.py @@ -0,0 +1,10 @@ +import sys +import gzip +import json + +for path in sys.argv[1:]: + with gzip.open(path, 'rb') as f: + obj = json.loads(f.read().decode('utf-8')) + obj = json.loads(obj) + with gzip.open(path, 'wb') as f: + f.write(json.dumps(obj).encode('utf-8')) -- cgit v1.2.3-101-g0448