Python异步编程--获取girlypic写真集

news2024/11/25 12:35:34

前言

在日常生活中，无论是网络配置、文件整理、web开发还是工具，时常用到Python写些脚本。

这次主要是分享下异步编程的经验，就拿获取girlypic的图片举例吧，也希望能给一些同学带来思考。

使用argparse而不是os.args，这个库能够优雅地获取命令行参数，不再需要os.args判断个数或者类型。
使用pathlib而不是os.path，这个库能够高效地处理各种文件操作，创建、修改、删除、路径拼接等。
使用logging而不是print，个人更加习惯用日志的方式打印信息。
使用aiohttp而不是requests，当你决定用异步的方式发送网络请求时，就用这个库吧。
使用aiofiles而不是open，当你决定用异步的方式读写文件时，就用这个库吧。
使用lxml，这个库能够使用Xpath语法帮助我们解析html内容。
使用asyncio.create_task而不是for循环，当一组任务不是顺序相关时，不必循环await执行，通过创建任务的方式异步执行。
使用类型提示，标注变量的类型，方便IDE检索。

以上是常用的库或者提示，没有好坏之分，主要是看个人的习惯，一旦一处异步，那就处处异步。

Python版本

当前稳定版本是：3.10.11

Python依赖

aiohttp==3.8.6
aiofiles==23.2.1
lxml==4.9.3

Python源码

import asyncio
import logging
import argparse
from pathlib import Path

import aiohttp
import aiofiles
from lxml import etree

logger = logging.getLogger()
logger.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(formatter)
logger.addHandler(console)

PROXY = "http://127.0.0.1:10808"
SAVE_FOLDER = Path("Downloads")
PICTURE_HOST = "https://girlygirlpic.com"
PICTURE_GATHER = PICTURE_HOST + "/ax/"
PICTURE_SEARCH = PICTURE_HOST + "/sx/"
PICTURE_URL = '//div[@class="post-media-body"]//a[@class="figure-link os-lightbox-activator"]/@href'
ALBUM_NAMES = '//div[@class="post-content-body"]/h4[@class="post-title entry-title"]/a[@class="on-popunder"]'
HEADER = {
    "Origin": PICTURE_HOST,
    "Connection": "close",
    "Cookie": "_user_language=Cn",
    "X-Requested-With": "XMLHttpRequest",
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/115.0"
}

async def download_picture(url: str, save_folder: Path) -> None:
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get(url, proxy=PROXY, headers=HEADER) as response:
                picture_name = url[url.rfind("/") + 1:]
                async with aiofiles.open(save_folder / picture_name, "wb") as handle:
                    async for chunk in response.content.iter_chunked(1024):
                        await handle.write(chunk)
        except Exception as exception:
            logger.error(f"{exception} [URL] {url}")

async def parse_album(url: str, save_folder: Path) -> None:
    album_id = url[url.rfind("/") + 1:]
    header = {**HEADER, **{"Referer": url}}
    async with aiohttp.ClientSession() as session:
        requests_body = {"album_id": album_id}
        async with session.post(PICTURE_GATHER, json=requests_body, proxy=PROXY, headers=header) as response:
            html_content = await response.text()
            image_href = etree.HTML(html_content).xpath(PICTURE_URL)
            logger.info(f"{len(image_href):03d} photos of album [{save_folder.name}]")
            task_list = [asyncio.create_task(download_picture(pic_url, save_folder)) for pic_url in image_href]
            await asyncio.wait(task_list) if len(task_list) > 0 else None
            logger.info(f"album saved [{save_folder.name}]")

async def get_albums(name: str) -> None:
    async with aiohttp.ClientSession() as session:
        requests_body = {"search_keys_tag": name}
        async with session.post(PICTURE_SEARCH, json=requests_body, proxy=PROXY, headers=HEADER) as response:
            html_content = await response.text()
            album_content = etree.HTML(html_content).xpath(ALBUM_NAMES)
            logger.info(f"{len(album_content):03d} albums about {name}")
            task_list = []
            for album in album_content:
                href = album.get("href")
                text = album.text
                save_folder = SAVE_FOLDER / name / text
                save_folder.mkdir(parents=True) if not save_folder.exists() else None
                task_list.append(asyncio.create_task(parse_album(href, save_folder)))
            await asyncio.wait(task_list) if len(task_list) > 0 else None

async def main(names: [str]) -> None:
    await asyncio.wait([asyncio.create_task(get_albums(name)) for name in names])
    logger.info("done")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="download pictures")
    parser.add_argument("name", type=str, nargs='+', help="girls name list")
    args = parser.parse_args()
    asyncio.run(main(args.name))

其中PICTURE_HOST就是域名，需要科学访问，所以PROXY自己配置吧。

运行方法很简单：python main.py xxxx xxxx xxxx，其中xxxx为姓名，多个人名按照空格隔开。

获取完毕后会在当前创建一个Downloads目录，图片会按照姓名以及相册名归类在里面。