上一章内容

用python从零开始做一个最简单的小说爬虫带GUI界面（2/3）_木木em哈哈的博客-CSDN博客前一章博客我们讲了怎么通过PyQt5来制作图形化界面，并且进行一些基本设置接下来两章我们主要讲核心爬虫代码的实现。https://blog.csdn.net/mumuemhaha/article/details/132457770?spm=1001.2014.3001.5501

前言

本章内容讲的是给出了小说文章链接的情况下，如何爬取小说

出现的一些问题

requests包爬取小说的不便之处

在最开始的时候包括我前段时间写的博客都是利用requests包进行爬取

但是这回出现一个问题

简单来说就是request是顺序执行的

必须要等到上一个网络的请求返回后才会执行下一个步骤

假设我要爬取的小说有2000个章节

每次返回请求并且处理信息都需要1秒的时间

那么总共就需要2000秒也就是半个多小时

要是中间再来个返回超时出现错误的

心态直接要爆炸

返回超时我们可以设置超时等待时间

但是占据大部分时间的依然是网络请求的延迟

那有什么方法可以解决呢

利用aiohttp包来异步爬取小说

介绍

异步是一种比多线程高效得多的并发模型，是无序的，为了完成某个任务，在执行的过程中，不同程序单元之间过程中无需通信协调，也能完成任务的方式，也就是说不相关的程序单元之间可以是异步的。

简单来说就是可以类比小学的一种数学——你可以再烧开水的时候洗菜，在煮饭的时候切菜的那类问题

在python程序中就是你在等待网络回复的数据包时候可以继续发送其他的数据包

起到资源利用趋于最大化的趋势

代码

具体的代码在这

这里只做初步介绍，具体包的使用不展开细讲

main.py

import sys
# PyQt5中使用的基本控件都在PyQt5.QtWidgets模块中
from PyQt5.QtWidgets import QApplication, QMainWindow
# 导入designer工具生成的login模块
from win import Ui_MainWindow
from test_1 import *
from test_3 import *
import time
class MyMainForm(QMainWindow, Ui_MainWindow):
    def __init__(self, parent=None):
        super(MyMainForm, self).__init__(parent)
        self.setupUi(self)
        self.Button_close.clicked.connect(self.close)
        self.Button_run.clicked.connect(self.F_run)

    def F_run(self):
        link_1=self.line_link.text()
        title_1=F_gettitle(link_1)
        self.text_result.setText(f"标题获取成功——{title_1}")
        # file_1=open(f'{title_1}.txt',mode='w',encoding='utf-8  ')
        test_1=F_getyuan(link_1)
        self.text_result.append("提取源代码成功")
        time.sleep(1)
        search_1=F_searchlink(test_1)
        self.text_result.append("提取文章链接成功")
        pachong(search_1,title_1)

if __name__ == "__main__":
    # 固定的，PyQt5程序都需要QApplication对象。sys.argv是命令行参数列表，确保程序可以双击运行
    app = QApplication(sys.argv)
    # 初始化
    myWin = MyMainForm()
    # 将窗口控件显示在屏幕上
    myWin.show()
    # 程序运行，sys.exit方法确保程序完整退出。
    sys.exit(app.exec_())

test_1.py

import requests
import re
import numpy as np
from lxml import etree
#获取文章标题
def F_gettitle(link_0):
    head_qb={
        'User-Agent':'Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Mobile Safari/537.36'
    }
    test_1=requests.get(url=link_0,headers=head_qb)
    test_yuan=test_1.text
    dom=etree.HTML(test_yuan)
    test_2=dom.xpath('/html/body/article[1]/div[2]/div[2]/h1/text()')
    return test_2[0]


#提取源代码
def F_getyuan(link_1):
    head_qb={
        'User-Agent':'Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Mobile Safari/537.36'
    }
    test_1=requests.get(url=link_1,headers=head_qb)
    test_yuan=test_1.text
    test_yuan=str(test_yuan)
    return test_yuan


#查询所有小说章节链接
def F_searchlink(link_2):
    re_1='<a id="haitung" href="(.*?)" rel="chapter">'
    re_1=re.compile(re_1)
    link_3=re.findall(re_1,link_2)
    link_max=np.array([])
    for link_1 in link_3:
        link_4=f'http://www.biquge66.net{link_1}'
        link_max=np.append(link_max,link_4)

    return link_max


# #输出文章内容
# def F_edittxt(link_3):
#     head_qb={
#         'User-Agent':'Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Mobile Safari/537.36'
#     }
#     trytimes = 3
#     for i in range(trytimes):
#         try:
#             proxies = None
#             test_1=requests.get(url=link_3,headers=head_qb, verify=False, proxies=None, timeout=3)
#             if test_1.status_code == 200:
#                 break
#         except:
#             print(f'requests failed {i} time')
#     #提取文章链接
#     re_2='<p>(.*?)</p>'
#     re_2=re.compile(re_2)
#     #提取文章标题
#     re_3='<h1 class="bookname">(.*?)</h1>'
#     re.compile(re_3)
#     test_2=np.array([])
#     test_3=np.array([])
#     test_2=re.findall(re_2,test_1.text)
#     test_3 = re.findall(re_3, test_1.text)
#     #放在数组的最后一个
#     test_2=np.append(test_3,test_2)
#     return test_2

test_3.py

import asyncio
import aiohttp
import re
import numpy as np

title=''


async def F_2(session,url):
    head_qb = {
        'User-Agent': 'Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Mobile Safari/537.36'
    }
    async with session.get(url,verify_ssl=False,headers=head_qb) as resqonse:
        global title
        text=await resqonse.text()
        text=str(text)
        re_2 = '<p>(.*?)</p>'
        re_2 = re.compile(re_2)
        # 提取文章标题
        re_3 = '<h1 class="bookname">(.*?)</h1>'
        re.compile(re_3)
        test_2 = np.array([])
        test_3 = np.array([])
        test_2 = re.findall(re_2, text)
        test_3 = re.findall(re_3, text)
        test_2 = np.append(test_3, test_2)
        for test_max in test_2:
            with open(f'{title}.txt',mode='a',encoding='utf-8') as file:
                file.writelines(test_max)


async def F_1(urls):
    async with aiohttp.ClientSession() as session:
        tasks=[asyncio.create_task(F_2(session,url)) for url in urls]
        await asyncio.wait(tasks)



def pachong(urls_1,title_1):
    global title
    title=title_1
    asyncio.run(F_1(urls_1))
    title=title_1