diff options
Diffstat (limited to 'src/shoki/py/providers')
| -rw-r--r-- | src/shoki/py/providers/__init__.py | 0 | ||||
| -rw-r--r-- | src/shoki/py/providers/fanbox.py | 130 | ||||
| -rw-r--r-- | src/shoki/py/providers/instagram.py | 96 | ||||
| -rw-r--r-- | src/shoki/py/providers/pixiv.py | 113 | ||||
| -rw-r--r-- | src/shoki/py/providers/searx.py | 93 | ||||
| -rw-r--r-- | src/shoki/py/providers/twitter.py | 424 | ||||
| -rw-r--r-- | src/shoki/py/providers/youtube.py | 122 |
7 files changed, 0 insertions, 978 deletions
diff --git a/src/shoki/py/providers/__init__.py b/src/shoki/py/providers/__init__.py deleted file mode 100644 index e69de29..0000000 --- a/src/shoki/py/providers/__init__.py +++ /dev/null diff --git a/src/shoki/py/providers/fanbox.py b/src/shoki/py/providers/fanbox.py deleted file mode 100644 index 7739be3..0000000 --- a/src/shoki/py/providers/fanbox.py +++ /dev/null @@ -1,130 +0,0 @@ -import urllib.parse -import log -from plugin import Plugin -from search import Search -from post import Post, Image - -BASE_URL = "https://api.fanbox.cc/" - -PAGES_PER_QUERY = 2 - -class FanboxSearch(Search): - def __init__(self, plugin, creator_id): - super().__init__() - self.p = plugin - self.creator_id = creator_id - self.pagination_pages = [] - self.pagination_complete = False - - def paginate(self): - res = self.p.get_json(BASE_URL + 'post.paginateCreator', { 'creatorId': self.creator_id }) - if not res or 'body' not in res or len(res['body']) == 0: - return False - for p in res['body']: - params = urllib.parse.parse_qs(urllib.parse.urlparse(p).query) - page = { - 'url': p, - 'params': params, - 'data': None, - 'limit': -1 if 'limit' not in params else int(params['limit'][0]) - } - self.pagination_pages.append(page) - return True - - def load_pagination_page(self, p): - if not p['data']: - res = self.p.get_json(BASE_URL + 'post.listCreator', p['params']) - if not res: - return False - p['data'] = res['body']['items'] - # The get_post_at_index logic assumes limit will always be exact. - p['limit'] = len(p['data']) - return True - - def get_post_at_index(self, index): - if not self.pagination_complete: - if not self.paginate(): - return None - self.pagination_complete = True - page = None - count = 0 - for p in self.pagination_pages: - if p['limit'] == -1: - if not self.load_pagination_page(p): - return None - ncount = count + p['limit'] - if ncount > index: - page = p - break - count = ncount - if not page or not self.load_pagination_page(page) or count + page['limit'] <= index: - return None - return page['data'][index - count] - - def add_images_to_map(self, body): - if 'imageMap' not in body: - return - for k in body['imageMap'].keys(): - self.p.image_map[k] = body['imageMap'][k]['originalUrl'] - - def load_page(self, index): - self.pages[index] = [] - s = index * PAGES_PER_QUERY - for i in range(s, s + PAGES_PER_QUERY): - p = self.get_post_at_index(i) - # Break to return what we have, which could be less than expected. - if not p: - self.completed = True - break - unique_id = 'fanbox:{}'.format(p['id']) - if unique_id not in self.p.unique_id_map: - res = self.p.get_json( - BASE_URL + 'post.info', { 'postId': p['id'] } - ) - if not res or 'body' not in res: - # TODO: Log this. - continue - if res['body']['type'] == 'article': - self.add_images_to_map(res['body']['body']) - self.p.unique_id_map[unique_id] = res['body'] - # TODO: Should store Post object here instead. - body = self.p.unique_id_map[unique_id] - text = '' - media = [] - # If body is None, that post is restricted (not subbed or from a higher tier than your sub) - if body['body']: - if body['type'] == 'image': - text = body['body']['text'] - for i in body['body']['images']: - media.append(Image(url=i['originalUrl'], thumbnail_url='')) - elif body['type'] == 'article': - for b in body['body']['blocks']: - if b['type'] == 'p': - text += b['text'] + '\n' - elif b['type'] == 'image': - media.append(Image(url=self.p.image_map[b['imageId']], thumbnail_url='')) - else: - log.warn('Unhandled block type.') - else: - log.warn('Unhandled post type.') - # TODO: Raw response - self.pages[index].append(Post( - unique_id=unique_id, url='', title=body['title'], text=text, media=media)) - return True - -class FanboxPlugin(Plugin): - def __init__(self, sessid): - super().__init__() - self.image_map = {} - self.headers.update({ - 'accept': 'application/json, text/plain, */*', - 'accept-encoding': 'gzip, deflate, br', - 'accept-language': 'ja,en-US;q=0.9,en;q=0.8', - 'origin': 'https://www.fanbox.cc', - 'referer': 'https://www.fanbox.cc/', - 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.102 Safari/537.36' - }) - self.cookies['FANBOXSESSID'] = sessid - - def search(self, query, *extra_args): - return FanboxSearch(self, query) diff --git a/src/shoki/py/providers/instagram.py b/src/shoki/py/providers/instagram.py deleted file mode 100644 index 0146ba3..0000000 --- a/src/shoki/py/providers/instagram.py +++ /dev/null @@ -1,96 +0,0 @@ -import os -import config -import log -from plugin import Plugin -from search import Search -from instagrapi import Client -from instagrapi.exceptions import UserNotFound, LoginRequired, ChallengeRequired -from post import User, Post, Image, Video - -REACH_LIMIT = 3 - -class InstagramSearch(Search): - def __init__(self, cl, pk): - super().__init__() - self.cl = cl - self.pk = pk - self.pagination_page = 0 - self.pagination_cursor = None - - def load_page(self, index): - if index >= self.pagination_page + REACH_LIMIT: - return False - request_satisfied = False - for _ in range(self.pagination_page, index + 1): - if self.pagination_page > 0 and not self.pagination_cursor: - self.completed = True - return False - try: - media, self.pagination_cursor = self.cl.user_medias_paginated_v1( - self.pk, 0, end_cursor=self.pagination_cursor) - except LoginRequired: - return False - except ChallengeRequired: - return False - self.pages[self.pagination_page] = [] - for m in media: - unique_id = 'instagram:p:{}'.format(m.id) - media = [] - if m.media_type == 8: # Album - for r in m.resources: - if r.media_type == 1: # Photo - media.append(Image(url=r.thumbnail_url)) - elif r.media_type == 2: # Video - media.append(Video(url=r.video_url, thumbnail_url=r.thumbnail_url)) - elif m.media_type == 2: # Video - #print(m.product_type) # feed, igtv, or clips - media.append(Video(url=m.video_url, thumbnail_url=m.thumbnail_url)) - elif m.media_type == 1: # Photo - media.append(Image(url=m.thumbnail_url)) - kwargs = {} - kwargs['unique_id'] = unique_id - kwargs['raw_responses'] = { 'tile': m.json() } - kwargs['url'] = 'https://www.instagram.com/p/{}'.format(m.code) - 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 - kwargs['author'] = user - kwargs['title'] = m.title - kwargs['text'] = m.caption_text - kwargs['media'] = media - self.pages[self.pagination_page].append(Post(**kwargs)) - if self.pagination_page == index: - request_satisfied = True - self.pagination_page += 1 - return request_satisfied - -class InstagramPlugin(Plugin): - def __init__(self): - super().__init__() - self.cl = Client() - self.cl.set_user_agent(config.INSTAGRAM_USER_AGENT) - - def init(self): - if os.path.exists(config.INSTAGRAM_SETTINGS_PATH): - self.cl.load_settings(config.INSTAGRAM_SETTINGS_PATH) - if self.cl.login(config.INSTAGRAM_USERNAME, config.INSTAGRAM_PASSWORD): - return True - if not self.cl.login(config.INSTAGRAM_USERNAME, config.INSTAGRAM_PASSWORD): - return False - self.cl.dump_settings(config.INSTAGRAM_SETTINGS_PATH) - return True - - def search(self, query, *extra_args): - try: - user = self.cl.user_info_by_username_v1(query) - except UserNotFound: - return None - except LoginRequired: - log.error('Login required.') - return None - except ChallengeRequired: - log.error('Challenge required.') - return None - return InstagramSearch(self.cl, user.pk) diff --git a/src/shoki/py/providers/pixiv.py b/src/shoki/py/providers/pixiv.py deleted file mode 100644 index 9276d61..0000000 --- a/src/shoki/py/providers/pixiv.py +++ /dev/null @@ -1,113 +0,0 @@ -import json -import requests -from post import User, Image, Post -from search import Search -from plugin import Plugin -from query_parser import QueryParser -from pixivpy3 import * - -class PixivBase(Search): - def __init__(self, userdata, arg): - super().__init__() - self.plugin = userdata - self.next_page = 0 - - def api_load_page(self, index): - return False, None - - def load_page(self, index): - request_satisfied = False - for i in range(self.next_page, index + 1): - if not self.next_qs: - break - self.pages[i] = [] - res, self.next_qs = self.api_load_page(i) - if not res: - break - if i == index: - request_satisfied = True - self.next_page += 1 - return request_satisfied - - def make_post(self, post): - if len(post['meta_pages']) == 0: - post['meta_pages'].append({ - 'image_urls': { - "medium": post['image_urls']['medium'], - "original": post['meta_single_page']['original_image_url'] - } - }) - kwargs = {} - kwargs['unique_id'] = 'pixiv:i:{}'.format(post['id']) - kwargs['raw_responses'] = { 'api': json.dumps(post) } - kwargs['url'] = 'https://www.pixiv.net/en/artworks/{}'.format(post['id']) - kwargs['author'] = User( - unique_id='pixiv:u:{}'.format(post['user']['id']), - username=post['user']['account'], display_name=post['user']['name']) - kwargs['title'] = post['title'] - kwargs['text'] = post['caption'] - kwargs['likes'] = post['total_bookmarks'] - if 'total_comments' in post: - kwargs['comments'] = post['total_comments'] - media = [] - for img in post['meta_pages']: - media.append(Image( - url=img['image_urls']['original'], thumbnail_url=img['image_urls']['medium'])) - kwargs['media'] = media - return Post(**kwargs) - -class PixivFollowing(PixivBase): - def __init__(self, userdata, arg): - super().__init__(userdata, arg) - self.next_qs = { "user_id": arg, "restrict": "public" } - - def api_load_page(self, index): - obj = self.plugin.api.user_following(**self.next_qs) - if 'user_previews' not in obj: - return False, None - for u in obj['user_previews']: - user = u['user'] - self.pages[index].append(User( - unique_id='pixiv:u:{}'.format(user['id']), username=user['account'], - display_name=user['name'])) - return True, self.plugin.api.parse_qs(obj['next_url']) - -class PixivUser(PixivBase): - def __init__(self, userdata, arg): - super().__init__(userdata, arg) - self.next_qs = { 'user_id': arg, 'type': 'illust' } - - def api_load_page(self, index): - obj = self.plugin.api.user_illusts(**self.next_qs) - if 'illusts' not in obj: - return False, None - for post in obj['illusts']: - self.pages[index].append(self.make_post(post)) - return True, self.plugin.api.parse_qs(obj['next_url']) - -class PixivBookmarks(PixivBase): - def __init__(self, userdata, arg): - super().__init__(userdata, arg) - self.next_qs = { 'user_id': arg, 'restrict': 'public' } - - def api_load_page(self, index): - obj = self.plugin.api.user_bookmarks_illust(**self.next_qs) - if 'illusts' not in obj: - return False, None - for post in obj['illusts']: - self.pages[index].append(self.make_post(post)) - return True, self.plugin.api.parse_qs(obj['next_url']) - -class PixivPlugin(Plugin): - def __init__(self, refresh_token): - super().__init__() - self.parser = QueryParser(self, 'following') - self.parser.add_command('following', PixivFollowing) - self.parser.add_command('user', PixivUser) - self.parser.add_command('bookmarks', PixivBookmarks) - self.api = AppPixivAPI() - self.api.requests = requests.Session() - self.api.auth(refresh_token=refresh_token) - - def search(self, query, *extra_args): - return self.parser.parse_query(query) diff --git a/src/shoki/py/providers/searx.py b/src/shoki/py/providers/searx.py deleted file mode 100644 index b536f28..0000000 --- a/src/shoki/py/providers/searx.py +++ /dev/null @@ -1,93 +0,0 @@ -import sys -sys.path.append('../../../subprojects/searxng') -import json -from searx import settings -from searx.utils import gen_useragent -from searx.engines import load_engine, register_engine -from plugin import Plugin -from search import Search -from post import Post, Image - -# TODO: Put these on the Plugin 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'] = { '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 SearxPlugin(Plugin): - 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/shoki/py/providers/twitter.py b/src/shoki/py/providers/twitter.py deleted file mode 100644 index 4770dda..0000000 --- a/src/shoki/py/providers/twitter.py +++ /dev/null @@ -1,424 +0,0 @@ -import json -import config -import http.cookiejar -from datetime import timezone - -import log -from provider import Provider -from search import Search -from post import PostType, DateType, PostRef, Post, User, Image, Video -from query_parser import QueryParser - -# Official Twitter API -#from twitter import Twitter2, TwitterError, OAuth - -# GraphQL API -import snscrape.modules.twitter -from snscrape.base import ScraperException -from snscrape.modules.twitter import (TwitterSearchScraper, TwitterProfileScraper, - TwitterUserScraper, TwitterTweetScraper, TweetRef) - -# Don't attempt to load a page 4 or more pages past the current page. -REACH_LIMIT = 4 - -# 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 TwitterScrapeBase(Search): - # Posts per emulated page because snscrape returns an iterator. - POSTS_PER_PAGE = 8 - - def __init__(self, userdata, arg): - super().__init__() - self.provider = userdata - self.cookies = self.provider.cookies - self.arg = arg - self.iterator = None - self.iter_page = 0 - self.sent_hashes = [] - - def parse_media(self, l): - media = [] - for m in l: - if isinstance(m, snscrape.modules.twitter.Photo): - media.append(Image(url=m.fullUrl, thumbnail_url=m.previewUrl)) - elif isinstance(m, snscrape.modules.twitter.Video): - videos = sorted(m.variants, key=lambda x: x.bitrate if x.bitrate else 0, reverse=True) - media.append(Video(url=videos[0].url, thumbnail_url=m.thumbnailUrl)) - elif isinstance(m, snscrape.modules.twitter.Gif): - gifs = sorted(m.variants, key=lambda x: x.bitrate if x.bitrate else 0, reverse=True) - media.append(Video(url=gifs[0].url, thumbnail_url=m.thumbnailUrl)) - return media - - def make_post(self, tweet): - kwargs = {} - kwargs['unique_id'] = 'twitter:t:{}'.format(tweet.id) - if isinstance(tweet, TweetRef): - return Post(**kwargs) - kwargs['raw_responses'] = tweet.rawResponses - h = tweet.rawResponses['hash'] - if h not in self.sent_hashes: - self.sent_hashes.append(h) - else: - del kwargs['raw_responses']['api'] - kwargs['url'] = tweet.url - kwargs['dates'] = { - DateType.CREATED: tweet.date.replace(tzinfo=timezone.utc).timestamp() - } - author = User(unique_id='twitter:u:{}'.format(tweet.user.id), username='@' + tweet.user.username, - profile_image_url=tweet.user.profileImageUrl) - if tweet.user.displayname: - author.display_name = tweet.user.displayname - kwargs['author'] = author - if tweet.retweetedTweet: - kwargs['type'] = PostType.REPOST - kwargs['post'] = PostRef(unique_id='twitter:t:{}'.format(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(unique_id='twitter:t:{}'.format(tweet.inReplyToTweetId)) - if tweet.quotedTweet: - kwargs['quoted'] = PostRef(unique_id='twitter:t:{}'.format(tweet.quotedTweet.id)) - return Post(**kwargs) - - def add_post(self, i, post): - self.pages[i]['posts'].append(post) - - def add_post_to_list(self, i, post): - self.add_post(i, post) - self.pages[i]['list'].append(post.unique_id) - - def load_page(self, index): - if not self.iterator or index - self.iter_page >= REACH_LIMIT: - return False - if index in self.pages: - return True - # It's possible to load some "pages", but not enough to satisfy the requested index. - request_satisfied = False - for i in range(self.iter_page, index + 1): - if self.completed: - break - self.pages[i] = { 'posts': [], 'list': [] } - 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_list(i, self.make_post(tweet)) - if tweet.retweetedTweet: - self.add_post(i, self.make_post(tweet.retweetedTweet)) - if tweet.quotedTweet: - self.add_post(i, self.make_post(tweet.quotedTweet)) - if i == index: - request_satisfied = True - self.iter_page += 1 - return request_satisfied - -class TwitterScrapeSearch(TwitterScrapeBase): - def __init__(self, userdata, arg): - super().__init__(userdata, arg) - self.iterator = TwitterSearchScraper(self.arg, top=True, cookies=self.cookies).get_items() - -class TwitterScrapeUser(TwitterScrapeBase): - def __init__(self, userdata, arg): - super().__init__(userdata, arg) - if (self.arg.startswith('@')): - self.arg = self.arg[1:] - self.iterator = TwitterProfileScraper(self.arg, cookies=self.cookies).get_items() - #self.iterator = TwitterUserScraper(self.arg, cookies=self.cookies).get_items() - -class TwitterScrapeTweet(TwitterScrapeBase): - def __init__(self, userdata, arg): - super().__init__(userdata, arg) - self.single = True - if (self.arg.startswith('https://')): - self.arg = self.arg[self.arg.rfind('/') + 1:] - question = self.arg.find('?') - if question >= 0: - self.arg = self.arg[:question] - self.iterator = TwitterTweetScraper(self.arg, cookies=self.cookies).get_items() - -class TwitterProvider(Provider): - def __init__(self, access_key, access_secret, consumer_key, consumer_secret, cookies_path): - super().__init__() - self.parser = QueryParser(self, 'search') - self.parser.add_command('search', TwitterScrapeSearch) - self.parser.add_command('user', TwitterScrapeUser) - self.parser.add_command('tweet', TwitterScrapeTweet) - #self.parser.add_command('timeline', TwitterTimeline) - #self.parser.add_command('likes', TwitterLikes) - #self.parser.add_command('following', TwitterFollowing) - #self.t = Twitter2(auth=OAuth(access_key, access_secret, consumer_key, consumer_secret), retry=True) - # Cookies are used by the scrape backend. - cookie_jar = http.cookiejar.MozillaCookieJar() - cookie_jar.load(filename=cookies_path, ignore_expires=True) - for c in cookie_jar: - self.cookies[c.name] = c.value - - def search(self, query, *extra_args): - return self.parser.parse_query(query) diff --git a/src/shoki/py/providers/youtube.py b/src/shoki/py/providers/youtube.py deleted file mode 100644 index 3cbf0c8..0000000 --- a/src/shoki/py/providers/youtube.py +++ /dev/null @@ -1,122 +0,0 @@ -import log -from provider import Provider -from search import Search -from post import Post, PostType, Image, 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): - pass - - def warning(self, msg): - log.warning(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, arg): - super().__init__() - self.provider = userdata - self.query = arg - - def load_page(self, index): - self.pages[index] = { 'posts': [], 'list': [] } - res = self.get_info() - if res: - url = get_playback_url(res) - if not url: - return False - media = [Video(url=url)] - self.pages[index]['posts'].append(Post(type=PostType.POST, unique_id='', url='', title='', text='', media=media)) - return True - -class YoutubeSearch(YoutubeBase): - def __init__(self, userdata, arg): - super().__init__(userdata, arg) - - def get_info(self): - return ydl.extract_info('ytsearch1:{}'.format(self.query), download=False) - -class YoutubeLink(YoutubeBase): - def __init__(self, userdata, arg): - super().__init__(userdata, arg) - - def get_info(self): - return ydl.extract_info(self.query, download=False) - -class YoutubeProvider(Provider): - def __init__(self): - super().__init__() - self.parser = QueryParser(self, 'search') - self.parser.add_command('search', YoutubeSearch) - self.parser.add_command('link', YoutubeLink) - - def search(self, query, *extra_args): - return self.parser.parse_query(query) |