标签：chapter 异步 info 协程进阶 url self 爬虫 book

顶点小说进阶

建议

这篇顶点小说进阶包括（数据入库、异步爬虫）

看之前可以先看我之前发布的文章（从零开始学习Python爬虫：顶点小说全网爬取实战）

入库

# 入库
def save_to_mysql(db_name, table_name, table_column_str, table_info_str):
    db = pymysql.connect(user='host', password='Lun18532104295', db=db_name)
    cursor = db.cursor()
    sql = f'insert into {table_name}({table_column_str}) values({table_info_str})'
    cursor.execute(sql)
    db.commit()

完善主线程

if __name__ == '__main__':
    # 获取小说分类url
    type_lists = get_type()
    # 分类url默认为第一页
    for first_page_url in type_lists:
        # 获取带分类的url的前半截
        type_url = first_page_url.split('1')[0]
        # 获取此分类下最大页
        max_page = get_max_page(first_page_url)
        # 生成此分类下每一页url
        for every_page in range(1, int(max_page[0]) + 1):
            every_page_url = f"{type_url}{every_page}/"
            # 获取小说列表页信息
            book_info_lists = get_book_info(every_page_url)
            # 获取章节列表url
            for book_info in book_info_lists:
                print(f"爬取小说:{book_info[1]}...")
                # 入库小说信息
                save_to_mysql('xiaoshuo', 'books', 'book_id, book_name, new_chapter, author, update_time, font_num, summary', ''.join(book_info))

                book_id = book_info[0]
                chapter_urls = get_chapter_urls(f"https://www.cdbxs.com/booklist/b/{book_id}/1")
                for chapter_url in chapter_urls:
                    # print(chapter_url)
                    chapter_info = get_chapter_info(chapter_url)
                    # 入库小说章节信息
                    print(chapter_info)
                    save_to_mysql('xiaoshuo', 'chapters', 'chapter_name,chapter_content', ''.join(chapter_info))

                    # print(f"title:{chapter_info[0]}")
                    # print(f"content:{chapter_info[1]}")

异步爬虫

介绍

爬虫是IO密集型任务。比如我们使用requests库来爬取某个站点的话，发出一个请求之后，程序必须要等待网站返回响应之后才能接着运行，而在等待响应的过程中，整个爬虫程序是一直等待的，实际上没有做任何的事情。
异步是提高程序运行效率的一种有效方法。

基本概念

阻塞
- 阻塞状态指程序未得到所需计算资源时被挂起的状态。程序在等待某个操作完成期间，自身无法继续处理其他的事情，则称该程序在操作上是阻塞的。
- 常见的阻塞形式有：网络I/O阻塞、磁盘I/O阻塞、用户输入阻塞等，阻塞是无处不在的，包括CPU切换上下文时，所有的进程都无法真正处理事情，他们会被阻塞。如果是多核CPU则正在执行上下文切换操作的核不可被利用。
非阻塞
- 程序在等待某操作过程中，自身不被阻塞，可以继续处理其他的事情，则称该程序在该操作上是非阻塞的。非阻塞并不是在任何程序级别、任何情况下都可以存在的，仅当程序封装的级别可以囊括独立的子程序单位时，它才可能存在非阻塞状态。
- 非阻塞的存在是因为阻塞存在，正因为某个操作阻塞导致的耗时与效率低下，我们才要把它变成非阻塞的
同步
- 不同程序单位为了完成某个任务，在执行过程中需靠某种通信方式以协调一致，我们称这些程序单位是同步执行的。
- 例如购物系统中更新商品库存，需要用“行锁”作为通信信号，让不同的更新请求强制排队顺序执行，那更新库存的操作是同步的。简而言之，同步意味着有序。
异步
- 为完成某个任务，不同程序单元之间过程中无需通信协调，也能完成任务的方式，不相关的程序单元之间可以是异步的。
- 例如，爬虫下载网页。调度程序调用下载程序后，即可调度其他任务，而无需与该下载任务保持通信以协调行为。不同网页的下载、保存等操作都是无关的，也无需相互通信协调。这些异步操作的完成时刻并不确定。简而言之，异步意味着无序。
多进程
- 多进程就是利用CPU的多核优势，在同一时间并行地执行多个任务，可以大大提高执行效率。
协程（Coroutine）
- 又称微线程、纤程，协程是一种用户态地轻量级线程。
- 协程拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来地时候，回复先前保存的寄存器上下文和栈。因此协程能保留上一次调用的状态，即所有局部状态的一个特定组合，每次过程重入时，就相当于进入上一次调用的状态。
- 协程本质上是一个单进程，协程相对于多进程来说，无需线程上下文切换的开销，无需原子操作锁定及同步的开销。我们可以使用协程来实现异步操作。比如在网络爬虫场景下，我们发出一个请求之后，需要等待一定的时间才能得到响应，但其实在这个等待过程中，程序可以干许多其他的事情，等到响应得到之后才切换回来继续处理，这样可以充分利用CPU和其他资源，这就是协程的优势。

协程

asyncio

event_loop：事件循环池，相当于一个无限循环，我们可以把一些函数注册到这个事件池中，当满足条件发生的时候，就会调用对应的处理方法。
coroutine：协程对象，我们可以将协程对象注册到事件循环池中，它会被事件循环池调用。我们可以用async关键字来定义一个方法，那这个方法在调用时不会立即被执行，而是返回一个协程对象。
task：任务，对协程对象的进一步封装，包含了任务的各个状态。
await：用来挂起阻塞方法的执行

import asyncio


async def execute(x):
    print(f"Number: {x}")

# 创建协程对象
coroutine = execute(10)
# 封装成任务（可省略，将协程对象放入事件循环池后自动封装为任务）
task = asyncio.ensure_future(coroutine)
print(task)
# 创建事件循环池
loop = asyncio.get_event_loop()
# 注册任务，开始执行
loop.run_until_complete(task)
print(task)

task对象的绑定回调操作

可以为某个task绑定一个回调方法

# 为task对象绑定回调函数
async def call_on():
    status = requests.get("https://www.baidu.com")
    return status

def call_back(task):
    print(f"status: {task.result()}")

# 创建协程对象
coroutine1 = call_on()
# 将协程对象封装为任务
task1 = asyncio.ensure_future(coroutine1)
# 创建事件循环池
loop = asyncio.get_event_loop()
# 为task对象绑定回调函数
task1.add_done_callback(call_back)
# 注册任务
loop.run_until_complete(task1)

异步爬虫实现（基于协程）

await后面的对象

一个原生的coroutine对象
一个返回coroutine对象的生成器
一个包含await方法的对象返回的一个迭代器

aiohttp

aiohttp是一个支持异步请求的库，利用它和asyncio配合我们可以方便地实现异步请求操作。下面以访问博客里面的文章，并返回response.text()为例，实现异步爬虫。

import time
import requests
import aiohttp
from lxml import etree
import asyncio
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s : %(message)s')
url = 'https://blog.csdn.net/nav/ai'
start_time = time.time()


# 获取博客里文章链接
def get_urls():
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 Edg/126.0.0.0'
    }
    source = requests.get(url=url, headers=headers).text
    urls = etree.HTML(source).xpath(
        "//div[@class='content']/a/@href")
    return urls


# 异步请求博客里文章链接(aiohttp)
async def request_page(url):
    logging.info(f'scraping {url}')
    async with aiohttp.ClientSession() as session:
        # 发起请求
        response = await session.get(url)
        return await response.text()


# main函数
def main():
    # 获取博客里文章链接
    urls = get_urls()
    # 创建任务列表
    tasks = [asyncio.ensure_future(request_page(url)) for url in urls]
    # 创建事件循环池
    loop = asyncio.get_event_loop()
    # 处理协程对象列表(不可少，用于并发运行多个可等待对象（比如协程），并将它们的结果收集到一个列表中。)
    results = asyncio.gather(*tasks)
    # 注册任务，开始执行
    loop.run_until_complete(results)


if __name__ == '__main__':
    main()
    end_time = time.time()
    logging.info(f"total time {end_time - start_time} seconds")

Question1：text和content啥区别？

Answer1：

响应对象中，有两个常用的属性：text 和 content。

text:

text 属性返回的是响应内容的字符串形式，通常是根据 HTTP 响应的内容推测出的字符编码来解码的文本。例如，如果服务器返回的是 HTML 内容，那么text 属性会返回解码后的 HTML 文本内容。
content:

content 属性返回的是响应内容的字节形式，即原始的未解码数据。这个属性通常用于获取非文本类型的内容，比如图片、音频、视频等文件。对于文本内容，你也可以通过 response.content.decode('utf-8') 等方法将其转换为字符串形式。
区别总结如下：

text: 返回解码后的文本内容，适用于文本类型的响应数据，如 HTML、JSON 等。
content: 返回原始的字节形式的响应内容，适用于任何类型的响应数据，包括文本和非文本。

Question2：为什么用aiohttp来配合asyncio而不是用requests？

aiohttp为异步请求库，requests为同步请求库。
要实现爬虫异步请求，需要用到await来将请求挂起，而await后不能跟同步请求，需要跟异步请求。

线程

异步爬虫实现（基于线程）

import time
import requests
from lxml import etree
import logging
from concurrent.futures import ThreadPoolExecutor

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s : %(message)s')
url = 'https://blog.csdn.net/nav/ai'
start_time = time.time()


# 获取博客里文章链接
def get_urls():
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 Edg/126.0.0.0'
    }
    source = requests.get(url=url, headers=headers).text
    urls = etree.HTML(source).xpath(
        "//div[@class='content']/a/@href")
    return urls


# 请求博客里文章链接
def request_page(url):
    logging.info(f'scraping {url}')
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 Edg/126.0.0.0'
    }
    return requests.get(url=url, headers=headers).text


# main函数
def main():
    # 获取博客里文章链接
    urls = get_urls()
    # 创建线程池
    with ThreadPoolExecutor(max_workers=6) as executor:
        executor.map(request_page, urls)


if __name__ == '__main__':
    main()
    end_time = time.time()
    logging.info(f"total time {end_time - start_time} seconds")

进程、线程、协程的关系与区别

进程：进程是计算机中的程序关于某数据集合上的一次运行活动，是系统进行资源分配和调度的基本单位，是操作系统结构的基础
线程（Thread）：线程有时被称为轻量级进程，是程序执行流的最小单元。一个标准的线程由线程ID、当前指令指针(PC)、寄存器集合和堆栈组成。
协程：协程是一种比线程更加轻量级的一种函数。正如一个进程可以拥有多个线程一样，一个线程可以拥有多个协程。协程不是被操作系统内核所管理的，而是完全由程序所控制的，即在用户态执行。这样带来的好处是：性能有大幅度的提升，因为不会像线程切换那样消耗资源。

联系：

  协程既不是进程也不是线程，协程仅是一个特殊的函数。协程、进程和线程不是一个维度的。
  一个进程可以包含多个线程，一个线程可以包含多个协程。虽然一个线程内的多个协程可以切换但是这多个协程是串行执行的，某个时刻只能有一个协程在运行，没法利用CPU的多核能力。
  线程与进程一样，也存在上下文切换问题。
  进程的切换者是操作系统，切换时机是根据操作系统自己的切换策略来决定的，用户是无感的。进程的切换内容包括页全局目录、内核栈和硬件上下文，切换内容被保存在内存中。 进程切换过程采用的是“从用户态到内核态再到用户态”的方式，切换效率低。
  线程的切换者是操作系统，切换时机是根据操作系统自己的切换策略来决定的，用户是无感的。线程的切换内容包括内核栈和硬件上下文。线程切换内容被保存在内核栈中。线程切换过程采用的是“从用户态到内核态再到用户态”的方式，切换效率中等。
  协程的切换者是用户(编程者或应用程序),切换时机是用户自己的程序来决定的。协程的切换内容是硬件上下文，切换内存被保存在用自己的变量(用户栈或堆)中。协程的切换过程只有用户态(即没有陷入内核态),因此切换效率高。

异步mysql（aiomysql）和同步mysql（pymsql）的使用

mysql操作为同步代码（也就是必须等待它的返回才会往下执行，如果一个SQL执行得比较久，那么会直接卡死这个线程），在异步环境中无法调用同步代码，所以需要调用异步mysql进行入库。

import asyncio
import aiomysql
# UUID是 通用唯一识别码(Universally Unique Identifier)的缩写.
import shortuuid
import pymysql


# 异步
# aiomysql做数据库连接的时候，需要这个loop对象
async def async_basic(loop):
    pool = await aiomysql.create_pool(
        host="127.0.0.1",
        port=3306,
        user='root',
        password='123456',
        db='test',
        loop=loop
    )
    async with pool.acquire() as conn:
        async with conn.cursor() as cursor:
            for x in range(10000):
                content = shortuuid.uuid()
                sql = f"insert into mybrank(brank) values('{content}')"
                # 执行sql语句
                await cursor.execute(sql)
            await conn.commit()

    # 关闭连接池
    pool.close()
    await pool.wait_closed()


# 同步
def sync_basic():
    conn = pymysql.connect(
        host='127.0.0.1',
        port=3306,
        user='root',
        password='123456',
        db='dvwa',
    )
    with conn.cursor() as cursor:
        for x in range(10000):
            content = shortuuid.uuid()
            sql = f"insert into guestbook(comment_id,comment,name) values(2,'asd','{content})"
            # 执行sql语句
            cursor.execute(sql)
        conn.commit()


if __name__ == '__main__':
    # 异步: 数量大时用异步
    loop = asyncio.get_event_loop()
    loop.run_until_complete(async_basic(loop))
    # 同步:sync_basic()
    sync_basic()

案例：顶点小说完善(基于aiohttp、aiomysql、协程)

建议：基于我上篇文章的顶点小说基础，进行异步优化。

异步优化思路：

aiohttp、协程：有一个scrape_api方法：异步请求一个网址获取页面源代码
获取到源代码正常xpath解析数据
基于aiomysql入库：有个init_pool方法：初始化数据库连接池，有个close_pool方法：关闭数据库连接池；入库代码改成异步
整体用类Spider实现，有属性session（异步网络请求）、semaphore（设置协程数量）
对于爬取每本小说的所有章节的标题和正文，作为任务放到事件循环池中并发执行

import asyncio
import logging
import time
import requests
from lxml import etree
import aiohttp
import aiomysql
from aiohttp import ContentTypeError

CONCURRENCY = 4

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s : %(message)s')


class Spider(object):
    def __init__(self):
        # 方便设置头部信息、代理IP、cookie信息等
        self.session = None
        # 设置协程数量
        self.semaphore = asyncio.Semaphore(CONCURRENCY)
        # 限制协程的并发数:
        # 如果并发数没有达到限制: 那么async with semaphore会瞬间执行完成,进入里面的正式代码中
        # 如果并发数已经达到了限制，那么其他的协程对象会阻塞在asyn with semaphore这个地方，直到正在运行的某个协程对象完成了，退出了，才会放行一个新的协程对象去替换掉这个已经完成的协程对象

    # 初始化数据库连接池
    async def init_pool(self):
        self.pool = await aiomysql.create_pool(
            host="127.0.0.1",
            port=3306,
            user="root",
            password="123456",
            db=f"dingdian",
            autocommit=True  # Ensure autocommit is set to True for aiomysql
        )
        # 在 aiomysql.create_pool 方法中，不需要显式传递 loop 参数。aiomysql 会自动使用当前的事件循环（即默认的全局事件循环）。

    # 关闭数据库连接池
    async def close_pool(self):
        if self.pool:
            self.pool.close()
            await self.pool.wait_closed()

    # 获取url源码
    async def scrape_api(self, url):
        # 设置协程数量
        async with self.semaphore:
            logging.info(f"scraping {url}")
            try:
                async with self.session.get(url) as response:
                    # 控制爬取（或请求）的速率，以避免对目标服务器造成过多的负荷或请求频率过高而被封禁或限制访问。
                    await asyncio.sleep(1)
                    # 在异步环境中，可能需要使用 response.content.read() 或 await response.text() 来获取文本内容。
                    return await response.text()
            except ContentTypeError as e:  # aiohttp 的 ContentTypeError 异常: 请求内容类型错误 或者 响应内容类型错误
                # exc_info=True 参数将导致 logging 模块记录完整的异常信息，包括栈跟踪，这对于调试非常有用。
                logging.error(f'error occurred while scraping {url}', exc_info=True)

    # 获取小说分类url
    async def get_type(self):
        url = "https://www.cdbxs.com/sort/"
        source = await self.scrape_api(url)
        href_lists = etree.HTML(source).xpath('//ul[@class="nav"]/li/a/@href')[2:-4]
        type_lists = []
        for href in href_lists:
            type_lists.append(f"{url}{href.split('/')[2]}/1/")
        # print(type_lists)
        return type_lists

    # 获取最大页
    async def get_max_page(self, first_page_url):
        source = await self.scrape_api(first_page_url)
        # print(source)
        max_page = etree.HTML(source).xpath('//a[13]/text()')
        return max_page

    # 获取小说列表页信息
    async def get_book_info(self, every_page_url):
        source = await self.scrape_api(every_page_url)
        book_lists = []

        lis = etree.HTML(source).xpath("//ul[@class='txt-list txt-list-row5']/li")
        for li in lis:
            book_id_url = li.xpath("span[@class='s2']/a/@href")[0]
            book_id = book_id_url.split('/')[3]
            # 书名
            book_name = li.xpath("span[@class='s2']/a/text()")[0]
            # 最新章节
            new_chapter = li.xpath("span[@class='s3']/a/text()")[0]
            # 作者
            author = li.xpath("span[@class='s4']/text()")[0]
            # 更新时间
            update_time = li.xpath("span[@class='s5']/text()")[0]

            source = await self.scrape_api(f"https://www.cdbxs.com{book_id_url}")
            # 字数
            font_num = etree.HTML(source).xpath("//p[6]/span/text()")[0]
            # 摘要
            summary = etree.HTML(source).xpath("//div[@class='desc xs-hidden']/text()")[0]

            # 以元组添加至 book_lists
            # print((book_id, book_name, new_chapter, author, update_time, font_num, summary))
            book_lists.append((book_id, book_name, new_chapter, author, update_time, font_num, summary))
        return book_lists

    # 获取章节urls
    async def get_chapter_urls(self, chapter_list_url):
        source = await self.scrape_api(chapter_list_url)
        # 章节url
        chapter_urls = map(lambda x: "https://www.cdbxs.com" + x, etree.HTML(source).xpath(
            "//div[@class='section-box'][2]/ul[@class='section-list fix']/li/a/@href | //div[@class='section-box'][1]/ul[@class='section-list fix']/li/a/@href"))

        return chapter_urls

    # 获取章节详情信息
    async def get_chapter_info(self, chapter_url):
        source = await self.scrape_api(chapter_url)
        # 标题
        title = etree.HTML(source).xpath("//h1[@class='title']/text()")
        # 正文
        content = ''.join(etree.HTML(source).xpath("//div[@id='nb_content']/dd//text()"))
        if title:
            return f'\'{title[0]}\'', f'\'{content}\''
        else:
            return '', f'\'{content}\''

    # 入库
    async def save_to_mysql(self, table_name, table_column_str, table_info_str):
        async with self.pool.acquire() as conn:
            async with conn.cursor() as cursor:
                sql = f'insert into {table_name}({table_column_str}) values{table_info_str}'
                # 执行SQL语句
                await cursor.execute(sql)
                await conn.commit()

    async def main(self):
        # headers
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 Edg/126.0.0.0"
        }
        # 建立异步请求需要的session（主要加header头信息以及代理，cookie等头信息）
        self.session = aiohttp.ClientSession(headers=headers)
        # 获取小说分类url
        type_lists = await self.get_type()
        # 分类url默认为第一页
        for first_page_url in type_lists:
            # 获取带分类的url的前半截
            type_url = first_page_url.split('1')[0]
            # 获取此分类下最大页
            max_page = await self.get_max_page(first_page_url)
            # 生成此分类下每一页url
            for every_page in range(1, int(max_page[0]) + 1):
                every_page_url = f"{type_url}{every_page}/"
                # 获取小说列表页信息
                book_info_lists = await self.get_book_info(every_page_url)
                # 获取章节列表url
                for book_info in book_info_lists:
                    print(f"爬取小说:{book_info[1]}...")
                    # 初始化数据库连接池
                    await self.init_pool()
                    # 入库小说信息
                    await self.save_to_mysql('books',
                                             'book_id, book_name, new_chapter, author, update_time, font_num, summary',
                                             book_info)

                    # 获取章节urls
                    book_id = book_info[0]
                    chapter_urls = await self.get_chapter_urls(f"https://www.cdbxs.com/booklist/b/{book_id}/1")
                    # 生成scrape_detail任务列表
                    scrape_detail_tasks = [asyncio.ensure_future(self.get_chapter_info(chapter_url)) for chapter_url in
                                           chapter_urls]
                    # 并发执行任务，获取结果
                    chapter_details = list(
                        await asyncio.gather(*scrape_detail_tasks))  # await asyncio.gather(*scrape_detail_tasks生成元组
                    # 入库
                    # 1.添加book_id 到 chapter_detail
                    for i in range(len(chapter_details)):
                        chapter_detail = list(chapter_details[i])
                        chapter_detail.append(book_id)
                        chapter_detail = tuple(chapter_detail)
                        chapter_details[i] = chapter_detail
                    # 2.保存至数据库
                    [await self.save_to_mysql('chapters', 'chapter_name,chapter_content, bid',
                                              chapter_detail) for chapter_detail in chapter_details]
        # 关闭连接池
        self.close_pool()
        # 关闭连接
        await self.session.close()


if __name__ == '__main__':
    # 开始时间
    start_time = time.time()
    # 初始化Spider
    spider = Spider()
    # 创建事件循环池
    loop = asyncio.get_event_loop()
    # 注册
    loop.run_until_complete(spider.main())
    # 结束时间
    end_time = time.time()
    logging.info(f'total time: {end_time - start_time}')

后续发布爬虫更多精致内容（按某培训机构爬虫课程顺序发布，欢迎关注后续发布）

更多精致内容，关注公众号：[CodeRealm]

标签：chapter,异步,info,协程,进阶,url,self,爬虫,book
From： https://www.cnblogs.com/CodeRealm/p/18289059

异步优化与数据入库：顶点小说爬虫进阶实战