Python——如何优雅的爬取公众号信息

这是奔跑的键盘侠的第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|%s\n" % (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

奔跑的键盘侠原创作品 | 尽情分享朋友圈 | 转载请联系授权

原文发布于微信公众号 - 奔跑的键盘侠(runningkeyboardhero)

原文发表时间:2019-09-11

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

扫码关注云+社区

领取腾讯云代金券