引言:
传统上,Scrapy作为Python中的一款强大爬虫框架,因其便捷而得到广泛应用,尤其是在小到中型项目中的效率与方便性上无可匹敌。但在面对需要更加个性化的复杂项目时,Scrapy的固有结构可能不够灵活。此外,爬虫本质上不过是一个涉及“获取数据、抓取内容、解析结果、执行逻辑以及数据存储”五个基本步骤的过程。如果你能够清晰识别并构建这五大块,其余的工作——如调度逻辑、网络请求,甚至是搭建分布式系统——都可通过模块化来高效执行。
第一章 抓取:
在许多场景下,抓取和解析的过程往往被合并在一起。如果面临的是一个数据量相对较小(100万条以内)的项目,这种做法可能并无大碍。然而,一旦涉及较大规模的数据处理,将两者分离将会显著提升系统的灵活性和效率。相对独立的模块化也使得后期维护、调度优化和问题修复变得更加简便。
在这里,我将分享一个异步抓取框架的构建经验,此框架特色在于:**异步处理**的高效性和**异常日志输出**的便利性。
另外,此代码可操作的空间:请求头/cookie/ip等,自行补充;(如果你把解析写到一块,很容易导致错误混淆,不易区分)
框架核心:
我设计的框架利用`asyncio`和`aiohttp`库进行高效的异步网页请求,配合`logging`模块打印详细的异常信息。如此一来,不仅大幅提升了数据抓取的速度,而且在出现错误时也能够轻松地定位问题,并有针对性地进行数据补充。
以下是框架的一个核心组件`SecondGet`,它负责异步请求网页并返回HTML内容。具体功能如下:
import logging
import asyncio
import aiohttp
class SecondGet():
'''
传入一个id, 分别对X个url进行异步抓取html(不执行解析,只返回html)。
'''
def __init__(self, id):
# 初始化日志
self.logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)
# 存储id和各种请求需要的头信息
self.id = id
self.headers = {
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36',
}
# 需要爬取的urls
self.urls = {
'XX': f'http://example2_{id}.html',
'XX': f'http://example3_{id}.html',
'XX': f'http://example4_{id}.html',
'XX': f'http://example5_{id}.html',
......
}
# 异步请求URL的函数
async def fetch_url(self, session, url):
try:
# 发起网络请求
async with session.get(url, timeout=3, headers=self.headers) as response:
# 返回响应的html文本
return await response.text()
except aiohttp.ClientError as e:
# 如果请求出错,记录异常信息到日志
self.logger.exception(f"爬取相关url出错: {url}")
# 创建异步任务并处理结果的函数
async def run(self):
# 存储结果的字典
results = {}
# 创建网络会话
async with aiohttp.ClientSession() as session:
tasks = [] # 存储所有任务
# 为每个URL创建一个任务
for name, url in self.urls.items():
task = asyncio.create_task(self.fetch_url(session, url))
tasks.append(task)
# 等待所有任务完成
responses = await asyncio.gather(*tasks, return_exceptions=True)
# 处理每个任务的结果
for name, response in zip(self.urls.keys(), responses):
if isinstance(response, Exception):
# 如果结果是一个异常就记录到文件中
with open('爬取异常.txt', 'a') as error_file:
error_file.write(f"{name} URL: {self.urls[name]}\n")
results[name] = None # 异常对应结果为None
else:
# 如果结果正常就存储到结果字典
results[name] = response
# 返回结果字典
return results
# 异步执行的主函数
async def main(id):
# 实例化SecondGet
second_get = SecondGet(id)
# 爬取数据并等待结果
results = await second_get.run()
# 返回抓取结果
return results
# 代码执行的入口
if __name__ == "__main__":
# 示例ID,可以修改为具体的目标ID
example_id = 'XXX'
# 运行主函数并获取最终结果
final_results = asyncio.run(main(example_id))
# 打印结果字典
print(final_results)