现在 python 写爬虫基本都是用scrapy，但是有时候一些小任务没必要用这么大的框架。而且希望代码自由度高一些，可以随心所欲写。这时候就需要用到 aiohttp 这个库，它基于 asyncio 编写,主要解决这类问题。

什么？你说 requests?这确实是一个优秀的http库，可它是阻塞式IO，太慢了。但是，其API设计优秀，文档健全，非常适合新手入门使用。

这是群里一位朋友需要的抓取淘宝主图的爬虫。随手写了一下，就全写在单文件内，没有以项目形式解耦分成很多文件。希望能给正在学习协程的朋友一点参考。

import os

import logging
import aiohttp
import asyncio
import aiofiles
import pandas as pd
from pyquery import PyQuery
from multiprocessing import cpu_count

# 【使用方法】：将要抓取主图的宝贝 ID 填入程序生成的 items.csv 内，一行一个。

# 日志等级及格式
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s 【%(levelname)s】=> %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)


class AsyncSpider(object):
    def __init__(self):
        self.BASE_DIR = os.path.dirname(os.path.abspath(__file__))
        self.BASE_URL = "https://detail.tmall.com/item.htm?id="
        self.checker()
        self.loader()
        self.total = 0
        self.failed = 0

    def checker(self):
        """
        检查必要文件及文件夹，如果不存在就自行创建。
        """
        folders = ["files", "images"]
        files = ["items.csv", "failed.csv"]

        for folder in folders:
            path = os.path.join(self.BASE_DIR, folder)
            if not os.path.exists(path):
                os.makedirs(path)

        for file in files:
            path = os.path.join(self.BASE_DIR, "files", file)
            if not os.path.exists(path):
                f = open(path, "w")
                exit("将要下载的宝贝 id 填写至 items.csv 后再运行程序。")
            f = open(os.path.join(self.BASE_DIR, "files", "failed.csv"), "w")
            f.close()

    def loader(self):
        try:
            path = os.path.join(self.BASE_DIR, "files", "items.csv")
            df = pd.read_csv(path, header=None)
            items = df[0].to_list()
            return items
        except Exception as e:
            if str(e) == "No columns to parse from file":
                logging.error("items.csv 数据为空，无可执行任务，程序退出。")
            else:
                logging.error(e)
            exit()

    async def error_handler(self, url):
        self.failed += 1
        item = url.split("id=")[-1] + "n"
        path = os.path.join(self.BASE_DIR, "files", "failed.csv")
        async with aiofiles.open(path, "ab") as f:
            await f.write(item.encode())
            await f.close()

    @staticmethod
    async def fetch(session, url):
        logging.info(f"Start: {url}")
        async with session.get(url) as response:
            if response.status in [200, 201]:
                # 如果 url 以『.jpg』结尾，则认定为图片地址，返回 byte。否则，认定为网址，返回 str 便于解析。
                if url.endswith(".jpg"):
                    return await response.read()
                else:
                    return await response.text()

    # 解析 html 获取主图 url
    @staticmethod
    async def parser(html):
        pq = PyQuery(html)
        # 获取所有图片的url，如果 url 中存在『_430x430q90』则认定为主图，将主图 url 返回
        for link in pq.items("img"):
            src = link.attr("src")
            if src:
                if "_430x430q90" in src or "400x400.jpg" in src:
                    return "https:" + src
        return

    # 图片存储
    async def save(self, session, url, path):
        # 获取图片内容
        img = await self.fetch(session, url)
        # 将内容异步存储
        async with aiofiles.open(path, "wb") as f:
            await f.write(img)
            await f.close()
            logging.info(f"Saved image {path}")

    async def main(self, url, semaphore):
        self.total += 1
        async with semaphore:
            async with aiohttp.ClientSession() as session:
                html = await self.fetch(session, url)
                image_src = await self.parser(html)
                if image_src:
                    image_name = url.split("id=")[-1] + ".jpg"
                    image_path = os.path.join(self.BASE_DIR, "images", image_name)
                    await self.save(session, image_src, image_path)
                else:
                    await self.error_handler(url)

    def run(self):
        items = self.loader()
        urls = map(lambda x: self.BASE_URL + str(x), items)

        semaphore = asyncio.Semaphore(cpu_count() * 2)
        loop = asyncio.get_event_loop()
        futures = [asyncio.ensure_future(self.main(url, semaphore)) for url in urls]
        tasks = asyncio.gather(*futures)
        loop.run_until_complete(tasks)
        loop.close()
        logging.info(f"运行完毕。任务数共 {self.total} 。失败 {self.failed}")


if __name__ == "__main__":
    async_spider = AsyncSpider()
    async_spider.run()