Python | 爬蟲框架 feapder 初體驗
嘿，大家好，我是咸魚，之前，我們寫爬蟲，用的最多的框架莫過于scrapy啦，今天我們用最近新出的爬蟲框架feapder來開發(fā)爬蟲，看下是怎樣的體驗。
目標網(wǎng)站：aHR0cHM6Ly93d3cubGFnb3UuY29tLw==
需求：采集職位列表與職位詳情，詳情需每7天更新一次
為了演示，以下只搜索與爬蟲相關(guān)的職位
1. 調(diào)研
1.1 列表頁面
首先我們需要看下頁面是否為動態(tài)渲染的，接口是否有反爬。
看是否為動態(tài)渲染的可以右鍵，顯示網(wǎng)頁源代碼，然后搜索網(wǎng)頁上的內(nèi)容源碼里是否存在，比如搜索列表的第一條知衣科技，匹配了0條，則初步判斷是動態(tài)渲染的
或者可以用feapder命令，下載網(wǎng)頁源碼，查看。
打開后的頁面為加載中
調(diào)用response.open()命令會在工作目錄下生產(chǎn)一個temp.html文件，內(nèi)容為當前請求返回的源碼，我們點擊查看，是一段js，有安全驗證。因此可以推斷出該網(wǎng)站有反爬，難度升級預警
feapder還支持使用curl命令請求，方式如下：
按F12，或者右鍵檢查，打開調(diào)試窗口，刷新頁面，點擊當前頁的請求，復制為curl，返回命令行窗口，輸入 feapder shell --然后粘貼剛剛復制的內(nèi)容
發(fā)現(xiàn)攜帶header，cookie也不行，可能是某些參數(shù)只能用一次吧。
調(diào)研結(jié)論：列表頁有反爬，頁面動態(tài)渲染
ps: 正常大神還會繼續(xù)調(diào)研，列表接口是什么，如何破解反爬，但因為我是小白，就先不糾結(jié)了
1.2 詳情頁面
與列表頁調(diào)研類似，結(jié)論是有反爬，但頁面不是動態(tài)渲染的
2. 創(chuàng)建項目
打開命令行工具，輸入：
> feapder create -p lagou-spider
lagou-spider 項目生成成功
生成項目如下：
我用的pycharm，先右鍵，將這個項目加入到工作區(qū)間。
（右鍵項目名，Mark Directory as -> Sources Root）
3. 寫列表頁爬蟲
3.1 創(chuàng)建爬蟲
> cd lagou-spider/spiders
> feapder create -s list_spider
ListSpider 生成成功
生成代碼如下：
import feapder
class ListSpider(feapder.AirSpider):
def start_requests(self):
yield feapder.Request("https://www.baidu.com")
def parse(self, request, response):
print(response)
if __name__ == "__main__":
ListSpider().start()
這是請求百度的例子，可直接運行
3.2 寫爬蟲
下發(fā)任務(wù)：
def start_requests(self):
yield feapder.Request("https://www.lagou.com/jobs/list_%E7%88%AC%E8%99%AB?labelWords=&fromSearch=true&suginput=", render=True)
注意到，我們在請求里攜帶了render參數(shù)，表示是否用瀏覽器渲染，因為這個列表頁是動態(tài)渲染的，又有反爬，我比較慫，所以采用了渲染模式，以避免掉頭發(fā)
編寫解析函數(shù)
觀察頁面結(jié)構(gòu)，寫出如下解析函數(shù)
def parse(self, request, response):
job_list = response.xpath("http://li[contains(@class, 'con_list_item')]")
for job in job_list:
job_name = job.xpath("./@data-positionname").extract_first()
company = job.xpath("./@data-company").extract_first()
salary = job.xpath("./@data-salary").extract_first()
job_url = job.xpath(".//a[@class='position_link']/@href").extract_first()
print(job_name, company, salary, job_url)
我們解析了職位名稱、公司、薪資、以及職位詳情地址，正常邏輯應(yīng)該將詳情地址作為任務(wù)下發(fā)，獲取詳情
def parse(self, request, response):
job_list = response.xpath("http://li[contains(@class, 'con_list_item')]")
for job in job_list:
job_name = job.xpath("./@data-positionname").extract_first()
company = job.xpath("./@data-company").extract_first()
salary = job.xpath("./@data-salary").extract_first()
job_url = job.xpath(".//a[@class='position_link']/@href").extract_first()
print(job_name, company, salary, job_url)
yield feapder.Request(
job_url, callback=self.parse_detail, cookies=response.cookies.get_dict()
) # 攜帶列表頁返回的cookie，回調(diào)函數(shù)指向詳情解析函數(shù)
def parse_detail(self, request, response):
print(response.text)
# TODO 解析詳情
但需求是詳情每7天更新一次，列表沒說要更新，因此為了優(yōu)化，將詳情單獨寫個爬蟲，本爬蟲只負責列表的數(shù)據(jù)和生產(chǎn)詳情的任務(wù)就好了
3.3 數(shù)據(jù)入庫
創(chuàng)建表
職位列表數(shù)據(jù)表 lagou_job_list
CREATE TABLE `lagou_job_list` (
`id` int(10) unsigned NOT NULL AUTO_INCREMENT COMMENT '自增id',
`job_name` varchar(255) DEFAULT NULL COMMENT '職位名稱',
`company` varchar(255) DEFAULT NULL COMMENT '公司',
`salary` varchar(255) DEFAULT NULL COMMENT '薪資',
`job_url` varchar(255) DEFAULT NULL COMMENT '職位地址',
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
詳情任務(wù)表 lagou_job_detail_task
CREATE TABLE `lagou_job_detail_task` (
`id` int(11) unsigned NOT NULL AUTO_INCREMENT,
`url` varchar(255) DEFAULT NULL,
`state` int(11) DEFAULT '0' COMMENT '任務(wù)狀態(tài)（0未做，1完成，2正在做，-1失?。?,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
數(shù)據(jù)入庫方式
數(shù)據(jù)入庫有很多方式，直接導入pymysql然后拼接sql語句入庫，或者使用框架自帶的MysqlDB。不過feapder有一種更方便的入庫方式，自動入庫
自動入庫AirSpider是不支持的，因為他比較輕量嘛，作者為了保持輕量的特性，暫時沒支持自動入庫。不過分布式爬蟲Spider是支持的，我們直接將繼承類改為Spider即可
class ListSpider(feapder.AirSpider):
改為
class ListSpider(feapder.Spider):
生成item
item是與表一一對應(yīng)的，與數(shù)據(jù)入庫機制有關(guān)，可用feapder命令生成。
首先配置下數(shù)據(jù)庫連接信息，在setting中配置的
生成item：
> cd items
> feapder create -i lagou_job_list
> feapder create -i lagou_job_detail_task
數(shù)據(jù)入庫
def parse(self, request, response):
job_list = response.xpath("http://li[contains(@class, 'con_list_item')]")
for job in job_list:
job_name = job.xpath("./@data-positionname").extract_first()
company = job.xpath("./@data-company").extract_first()
salary = job.xpath("./@data-salary").extract_first()
job_url = job.xpath(".//a[@class='position_link']/@href").extract_first()
# 列表數(shù)據(jù)
list_item = lagou_job_list_item.LagouJobListItem()
list_item.job_name = job_name
list_item.company = company
list_item.salary = salary
list_item.job_url = job_url
yield list_item # 直接返回，框架實現(xiàn)批量入庫
# 詳情任務(wù)
detail_task_item = lagou_job_detail_task_item.LagouJobDetailTaskItem()
detail_task_item.url = job_url
yield detail_task_item # 直接返回，框架實現(xiàn)批量入庫
以yield item的方式將數(shù)據(jù)返回給框架，框架自動批量入庫
3.4 整體代碼
import feapder
from items import *
class ListSpider(feapder.Spider):
def start_requests(self):
yield feapder.Request(
"https://www.lagou.com/jobs/list_%E7%88%AC%E8%99%AB?labelWords=&fromSearch=true&suginput=",
render=True,
)
def parse(self, request, response):
job_list = response.xpath("http://li[contains(@class, 'con_list_item')]")
for job in job_list:
job_name = job.xpath("./@data-positionname").extract_first()
company = job.xpath("./@data-company").extract_first()
salary = job.xpath("./@data-salary").extract_first()
job_url = job.xpath(".//a[@class='position_link']/@href").extract_first()
# 列表數(shù)據(jù)
list_item = lagou_job_list_item.LagouJobListItem()
list_item.job_name = job_name
list_item.company = company
list_item.salary = salary
list_item.job_url = job_url
yield list_item # 直接返回，框架實現(xiàn)批量入庫
# 詳情任務(wù)
detail_task_item = lagou_job_detail_task_item.LagouJobDetailTaskItem()
detail_task_item.url = job_url
yield detail_task_item # 直接返回，框架實現(xiàn)批量入庫
if __name__ == "__main__":
spider = ListSpider(redis_key="feapder:lagou_list")
spider.start()
redis_key為任務(wù)隊列在redis中存放的位置。
直接運行，觀察到數(shù)據(jù)已經(jīng)自動入庫了
4. 寫詳情爬蟲
與列表頁爬蟲不同，詳情數(shù)據(jù)需要每7天更新一次。
為了做時序數(shù)據(jù)展示，我們每7天采集一次數(shù)據(jù)，數(shù)據(jù)需攜帶批次信息，將數(shù)據(jù)按照7天維度劃分
在沒接觸feapder框架前，我們需要考慮將任務(wù)從詳情任務(wù)表中分批拿出發(fā)給爬蟲，還需要維護任務(wù)的狀態(tài)，以及上面提及的批次信息。并且為了保證數(shù)據(jù)的時效性，需要對采集進度進行監(jiān)控，寫個爬蟲十分繁瑣。
那么feapder如何做呢？為了節(jié)省篇幅，直接給出完整代碼：
import feapder
from items import *
class DetailSpider(feapder.BatchSpider):
def start_requests(self, task):
task_id, url = task
yield feapder.Request(url, task_id=task_id, render=True)
def parse(self, request, response):
job_name = response.xpath('//div[@class="job-name"]/@title').extract_first().strip()
detail = response.xpath('string(//div[@class="job-detail"])').extract_first().strip()
item = lagou_job_detail_item.LagouJobDetailItem()
item.title = job_name
item.detail = detail
item.batch_date = self.batch_date # 獲取批次信息，批次信息框架自己維護
yield item # 自動批量入庫
yield self.update_task_batch(request.task_id, 1) # 更新任務(wù)狀態(tài)
if __name__ == "__main__":
spider = DetailSpider(
redis_key="feapder:lagou_detail", # redis中存放任務(wù)等信息的根key
task_table="lagou_job_detail_task", # mysql中的任務(wù)表
task_keys=["id", "url"], # 需要獲取任務(wù)表里的字段名，可添加多個
task_state="state", # mysql中任務(wù)狀態(tài)字段
batch_record_table="lagou_detail_batch_record", # mysql中的批次記錄表
batch_name="詳情爬蟲(周全)", # 批次名字
batch_interval=7, # 批次周期 天為單位 若為小時 可寫 1 / 24
)
# 下面兩個啟動函數(shù) 相當于 master、worker。需要分開運行
# spider.start_monitor_task() # 下發(fā)及監(jiān)控任務(wù)
spider.start() # 采集
我們分別運行spider.start_monitor_task()與spider.start()，待爬蟲結(jié)束后，觀察數(shù)據(jù)庫
任務(wù)表：lagou_job_detail_task
任務(wù)均已完成了，框架有任務(wù)丟失重發(fā)機制，直到所有任務(wù)均已做完
數(shù)據(jù)表：lagou_job_detail:
數(shù)據(jù)里攜帶了批次時間信息，我們可以根據(jù)這個時間來對數(shù)據(jù)進行劃分。當前批次為3月19號，若7天一批次，則下一批次為3月26號。
在本批次期間重復啟動爬蟲，若無新任務(wù)，爬蟲不會抓取spider.start_monitor_task()
spider.start()
批次表：lagou_detail_batch_record
批次表為啟動參數(shù)中指定的，自動生成。批次表里詳細記錄了每個批次的抓取狀態(tài)，如任務(wù)總量、已做量、失敗量、是否已完成等信息
5. 整合
目前列表爬蟲與詳情爬蟲都寫完了，運行入口分布在兩個文件里，管理起來比較亂，feapder建議寫到統(tǒng)一寫到main.py里
from feapder import ArgumentParser
from spiders import *
def crawl_list():
"""
列表爬蟲
"""
spider = list_spider.ListSpider(redis_key="feapder:lagou_list")
spider.start()
def crawl_detail(args):
"""
詳情爬蟲
@param args: 1 / 2 / init
"""
spider = detail_spider.DetailSpider(
redis_key="feapder:lagou_detail", # redis中存放任務(wù)等信息的根key
task_table="lagou_job_detail_task", # mysql中的任務(wù)表
task_keys=["id", "url"], # 需要獲取任務(wù)表里的字段名，可添加多個
task_state="state", # mysql中任務(wù)狀態(tài)字段
batch_record_table="lagou_detail_batch_record", # mysql中的批次記錄表
batch_name="詳情爬蟲(周全)", # 批次名字
batch_interval=7, # 批次周期 天為單位 若為小時 可寫 1 / 24
)
if args == 1:
spider.start_monitor_task()
elif args == 2:
spider.start()
if __name__ == "__main__":
parser = ArgumentParser(description="xxx爬蟲")
parser.add_argument(
"--crawl_list", action="store_true", help="列表爬蟲", function=crawl_list
)
parser.add_argument(
"--crawl_detail", type=int, nargs=1, help="詳情爬蟲(1|2）", function=crawl_detail
)
parser.start()
查看啟動命令：
> python3 main.py --help
usage: main.py [-h] [--crawl_list] [--crawl_detail CRAWL_DETAIL]
xxx爬蟲
optional arguments:
-h, --help show this help message and exit
--crawl_list 列表爬蟲
--crawl_detail CRAWL_DETAIL
詳情爬蟲(1|2）
啟動列表爬蟲：
python3 main.py --crawl_list
啟動詳情爬蟲master
python3 main.py --crawl_detail 1
啟動詳情爬蟲worker
python3 main.py --crawl_detail 2
總結(jié)
本文拿某招聘網(wǎng)站舉例，介紹了使用feapder采集數(shù)據(jù)整個過程。其中涉及到AirSpider、Spider、BatchSpider三種爬蟲的使用。
AirSpider爬蟲比較輕量，學習成本低。面對一些數(shù)據(jù)量較少，無需斷點續(xù)爬，無需分布式采集的需求，可采用此爬蟲。
Spider是一款基于redis的分布式爬蟲，適用于海量數(shù)據(jù)采集，支持斷點續(xù)爬、爬蟲報警、數(shù)據(jù)自動入庫等功能
BatchSpider是一款分布式批次爬蟲，對于需要周期性采集的數(shù)據(jù)，優(yōu)先考慮使用本爬蟲。
feapder除了支持瀏覽器渲染下載外，還支持pipeline，用戶可自定義，方便對接其他數(shù)據(jù)庫
框架內(nèi)置豐富的報警，爬蟲有問題時及時通知到我們，以保證數(shù)據(jù)的時效性
實時計算爬蟲抓取速度，估算剩余時間，在指定的抓取周期內(nèi)預判是否會超時
爬蟲卡死報警
爬蟲任務(wù)失敗數(shù)過多報警，可能是由于網(wǎng)站模板改動或封堵導致
下載情況監(jiān)控
關(guān)于feapder使用說明，詳見官方文檔：https://boris.org.cn/feapder/#/
本文項目地址可在后臺回復【feapder】獲取