diff options
Diffstat (limited to 'src/portal/py/modules')
| -rw-r--r-- | src/portal/py/modules/__init__.py | 12 | ||||
| -rw-r--r-- | src/portal/py/modules/fanbox.py | 10 | ||||
| -rw-r--r-- | src/portal/py/modules/instagram.py | 18 | ||||
| -rw-r--r-- | src/portal/py/modules/patreon.py | 8 | ||||
| -rw-r--r-- | src/portal/py/modules/pixiv_app.py | 151 | ||||
| -rw-r--r-- | src/portal/py/modules/pixiv_web.py | 78 | ||||
| -rw-r--r-- | src/portal/py/modules/twitter.py | 61 |
7 files changed, 177 insertions, 161 deletions
diff --git a/src/portal/py/modules/__init__.py b/src/portal/py/modules/__init__.py index fee4c22..8a1e3cb 100644 --- a/src/portal/py/modules/__init__.py +++ b/src/portal/py/modules/__init__.py @@ -1,9 +1,9 @@ import config from modules.youtube import YoutubeModule -#from modules.twitter import TwitterScrapeModule -#from modules.pixiv_app import PixivAppModule -#from modules.pixiv_web import PixivWebModule +from modules.twitter import TwitterScrapeModule +from modules.pixiv_app import PixivAppModule +from modules.pixiv_web import PixivWebModule #from modules.fanbox import FanboxModule #from modules.instagram import InstagramModule #from modules.patreon import PatreonModule @@ -13,9 +13,9 @@ ALL_MODULES = { # 'twitter': (TwitterModule( # config.TWITTER_ACCESS_TOKEN, config.TWITTER_ACCESS_TOKEN_SECRET, # config.TWITTER_CONSUMER_TOKEN, config.TWITTER_CONSUMER_TOKEN_SECRET), []), -# 'twitter': (TwitterScrapeModule(config.TWITTER_COOKIES_PATH), []), -# 'pixiv_web': (PixivWebModule(config.PIXIV_SESSID, config.PIXIV_USERID), []), -# 'pixiv_app': (PixivAppModule(config.PIXIV_REFRESH_TOKEN), []), + 'twitter': (TwitterScrapeModule(config.TWITTER_COOKIES_PATH), []), + 'pixiv_web': (PixivWebModule(config.PIXIV_SESSID, config.PIXIV_USERID), []), + 'pixiv_app': (PixivAppModule(config.PIXIV_REFRESH_TOKEN), []), # 'fanbox': (FanboxModule(config.FANBOX_SESSID, config.FANBOX_CF_CLEARANCE), []), # 'instagram': (InstagramModule( # config.INSTAGRAM_USER_AGENT, config.INSTAGRAM_SETTINGS_PATH, diff --git a/src/portal/py/modules/fanbox.py b/src/portal/py/modules/fanbox.py index 9d233bb..01e2b8c 100644 --- a/src/portal/py/modules/fanbox.py +++ b/src/portal/py/modules/fanbox.py @@ -31,7 +31,7 @@ class FanboxBase(Search): unique_id = f'fanbox:p:{post_id}' kwargs['unique_id'] = unique_id kwargs['raw_responses'] = {} - kwargs['raw_responses']['api'] = json.dumps(data) + kwargs['raw_responses']['api'] = json.dumps(data, separators=(',', ':'), indent=None) publish_date = reformat_pixiv_date(data['publishedDatetime']).timestamp() current_date = get_current_utc_time().timestamp() kwargs['dates'] = [ @@ -56,12 +56,12 @@ class FanboxBase(Search): media = {} if data['type'] == 'image': text = data['body']['text'] - for i, value in enumerate(data['body']['images']): - media[str(i)] = Image(url=Url(value['originalUrl'], value['extension']), thumbnail_url=Url(value['thumbnailUrl'])) + for index, value in enumerate(data['body']['images']): + media[str(index)] = Image(url=Url(value['originalUrl'], value['extension']), thumbnail_url=Url(value['thumbnailUrl'])) elif data['type'] == 'file': text = data['body']['text'] - for i, value in enumerate(data['body']['files']): - media[str(i)] = File(url=Url(value['url'], value['extension']), name=value['name']) + for index, value in enumerate(data['body']['files']): + media[str(index)] = File(url=Url(value['url'], value['extension']), name=value['name']) elif data['type'] == 'article': for block in data['body']['blocks']: if block['type'] == 'p': diff --git a/src/portal/py/modules/instagram.py b/src/portal/py/modules/instagram.py index 014ffc1..7d71f9f 100644 --- a/src/portal/py/modules/instagram.py +++ b/src/portal/py/modules/instagram.py @@ -22,7 +22,7 @@ class InstagramSearch(Search): if num >= self.page + self.REACH_LIMIT: return False request_satisfied = False - for i in range(self.page, num + 1): + for index in range(self.page, num + 1): try: media, self.cursor = self.module.cl.user_medias_paginated_v1(self.pk, 0, end_cursor=self.cursor) #media, self.cursor = self.module.cl.user_clips_paginated_v1(self.pk, amount=12, end_cursor=self.cursor) @@ -34,24 +34,24 @@ class InstagramSearch(Search): return False if not self.cursor: self.completed = True - self.pages[i] = [] + self.pages[index] = [] for m in media: kwargs = {} kwargs['type'] = PostType.POST - unique_id = 'instagram:p:{}'.format(m.id) - #unique_id = 'instagram:r:{}'.format(m.id) + unique_id = f'instagram:p:{m.id}' + #unique_id = f'instagram:r:{m.id}' kwargs['unique_id'] = unique_id kwargs['raw_responses'] = {} kwargs['raw_responses']['tile'] = m.json() - kwargs['url'] = 'https://www.instagram.com/p/{}'.format(m.code) - #kwargs['url'] = 'https://www.instagram.com/reel/{}'.format(m.code) + kwargs['url'] = f'https://www.instagram.com/p/{m.code}' + #kwargs['url'] = f'https://www.instagram.com/reel/{m.code}' create_date = m.taken_at.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) ] - user = User(unique_id='instagram:u:{}'.format(m.user.pk)) + user = User(unique_id=f'instagram:u:{m.user.pk}') if m.user.username: user.username = m.user.username if m.user.full_name: @@ -79,9 +79,9 @@ class InstagramSearch(Search): kwargs['media'] = media post = Post(**kwargs) self.module.add_to_map(unique_id, post) - self.pages[i].append(unique_id) + self.pages[index].append(unique_id) self.page += 1 - if i == num: + if index == num: request_satisfied = True return request_satisfied diff --git a/src/portal/py/modules/patreon.py b/src/portal/py/modules/patreon.py index 0e4345a..bf2f0df 100644 --- a/src/portal/py/modules/patreon.py +++ b/src/portal/py/modules/patreon.py @@ -126,7 +126,7 @@ class PatreonUser(PatreonBase): return False request_satisfied = False - for i in range(self.page, num + 1): + for index in range(self.page, num + 1): url = f'{BASE_URL}/posts' #'filter[accessible_by_user_id]': self.module.user_id, params = { @@ -187,17 +187,17 @@ class PatreonUser(PatreonBase): else: log.warn(f'Unhandled include type {inc['type']}.') - self.pages[i] = [] + self.pages[index] = [] for entry in obj['data']: if entry['type'] != 'post': log.warn(f'Unhandled entry type {entry['type']}.') continue post = self.create_post(entry, hash) if post: - self.pages[i].append(post.unique_id) + self.pages[index].append(post.unique_id) self.page += 1 - if i == num: + if index == num: request_satisfied = True return request_satisfied diff --git a/src/portal/py/modules/pixiv_app.py b/src/portal/py/modules/pixiv_app.py index 02d5254..7bd341b 100644 --- a/src/portal/py/modules/pixiv_app.py +++ b/src/portal/py/modules/pixiv_app.py @@ -1,9 +1,9 @@ import json from typing import Optional, Any from base import USER_AGENT, Search, Module, ParsedJson -from post import Url, PostType, DateType, User, Post, Image, Tag, TagType +from post import Url, PostType, Date, DateType, DateMeta, User, Post, Image, Tag, TagType from query_parser import QueryParser -from pixivpy3 import AppPixivAPI +from pixivpy3 import AppPixivAPI, models from modules.common import get_current_utc_time, reformat_pixiv_date class PixivAppBase(Search): @@ -18,63 +18,77 @@ class PixivAppBase(Search): def request_page(self, num: int) -> bool: request_satisfied = False - for i in range(self.page, num + 1): + for index in range(self.page, num + 1): if not self.next_qs: break - if i in self.pages: + if index in self.pages: continue - self.pages[i] = [] - success, self.next_qs = self.api_request_page(i) + self.pages[index] = [] + success, self.next_qs = self.api_request_page(index) if not success: self.errored = True break - if i == num: + if index == num: request_satisfied = True self.page += 1 return request_satisfied - def create_media_url(self, url: str) -> Url: - return Url(url, url.split('.')[-1]) - def create_tag(self, data: ParsedJson) -> Tag: - tag = Tag(type=TagType.GENERAL, name=data['name']) + kwargs = {} + kwargs['type'] = TagType.GENERAL + kwargs['name'] = data['name'] + kwargs['alts'] = {} if data['translated_name']: - tag.alts['en'] = data['translated_name'] - return tag + kwargs['alts']['en'] = data['translated_name'] + return Tag(**kwargs) - 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'] - } - }) + # Other objects returned from pixivpy3 are likely also ModelT but it seems + # so inconsistent internally for now keep some params as ParsedJson. + def create_post(self, data: ParsedJson | models.ModelT, retrieved_date: Optional[float] = None) -> Post: + is_deleted = not data['visible'] kwargs = {} + kwargs['module'] = 'pixiv_app' kwargs['type'] = PostType.POST - unique_id = 'pixiv:i:{}'.format(data['id']) + unique_id = f'pixiv:i:{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: reformat_pixiv_date(data['create_date']).timestamp(), - DateType.RETRIEVED: get_current_utc_time().timestamp() - } + try: + kwargs['raw_responses']['api'] = json.dumps(data, separators=(',', ':'), indent=None) + except TypeError: + kwargs['raw_responses']['api'] = data.model_dump_json(indent=None) + kwargs['url'] = f'https://www.pixiv.net/en/artworks/{data['id']}' + if not retrieved_date: + retrieved_date = get_current_utc_time().timestamp() + kwargs['dates'] = [ + Date(DateType.RETRIEVED, (retrieved_date, retrieved_date), DateMeta.NONE) + ] + if not is_deleted: + create_date = reformat_pixiv_date(data['create_date']).timestamp() + kwargs['dates'].append(Date(DateType.CREATED, (create_date, create_date), DateMeta.NONE)) 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'] + if not is_deleted: + user.profile_picture_url = Url(data['user']['profile_image_urls']['medium']) kwargs['author'] = user kwargs['title'] = data['title'] kwargs['text'] = data['caption'] - kwargs['likes'] = data['total_bookmarks'] + if not is_deleted: # Default to None instead of 0. + kwargs['bookmarks'] = data['total_bookmarks'] + kwargs['views'] = data['total_view'] 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 + kwargs['tags'] = [self.create_tag(t) for t in data['tags']] + if not is_deleted: + 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'] + } + }) + media = {} + for index, page in enumerate(data['meta_pages']): + media[str(index)] = Image(url=Url(page['image_urls']['original']), thumbnail_url=Url(page['image_urls']['medium'])) + kwargs['media'] = media post = Post(**kwargs) self.module.add_to_map(unique_id, post) return post @@ -82,16 +96,23 @@ class PixivAppBase(Search): 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']) + user.profile_picture_url = Url(data['profile_image_urls']['medium']) self.module.add_to_map(unique_id, user) return user + def remake_post(self, raw_responses: ParsedJson, original_date: float) -> Optional[Post]: + api = json.loads(raw_responses['api']) + return self.create_post(api, original_date) + class PixivAppSearch(PixivAppBase): def __init__(self, userdata: Any, arg: str) -> None: 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['sort'] = 'popular_desc' + # https://github.com/upbit/pixivpy/issues/352 + # We want to include AI results for completeness, to be ignored further down the chain. + #self.next_qs['search_ai_type'] = 1 self.next_qs['start_date'] = '2019-01-01' self.next_qs['end_date'] = '2019-12-31' @@ -99,27 +120,10 @@ class PixivAppSearch(PixivAppBase): 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']: + 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) -> None: - 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) -> None: super().__init__(userdata) @@ -131,9 +135,7 @@ class PixivAppUser(PixivAppBase): 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']: + 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']) @@ -148,9 +150,7 @@ class PixivAppBookmarks(PixivAppBase): 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']: + 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']) @@ -165,22 +165,39 @@ class PixivAppFollowing(PixivAppBase): 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']: + 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 PixivAppIllust(PixivAppBase): + def __init__(self, userdata: Any, arg: str) -> None: + 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) + self.pages[num].append(self.create_post(obj.illust).unique_id) + return True, None + class PixivAppModule(Module): def __init__(self, refresh_token: str) -> None: 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.parser.add_command('illust', PixivAppIllust) + headers = { + "app-os": "ios", + "app-os-version": "16.7.2", + "app-version": "7.19.6", + "user-agent": "PixivIOSApp/7.19.6 (iOS 16.7.2; iPhone13,2)" + } + self.api: AppPixivAPI = AppPixivAPI(headers=headers) self.api.auth(refresh_token=refresh_token) def search(self, query: str, *extra_args: Any) -> Optional[Search]: diff --git a/src/portal/py/modules/pixiv_web.py b/src/portal/py/modules/pixiv_web.py index 3c04202..e7399dc 100644 --- a/src/portal/py/modules/pixiv_web.py +++ b/src/portal/py/modules/pixiv_web.py @@ -33,8 +33,8 @@ class PixivWebBase(Search): def parse_pages(self, data: ParsedJson) -> dict[str, Media]: media = {} - for i, m in enumerate(data): - media[str(i)] = Image(url=Url(m['urls']['original']), thumbnail_url=Url(m['urls']['small'])) + for index, m in enumerate(data): + media[str(index)] = Image(url=Url(m['urls']['original']), thumbnail_url=Url(m['urls']['small'])) return media def parse_ugoira(self, data: ParsedJson, thumbnail_url: str) -> Animation: @@ -43,7 +43,7 @@ class PixivWebBase(Search): animation.frames.append((frame['file'], frame['delay'])) return animation - def seek_profile_picture_url(self, user_illusts: ParsedJson) -> Optional[Url]: + def search_profile_picture_url(self, user_illusts: ParsedJson) -> Optional[Url]: for illust in user_illusts.values(): if illust is None: continue @@ -52,7 +52,7 @@ class PixivWebBase(Search): return None def create_tag(self, data: ParsedJson) -> Tag: - kwargs= {} + kwargs = {} kwargs['type'] = TagType.GENERAL kwargs['name'] = data['tag'] kwargs['alts'] = {} @@ -63,26 +63,27 @@ class PixivWebBase(Search): kwargs['alts']['en'] = data['translation']['en'] return Tag(**kwargs) - def create_post(self, data: ParsedJson, pages: Optional[ParsedJson] = None, ugoira: Optional[ParsedJson] = None) -> Optional[Post]: + def create_post(self, data: ParsedJson, pages: Optional[ParsedJson] = None, ugoira: Optional[ParsedJson] = None, retrieved_date: Optional[float] = 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['raw_responses']['api'] = json.dumps(data, separators=(',', ':'), indent=None) kwargs['url'] = f'https://www.pixiv.net/{LANG}/artworks/{data['illustId']}' create_date = reformat_pixiv_date(data['createDate']).timestamp() - current_date = get_current_utc_time().timestamp() + if not retrieved_date: + retrieved_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) + Date(DateType.RETRIEVED, (retrieved_date, retrieved_date), DateMeta.NONE) ] upload_date = reformat_pixiv_date(data['uploadDate']).timestamp() if upload_date != create_date: kwargs['dates'].append(Date(DateType.EDITED, (upload_date, upload_date), DateMeta.NONE)) 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']) + profile_picture_url = self.search_profile_picture_url(data['userIllusts']) if profile_picture_url: user.profile_picture_url = profile_picture_url kwargs['author'] = user @@ -124,19 +125,6 @@ class PixivWebBase(Search): 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)) @@ -151,6 +139,16 @@ class PixivWebBase(Search): self.module.add_to_map(unique_id, user) return user + 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) + return self.create_post(api, pages, ugoira, original_date) + class PixivWebSearch(PixivWebBase): def __init__(self, userdata: Any, arg: str) -> None: super().__init__(userdata) @@ -178,23 +176,6 @@ class PixivWebSearch(PixivWebBase): self.pages[num].append(post.unique_id) return True -class PixivWebIllust(PixivWebBase): - def __init__(self, userdata: Any, arg: str) -> None: - 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) -> None: super().__init__(userdata) @@ -285,15 +266,32 @@ class PixivWebFollowing(PixivWebBase): self.pages[num].append(self.create_user(user).unique_id) return True +class PixivWebIllust(PixivWebBase): + def __init__(self, userdata: Any, arg: str) -> None: + 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 PixivWebModule(Module): def __init__(self, sessid: str, user_id: str) -> None: 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('bookmarks', PixivWebBookmarks) self.parser.add_command('following', PixivWebFollowing) + self.parser.add_command('illust', PixivWebIllust) self.headers['User-Agent'] = USER_AGENT self.headers['Accept-Encoding'] = 'gzip, deflate, br, zstd' self.headers['Accept-Language'] = 'en-US,en;q=0.5' diff --git a/src/portal/py/modules/twitter.py b/src/portal/py/modules/twitter.py index 2f45ec1..93286c3 100644 --- a/src/portal/py/modules/twitter.py +++ b/src/portal/py/modules/twitter.py @@ -3,15 +3,15 @@ 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, DateType, PostRef, Post, User, Media, Image, Video +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.modules import twitter from snscrape.base import ScraperException +from snscrape.modules import twitter from snscrape.modules.twitter import (Tweet, TweetRef, Tombstone, UserRef, - TwitterSearchScraper, TwitterProfileScraper, TwitterUserScraper, TwitterTweetScraper) + TwitterSearchScraper, TwitterSearchScraperMode, TwitterProfileScraper, TwitterUserScraper, TwitterTweetScraper) class TwitterScrapeBase(Search): # Posts per emulated page because snscrape returns an iterator. @@ -20,7 +20,7 @@ class TwitterScrapeBase(Search): # Don't allow requests for a page more than this amount past the current page. REACH_LIMIT = 4 - def __init__(self, userdata: Any): + def __init__(self, userdata: Any) -> None: super().__init__() self.module: TwitterScrapeModule = userdata self.iterator: Iterator[Tweet | TweetRef | Tombstone] @@ -28,38 +28,39 @@ class TwitterScrapeBase(Search): def create_media_url(self, url: str) -> Url: format = url.find('format=') + ext = None if format >= 0: - ext = url[format + 7:format + 10] - else: - question = url.rfind('?') - if question >= 0: - url = url[0:question] - ext = url.split('.')[-1] + 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 i, m in enumerate(data): + for index, 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)) + 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(i)] = Video(url=self.create_media_url(videos[0].url), thumbnail_url=self.create_media_url(m.thumbnailUrl)) + 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 - # @TODO: Replace correct? - kwargs['dates'] = { - DateType.CREATED: tweet.date.replace(tzinfo=timezone.utc).timestamp(), - DateType.RETRIEVED: get_current_utc_time().timestamp() - } + 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 @@ -116,40 +117,40 @@ class TwitterScrapeBase(Search): if num - self.page >= self.REACH_LIMIT: return False request_satisfied = False - for i in range(self.page, num + 1): + for index in range(self.page, num + 1): if self.completed: break - if i in self.pages: + if index in self.pages: continue - self.pages[i] = [] - if not self.step_iterator_for_page(i): + self.pages[index] = [] + if not self.step_iterator_for_page(index): break - self.page = i - if i == num: + self.page = index + if index == num: request_satisfied = True return request_satisfied class TwitterScrapeSearch(TwitterScrapeBase): - def __init__(self, userdata: Any, arg: str): + def __init__(self, userdata: Any, arg: str) -> None: super().__init__(userdata) - self.iterator = TwitterSearchScraper(arg, top=True, cookies=self.module.cookies).get_items() + self.iterator = TwitterSearchScraper(arg, mode=TwitterSearchScraperMode.TOP, cookies=self.module.cookies).get_items() class TwitterScrapeUser(TwitterScrapeBase): - def __init__(self, userdata: Any, arg: str): + 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): + 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): + def __init__(self, userdata: Any, arg: str) -> None: super().__init__(userdata) if arg.startswith('https://'): arg = arg[arg.rfind('/') + 1:] @@ -159,7 +160,7 @@ class TwitterScrapeTweet(TwitterScrapeBase): self.iterator = TwitterTweetScraper(arg, cookies=self.module.cookies).get_items() class TwitterScrapeModule(Module): - def __init__(self, cookies_path: str): + def __init__(self, cookies_path: str) -> None: super().__init__() self.parser: QueryParser = QueryParser(self, 'search') self.parser.add_command('search', TwitterScrapeSearch) |