#! /usr/bin/env python
# -*- coding: utf-8 -*-
import asyncio
import aiohttp
import re
import aiofiles
import time
# time是同步的,不要进入异步代码
LOOP = asyncio.get_event_loop()
#
HEADERS = {"Cookie": "__cdnuid=efaf90be3615c5e79e92852f271af777; jieqiVisitTime=jieqiArticlesearchTime%3D1534652382",
"Referer": "http://www.biquge.com.tw/",
"User-Agent": "Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.99 Mobile Safari/537.36"}
BASE_URL = "http://www.biquge.com.tw"
async def get_html(session, url):
try:
# 6.31415926535 66.30009722709656秒 ----左边是时间没有控制好的错误例子,好像重复执行了500多次
async with session.get(url=url, timeout=10.31415926535) as resp:
# 这个timeout非常重要,笔趣阁的服务器是有点渣,他既不拒绝你又不答应你,时间自己好好考虑
# 我这里采用的是回调,可以说,如果不懂异常处理和回调,你不要看了
if not resp.status // 100 == 2:
print(resp.status)
print("爬取", url, "出现错误")
else:
resp.encoding = 'gb18030'
text = await resp.text()
return text
except Exception as e:
print("出现错误", e)
await get_html(session, url)
def parse_url(html):
# 解析出来url
pattern_url = re.compile('<dd><a href="(.*?\.html)">.*?</a></dd>')
url_list = re.findall(pattern=pattern_url, string=html)
return [BASE_URL + each for each in url_list]
def novel_content(html):
# 返回两个列表
pattern_title = re.compile('<h1>([\s\S]*?)</h1>')
pattern_content = re.compile('(?<= )(.*?)(?=<br)')
title_list = re.findall(pattern_title, html)
content_list = re.findall(pattern_content, html)
return title_list, content_list
async def download(title_list, content_list):
# 下载 第二次测试下载只花了十三秒 643万字到手 13.034078598022461
async with aiofiles.open(r'C:\Users\2191751138\Desktop\个人数据\小说\执魔\{}.txt'.format(title_list[0]), 'a',
encoding='utf-8') as f:
# 列表直接写入,不拆了,浪费时间的事情不做
await f.write('{}'.format(str(content_list)))
# print(title, "下载完成")
async def get_content(session, chapter_url):
# 错误处理很重要
try:
html = await get_html(session=session, url=chapter_url)
title_list, content_list = novel_content(html)
except Exception as e:
print("出现错误,", e)
await get_content(session, chapter_url)
else:
# 进行下载
await download(title_list, content_list)
print(title_list, url)
async def run(url):
async with aiohttp.ClientSession() as session:
session.headers = HEADERS
html = await get_html(session, url)
url_list = parse_url(html)
tasks = []
for each in url_list:
tasks.append(asyncio.ensure_future(get_content(session, each)))
await asyncio.gather(*tasks)
if __name__ == '__main__':
func = lambda: time.time()
start = func()
url = "http://www.biquge.com.tw/14_14055/" # 三寸人间
url=input("请输入笔趣阁书籍url:")
# 不要尝试章节过多的,一方面写的也不怎么样,一方面这里有个缺点
LOOP.run_until_complete(run(url))
print(func() - start)
# 我qq在上面,有不懂可以问我,但是图方便可以直接到下面提问,但是我不一定能回答出来
# 注意一下,网络问题还是很大的,上限我不知道,估计应该是10以内,下限的话也不知道,我这边最慢78秒
'''
还是想说点什么
这是我最认真的一次写文章,我是面向百度编程的,看了一下其他的异步版本,没有看见过py3的异步实战,我觉得py2的异步实战很臃肿,
什么基于gevent等等啊,其实我想强调一点,在效率上py3是可以吊打py2的,希望可以有更多的人用py3尽早淘汰py2,
就我这个13秒643万字(我不知道是不是我的上限)可以吊打百分之九十的所谓py2的异步编程了
最近听到了python发展的一个好消息,facebook的所有代码已经一半以上是py3了,
其实我可以说没有用过py2编程的,我也不会考虑,让py2早点被淘汰吧,我这么感慨其实不得已,我找了一天的资料,没有找到py3异步编程的实战
都是搞什么测试,不过还是谢谢他们'''
python3异步实现高并发爬取笔趣阁小说643万字平均26秒直接下载,第二次测试只花了不足14秒
猜你喜欢
转载自blog.csdn.net/qq_42395490/article/details/81841162
今日推荐
周排行