summaryrefslogtreecommitdiff
path: root/src/portal/py/modules
diff options
context:
space:
mode:
Diffstat (limited to 'src/portal/py/modules')
-rw-r--r--src/portal/py/modules/__init__.py22
-rw-r--r--src/portal/py/modules/common.py9
-rw-r--r--src/portal/py/modules/fanbox.py192
-rw-r--r--src/portal/py/modules/instagram.py154
-rw-r--r--src/portal/py/modules/pixiv_app.py201
-rw-r--r--src/portal/py/modules/pixiv_web.py263
-rw-r--r--src/portal/py/modules/searx.py95
-rw-r--r--src/portal/py/modules/twitter.py190
-rw-r--r--src/portal/py/modules/twitter_api.py269
-rw-r--r--src/portal/py/modules/youtube.py133
10 files changed, 1528 insertions, 0 deletions
diff --git a/src/portal/py/modules/__init__.py b/src/portal/py/modules/__init__.py
new file mode 100644
index 0000000..480660c
--- /dev/null
+++ b/src/portal/py/modules/__init__.py
@@ -0,0 +1,22 @@
+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.fanbox import FanboxModule
+from modules.instagram import InstagramModule
+
+ALL_MODULES = {
+ 'youtube': (YoutubeModule(), []),
+# '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), []),
+# '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), [])
+}
diff --git a/src/portal/py/modules/common.py b/src/portal/py/modules/common.py
new file mode 100644
index 0000000..62c3d5d
--- /dev/null
+++ b/src/portal/py/modules/common.py
@@ -0,0 +1,9 @@
+from datetime import datetime, timezone
+
+def parse_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")
+
+def get_current_utc_time() -> datetime:
+ return datetime.now(timezone.utc)
diff --git a/src/portal/py/modules/fanbox.py b/src/portal/py/modules/fanbox.py
new file mode 100644
index 0000000..01e66c3
--- /dev/null
+++ b/src/portal/py/modules/fanbox.py
@@ -0,0 +1,192 @@
+import json
+import httpx
+from typing import Optional, Any
+from json.decoder import JSONDecodeError
+from base import Search, Module, Method, ParsedJson
+from post import MediaUrl, PostType, DateType, Post, User, Image, File, Tag
+from query_parser import QueryParser
+from modules.common import parse_pixiv_date, get_current_utc_time
+
+BASE_URL = "https://api.fanbox.cc"
+
+class FanboxBase(Search):
+ def __init__(self, userdata: Any):
+ super().__init__()
+ self.module: FanboxModule = userdata
+
+ def check_api_response(self, response: Optional[httpx.Response]) -> Optional[ParsedJson]:
+ if not response:
+ return None
+ try:
+ obj = response.json()
+ except JSONDecodeError:
+ return None
+ return obj['body']
+
+ def create_media_url(self, url: str) -> MediaUrl:
+ return MediaUrl(url, url.split('.')[-1])
+
+ def create_post(self, data: ParsedJson) -> Optional[Post]:
+ kwargs = {}
+ post_id = data['id']
+ unique_id = f'fanbox:p:{post_id}'
+ kwargs['unique_id'] = unique_id
+ kwargs['raw_responses'] = {}
+ kwargs['raw_responses']['api'] = json.dumps(data)
+ publish_date = parse_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()
+ 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'])
+ user.profile_picture_url = self.create_media_url(data['user']['iconUrl'])
+ kwargs['author'] = user
+ kwargs['title'] = data['title']
+ kwargs['likes'] = data['likeCount']
+ kwargs['comments'] = data['commentCount']
+ kwargs['tags'] = [Tag(name=tag) for tag in data['tags']]
+ if data['isRestricted']:
+ kwargs['type'] = PostType.PREVIEW
+ else:
+ kwargs['type'] = PostType.POST
+ text = ''
+ media = {}
+ 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']))
+ 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'])
+ elif data['type'] == 'article':
+ for block in data['body']['blocks']:
+ if block['type'] == 'p':
+ text += block['text'] + '\n'
+ elif block['type'] == 'image':
+ text += f'image::{block['imageId']}[]' + '\n'
+ 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']))
+ for key, value in data['body']['fileMap'].items():
+ media[key] = File(url=MediaUrl(value['url'], value['extension']), name=value['name'])
+ kwargs['text'] = text
+ kwargs['media'] = media
+ post = Post(**kwargs)
+ self.module.add_to_map(unique_id, post)
+ return post
+
+ def request_post(self, post_id: int) -> Optional[Post]:
+ url = f'{BASE_URL}/post.info?postId={post_id}'
+ obj = self.check_api_response(self.module.do_request(Method.GET, url))
+ if not obj:
+ return None
+ return self.create_post(obj)
+
+ def create_user(self, data: ParsedJson) -> User:
+ unique_id = f'fanbox:u:{data['id']}'
+ user = User(unique_id=unique_id, username=data['creatorId'], display_name=data['user']['name'])
+ user.profile_picture_url = self.create_media_url(data['user']['iconUrl'])
+ self.module.add_to_map(unique_id, user)
+ return user
+
+class FanboxPost(FanboxBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ try:
+ self.id: int = int(arg)
+ except ValueError:
+ self.errored = True
+
+ def request_page(self, num: int) -> bool:
+ post = self.request_post(self.id)
+ if not post:
+ self.errored = True
+ return False
+ self.pages[num] = [post.unique_id]
+ self.completed = True
+ return True
+
+class FanboxUser(FanboxBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ self.id = arg
+ self.pagination = []
+
+ def request_pagination(self) -> bool:
+ url = f'{BASE_URL}/post.paginateCreator?creatorId={self.id}'
+ obj = self.check_api_response(self.module.do_request(Method.GET, url))
+ if not obj:
+ return False
+ for link in obj:
+ self.pagination.append(link)
+ return True
+
+ def request_page(self, num: int) -> bool:
+ if len(self.pagination) == 0:
+ if not self.request_pagination():
+ self.errored = True
+ return False
+ if num >= len(self.pagination):
+ return False
+ obj = self.check_api_response(self.module.do_request(Method.GET, self.pagination[num]))
+ if not obj:
+ self.errored = True
+ return False
+ self.pages[num] = []
+ for item in obj['items']:
+ post = self.request_post(item['id'])
+ if post:
+ self.pages[num].append(post.unique_id)
+ if len(self.pages) == len(self.pagination):
+ self.completed = True
+ return True
+
+class FanboxSupporting(FanboxBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+
+ def request_page(self, num: int) -> bool:
+ url = f'{BASE_URL}/plan.listSupporting'
+ obj = self.check_api_response(self.module.do_request(Method.GET, url))
+ if not obj:
+ self.errored = True
+ return False
+ self.pages[num] = []
+ for user in obj:
+ self.pages[num].append(self.create_user(user).unique_id)
+ self.completed = True
+ return True
+
+class FanboxModule(Module):
+ def __init__(self, sessid: str):
+ super().__init__()
+ self.parser: QueryParser = QueryParser(self, 'post')
+ 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-Language'] = 'en-US,en;q=0.5'
+ self.headers['Origin'] = 'https://www.fanbox.cc'
+ self.headers['Referer'] = 'https://www.fanbox.cc/'
+ self.headers['User-Agent'] = 'Mozilla/5.0 (Windows NT 10.0; rv:122.0) Gecko/20100101 Firefox/122.0'
+ self.cookies['FANBOXSESSID'] = sessid
+
+ def search(self, query: str, *extra_args: Any) -> Optional[Search]:
+ return self.parser.parse_query(query)
+
+ def get_download(self, unique_id: str, key: str) -> Optional[dict[str, Any]]:
+ if unique_id not in self.unique_id_map:
+ return None
+ post = self.unique_id_map[unique_id]
+ if key not in post.media:
+ return None
+ return {
+ 'urls': [post.media[key].url],
+ 'headers': {},
+ 'cookies': {}
+ }
diff --git a/src/portal/py/modules/instagram.py b/src/portal/py/modules/instagram.py
new file mode 100644
index 0000000..325caf6
--- /dev/null
+++ b/src/portal/py/modules/instagram.py
@@ -0,0 +1,154 @@
+import os
+import log
+import email.utils
+from typing import Optional, Any
+from pathlib import Path
+from base import Module, Search
+from post import MediaUrl, 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
+
+class InstagramSearch(Search):
+ REACH_LIMIT = 3
+
+ def __init__(self, module: Any, pk: str):
+ super().__init__()
+ self.module: InstagramModule = module
+ self.pk: str = pk
+ self.page: int = 0
+ self.cursor: Any = None
+
+ def create_media_url(self, url: str) -> MediaUrl:
+ question = url.find('?')
+ if question >= 0:
+ ext = url[:question].split('.')[-1]
+ else:
+ ext = url.split('.')[-1]
+ return MediaUrl(url, ext)
+
+ def request_page(self, num: int) -> bool:
+ if num >= self.page + self.REACH_LIMIT:
+ return False
+ request_satisfied = False
+ 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)
+ except LoginRequired:
+ self.errored = True
+ return False
+ except ChallengeRequired:
+ self.errored = True
+ return False
+ if not self.cursor:
+ self.completed = True
+ self.pages[i] = []
+ for m in media:
+ kwargs = {}
+ kwargs['type'] = PostType.POST
+ unique_id = 'instagram:p:{}'.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['dates'] = {
+ DateType.CREATED: m.taken_at.timestamp(),
+ DateType.RETRIEVED: get_current_utc_time().timestamp()
+ }
+ user = User(unique_id='instagram:u:{}'.format(m.user.pk))
+ if m.user.username:
+ user.username = m.user.username
+ if m.user.full_name:
+ user.display_name = m.user.full_name
+ if m.user.profile_pic_url:
+ user.profile_picture_url = self.create_media_url(str(m.user.profile_pic_url))
+ kwargs['author'] = user
+ kwargs['title'] = m.title
+ kwargs['text'] = m.caption_text
+ kwargs['likes'] = m.like_count
+ kwargs['comments'] = m.comment_count
+ media = {}
+ if m.media_type == 8: # Album
+ for k, r in enumerate(m.resources):
+ if r.media_type == 1: # Photo
+ media[str(k)] = Image(url=self.create_media_url(str(r.thumbnail_url)))
+ elif r.media_type == 2: # Video
+ media[str(k)] = Video(url=self.create_media_url(str(r.video_url)), thumbnail_url=self.create_media_url(str(r.thumbnail_url)))
+ elif m.media_type == 2: # Video
+ kwargs['views'] = m.play_count
+ media[str(0)] = Video(url=self.create_media_url(str(m.video_url)), thumbnail_url=self.create_media_url(str(m.thumbnail_url)))
+ elif m.media_type == 1: # Photo
+ candidates = sorted(m.image_versions2['candidates'], key=lambda x: x['width'], reverse=True)
+ media[str(0)] = Image(url=self.create_media_url(candidates[0]['url']), thumbnail_url=self.create_media_url(candidates[1]['url']))
+ kwargs['media'] = media
+ post = Post(**kwargs)
+ self.module.add_to_map(unique_id, post)
+ self.pages[i].append(unique_id)
+ self.page += 1
+ if i == num:
+ request_satisfied = True
+ return request_satisfied
+
+class InstagramModule(Module):
+ def __init__(self, user_agent: str, settings: Path, session_id: str):
+ super().__init__()
+ self.cl: Client = Client()
+ # https://specdevice.com/showspec.php?id=c318-0c39-0033-c5870033c587
+ device_set = {
+ 'app_version': '324.0.0.0.16',
+ 'android_version': 33,
+ 'android_release': '13.0.0',
+ 'dpi': '476dpi',
+ 'resolution': '1440x3120',
+ 'manufacturer': 'Google',
+ 'device': 'cheetah',
+ 'model': 'Pixel 7 Pro',
+ 'cpu': 'cheetah',
+ 'version_code': '9981770'
+ }
+ user_agent = f'Instagram {device_set['app_version']} Android ({device_set['android_version']}/{device_set['android_release']}; {device_set['dpi']}; {device_set['resolution']}; {device_set['manufacturer']}; {device_set['device']}; {device_set['model']}; {device_set['cpu']}; en_US; {device_set['version_code']})'
+ self.cl.set_country('US')
+ self.cl.set_country_code(1) # Phone code
+ self.cl.set_locale('en_US')
+ self.cl.set_timezone_offset(-14400) # New_York GMT-4
+ self.cl.set_user_agent(user_agent)
+ self.cl.set_device(device = device_set)
+ self.settings: Path = settings
+ self.session_id: str = session_id
+
+ def init(self) -> bool:
+ if os.path.exists(self.settings):
+ self.cl.load_settings(self.settings)
+ if not self.cl.login_by_sessionid(self.session_id):
+ return False
+ #if not self.cl.login(self.username, self.password):
+ # return False
+ self.cl.dump_settings(self.settings)
+ return True
+
+ def search(self, query: str, *extra_args: Any) -> Optional[Search]:
+ try:
+ user = self.cl.user_info_by_username_v1(query)
+ except UserNotFound:
+ log.warn('User not found.')
+ return None
+ except LoginRequired:
+ log.error('Fetching user failed: Login required.')
+ return None
+ except ChallengeRequired:
+ log.error('Fetching user failed: Challenge required.')
+ return None
+ return InstagramSearch(self, user.pk)
+
+ def get_download(self, unique_id: str, key: str) -> Optional[dict[str, Any]]:
+ if unique_id not in self.unique_id_map:
+ return None
+ post = self.unique_id_map[unique_id]
+ if key not in post.media:
+ return None
+ return {
+ 'urls': [post.media[key].url],
+ 'headers': {
+ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; rv:122.0) Gecko/20100101 Firefox/122.0'
+ }
+ }
diff --git a/src/portal/py/modules/pixiv_app.py b/src/portal/py/modules/pixiv_app.py
new file mode 100644
index 0000000..8be3f9c
--- /dev/null
+++ b/src/portal/py/modules/pixiv_app.py
@@ -0,0 +1,201 @@
+import json
+from typing import Optional, Any
+from base import Search, Module, ParsedJson
+from post import MediaUrl, 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
+
+class PixivAppBase(Search):
+ def __init__(self, userdata: Any):
+ super().__init__()
+ self.module: PixivAppModule = userdata
+ self.page: int = 0
+ self.next_qs: Optional[dict[str, Any]] = None
+
+ def api_request_page(self, num: int) -> tuple[bool, Optional[dict[str, Any]]]:
+ return False, None
+
+ def request_page(self, num: int) -> bool:
+ request_satisfied = False
+ for i in range(self.page, num + 1):
+ if not self.next_qs:
+ break
+ if i in self.pages:
+ continue
+ self.pages[i] = []
+ success, self.next_qs = self.api_request_page(i)
+ if not success:
+ self.errored = True
+ break
+ if i == num:
+ request_satisfied = True
+ self.page += 1
+ return request_satisfied
+
+ def create_media_url(self, url: str) -> MediaUrl:
+ return MediaUrl(url, url.split('.')[-1])
+
+ def create_tag(self, data: ParsedJson) -> Tag:
+ tag = Tag(type=TagType.GENERAL, name=data['name'])
+ if data['translated_name']:
+ tag.alts['en'] = data['translated_name']
+ return tag
+
+ def create_post(self, data: ParsedJson) -> Post:
+ if len(data['meta_pages']) == 0:
+ data['meta_pages'].append({
+ 'image_urls': {
+ 'medium': data['image_urls']['medium'],
+ 'original': data['meta_single_page']['original_image_url']
+ }
+ })
+ kwargs = {}
+ kwargs['type'] = PostType.POST
+ unique_id = 'pixiv:i:{}'.format(data['id'])
+ kwargs['unique_id'] = unique_id
+ kwargs['raw_responses'] = {}
+ 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.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'])
+ user.profile_picture_url = data['user']['profile_image_urls']['medium']
+ kwargs['author'] = user
+ kwargs['title'] = data['title']
+ kwargs['text'] = data['caption']
+ kwargs['likes'] = data['total_bookmarks']
+ if 'total_comments' in data:
+ kwargs['comments'] = data['total_comments']
+ kwargs['views'] = data['total_view']
+ kwargs['tags'] = [self.create_tag(tag) for tag in data['tags']]
+ media = {}
+ for i, page in enumerate(data['meta_pages']):
+ media[str(i)] = Image(url=self.create_media_url(page['image_urls']['original']), thumbnail_url=self.create_media_url(page['image_urls']['medium']))
+ kwargs['media'] = media
+ post = Post(**kwargs)
+ self.module.add_to_map(unique_id, post)
+ return post
+
+ def create_user(self, data: ParsedJson) -> User:
+ unique_id = f'pixiv:u:{data['id']}'
+ user = User(unique_id=unique_id, username=data['account'], display_name=data['name'])
+ self.module.add_to_map(unique_id, user)
+ return user
+
+class PixivAppSearch(PixivAppBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ self.next_qs: Optional[dict[str, Any]] = {}
+ self.next_qs['word'] = arg
+ self.next_qs['sort'] = 'popular_desc'
+ self.next_qs['search_ai_type'] = 1
+ self.next_qs['start_date'] = '2019-01-01'
+ self.next_qs['end_date'] = '2019-12-31'
+
+ def api_request_page(self, num: int) -> tuple[bool, Optional[dict[str, Any]]]:
+ if not self.next_qs:
+ return False, None
+ obj = self.module.api.search_illust(**self.next_qs)
+ if 'illusts' not in obj:
+ return False, None
+ for post in obj['illusts']:
+ self.pages[num].append(self.create_post(post).unique_id)
+ return True, self.module.api.parse_qs(obj['next_url'])
+
+class PixivAppIllust(PixivAppBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ self.next_qs: Optional[dict[str, Any]] = {}
+ self.next_qs['illust_id'] = arg
+
+ def api_request_page(self, num: int) -> tuple[bool, Optional[dict[str, Any]]]:
+ if not self.next_qs:
+ return False, None
+ obj = self.module.api.illust_detail(**self.next_qs)
+ if 'illust' not in obj:
+ return False, None
+ self.pages[num].append(self.create_post(obj['illust']).unique_id)
+ return True, None
+
+class PixivAppUser(PixivAppBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ self.next_qs: Optional[dict[str, Any]] = {}
+ self.next_qs['user_id'] = arg
+ self.next_qs['type'] = 'illust'
+
+ def api_request_page(self, num: int) -> tuple[bool, Optional[dict[str, Any]]]:
+ if not self.next_qs:
+ return False, None
+ obj = self.module.api.user_illusts(**self.next_qs)
+ if 'illusts' not in obj:
+ return False, None
+ for post in obj['illusts']:
+ self.pages[num].append(self.create_post(post).unique_id)
+ return True, self.module.api.parse_qs(obj['next_url'])
+
+class PixivAppBookmarks(PixivAppBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ self.next_qs: Optional[dict[str, Any]] = {}
+ self.next_qs['user_id'] = arg
+ self.next_qs['restrict'] = 'public'
+
+ def api_request_page(self, num: int) -> tuple[bool, Optional[dict[str, Any]]]:
+ if not self.next_qs:
+ return False, None
+ obj = self.module.api.user_bookmarks_illust(**self.next_qs)
+ if 'illusts' not in obj:
+ return False, None
+ for post in obj['illusts']:
+ self.pages[num].append(self.create_post(post).unique_id)
+ return True, self.module.api.parse_qs(obj['next_url'])
+
+class PixivAppFollowing(PixivAppBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ self.next_qs: Optional[dict[str, Any]] = {}
+ self.next_qs['user_id'] = arg
+ self.next_qs['restrict'] = 'public'
+
+ def api_request_page(self, num: int) -> tuple[bool, Optional[dict[str, Any]]]:
+ if not self.next_qs:
+ return False, None
+ obj = self.module.api.user_following(**self.next_qs)
+ if 'user_previews' not in obj:
+ return False, None
+ for user in obj['user_previews']:
+ self.pages[num].append(self.create_user(user['user']).unique_id)
+ return True, self.module.api.parse_qs(obj['next_url'])
+
+class PixivAppModule(Module):
+ def __init__(self, refresh_token: str):
+ super().__init__()
+ self.parser: QueryParser = QueryParser(self, 'search')
+ self.parser.add_command('search', PixivAppSearch)
+ self.parser.add_command('illust', PixivAppIllust)
+ self.parser.add_command('user', PixivAppUser)
+ self.parser.add_command('bookmarks', PixivAppBookmarks)
+ self.parser.add_command('following', PixivAppFollowing)
+ self.api: AppPixivAPI = AppPixivAPI()
+ self.api.auth(refresh_token=refresh_token)
+
+ def search(self, query: str, *extra_args: Any) -> Optional[Search]:
+ return self.parser.parse_query(query)
+
+ def get_download(self, unique_id: str, key: str) -> Optional[dict[str, Any]]:
+ if unique_id not in self.unique_id_map:
+ return None
+ post = self.unique_id_map[unique_id]
+ if key not in post.media:
+ return None
+ return {
+ 'urls': [post.media[key].url],
+ 'headers': {
+ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; rv:122.0) Gecko/20100101 Firefox/122.0',
+ 'Referer': 'https://www.pixiv.net/'
+ }
+ }
diff --git a/src/portal/py/modules/pixiv_web.py b/src/portal/py/modules/pixiv_web.py
new file mode 100644
index 0000000..7bc043b
--- /dev/null
+++ b/src/portal/py/modules/pixiv_web.py
@@ -0,0 +1,263 @@
+import log
+import json
+import httpx
+import urllib.parse
+from typing import Optional, Any
+from json.decoder import JSONDecodeError
+from base import Search, Module, Method, ParsedJson
+from post import MediaUrl, 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
+
+BASE_URL = 'https://www.pixiv.net/ajax'
+LANG = 'en'
+VERSION = '5f53980f6b59c9376220b6d86e8feb5ea9e22e10'
+
+class PixivWebBase(Search):
+ def __init__(self, userdata: Any):
+ super().__init__()
+ self.module: PixivWebModule = userdata
+
+ def check_api_response(self, response: Optional[httpx.Response]) -> Optional[ParsedJson]:
+ if not response:
+ return None
+ try:
+ obj = response.json()
+ except JSONDecodeError:
+ return None
+ if obj['error']:
+ log.error(obj['message'])
+ return None
+ return obj['body']
+
+ def create_media_url(self, url: str) -> MediaUrl:
+ return MediaUrl(url, url.split('.')[-1])
+
+ def parse_pages(self, data: ParsedJson) -> dict[str, Media]:
+ media = {}
+ for i, m in enumerate(data):
+ media[str(i)] = Image(url=self.create_media_url(m['urls']['original']), thumbnail_url=self.create_media_url(m['urls']['small']))
+ return media
+
+ def parse_ugoira(self, data: ParsedJson, thumbnail_url: str) -> Animation:
+ animation = Animation(url=self.create_media_url(data['originalSrc']), thumbnail_url=self.create_media_url(thumbnail_url))
+ for frame in data['frames']:
+ animation.frames.append((frame['file'], frame['delay']))
+ return animation
+
+ def seek_profile_picture_url(self, user_illusts: ParsedJson) -> Optional[MediaUrl]:
+ for illust in user_illusts.values():
+ if illust is None:
+ continue
+ if 'profileImageUrl' in illust:
+ return self.create_media_url(illust['profileImageUrl'])
+ return None
+
+ def create_tag(self, data: ParsedJson) -> Tag:
+ tag = Tag(type=TagType.GENERAL, name=data['tag'])
+ if 'romaji' in data:
+ tag.alts['romaji'] = data['romaji']
+ if 'translation' in data:
+ if 'en' in data['translation']:
+ tag.alts['en'] = data['translation']['en']
+ return tag
+
+ def create_post(self, data: ParsedJson, pages: Optional[ParsedJson]=None, ugoira: Optional[ParsedJson]=None) -> Optional[Post]:
+ kwargs = {}
+ kwargs['type'] = PostType.POST
+ illust_id = data['illustId']
+ unique_id = f'pixiv:i:{illust_id}'
+ kwargs['unique_id'] = unique_id
+ 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()
+ kwargs['dates'] = {
+ DateType.CREATED: create_date,
+ DateType.RETRIEVED: get_current_utc_time().timestamp()
+ }
+ upload_date = parse_pixiv_date(data['uploadDate']).timestamp()
+ if upload_date != create_date:
+ kwargs['dates'][DateType.EDITED] = upload_date
+ 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:
+ user.profile_picture_url = profile_picture_url
+ kwargs['author'] = user
+ kwargs['title'] = data['illustTitle']
+ kwargs['text'] = data['illustComment']
+ kwargs['likes'] = data['likeCount']
+ kwargs['bookmarks'] = data['bookmarkCount']
+ kwargs['comments'] = data['commentCount']
+ kwargs['views'] = data['viewCount']
+ kwargs['tags'] = [self.create_tag(tag) for tag in data['tags']['tags']]
+ if data['illustType'] == 2: # Ugoira
+ if not ugoira:
+ url = f'{BASE_URL}/illust/{illust_id}/ugoira_meta?lang={LANG}&version={VERSION}'
+ ugoira = self.check_api_response(self.module.do_request(Method.GET, url))
+ if not ugoira:
+ return None
+ kwargs['raw_responses']['ugoira'] = json.dumps(ugoira)
+ kwargs['media'] = { '0': self.parse_ugoira(ugoira, data['urls']['original']) }
+ elif data['pageCount'] > 1:
+ if not pages:
+ url = f'{BASE_URL}/illust/{illust_id}/pages?lang={LANG}&version={VERSION}'
+ pages = self.check_api_response(self.module.do_request(Method.GET, url))
+ if not pages:
+ return None
+ kwargs['raw_responses']['pages'] = json.dumps(pages)
+ kwargs['media'] = self.parse_pages(pages)
+ else:
+ kwargs['media'] = self.parse_pages([{ 'urls': data['urls'] }])
+ post = Post(**kwargs)
+ self.module.add_to_map(unique_id, post)
+ return post
+
+ def remake_post(self, raw_responses: ParsedJson, original_date: float) -> Optional[Post]:
+ api = json.loads(raw_responses['api'])
+ pages = raw_responses.get('pages', None)
+ if pages:
+ pages = json.loads(pages)
+ ugoira = raw_responses.get('ugoira', None)
+ if ugoira:
+ ugoira = json.loads(ugoira)
+ post = self.create_post(api, pages, ugoira)
+ if post:
+ post.dates[DateType.RETRIEVED] = original_date
+ return post
+
+ def request_illust(self, illust_id: int) -> Optional[Post]:
+ url = f'{BASE_URL}/illust/{illust_id}?lang={LANG}&version={VERSION}'
+ obj = self.check_api_response(self.module.do_request(Method.GET, url))
+ if not obj:
+ return None
+ return self.create_post(obj)
+
+ def create_user(self, data: ParsedJson) -> User:
+ unique_id=f'pixiv:u:{data['userId']}'
+ user = User(unique_id=unique_id, display_name=data['userName'])
+ user.profile_picture_url = self.create_media_url(data['profileImageUrl'])
+ self.module.add_to_map(unique_id, user)
+ return user
+
+class PixivWebSearch(PixivWebBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ self.query: str = arg
+ self.last_page: Optional[int] = None
+
+ def request_page(self, num: int) -> bool:
+ if self.last_page and num >= self.last_page:
+ return False
+ start = '2019-01-01'
+ end = '2019-12-31'
+ order = 'popular_d'
+ url = f'{BASE_URL}/search/artworks/{urllib.parse.quote(self.query)}?word={self.query}&order={order}&mode=all&scd={start}&ecd={end}&p={num + 1}&csw=0&s_mode=s_tag&type=all&lang={LANG}&version={VERSION}'
+ obj = self.check_api_response(self.module.do_request(Method.GET, url))
+ if not obj:
+ self.errored = True
+ return False
+ self.last_page = obj['illustManga']['lastPage']
+ self.pages[num] = []
+ for partial_post in obj['illustManga']['data']:
+ post = self.request_illust(int(partial_post['id']))
+ if not post:
+ self.errored = True
+ return False
+ self.pages[num].append(post.unique_id)
+ return True
+
+class PixivWebIllust(PixivWebBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ try:
+ self.id: int = int(arg)
+ except ValueError:
+ self.errored = True
+
+ def request_page(self, num: int) -> bool:
+ post = self.request_illust(self.id)
+ if not post:
+ self.errored = True
+ return False
+ self.pages[num] = [post.unique_id]
+ self.completed = True
+ return True
+
+class PixivWebUser(PixivWebBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ try:
+ self.id: int = int(arg)
+ except ValueError:
+ self.errored = True
+
+ def request_page(self, num: int) -> bool:
+ url = f'{BASE_URL}/user/{self.id}/profile/all?lang={LANG}&version={VERSION}'
+ obj = self.check_api_response(self.module.do_request(Method.GET, url))
+ if not obj:
+ self.errored = True
+ return False
+ self.pages[num] = []
+ if obj['manga']:
+ for illust in obj['manga'].keys():
+ post = self.request_illust(int(illust))
+ if not post:
+ self.errored = True
+ return False
+ self.pages[num].append(post.unique_id)
+ self.completed = True
+ return True
+
+class PixivWebFollowing(PixivWebBase):
+ LIMIT = 24
+
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ try:
+ self.id: int = int(arg)
+ except ValueError:
+ self.errored = True
+
+ def request_page(self, num: int) -> bool:
+ url = f'{BASE_URL}/user/{self.id}/following?offset={self.LIMIT * num}&limit={self.LIMIT}&rest=show&tag=&acceptingRequests=0&lang={LANG}&version={VERSION}'
+ obj = self.check_api_response(self.module.do_request(Method.GET, url))
+ if not obj:
+ self.errored = True
+ return False
+ self.pages[num] = []
+ for user in obj['users']:
+ self.pages[num].append(self.create_user(user).unique_id)
+ return True
+
+class PixivWebModule(Module):
+ def __init__(self, sessid: str):
+ super().__init__()
+ self.parser: QueryParser = QueryParser(self, 'search')
+ self.parser.add_command('search', PixivWebSearch)
+ 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'
+ self.headers['Accept-Language'] = 'en-US,en;q=0.5'
+ self.headers['Referer'] = 'https://www.pixiv.net/'
+ self.headers['User-Agent'] = 'Mozilla/5.0 (Windows NT 10.0; rv:122.0) Gecko/20100101 Firefox/122.0'
+ self.headers['x-user-id'] = '22781328'
+ self.cookies['PHPSESSID'] = sessid
+
+ def search(self, query: str, *extra_args: Any) -> Optional[Search]:
+ return self.parser.parse_query(query)
+
+ def get_download(self, unique_id: str, key: str) -> Optional[dict[str, Any]]:
+ if unique_id not in self.unique_id_map:
+ return None
+ post = self.unique_id_map[unique_id]
+ if key not in post.media:
+ return None
+ return {
+ 'urls': [post.media[key].url],
+ 'headers': {
+ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; rv:122.0) Gecko/20100101 Firefox/122.0',
+ 'Referer': 'https://www.pixiv.net/'
+ }
+ }
diff --git a/src/portal/py/modules/searx.py b/src/portal/py/modules/searx.py
new file mode 100644
index 0000000..4cc0b04
--- /dev/null
+++ b/src/portal/py/modules/searx.py
@@ -0,0 +1,95 @@
+import sys
+import json
+
+from base import Module, Search
+from post import Post, Image
+
+sys.path.append('../../vendor/searxng')
+from searx import settings
+from searx.utils import gen_useragent
+from searx.engines import load_engine, register_engine
+
+# TODO: Put these on the Provider object.
+ENGINES = {}
+
+ENGINE_INITIAL_INDEX = {
+ 'google images': 0,
+ 'google': 1,
+ 'bing images': 0
+}
+
+class SearxSearch(Search):
+ def __init__(self, plugin, query, engine):
+ super().__init__()
+ global ENGINES
+ self.query = query
+ self.plugin = plugin
+ self.key = engine.replace(' ', '_')
+ self.engine = ENGINES[engine]
+ self.index = ENGINE_INITIAL_INDEX[engine]
+
+ def load_page(self, index):
+ self.pages[index] = []
+ res = self.plugin.send_http_request(self.engine.request(self.query, {
+ 'language': 'en-US',
+ 'safesearch': 0,
+ 'time_range': None,
+ 'pageno': self.index + index,
+ 'cookies': self.plugin.cookies,
+ 'headers': self.plugin.headers,
+ 'data': None
+ }))
+ if not res:
+ return False
+ try:
+ ret = self.engine.response(res)
+ except:
+ return False
+ if len(ret) == 0:
+ # TODO: Test.
+ self.completed = True
+ return False
+ for i, p in enumerate(ret):
+ if 'url' not in p:
+ continue
+ # Don't add this to the global map because it's not actually unique.
+ unique_id = '{}:{}_{}:{}'.format(self.key, self.query, i, index)
+ media = []
+ if 'img_src' in p:
+ media.append(Image(url=p['img_src'], thumbnail_url=''))
+ elif 'image_url' in p:
+ media.append(Image(url=p['image_url'], thumbnail_url=''))
+ kwargs = {}
+ kwargs['unique_id'] = unique_id
+ kwargs['raw_responses'] = {}
+ kwargs['raw_responses']['result'] = json.dumps(p)
+ kwargs['url'] = p['url']
+ kwargs['title'] = p['title']
+ kwargs['text'] = p['content']
+ kwargs['media'] = media
+ self.pages[index].append(Post(**kwargs))
+ return True
+
+class SearxModule(Module):
+ def __init__(self):
+ super().__init__()
+ global ENGINES
+ self.headers = {
+ 'User-Agent': gen_useragent(),
+ 'Accept-Language': 'en-US,en;q=0.5',
+ }
+ for engine_data in settings['engines']:
+ engine = load_engine(engine_data)
+ if engine:
+ register_engine(engine)
+ ENGINES[engine.name] = engine
+
+ def send_http_request(self, params):
+ self.cookies.update(params['cookies'])
+ self.headers.update(params['headers'])
+ return self.get(params['url'], params=params['data'])
+
+ def search(self, query, *extra_args):
+ if not extra_args:
+ return None
+ return SearxSearch(self, query, extra_args[0])
diff --git a/src/portal/py/modules/twitter.py b/src/portal/py/modules/twitter.py
new file mode 100644
index 0000000..7e77253
--- /dev/null
+++ b/src/portal/py/modules/twitter.py
@@ -0,0 +1,190 @@
+import log
+import http.cookiejar
+from typing import Optional, Any, Iterator
+from datetime import timezone
+from base import Search, Module
+from post import MediaUrl, PostType, DateType, PostRef, Post, User, Media, Image, Video
+from query_parser import QueryParser
+from modules.common import get_current_utc_time
+
+# GraphQL API
+from snscrape.modules import twitter
+from snscrape.base import ScraperException
+from snscrape.modules.twitter import (Tweet, TweetRef, Tombstone, UserRef,
+ TwitterSearchScraper, TwitterProfileScraper, TwitterUserScraper, TwitterTweetScraper)
+
+class TwitterScrapeBase(Search):
+ # Posts per emulated page because snscrape returns an iterator.
+ POSTS_PER_PAGE = 8
+
+ # Don't allow requests for a page more than this amount past the current page.
+ REACH_LIMIT = 4
+
+ def __init__(self, userdata: Any):
+ super().__init__()
+ self.module: TwitterScrapeModule = userdata
+ self.iterator: Iterator[Tweet | TweetRef | Tombstone]
+ self.page: int = 0
+
+ def create_media_url(self, url: str) -> MediaUrl:
+ format = url.find('format=')
+ if format >= 0:
+ ext = url[format + 7:format + 10]
+ else:
+ question = url.rfind('?')
+ if question >= 0:
+ url = url[0:question]
+ ext = url.split('.')[-1]
+ return MediaUrl(url, ext)
+
+ def parse_media(self, data: list[twitter.Medium]) -> dict[str, Media]:
+ media = {}
+ for i, m in enumerate(data):
+ if isinstance(m, twitter.Photo):
+ media[str(i)] = Image(url=self.create_media_url(m.fullUrl), thumbnail_url=self.create_media_url(m.previewUrl))
+ elif isinstance(m, twitter.Video) or isinstance(m, twitter.Gif):
+ videos = sorted(m.variants, key=lambda x: x.bitrate if x.bitrate else 0, reverse=True)
+ media[str(i)] = Video(url=self.create_media_url(videos[0].url), thumbnail_url=self.create_media_url(m.thumbnailUrl))
+ return media
+
+ def create_post(self, tweet: Tweet | TweetRef | Tombstone) -> Post:
+ kwargs = {}
+ kwargs['unique_id'] = f'twitter:t:{tweet.id}'
+ if isinstance(tweet, TweetRef) or isinstance(tweet, Tombstone):
+ kwargs['type'] = PostType.TOMBSTONE
+ return Post(**kwargs)
+ kwargs['raw_responses'] = tweet.rawResponses
+ kwargs['url'] = tweet.url
+ # TODO: replace correct?
+ kwargs['dates'] = {
+ DateType.CREATED: tweet.date.replace(tzinfo=timezone.utc).timestamp(),
+ DateType.RETRIEVED: get_current_utc_time().timestamp()
+ }
+ author = User(unique_id=f'twitter:u:{tweet.user.id}')
+ if not isinstance(tweet.user, UserRef):
+ author.username = tweet.user.username
+ if tweet.user.displayname:
+ author.display_name = tweet.user.displayname
+ if tweet.user.profileImageUrl:
+ author.profile_picture_url = self.create_media_url(tweet.user.profileImageUrl)
+ kwargs['author'] = author
+ if tweet.retweetedTweet:
+ kwargs['type'] = PostType.REPOST
+ kwargs['post'] = PostRef(f'twitter:t:{tweet.retweetedTweet.id}')
+ return Post(**kwargs)
+ kwargs['type'] = PostType.POST
+ kwargs['text'] = tweet.rawContent
+ kwargs['likes'] = tweet.likeCount
+ kwargs['reposts'] = tweet.retweetCount
+ kwargs['quotes'] = tweet.quoteCount
+ kwargs['comments'] = tweet.replyCount
+ kwargs['views'] = tweet.viewCount
+ if tweet.media:
+ kwargs['media'] = self.parse_media(tweet.media)
+ if tweet.inReplyToTweetId:
+ kwargs['in_reply_to'] = PostRef(f'twitter:t:{tweet.inReplyToTweetId}')
+ if tweet.quotedTweet:
+ kwargs['quoted'] = PostRef(f'twitter:t:{tweet.quotedTweet.id}')
+ return Post(**kwargs)
+
+ def add_post_to_page(self, num: int, item: Post | User) -> None:
+ self.module.add_to_map(item.unique_id, item)
+ self.pages[num].append(item.unique_id)
+
+ def step_iterator_for_page(self, num: int) -> bool:
+ for _ in range(0, self.POSTS_PER_PAGE):
+ try:
+ tweet = next(self.iterator)
+ except StopIteration:
+ self.completed = True
+ break
+ except ScraperException as e:
+ log.error(repr(e))
+ continue
+ self.add_post_to_page(num, self.create_post(tweet))
+ if isinstance(tweet, TweetRef) or isinstance(tweet, Tombstone):
+ continue
+ if tweet.retweetedTweet:
+ retweet = self.create_post(tweet.retweetedTweet)
+ self.module.add_to_map(retweet.unique_id, retweet)
+ if tweet.quotedTweet:
+ quote = self.create_post(tweet.quotedTweet)
+ self.module.add_to_map(quote.unique_id, quote)
+ return len(self.pages[num]) > 0
+
+ def request_page(self, num: int) -> bool:
+ if num - self.page >= self.REACH_LIMIT:
+ return False
+ request_satisfied = False
+ for i in range(self.page, num + 1):
+ if self.completed:
+ break
+ if i in self.pages:
+ continue
+ self.pages[i] = []
+ if not self.step_iterator_for_page(i):
+ break
+ self.page = i
+ if i == num:
+ request_satisfied = True
+ return request_satisfied
+
+class TwitterScrapeSearch(TwitterScrapeBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ self.iterator = TwitterSearchScraper(arg, top=True, cookies=self.module.cookies).get_items()
+
+class TwitterScrapeUser(TwitterScrapeBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ if arg.startswith('@'):
+ arg = arg[1:]
+ self.iterator = TwitterUserScraper(arg, cookies=self.module.cookies).get_items()
+
+class TwitterScrapeProfile(TwitterScrapeBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ if arg.startswith('@'):
+ arg = arg[1:]
+ self.iterator = TwitterProfileScraper(arg, cookies=self.module.cookies).get_items()
+
+class TwitterScrapeTweet(TwitterScrapeBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ if arg.startswith('https://'):
+ arg = arg[arg.rfind('/') + 1:]
+ question = arg.find('?')
+ if question >= 0:
+ arg = arg[:question]
+ self.iterator = TwitterTweetScraper(arg, cookies=self.module.cookies).get_items()
+
+class TwitterScrapeModule(Module):
+ def __init__(self, cookies_path: str):
+ super().__init__()
+ self.parser: QueryParser = QueryParser(self, 'search')
+ self.parser.add_command('search', TwitterScrapeSearch)
+ self.parser.add_command('user', TwitterScrapeUser)
+ self.parser.add_command('profile', TwitterScrapeProfile)
+ self.parser.add_command('tweet', TwitterScrapeTweet)
+ if cookies_path:
+ cookie_jar = http.cookiejar.MozillaCookieJar()
+ cookie_jar.load(filename=cookies_path, ignore_expires=True)
+ for c in cookie_jar:
+ if c.value:
+ self.cookies[c.name] = c.value
+
+ def search(self, query: str, *extra_args: Any) -> Optional[Search]:
+ return self.parser.parse_query(query)
+
+ def get_download(self, unique_id: str, key: str) -> Optional[dict[str, Any]]:
+ if unique_id not in self.unique_id_map:
+ return None
+ post = self.unique_id_map[unique_id]
+ if key not in post.media:
+ return None
+ return {
+ 'urls': [post.media[key].url],
+ 'headers': {
+ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; rv:122.0) Gecko/20100101 Firefox/122.0'
+ }
+ }
diff --git a/src/portal/py/modules/twitter_api.py b/src/portal/py/modules/twitter_api.py
new file mode 100644
index 0000000..6d04855
--- /dev/null
+++ b/src/portal/py/modules/twitter_api.py
@@ -0,0 +1,269 @@
+from base import Search, Module
+from query_parser import QueryParser
+
+# Official Twitter API
+from twitter import Twitter2, TwitterError, OAuth
+
+# Quite incomplete API based backend. Should *not* be used for archiving.
+# HOLD: I can't test this without paying $100 for the X API WTFFF.
+'''
+class TwitterBase(Search):
+ ALL_PARAMS = {
+ 'tweet.fields': 'attachments,author_id,context_annotations,conversation_id,created_at,entities,geo,id,in_reply_to_user_id,lang,public_metrics,possibly_sensitive,referenced_tweets,reply_settings,source,text,withheld',
+ 'user.fields': 'created_at,description,entities,id,location,name,pinned_tweet_id,profile_image_url,protected,public_metrics,url,username,verified,withheld',
+ 'media.fields': 'duration_ms,height,media_key,preview_image_url,type,url,width,public_metrics',
+ 'place.fields': 'contained_within,country,country_code,full_name,geo,id,name,place_type',
+ 'poll.fields': 'duration_minutes,end_datetime,id,options,voting_status',
+ }
+ ALL_EXPRESSIONS = 'author_id,referenced_tweets.id,referenced_tweets.id.author_id,entities.mentions.username,attachments.poll_ids,attachments.media_keys,in_reply_to_user_id,geo.place_id'
+
+ SOME_PARAMS = {
+ 'tweet.fields': 'attachments,author_id,text,entities,referenced_tweets',
+ 'user.fields': 'id,name,profile_image_url,url,username',
+ 'media.fields': 'duration_ms,height,media_key,preview_image_url,type,url,width,public_metrics',
+ 'place.fields': '',
+ 'poll.fields': '',
+ }
+ SOME_EXPRESSIONS = 'author_id,referenced_tweets.id,referenced_tweets.id.author_id,entities.mentions.username,attachments.media_keys,in_reply_to_user_id'
+
+ def __init__(self, userdata, arg):
+ super().__init__()
+ self.provider = userdata
+ self.params = self.SOME_PARAMS.copy()
+ self.media_map = {}
+ self.user_id = None
+ self.pagination_page = 0
+ self.pagination_token = None
+ self.arg = arg
+
+ def add_attachemnts(self, l, r, data):
+ if 'media_keys' not in data:
+ return
+ for m in data['media_keys']:
+ if m in r:
+ continue
+ r.append(m)
+ if m in self.media_map:
+ l.append(Image(url=self.media_map[m], thumbnail_url=''))
+
+ def add_entity_urls(self, data):
+ if 'urls' in data:
+ for u in data['urls']:
+ if 'media_key' in u and u['media_key'] not in self.media_map:
+ self.media_map[u['media_key']] = u['expanded_url']
+
+ def make_post(self, tweet):
+ print(json.dumps(tweet, indent=4))
+ media = []
+ repeats = []
+ if 'referenced_tweets' in t:
+ print(len(referenced_tweets))
+ #for rt in t['referenced_tweets']:
+ # if rt['id'] in self.tweet_map:
+ # rrt = self.tweet_map[rt['id']]
+ # if 'entities' in rrt:
+ # self.add_entity_urls(rrt['entities'])
+ # if 'attachments' in rrt:
+ # self.add_attachemnts(media, repeats, rrt['attachments'])
+ if 'attachments' in t:
+ self.add_attachemnts(media, repeats, t['attachments'])
+ unique_id = 'twitter:t:{}'.format(t['id'])
+ url = 'https://twitter.com/{}/status/{}'.format(t['author_id'], t['id'])
+ kwargs = {}
+ kwargs['unique_id'] = unique_id
+ kwargs['raw_responses'] = { 'tweet' : json.dumps(t) }
+ kwargs['url'] = url
+ kwargs['author'] = User(unique_id='twitter:u:{}'.format(t['author_id']))
+ kwargs['title'] = ''
+ kwargs['text'] = t['text']
+ kwargs['media'] = media
+ return Post(**kwargs)
+
+ def add_items_from_search(self, data):
+ if 'includes' in data:
+ includes = data['includes']
+ if 'media' in includes:
+ for i in includes['media']:
+ if 'url' in i:
+ self.media_map[i['media_key']] = i['url']
+ #if 'tweets' in includes:
+ # for t in includes['tweets']:
+ # self.tweet_map[t['id']] = t
+ if 'data' not in data or not data['data']:
+ return False
+ l = data['data'] if type(data['data']) == list else [data['data']]
+ for t in l:
+ if 'entities' in t:
+ self.add_entity_urls(t['entities'])
+ post = self.make_post(t)
+ self.pages[self.pagination_page].append(post)
+ return True
+
+ def add_user_ids_from_search(self, data):
+ if not data['data']:
+ return False
+ for u in data['data']:
+ self.pages[self.pagination_page].append(User(
+ 'twitter:u:{}'.format(u['id']), username=u['username'], display_name=u['name']))
+ return True
+
+ def should_process_index(self, index):
+ if index - self.pagination_page >= REACH_LIMIT:
+ return False
+ if self.pagination_page > 0 and not self.pagination_token:
+ self.completed = True
+ return False
+ if self.pagination_token:
+ self.params['pagination_token'] = self.pagination_token
+ self.pages[self.pagination_page] = []
+ return True
+
+ def handle_data(self, data, user_ids=False):
+ if 'next_token' not in data['meta'].keys():
+ self.pagination_token = None
+ else:
+ self.pagination_token = data['meta']['next_token']
+ if user_ids:
+ if not self.add_user_ids_from_search(data):
+ return False
+ else:
+ if not self.add_items_from_search(data):
+ return False
+ self.pagination_page += 1
+ return True
+
+ def get_user_id(self, username):
+ if self.user_id:
+ return True
+ try:
+ data = self.provider.t.users.by.username._username(
+ _username=username, _timeout=config.TWITTER_TIMEOUT)
+ except TwitterError as e:
+ log.error(repr(e))
+ return False
+ self.user_id = data['data']['id']
+ return True
+
+class TwitterSearch(TwitterBase):
+ def load_page(self, index):
+ for _ in range(self.pagination_page, index + 1):
+ if not self.should_process_index(index):
+ return False
+ try:
+ data = self.provider.t.tweets.search.recent(
+ query=self.arg, expansions=self.SOME_EXPRESSIONS, params=self.params,
+ sort_order='relevancy', max_results=25, _timeout=config.TWITTER_TIMEOUT)
+ except TwitterError as e:
+ log.error(repr(e))
+ return False
+ return self.handle_data(data)
+
+class TwitterUser(TwitterBase):
+ def __init__(self, userdata, arg):
+ if arg.startswith('@'):
+ arg = arg[1:]
+ super().__init__(userdata, arg)
+
+ def load_page(self, index):
+ for _ in range(self.pagination_page, index + 1):
+ if not self.should_process_index(index):
+ return False
+ if not self.get_user_id(self.arg):
+ return False
+ try:
+ data = self.provider.t.users._id.tweets(
+ _id=self.user_id, expansions=self.SOME_EXPRESSIONS, params=self.params,
+ max_results=25, _timeout=config.TWITTER_TIMEOUT)
+ except TwitterError as e:
+ log.error(repr(e))
+ return False
+ return self.handle_data(data)
+
+class TwitterTimeline(TwitterBase):
+ def __init__(self, userdata, arg):
+ if not arg:
+ arg = 'pizzabelly'
+ super().__init__(userdata, arg)
+ self.params['exclude'] = 'replies'
+
+ def load_page(self, index):
+ for _ in range(self.pagination_page, index + 1):
+ if not self.should_process_index(index):
+ return False
+ if not self.get_user_id(self.arg):
+ return False
+ try:
+ data = self.provider.t.users._id.timelines.reverse_chronological(
+ _id=self.user_id, expansions=self.SOME_EXPRESSIONS, params=self.params,
+ max_results=25, _timeout=config.TWITTER_TIMEOUT)
+ except TwitterError as e:
+ log.error(repr(e))
+ return False
+ return self.handle_data(data)
+
+class TwitterLikes(TwitterBase):
+ def __init__(self, userdata, arg):
+ if arg.startswith('@'):
+ arg = arg[1:]
+ super().__init__(userdata, arg)
+
+ def load_page(self, index):
+ for _ in range(self.pagination_page, index + 1):
+ if not self.should_process_index(index):
+ return False
+ if not self.get_user_id(self.arg):
+ return False
+ try:
+ data = self.provider.t.users._id.liked_tweets(
+ _id=self.user_id, expansions=self.SOME_EXPRESSIONS, params=self.params,
+ max_results=25, _timeout=config.TWITTER_TIMEOUT)
+ except TwitterError as e:
+ log.error(repr(e))
+ return False
+ return self.handle_data(data)
+
+class TwitterTweet(TwitterBase):
+ def load_page(self, index):
+ for _ in range(self.pagination_page, index + 1):
+ if not self.should_process_index(index):
+ return False
+ try:
+ data = self.provider.t.tweets(
+ ids=self.arg, expansions=self.SOME_EXPRESSIONS, params=self.params,
+ _timeout=config.TWITTER_TIMEOUT)
+ except TwitterError as e:
+ log.error(repr(e))
+ return False
+ return self.handle_data(data)
+
+class TwitterFollowing(TwitterBase):
+ def __init__(self, userdata, arg):
+ if arg.startswith('@'):
+ arg = arg[1:]
+ super().__init__(userdata, arg)
+ del self.params['media.fields']
+ del self.params['place.fields']
+ del self.params['poll.fields']
+
+ def load_page(self, index):
+ for _ in range(self.pagination_page, index + 1):
+ if not self.should_process_index(index):
+ return False
+ if not self.get_user_id(self.arg):
+ return False
+ try:
+ data = self.provider.t.users._id.following(
+ _id=self.user_id, params=self.params, max_results=25, _timeout=config.TWITTER_TIMEOUT)
+ except TwitterError as e:
+ log.error(repr(e))
+ return False
+ return self.handle_data(data, True)
+'''
+
+class TwitterApiModule(Module):
+ def __init__(self, access_key: str, access_secret: str, consumer_key: str, consumer_secret: str):
+ self.t: Twitter2 = Twitter2(auth=OAuth(access_key, access_secret, consumer_key, consumer_secret), retry=True)
+ self.parser: QueryParser = QueryParser(self, 'timeline')
+ #self.parser.add_command('timeline', TwitterTimeline)
+ #self.parser.add_command('likes', TwitterLikes)
+ #self.parser.add_command('following', TwitterFollowing)
diff --git a/src/portal/py/modules/youtube.py b/src/portal/py/modules/youtube.py
new file mode 100644
index 0000000..d14dfe8
--- /dev/null
+++ b/src/portal/py/modules/youtube.py
@@ -0,0 +1,133 @@
+import log
+from typing import Optional, Any
+from base import Search, Module, ParsedJson
+from post import MediaUrl, Post, PostType, Media, Video
+from query_parser import QueryParser
+from yt_dlp import YoutubeDL
+
+class YDLLogger():
+ def debug(self, msg):
+ if msg.startswith('[debug] '):
+ log.debug(msg)
+ else:
+ log.info(msg)
+
+ def info(self, msg):
+ log.info(msg)
+
+ def warning(self, msg):
+ log.warn(msg)
+
+ def error(self, msg):
+ log.error(msg)
+
+ydl_opts = {
+ 'quiet': False,
+ 'logger': YDLLogger(),
+ 'cachedir': False,
+# '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]
+
+ if 'formats' not in data:
+ if 'url' in data:
+ return data['url']
+ return None
+
+ # Filter out hls temporarily.
+ 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']
+
+ # audio_ext?
+ has_audio = list(filter(lambda f: 'acodec' not in f or f['acodec'] != 'none', 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
+ 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
+
+ return url
+
+class YoutubeBase(Search):
+ def __init__(self, userdata: Any):
+ super().__init__()
+ self.module: YoutubeModule = userdata
+
+ def get_info(self) -> Optional[ParsedJson]:
+ return None
+
+ def request_page(self, num: int) -> bool:
+ info = self.get_info()
+ if not info:
+ return False
+ url = get_playback_url(info)
+ if not url:
+ return False
+ media: dict[str, Media] = { '0': Video(url=MediaUrl(url=url)) }
+ unique_id = f'youtube:v:{info['id']}'
+ 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
+
+class YoutubeSearch(YoutubeBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ self.query: str = arg
+
+ def get_info(self) -> Optional[ParsedJson]:
+ return ydl.extract_info(f'ytsearch1:{self.query}', download=False)
+
+class YoutubeLink(YoutubeBase):
+ def __init__(self, userdata: Any, arg: str):
+ super().__init__(userdata)
+ self.link: str = arg
+
+ def get_info(self) -> Optional[ParsedJson]:
+ return ydl.extract_info(self.link, download=False)
+
+class YoutubeModule(Module):
+ def __init__(self):
+ super().__init__()
+ self.parser: QueryParser = QueryParser(self, 'search')
+ self.parser.add_command('search', YoutubeSearch)
+ self.parser.add_command('link', YoutubeLink)
+
+ def search(self, query: str, *extra_args: Any) -> Optional[Search]:
+ return self.parser.parse_query(query)
+
+ def get_download(self, unique_id: str, key: str) -> Optional[dict[str, Any]]:
+ if unique_id not in self.unique_id_map:
+ return None
+ return None
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 23:31:38 +0000