python3網絡爬蟲aiohttp實戰案例

提示：文章寫完後，目錄可以自動生成，如何生成可參考右邊的幫助文檔

文章目錄

• 前言
• 一、觀察需要爬取的頁面的資料包
• 二、基本配置
• 三、生成不同頁面的url，并擷取響應資料
• 四、獲得評論資訊和資料儲存
• 五、主函數調用
• 六、運作結果

前言

本文介紹了使用aiohttp庫實作異步爬蟲爬取網站圖書的評論，及其代碼的詳細解釋（案例來源于python3網絡爬蟲開發實戰，本人對其進行了改編）

一、觀察需要爬取的頁面的資料包

1、通過觀察發現頁面的資料包是ajax資料包，而且我們需要的組成詳情頁的url的id也在ajax資料包的響應當中

2、進入詳情頁中，頁面的資料還是一個ajax資料包，我們想要的評論資料也在ajax資料包的響應當中。

這裡注意一個問題：詳情頁的url和ajax資料包的url是不一樣的

二、基本配置

1、引入需要使用的庫

import aiohttp
import asyncio
import nest_asyncio
import time
import json
import logging##這個是日志檔案的庫，可以記錄程式運作的過程
           

2、定義必要的資訊

nest_asyncio.apply()##jupyter notebook上不加此句話會報錯
logging.basicConfig(level=logging.INFO,format='%(asctime)s-%(levelname)s:%(message)s')
index_url='https://spa5.scrape.center/api/book/?limit=18&offset={offset}'#索引頁的url
detail_url= 'https://spa5.scrape.center/api/book/{id}'#詳情頁的url
page_size=18#一頁有18本圖書
page_number=2#設定爬取的頁數是2頁
concurrency=5#設定最大的并大數量是5
           

``

三、生成不同頁面的url，并擷取響應資料

1、抓取ajax資料包中的響應資料

async def scrape_api(url):
    async with semaphore:
        try:
            logging.info('scraping %s',url)
            async with session.get(url) as response:
                return await response.json()
        except aiohttp.ClientError:
            logging.error('error occured while scraping %s',url,exc_info=True)
           

2、由于需要爬取不同頁面的圖書資訊，是以url在參數上會有不同，不同頁面的url也不同

##得到不同頁面的url
async def scrape_index(page):
    url=index_url.format(offset=page_size*(page-1))##拼接成完整的詳情頁的url
    return await scrape_api(url)
           

四、獲得評論資訊和資料儲存

#處理擷取到的響應資料
async def save_file(data):
    list_content=data.get('comments')
    for item in list_content:
         with open('comment.txt',mode='a',encoding='utf-8') as fp:
            fp.write(item.get('content'))
            fp.write('\r\n')
#擷取詳情頁的評論資訊   
async def scrape_detail(id):
    url=detail_url.format(id=id)
    data=await scrape_api(url)
    await save_file(data)
           

五、主函數調用

async def main():
    global session
    session=aiohttp.ClientSession()#建立一個ClientSession對象，用于獲得網頁響應，類似于使用get
    scrape_index_tasks=[asyncio.ensure_future(scrape_index(page)) for page in range(1,page_number+1)]#擷取爬取頁面的任務
    results=await asyncio.gather(*scrape_index_tasks)##将任務進行聚合，相當于放到一個隊列中
    logging.info('results %s',json.dumps(results,ensure_ascii=False,indent=2))
    ids=[]
    for index_data in results:
        if not index_data:
            continue
        for item in index_data.get('results'):#擷取資料包的響應資料是一個字典，字典的值是一個清單
            ids.append(item.get('id'))#清單中又是一個字典，找到id
    scrape_detail_tasks=[asyncio.ensure_future(scrape_detail(id)) for id in ids]##建立擷取詳情頁的任務
    await asyncio.wait(scrape_detail_tasks)##這裡是io操作比較密集的地方，需要挂起
    await session.close()
    
if __name__=='__main__':
    asyncio.run(main())
           

六、運作結果