爬虫爬取公众号文章

news2024/9/25 1:22:55

前言

自从chatGPT出现后，对于文本处理的能力直接上升了一个维度。在这之前，我们爬取到网络上的文本内容之后，都需要写一个文本清理的程序，对文本进行清洗，而现在，有了chatGPT的加持，我们只需要几秒，就可以很方便对所有类型，所有格式的文本内容，完成清洗，去除那些html标签等。
对于清洗后的文章，我们仍然可以做很多事情，比如，提取核心观点，文章改写等操作，使用chatGPT都可以很轻松的解决。

早在2019年，我就写过一篇文章，介绍爬虫爬取公众号文章的方法，现在看来依然适用，只不过，当时我们对于爬到的文章，后续基本没有什么处理了，但是现在不同了，我们可以做很多事情，不过前提是不要违反法律法规。

获取文章的URL

爬虫爬取网络资源，最重要的就是可以找到目标网站的url地址，然后寻找规律，遍历地址逐个或多线程进行爬取。一般获取地址的方式，一是通过网页分页，推断出url地址的规律吗，比如通过参数pageNum = num等，只需要将num进行累加即可。另外一种是隐藏在HTML中的，我们需要解析出当前网页的超链接，如< a href="url">，取出url作为后续爬取的地址。

遗憾的是，微信公众号对于这两种方式都很难使用，我们可以在网页中打开一个微信公众号地址，可以看下它的url格式，

https://mp.weixin.qq.com/s?__biz=MzIxMTgyODczNg==&mid=2247483660&idx=1&sn=2c14b9b416e2d8eeed0cbbc6f44444e9&chksm=974e2863a039a1752605e6e76610eb39e5855c473edab16ab2c7c8aa2f624b4d892d26130110&token=20884314&lang=zh_CN#rd

除了前面的域名，后面的参数完全就是没有规律可循，而且，在一篇文章中，也没有办法链接到下一篇文章，我们无法从一篇文章开始，爬取到这个公众号下的所有文章。

但好在，我们依然又办法批量获取到某个公众号下的所有文章地址，我们只需要将其保存下来，后续再遍历爬取就变得轻松很多。

1.首先，你要有一个公众号，如果没有的话，可以注册一个，这个是前提条件。注册公众号的步骤比较简单，可以自行操作。
2.注册好之后，登录微信公众平台，点击左侧草稿箱，新建图文
在这里插入图片描述
3.进入新建图文页面，点击顶端的超链接，出现一个弹窗，选择公众号文章，输入你要爬取的公众号的名称，如下所示：

4.选中之后，你就可以看到这个公众号下的所有文章的列表了，这个时候，打开F12，查看网页里面的网络请求。
在这里插入图片描述
当你点击下一页的时候，可以看到请求的url，还有携带的业务参数以及请求头参数。

上面就是一些业务参数，这些参数很容易理解，比较重要的是begin指的是从第几个开始查询，count指的是一次列表查询多少个，fakeid是公众号的唯一标识，每个公众号不一样，如果爬取其他公众号的，只需要更改这个参数就可以了。其中random可以省略。还可以看到相应结果：
在这里插入图片描述

编写代码

有了上述这些信息，我们就可以编写代码了，我使用的是python3.8，首先，定义url，header和所需要的参数

# 目标url
url = "https://mp.weixin.qq.com/cgi-bin/appmsg"
# 请求头参数
headers = {
  "Cookie": "ua_id=YF6RyP41YQa2QyQHAAAAAGXPy_he8M8KkNCUbRx0cVU=; pgv_pvi=2045358080; pgv_si=s4132856832; uuid=48da56b488e5c697909a13dfac91a819; bizuin=3231163757; ticket=5bd41c51e53cfce785e5c188f94240aac8fad8e3; ticket_id=gh_d5e73af61440; cert=bVSKoAHHVIldcRZp10_fd7p2aTEXrTi6; noticeLoginFlag=1; remember_acct=mf1832192%40smail.nju.edu.cn; data_bizuin=3231163757; data_ticket=XKgzAcTceBFDNN6cFXa4TZAVMlMlxhorD7A0r3vzCDkS++pgSpr55NFkQIN3N+/v; slave_sid=bU0yeTNOS2VxcEg5RktUQlZhd2xheVc5bjhoQTVhOHdhMnN2SlVIZGRtU3hvVXJpTWdWakVqcHowd3RuVF9HY19Udm1PbVpQMGVfcnhHVGJQQTVzckpQY042QlZZbnJzel9oam5SdjRFR0tGc0c1eExKQU9ybjgxVnZVZVBtSmVnc29ZcUJWVmNWWEFEaGtk; slave_user=gh_d5e73af61440; xid=93074c5a87a2e98ddb9e527aa204d0c7; openid2ticket_obaWXwJGb9VV9FiHPMcNq7OZzlzY=lw6SBHGUDQf1lFHqOeShfg39SU7awJMxhDVb4AbVXJM=; mm_lang=zh_CN",
  "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.62 Safari/537.36",
}

# 业务参数
data = {
    "token": "1378111188",
    "lang": "zh_CN",
    "f": "json",
    "ajax": "1",
    "action": "list_ex",
    "begin": "0",
    "count": "5",
    "query": "",
    "fakeid": "MzU5MDUzMTk5Nw==",
    "type": "9",
}

cookie和token需要根据自己请求URL中的进行更改，然后发送请求获取响应，并对响应结果进行解析。

content_list = []
for i in range(20):
    data["begin"] = i*5
    time.sleep(5)
    # 使用get方法进行提交
    content_json = requests.get(url, headers=headers, params=data).json()
    # 返回了一个json，里面是每一页的数据
    for item in content_json["app_msg_list"]:    
    # 提取每页文章的标题及对应的url
        items = []
        items.append(item["title"])
        items.append(item["link"])
        t = time.localtime(item["create_time"])
        items.append(time.strftime("%Y-%m-%d %H:%M:%S", t))
        content_list.append(items)

第一个for循环是爬取的页数，这里建议一次爬取20页，每一页5条数据，就是100篇文章了，首先需要看好公众号的历史文章列表一共有多少页，这个数只能小于页数。更改data[“begin”]，表示从第几条开始，每次5条，注意爬取不能够太多和太频繁，所以需要每次爬取等待几秒钟，不然会被封ip封cookie，严重还会封公众号。

最后，我们只需要把标题和url保存起来就可以了，保存起来后，就可以逐个爬取了。

name = ['title', 'link', 'create_time']
test = pd.DataFrame(columns=name, data=content_list)
test.to_csv("url.csv", mode='a', encoding='utf-8')

为了获取所有的历史文章，需要得到文章总数，从app_msg_cnt中获取，然后计算一共有多少页，就可以一次性爬取所有文章了。

content_json = requests.get(url, headers=headers, params=data).json()
count = int(content_json["app_msg_cnt"])
print(count)
page = int(math.ceil(count / 5))
print(page)

为了抓取的不够频繁，我们需要不断sleep一段时间，我们可以爬取10次之后，就让程序sleep几秒

if (i > 0) and (i % 10 == 0):
        name = ['title', 'link', 'create_time']
        test = pd.DataFrame(columns=name, data=content_list)
        test.to_csv("url.csv", mode='a', encoding='utf-8')
        print("第" + str(i) + "次保存成功")
        content_list = []
        time.sleep(random.randint(60,90))
    else:
        time.sleep(random.randint(15,25))