import json from typing import Optional, Any from base import USER_AGENT, Search, Module, ParsedJson from post import Url, PostType, Date, DateType, DateMeta, User, Post, Image, Tag, TagType from query_parser import QueryParser from pixivpy3 import AppPixivAPI, models from modules.common import get_current_utc_time, reformat_pixiv_date class PixivAppBase(Search): def __init__(self, userdata: Any) -> None: super().__init__() self.module: PixivAppModule = userdata self.page: int = 0 self.next_qs: Optional[dict[str, Any]] = None def api_request_page(self, num: int) -> tuple[bool, Optional[dict[str, Any]]]: return False, None def request_page(self, num: int) -> bool: request_satisfied = False for index in range(self.page, num + 1): if not self.next_qs: break if index in self.pages: continue self.pages[index] = [] success, self.next_qs = self.api_request_page(index) if not success: self.errored = True break if index == num: request_satisfied = True self.page += 1 return request_satisfied def create_tag(self, data: ParsedJson) -> Tag: kwargs = {} kwargs['type'] = TagType.GENERAL kwargs['name'] = data['name'] kwargs['alts'] = {} if data['translated_name']: kwargs['alts']['en'] = data['translated_name'] return Tag(**kwargs) # 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 = f'pixiv:i:{data['id']}' kwargs['unique_id'] = unique_id kwargs['raw_responses'] = {} 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']) 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'] 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['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 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' # 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' def api_request_page(self, num: int) -> tuple[bool, Optional[dict[str, Any]]]: if not self.next_qs: return False, None obj = self.module.api.search_illust(**self.next_qs) 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 PixivAppUser(PixivAppBase): def __init__(self, userdata: Any, arg: str) -> None: super().__init__(userdata) self.next_qs: Optional[dict[str, Any]] = {} self.next_qs['user_id'] = arg self.next_qs['type'] = 'illust' def api_request_page(self, num: int) -> tuple[bool, Optional[dict[str, Any]]]: if not self.next_qs: return False, None obj = self.module.api.user_illusts(**self.next_qs) for post in obj.illusts: self.pages[num].append(self.create_post(post).unique_id) return True, self.module.api.parse_qs(obj['next_url']) class PixivAppBookmarks(PixivAppBase): def __init__(self, userdata: Any, arg: str) -> None: super().__init__(userdata) self.next_qs: Optional[dict[str, Any]] = {} self.next_qs['user_id'] = arg self.next_qs['restrict'] = 'public' def api_request_page(self, num: int) -> tuple[bool, Optional[dict[str, Any]]]: if not self.next_qs: return False, None obj = self.module.api.user_bookmarks_illust(**self.next_qs) for post in obj.illusts: self.pages[num].append(self.create_post(post).unique_id) return True, self.module.api.parse_qs(obj['next_url']) class PixivAppFollowing(PixivAppBase): def __init__(self, userdata: Any, arg: str) -> None: super().__init__(userdata) self.next_qs: Optional[dict[str, Any]] = {} self.next_qs['user_id'] = arg self.next_qs['restrict'] = 'public' def api_request_page(self, num: int) -> tuple[bool, Optional[dict[str, Any]]]: if not self.next_qs: return False, None obj = self.module.api.user_following(**self.next_qs) 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('user', PixivAppUser) self.parser.add_command('bookmarks', PixivAppBookmarks) self.parser.add_command('following', PixivAppFollowing) 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]: 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, 'Referer': 'https://www.pixiv.net/' } }