一個站點不夠學？那就在用Python增加一個采集目標，一派話題廣場+某金融論壇話題廣場爬蟲

本次的目標站點原計劃是一個比較簡單的站點，后來發現有點太簡單了，就額外增加了一個案例，學一個贈一個，本篇博客核心用到的技術依舊是隊列 queue 技術。

目標站點【一派話題廣場】分析

本篇博客的第一個采集目標站點是：https://sspai.com/matrix/pods，少數派網站的一個子級欄目。

目標數據所在界面如下圖所示：

通過開發者工具，不斷下拉加載頁面，得到的接口請求規則如下：

https://sspai.com/api/v1/bullet/search/page/get?type=0&limit=10&offset=0&created_at=0 https://sspai.com/api/v1/bullet/search/page/get?type=0&limit=10&offset=10&created_at=0 https://sspai.com/api/v1/bullet/search/page/get?type=0&limit=10&offset=20&created_at=0 https://sspai.com/api/v1/bullet/search/page/get?type=0&limit=10&offset=30&created_at=0

其中參數除 offset 變化外，其余無變化，其中 limit 參數應該為每個數據量，基于此邏輯，請求接口可以通過代碼進行批量生成，實測過程發現數據量也不大，只有 6 頁。

下述代碼采用了后進先出隊列 LifoQueue，沒有特殊原因，單純給大家展示一下用法。

# 初始化一個隊列 q = LifoQueue(maxsize=0) # 批量生成請求地址鏈接 for page in range(1, 7): # https://sspai.com/api/v1/bullet/search/page/get?type=0&limit=10&offset=0&created_at=0 q.put('https://sspai.com/api/v1/bullet/search/page/get?type=0&limit=10&offset={}&created_at=0'.format((page-1)*10))

請求地址批量生成完畢之后，可以開始獲取接口返回的數據了，具體代碼如下，核心關注 main 部分內容。

# 請求頭函數，可以參考之前的系列文章 def get_headers(): uas = [ "Mozilla/5.0 (compatible; Baiduspider/2.0; +http://www.baidu.com/search/spider.html)", "Mozilla/5.0 (compatible; Baiduspider-render/2.0; +http://www.baidu.com/search/spider.html)" ] ua = random.choice(uas) headers = { "user-agent": ua } return headers # 儲存數據 def save(text): # 文件名使用時間戳命名 with open(f'{time.time()}.json', 'a+', encoding='utf-8') as f: f.write(text) print(text, "--- 保存成功") if __name__ == "__main__": # 判斷隊列是否為空，空則停止循環 while q.qsize() > 0: # 獲取一個 URL url = q.get() # 通知任務已經完成 q.task_done() # 獲取相應數據 res = requests.get(url=url, headers=get_headers(), timeout=10) # 調用保存函數 save(res.text) q.join() print("所有任務都已完成")

因為上述案例實在太簡單，連多線程都不用，所以我基于【話題廣場】關鍵字，查詢是否還有其它可用于 學習目的 相關站點，結果還真被我發現一個。

話題廣場 - 集思錄

集思錄，一個以數據為本的投資社區，https://www.jisilu.cn/topic/。

每一個話題下面，都有很多問題，完美符合生產者消費者模型。

編碼邏輯分析

由于在列表頁之前，還存在一個層級-【熱門話題】，所以需要提前準備好待抓取隊列。

# 熱門話題列表頁待抓取鏈接 hot_subjects = Queue(maxsize=0) for i in range(1, 11): url = f'https://www.jisilu.cn/topic/square/id-hot__feature_id-__page-{i}' hot_subjects.put(url)

接下來生產函數用于產生列表頁數據，到提前初始化好的 q_data_ids 隊列中，其中 get_headers 函數，參考前文即可。

def get_headers(): uas = [ "Mozilla/5.0 (compatible; Baiduspider/2.0; +http://www.baidu.com/search/spider.html)", "Mozilla/5.0 (compatible; Baiduspider-render/2.0; +http://www.baidu.com/search/spider.html)" ] ua = random.choice(uas) headers = { "user-agent": ua } return headers # 初始化一個隊列 q_data_ids = Queue(maxsize=0) # 生產函數，用于產生話題列表頁 URL def producer(): while hot_subjects.qsize() > 0: # 取得一個分類頁請求地址 list_url = hot_subjects.get() hot_subjects.task_done() print("正在解析：", list_url) # 獲取話題列表頁相應數據 res = requests.get(list_url, headers=get_headers(), timeout=3) # 解析，獲取話題詳情頁進入的關鍵參數 `data-id`，代碼后有對于該關鍵字的說明 element = etree.HTML(res.text) data_ids = element.xpath('//a[@class="aw-topic-name"]/@data-id') for data_id in data_ids: q_data_ids.put(data_id)

上述代碼最重要的是捕獲 data-id，即熱門話題的 ID 隊列，該 ID 值會用在消費者函數中，用于請求詳情頁數據。

接下來就是消費者函數的實現，該邏輯主要用于抓取下圖所示數據。

以上數據在測試過程中發現如下接口格式，其中 data_id 在生產者中產生，start_page 由循環迭代生成。

https://www.jisilu.cn/question/ajax/discuss/sort_type-new__topic_id-{data_id}__page-{start_page}

上述地址即為最后目標數據所在地址，我們需要拼湊出標準地址，然后再對 start_page 頁碼，進行迭代，遍歷獲取全部數據。

學習時請注意下述代碼注釋，技術層級難度不大，重點為實現邏輯。

# 消費者函數 def consumer(): # 死循環讀取 data_id 值，用于拼湊話題詳情頁數據 while True: # 取一個分類ID data_id = q_data_ids.get() q_data_ids.task_done() if data_id is None: break # start_page 初始值設置為 1，即從第一個開始讀取數據 start_page = 1 # URL 拼接 url = f'https://www.jisilu.cn/question/ajax/discuss/sort_type-new__topic_id-{data_id}__page-{start_page}' res = requests.get(url=url, headers=get_headers(), timeout=5) text = res.text # 通過判斷 text 是否為空，確定是否繼續解析數據 while len(text) > 0: url = f'https://www.jisilu.cn/question/ajax/discuss/sort_type-new__topic_id-{data_id}__page-{start_page}' res = requests.get(url=url, headers=get_headers(), timeout=5) # print(res.url) text = res.text start_page += 1 # 如果 text 不為空，則解析數據，并存儲數據 if len(text)>0: element = etree.HTML(res.text) titles = element.xpath('//h4/a/text()') urls = element.xpath('//h4/a/@href') names = element.xpath('//a[@class="aw-user-name"]/text()') data = zip(titles,names,urls) save_list = [f"{item[0]},{item[1]},{item[2]}\n" for item in data] long_str = "".join(save_list) with open("./data.csv","a+",encoding="utf-8") as f: f.write(long_str)

該程序實現的步驟與邏輯相對會繞一些，故通過下圖你可以再復盤進行理解，按照步驟 1,2,3 進行學習。

函數運行可是采用多線程實現，具體如下：

# 開啟2個生產者線程 for p_in in range(1, 3): p = threading.Thread(target=producer) p.start() # 開啟2個消費者線程 for p_in in range(1, 2): p = threading.Thread(target=consumer) p.start()

Python 金融專區

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。