- Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathscraper_fantia.py
More file actions
Latest commit
243 lines (214 loc) · 10.8 KB
/
Copy pathscraper_fantia.py
File metadata and controls
243 lines (214 loc) · 10.8 KB
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
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
importconcurrent.futures
importre
importwebbrowser
fromdatetimeimportdatetime
frompathlibimportPath
fromurllib.parseimporturljoin
# from natsort import natsorted
frombs4importBeautifulSoup
fromutilimportdownload, get_webname, requests_retry_session, safeify
API_POSTS="https://fantia.jp/api/v1/posts/{}"
API_FANCLUB="https://fantia.jp/api/v1/fanclubs/{}"
HTML_POSTLIST="https://fantia.jp/fanclubs/{}/posts?page={}"
DEFAULT_UA='Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.99 Safari/537.36'
# Default templates for directory and filename formatting
# Directory template variables:
# {fanclub_id}, {fanclub_name}, {fanclub_full_name}, {creator_name}
# {post_id}, {post_title}, {post_date}, {rating}
# Filename template variables (in addition to directory ones):
# {content_id}, {idx}, {stem_cleaned}
# Note: extension (.ext) is always appended automatically and not part of the template
DEFAULT_DIR_TEMPLATE='{fanclub_full_name} ({fanclub_id})/{post_id} {post_title}'
DEFAULT_FILENAME_TEMPLATE='{content_id}{idx} {stem_cleaned}'
classFantiaDownloader:
def__init__(self, key, fanclub=None, output='.', dir_template=None, filename_template=None, skip_existing=True, quick_stop=True):
super().__init__()
self.key=key
self.fanclub=fanclub
ifnotself.fanclub:
print('[W] no fanclub id is given. "downloadAll()" won\'t work.')
self.output=Path(output)
self.dir_template=dir_templateorDEFAULT_DIR_TEMPLATE
self.filename_template=filename_templateorDEFAULT_FILENAME_TEMPLATE
self.skip_existing=skip_existing
self.quick_stop=skip_existingandquick_stop
self.fanclub_info=None
self.session=requests_retry_session()
self.session.headers['User-Agent'] =DEFAULT_UA
self.session.cookies.update({"_session_id": self.key})
self.token=None
self.__set_token__()
def__set_token__(self):
r=self.fetch('https://fantia.jp/')
soup=BeautifulSoup(r.content, 'html.parser')
ifele:=soup.select_one('meta[name="csrf-token"]'):
print(f'Set csrf-token = {ele["content"]}')
self.token=ele['content']
self.session.headers['x-csrf-token'] =self.token
else:
raiseException('Failed to obtain csrf-token!!')
deffetch(self, url, headers=None):
returnself.session.get(url, headers=headers)
defdownload_all(self):
ifnotself.fanclub:
print('No fanclub id is given. "downloadAll()" won\'t work.')
return
ifnotself.fanclub_info:
withself.fetch(API_FANCLUB.format(self.fanclub)) asr:
self.update_fanclub_info(r.json()['fanclub'])
# Get base output dir (fanclub level) for checking existing downloads
# Extract only the fanclub-level part of the template (before any post-level placeholders)
fanclub_subs=self._get_fanclub_substitutes()
# Split dir_template and only format the parts that don't require post info
template_parts=self.dir_template.split('/')
base_parts= []
forpartintemplate_parts:
ifany(pinpartforpin ['{post_id}', '{post_title}', '{post_date}', '{rating}']):
break
base_parts.append(part.format(**fanclub_subs))
base_dir=self.output/'/'.join(base_parts) ifbase_partselseself.output
print(f'Fanclub {self.fanclub}: download all to {base_dir}.')
existing_ids= []
ifbase_dir.exists():
# Check for subdirectories starting with post_id (new structure)
forfinbase_dir.iterdir():
iff.is_dir() and (m:=re.match(r'^(\d+)\b', f.name)):
existing_ids.append(m.group(1))
# Also check for files starting with post_id (old flat structure)
forfinbase_dir.iterdir():
iff.is_file() and (m:=re.match(r'^(\d+)\b', f.name)):
existing_ids.append(m.group(1))
existing_ids=list(dict.fromkeys(existing_ids))
existing_ids=set(map(int, existing_ids))
print(f'{len(existing_ids)} ID(s) have already been downloaded.')
deffetch_all():
results= []; page=1
whileTrue:
print(f"Attempting to fetch page {page}...")
out=self.fetch_gallery_page(page)
ifout:
# do NOT sort; since the order is not id_desc
# out = natsorted(out, reverse=True)
foridinout:
ifself.quick_stopandidinexisting_ids:
print(f'Encountered existing id {id}. Quick stop.')
returnresults
results.append(id)
page+=1
else:
returnresults
results=fetch_all()
results=list(dict.fromkeys(results))
print(f"Get {len(results)} ID(s) from list.")
skipped= []; to_be_dl= []
foridinresults:
ifself.skip_existingandidinexisting_ids:
skipped.append(id)
else:
to_be_dl.append(id)
ifskipped:
print(f"Skip {len(skipped)} existing ID(s).")
ifto_be_dl:
# Only create the output directory if we're actually going to download something
self.output.mkdir(parents=True, exist_ok=True)
withconcurrent.futures.ThreadPoolExecutor(max_workers=10) asex:
foridinto_be_dl:
ex.submit(self.get_post_photos, id)
deffetch_gallery_page(self, page):
url=HTML_POSTLIST.format(self.fanclub, page)
r=self.fetch(url)
r.encoding='utf-8'
html=r.text
returnsorted(map(int, re.findall(r'\/posts\/(?P<id>[0-9]{1,8})"', html)), reverse=True)
defupdate_fanclub_info(self, d):
self.fanclub_info=d
def_get_fanclub_substitutes(self):
"""Get template substitutes from fanclub info."""
ifnotself.fanclub_info:
return {}
returndict(
fanclub_full_name=safeify(str(self.fanclub_info['fanclub_name_with_creator_name'])),
fanclub_name=safeify(str(self.fanclub_info['fanclub_name'])),
creator_name=safeify(str(self.fanclub_info['creator_name'])),
fanclub_id=self.fanclub_info['id']
)
def_get_post_substitutes(self, post_data):
"""Get template substitutes from post data."""
# Parse posted_at date (format: "Fri, 16 Aug 2024 07:15:36 +0900")
post_date=''
ifposted_at:=post_data.get('posted_at'):
try:
dt=datetime.strptime(posted_at, "%a, %d %b %Y %H:%M:%S %z")
post_date=dt.strftime('%Y%m%d')
exceptValueError:
post_date=''
returndict(
post_id=post_data.get('id', ''),
post_title=safeify(str(post_data.get('title', ''))),
post_date=post_date,
rating=post_data.get('rating', '')
)
def_format_dir(self, post_substitutes):
"""Format directory path using template and substitutes."""
subs= {**self._get_fanclub_substitutes(), **post_substitutes}
returnself.dir_template.format(**subs)
def_format_filename(self, post_substitutes, content_id, idx_string, stem_cleaned, ext):
"""Format filename using template and substitutes."""
subs= {
**self._get_fanclub_substitutes(),
**post_substitutes,
'content_id': content_id,
'idx': idx_string,
'stem_cleaned': stem_cleaned
}
# Format the template, strip to remove trailing spaces, then append extension
filename_without_ext=self.filename_template.format(**subs).strip()
returnf'{filename_without_ext}.{ext}'
defget_post_photos(self, id):
print(f'Fetching post {id}...')
whileTrue:
d=self.fetch(API_POSTS.format(id), headers={'x-requested-with': 'XMLHttpRequest'}).json()
if'redirect'ind:
recaptcha_url=urljoin(API_POSTS, d["redirect"])
# open recaptcha_url in browser
webbrowser.open(recaptcha_url)
input(f'Please solve the recaptcha at {recaptcha_url} and then press any key')
else:
break
if'error_text'ind:
print(f'Error: {d["error_text"]}')
raiseException(f'Error: {d["error_text"]}')
ifnotself.fanclub_info:
self.update_fanclub_info(d['post']['fanclub'])
post_data=d['post']
post_subs=self._get_post_substitutes(post_data)
output_dir=self.output/self._format_dir(post_subs)
# Ensure multi-level directory exists
output_dir.mkdir(parents=True, exist_ok=True)
withconcurrent.futures.ThreadPoolExecutor(max_workers=3) asex:
ifthumb:=post_data.get('thumb', None):
img_url=thumb['original']
stem, ext=get_webname(img_url).rsplit('.', 1)
cover_filename=f'!cover.{ext}'
ex.submit(download, img_url, filename=output_dir/cover_filename, verbose=1)
ifpost_contents:=post_data.get('post_contents', None):
forcinpost_contents:
cid=c['id']
ifphotos:=c.get('post_content_photos', None):
foridx, pinenumerate(photos, 1):
img_url=p['url']['original']
stem, ext=get_webname(img_url).rsplit('.', 1)
# Clean up the filename; remove all the UUID-ish garbage
stem_cleaned=re.sub(r'^[0-9a-fA-F]{8}_(.+)$', r'\1', stem)
stem_cleaned=re.sub(r'[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}', '', stem_cleaned)
stem_cleaned=stem_cleaned.strip()
idx_string='_'+str(idx).zfill(len(str(len(photos)))) iflen(photos) >1else''
filename=self._format_filename(post_subs, cid, idx_string, stem_cleaned, ext)
ex.submit(download, img_url, filename=output_dir/filename, verbose=1)
if'download_uri'inc:
dl_url=urljoin('https://fantia.jp', c['download_uri'])
# For download files, use original filename with content_id prefix
dl_filename=f'{cid}{c["filename"]}'
ex.submit(download, dl_url, filename=output_dir/dl_filename, session=self.session, verbose=1)
if__name__=="__main__":
pass