summaryrefslogtreecommitdiff
path: root/src/portal/py
diff options
context:
space:
mode:
Diffstat (limited to 'src/portal/py')
-rw-r--r--src/portal/py/base.py25
-rw-r--r--src/portal/py/modules/__init__.py16
-rw-r--r--src/portal/py/modules/common.py2
-rw-r--r--src/portal/py/modules/fanbox.py32
-rw-r--r--src/portal/py/modules/instagram.py10
-rw-r--r--src/portal/py/modules/patreon.py104
-rw-r--r--src/portal/py/modules/pixiv_app.py12
-rw-r--r--src/portal/py/modules/pixiv_web.py25
-rw-r--r--src/portal/py/modules/twitter.py6
-rw-r--r--src/portal/py/modules/youtube.py118
-rw-r--r--src/portal/py/post.py116
-rw-r--r--src/portal/py/test.py1
-rw-r--r--src/portal/py/tests/__init__.py0
-rw-r--r--src/portal/py/tests/archive_query.py263
-rw-r--r--src/portal/py/tests/logger.py15
-rw-r--r--src/portal/py/tests/old_twitter_api.py271
-rw-r--r--src/portal/py/tests/rewrite_post.py56
-rw-r--r--src/portal/py/tests/test.py103
-rw-r--r--src/portal/py/tests/unescape_json.py10
19 files changed, 1035 insertions, 150 deletions
diff --git a/src/portal/py/base.py b/src/portal/py/base.py
index cc883af..7869521 100644
--- a/src/portal/py/base.py
+++ b/src/portal/py/base.py
@@ -1,21 +1,24 @@
import log
import time
+import json
import httpx
+import binascii
import threading
from enum import Enum
+from blake3 import blake3
from collections import OrderedDict
from typing import Optional, Self, Any
from post import Post, User
-USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; rv:122.0) Gecko/20100101 Firefox/122.0'
+type ParsedJson = Any
+
+USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; rv:131.0) Gecko/20100101 Firefox/131.0'
class Method(Enum):
HEAD = "HEAD"
GET = "GET"
POST = "POST"
-type ParsedJson = Any
-
class Search():
def __init__(self):
self.pages: dict[int, list[str]] = {}
@@ -59,8 +62,9 @@ class Module():
def __init__(self):
self.headers: dict[str, str] = {}
self.cookies: dict[str, str] = {}
+ self.mutex: threading.Lock = threading.Lock()
+ self.raw_responses: dict[str, str] = {}
self.unique_id_map: OrderedDict = OrderedDict()
- self.map_mutex: threading.Lock = threading.Lock()
self.session: httpx.Client = httpx.Client(follow_redirects=True, timeout=20.0, http2=True)
def init(self) -> bool:
@@ -84,6 +88,7 @@ class Module():
response = None
continue
if response.status_code == 404 or response.status_code == 403: # Shortcut 404, 403
+ log.warn(f'Returning None, {response.status_code}')
return None
if int(response.status_code / 100) != 2: # Non-200 response
log.error(f'Non-200 status code ({response.status_code}), retrying in 5 seconds...')
@@ -101,8 +106,16 @@ class Module():
break
return response
+ def add_raw_response(self, data: ParsedJson) -> str:
+ with self.mutex:
+ dump = json.dumps(data)
+ hash = blake3(dump.encode('utf-8')).digest()
+ hash_str = binascii.hexlify(hash).decode('ascii')
+ self.raw_responses[hash_str] = dump
+ return hash_str
+
def add_to_map(self, unique_id: str, item: Post | User):
- with self.map_mutex:
+ with self.mutex:
self.unique_id_map[unique_id] = item
if len(self.unique_id_map) > 10240:
self.unique_id_map.popitem(last=False)
@@ -111,7 +124,7 @@ class Module():
return None
def get_item(self, unique_id: str) -> Optional[Post | User]:
- with self.map_mutex:
+ with self.mutex:
return self.unique_id_map.get(unique_id, None)
def get_download(self, unique_id: str, key: str) -> Optional[dict[str, Any]]:
diff --git a/src/portal/py/modules/__init__.py b/src/portal/py/modules/__init__.py
index 994b25a..1a9e9ca 100644
--- a/src/portal/py/modules/__init__.py
+++ b/src/portal/py/modules/__init__.py
@@ -1,11 +1,11 @@
import config
from modules.youtube import YoutubeModule
-from modules.twitter import TwitterScrapeModule
-from modules.pixiv_app import PixivAppModule
-from modules.pixiv_web import PixivWebModule
+#from modules.twitter import TwitterScrapeModule
+#from modules.pixiv_app import PixivAppModule
+#from modules.pixiv_web import PixivWebModule
from modules.fanbox import FanboxModule
-from modules.instagram import InstagramModule
+#from modules.instagram import InstagramModule
from modules.patreon import PatreonModule
ALL_MODULES = {
@@ -13,12 +13,12 @@ ALL_MODULES = {
# 'twitter': (TwitterModule(
# config.TWITTER_ACCESS_TOKEN, config.TWITTER_ACCESS_TOKEN_SECRET,
# config.TWITTER_CONSUMER_TOKEN, config.TWITTER_CONSUMER_TOKEN_SECRET), []),
- 'twitter': (TwitterScrapeModule(config.TWITTER_COOKIES_PATH), []),
- 'pixiv_web': (PixivWebModule(config.PIXIV_SESSID, config.PIXIV_USERID), []),
+# 'twitter': (TwitterScrapeModule(config.TWITTER_COOKIES_PATH), []),
+# 'pixiv_web': (PixivWebModule(config.PIXIV_SESSID, config.PIXIV_USERID), []),
# 'pixiv_app': (PixivAppModule(config.PIXIV_REFRESH_TOKEN), []),
# 'fanbox': (FanboxModule(config.FANBOX_SESSID), []),
# 'instagram': (InstagramModule(
# config.INSTAGRAM_USER_AGENT, config.INSTAGRAM_SETTINGS_PATH,
-# config.INSTAGRAM_SESSION_ID), [])
- 'patreon': (PatreonModule(config.PATREON_SESSION_ID, config.PATREON_UUID, config.PATREON_USER_ID), [])
+# config.INSTAGRAM_SESSION_ID), []),
+# 'patreon': (PatreonModule(config.PATREON_SESSION_ID, config.PATREON_UUID, config.PATREON_USER_ID), [])
}
diff --git a/src/portal/py/modules/common.py b/src/portal/py/modules/common.py
index 62c3d5d..3b87884 100644
--- a/src/portal/py/modules/common.py
+++ b/src/portal/py/modules/common.py
@@ -1,6 +1,6 @@
from datetime import datetime, timezone
-def parse_pixiv_date(date_str: str) -> datetime:
+def reformat_pixiv_date(date_str: str) -> datetime:
rindex = date_str.rfind(':')
date_str = date_str[:rindex] + date_str[rindex + 1:]
return datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%S%z")
diff --git a/src/portal/py/modules/fanbox.py b/src/portal/py/modules/fanbox.py
index 4323d6b..bf55b9d 100644
--- a/src/portal/py/modules/fanbox.py
+++ b/src/portal/py/modules/fanbox.py
@@ -3,9 +3,9 @@ import httpx
from typing import Optional, Any
from json.decoder import JSONDecodeError
from base import USER_AGENT, Search, Module, Method, ParsedJson
-from post import MediaUrl, PostType, DateType, Post, User, Image, File, Tag
+from post import Url, PostType, DateType, Post, User, Image, File, Tag
from query_parser import QueryParser
-from modules.common import parse_pixiv_date, get_current_utc_time
+from modules.common import reformat_pixiv_date, get_current_utc_time
BASE_URL = "https://api.fanbox.cc"
@@ -23,8 +23,8 @@ class FanboxBase(Search):
return None
return obj['body']
- def create_media_url(self, url: str) -> MediaUrl:
- return MediaUrl(url, url.split('.')[-1])
+ def create_media_url(self, url: str) -> Url:
+ return Url(url, url.split('.')[-1])
def create_post(self, data: ParsedJson) -> Optional[Post]:
kwargs = {}
@@ -33,12 +33,12 @@ class FanboxBase(Search):
kwargs['unique_id'] = unique_id
kwargs['raw_responses'] = {}
kwargs['raw_responses']['api'] = json.dumps(data)
- publish_date = parse_pixiv_date(data['publishedDatetime']).timestamp()
+ publish_date = reformat_pixiv_date(data['publishedDatetime']).timestamp()
kwargs['dates'] = {
DateType.CREATED: publish_date,
DateType.RETRIEVED: get_current_utc_time().timestamp()
}
- update_date = parse_pixiv_date(data['updatedDatetime']).timestamp()
+ update_date = reformat_pixiv_date(data['updatedDatetime']).timestamp()
if update_date != publish_date:
kwargs['dates'][DateType.EDITED] = update_date
user = User(unique_id=f'fanbox:u:{data['user']['userId']}', username=data['user']['name'])
@@ -57,11 +57,11 @@ class FanboxBase(Search):
if data['type'] == 'image':
text = data['body']['text']
for i, value in enumerate(data['body']['images']):
- media[str(i)] = Image(url=MediaUrl(value['originalUrl'], value['extension']), thumbnail_url=self.create_media_url(value['thumbnailUrl']))
+ media[str(i)] = Image(url=Url(value['originalUrl'], value['extension']), thumbnail_url=self.create_media_url(value['thumbnailUrl']))
elif data['type'] == 'file':
text = data['body']['text']
for i, value in enumerate(data['body']['files']):
- media[str(i)] = File(url=MediaUrl(value['url'], value['extension']), name=value['name'])
+ media[str(i)] = File(url=Url(value['url'], value['extension']), name=value['name'])
elif data['type'] == 'article':
for block in data['body']['blocks']:
if block['type'] == 'p':
@@ -71,9 +71,9 @@ class FanboxBase(Search):
elif block['type'] == 'url_embed':
text += f'embed::{block['urlEmbedId']}[]' + '\n'
for key, value in data['body']['imageMap'].items():
- media[key] = Image(url=MediaUrl(value['originalUrl'], value['extension']), thumbnail_url=self.create_media_url(value['thumbnailUrl']))
+ media[key] = Image(url=Url(value['originalUrl'], value['extension']), thumbnail_url=self.create_media_url(value['thumbnailUrl']))
for key, value in data['body']['fileMap'].items():
- media[key] = File(url=MediaUrl(value['url'], value['extension']), name=value['name'])
+ media[key] = File(url=Url(value['url'], value['extension']), name=value['name'])
kwargs['text'] = text
kwargs['media'] = media
post = Post(**kwargs)
@@ -138,7 +138,7 @@ class FanboxUser(FanboxBase):
self.errored = True
return False
self.pages[num] = []
- for item in obj['items']:
+ for item in obj:
post = self.request_post(item['id'])
if post:
self.pages[num].append(post.unique_id)
@@ -169,12 +169,20 @@ class FanboxModule(Module):
self.parser.add_command('post', FanboxPost)
self.parser.add_command('user', FanboxUser)
self.parser.add_command('supporting', FanboxSupporting)
- self.headers['Accept-Encoding'] = 'gzip, deflate, br'
+ self.headers['Accept-Encoding'] = 'gzip, deflate, br, zstd'
self.headers['Accept-Language'] = 'en-US,en;q=0.5'
+ self.headers['Alt-Used'] = 'api.fanbox.cc'
self.headers['Origin'] = 'https://www.fanbox.cc'
self.headers['Referer'] = 'https://www.fanbox.cc/'
self.headers['User-Agent'] = USER_AGENT
self.cookies['FANBOXSESSID'] = sessid
+ self.cookies['privacy_policy_agreement'] = '7'
+ self.cookies['privacy_policy_notification'] = '0'
+ self.cookies['p_ab_id'] = '0'
+ self.cookies['p_ab_id_2'] = '6'
+ self.cookies['p_ab_d_id'] = '251960935'
+ self.cookies['cf_clearance'] = ''
+ self.cookies['__cf_bm'] = ''
def search(self, query: str, *extra_args: Any) -> Optional[Search]:
return self.parser.parse_query(query)
diff --git a/src/portal/py/modules/instagram.py b/src/portal/py/modules/instagram.py
index d107566..28f7def 100644
--- a/src/portal/py/modules/instagram.py
+++ b/src/portal/py/modules/instagram.py
@@ -1,10 +1,9 @@
import os
import log
-import email.utils
from typing import Optional, Any
from pathlib import Path
from base import USER_AGENT, Module, Search
-from post import MediaUrl, PostType, DateType, Post, User, Image, Video
+from post import Url, PostType, DateType, Post, User, Image, Video
from modules.common import get_current_utc_time
from instagrapi import Client
from instagrapi.exceptions import UserNotFound, LoginRequired, ChallengeRequired
@@ -19,13 +18,13 @@ class InstagramSearch(Search):
self.page: int = 0
self.cursor: Any = None
- def create_media_url(self, url: str) -> MediaUrl:
+ def create_media_url(self, url: str) -> Url:
question = url.find('?')
if question >= 0:
ext = url[:question].split('.')[-1]
else:
ext = url.split('.')[-1]
- return MediaUrl(url, ext)
+ return Url(url, ext)
def request_page(self, num: int) -> bool:
if num >= self.page + self.REACH_LIMIT:
@@ -34,6 +33,7 @@ class InstagramSearch(Search):
for i in range(self.page, num + 1):
try:
media, self.cursor = self.module.cl.user_medias_paginated_v1(self.pk, 0, end_cursor=self.cursor)
+ #media, self.cursor = self.module.cl.user_clips_paginated_v1(self.pk, amount=12, end_cursor=self.cursor)
except LoginRequired:
self.errored = True
return False
@@ -47,10 +47,12 @@ class InstagramSearch(Search):
kwargs = {}
kwargs['type'] = PostType.POST
unique_id = 'instagram:p:{}'.format(m.id)
+ #unique_id = 'instagram:r:{}'.format(m.id)
kwargs['unique_id'] = unique_id
kwargs['raw_responses'] = {}
kwargs['raw_responses']['tile'] = m.json()
kwargs['url'] = 'https://www.instagram.com/p/{}'.format(m.code)
+ #kwargs['url'] = 'https://www.instagram.com/reel/{}'.format(m.code)
kwargs['dates'] = {
DateType.CREATED: m.taken_at.timestamp(),
DateType.RETRIEVED: get_current_utc_time().timestamp()
diff --git a/src/portal/py/modules/patreon.py b/src/portal/py/modules/patreon.py
index 40c3255..f33ec18 100644
--- a/src/portal/py/modules/patreon.py
+++ b/src/portal/py/modules/patreon.py
@@ -1,9 +1,12 @@
-import json
+import log
import httpx
+import sys
from json import JSONDecodeError
from typing import Optional, Any
+from datetime import datetime
from base import USER_AGENT, Search, Module, Method, ParsedJson
-from post import MediaUrl, PostType, DateType, Post, User, Media, Image, Animation, Tag, TagType
+from modules.common import get_current_utc_time
+from post import Date, DateType, DateMeta, Post, User, Url, Media, File
from query_parser import QueryParser
BASE_URL = 'https://www.patreon.com/api'
@@ -26,7 +29,7 @@ class PatreonBase(Search):
return None
return obj
- def search_user(self, query: str) -> Optional[User]:
+ def search_user(self, query: str) -> tuple[Optional[User], Optional[str]]:
url = f'{BASE_URL}/search'
params = {
'q': query,
@@ -37,32 +40,76 @@ class PatreonBase(Search):
}
obj = self.check_api_response(self.module.do_request(Method.GET, url, params=params))
if not obj:
- return None
+ return None, None
for result in obj['data']:
+ username = result['attributes']['creator_name']
+ display_name = result['attributes']['name']
+ # Only accept exact matches.
+ if username.lower() != query.lower() and display_name.lower() != query.lower():
+ continue
prefix = result['id'].find('campaign_')
if prefix >= 0:
campaign = result['id'][prefix + len('campaign_'):]
unique_id = 'patreon:c:' + campaign
- user = User(unique_id=unique_id, username=result['attributes']['creator_name'], display_name=result['attributes']['name'])
- return user
- return None
+ user = User(unique_id=unique_id, username=username, display_name=display_name)
+ return user, campaign
+ return None, None
class PatreonUser(PatreonBase):
def __init__(self, userdata: Any, arg: str):
super().__init__(userdata)
self.username: str = arg
self.user: Optional[User] = None
- self.campaign: str
+ self.campaign: Optional[str] = None
+ self.cursor: Optional[str] = None
+ self.attachments: dict[str, Media] = {}
+
+ # post_type
+ # image_file: https://www.patreon.com/posts/emma-again-29391136
+ # text_only: https://www.patreon.com/posts/hello-update-29391184
+
+ def create_post(self, data: ParsedJson, hash: str) -> Optional[Post]:
+ kwargs = {}
+ unique_id = f'patreon:p:{data['id']}'
+ kwargs['unique_id'] = unique_id
+ kwargs['raw_responses'] = {}
+ kwargs['raw_responses']['api'] = hash
+ kwargs['url'] = data['attributes']['url']
+ published_at = datetime.strptime(data['attributes']['published_at'], "%Y-%m-%dT%H:%M:%S.%f%z").timestamp()
+ current_time = get_current_utc_time().timestamp()
+ kwargs['dates'] = [
+ Date(DateType.CREATED, (published_at, published_at), DateMeta.NONE),
+ Date(DateType.RETRIEVED, (current_time, current_time), DateMeta.NONE),
+ ]
+ kwargs['author'] = self.user
+ kwargs['title'] = data['attributes']['title']
+ kwargs['likes'] = data['attributes']['like_count']
+ kwargs['comments'] = data['attributes']['comment_count']
+ kwargs['text'] = data['attributes']['content']
+ post_type = data['attributes']['post_type']
+ media = {}
+ #if 'access_rules' in data['relationships']:
+ # if not (len(data['relationships']['access_rules']['data']) == 1 and data['relationships']['access_rules']['data'][0]['id'] == '9816126'):
+ # return None
+ if 'media' in data['relationships']:
+ for m in data['relationships']['media']['data']:
+ media[m['id']] = self.attachments[m['id']]
+ if 'attachments' in data['relationships']:
+ for a in data['relationships']['attachments']['data']:
+ media[a['id']] = self.attachments[a['id']]
+ kwargs['media'] = media
+ post = Post(**kwargs)
+ self.module.add_to_map(unique_id, post)
+ return post
def request_page(self, num: int) -> bool:
if num >= self.page + self.REACH_LIMIT:
return False
if not self.user:
- user = self.search_user(self.username)
- if not user:
+ self.user, self.campaign = self.search_user(self.username)
+ if not self.user:
self.errored = True
return False
- self.campaign = user.unique_id[len('patreon:c:'):]
request_satisfied = False
for i in range(self.page, num + 1):
url = f'{BASE_URL}/posts'
@@ -81,7 +128,7 @@ class PatreonUser(PatreonBase):
'filter[contains_exclusive_posts]': 'true',
'filter[is_draft]': 'false',
'filter[accessible_by_user_id]': self.module.user_id,
- 'sort': '-published_at',
+ 'sort': 'published_at',
'json-api-version': '1.0',
'json-api-use-default-includes': 'false'
}
@@ -91,8 +138,27 @@ class PatreonUser(PatreonBase):
if not obj:
self.errored = True
return False
- print(json.dumps(obj, indent=4))
- return True
+ hash = self.module.add_raw_response(obj)
+ self.cursor = obj['meta']['pagination']['cursors']['next']
+ self.pages[i] = []
+ for inc in obj['included']:
+ if inc['type'] == 'attachment':
+ self.attachments[inc['id']] = File(url=Url(inc['attributes']['url']), name=inc['attributes']['name'])
+ elif inc['type'] == 'media':
+ self.attachments[inc['id']] = Media(url=Url(inc['attributes']['download_url']), thumbnail_url=Url(inc['attributes']['image_urls']['thumbnail']))
+ else:
+ log.error(f'Unhandled inc type {inc['type']}.')
+ for entry in obj['data']:
+ if entry['type'] != 'post':
+ log.error(f'Unhandled entry type {entry['type']}.')
+ continue
+ post = self.create_post(entry, hash)
+ if post:
+ self.pages[i].append(post.unique_id)
+ self.page += 1
+ if i == num:
+ request_satisfied = True
+ return request_satisfied
class PatreonModule(Module):
def __init__(self, session_id: str, uuid: str, user_id: str):
@@ -100,7 +166,8 @@ class PatreonModule(Module):
self.user_id: str = user_id
self.parser: QueryParser = QueryParser(self, 'user')
self.parser.add_command('user', PatreonUser)
- self.headers['Accept-Encoding'] = 'gzip, deflate, br'
+ self.headers['User-Agent'] = USER_AGENT
+ self.headers['Accept-Encoding'] = 'gzip, deflate, br, zstd'
self.headers['Accept-Language'] = 'en-US,en;q=0.5'
self.headers['x-patreon-uuid'] = uuid
self.cookies['patreon_locale_code'] = 'en-US'
@@ -116,9 +183,4 @@ class PatreonModule(Module):
post = self.unique_id_map[unique_id]
if key not in post.media:
return None
- return {
- 'urls': [post.media[key].url],
- 'headers': {
- 'User-Agent': USER_AGENT
- }
- }
+ return { 'urls': [post.media[key].url], 'headers': self.headers, 'cookies': self.cookies }
diff --git a/src/portal/py/modules/pixiv_app.py b/src/portal/py/modules/pixiv_app.py
index c84030d..ef5886d 100644
--- a/src/portal/py/modules/pixiv_app.py
+++ b/src/portal/py/modules/pixiv_app.py
@@ -1,10 +1,10 @@
import json
from typing import Optional, Any
from base import USER_AGENT, Search, Module, ParsedJson
-from post import MediaUrl, PostType, DateType, User, Post, Image, Tag, TagType
+from post import Url, PostType, DateType, User, Post, Image, Tag, TagType
from query_parser import QueryParser
-from pixivpy3 import *
-from modules.common import get_current_utc_time, parse_pixiv_date
+from pixivpy3 import AppPixivAPI
+from modules.common import get_current_utc_time, reformat_pixiv_date
class PixivAppBase(Search):
def __init__(self, userdata: Any):
@@ -33,8 +33,8 @@ class PixivAppBase(Search):
self.page += 1
return request_satisfied
- def create_media_url(self, url: str) -> MediaUrl:
- return MediaUrl(url, url.split('.')[-1])
+ def create_media_url(self, url: str) -> Url:
+ return Url(url, url.split('.')[-1])
def create_tag(self, data: ParsedJson) -> Tag:
tag = Tag(type=TagType.GENERAL, name=data['name'])
@@ -58,7 +58,7 @@ class PixivAppBase(Search):
kwargs['raw_responses']['api'] = json.dumps(data)
kwargs['url'] = 'https://www.pixiv.net/en/artworks/{}'.format(data['id'])
kwargs['dates'] = {
- DateType.CREATED: parse_pixiv_date(data['create_date']).timestamp(),
+ DateType.CREATED: reformat_pixiv_date(data['create_date']).timestamp(),
DateType.RETRIEVED: get_current_utc_time().timestamp()
}
user = User(unique_id=f'pixiv:u:{data['user']['id']}', username=data['user']['account'], display_name=data['user']['name'])
diff --git a/src/portal/py/modules/pixiv_web.py b/src/portal/py/modules/pixiv_web.py
index be68b7e..7dbb8bb 100644
--- a/src/portal/py/modules/pixiv_web.py
+++ b/src/portal/py/modules/pixiv_web.py
@@ -5,9 +5,9 @@ import urllib.parse
from typing import Optional, Any
from json.decoder import JSONDecodeError
from base import USER_AGENT, Search, Module, Method, ParsedJson
-from post import MediaUrl, PostType, DateType, Post, User, Media, Image, Animation, Tag, TagType
+from post import Url, PostType, DateType, Post, User, Media, Image, Animation, Tag, TagType
from query_parser import QueryParser
-from modules.common import parse_pixiv_date, get_current_utc_time
+from modules.common import reformat_pixiv_date, get_current_utc_time
BASE_URL = 'https://www.pixiv.net/ajax'
LANG = 'en'
@@ -30,8 +30,8 @@ class PixivWebBase(Search):
return None
return obj['body']
- def create_media_url(self, url: str) -> MediaUrl:
- return MediaUrl(url, url.split('.')[-1])
+ def create_media_url(self, url: str) -> Url:
+ return Url(url, url.split('.')[-1])
def parse_pages(self, data: ParsedJson) -> dict[str, Media]:
media = {}
@@ -45,7 +45,7 @@ class PixivWebBase(Search):
animation.frames.append((frame['file'], frame['delay']))
return animation
- def seek_profile_picture_url(self, user_illusts: ParsedJson) -> Optional[MediaUrl]:
+ def seek_profile_picture_url(self, user_illusts: ParsedJson) -> Optional[Url]:
for illust in user_illusts.values():
if illust is None:
continue
@@ -71,14 +71,15 @@ class PixivWebBase(Search):
kwargs['raw_responses'] = {}
kwargs['raw_responses']['api'] = json.dumps(data)
kwargs['url'] = f'https://www.pixiv.net/{LANG}/artworks/{data['illustId']}'
- create_date = parse_pixiv_date(data['createDate']).timestamp()
+ create_date = reformat_pixiv_date(data['createDate']).timestamp()
+ current_date = get_current_utc_time().timestamp()
kwargs['dates'] = {
- DateType.CREATED: create_date,
- DateType.RETRIEVED: get_current_utc_time().timestamp()
+ DateType.CREATED: (create_date, create_date, 0),
+ DateType.RETRIEVED: (current_date, current_date, 0)
}
- upload_date = parse_pixiv_date(data['uploadDate']).timestamp()
+ upload_date = reformat_pixiv_date(data['uploadDate']).timestamp()
if upload_date != create_date:
- kwargs['dates'][DateType.EDITED] = upload_date
+ kwargs['dates'][DateType.EDITED] = (upload_date, upload_date, 0)
user = User(unique_id=f'pixiv:u:{data['userId']}', username=data['userAccount'], display_name=data['userName'])
profile_picture_url = self.seek_profile_picture_url(data['userIllusts'])
if profile_picture_url:
@@ -242,10 +243,10 @@ class PixivWebModule(Module):
self.parser.add_command('illust', PixivWebIllust)
self.parser.add_command('user', PixivWebUser)
self.parser.add_command('following', PixivWebFollowing)
- self.headers['Accept-Encoding'] = 'gzip, deflate, br' # add 'zstd' when httpx update
+ self.headers['User-Agent'] = USER_AGENT
+ self.headers['Accept-Encoding'] = 'gzip, deflate, br, zstd'
self.headers['Accept-Language'] = 'en-US,en;q=0.5'
self.headers['Referer'] = 'https://www.pixiv.net/'
- self.headers['User-Agent'] = USER_AGENT
self.headers['x-user-id'] = user_id
self.cookies['PHPSESSID'] = sessid
diff --git a/src/portal/py/modules/twitter.py b/src/portal/py/modules/twitter.py
index 2996d0e..386bc5b 100644
--- a/src/portal/py/modules/twitter.py
+++ b/src/portal/py/modules/twitter.py
@@ -3,7 +3,7 @@ import http.cookiejar
from typing import Optional, Any, Iterator
from datetime import timezone
from base import USER_AGENT, Search, Module
-from post import MediaUrl, PostType, DateType, PostRef, Post, User, Media, Image, Video
+from post import Url, PostType, DateType, PostRef, Post, User, Media, Image, Video
from query_parser import QueryParser
from modules.common import get_current_utc_time
@@ -26,7 +26,7 @@ class TwitterScrapeBase(Search):
self.iterator: Iterator[Tweet | TweetRef | Tombstone]
self.page: int = 0
- def create_media_url(self, url: str) -> MediaUrl:
+ def create_media_url(self, url: str) -> Url:
format = url.find('format=')
if format >= 0:
ext = url[format + 7:format + 10]
@@ -35,7 +35,7 @@ class TwitterScrapeBase(Search):
if question >= 0:
url = url[0:question]
ext = url.split('.')[-1]
- return MediaUrl(url, ext)
+ return Url(url, ext)
def parse_media(self, data: list[twitter.Medium]) -> dict[str, Media]:
media = {}
diff --git a/src/portal/py/modules/youtube.py b/src/portal/py/modules/youtube.py
index d14dfe8..f809ff6 100644
--- a/src/portal/py/modules/youtube.py
+++ b/src/portal/py/modules/youtube.py
@@ -1,7 +1,8 @@
import log
+import json
from typing import Optional, Any
from base import Search, Module, ParsedJson
-from post import MediaUrl, Post, PostType, Media, Video
+from post import Url, Post, PostType, Media, Video
from query_parser import QueryParser
from yt_dlp import YoutubeDL
@@ -25,65 +26,53 @@ ydl_opts = {
'quiet': False,
'logger': YDLLogger(),
'cachedir': False,
-# 'cookiefile': '',
+ 'socket_timeout': 10
+# 'cookiefile': ''
}
-ydl = YoutubeDL(ydl_opts)
-
-def get_playback_url(data, video=True):
- if 'entries' in data:
- if len(data['entries']) == 0:
- return None
- data = data['entries'][0]
+# https://github.com/yt-dlp/yt-dlp/issues/4103
- if 'formats' not in data:
- if 'url' in data:
- return data['url']
- return None
+ydl = YoutubeDL(ydl_opts)
- # Filter out hls temporarily.
- data['formats'] = list(filter(lambda f: not f['protocol'].startswith('m3u8'), data['formats']))
+def get_playback_url(data, video=True) -> Optional[str]:
+ data['formats'] = list(filter(
+ lambda f: not f['protocol'].startswith('m3u8'), data['formats']))
if len(data['formats']) == 0:
return None
- url = data['formats'][0]['url']
+ result = data['formats'][-1]['url']
- # audio_ext?
- has_audio = list(filter(lambda f: 'acodec' not in f or f['acodec'] != 'none', data['formats']))
+ has_audio = list(filter(lambda f:
+ (f['acodec'] and f['acodec'] != 'none') or
+ (f['audio_ext'] and f['audio_ext'] != 'none') or
+ (f['abr']), data['formats']))
if video:
- if len(has_audio) > 0:
- data['formats'] = has_audio
- try:
- data['formats'] = list(filter(lambda f: 'quality' in f, data['formats']))
- url = max(data['formats'], key=lambda f: f['quality'])['url']
- except:
- pass
+ data['formats'] = list(filter(lambda f:
+ (f['vcodec'] and f['vcodec'] != 'none') or
+ (f['video_ext'] and f['video_ext'] != 'none') or
+ (f['vbr']), has_audio))
else:
- if len(has_audio) == 0:
- return None
- data['formats'] = has_audio
- try:
- audio_only = list(filter(lambda f: f['vcodec'] == 'none', data['formats']))
- if len(audio_only) > 0:
- data['formats'] = audio_only
- else:
- data['formats'] = list(filter(lambda f: f['ext'] in ['mp4'], data['formats']))
- except:
- pass
- try:
- data['formats'] = list(filter(lambda f: 'abr' in f, data['formats']))
- url = max(data['formats'], key=lambda f: f['abr'])['url']
- except:
- pass
+ data['formats'] = list(filter(lambda f:
+ (f['vcodec'] and f['vcodec'] == 'none') and
+ (f['video_ext'] and f['video_ext'] == 'none') and
+ (not f['vbr']), has_audio))
+
+ if len(data['formats']) > 0:
+ selection = max(data['formats'], key=lambda f: 0 if 'quality' not in f else f['quality'])
+ log.info(json.dumps(selection, indent=4))
+ result = selection['url']
+ else:
+ log.warn('Defaulting in Youtube get_playback_url().')
- return url
+ return result
class YoutubeBase(Search):
def __init__(self, userdata: Any):
super().__init__()
self.module: YoutubeModule = userdata
+ self.guessed_index: int = 0
def get_info(self) -> Optional[ParsedJson]:
return None
@@ -92,11 +81,25 @@ class YoutubeBase(Search):
info = self.get_info()
if not info:
return False
- url = get_playback_url(info)
+ if 'entries' in info:
+ if len(info['entries']) == 0:
+ return False
+ if len(info['entries']) <= self.guessed_index:
+ self.guessed_index = 0
+ info = info['entries'][self.guessed_index]
+ if 'formats' not in info and 'url' in info:
+ self.link = info['url']
+ info = self.get_info()
+ if not info:
+ return False
+ if 'direct' in info and info['direct']:
+ url = info['url']
+ else:
+ url = get_playback_url(info, video=False)
if not url:
return False
- media: dict[str, Media] = { '0': Video(url=MediaUrl(url=url)) }
unique_id = f'youtube:v:{info['id']}'
+ media: dict[str, Media] = { '0': Video(url=Url(url)) }
self.module.add_to_map(unique_id, Post(type=PostType.POST, unique_id=unique_id, url='', title=info['title'], text='', media=media))
self.pages[num] = [unique_id]
return True
@@ -107,7 +110,13 @@ class YoutubeSearch(YoutubeBase):
self.query: str = arg
def get_info(self) -> Optional[ParsedJson]:
- return ydl.extract_info(f'ytsearch1:{self.query}', download=False)
+ ydl.params['extract_flat'] = False
+ try:
+ info = ydl.extract_info(f'ytsearch1:{self.query}', download=False)
+ except Exception as e:
+ log.error(repr(e))
+ return None
+ return info
class YoutubeLink(YoutubeBase):
def __init__(self, userdata: Any, arg: str):
@@ -115,7 +124,24 @@ class YoutubeLink(YoutubeBase):
self.link: str = arg
def get_info(self) -> Optional[ParsedJson]:
- return ydl.extract_info(self.link, download=False)
+ index = self.link.find('&index=')
+ if index >= 0:
+ sub = self.link[index + 7:]
+ end = sub.find('&')
+ if end >= 0:
+ sub = sub[:end]
+ try:
+ self.guessed_index = int(sub, 10) - 1
+ except ValueError:
+ pass
+ log.info(f'Guessed index: {self.guessed_index}')
+ ydl.params['extract_flat'] = 'in_playlist'
+ try:
+ info = ydl.extract_info(self.link, download=False)
+ except Exception as e:
+ log.error(repr(e))
+ return None
+ return info
class YoutubeModule(Module):
def __init__(self):
diff --git a/src/portal/py/post.py b/src/portal/py/post.py
index 8149e4c..76e82b9 100644
--- a/src/portal/py/post.py
+++ b/src/portal/py/post.py
@@ -1,22 +1,38 @@
import dataclasses
from dataclasses import field
from typing import Optional, Any
-from enum import Enum
+from enum import IntEnum, IntFlag
from json import JSONEncoder
-class PostType(int, Enum):
+'''
+List of Things:
+ - Social media posts
+ - Albums
+ - Loose files
+'''
+
+class PostType(IntEnum):
UNKNOWN = 0
POST = 1
REPOST = 2
PREVIEW = 3
TOMBSTONE = 4
-class DateType(int, Enum):
- CREATED = 0
- EDITED = 1
- RETRIEVED = 2
+class DateType(IntEnum):
+ UNKNOWN = 0
+ CREATED = 1
+ EDITED = 2
+ RETRIEVED = 3
+
+class DateMeta(IntFlag):
+ NONE = 0
+ ESTIMATE = 1
+ GUESS = 1 << 1
+ LOOSE_PRECISION = 1 << 2
+ EDITED_AT_UNKNOWN_TIME = 1 << 3
+ ASSUMED_TYPE = 1 << 4
-class MediaType(int, Enum):
+class MediaType(IntEnum):
UNKNOWN = 0
FILE = 1
AUDIO = 2
@@ -25,7 +41,7 @@ class MediaType(int, Enum):
VIDEO_SPLIT = 5
ANIMATION = 6
-class TagType(int, Enum):
+class TagType(IntEnum):
UNKNOWN = 0
GENERAL = 1
ARTIST = 2
@@ -37,15 +53,33 @@ class TagType(int, Enum):
def df(c: Any) -> Any:
return field(default_factory=lambda: c)
+def default_url_ext_guess(url):
+ # TODO: List of known extensions.
+ question = url.rfind('?')
+ if question >= 0:
+ guess = url[0:question].rsplit('.')[-1]
+ else:
+ guess = url.rsplit('.')[-1]
+ return guess
+
@dataclasses.dataclass
-class MediaUrl():
- url: str = ''
- ext: str = 'unknown'
+class Url():
+ url: str
+ ext: str
+
+ def __init__(self, url: str, ext: str = 'unknown'):
+ self.url = url
+ ext_guess = default_url_ext_guess(url)
+ if ext_guess:
+ self.ext = ext_guess
+ else:
+ self.ext = ext
@dataclasses.dataclass
class Media():
type: MediaType = MediaType.UNKNOWN
- url: MediaUrl = df(MediaUrl())
+ url: Url = df(Url(''))
+ thumbnail_url: Url = df(Url(''))
@dataclasses.dataclass
class File(Media):
@@ -59,24 +93,20 @@ class Audio(Media):
@dataclasses.dataclass
class Image(Media):
type: MediaType = MediaType.IMAGE
- thumbnail_url: MediaUrl = df(MediaUrl())
@dataclasses.dataclass
class Video(Media):
type: MediaType = MediaType.VIDEO
- thumbnail_url: MediaUrl = df(MediaUrl())
@dataclasses.dataclass
class VideoSplit(Media):
type: MediaType = MediaType.VIDEO_SPLIT
- audio_url: MediaUrl = df(MediaUrl())
- subtitle_url: MediaUrl = df(MediaUrl())
- thumbnail_url: MediaUrl = df(MediaUrl())
+ audio_url: Url = df(Url(''))
+ subtitle_url: Url = df(Url(''))
@dataclasses.dataclass
class Animation(Media):
type: MediaType = MediaType.ANIMATION
- thumbnail_url: MediaUrl = df(MediaUrl())
frames: list[tuple[str, int]] = df([])
@dataclasses.dataclass
@@ -85,31 +115,59 @@ class Tag():
name: str = ''
alts: dict[str, str] = df({})
+'''
+Date(DateType.CREATED, (1396411200.0, 1396756800.0), DateMeta.ESTIMATE)
+ - Post is estimated to have been created at some point between April 2nd and 6th 2014.
+
+Date(DateType.EDITED, (1396411200.0, 1550247862.0), DateMeta.EDITED_AT_UNKNOWN_TIME)
+ - We have no clue when the post was edited, but it had to have been between the time it was created and now.
+ - EDITED_AT_UNKNOWN_TIME should be used when either of CREATED and/or RETRIEVED are used as placeholders.
+ It can be combined with GUESS or ESTIMATE if, for example, there is some indication about the end of the range.
+
+Date(DateType.RETRIEVED, (1550247862.0, 1550247862.0), DateMeta.NONE)
+ - Time of processing this post. Will likely be slightly after it was actually retrived but that is negligible.
+ - RETRIEVED can still have a range and meta if applicable.
+
+Date(DateType.CREATED, (1483228800.0, 1546300800.0), DateMeta.LOOSE_PRECISION)
+ - Date is exactly between "01 Jan 2017 12:00:00 AM UTC" and "01 Jan 2019 12:00:00 AM UTC".
+ Combined with LOOSE_PRECISION, the application will assume a meaning of "between 2017 and 2019".
+
+Date(DateType.CREATED, (1145059200.0, 1145059200.0), DateMeta.ASSUMED_TYPE)
+ - The date of April 15th 2006 is associated with the resource. We are assuming that's the creation date.
+'''
+@dataclasses.dataclass
+class Date():
+ type: DateType = DateType.UNKNOWN
+ range: tuple[float, float] = (0.0, 0.0)
+ meta: DateMeta = DateMeta.NONE
+ note: str = ''
+
@dataclasses.dataclass
class User():
unique_id: str = ''
username: str = ''
display_name: str = ''
- profile_picture_url: MediaUrl = df(MediaUrl())
+ profile_picture_url: Url = df(Url(''))
@dataclasses.dataclass
class PostRef():
unique_id: str = ''
-# V4 Ideas:
-# - Date estimate and range
-# - Edited but unknown when
-# - Generally an estimate/guess
-# - Formated text/body
-
+'''
+V4:
+ - Improved date.
+ - Ends when all current modules are ported and/or completed.
+V5:
+ - Formatted body/text.
+'''
@dataclasses.dataclass
class Post():
- version: int = 3
+ version: int = 4
type: PostType = PostType.UNKNOWN
unique_id: str = ''
raw_responses: dict[str, str] = df({})
url: str = ''
- dates: dict[DateType, float] = df({})
+ dates: list[Date] = df([])
author: User = df(User())
title: str = ''
text: str = ''
@@ -128,8 +186,4 @@ class Post():
class PostEncoder(JSONEncoder):
def default(self, o):
- if isinstance(o, Enum):
- return o.value
- if dataclasses.is_dataclass(o):
- return dataclasses.asdict(o)
return o.__dict__
diff --git a/src/portal/py/test.py b/src/portal/py/test.py
new file mode 100644
index 0000000..c5687b1
--- /dev/null
+++ b/src/portal/py/test.py
@@ -0,0 +1 @@
+import tests.test
diff --git a/src/portal/py/tests/__init__.py b/src/portal/py/tests/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/src/portal/py/tests/__init__.py
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'))
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 23:24:36 +0000