Python——如何優雅的爬取公眾號資訊

• 2019 年 10 月 4 日
• 筆記

這是奔跑的鍵盤俠的第134篇文章

作者|我是奔跑的鍵盤俠

來源|奔跑的鍵盤俠（ID：runningkeyboardhero）

轉載請聯繫授權（微信ID：ctwott）

最近兩個周業餘時間在趕的一個項目，因為精力有限所以進展緩慢，索性就先把接近完善的這部分程式碼，先分享出來吧。

寫個爬蟲來爬取公眾號資訊，不知道會不會被公眾號後台K

且看且珍惜吧。

先貼一下目錄：

├── README  ├── wechat_crawler #微信公眾號爬蟲目錄     ├── __init__.py     ├── data #數據存儲目錄     │   ├── __init__.py

   │   ├── account.py #以字典形式存儲個人公眾號用戶名、密碼資訊。

   │   ├── cookie.txt  #記錄網頁登陸的cookie資訊。

   │   ├── 奔跑的鍵盤俠.txt  #運行程式碼後爬取的公眾號文章資訊。

   │   └── 十點讀書.txt  #運行程式碼後爬取的公眾號文章資訊。

   └─ ─ crawler #爬蟲主程式碼

     └── __init__.py

      └── crawler.py #包含登陸、爬取公眾號文章核心程式碼。

這個實現的前提，是要通過個人微信公眾號，在後台搜索其他公眾號的文章清單，得以實現。另外一個途徑呢，就是通過搜狗微信，最早我也是從這條途徑下手的，結果夭折了

按照原計劃是要爬取完幾個目標公眾號的帖子，然後分別再爬取對應帖子中的數據，最後再清洗數據、數據分析。這麼久只搞定了第一步，而且還有個半大不小的問題待解決……

1

coding

#!/usr/bin/env python3.6  # -*- coding: utf-8 -*-  # @Time    : 2019-09-10 18:37  # @Author  : Ed Frey  # @File    : crawler.py  # @Software: PyCharm  from selenium import webdriver  import time  import json  import random  import requests  import re  import urllib3  import math  import os,sys      BASE_DIR = os.path.abspath("..")  sys.path.append(BASE_DIR)  from db.account import ACCOUNT    class Wechat_Crawl:        def __init__(self,ACCOUNT,cookie_path,subscription_account):          self.__username = ACCOUNT["username"]          self.__password = ACCOUNT["password"]          self.begin = 0          self.i = 1          self.cookie_path = cookie_path          self.qurey = subscription_account          self.url =  'https://mp.weixin.qq.com'          self.search_url = 'https://mp.weixin.qq.com/cgi-bin/searchbiz?'          self.appmsg_url = 'https://mp.weixin.qq.com/cgi-bin/appmsg?'          self.header = {              "HOST": "mp.weixin.qq.com",              "User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36"          }        def wechat_login(self):          '''          Login wechat by automatically inputing accounts and keywords and manully scanning QR code， and then you can          get the cookie information, save it in local file in order to simulate loginning and crawling……          :param __username:          :param __password:          :return:          '''          print("瀏覽器將自動打開並跳轉至微信公眾號登錄頁面……")          time.sleep(1)          driver = webdriver.Chrome()          driver.get("https://mp.weixin.qq.com/")          time.sleep(2)          print("正在自動輸入帳號、密碼......請勿操作")          driver.find_element_by_name("account").clear()          driver.find_element_by_name("account").send_keys(self.__username)          time.sleep(1)          driver.find_element_by_name("password").clear()          driver.find_element_by_name("password").send_keys(self.__password)          time.sleep(1)          driver.find_element_by_class_name("frm_checkbox_label").click()          time.sleep(2)          driver.find_element_by_class_name("btn_login").click()          print("請拿手機掃碼二維碼登錄公眾號")          time.sleep(15)          print("登錄成功")            cookies = driver.get_cookies()          info = {}          for cookie in cookies:              info[cookie['name']] = cookie['value']          cookie_info = json.dumps(info)          print(cookie_info)          with open(cookie_path, 'w+', encoding='utf-8') as f:              f.write(cookie_info)              f.flush()          print("cookies已存入cookie.txt",flush=True)          driver.quit()        def get_cookie(self):          if not os.path.isfile(self.cookie_path):              print("指定路徑未發現cookie文件，請重新掃描登陸……")              self.wechat_login()          with open(self.cookie_path, 'r', encoding='utf-8') as f:              cookie = f.read()          self.cookies = json.loads(cookie)          def _session(self):          urllib3.disable_warnings()          # requests庫urllib編寫，是對urllib進行了封裝，在urllib2版本對https的處理非常簡單，只需要在請求的時候加上verify = False即可，          # 這個參數的意思是忽略https安全證書的驗證，也就是不驗證證書的可靠性，直接請求，這其實是不安全的，因為證書可以偽造，不驗證的話就不          # 能保證數據的真實性。          # 在urllib3版本，官方強制驗證https的安全證書，如果沒有通過是不能通過請求的，雖然添加忽略驗證的參數，但是依然會給出醒目的Warning          # urllib3.disable_warnings()可以禁用urllib3的警告。          session = requests.Session()          session.keep_alive = False          session.adapters.DEFAULT_RETRIES = 511          self.session = session        def get_token(self):          '''          to get the token from the loginned page.          :param cookie_path:          :return: token          '''          time.sleep(1)          response = self.session.get(url=self.url, cookies=self.cookies, verify=False)          url = str(response.url)          pattern = r'token=([0-9]+)'          self.token = re.findall(pattern,url)[0]            return self.token        def get_fakedid(self):          query_id = {              'action': 'search_biz',              'token': self.token,              'lang': 'zh_CN',              'f': 'json',              'ajax': '1',              'random': random.random(),              'query': self.qurey,              'begin': '0',              'count': '5'          }          search_response = self.session.get(              self.search_url,              headers=self.header,              cookies=self.cookies,              params=query_id)          lists = search_response.json()['list'][0]            self.fakeid = lists['fakeid']        def get_args(self):          self.get_cookie()          self._session()          self.get_token()          self.get_fakedid()        def get_info(self,output_path):          self.data = {              "token": self.token,              "lang": "zh_CN",              "f": "json",              "ajax": "1",              "action": "list_ex",              "begin": self.begin,              "count": "5",              "query": "",              "fakeid": self.fakeid,              "type": "9",          }          res = requests.get(self.appmsg_url, cookies=self.cookies, headers=self.header, params=self.data)          try:              json = res.json()              count = json['app_msg_cnt']                for item in json["app_msg_list"]:                  create_date = time.strftime("%Y-%m-%d", time.localtime(item['create_time']))                  update_date = time.strftime("%Y-%m-%d", time.localtime(item['update_time']))                  title = item['title'].replace("n", "")                  link = item['link']                  with open(output_path, 'a', encoding='utf-8') as fh:                      article_info = "%s|%s|%s|%sn" % (create_date, update_date, title, link)                      fh.write(article_info)                      fh.flush()                loop_count = math.ceil(count / 5)              total_page = loop_count+1              while loop_count > 1:                  time.sleep(5)                  loop_count -= 1                  self.begin = 5 * self.i                  print("已完成下載第%s/%s頁..." % (self.i, total_page))                  self.i += 1                  self.get_info(output_path)            except:              print(res.json()['base_resp']['err_msg'])              print("Try it again after two hours.")              exit()        def run(self,output_path):          self.get_args()          self.get_info(output_path)    if __name__ == '__main__':        subscription_account = "奔跑的鍵盤俠"      cookie_path = BASE_DIR + os.sep + 'db' + os.sep + 'cookie.txt'      output_path = BASE_DIR + os.sep + 'db' + os.sep + 'data' + os.sep + subscription_account + '.txt'      wc = Wechat_Crawl(ACCOUNT,cookie_path,subscription_account)      print("開始爬取公眾號:",subscription_account)      wc.run(output_path)      print("爬取完成")

2

調試問題匯總

現在主流網站都有反爬機制，爬取過於頻繁就會被禁止訪問。於是乎，公眾號平台肯定也會遇到這個問題，個人感覺無解，除非參數設置大於反爬機制的邊界值……

程式碼中get_info函數有設置一個延時5秒，經過實測，即使設置在30秒左右（隨機數）的延時，依舊會被檢測到。

反正吧，測試環節就耗費了較多精力，而且一旦被禁，可能要隔挺長一段時間才能恢復，昨晚一個測試，間隔了大概24小時登陸帳號才能繼續爬取數據，實在折騰不起……

運行結果的txt文件，簡單截取一部分：

感興趣的小夥伴可以自行測試，最後一個欄位的鏈接，是可以直接訪問的哦。

比如一個公眾號有成百上千個帖子，txt文檔翻一下，很快就找到自己的目標帖子，然後直接點鏈接閱讀即可，省去手機上翻頁的苦惱

當然如果是某些提供數據的訂閱號，可以從這幾百個帖子鏈接中，直接解析出需要的數據，這個相對而言就比較簡單，就不再贅述了

-END-

© Copyright

奔跑的鍵盤俠原創作品 | 盡情分享朋友圈 | 轉載請聯繫授權