1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
|
import log
import json
from typing import Optional, Any
from json.decoder import JSONDecodeError
from base import USER_AGENT, HttpResponse, 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 parse_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[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['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 = parse_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 = parse_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.getComments?postId={post_id}&limit=10'
obj = self.check_api_response(self.module.do_request(Method.GET, url))
if not obj:
return None
'''
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)
if len(self.pagination) == 0:
log.warn(f'Found 0 posts for user.')
return True
def request_page(self, num: int) -> bool:
if len(self.pagination) == 0:
if not self.request_pagination():
log.error('Failed to load user\'s posts.')
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)
# @TODO: At some point verify if tls-client supports zstd.
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['cf_clearance'] = cf_clearance
self.cookies['privacy_policy_agreement'] = '7'
self.cookies['privacy_policy_notification'] = '0'
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': {}
}
|