import log from json import JSONDecodeError from typing import Optional, Any from datetime import datetime from base import USER_AGENT, HttpResponse, Search, Module, Method, ParsedJson from modules.common import get_current_utc_time from post import Date, DateType, DateMeta, Post, User, Url, Media, File from query_parser import QueryParser BASE_URL = 'https://www.patreon.com/api' def parse_patreon_date(date_str: str) -> datetime: return datetime.strptime(date_str, '%Y-%m-%dT%H:%M:%S.%f%z') class PatreonBase(Search): REACH_LIMIT = 3 def __init__(self, userdata: Any) -> None: super().__init__() self.module: PatreonModule = userdata self.page: int = 0 self.cursor: Optional[str] = None def check_api_response(self, response: Optional[HttpResponse]) -> 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 search_user(self, query: str) -> tuple[Optional[User], Optional[str]]: url = f'{BASE_URL}/search' params = { 'q': query, 'page[number]': '1', 'json-api-version': '1.0', 'json-api-use-default-includes': 'false', 'include': '[]' } obj = self.check_api_response(self.module.do_request(Method.GET, url, params=params)) if not obj: return None, None for result in obj['data']: username = result['attributes']['creator_name'] display_name = result['attributes']['name'] url_name = result['attributes']['url'].split('/')[-1] # Only accept exact matches. if username.lower() != query.lower() and display_name.lower() != query.lower() and url_name != query.lower(): continue prefix = result['id'].find('campaign_') if prefix >= 0: campaign = result['id'][prefix + len('campaign_'):] unique_id = 'patreon:c:' + campaign user = User(unique_id=unique_id, username=username, display_name=display_name) return user, campaign return None, None class PatreonUser(PatreonBase): def __init__(self, userdata: Any, arg: str) -> None: super().__init__(userdata) self.username: str = arg self.user: Optional[User] = None self.campaign: Optional[str] = None self.cursor: Optional[str] = None self.attachments: dict[str, Media] = {} # @TODO: # post_type # image_file: https://www.patreon.com/posts/emma-again-29391136 # text_only: https://www.patreon.com/posts/hello-update-29391184 # tags def create_post(self, data: ParsedJson, hash: str) -> Optional[Post]: #if 'access_rules' in data['relationships']: # if not (len(data['relationships']['access_rules']['data']) == 1 and data['relationships']['access_rules']['data'][0]['id'] == '9816126'): # return None kwargs = {} unique_id = f'patreon:p:{data['id']}' kwargs['unique_id'] = unique_id kwargs['raw_responses'] = {} kwargs['raw_responses']['api'] = hash kwargs['url'] = data['attributes']['url'] published_at = parse_patreon_date(data['attributes']['published_at']).timestamp() current_time = get_current_utc_time().timestamp() kwargs['dates'] = [ Date(DateType.CREATED, (published_at, published_at), DateMeta.NONE), Date(DateType.RETRIEVED, (current_time, current_time), DateMeta.NONE) ] if 'edited_at' in data['attributes']: edited_at = parse_patreon_date(data['attributes']['edited_at']).timestamp() kwargs['dates'].append(Date(DateType.EDITED, (edited_at, edited_at), DateMeta.NONE)) kwargs['author'] = self.user kwargs['title'] = data['attributes']['title'] kwargs['likes'] = data['attributes']['like_count'] if 'comment_count' in data['attributes']: kwargs['comments'] = data['attributes']['comment_count'] kwargs['text'] = '' if 'content' in data['attributes']: kwargs['text'] = data['attributes']['content'] #post_type = data['attributes']['post_type'] media = {} if data['relationships']['media']['data']: for m in data['relationships']['media']['data']: if m['id'] in self.attachments: media[m['id']] = self.attachments[m['id']] if 'attachments' in data['relationships'] and data['relationships']['attachments']['data']: for a in data['relationships']['attachments']['data']: if a['id'] in self.attachments: media[a['id']] = self.attachments[a['id']] kwargs['media'] = media post = Post(**kwargs) self.module.add_to_map(unique_id, post) return post def request_page(self, num: int) -> bool: if num >= self.page + self.REACH_LIMIT: return False if not self.user: self.user, self.campaign = self.search_user(self.username) if not self.user: self.errored = True return False request_satisfied = False for index in range(self.page, num + 1): url = f'{BASE_URL}/posts' #'filter[accessible_by_user_id]': self.module.user_id, params = { 'include': 'campaign,access_rules,access_rules.tier.null,attachments_media,audio,audio_preview.null,drop,images,media,native_video_insights,poll.choices,poll.current_user_responses.user,poll.current_user_responses.choice,poll.current_user_responses.poll,user,user_defined_tags,ti_checks,video.null,content_unlock_options.product_variant.null', 'fields[campaign]': 'currency,show_audio_post_download_links,avatar_photo_url,avatar_photo_image_urls,earnings_visibility,is_nsfw,is_monthly,name,url', 'fields[post]': 'change_visibility_at,comment_count,commenter_count,content,created_at,current_user_can_comment,current_user_can_delete,current_user_can_report,current_user_can_view,current_user_comment_disallowed_reason,current_user_has_liked,embed,image,insights_last_updated_at,is_paid,like_count,meta_image_url,min_cents_pledged_to_view,monetization_ineligibility_reason,post_file,post_metadata,published_at,patreon_url,post_type,pledge_url,preview_asset_type,thumbnail,thumbnail_url,teaser_text,title,upgrade_url,url,was_posted_by_campaign_owner,has_ti_violation,moderation_status,post_level_suspension_removal_date,pls_one_liners_by_category,video,video_preview,view_count,content_unlock_options,is_new_to_current_user,watch_state', 'fields[post_tag]': 'tag_type,value', 'fields[user]': 'image_url,full_name,url', 'fields[access_rule]': 'access_rule_type,amount_cents', 'fields[media]': 'id,image_urls,display,download_url,metadata,file_name', 'fields[native_video_insights]': 'average_view_duration,average_view_pct,has_preview,id,last_updated_at,num_views,preview_views,video_duration', 'fields[content-unlock-option]': 'content_unlock_type', 'fields[product-variant]': 'price_cents,currency_code,checkout_url,is_hidden,published_at_datetime,content_type,orders_count,access_metadata', 'filter[campaign_id]': self.campaign, 'filter[contains_exclusive_posts]': 'true', 'filter[is_draft]': 'false', 'sort': '-published_at', 'json-api-use-default-includes': 'false', 'json-api-version': '1.0' } if self.cursor: params['page[cursor]'] = self.cursor obj = self.check_api_response(self.module.do_request(Method.GET, url, params=params)) if not obj: self.errored = True return False hash = self.module.add_raw_response(obj) try: # Cursor should be 'null' at the end of pagination. self.cursor = obj['meta']['pagination']['cursors']['next'] except KeyError: self.cursor = None for inc in obj['included']: if inc['type'] == 'attachment': self.attachments[inc['id']] = File(url=Url(inc['attributes']['url']), name=inc['attributes']['name']) elif inc['type'] == 'media': if 'download_url' in inc['attributes']: media_url = inc['attributes']['download_url'] elif inc['attributes'].get('image_urls', None): media_url = inc['attributes']['image_urls']['original'] elif 'default_thumbnail' in inc['attributes']['display']: media_url = inc['attributes']['display']['default_thumbnail']['url'] else: continue if 'default_thumbnail' in inc['attributes']['display']: thumbnail_url = inc['attributes']['display']['default_thumbnail']['url'] elif inc['attributes']['image_urls']: thumbnail_url = inc['attributes']['image_urls']['thumbnail'] else: thumbnail_url = '' self.attachments[inc['id']] = Media(url=Url(media_url), thumbnail_url=Url(thumbnail_url)) else: log.warn(f'Unhandled include type {inc['type']}.') 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[index].append(post.unique_id) self.page += 1 if index == num: request_satisfied = True if not self.cursor: self.completed = True break return request_satisfied class PatreonModule(Module): def __init__(self, session_id: str, uuid: str, user_id: str) -> None: super().__init__() self.user_id: str = user_id self.parser: QueryParser = QueryParser(self, 'user') self.parser.add_command('user', PatreonUser) self.headers['User-Agent'] = USER_AGENT self.headers['Accept-Encoding'] = 'gzip, deflate, br, zstd' self.headers['Accept-Language'] = 'en-US,en;q=0.5' self.headers['x-patreon-uuid'] = uuid self.cookies['patreon_locale_code'] = 'en-US' self.cookies['patreon_location_country_code'] = 'US' self.cookies['session_id'] = session_id 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': self.headers, 'cookies': self.cookies }