Python实现微信公众号群发图片与文本消息的完整流程
本文详细介绍使用Python调用微信接口实现公众号群发图片与文本消息的方法。内容涵盖获取access_token、上传图片获取media_id、查询用户分组及筛选有效分组,并通过完整代码示例演示如何向指定分组群发文本和图片,同时解析测试号权限限制导致的错误码。
本文通过一个具体实例,详细讲解如何利用Python实现微信公众号的群发图片与文本消息功能,供读者参考学习。
在调用微信公众号相关API时,必须附加有效的access_token参数。因此,实现群发功能的第一步是获取该令牌,具体实现如下所示。
#获取微信access_token
def get_token():
payload_access_token={
'grant_type':'client_credential',
'appid':'xxxxxxxxxxxxx',
'secret':'xxxxxxxxxxxxx'
}
token_url='https://api.weixin.qq.com/cgi-bin/token'
r=requests.get(token_url,params=payload_access_token)
dict_result= (r.json())
return dict_result['access_token']
群发图片消息时,必须提供已上传至微信服务器的图片media_id。需要注意的是,上传图片必须通过指定接口:https://api.weixin.qq.com/cgi-bin/material/add_material 完成。
#获取上传文件的media_ID
#群发图片的时候,必须使用该api提供的media_ID
def get_media_ID(path):
img_url='https://api.weixin.qq.com/cgi-bin/material/add_material'
payload_img={
'access_token':get_token(),
'type':'image'
}
data ={'media':open(path,'rb')}
r=requests.post(url=img_url,params=payload_img,files=data)
dict =r.json()
return dict['media_id']
对于订阅号而言,执行群发操作必须基于用户分组ID。因此,首先需要获取公众号下所有的用户分组信息。
#查询所有用户分组信息
def get_group_id():
url="https://api.weixin.qq.com/cgi-bin/groups/get"
payload_id={
'access_token':get_token()
}
r=requests.get(url=url,params=payload_id)
result=r.json()
return result['groups']
执行群发时需选定一个分组,本例中策略是选择第一个用户数不为0的有效分组进行消息推送。
#返回第一个有效的group 分组id
def get_first_group_id():
groups =get_group_id()
group_id =0
for group in groups:
if(group['count']!=0):
group_id=group['id']
break;
return group_id
以下代码实现了向第一个有效分组群发文本消息的功能。
def send_txt_to_first_group(str='Hello World!'):
group_id =get_first_group_id()
pay_send_all={
"filter":{
"is_to_all":False,
"group_id":group_id
},
"text":{
"content":str
},
"msgtype":"text"
}
url="https://api.weixin.qq.com/cgi-bin/message/mass/sendall?access_token="+get_token()
#需要指定json编码的时候不会对中文转码为unicode,否则群发的消息会显示为unicode码,不能正确显示
r=requests.post(url=url,data=json.dumps(pay_send_all,ensure_ascii=False,indent=2))#此处的必须指定此参数
result=r.json()
#根据返回码的内容是否为0判断是否成功
return result['errcode']==0
以下代码实现了向第一个有效分组群发图片消息的功能。
def send_img_to_first_group(path='/home/fit/Desktop/test.jpg'):
group_id =get_first_group_id()
pay_send_all={
"filter":{
"is_to_all":False,
"group_id":group_id
},
"image":{
"media_id":get_media_ID(path)
},
"msgtype":"image"
}
url="https://api.weixin.qq.com/cgi-bin/message/mass/sendall?access_token="+get_token()
r=requests.post(url=url,data=json.dumps(pay_send_all))
result=r.json()
#根据返回码的内容是否为0判断是否成功
return result['errcode']==0
以下是整合上述逻辑的完整代码示例。
# -*- coding: utf-8 -*-
import requests
#首先获取access_token
import json
#获取微信access_token
def get_token():
payload_access_token={
'grant_type':'client_credential',
'appid':'xxxxxxxxxx',
'secret':'xxxxxxxxx'
}
token_url='https://api.weixin.qq.com/cgi-bin/token'
r=requests.get(token_url,params=payload_access_token)
dict_result= (r.json())
return dict_result['access_token']
#获取上传文件的media_ID
#群发图片的时候,必须使用该api提供的media_ID
def get_media_ID(path):
img_url='https://api.weixin.qq.com/cgi-bin/material/add_material'
payload_img={
'access_token':get_token(),
'type':'image'
}
data ={'media':open(path,'rb')}
r=requests.post(url=img_url,params=payload_img,files=data)
dict =r.json()
return dict['media_id']
#查询所有用户分组信息
def get_group_id():
url="https://api.weixin.qq.com/cgi-bin/groups/get"
payload_id={
'access_token':get_token()
}
r=requests.get(url=url,params=payload_id)
result=r.json()
return result['groups']
#返回第一个有效的group 分组id
def get_first_group_id():
groups =get_group_id()
group_id =0
for group in groups:
if(group['count']!=0):
group_id=group['id']
break;
return group_id
def send_img_to_first_group(path='/home/fit/Desktop/test.jpg'):
group_id =get_first_group_id()
pay_send_all={
"filter":{
"is_to_all":False,
"group_id":group_id
},
"image":{
"media_id":get_media_ID(path)
},
"msgtype":"image"
}
url="https://api.weixin.qq.com/cgi-bin/message/mass/sendall?access_token="+get_token()
r=requests.post(url=url,data=json.dumps(pay_send_all))
result=r.json()
print result
#根据返回码的内容是否为0判断是否成功
return result['errcode']==0
def send_txt_to_first_group(str='Hello World!'):
group_id =get_first_group_id()
pay_send_all={
"filter":{
"is_to_all":False,
"group_id":group_id
},
"text":{
"content":str
},
"msgtype":"text"
}
url="https://api.weixin.qq.com/cgi-bin/message/mass/sendall?access_token="+get_token()
#需要指定json编码的时候不会对中文转码为unicode,否则群发的消息会显示为unicode码,不能正确显示
r=requests.post(url=url,data=json.dumps(pay_send_all,ensure_ascii=False,indent=2))#此处的必须指定此参数
result=r.json()
#根据返回码的内容是否为0判断是否成功
return result['errcode']==0
if(send_txt_to_first_group("祝你合家欢乐,幸福美满!")):
print 'success!'
else:
print 'fail!'
附录:在使用微信测试订阅号测试群发图片接口时,可能返回如下错误码。
{u'errcode': 45028, u'errmsg': u'has no masssend quota hint: [OKvFdA0813ge12]'}
该错误表明测试订阅号本身不具备群发图文消息的权限,并非接口调用逻辑存在错误。
补充说明:
作者GitHub仓库地址:https://github.com/zhoudayang
对Python技术感兴趣的读者,还可参考本站专题:《Python字符串操作技巧汇总》、《Python编码操作技巧总结》、《Python数据结构与算法教程》、《Python函数使用技巧总结》以及《Python入门与进阶经典教程》。
希望本文所述的实现方法能对大家的Python编程实践提供有益帮助。
本站声明
本文内容仅供参考,不代表本站观点。转载或引用请注明出处。如发现内容问题,请通过联系我们页面反馈。













