- A+
所属分类：python
最近在QQ阅读看《我的徒弟都是大反派》和《假装自己是学霸》。由于第二本刚看不久，而第70章之后收费，算了下第二本的阅读成本有点小贵，就只好暂时告别正版了😓。下面就演示如何python3提取小说到txt文件。
Contents
先找个免费阅读的网站
百度了下，找到了这个https://www.biqumo.com/34_34964/。稍微F12查看了下，章节目录和章节内容都是直接在html页面可以直接找到，不需要经过js代码执行特定方法后才能获取。所以针对这种网站提取操作特别简单，没有任何难度。所以，为了同以往的文章稍作区别，py3实现代码采用多线程的方式来进行爬取。
多线程爬取大致思路
- 先提取小说所有章节目录
- 创建N个爬取线程
- 将总章节数M分配给N个爬取线程
- 每个爬取线程爬取各自负责的C个章节数，并写到各自的文件
- 合并爬取线程已经写入的文件
注意事项
主要是针对中文编码问题，针对该网站，进行了测试，'utf-8','gbk'编码均不能正常decode，最后采用'gb180130'编码完美decode。
PS：常用中文编码有'utf-8'，'gbk'，'gb2312'，'gb18030'，'big5'，'big5hkscs'，一种不行就换另一种测试。
完整实现代码
# -*- coding: utf-8 -*-
import requests
import threading
import time
import os
from bs4 import BeautifulSoup
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko)\
Chrome/59.0.3071.112 Safari/537.36 Vivaldi/1.91.867.48'
}
# 所有章节信息
links = {}
# 首页地址
base_url = 'https://www.biqumo.com'
# 存放结果文件的目标文件夹,需要先行创建目录
des_dir = 'jiazhuangzijishixueba'
# 几种常用的中文编码
# utf8
# gbk
# gb2312
# gb18030
# big5
# big5hkscs
# 网站使用的中文字符编码
cn_charset = 'gb18030'
# 提取所有章节的地址信息
def get_urls():
res = requests.get('https://www.biqumo.com/34_34964/', headers=headers).content.decode(cn_charset)
soup = BeautifulSoup(res, "lxml")
chapters = soup.select('body > div.listmain > dl > dd > a')
begin = False
cnt = 0
for chapter in chapters:
if chapter.text.find('第一章') != -1:
begin = True
if begin:
links[cnt] = {
'a': chapter.get('href'),
'title': chapter.text
}
cnt = cnt + 1
print('章节总数', cnt, '\n')
return cnt
# 提取指定章节内容
def download_content(url, des_file):
try:
res = requests.get(url, headers=headers).content.decode(cn_charset)
except Exception:
raise Exception(url)
soup = BeautifulSoup(res, "lxml")
content_obj = soup.select('#content')
for content in content_obj:
text = content.text
text = text.replace(' ', ' ').replace('/p>', '').replace('<!--', '').replace('←', '')
text = text.replace('天才一秒记住本站地址：www.biqumo.com 笔趣阁手机版阅读网址：m.biqumo.com', '')
text = text.replace(url, '')
text = text.replace('<br/><br/>', '\r\n')
text = text.replace(' ', '\r\n')
text = text + '\r\n'
des_file.write(text.encode('utf-8'))
# 线程各自下载一部分章节
def download_chapters(thread_id, begin_idx, end_idx):
print("Thread id:%s begin_idex:%s end_idx:%s" % (thread_id, begin_idx, end_idx))
idx = begin_idx
des_file = open('{}/book_{}.txt'.format(des_dir, thread_id), 'wb')
while idx < end_idx:
url = '{}{}'.format(base_url, links[idx]['a'])
title = links[idx]['title']
des_file.write(title.encode('utf-8'))
print('Thread id:{}, title:{}, url:{}'.format(thread_id, title, url))
download_content(url, des_file)
idx += 1
time.sleep(0.05)
des_file.close()
class MyThread(threading.Thread):
def __init__(self, thread_id, begin_idx, end_idx):
threading.Thread.__init__(self)
self.threadId = thread_id
self.beginIdx = begin_idx
self.endIdx = end_idx
def run(self):
download_chapters(self.threadId, self.beginIdx, self.endIdx)
# 获取总章节数
chapter_cnt = get_urls()
threads = []
# 开启多少个线程
thread_cnt = 8
if thread_cnt > chapter_cnt:
raise Exception("线程数量必须小于或者等于章节数")
# 平均每个线程下载多少个章节
avg_chapter_num = chapter_cnt // thread_cnt
# 最后还有多少个章节没有被分配
remain_num = chapter_cnt % thread_cnt
print(avg_chapter_num, remain_num)
# 根据配置的线程数创建线程
st_idx = 0
for i in range(thread_cnt):
ed_idx = st_idx + avg_chapter_num
# 如果是最后一个线程，则将未分配的章节也让最后一个线程去下载
if i == thread_cnt-1:
ed_idx = chapter_cnt
thread = MyThread(i, st_idx, ed_idx)
# 添加线程到线程列表
threads.append(thread)
# 开启新线程
thread.start()
st_idx = ed_idx
print("等待所有线程完成")
for t in threads:
t.join()
# 合并每个线程输出的文件
des = open('{}/假装自己是学霸.txt'.format(des_dir), 'wb')
for i in range(thread_cnt):
tmp_file_path = '{}/book_{}.txt'.format(des_dir, i)
# 读取已经下载的中间文件并合并到主文件
f = open(tmp_file_path, 'rb')
des.write(f.read())
f.close()
# 清理中间文件
os.remove(tmp_file_path)
des.close()
print("文件处理完毕，欢迎再次使用")
运行程序稍作等待，即可提取完毕。然后就可以愉快的去阅读了(^__^*)。
