From 2cc385b71e161eae39915bf24de53d5b704e7ea0 Mon Sep 17 00:00:00 2001 From: Andrew Opalach Date: Tue, 5 Aug 2025 10:50:08 -0400 Subject: Twitter module work, various cleanups and comment clarifications Signed-off-by: Andrew Opalach --- src/portal/py/base.py | 3 +- src/portal/py/modules/__init__.py | 17 +- src/portal/py/modules/twitter.py | 263 ++++++++++++------------------- src/portal/py/modules/twitter_api.py | 269 -------------------------------- src/portal/py/modules/twitter_common.py | 177 +++++++++++++++++++++ src/portal/py/modules/twitter_scrape.py | 191 +++++++++++++++++++++++ src/portal/py/modules/youtube.py | 3 + src/portal/py/tests/archive_query.py | 32 +++- src/portal/py/tests/logger.py | 15 -- src/portal/py/tests/test.py | 8 - src/portal/py/tests/touch_all.py | 40 ++--- src/portal/requirements.txt | 1 + 12 files changed, 529 insertions(+), 490 deletions(-) delete mode 100644 src/portal/py/modules/twitter_api.py create mode 100644 src/portal/py/modules/twitter_common.py create mode 100644 src/portal/py/modules/twitter_scrape.py delete mode 100644 src/portal/py/tests/logger.py delete mode 100644 src/portal/py/tests/test.py (limited to 'src/portal') diff --git a/src/portal/py/base.py b/src/portal/py/base.py index f40bc00..a146e8c 100644 --- a/src/portal/py/base.py +++ b/src/portal/py/base.py @@ -110,8 +110,7 @@ class Module(): def add_raw_response(self, data: ParsedJson) -> str: with self.mutex: dump = json.dumps(data, separators=(',', ':'), indent=None) - hash = blake3(dump.encode('utf-8')).digest() - hash_str = binascii.hexlify(hash).decode('ascii') + hash_str = binascii.hexlify(blake3(dump.encode('utf-8')).digest()).decode('ascii') self.raw_responses[hash_str] = dump return hash_str diff --git a/src/portal/py/modules/__init__.py b/src/portal/py/modules/__init__.py index 8a1e3cb..a9b50eb 100644 --- a/src/portal/py/modules/__init__.py +++ b/src/portal/py/modules/__init__.py @@ -1,21 +1,20 @@ import config from modules.youtube import YoutubeModule -from modules.twitter import TwitterScrapeModule -from modules.pixiv_app import PixivAppModule -from modules.pixiv_web import PixivWebModule +#from modules.twitter_scrape import TwitterScrapeModule +#from modules.twitter import TwitterModule +#from modules.pixiv_app import PixivAppModule +#from modules.pixiv_web import PixivWebModule #from modules.fanbox import FanboxModule #from modules.instagram import InstagramModule #from modules.patreon import PatreonModule 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, config.PIXIV_USERID), []), - 'pixiv_app': (PixivAppModule(config.PIXIV_REFRESH_TOKEN), []), +# 'twitter_scrape': (TwitterScrapeModule(config.TWITTER_COOKIES_PATH), []), +# 'twitter': (TwitterModule(config.TWITTER_COOKIES_PATH), []), +# 'pixiv_web': (PixivWebModule(config.PIXIV_SESSID, config.PIXIV_USERID), []), +# 'pixiv_app': (PixivAppModule(config.PIXIV_REFRESH_TOKEN), []), # 'fanbox': (FanboxModule(config.FANBOX_SESSID, config.FANBOX_CF_CLEARANCE), []), # 'instagram': (InstagramModule( # config.INSTAGRAM_USER_AGENT, config.INSTAGRAM_SETTINGS_PATH, diff --git a/src/portal/py/modules/twitter.py b/src/portal/py/modules/twitter.py index 93286c3..f756d13 100644 --- a/src/portal/py/modules/twitter.py +++ b/src/portal/py/modules/twitter.py @@ -1,178 +1,121 @@ -import log +import json +import httpx import http.cookiejar from typing import Optional, Any, Iterator -from datetime import timezone -from base import USER_AGENT, Search, Module -from post import Url, PostType, Date, DateType, DateMeta, PostRef, Post, User, Media, Image, Video +from base import USER_AGENT, Search, Module, ParsedJson, Method +from post import Post from query_parser import QueryParser -from modules.common import get_current_utc_time - -# GraphQL API -from snscrape.base import ScraperException -from snscrape.modules import twitter -from snscrape.modules.twitter import (Tweet, TweetRef, Tombstone, UserRef, - TwitterSearchScraper, TwitterSearchScraperMode, 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 +from modules.twitter_common import parse_graphql_raw_response +class TwitterBase(Search): def __init__(self, userdata: Any) -> None: super().__init__() - self.module: TwitterScrapeModule = userdata - self.iterator: Iterator[Tweet | TweetRef | Tombstone] - self.page: int = 0 - - def create_media_url(self, url: str) -> Url: - format = url.find('format=') - ext = None - if format >= 0: - ext = url[format + 7:] - amp = ext.find('&') - if amp >= 0: - ext = ext[:amp] - return Url(url, ext) - - def parse_media(self, data: list[twitter.Medium]) -> dict[str, Media]: - media = {} - for index, m in enumerate(data): - if isinstance(m, twitter.Photo): - media[str(index)] = 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(index)] = 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['module'] = 'twitter_scrape' - 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 - create_date = tweet.date.replace(tzinfo=timezone.utc).timestamp() - current_date = get_current_utc_time().timestamp() - kwargs['dates'] = [ - Date(DateType.CREATED, (create_date, create_date), DateMeta.NONE), - Date(DateType.RETRIEVED, (current_date, current_date), DateMeta.NONE) - ] - 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) + self.module: TwitterModule = userdata + self.cursor: Optional[dict] = None - 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 index in range(self.page, num + 1): - if self.completed: - break - if index in self.pages: - continue - self.pages[index] = [] - if not self.step_iterator_for_page(index): - break - self.page = index - if index == num: - request_satisfied = True - return request_satisfied - -class TwitterScrapeSearch(TwitterScrapeBase): - def __init__(self, userdata: Any, arg: str) -> None: - super().__init__(userdata) - self.iterator = TwitterSearchScraper(arg, mode=TwitterSearchScraperMode.TOP, cookies=self.module.cookies).get_items() +def build_params(params: dict) -> dict: + return json.dumps(params, separators=(',', ':'), indent=None) -class TwitterScrapeUser(TwitterScrapeBase): +class TwitterUser(TwitterBase): def __init__(self, userdata: Any, arg: str) -> None: super().__init__(userdata) - if arg.startswith('@'): - arg = arg[1:] - self.iterator = TwitterUserScraper(arg, cookies=self.module.cookies).get_items() + self.user_id = '' -class TwitterScrapeProfile(TwitterScrapeBase): - def __init__(self, userdata: Any, arg: str) -> None: - super().__init__(userdata) - if arg.startswith('@'): - arg = arg[1:] - self.iterator = TwitterProfileScraper(arg, cookies=self.module.cookies).get_items() + def check_api_response(self, response: Optional[httpx.Response]) -> Optional[ParsedJson]: + if not response: + return None + try: + obj = response.json() + except JSONDecodeError as e: + log.error(f'JSONDecodeError: {e}.') + return None + return obj + + def do_request(self) -> Optional[ParsedJson]: + variables = { + 'userId': f'{self.user_id}', + 'count': 20, + 'includePromotedContent': False, + 'withClientEventToken': False, + 'withBirdwatchNotes': False, + 'withVoice': True + } + if self.cursor: + variables['cursor'] = self.cursor + features = { + 'rweb_video_screen_enabled': False, + 'payments_enabled': False, + 'profile_label_improvements_pcf_label_in_post_enabled': True, + 'rweb_tipjar_consumption_enabled': True, + 'verified_phone_label_enabled': False, + 'creator_subscriptions_tweet_preview_api_enabled': True, + 'responsive_web_graphql_timeline_navigation_enabled': True, + 'responsive_web_graphql_skip_user_profile_image_extensions_enabled': False, + 'premium_content_api_read_enabled': False, + 'communities_web_enable_tweet_community_results_fetch': True, + 'c9s_tweet_anatomy_moderator_badge_enabled': True, + 'responsive_web_grok_analyze_button_fetch_trends_enabled': False, + 'responsive_web_grok_analyze_post_followups_enabled': True, + 'responsive_web_jetfuel_frame': True, + 'responsive_web_grok_share_attachment_enabled': True, + 'articles_preview_enabled': True, + 'responsive_web_edit_tweet_api_enabled': True, + 'graphql_is_translatable_rweb_tweet_is_translatable_enabled': True, + 'view_counts_everywhere_api_enabled': True, + 'longform_notetweets_consumption_enabled': True, + 'responsive_web_twitter_article_tweet_consumption_enabled': True, + 'tweet_awards_web_tipping_enabled': False, + 'responsive_web_grok_show_grok_translated_post': False, + 'responsive_web_grok_analysis_button_from_backend': True, + 'creator_subscriptions_quote_tweet_preview_enabled': False, + 'freedom_of_speech_not_reach_fetch_enabled': True, + 'standardized_nudges_misinfo': True, + 'tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled': True, + 'longform_notetweets_rich_text_read_enabled': True, + 'longform_notetweets_inline_media_enabled': True, + 'responsive_web_grok_image_annotation_enabled': True, + 'responsive_web_grok_community_note_auto_translation_is_enabled': False, + 'responsive_web_enhance_cards_enabled': False + } + fieldToggles = { + 'withArticlePlainText': False + } + params = { + 'variables': build_params(variables), + 'features': build_params(features), + 'fieldToggles': build_params(fieldToggles) + } + url = "https://x.com/i/api/graphql/DdW3L2nfGW6FZDcSSSG4yg/UserMedia" + return self.check_api_response(self.module.do_request(Method.GET, url, params=params)) -class TwitterScrapeTweet(TwitterScrapeBase): - def __init__(self, userdata: Any, arg: str) -> None: - 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() + #print(json.dumps(media, indent=4)) + def request_page(self, num: int) -> bool: + media = self.do_request() + #print(json.dumps(media, indent=4)) + response_hash = self.module.add_raw_response(media) + tps = parse_graphql_raw_response(media, response_hash, None, self.module) + if not tps.cursor: + self.completed = True + if len(tps.page) > 0: + self.cursor = tps.cursor + else: + self.completed = True + return False + self.pages[num] = tps.page + return True -class TwitterScrapeModule(Module): +class TwitterModule(Module): def __init__(self, cookies_path: str) -> None: 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 + self.parser: QueryParser = QueryParser(self, 'user') + self.parser.add_command('user', TwitterUser) + # @TODO: Cookie import from browser. + #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) diff --git a/src/portal/py/modules/twitter_api.py b/src/portal/py/modules/twitter_api.py deleted file mode 100644 index 6d04855..0000000 --- a/src/portal/py/modules/twitter_api.py +++ /dev/null @@ -1,269 +0,0 @@ -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/twitter_common.py b/src/portal/py/modules/twitter_common.py new file mode 100644 index 0000000..5d1587d --- /dev/null +++ b/src/portal/py/modules/twitter_common.py @@ -0,0 +1,177 @@ +import sys +import json +import email.utils +from json import JSONDecodeError +from typing import Optional, Any +from base import USER_AGENT, Search, Module, Method, ParsedJson +from post import Url, PostType, PostRef, Date, DateType, DateMeta, Post, User, Media, Image, Video, PostEncoder +from modules.common import get_current_utc_time + +class TwitterParserState(): + def __init__(self, response_hash: str, mtime: Optional[float], module: Module): + self.response_hash = response_hash; + self.mtime: float = mtime if mtime else get_current_utc_time().timestamp() + self.module: Module = module + self.cursor: Optional[dict] = None + self.page: list[str] = [] + +def parse_media_url_https(url: str) -> (Url, Url): + question = url.rfind('?') + if question >= 0: + url = url[0:question] + ext = url.rsplit('.')[-1] + return Url(f'{url}?format={ext}&name=orig', ext), Url(f'{url}?format={ext}&name=small', ext) + +def create_media_graphql(m: ParsedJson) -> Media: + original, thumbnail = parse_media_url_https(m['media_url_https']) + if m['type'] == 'photo': + return Image(url=original, thumbnail_url=thumbnail) + elif m['type'] == 'video' or m['type'] == 'animated_gif': + variants = sorted(m['video_info']['variants'], key=lambda x: x['bitrate'] if 'bitrate' in x else 0, reverse=True) + return Video(url=Url(variants[0]['url']), thumbnail_url=thumbnail) + return None + +def create_post_graphql(data: ParsedJson, tps: TwitterParserState) -> str: + kwargs = {} + kwargs['module'] = 'twitter' + tweet_id = data['rest_id'] + unique_id = f'twitter:t:{tweet_id}' + kwargs['unique_id'] = unique_id + kwargs['raw_responses'] = {} + kwargs['raw_responses']['graphql'] = tps.response_hash + kwargs['dates'] = [Date(DateType.RETRIEVED, (tps.mtime, tps.mtime), DateMeta.NONE)] + if 'core' in data and 'user_results' in data['core']: + user_data = data['core']['user_results']['result'] + if user_data['__typename'] != 'User': + print(f'Unexpected type {user_data['__typename']} when expecting User') + user_id = user_data['rest_id'] + user = User(unique_id=f'twitter:u:{user_id}') + if 'core' in user_data: + user.username = user_data['core']['screen_name'] + user.name = user_data['core']['name'] + else: + user.username = user_data['legacy']['screen_name'] + user.name = user_data['legacy']['name'] + if 'avatar' in user_data: + user.profile_picture_url = Url(user_data['avatar']['image_url']) + else: + user.profile_picture_url = Url(user_data['legacy']['profile_image_url_https']) + kwargs['url'] = f'https://twitter.com/{user.username}/status/{tweet_id}' + kwargs['author'] = user + if 'legacy' not in data or ('__typename' in data and (data['__typename'] == 'TweetTombstone' or data['__typename'] == 'TweetUnavailable')): + kwargs['type'] = PostType.TOMBSTONE + post = Post(**kwargs) + tps.module.add_to_map(unique_id, post) + return unique_id + legacy = data['legacy'] + create_date = email.utils.parsedate_to_datetime(legacy['created_at']).timestamp() + kwargs['dates'].append(Date(DateType.CREATED, (create_date, create_date), DateMeta.NONE)) + # @TODO: Find example of retweeted_status_id. + is_retweet = 'retweeted_status_result' in legacy or 'retweeted_status_id' in legacy + if is_retweet: + kwargs['type'] = PostType.REPOST + if 'retweeted_status_result' in legacy and 'result' in legacy['retweeted_status_result']: + retweet_id = PostRef(create_post_graphql(legacy['retweeted_status_result']['result'], tps)) + elif 'retweeted_status_id' in legacy: + retweet_id = PostRef(f'twitter:t:{legacy['retweeted_status_id']}') + else: + print('Unknown ID for supposed retweet.') + retweet_id = PostRef() + post = Post(**kwargs) + tps.module.add_to_map(unique_id, post) + return unique_id + kwargs['type'] = PostType.POST + kwargs['text'] = legacy.get('full_text', '') + kwargs['likes'] = legacy.get('favorite_count', None) + kwargs['bookmarks'] = legacy.get('bookmark_count', None) + kwargs['reposts'] = legacy.get('retweet_count', None) + kwargs['quotes'] = legacy.get('quote_count', None) + kwargs['comments'] = legacy.get('reply_count', None) + if 'views' in data: + kwargs['views'] = data['views'].get('count', None) + merged_entities = [] + if 'extended_entities' in legacy and 'media' in legacy['extended_entities']: + merged_entities.extend(legacy['extended_entities']['media']) + merged_entities.extend(legacy['entities'].get('media', [])) + seen_media_ids = [] + kwargs['media'] = {} + for m in merged_entities: + if m['id_str'] in seen_media_ids: + continue + seen_media_ids.append(m['id_str']) + kwargs['media'][m.get('media_key', m['id_str'])] = create_media_graphql(m) + # quotedRefResult references the tweet that quoted this tweet, not the tweet this tweet is quoting. + has_quoted = 'quoted_status_result' in legacy or 'quoted_status_id_str' in legacy + if has_quoted: + if 'quoted_status_result' in legacy and 'result' in legacy['quoted_status_result']: + quoted_id = PostRef(create_post_graphql(legacy['quoted_status_result']['result'], tps)) + elif 'quoted_status_id_str' in legacy: + quoted_id = PostRef(f'twitter:t:{legacy['quoted_status_id_str']}') + else: + print('Unknown ID for supposed quoted tweet.') + quoted_id = PostRef() + if 'in_reply_to_status_id_str' in legacy: + kwargs['in_reply_to'] = PostRef(f'twitter:t:{legacy['in_reply_to_status_id_str']}') + post = Post(**kwargs) + tps.module.add_to_map(unique_id, post) + return unique_id + +def parse_graphql_tweet(tweet: ParsedJson, tps: TwitterParserState): + entry_id = tweet['entryId'] + if 'item' not in tweet or 'itemContent' not in tweet['item']: + print(f'Unhandled Timeline entry {entry_id}') + return + content = tweet['item']['itemContent'] + if content['__typename'] != 'TimelineTweet': + print(f'Unhandled Timeline content type {content['__typename']}') + return + result = content['tweet_results']['result'] + if result['__typename'] != 'Tweet': + print(f'Unhandled Tweet type {result['__typename']}') + return + tps.page.append(create_post_graphql(result, tps)) + +def parse_graphql_timeline_v2(timeline: ParsedJson, tps: TwitterParserState): + for ins in timeline['timeline']['instructions']: + ins_type = ins['type'] + nops = ['TimelineClearCache', 'TimelineTerminateTimeline', 'TimelinePinEntry'] + if ins_type in nops: + print(f'no-op: {ins_type}') + elif ins_type == 'TimelineAddEntries': + for entry in ins['entries']: + entry_id = entry['entryId'] + if entry_id.startswith('cursor-top-'): + pass + elif entry_id.startswith('cursor-bottom-'): + tps.cursor = entry['content']['value'] + elif 'content' in entry and entry['content']['entryType'] == 'TimelineTimelineModule': + for item in entry['content']['items']: + parse_graphql_tweet(item, tps) + elif ins_type == 'TimelineAddToModule': + for item in ins['moduleItems']: + parse_graphql_tweet(item, tps) + else: + print(f'Unknown instruction: {ins_type}') + +def parse_graphql_raw_response(obj: ParsedJson, response_hash: str, mtime: Optional[float], module: Module) -> TwitterParserState: + tps = TwitterParserState(response_hash, mtime, module) + if 'data' in obj: + if 'user' in obj['data']: + user = obj['data']['user'] + if 'result' in user: + parse_graphql_timeline_v2( + user['result']['timeline_v2'] if 'timeline_v2' in user['result'] else user['result']['timeline'], tps) + if len(tps.page) == 0: + print(f'Response with 0 tweets:\n{json.dumps(obj)}') + else: + print('Invalid graphql user.') + else: + print(f'Unexpected response:\n{json.dumps(obj)}') + return tps + +def parse_v1_raw_response(obj: ParsedJson, reponse_hash: str, mtime: Optional[float]): + pass + +#target = sys.argv[1] +#with open(target, 'r') as f: +# parse_graphql_raw_response(json.loads(f.read())[0], 'AAAAAAAA', None, Module()) diff --git a/src/portal/py/modules/twitter_scrape.py b/src/portal/py/modules/twitter_scrape.py new file mode 100644 index 0000000..93286c3 --- /dev/null +++ b/src/portal/py/modules/twitter_scrape.py @@ -0,0 +1,191 @@ +import log +import http.cookiejar +from typing import Optional, Any, Iterator +from datetime import timezone +from base import USER_AGENT, Search, Module +from post import Url, PostType, Date, DateType, DateMeta, PostRef, Post, User, Media, Image, Video +from query_parser import QueryParser +from modules.common import get_current_utc_time + +# GraphQL API +from snscrape.base import ScraperException +from snscrape.modules import twitter +from snscrape.modules.twitter import (Tweet, TweetRef, Tombstone, UserRef, + TwitterSearchScraper, TwitterSearchScraperMode, 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) -> None: + super().__init__() + self.module: TwitterScrapeModule = userdata + self.iterator: Iterator[Tweet | TweetRef | Tombstone] + self.page: int = 0 + + def create_media_url(self, url: str) -> Url: + format = url.find('format=') + ext = None + if format >= 0: + ext = url[format + 7:] + amp = ext.find('&') + if amp >= 0: + ext = ext[:amp] + return Url(url, ext) + + def parse_media(self, data: list[twitter.Medium]) -> dict[str, Media]: + media = {} + for index, m in enumerate(data): + if isinstance(m, twitter.Photo): + media[str(index)] = 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(index)] = 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['module'] = 'twitter_scrape' + 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 + create_date = tweet.date.replace(tzinfo=timezone.utc).timestamp() + current_date = get_current_utc_time().timestamp() + kwargs['dates'] = [ + Date(DateType.CREATED, (create_date, create_date), DateMeta.NONE), + Date(DateType.RETRIEVED, (current_date, current_date), DateMeta.NONE) + ] + 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 index in range(self.page, num + 1): + if self.completed: + break + if index in self.pages: + continue + self.pages[index] = [] + if not self.step_iterator_for_page(index): + break + self.page = index + if index == num: + request_satisfied = True + return request_satisfied + +class TwitterScrapeSearch(TwitterScrapeBase): + def __init__(self, userdata: Any, arg: str) -> None: + super().__init__(userdata) + self.iterator = TwitterSearchScraper(arg, mode=TwitterSearchScraperMode.TOP, cookies=self.module.cookies).get_items() + +class TwitterScrapeUser(TwitterScrapeBase): + def __init__(self, userdata: Any, arg: str) -> None: + 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) -> None: + 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) -> None: + 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) -> None: + 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': USER_AGENT + } + } diff --git a/src/portal/py/modules/youtube.py b/src/portal/py/modules/youtube.py index 9d5a135..77e17b3 100644 --- a/src/portal/py/modules/youtube.py +++ b/src/portal/py/modules/youtube.py @@ -5,6 +5,7 @@ from base import Search, Module, ParsedJson from post import Url, Post, PostType, Media, Video from query_parser import QueryParser from yt_dlp import YoutubeDL +from yt_dlp.cookies import extract_cookies_from_browser class YDLLogger(): def debug(self, message: str) -> None: @@ -31,6 +32,8 @@ ydl_opts = { # 'cookiefile': '' } +#print(extract_cookies_from_browser('firefox')) + # https://github.com/yt-dlp/yt-dlp/issues/4103 ydl = YoutubeDL(ydl_opts) diff --git a/src/portal/py/tests/archive_query.py b/src/portal/py/tests/archive_query.py index 3de4329..8ecbb52 100644 --- a/src/portal/py/tests/archive_query.py +++ b/src/portal/py/tests/archive_query.py @@ -1,20 +1,35 @@ import os import sys import signal -import json import threading +import json import gzip from datetime import datetime, timezone sys.path.append('../') from base import Method from post import PostEncoder, PostType, DateType, User from modules import ALL_MODULES -from tests.logger import Logger + +class Logger(): + def __init__(self, path): + self.file = open(path, 'a+') + self.mutex = threading.Lock() + + def write(self, message): + with self.mutex: + print(message) + self.file.write(message + '\n') + self.file.flush() + + def close(self): + self.file.close() mode = 'pixiv_web' #mode = 'fanbox' #mode = 'patreon' #mode = 'instagram' +#mode = 'twitter' +#cmd = 'profile' cmd = 'user' #cmd = 'bookmarks' #cmd = 'search' @@ -151,8 +166,8 @@ class QueryDownloadThread(threading.Thread): while completed: self.log.write(f'Starting page {num} of {arg}.') page = search.get_page(num) - for hash, response in module.raw_responses.items(): - raw_response_path = f'{self.output_dir}/raw_responses/{hash}.json.gz' + for hash_str, response in module.raw_responses.items(): + raw_response_path = f'{self.output_dir}/raw_responses/{hash_str}.json.gz' if not os.path.isfile(raw_response_path): if not self.write_to_file(raw_response_path, 'wb+', response, True): completed = False @@ -161,8 +176,8 @@ class QueryDownloadThread(threading.Thread): break num += 1 for unique_id in page: - if unique_id == 'twitter:t:1597637140833529856': - self.log.write('---------Hit target---------') + #if unique_id == '': + # self.log.write('---------Hit target---------') post = module.get_item(unique_id) if not post: self.log.write(f'Missing post with id {unique_id}.') @@ -182,8 +197,8 @@ class QueryDownloadThread(threading.Thread): #RUNTIME_PATH = './run' RUNTIME_PATH = '/mnt/store/files/tmp/run' -LOG_FILE = f'{RUNTIME_PATH}/archive4.log' -ARG_FILE = f'{RUNTIME_PATH}/completed_args4.log' +LOG_FILE = f'{RUNTIME_PATH}/archive5.log' +ARG_FILE = f'{RUNTIME_PATH}/completed_args5.log' def read_completed_args(path): args = [] @@ -214,6 +229,7 @@ def download_args(): output_dir = sys.argv[1] if not os.path.isdir(output_dir): os.mkdir(output_dir) + if not os.path.isdir(f'{output_dir}/raw_responses'): os.mkdir(f'{output_dir}/raw_responses') args = [] diff --git a/src/portal/py/tests/logger.py b/src/portal/py/tests/logger.py deleted file mode 100644 index 3adf792..0000000 --- a/src/portal/py/tests/logger.py +++ /dev/null @@ -1,15 +0,0 @@ -import threading - -class Logger(): - def __init__(self, path): - self.file = open(path, 'a+') - self.mutex = threading.Lock() - - def write(self, message): - with self.mutex: - print(message) - self.file.write(message + '\n') - self.file.flush() - - def close(self): - self.file.close() diff --git a/src/portal/py/tests/test.py b/src/portal/py/tests/test.py deleted file mode 100644 index d3e5640..0000000 --- a/src/portal/py/tests/test.py +++ /dev/null @@ -1,8 +0,0 @@ -#from twitter.scraper import Scraper -#scraper = Scraper(cookies = { -# 'ct0': '', -# 'auth_token': '' -#}) -# -#media = scraper.media([1557919548904419328], limit=10) -#print(media) diff --git a/src/portal/py/tests/touch_all.py b/src/portal/py/tests/touch_all.py index 6d6bfa1..158c7dd 100644 --- a/src/portal/py/tests/touch_all.py +++ b/src/portal/py/tests/touch_all.py @@ -3,36 +3,24 @@ from typing import Optional, Any from post import PostEncoder from modules import ALL_MODULES -''' -post = module.get_item(page[0]) -for key in post.media.keys(): - download_params = module.get_download(post.unique_id, key) - for url in download_params['urls']: - #response = module.do_request(Method.GET, url.url) - r = httpx.get(url.url, headers=download_params['headers']) - if r.status_code != 200: - print('error{}'.format(r.status_code)) - else: - with open(f'test_{key}.{url.ext}', 'wb+') as f: - f.write(r.content) -''' - def print_page(module: Any, page: Optional[list[str]]) -> None: if not page: print('Page is None') return - count = 0 for unique_id in page: post = module.get_item(unique_id) print(json.dumps(post, indent=4, cls=PostEncoder)) - count = count + 1 - if count == 2: - break -''' Blocked on certain endpoints, snscrape is outdated. module = ALL_MODULES['twitter'][0] module.init() +search = module.search('user:de7_7_7e') +print_page(module, search.get_page(0)) + +''' Blocked on most endpoints, snscrape is outdated. +module = ALL_MODULES['twitter_scrape'][0] +module.init() + search = module.search('search:#小関麗奈誕生祭2024') print_page(module, search.get_page(0)) search = module.search('user:xuuikie') @@ -77,3 +65,17 @@ print_page(module, search.get_page(0)) search = module.search('illust:100455669') # Deleted. print_page(module, search.get_page(0)) ''' + +''' +post = module.get_item(page[0]) +for key in post.media.keys(): + download_params = module.get_download(post.unique_id, key) + for url in download_params['urls']: + #response = module.do_request(Method.GET, url.url) + r = httpx.get(url.url, headers=download_params['headers']) + if r.status_code != 200: + print('error{}'.format(r.status_code)) + else: + with open(f'test_{key}.{url.ext}', 'wb+') as f: + f.write(r.content) +''' diff --git a/src/portal/requirements.txt b/src/portal/requirements.txt index 0d31ac1..c2e2915 100644 --- a/src/portal/requirements.txt +++ b/src/portal/requirements.txt @@ -1,2 +1,3 @@ httpx[http2,brotli,zstd] blake3 +pillow -- cgit v1.2.3-101-g0448