江苏龙网

  1. 首页 > 生活 > >

怎样用Python 写一个爬虫框架( 二 )



选择刚刚pipenv配置好的python解释器：
怎样用Python 写一个爬虫框架

此时可以完整地看到项目代码：
怎样用Python 写一个爬虫框架

好，环境以及源码准备完毕，接下来将结合代码讲述一个爬虫框架的编写流程
Request \u0026amp; Response Request类的目的是对aiohttp加一层封装进行模拟请求，功能如下：
封装GET、POST两种请求方式增加回调机制自定义重试次数、休眠时间、超时、重试解决方案、请求是否成功验证等功能将返回的一系列数据封装成Response类返回接下来就简单了，不过就是实现上述需求，首先，需要实现一个函数来抓取目标url，比如命名为fetch:
import asyncioimport aiohttpimport async_timeoutfrom typing import Coroutineclass Request: # Default config REQUEST_CONFIG = { \u0026#39;RETRIES\u0026#39;: 3, \u0026#39;DELAY\u0026#39;: 0, \u0026#39;TIMEOUT\u0026#39;: 10, \u0026#39;RETRY_FUNC\u0026#39;: Coroutine, \u0026#39;VALID\u0026#39;: Coroutine } METHOD = def __init__(self, url, method=\u0026#39;GET\u0026#39;, request_config=None, request_session=None): self.url = url self.method = method.upper() self.request_config = request_config or self.REQUEST_CONFIG self.request_session = request_session @property def current_request_session(self): if self.request_session is None: self.request_session = aiohttp.ClientSession() self.close_request_session = True return self.request_session async def fetch(self): """Fetch all the information by using aiohttp""" if self.request_config.get(\u0026#39;DELAY\u0026#39;, 0) \u0026gt; 0: await asyncio.sleep(self.request_config) timeout = self.request_config.get(\u0026#39;TIMEOUT\u0026#39;, 10) async with async_timeout.timeout(timeout): resp = await self._make_request() try: resp_data = https://www.zhihu.com/api/v4/questions/28548989/await resp.text() except UnicodeDecodeError: resp_data = await resp.read() resp_dict = dict( rl=self.url, method=self.method, encoding=resp.get_encoding(), html=resp_data, cookies=resp.cookies, headers=resp.headers, status=resp.status, history=resp.history ) await self.request_session.close() return type(/u0026#39;Response/u0026#39;, (), resp_dict) async def _make_request(self): if self.method == /u0026#39;GET/u0026#39;: request_func = self.current_request_session.get(self.url) else: request_func = self.current_request_session.post(self.url) resp = await request_func return respif __name__ == /u0026#39;__main__/u0026#39;: loop = asyncio.get_event_loop() resp = loop.run_until_complete(Request(/u0026#39;https://docs.python-ruia.org/u0026#39;).fetch()) print(resp.status)


推荐阅读


上一篇：怎样评价小米智能台灯的海报

下一篇：程序里咋理解包，模块，类，函数和参数的关系，如果理解成一层层的调用或者传递是咋实现的求形象理解