# 前言
在前面是实战中,我们都是爬取的没有人机校验机制的网站,如果遇到有人机验证的网站是无法爬取的,具体我们在 bs4 基础上,增加 Web Driver 自动化调用浏览器模拟我们的真人请求,对内容进行爬取。
前面的内容:点击跳转
# 实战内容
爬取的网站:http://www.beqege.com/28970/
还是笔趣阁,不过这个比起之前的有人机校验,反爬虫机制,使用之前的代码框架显然不足以满足我们的需求,只会给你返回错误。
# 给代码加入 WebDriver
使用以下代码可实现对这种人机校验简单爬取,同时加入了自动化控制浏览器的 WebDriver 模块、多线程模块。可以多线程的方式爬取网站内容,加快爬取速度,线程可自定义。
博主这里习惯使用的是 Google Chrome 浏览器,所以是调用的 Chrome,请根据自己的实际使用环境决定最后调用的浏览器!
from bs4 import BeautifulSoup | |
from selenium import webdriver | |
import concurrent.futures | |
import time | |
def get_chapter_content(chapter_url): | |
driver = webdriver.Chrome() # 我这里使用的是 Google Chrome,请根据实际情况修改为你的浏览器。 | |
driver.get(chapter_url) | |
time.sleep(5) | |
soup = BeautifulSoup(driver.page_source, "html.parser") | |
chapter_title = soup.title.text.strip().replace("_笔趣阁_beqege.com", "") | |
content_paragraphs = soup.select("#content p") | |
driver.quit() | |
chapter_content = f"{chapter_title}\n\n" | |
for paragraph in content_paragraphs: | |
text = paragraph.text.strip().replace("_笔趣阁_beqege.com", "") | |
if "小说网..org,最快更新至尊神医之帝君要下嫁最新章节!" in text: | |
continue | |
chapter_content += f"{text}\n" | |
return chapter_content | |
base_url = "https://www.beqege.com/28970/" | |
driver = webdriver.Chrome() | |
driver.get(base_url) | |
time.sleep(5) | |
soup = BeautifulSoup(driver.page_source, "html.parser") | |
title = soup.title.text.strip().replace("_笔趣阁_beqege.com", "") | |
chapter_list = soup.select("#list dl dd a") | |
driver.quit() | |
file_path = "爬取.txt" | |
with open(file_path, "w", encoding="utf-8") as file: | |
file.write(f"小说标题: {title}\n\n") | |
# 设置自定义的窗口数,例如 3 个窗口 (线程) | |
max_workers = 3 | |
# 使用 ThreadPoolExecutor 创建线程池,指定窗口数 | |
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: | |
# 使用 map 方法并行爬取章节内容 | |
chapter_contents = executor.map(get_chapter_content, [base_url + chapter["href"] for chapter in chapter_list]) | |
# 将爬取到的内容写入文件 | |
for content in chapter_contents: | |
file.write(content) | |
print(f"\n全部章节已爬取并写入到 {file_path}") |
# 爬虫时隐藏浏览器窗口
以上代码的不足之处在于每次爬取一个新的章节就一口气打开一个新的窗口,如果你是多线程那么会打开多个窗口。可能会影响我们的一些操作,那么我们如何隐藏这个窗口弹出,让它静默执行呢?
from bs4 import BeautifulSoup | |
from selenium import webdriver | |
from selenium.webdriver.chrome.options import Options | |
import concurrent.futures | |
import time | |
def get_chapter_content(chapter_url): | |
options = Options() | |
options.add_argument('--headless') # 添加该选项实现无头模式,即不显示浏览器窗口 | |
driver = webdriver.Chrome(options=options) | |
driver.get(chapter_url) | |
time.sleep(5) | |
soup = BeautifulSoup(driver.page_source, "html.parser") | |
chapter_title = soup.title.text.strip().replace("_笔趣阁_beqege.com", "") | |
content_paragraphs = soup.select("#content p") | |
driver.quit() | |
chapter_content = f"{chapter_title}\n\n" | |
for paragraph in content_paragraphs: | |
text = paragraph.text.strip().replace("_笔趣阁_beqege.com", "") | |
if "小说网..org,最快更新至尊神医之帝君要下嫁最新章节!" in text: | |
continue | |
chapter_content += f"{text}\n" | |
return chapter_content | |
base_url = "https://www.beqege.com/28970/" | |
options = Options() | |
options.add_argument('--headless') # 添加该选项实现无头模式,即不显示浏览器窗口 | |
driver = webdriver.Chrome(options=options) | |
driver.get(base_url) | |
time.sleep(5) | |
soup = BeautifulSoup(driver.page_source, "html.parser") | |
title = soup.title.text.strip().replace("_笔趣阁_beqege.com", "") | |
chapter_list = soup.select("#list dl dd a") | |
driver.quit() | |
file_path = "爬取.txt" | |
with open(file_path, "w", encoding="utf-8") as file: | |
file.write(f"小说标题: {title}\n\n") | |
# 设置自定义的窗口数,例如 3 个窗口 (线程) | |
max_workers = 3 | |
# 使用 ThreadPoolExecutor 创建线程池,指定窗口数 | |
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: | |
# 使用 map 方法并行爬取章节内容 | |
chapter_contents = executor.map(get_chapter_content, [base_url + chapter["href"] for chapter in chapter_list]) | |
# 将爬取到的内容写入文件 | |
for content in chapter_contents: | |
file.write(content) | |
print(f"\n全部章节已爬取并写入到 {file_path}") |
# 对本次代码分析 (未包含后期加入的隐藏窗口库)
本次 Python 脚本旨在从一个特定包含想要的小说的网站中提取章节内容。以下为代码分析:
- 导入库:
bs4
和BeautifulSoup
用于网页抓取。selenium
用于浏览器自动化。concurrent.futures
用于并行执行任务。time
用于引入延迟。
from bs4 import BeautifulSoup | |
from selenium import webdriver | |
import concurrent.futures | |
import time |
- 定义函数
get_chapter_content
:- 此函数以
chapter_url
作为输入。 - 使用 Selenium 在 Chrome 浏览器中打开 URL。
- 经过 5 秒的延迟(使用
time.sleep(5)
),使用 BeautifulSoup 提取页面源代码。 - 从页面源代码中提取章节标题和内容段落。
- 然后关闭浏览器。
- 函数返回格式化的章节内容。
- 此函数以
def get_chapter_content(chapter_url): | |
# ... | |
# (提取章节内容的代码) | |
# ... | |
return chapter_content |
- 设置基础 URL 并抓取章节列表:
- 脚本设置了
base_url
并使用 Selenium 在 Chrome 浏览器中打开 URL。 - 经过 5 秒的延迟后,使用 BeautifulSoup 提取页面源代码。
- 提取小说标题和章节链接列表。
- 然后关闭浏览器。
- 脚本设置了
base_url = "https://www.beqege.com/28970/" | |
# ... | |
# (提取小说标题和章节链接的代码) | |
# ... |
- 将小说标题写入文件:
- 脚本以写入模式打开一个文件("爬取.txt")并将小说标题写入其中。
file_path = "爬取.txt" | |
with open(file_path, "w", encoding="utf-8") as file: | |
file.write(f"小说标题: {title}\n\n") |
- 并行执行章节内容提取:
- 将
max_workers
变量设置为并发线程的数量(在本例中为 3)。 - 使用
ThreadPoolExecutor
创建一个具有指定工作线程数的线程池。 - 使用
map
方法并行执行get_chapter_content
函数以获取每个章节的内容。 - 然后将章节内容写入文件。
- 将
max_workers = 3 | |
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: | |
chapter_contents = executor.map(get_chapter_content, [base_url + chapter["href"] for chapter in chapter_list]) | |
for content in chapter_contents: | |
file.write(content) |
- 打印完成消息:
- 脚本打印一条消息,指示所有章节已经抓取并写入文件。
print(f"\n全部章节已爬取并写入到 {file_path}") |
# 结语
tpis:网络爬取可能违反某些网站的服务条款,使用这类工具时务必确保合规性。此外,网站结构可能随时间而变化,如果 HTML 结构修改,代码可能需要调整。
本博客一切文章内容仅用于学习交流使用,请尊重技术初衷。