开发开发PythonPython爬取K站(Konachan)原图
Sukap
24/2/22 01:57:33 更新：
重写了一遍，加入了多线程下载和错误统计改完还是史
2024/02/23 17:15:46 更新：
- 解决了报错
Max retries exceeded with url 的问题
- 补充了下载图片截止至当前时间
写博文定定少不了一个好康的Banner，但是又没啥资源，想着去爬点资源顺带着整一个API出来自个用，这篇水文就诞生了
下手对象是从A、B、C、D、E、F、G、K、M、P站中选出的：Konachan，也就是K站
K站有两个，分别是大众级(G)的 https://konachan.net ，和限制级(R-18) https://konachan.com
本文选用的是大众级(G)的 https://konachan.net 进行爬取，当然也能无缝切换到限制级站点
源码中是按月份进行爬取的，每个月的所有图片会保存在同一个目录里面
代码逻辑如下：
- 设置初始日期为
2024-1-1，page默认值为1，将两个参数传入链接tags和page中
- 常量
FILE_PATH指的是文件的下载目录，我用的值为C:\\Users\\Sukap\\Desktop\\konachan，自行修改为自己的目录
- 同一个月份的所有图片都会存在一个文件夹里边，程序会自动创建每个月的文件夹，例如说
202401
- 请求链接后，返回JSON数据，控制台会打印
Status Code和url方便检查K站是否成功响应
- 加入了记录异常的操作，会以
JSON格式保存在下载目录中的error.json中
- 解析出图片的
原图链接和图片ID后，自动开启多线程下载，图片会自动重命名为图片ID
通俗点讲就是：
请求24年1月1日的图片数据，然后下载进202401这个文件夹中，1月1日的图片下载完成后会自动下载后面日期的图片
整个月的照片都会汇聚在一个文件夹里边，2月份的所有图片也会汇聚在202402这个文件夹里
控制台的输出：
注意：
- Q：Status Code: 403
A：国内无法直连，先挂代理
- Q：控制台报错：SSLError
A：代理抽风，关掉再开或者换个节点
源码已在Github开源：sukaps/Konachan_Crawler
遵循GPL-3.0开源协议。
下面是源码：
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
| import json
import os
import requests
from datetime import datetime, timedelta
import concurrent.futures
import urllib3
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
FILE_PATH = 'C:\\Users\\Sukap\\Desktop\\konachan\\'
base_url = 'https://konachan.net/post.json?tags=date:{}-{}-{}&page={}'
current_date = datetime.strptime('2024-01-05', '%Y-%m-%d')
current_page = 1
session = requests.Session()
retry_strategy = Retry(total=3, backoff_factor=1)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount('https://', adapter)
def error_log(error_type, msg=None, status_code=None, url=None):
json_path = FILE_PATH + 'error.json'
try:
with open(json_path, 'r', encoding='utf-8') as file:
json_data = file.read()
if json_data.strip():
json_data = json.loads(json_data)
else:
json_data = []
except (FileNotFoundError, json.decoder.JSONDecodeError):
json_data = []
if status_code:
json_data.append({'error_type': error_type, 'status_code': status_code, 'msg': msg, 'url': url})
else:
json_data.append({'error_type': error_type, 'msg': msg})
with open(json_path, 'w', encoding='utf-8') as file:
json.dump(json_data, file, ensure_ascii=False)
def download_image_in_thread(url, folder_path, image_id):
print(f"ID: {image_id} 正在下载...")
response = session.get(url)
if response.status_code == 200:
with open(os.path.join(folder_path, f"{image_id}.jpg"), 'wb') as file:
file.write(response.content)
print(f"ID: {image_id} 下载完成")
else:
print(f"ID: {image_id} 下载失败，状态码：{response.status_code}")
error_log('图片下载失败', f"ID: {image_id} 下载失败", response.status_code, url)
def main():
global current_date, current_page
try:
while True:
url = base_url.format(current_date.year, current_date.month, current_date.day, current_page)
print(f"URL: {url}")
response = session.get(url)
print(f"Status Code: {response.status_code}")
response.raise_for_status()
data = response.json()
if not data:
current_date += timedelta(days=1)
current_page = 1
else:
max_workers = 6
with concurrent.futures.ThreadPoolExecutor(max_workers) as executor:
for item in data:
post_id = item.get('id')
file_url = item.get('file_url')
folder_name = f"{current_date.year}{current_date.month:02d}"
folder_path = os.path.join(FILE_PATH, folder_name)
if not os.path.exists(folder_path):
os.makedirs(folder_path)
executor.submit(download_image_in_thread, file_url, folder_path, post_id)
if current_date.date() == datetime.now().date():
break
current_page += 1
except requests.exceptions.RequestException as e:
print('RequestException' + str(e))
error_log('请求异常', str(e))
except ValueError as e:
error_log('数据解析失败', str(e))
if __name__ == "__main__":
main()
|
水文结束，边爬图片边写随机图片API（大概是从存储库拉取图片至服务器后端压缩后输出减小体积）
