时间:2021-05-22
微信公众号共有三种,服务号、订阅号、企业号。它们在获取AccessToken上各有不同。
其中订阅号比较坑,它的AccessToken是需定时刷新,重复获取将导致上次获取的AccessToken失效。
而企业号就比较好,AccessToken有效期同样为7200秒,但有效期内重复获取返回相同结果。
为兼容这两种方式,因此按照订阅号的方式处理。
处理办法与接口文档中的要求相同:
为了保密appsecrect,第三方需要一个access_token获取和刷新的中控服务器。
而其他业务逻辑服务器所使用的access_token均来自于该中控服务器,不应该各自去刷新,否则会造成access_token覆盖而影响业务。
下面的代码以企业号为例,将access_token储存在sqlite3数据库中,相比储存在文本中,放在数
据库里,可以为后期存放其他数据提供向后兼容。如果放在文本中,则不如放在数据库中灵活。
设计思路和使用方法:
自动创建sqlite3数据库,包括表结构和数据,并能在数据库表结构不存在或者数据不存在或遭删除的情况下,创建新的可用的数据
尽可能的保证Class中每一个可执行的函数单独调用都能成功。
Class中只将真正能被用到的方法和变量设置为public的。
使用时只需要修改此文件中的weixin_qy_CorpID和weixin_qy_Secret改成自己的,并import此文件,使
用WeiXinTokenClass().get()方法即可得到access_token。
Python实现通过微信企业号发送文本消息的Class
编程要点和调用方法:
支持发送中文,核心语句“payload = json.dumps(self.data, encoding='utf-8', ensure_ascii=False)”,关键字“python json 中文”
这个Class只有一个公共方法send()。
使用方法:import这个class,然后调用send方法即可,方法参数只需要两个,给谁(多UserID用"|"隔开),内容是什么,例如:
python调用mongodb发送微信企业号
python2.x
注意:data变量里, agent_id为刚刚创建的应用id(可在web页面看到)
toparty即为目标部门,或者可以用touser,totag指定目标账户
比较简单的调用,已实测,可以使用。
ZABBIX 微信报警 插件
#!/usr/bin/env python# -*- coding:utf-8 -*-# __author__ = '懒懒的天空'import requestsimport sysimport jsonfrom conf.INIFILES import read_config, write_configimport osimport datetimefrom conf.BLog import Logreload(sys)sys.setdefaultencoding('utf-8')class WeiXin(object): def __init__(self, corpid, corpsecret): # 初始化的时候需要获取corpid和corpsecret,需要从管理后台获取 self.__params = { 'corpid': corpid, 'corpsecret': corpsecret } self.url_get_token = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken' self.url_send = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?' self.__token = self.__get_token() self.__token_params = { 'access_token': self.__token } def __raise_error(self, res): raise Exception('error code: %s,error message: %s' % (res.json()['errcode'], res.json()['errmsg'])) global senderr sendstatus = False senderr = 'error code: %s,error message: %s' % (res.json()['errcode'], res.json()['errmsg']) def __get_token(self): # print self.url_get_token headers = {'content-type': 'application/json'} res = requests.get(self.url_get_token, headers=headers, params=self.__params) try: return res.json()['access_token'] except: self.__raise_error(res.content) def send_message(self, agentid, messages, userid='', toparty=''): payload = { 'touser': userid, 'toparty': toparty, 'agentid': agentid, "msgtype": "news", "news": messages } headers = {'content-type': 'application/json'} data = json.dumps(payload, ensure_ascii=False).encode('utf-8') params = self.__token_params res = requests.post(self.url_send, headers=headers, params=params, data=data) try: return res.json() except: self.__raise_error(res)def main(send_to, subject, content): try: global sendstatus global senderr data = '' messages = {} body = {} config_file_path = get_path() CorpID = read_config(config_file_path, 'wei', "CorpID") CorpSecret = read_config(config_file_path, 'wei', "CorpSecret") agentid = read_config(config_file_path, 'wei', "agentid") web = read_config(config_file_path, 'wei', "web") content = json.loads(content) messages["message_url"] = web body["url"] = web + "history.php?action=showgraph&itemids[]=" + content[u'监控ID'] warn_message = '' if content[u'当前状态'] == 'PROBLEM': body["title"] = "服务器故障" warn_message += subject + '\n' warn_message += '详情:\n' warn_message += '告警等级:'+ content[u'告警等级'] + '\n' warn_message += '告警时间:'+ content[u'告警时间'] + '\n' warn_message += '告警地址:'+ content[u'告警地址'] + '\n' warn_message += '持续时间:'+ content[u'持续时间'] + '\n' warn_message += '监控项目:'+ content[u'监控项目'] + '\n' warn_message += content[u'告警主机'] + '故障(' + content[u'事件ID']+ ')' else: body["title"] = "服务器恢复" warn_message += subject + '\n' warn_message += '详情:\n' warn_message += '告警等级:'+ content[u'告警等级'] + '\n' warn_message += '恢复时间:'+ content[u'恢复时间'] + '\n' warn_message += '告警地址:'+ content[u'告警地址'] + '\n' warn_message += '持续时间:'+ content[u'持续时间'] + '\n' warn_message += '监控项目:'+ content[u'监控项目'] + '\n' warn_message += content[u'告警主机'] + '恢复(' + content[u'事件ID']+ ')' body['description'] = warn_message data = [] data.append(body) messages['articles'] = data wx = WeiXin(CorpID, CorpSecret) data = wx.send_message(toparty=send_to, agentid=agentid, messages=messages) sendstatus = True except Exception, e: senderr = str(e) sendstatus = False logwrite(sendstatus, data)def get_path(): path = os.path.dirname(os.path.abspath(sys.argv[0])) config_path = path + '/config.ini' return config_pathdef logwrite(sendstatus, content): logpath = '/var/log/zabbix/weixin' if not sendstatus: content = senderr t = datetime.datetime.now() daytime = t.strftime('%Y-%m-%d') daylogfile = logpath+'/'+str(daytime)+'.log' logger = Log(daylogfile, level="info", is_console=False, mbs=5, count=5) os.system('chown zabbix.zabbix {0}'.format(daylogfile)) logger.info(content)if __name__ == "__main__": if len(sys.argv) > 1: send_to = sys.argv[1] subject = sys.argv[2] content = sys.argv[3] logwrite(True, content) main(send_to, subject, content)python实现微信企业号的文本消息推送
#!/usr/bin/python# _*_coding:utf-8 _*_import urllib2import jsonimport sysreload(sys)sys.setdefaultencoding('utf-8')def gettoken(corpid, corpsecret): gettoken_url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=' + corpid + '&corpsecret=' + corpsecret try: token_file = urllib2.urlopen(gettoken_url) except urllib2.HTTPError as e: print e.code print e.read().decode("utf8") sys.exit() token_data = token_file.read().decode('utf-8') token_json = json.loads(token_data) token_json.keys() token = token_json['access_token'] return tokendef senddata(access_token, user, party, agent, subject, content): send_url = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=' + access_token send_values = "{\"touser\":\"" + user + "\",\"toparty\":\"" + party + "\",\"totag\":\"\",\"msgtype\":\"text\",\"agentid\":\"" + agent + "\",\"text\":{\"content\":\"" + subject + "\n" + content + "\"},\"safe\":\"0\"}" send_request = urllib2.Request(send_url, send_values) response = json.loads(urllib2.urlopen(send_request).read()) print str(response)if __name__ == '__main__': user = str(sys.argv[1]) # 参数1:发送给用户的账号,必须关注企业号,并对企业号有发消息权限 party = str(sys.argv[2]) # 参数2:发送给组的id号,必须对企业号有权限 agent = str(sys.argv[3]) # 参数3:企业号中的应用id subject = str(sys.argv[4]) # 参数4:标题【消息内容的一部分】 content = str(sys.argv[5]) # 参数5:文本具体内容 corpid = 'CorpID' # CorpID是企业号的标识 corpsecret = 'corpsecretSecret' # corpsecretSecret是管理组凭证密钥 try: accesstoken = gettoken(corpid, corpsecret) senddata(accesstoken, user, party, agent, subject, content) except Exception, e: print str(e) + "Error Please Check \"corpid\" or \"corpsecret\" Config"Nagios调用Python程序控制微信公众平台发布报警信息
vim Notify-host-by-weixin-party.py import urllib.requestimport jsonimport sys#以上是导入模块#创建获取AccessToken的方法def gettoken(corp_id,corp_secret): gettoken_url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=' + corp_id + '&corpsecret=' + corp_secret try: token_file = urllib.request.urlopen(gettoken_url) except urllib.error.HTTPError as e: print(e.code) print(e.read().decode("utf8")) token_data = token_file.read().decode('utf-8') token_json = json.loads(token_data) token_json.keys() token = token_json['access_token'] return token#这里是发送消息的方法def senddata(access_token,notify_str): send_url = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=' + access_token#我传入的参数是一段字符串每个信息用separator连起来,只要再用字符串的split("separator")方法分开信息就可以了。 notifydata = notify_str.split("separator") party = notifydata[0] cationtype = notifydata[1] name = notifydata[2] state = notifydata[3] address = notifydata[4] output = notifydata[5] datatime = notifydata[6]# content = '[擦汗]Host Notification[擦汗]\n\n类型: ' + cationtype + '\n主机名: ' + name + '\n状态: ' + state + '\nIP地址: ' + address + '\n摘要: ' + output + '\n时间: ' + datatime + '\n' if cationtype == "RECOVERY": content = '[嘘]' + address + ' is ' + state + '[嘘]\n\nIP地址: ' + address + '\n主要用途: ' + name + '\n当前状态: ' + state + '\n\n日志摘要: ' + output + '\n检测时间: ' + datatime + '\n' else: content = '[擦汗]' + address + ' is ' + state + '[擦汗]\n\nIP地址: ' + address + '\n主要用途: ' + name + '\n当前状态: ' + state + '\n\n日志摘要: ' + output + '\n检测时间: ' + datatime + '\n' send_values = { "toparty":party, "totag":"2", "msgtype":"text", "agentid":"15", "text":{ "content":content }, "safe":"0" } send_data = json.dumps(send_values, ensure_ascii=False).encode(encoding='UTF8')#设置为非ascii解析,使其支持中文 send_request = urllib.request.Request(send_url, send_data) response = urllib.request.urlopen(send_request)#这个是返回微信公共平台的信息,调试时比较有用 msg = response.read() return msgdefault_encoding = 'utf-8'if sys.getdefaultencoding() != default_encoding: reload(sys) sys.setdefaultencoding(default_encoding)#我编辑的脚本是要获取nagios传入的一段参数的(字符串),下面这条代码是获取执行脚本后获取的第一个参数(经测试nagios只能传入一个参进python,所以把所有包括用户名跟报警主机报警信息放进一个字符串里)notifystr = str(sys.argv[1])corpid = 'wxb6162862801114c9da602' corpsecret = '2nCsNcHxepBCV4U9Lcf-23By1RGzU1Zs422tdJpKTQzqjQ1b26IFxP76ydG2rKkchGN6E'accesstoken = gettoken(corpid,corpsecret)msg = senddata(accesstoken,notifystr)print(msg)[root@localhost python]# vim Notify-service-by-weixin-party.py import urllib.requestimport jsonimport sysdef gettoken(corp_id,corp_secret): gettoken_url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=' + corp_id + '&corpsecret=' + corp_secret try: token_file = urllib.request.urlopen(gettoken_url) except urllib.error.HTTPError as e: print(e.code) print(e.read().decode("utf8")) token_data = token_file.read().decode('utf-8') token_json = json.loads(token_data) token_json.keys() token = token_json['access_token'] return tokendef senddata(access_token,notify_str): send_url = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=' + access_token notifydata = notify_str.split("separator") party = notifydata[0] cationtype = notifydata[1] desc = notifydata[2] alias = notifydata[3] address = notifydata[4] state = notifydata[5] datatime = notifydata[6] output = notifydata[7]# content ='[擦汗]Service Notification [擦汗]\n\n类型: ' + cationtype + '\n\n服务名: ' + desc + '\n主机名: ' + alias + '\nIP址: ' + address + '\n状态: ' + state + '\n时间: ' + datatime + '\n摘要:\n' + output + '\n' if cationtype == "RECOVERY": content ='[鼓掌]' + desc + ' is ' + state + '[鼓掌]\n\nIP地址: ' + address + '\n主要用途: ' + alias + '\n服务状态: ' + desc + ' is ' + state + '\n检测时间: ' + datatime + '\n日志摘要: \n' + output + '\n' else: content ='[擦汗]' + desc + ' is ' + state + '[擦汗]\n\nIP地址: ' + address + '\n主要用途: ' + alias + '\n服务状态: ' + desc + ' is ' + state + '\n检测时间: ' + datatime + '\n日志摘要: \n' + output + '\n' send_values = { "toparty":party, "totag":"2", "msgtype":"text", "agentid":"15", "text":{ "content":content }, "safe":"0" } send_data = json.dumps(send_values, ensure_ascii=False).encode(encoding='UTF8') send_request = urllib.request.Request(send_url, send_data) response = urllib.request.urlopen(send_request) msg = response.read() return msgdefault_encoding = 'utf-8'if sys.getdefaultencoding() != default_encoding: reload(sys) sys.setdefaultencoding(default_encoding)notifystr = str(sys.argv[1])corpid = 'wxb616286d28ds01114c9da602'corpsecret = '2nCsNcHxepBCdtgV4U9Lcf-23By1RGzUgh1Zs422tdJpKTQzqjQ1b26IFxP76ydG2rKkchGN6E'accesstoken = gettoken(corpid,corpsecret)msg = senddata(accesstoken,notifystr)print(msg)shell和Python调用企业微信服务号进行报警
#!/bin/bashcorpid="wxd6b3"corpsecret="aJTaPaGjW"access_token=`curl -s "https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=$corpid&corpsecret=$corpsecret" |jq '.access_token' | awk -F'"' '{print $2}'`curl -l -H "Content-type: application/json" -X POST -d '{"touser":"@all","msgtype":"text","toparty":"14","agentid":"14","text":{"content":"测试"} , "safe":"0"}' "Python脚本如下:
# coding:utf-8import sysimport urllib2import timeimport jsonimport requestsreload(sys)sys.setdefaultencoding('utf-8')#title = sys.argv[2] # 位置参数获取title 适用于zabbix#content = sys.argv[3] # 位置参数获取content 适用于zabbixtitle = "title 测试" # 位置参数获取title 适用于zabbixcontent = "content 测试" # 位置参数获取content 适用于zabbixclass Token(object): # 获取token def __init__(self, corpid, corpsecret): self.baseurl = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={0}&corpsecret={1}'.format( corpid, corpsecret) self.expire_time = sys.maxint def get_token(self): if self.expire_time > time.time(): request = urllib2.Request(self.baseurl) response = urllib2.urlopen(request) ret = response.read().strip() ret = json.loads(ret) if 'errcode' in ret.keys(): print >> ret['errmsg'], sys.stderr sys.exit(1) self.expire_time = time.time() + ret['expires_in'] self.access_token = ret['access_token'] return self.access_tokendef send_msg(title, content): # 发送消息 corpid = "" # 填写自己应用的 corpsecret = "" # 填写自己应用的 qs_token = Token(corpid=corpid, corpsecret=corpsecret).get_token() url = "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={0}".format( qs_token) payload = { "touser": "@all", "msgtype": "text", "agentid": "14", "text": { "content": "标题:{0}\n 内容:{1}".format(title, content) }, "safe": "0" } ret = requests.post(url, data=json.dumps(payload, ensure_ascii=False)) print ret.json()if __name__ == '__main__': # print title, content send_msg(title, content)python利用微信订阅号报警
声明:本页内容来源网络,仅供用户参考;我单位不保证亦不表示资料全面及准确无误,也不保证亦不表示这些资料为最新信息,如因任何原因,本网内容或者用户因倚赖本网内容造成任何损失或损害,我单位将不会负任何法律责任。如涉及版权问题,请提交至online#300.cn邮箱联系删除。
开微店,就是使用微商城,也就是利用微信公众号进行商品或者服务买卖的功能。公众号开设微商城,是使用高级功能实现的一种功能,这种功能的实现如果加入了微信支付,就闭环
代发广告的微信公众号,代发广告的微信公众号怎么发广告?现在微信公众号推出了一些新的微信推广功能,我们就可以利用这些扩大我们的广告影响力哦,那代发广告的微信公众号
微信公众号与QQ账号互通,让商家可以在微信公众平台实现和特定群体的文字、图片、语音、视频等全方位的沟通、互动。微信公众号是现今一个新的营销推广方式,利用好公众号
微信公众号页面模板是微信公众号里面的功能哦,如果你开启了微信公众号文章原创功能,那么你是可以去开通页面模板功能的,很多的微信公众号管理员不知道微信公众号页面
微信公众号让商家可以在微信公众平台实现和特定群体的文字、图片、语音、视频等全方位的沟通、互动。微信公众号是现今一个新的营销推广方式,利用好公众号能大大的提升商家