summaryrefslogtreecommitdiff
path: root/src/portal/py
diff options
context:
space:
mode:
Diffstat (limited to 'src/portal/py')
-rw-r--r--src/portal/py/__init__.py0
-rw-r--r--src/portal/py/base.py116
-rw-r--r--src/portal/py/config.def.py20
-rw-r--r--src/portal/py/log.py34
-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
-rw-r--r--src/portal/py/post.py132
-rw-r--r--src/portal/py/query_parser.py45
16 files changed, 1875 insertions, 0 deletions
diff --git a/src/portal/py/__init__.py b/src/portal/py/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/src/portal/py/__init__.py
diff --git a/src/portal/py/base.py b/src/portal/py/base.py
new file mode 100644
index 0000000..c0a0b4f
--- /dev/null
+++ b/src/portal/py/base.py
@@ -0,0 +1,116 @@
+import log
+import time
+import httpx
+import threading
+from enum import Enum
+from collections import OrderedDict
+from typing import Optional, Self, Any
+from post import Post, User
+
+class Method(Enum):
+ HEAD = "HEAD"
+ GET = "GET"
+ POST = "POST"
+
+ParsedJson = Any
+
+class Search():
+ def __init__(self):
+ self.pages: dict[int, list[str]] = {}
+ self.completed: bool = False
+ self.errored: bool = False
+ self.iter_page: int = 0
+ self.iter_index: int = 0
+
+ def __iter__(self) -> Self:
+ self.iter_page = 0
+ self.iter_index = 0
+ return self
+
+ def __next__(self) -> str:
+ page = self.get_page(self.iter_page)
+ if not page:
+ raise StopIteration
+ unique_id = page[self.iter_index]
+ self.iter_index += 1
+ if self.iter_index >= len(page):
+ self.iter_page += 1
+ self.iter_index = 0
+ return unique_id
+
+ def request_page(self, num: int) -> bool:
+ return False
+
+ def get_page(self, num: int) -> Optional[list[str]]:
+ if self.errored:
+ return None
+ if num not in self.pages:
+ if self.completed:
+ return None
+ if not self.request_page(num):
+ if not self.completed:
+ self.errored = True
+ return None
+ return self.pages[num]
+
+class Module():
+ def __init__(self):
+ self.headers: dict[str, str] = {}
+ self.cookies: 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:
+ return True
+
+ def do_request(self, method: Method, url: str, params: Optional[dict[str, str]] = None, retries: int = 2) -> Optional[httpx.Response]:
+ log.debug(f'HTTP {method.value} request {url} {params}.')
+ response: Optional[httpx.Response] = None
+ for _ in range(0, retries + 1):
+ try:
+ response = self.session.request(method.value, url, headers=self.headers, cookies=self.cookies, params=params)
+ except Exception as e:
+ log.info(f'Connection exception ({e}), retrying in 20 seconds...')
+ time.sleep(20)
+ continue
+ if not response:
+ continue
+ if response.status_code == 429:
+ log.info('Got too many requests, waiting for 3 minutes...')
+ time.sleep(60 * 3)
+ response = None
+ continue
+ if response.status_code == 404 or response.status_code == 403: # Shortcut 404, 403
+ 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...')
+ time.sleep(5)
+ response = None
+ continue
+ try:
+ response.read()
+ except Exception as e:
+ log.info(f'Read exception ({e}), retrying in 20 seconds...')
+ time.sleep(20)
+ response = None
+ continue
+ else:
+ break
+ return response
+
+ def add_to_map(self, unique_id: str, item: Post | User):
+ with self.map_mutex:
+ self.unique_id_map[unique_id] = item
+ if len(self.unique_id_map) > 10240:
+ self.unique_id_map.popitem(last=False)
+
+ def search(self, query: str, *extra_args: Any) -> Optional[Search]:
+ return None
+
+ def get_item(self, unique_id: str) -> Optional[Post | User]:
+ with self.map_mutex:
+ return self.unique_id_map.get(unique_id, None)
+
+ def get_download(self, unique_id: str, key: str) -> Optional[dict[str, Any]]:
+ return None
diff --git a/src/portal/py/config.def.py b/src/portal/py/config.def.py
new file mode 100644
index 0000000..c4e11c6
--- /dev/null
+++ b/src/portal/py/config.def.py
@@ -0,0 +1,20 @@
+from pathlib import Path
+
+BASE_DIR = ''
+
+TWITTER_ACCESS_TOKEN = ''
+TWITTER_ACCESS_TOKEN_SECRET = ''
+TWITTER_CONSUMER_TOKEN = ''
+TWITTER_CONSUMER_TOKEN_SECRET = ''
+TWITTER_COOKIES_PATH = '{}/cookies_twitter.txt'.format(BASE_DIR)
+TWITTER_TIMEOUT = 5 # seconds
+
+FANBOX_SESSID = ''
+
+PIXIV_REFRESH_TOKEN = ''
+
+INSTAGRAM_USERNAME = ''
+INSTAGRAM_PASSWORD = ''
+INSTAGRAM_SESSION_ID = ''
+INSTAGRAM_SETTINGS_PATH = Path('{}/ig_settings.json'.format(BASE_DIR))
+INSTAGRAM_USER_AGENT = ''
diff --git a/src/portal/py/log.py b/src/portal/py/log.py
new file mode 100644
index 0000000..b693a4c
--- /dev/null
+++ b/src/portal/py/log.py
@@ -0,0 +1,34 @@
+class DefaultLogger():
+ def debug(self, msg):
+ print(msg)
+
+ def info(self, msg):
+ print(msg)
+
+ def warn(self, msg):
+ print(msg)
+
+ def error(self, msg):
+ print(msg)
+
+LOGGER = DefaultLogger()
+
+def set_logger(logger):
+ global LOGGER
+ LOGGER = logger
+
+def debug(msg):
+ global LOGGER
+ LOGGER.debug(msg)
+
+def info(msg):
+ global LOGGER
+ LOGGER.info(msg)
+
+def warn(msg):
+ global LOGGER
+ LOGGER.warn(msg)
+
+def error(msg):
+ global LOGGER
+ LOGGER.error(msg)
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
diff --git a/src/portal/py/post.py b/src/portal/py/post.py
new file mode 100644
index 0000000..b064f97
--- /dev/null
+++ b/src/portal/py/post.py
@@ -0,0 +1,132 @@
+import dataclasses
+from dataclasses import field
+from typing import Optional
+from enum import Enum
+from json import JSONEncoder
+
+class PostType(int, Enum):
+ UNKNOWN = 0
+ POST = 1
+ REPOST = 2
+ PREVIEW = 3
+ TOMBSTONE = 4
+
+class DateType(int, Enum):
+ CREATED = 0
+ EDITED = 1
+ RETRIEVED = 2
+
+class MediaType(int, Enum):
+ UNKNOWN = 0
+ FILE = 1
+ AUDIO = 2
+ IMAGE = 3
+ VIDEO = 4
+ VIDEO_SPLIT = 5
+ ANIMATION = 6
+
+class TagType(int, Enum):
+ UNKNOWN = 0
+ GENERAL = 1
+ ARTIST = 2
+ CHARACTER = 3
+ COPYRIGHT = 4
+ META = 5
+ DEPRECATED = 6
+
+@dataclasses.dataclass
+class MediaUrl():
+ url: str = ''
+ ext: str = 'unknown'
+
+@dataclasses.dataclass
+class Media():
+ type: MediaType = MediaType.UNKNOWN
+ url: MediaUrl = field(default_factory=lambda: MediaUrl())
+
+@dataclasses.dataclass
+class File(Media):
+ type: MediaType = MediaType.FILE
+ name: str = ''
+
+@dataclasses.dataclass
+class Audio(Media):
+ type: MediaType = MediaType.AUDIO
+
+@dataclasses.dataclass
+class Image(Media):
+ type: MediaType = MediaType.IMAGE
+ thumbnail_url: MediaUrl = field(default_factory=lambda: MediaUrl())
+
+@dataclasses.dataclass
+class Video(Media):
+ type: MediaType = MediaType.VIDEO
+ thumbnail_url: MediaUrl = field(default_factory=lambda: MediaUrl())
+
+@dataclasses.dataclass
+class VideoSplit(Media):
+ type: MediaType = MediaType.VIDEO_SPLIT
+ audio_url: MediaUrl = field(default_factory=lambda: MediaUrl())
+ subtitle_url: MediaUrl = field(default_factory=lambda: MediaUrl())
+ thumbnail_url: MediaUrl = field(default_factory=lambda: MediaUrl())
+
+@dataclasses.dataclass
+class Animation(Media):
+ type: MediaType = MediaType.ANIMATION
+ thumbnail_url: MediaUrl = field(default_factory=lambda: MediaUrl())
+ frames: list[tuple[str, int]] = field(default_factory=lambda: [])
+
+@dataclasses.dataclass
+class Tag():
+ type: TagType = TagType.UNKNOWN
+ name: str = ''
+ alts: dict[str, str] = field(default_factory=lambda: {})
+
+@dataclasses.dataclass
+class User():
+ unique_id: str = ''
+ username: str = ''
+ display_name: str = ''
+ profile_picture_url: MediaUrl = field(default_factory=lambda: MediaUrl())
+
+@dataclasses.dataclass
+class PostRef():
+ unique_id: str = ''
+
+# V4 Ideas:
+# - Date estimate and range
+# - Edited but unknown when
+# - Generally an estimate/guess
+# - Formated text/body
+
+@dataclasses.dataclass
+class Post():
+ version: int = 3
+ type: PostType = PostType.UNKNOWN
+ unique_id: str = ''
+ raw_responses: dict[str, str] = field(default_factory=lambda: {})
+ url: str = ''
+ dates: dict[DateType, float] = field(default_factory=lambda: {})
+ author: User = field(default_factory=lambda: User())
+ title: str = ''
+ text: str = ''
+ likes: Optional[int] = None
+ bookmarks: Optional[int] = None
+ reposts: Optional[int] = None
+ quotes: Optional[int] = None
+ comments: Optional[int] = None
+ views: Optional[int] = None
+ tags: list[Tag] = field(default_factory=lambda: [])
+ links: list[str] = field(default_factory=lambda: [])
+ media: dict[str, Media] = field(default_factory=lambda: {})
+ post: PostRef = field(default_factory=lambda: PostRef())
+ quoted: PostRef = field(default_factory=lambda: PostRef())
+ in_reply_to: PostRef = field(default_factory=lambda: PostRef())
+
+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/query_parser.py b/src/portal/py/query_parser.py
new file mode 100644
index 0000000..c16ded2
--- /dev/null
+++ b/src/portal/py/query_parser.py
@@ -0,0 +1,45 @@
+from typing import Optional, Any, Callable
+from base import Search
+
+type QueryCommand = Callable[[Any, Optional[str]], Search]
+
+class QueryParser():
+ def __init__(self, userdata: Any, default_command: str):
+ self.commands: dict[str, QueryCommand] = {}
+ self.escaped_commands: dict[str, str] = {}
+ self.userdata: Any = userdata
+ self.default_command: str = default_command
+
+ def add_command(self, name, search) -> None:
+ self.commands[name] = search
+ self.escaped_commands['\\' + name] = name
+
+ def parse_query(self, query: str) -> Search:
+ arg = query
+ search = self.commands[self.default_command]
+
+ sp = query.split(' ')
+
+ for i, word in enumerate(sp):
+ col = word.find(':')
+ if col == -1:
+ continue
+ command = word[0:col]
+ if command in self.escaped_commands:
+ query = query.replace(command, self.escaped_commands[command], 1)
+ continue
+ if command in self.commands:
+ if col + 1 < len(word):
+ arg = word[col+1:]
+ else:
+ arg = None
+ search = self.commands[command]
+ if i == 0:
+ if len(sp) > 1:
+ query = query.replace(word + ' ', '', 1)
+ else:
+ query = query.replace(word, '', 1)
+ else:
+ query = query.replace(' ' + word + ' ', '', 1)
+
+ return search(self.userdata, arg)
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 22:31:09 +0000