閱讀文本大概需要 10 分鐘。
01 抓取目標(biāo)
場(chǎng)景：有時(shí)候我們想爬取某個(gè)大 V 的發(fā)布的全部的文章進(jìn)行學(xué)習(xí)或者分析。
這個(gè)爬蟲任務(wù)我們需要借助「 Charles 」這個(gè)抓包工具，設(shè)置好手機(jī)代理 IP 去請(qǐng)求某個(gè)頁面，通過分析，模擬請(qǐng)求，獲取到實(shí)際的數(shù)據(jù)。
我們要爬取文章的作者、文章標(biāo)題、封面圖、推送時(shí)間、文件內(nèi)容、閱讀量、點(diǎn)贊數(shù)、評(píng)論數(shù)、文章實(shí)際鏈接等數(shù)據(jù)，最后要把數(shù)據(jù)存儲(chǔ)到「 MongoDB 」數(shù)據(jù)庫中。
02 準(zhǔn)備工作
首先，在 PC 上下載 Charles，并獲取本地的 IP 地址。
然后，手機(jī)連上同一個(gè)網(wǎng)段，并手動(dòng)設(shè)置代理 IP，端口號(hào)默認(rèn)填 8888 。最后配置 PC 和手機(jī)上的證書及 SSL Proxying，保證能順利地抓到 HTTPS 的請(qǐng)求。具體的方法可以參考下面的文章。
「http://www.itdecent.cn/p/595e8b556a60?from=timeline&isappinstalled=0 」
03 爬取思路
首先我們選中一個(gè)微信公眾號(hào)，依次點(diǎn)擊右上角的頭像、歷史消息，就可以進(jìn)入到全部消息的主界面。默認(rèn)展示的是前 10 天歷史消息。
然后可以查看 Charles 抓取的請(qǐng)求數(shù)據(jù)，可以通過「 mp.weixin.qq.com 」去過濾請(qǐng)求，獲取到消息首頁發(fā)送的請(qǐng)求及請(qǐng)求方式及響應(yīng)內(nèi)容。
繼續(xù)往下滾動(dòng)頁面，可以加載到下一頁的數(shù)據(jù)，同樣可以獲取到請(qǐng)求和響應(yīng)的數(shù)據(jù)。
爬取的數(shù)據(jù)最后要保存在 MongoDB 文檔型數(shù)據(jù)庫中，所以不需要建立數(shù)據(jù)模型，只需要安裝軟件和開啟服務(wù)就可以了。MongoDB 的使用教程可以參考下面的鏈接：
「 http://www.itdecent.cn/p/4c5deb1b7e7c 」
為了操作 MongoDB 數(shù)據(jù)庫，這里使用「 MongoEngine 」這個(gè)類似于關(guān)系型數(shù)據(jù)庫中的 ORM 框架來方便我們處理數(shù)據(jù)。
pip3 install mongoengine
04 代碼實(shí)現(xiàn)
從上面的分析中可以知道首頁消息、更多頁面消息的請(qǐng)求 URL 規(guī)律如下：
# 由于微信屏蔽的關(guān)鍵字， 字段 netloc + path 用 ** 代替# 首頁請(qǐng)求urlhttps://**?action=home&__biz=MzIxNzYxMTU0OQ==&scene=126&bizpsid=0&sessionid=1545633855&subscene=0&devicetype=iOS12.1.2&version=17000027&lang=zh_CN&nettype=WIFI&a8scene=0&fontScale=100&pass_ticket=U30O32QRMK6dba2iJ3ls6A3PRbrhksX%2B7D8pF3%2Bu3uXSKvSAa1hnHzfsSClawjKg&wx_header=1# 第二頁請(qǐng)求urlhttps://**?action=getmsg&__biz=MzIxNzYxMTU0OQ==&f=json&offset=10&count=10&is_ok=1&scene=126&uin=777&key=777&pass_ticket=U30O32QRMK6dba2iJ3ls6A3PRbrhksX%2B7D8pF3%2Bu3uXSKvSAa1hnHzfsSClawjKg&wxtoken=&appmsg_token=988_rETfljlGIZqE%252F6MobN1rEtqBx5Ai9wBDbbH_sw~~&x5=0&f=json# 第三頁請(qǐng)求urlhttps://**?action=getmsg&__biz=MzIxNzYxMTU0OQ==&f=json&offset=21&count=10&is_ok=1&scene=126&uin=777&key=777&pass_ticket=U30O32QRMK6dba2iJ3ls6A3PRbrhksX%2B7D8pF3%2Bu3uXSKvSAa1hnHzfsSClawjKg&wxtoken=&appmsg_token=988_rETfljlGIZqE%252F6MobN1rEtqBx5Ai9wBDbbH_sw~~&x5=0&f=json
可以通過把 offset 設(shè)置為可變數(shù)據(jù)，請(qǐng)求所有頁面的數(shù)據(jù) URL可以寫成下面的方式：
https://**?action=getmsg&__biz=MzIxNzYxMTU0OQ==&f=json&offset={}&count=10&is_ok=1&scene=126&uin=777&key=777&pass_ticket=U30O32QRMK6dba2iJ3ls6A3PRbrhksX%2B7D8pF3%2Bu3uXSKvSAa1hnHzfsSClawjKg&wxtoken=&appmsg_token=988_rETfljlGIZqE%252F6MobN1rEtqBx5Ai9wBDbbH_sw~~&x5=0&f=json
另外，通過 Charles 獲取到請(qǐng)求頭。由于微信的反爬機(jī)制，這里的 Cookie 和 Referer 有一定的時(shí)效性，需要定時(shí)更換。
self.headers = { 'Host': 'mp.weixin.qq.com', 'Connection': 'keep-alive', 'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 12_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/16B92 MicroMessenger/6.7.4(0x1607042c) NetType/WIFI Language/zh_CN', 'Accept-Language': 'zh-cn', 'X-Requested-With': 'XMLHttpRequest', 'Cookie': 'devicetype=iOS12.1; lang=zh_CN; pass_ticket=fXbGiNdtFY050x9wsyhMnmaSyaGbSIXNzubjPBqiD+c8P/2GyKpUSimrtIKQJsQt; version=16070430; wap_sid2=CMOw8aYBElx2TWQtOGJfNkp3dmZHb3dyRnpRajZsVlVGX0pQem4ycWZSNzNFRmY3Vk9zaXZUM0Y5b0ZpbThVeWgzWER6Z0RBbmxqVGFiQ01ndFJyN01LNU9PREs3OXNEQUFBfjC409ngBTgNQJVO; wxuin=349984835; wxtokenkey=777; rewardsn=; pac_uid=0_f82bd5abff9aa; pgv_pvid=2237276040; tvfe_boss_uuid=05faefd1e90836f4', 'Accept': '*/*', 'Referer': 'https://**?action=home&__biz=MzIxNzYxMTU0OQ==&scene=126&sessionid=1544890100&subscene=0&devicetype=iOS12.1&version=16070430&lang=zh_CN&nettype=WIFI&a8scene=0&fontScale=100&pass_ticket=pg%2B0C5hdqENXGO6Fq1rED9Ypx20C2vuodaL8DCwZwVe22sv9OtWgeL5YLjUujPOR&wx_header=1' }
最后通過 requests 去模擬發(fā)送請(qǐng)求。
response = requests.get(current_request_url, headers=self.headers, verify=False)result = response.json()
通過 Charles 返回的數(shù)據(jù)格式可以得知消息列表的數(shù)據(jù)存儲(chǔ)在 general_msg_list 這個(gè) Key 下面。因此可以需要拿到數(shù)據(jù)后進(jìn)行解析操作。
has_next_page 字段可以判斷是否存在下一頁的數(shù)據(jù)；如果有下一頁的數(shù)據(jù)，可以繼續(xù)爬取，否則終止爬蟲程序。
ps：由于 Wx 反爬做的很完善，所以盡量降低爬取的速度。
response = requests.get(current_request_url, headers=self.headers, verify=False) result = response.json() if result.get("ret") == 0: msg_list = result.get('general_msg_list') # 保存數(shù)據(jù) self._save(msg_list) self.logger.info("獲取到一頁數(shù)據(jù)成功, data=%s" % (msg_list)) # 獲取下一頁數(shù)據(jù) has_next_page = result.get('can_msg_continue') if has_next_page == 1: # 繼續(xù)爬取寫一頁的數(shù)據(jù)【通過next_offset】 next_offset = result.get('next_offset') # 休眠2秒，繼續(xù)爬下一頁 time.sleep(2) self.spider_more(next_offset) else: # 當(dāng) has_next 為 0 時(shí)，說明已經(jīng)到了最后一頁，這時(shí)才算爬完了一個(gè)公眾號(hào)的所有歷史文章 print('爬取公號(hào)完成！') else: self.logger.info('無法獲取到更多內(nèi)容，請(qǐng)更新cookie或其他請(qǐng)求頭信息')
由于獲取到的列表數(shù)據(jù)是一個(gè)字符串，需要通過 json 庫去解析，獲取有用的數(shù)據(jù)。
def _save(self, msg_list): """ 數(shù)據(jù)解析 :param msg_list: :return: """ # 1.去掉多余的斜線,使【鏈接地址】可用 msg_list = msg_list.replace("\/", "/") data = json.loads(msg_list) # 2.獲取列表數(shù)據(jù) msg_list = data.get("list") for msg in msg_list: # 3.發(fā)布時(shí)間 p_date = msg.get('comm_msg_info').get('datetime') # 注意：非圖文消息沒有此字段 msg_info = msg.get("app_msg_ext_info") if msg_info: # 圖文消息 # 如果是多圖文推送，把第二條第三條也保存 multi_msg_info = msg_info.get("multi_app_msg_item_list") # 如果是多圖文，就從multi_msg_info中獲取數(shù)據(jù)插入；反之直接從app_msg_ext_info中插入 if multi_msg_info: for multi_msg_item in multi_msg_info: self._insert(multi_msg_item, p_date) else: self._insert(msg_info, p_date) else: # 非圖文消息 # 轉(zhuǎn)換為字符串再打印出來 self.logger.warning(u"此消息不是圖文推送，data=%s" % json.dumps(msg.get("comm_msg_info")))
最后一步是將數(shù)據(jù)保存保存到 MongoDB 數(shù)據(jù)庫中。首先要?jiǎng)?chuàng)建一個(gè) Model 保存我們需要的數(shù)據(jù)。
from datetime import datetimefrom mongoengine import connectfrom mongoengine import DateTimeFieldfrom mongoengine import Documentfrom mongoengine import IntFieldfrom mongoengine import StringFieldfrom mongoengine import URLField__author__ = 'xag'# 權(quán)限連接數(shù)據(jù)庫【數(shù)據(jù)庫設(shè)置了權(quán)限，這里必須指定用戶名和密碼】response = connect('admin', host='localhost', port=27017,username='root', password='xag')class Post(Document): """ 文章【模型】 """ title = StringField() # 標(biāo)題 content_url = StringField() # 文章鏈接 source_url = StringField() # 原文鏈接 digest = StringField() # 文章摘要 cover = URLField(validation=None) # 封面圖 p_date = DateTimeField() # 推送時(shí)間 author = StringField() # 作者 content = StringField() # 文章內(nèi)容 read_num = IntField(default=0) # 閱讀量 like_num = IntField(default=0) # 點(diǎn)贊數(shù) comment_num = IntField(default=0) # 評(píng)論數(shù) reward_num = IntField(default=0) # 點(diǎn)贊數(shù) c_date = DateTimeField(default=datetime.now) # 數(shù)據(jù)生成時(shí)間 u_date = DateTimeField(default=datetime.now) # 數(shù)據(jù)最后更新時(shí)間
使用命令行開啟數(shù)據(jù)庫服務(wù)，然后就可以往數(shù)據(jù)庫寫入數(shù)據(jù)了。
def _insert(self, msg_info, p_date): """ 數(shù)據(jù)插入到 MongoDB 數(shù)據(jù)庫中 :param msg_info: :param p_date: :return: """ keys = ['title', 'author', 'content_url', 'digest', 'cover', 'source_url'] # 獲取有用的數(shù)據(jù),構(gòu)建數(shù)據(jù)模型 data = sub_dict(msg_info, keys) post = Post(**data) # 時(shí)間格式化 date_pretty = datetime.fromtimestamp(p_date) post["p_date"] = date_pretty self.logger.info('save data %s ' % post.title) # 保存數(shù)據(jù) try: post.save() except Exception as e: self.logger.error("保存失敗 data=%s" % post.to_json(), exc_info=True)
05 爬取結(jié)果
推薦使用工具 Robo3T 連接 MongoDB 數(shù)據(jù)庫，可以查看到公號(hào)文章數(shù)據(jù)已經(jīng)全部保存到數(shù)據(jù)庫中。
本文首發(fā)于公眾號(hào)「 AirPython 」，后臺(tái)回復(fù)「 公號(hào)1 」即可獲取完整代碼。
