大批量爬取Pixiv推荐插图

pixiv.py

from pixivpy3 import AppPixivAPI
from concurrent.futures import ThreadPoolExecutor
import httpx
import os

aapi = AppPixivAPI()
aapi.auth(refresh_token="x")

def get_recommend_illusts():
    json_result = aapi.illust_recommended()
    results = []
    for illust in json_result.illusts:
        if illust["meta_single_page"].get("original_image_url") is not None:
            results.append((illust.meta_single_page['original_image_url'], f'{illust.id}.png'))
        else:
            results.append((illust.image_urls['large'], f'{illust.id}.png'))
    return results

def get_url(url: str, filename: str):
    url = url.replace("i.pximg.net", "proxy.pixivel.moe")
    if os.path.exists(os.path.join("cache", filename)):
        print(f"Skipping {url} {filename}...")
        return
    print(f"Downloading {url} {filename}...")
    i = 0
    while i <=3:
        res = httpx.get(url, proxies="http://127.0.0.1:7890")
        if res.status_code == 200:
            with open(os.path.join("cache", filename), 'wb') as f:
                f.write(res.content)
            print(f"Downloaded {url} {filename}!")
            return
        else:
            i += 1

    print(f"Failed to download {url} {filename} after 3 retries.")

if __name__ == '__main__':
    if not os.path.exists("cache"):
        os.mkdir("cache")

    with ThreadPoolExecutor(max_workers=64) as executor:
        for time in range(100):
            results = get_recommend_illusts()
            for result in results:
                executor.submit(get_url, *result)