import log import json import httpx from typing import Optional, Any from json.decoder import JSONDecodeError from base import USER_AGENT, Search, Module, Method, ParsedJson from post import Url, PostType, Date, DateType, DateMeta, Post, User, Image, File, Tag from query_parser import QueryParser from modules.common import reformat_pixiv_date, get_current_utc_time BASE_URL = "https://api.fanbox.cc" class FanboxBase(Search): def __init__(self, userdata: Any) -> None: super().__init__() self.module: FanboxModule = userdata def check_api_response(self, response: Optional[httpx.Response]) -> Optional[ParsedJson]: if not response: return None try: obj = response.json() except JSONDecodeError as e: log.error(f'JSONDecodeError: {e}.') return None return obj['body'] def create_post(self, data: ParsedJson) -> Optional[Post]: kwargs = {} post_id = data['id'] unique_id = f'fanbox:p:{post_id}' kwargs['unique_id'] = unique_id kwargs['raw_responses'] = {} kwargs['raw_responses']['api'] = json.dumps(data, separators=(',', ':'), indent=None) publish_date = reformat_pixiv_date(data['publishedDatetime']).timestamp() current_date = get_current_utc_time().timestamp() kwargs['dates'] = [ Date(DateType.CREATED, (publish_date, publish_date), DateMeta.NONE), Date(DateType.RETRIEVED, (current_date, current_date), DateMeta.NONE) ] update_date = reformat_pixiv_date(data['updatedDatetime']).timestamp() if update_date != publish_date: kwargs['dates'].append(Date(DateType.EDITED, (update_date, update_date), DateMeta.NONE)) user = User(unique_id=f'fanbox:u:{data['user']['userId']}', username=data['user']['name']) user.profile_picture_url = Url(data['user']['iconUrl']) kwargs['author'] = user kwargs['title'] = data['title'] kwargs['likes'] = data['likeCount'] kwargs['comments'] = data['commentCount'] kwargs['tags'] = [Tag(name=tag) for tag in data['tags']] if data['isRestricted']: kwargs['type'] = PostType.PREVIEW else: kwargs['type'] = PostType.POST text = '' media = {} if data['type'] == 'image': text = data['body']['text'] for 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 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': text += block['text'] + '\n' elif block['type'] == 'image': text += f'image::{block['imageId']}[]' + '\n' elif block['type'] == 'url_embed': text += f'embed::{block['urlEmbedId']}[]' + '\n' for key, value in data['body']['imageMap'].items(): media[key] = Image(url=Url(value['originalUrl'], value['extension']), thumbnail_url=Url(value['thumbnailUrl'])) for key, value in data['body']['fileMap'].items(): media[key] = File(url=Url(value['url'], value['extension']), name=value['name']) kwargs['text'] = text kwargs['media'] = media post = Post(**kwargs) self.module.add_to_map(unique_id, post) return post def request_post(self, post_id: int) -> Optional[Post]: url = f'{BASE_URL}/post.info?postId={post_id}' obj = self.check_api_response(self.module.do_request(Method.GET, url)) if not obj: return None return self.create_post(obj) def create_user(self, data: ParsedJson) -> User: unique_id = f'fanbox:u:{data['id']}' user = User(unique_id=unique_id, username=data['creatorId'], display_name=data['user']['name']) user.profile_picture_url = Url(data['user']['iconUrl']) self.module.add_to_map(unique_id, user) return user class FanboxPost(FanboxBase): 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_post(self.id) if not post: self.errored = True return False self.pages[num] = [post.unique_id] self.completed = True return True class FanboxUser(FanboxBase): def __init__(self, userdata: Any, arg: str) -> None: super().__init__(userdata) self.id = arg self.pagination = [] def request_pagination(self) -> bool: url = f'{BASE_URL}/post.paginateCreator?creatorId={self.id}' obj = self.check_api_response(self.module.do_request(Method.GET, url)) if not obj: return False for link in obj: self.pagination.append(link) return True def request_page(self, num: int) -> bool: if len(self.pagination) == 0: if not self.request_pagination(): self.errored = True return False if num >= len(self.pagination): return False obj = self.check_api_response(self.module.do_request(Method.GET, self.pagination[num])) if not obj: self.errored = True return False self.pages[num] = [] for item in obj: post = self.request_post(item['id']) if post: self.pages[num].append(post.unique_id) if len(self.pages) == len(self.pagination): self.completed = True return True class FanboxSupporting(FanboxBase): def __init__(self, userdata: Any, arg: str) -> None: super().__init__(userdata) def request_page(self, num: int) -> bool: url = f'{BASE_URL}/plan.listSupporting' obj = self.check_api_response(self.module.do_request(Method.GET, url)) if not obj: self.errored = True return False self.pages[num] = [] for user in obj: self.pages[num].append(self.create_user(user).unique_id) self.completed = True return True class FanboxModule(Module): def __init__(self, sessid: str, cf_clearance: str) -> None: super().__init__() self.parser: QueryParser = QueryParser(self, 'post') self.parser.add_command('post', FanboxPost) self.parser.add_command('user', FanboxUser) self.parser.add_command('supporting', FanboxSupporting) self.headers['Accept-Encoding'] = 'gzip, deflate, br, zstd' self.headers['Accept-Language'] = 'en-US,en;q=0.5' self.headers['Alt-Used'] = 'api.fanbox.cc' self.headers['Origin'] = 'https://www.fanbox.cc' self.headers['Referer'] = 'https://www.fanbox.cc/' self.headers['User-Agent'] = USER_AGENT self.cookies['FANBOXSESSID'] = sessid self.cookies['privacy_policy_agreement'] = '7' self.cookies['privacy_policy_notification'] = '0' self.cookies['cf_clearance'] = cf_clearance def search(self, query: str, *extra_args: Any) -> Optional[Search]: return self.parser.parse_query(query) def get_download(self, unique_id: str, key: str) -> Optional[dict[str, Any]]: if unique_id not in self.unique_id_map: return None post = self.unique_id_map[unique_id] if key not in post.media: return None return { 'urls': [post.media[key].url], 'headers': {}, 'cookies': {} }