这是我第一条写关于爬虫的文章
一、抓取微信公众号的文章有如下几点需要注意的地方
1、利用微信接口进行抓取，每天-每个公众号只能调用1000次，而且，不能连续调用，也就是说，你调用300次左右，可能会被封一个小时左右，但总的来说，一天调用的上线是1000次
2、可能在你调用100多次的时候，会出现一个反爬的情况，就是当你获取文章列表的时候，点击获取，发现返回来的是空值，这时候，就要写个递归，继续调用，经过3-4次就会成功(亲身经历)
3、ip代理是必要的，无论是调用微信的接口还是搜狗微信，如果想持续并且获得大量的数据，没有ip，爬不了多少数据的
4、接下来，我将贴出我的代码
from selenium import webdriver
import requests
import time
import re
import threading
import random
import redis
from bs4 import BeautifulSoup
import sys
import json
import urllib3
from fake_useragent import UserAgent
from pymysql import connect
from mon.proxy import Proxy
from mon.proxy import ProxyType
from email.mime.text import MIMEText
from email.header import Header
from email.utils import parseaddr, formataddr
import smtplib
sys.setrecursionlimit(10000)
class Wechat(object):
def __init__(self):
#从redis里获取ip
self.redis_cli = redis.Redis(host="120.77.216.243", port=6480, db=1, password="********",
decode_responses=True)
self.db = connect(host="localhost", port=3306, db="simhash", user="root",
password="zy79117911",
charset="utf8")
self.cursor = self.db.cursor()
self.useragent = UserAgent()
# 微信公众号账号
self.user = "你自己的公众号账号"
# 账号密码
self.password = "密码"
self.isgo = False
self.bool = 1
def __del__(self):
# self.db.close()
pass
def weChat_login(self):
#定义一个空的字典，存放cookies内容
post={}
proxy = Proxy(
{
'proxyType': ProxyType.MANUAL,
'httpProxy': self.get_IP()
}
)
driver = webdriver.Firefox(proxy=proxy)
#打开微信公众号登录页面
try:
driver.get('微信公众平台')
except:
self.weChat_login()
return
print('正在获取微信公众号登录账号和密码...')
# 清空账号框中的内容
driver.find_element_by_xpath("./*//input[@name='account']").clear()
# 自动填入登录用户名
driver.find_element_by_xpath("./*//input[@name='account']").send_keys(self.user)
# 清空密码框中的内容
driver.find_element_by_xpath("./*//input[@name='password']").clear()
# 自动填入登录密码
driver.find_element_by_xpath("./*//input[@name='password']").send_keys(self.password)
# print("请在登录界面点击:记住账号")
# time.sleep(10)
#自动点击登录按钮进行登录
driver.find_element_by_class_name('btn_login').click()
# 在自动输完密码之后需要手动点一下记住我
#拿手机扫二维码登录公众号
print("请拿手机扫码二维码登录公众号")
time.sleep(20)
#发送邮件提示微信公众号登陆成功
self.send_email('微信公众号登陆成功')
# driver.get('微信公众平台')
# 获取cookies
cookie_items = driver.get_cookies()
for cookie_item in cookie_items:
post[cookie_item['name']] = cookie_item['value']
cookie_str = json.dumps(post)
with open('cookie.txt', 'w+', encoding='utf-8') as f:
f.write(cookie_str)
print("cookies信息已保存到本地")
def get_content(self, query):
headers = {
"HOST": "微信公众平台",
"User-Agent": self.useragent.random,
"Connection": "close"
}
#去除警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
#query为要爬取的公众号名称
#公众号主页
url = '微信公众平台'
#设置headers
#读取上一步获取到的cookies
with open('cookie.txt', 'r', encoding='utf-8') as f:
cookie = f.read()
cookies = json.loads(cookie)
# 增加重试连接次数
session = requests.Session()
session.keep_alive = False
session.adapters.DEFAULT_RETRIES = 511
time.sleep(5)
#登录之后的微信公众号首页url变化为：微信公众平台
response = requests.get(url=url, cookies=cookies, verify=False)
print(response)
token = re.findall(r'token=(\d+)', str(response.url))[0]
#搜索微信公众号的接口地址
search_url = '微信公众平台
#搜索微信公众号接口需要传入的参数，有三个变量：微信公众号token、随机数random、搜索的微信公众号名字
query_id = {
'action': 'search_biz',
'token' : token,
'lang': 'zh_CN',
'f': 'json',
'ajax': '1',
'random': random.random(),
'query': query,
'begin': '0',
'count': '5'
}
#打开搜索微信公众号接口地址，需要传入相关参数信息如：cookies、params、headers
search_response = self.select_name(search_url, cookies, headers, query_id)
#取搜索结果中的第一个公众号
lists = search_response.json().get('list')[0]
official_account = lists.get('nickname')
print(lists.get('nickname'))
#获取这个公众号的fakeid，后面爬取公众号文章需要此字段
fakeid = lists.get('fakeid')
#微信公众号文章接口地址
appmsg_url = '微信公众平台
#搜索文章需要传入几个参数：登录的公众号token、要爬取文章的公众号fakeid、随机数random
query_id_data = {
'token': token,
'lang': 'zh_CN',
'f': 'json',
'ajax': '1',
'random': random.random(),
'action': 'list_ex',
'begin': '0',#不同页，此参数变化，变化规则为每页加5
'count': '5',
'query': '',
'fakeid': fakeid,
'type': '9'
}
#打开搜索的微信公众号文章列表页
appmsg_response = self.get_first_official_account(appmsg_url, cookies, headers, query_id_data)
#访问频繁,跳出，并发邮件通知
if appmsg_response == True:
return True
#微信的反爬手段
while True:
if appmsg_response == None:
time.sleep(10)
print("返回来的值为空")
appmsg_response = self.get_first_official_account(appmsg_url, cookies, headers, query_id_data)
else:
break
print("这一步返回来的值：", type(appmsg_response), appmsg_response)
print(appmsg_response.json())
# print("======", appmsg_response.text)
# print(appmsg_response)
#获取文章总数
max_num = int(appmsg_response.json().get('app_msg_cnt'))
print("总文章数：", max_num)
begin = 1
#当前系统的时间戳
start_time = int(time.time())
while True:
print("调用微信接口第", self.bool, "次")
time.sleep(random.randint(5, 8))
query_id_data = {
'token': token,
'lang': 'zh_CN',
'f': 'json',
'ajax': '1',
'random': random.random(),
'action': 'list_ex',
'begin': '{}'.format(str(begin)),
'count': '5',
'query': '',
'fakeid': fakeid,
'type': '9'
}
print('爬取到第', begin, "条数据")
query_fakeid_response = self.parse_page(appmsg_url, cookies, self.get_ip(), headers, query_id_data)
# print("看一下返回的状态", type(query_fakeid_response), query_fakeid_response)
if query_fakeid_response == None:
print("8888888888888888888888888888888888888")
query_fakeid_response = self.parse_page(appmsg_url, cookies, self.get_ip(), headers,
query_id_data)
print("77777777777777", query_fakeid_response)
try:
print("66666666666", query_fakeid_response)
fakeid_list = query_fakeid_response.json().get('app_msg_list')
if fakeid_list == None:
print(query_fakeid_response.text)
time.sleep(10)
print("返回数据为空，睡眠10秒重新启动...")
continue
except Exception as e:
# print("=========：", query_fakeid_response.text)
print("中途遇到错误，异常：", e)
continue
print("--------", fakeid_list)
for item in fakeid_list:
self.bool += 1
time.sleep(2)
source_url = item.get('link')
title = item.get('title')
publish_time = item.get('update_time')
pic_url = item.get('cover')
#查找从-09-01日以来，发布的文章
# if start_time - publish_time < 259200:
if publish_time > 1441036800:
#考虑到公众号的数据量不是很大，就不用批量插入了
time_local = time.localtime(publish_time)
# 转换成新的时间格式(12月20日15:58:33)
dt = time.strftime("%Y-%m-%d %H:%M:%S", time_local)
sql = "insert into wechat (source_url, title, official_account, publish_time, pic_url) values (%s,%s,%s,%s,%s)" % ("'"+source_url+"'", "'"+title+"'", "'" + official_account + "'", "'"+str(dt)+"'", "'"+pic_url+"'")
print(sql)
try:
self.cursor.execute(sql)
mit()
print("插入mysql数据库成功...")
begin += 1
continue
except:
begin += 1
print("数据已经存在，不用重复插入")
continue
else:
print(".............抓取", query, "公众号已完成................")
return
if max_num <= begin:
print("已抓取完该公众号的文章")
return
time.sleep(2)
def get_first_official_account(self, appmsg_url, cookies, headers, query_id_data):
try:
appmsg_response = requests.get(url=appmsg_url, proxies=self.get_ip(), cookies=cookies, headers=headers, params=query_id_data,
verify=False, timeout=10)
if appmsg_response.json().get('app_msg_cnt') == None:
print("ip被封，继续迭代...睡眠5~10，秒")
# 发送邮件，提醒爬虫异常
self.send_email('公众号访问接口频繁')
self.isgo = True
return self.isgo
else:
print("------>", appmsg_response.text)
return appmsg_response
except Exception as e:
print("发送邮件？")
print("异常...即将发送邮件提醒", e)
self.get_first_official_account(appmsg_url, cookies, headers, query_id_data)
def parse_page(self, appmsg_url, cookies, proxies, headers, query_id_data):
try:
query_fakeid_response = requests.get(appmsg_url, cookies=cookies, proxies=proxies, headers=headers,
params=query_id_data, verify=False, timeout=30)
if query_fakeid_response.json().get('app_msg_list') == None or query_fakeid_response == None:
print("query_fakeid_response返回来的值不正确", query_fakeid_response.text, headers)
self.parse_page(appmsg_url, cookies, self.get_ip(), headers, query_id_data)
else:
print("<<<<<<<<<<<<<
return query_fakeid_response
except:
print(">>>>>>>>>>>>>>>>>有异常，递归开始>>>>>>>>>>>>>>>>>", headers)
self.parse_page(appmsg_url, cookies, self.get_ip(), headers, query_id_data)
#获取搜索出来的第一个公众号
def select_name(self, search_url, cookies, headers, query_id):
try:
print("2222222222222")
search_response = requests.get(search_url, cookies=cookies, headers=headers, params=query_id, verify=False, timeout=30)
return search_response
except:
self.send_email()
self.select_name(search_url, cookies, headers, query_id)
def get_ip(self):
ip = self.redis_cli.srandmember("IP")
print('ip:', ip)
proxies = {
"https": "https://{}".format(ip),
}
# print(proxies)
return proxies
def get_IP(self):
ip = self.redis_cli.srandmember("IP")
return ip
def get_Time(self):
print(".")
global timer
timer = threading.Timer(1, self.get_Time)
timer.start()
if __name__=="__main__":
wechat = Wechat()
wechat.weChat_login()
flag = True
while flag:
#公众号列表
gznames = [
'腾讯','百度','新浪'
]
for gzname in gznames:
bool = wechat.get_content(gzname)
if bool == True:
flag = False
break
# time.sleep(60)
print("抓取", gzname, "公众号的历史文章完成")
# print("抓取公众号完成,睡眠一天，然后继续抓取,连续抓取会被封号")
time.sleep(43200)
代码就如上面所示了，操作步骤:
url:
注册登陆后：
1、在首页的，点击素材管理：
2、点击：
3、点击：
4:根据下图操作：
点击公众号的名称，就会显示所有的历史文章信息
上面的代码就是我爬取微信公众号的代码，如有疑问，可以评论或者私聊我
尚未解决，需要等高人指点：
微信公众号登陆的有效期，好像不到24小时，超过这个期限，就要重新扫码登陆，这是有点不方面的，未能做到全自动，这个需要等待有缘人来帮忙解决了。
