python程序猿：利用微信公众号实现报警功能
Posted huohuohuo1
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了python程序猿：利用微信公众号实现报警功能相关的知识，希望对你有一定的参考价值。
微信公众号共有三种，服务号、订阅号、企业号。它们在获取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。
1 #!/usr/bin/python 2 # encoding: utf-8 3 # -*- coding: utf8 -*- 4 #Python学习群125240963每天更新资料，包括2018最新企业级项目案例，同千人一起交流。 5 6 import os 7 import sqlite3 8 import sys 9 import urllib 10 import urllib2 11 import json 12 import datetime 13 14 # import time 15 16 enable_debug = True 17 18 19 def debug(msg, code=None): 20 if enable_debug: 21 if code is None: 22 print "message: %s" % msg 23 else: 24 print "message: %s, code: %s " % (msg, code) 25 26 27 AUTHOR_MAIL = "[email protected]" 28 29 weixin_qy_CorpID = "your_corpid" 30 weixin_qy_Secret = "your_secret" 31 32 # Build paths inside the project like this: os.path.join(BASE_DIR, ...) 33 BASE_DIR = os.path.dirname(os.path.abspath(__file__)) 34 35 # Database 36 # https://docs.djangoproject.com/en/1.9/ref/settings/#databases 37 38 DATABASES = { 39 ‘default‘: { 40 ‘ENGINE‘: ‘db.backends.sqlite3‘, 41 ‘NAME‘: os.path.join(BASE_DIR, ‘.odbp_db.sqlite3‘), 42 } 43 } 44 45 sqlite3_db_file = str(DATABASES[‘default‘][‘NAME‘]) 46 47 48 def sqlite3_conn(database): 49 try: 50 conn = sqlite3.connect(database) 51 except sqlite3.Error: 52 print >> sys.stderr, """ 53 There was a problem connecting to Database: 54 55 %s 56 57 The error leading to this problem was: 58 59 %s 60 61 It‘s possible that this database is broken or permission denied. 62 63 If you cannot solve this problem yourself, please mail to: 64 65 %s 66 67 """ % (database, sys.exc_value, AUTHOR_MAIL) 68 sys.exit(1) 69 else: 70 return conn 71 72 73 def sqlite3_commit(conn): 74 return conn.commit() 75 76 77 def sqlite3_close(conn): 78 return conn.close() 79 80 81 def sqlite3_execute(database, sql): 82 try: 83 sql_conn = sqlite3_conn(database) 84 sql_cursor = sql_conn.cursor() 85 sql_cursor.execute(sql) 86 sql_conn.commit() 87 sql_conn.close() 88 except sqlite3.Error as e: 89 print e 90 sys.exit(1) 91 92 93 def sqlite3_create_table_token(): 94 sql_conn = sqlite3_conn(sqlite3_db_file) 95 sql_cursor = sql_conn.cursor() 96 sql_cursor.execute(‘‘‘CREATE TABLE "main"."weixin_token" ( 97 "id" INTEGER , 98 "access_token" TEXT, 99 "expires_in" TEXT, 100 "expires_on" TEXT, 101 "is_expired" INTEGER 102 ) 103 ; 104 ‘‘‘) 105 sqlite3_commit(sql_conn) 106 sqlite3_close(sql_conn) 107 108 109 def sqlite3_create_table_account(): 110 sql_conn = sqlite3_conn(sqlite3_db_file) 111 sql_cursor = sql_conn.cursor() 112 sql_cursor.execute(‘‘‘CREATE TABLE "main"."weixin_account" ( 113 "id" INTEGER, 114 "name" TEXT, 115 "corpid" TEXT, 116 "secret" TEXT, 117 "current" INTEGER 118 ) 119 ; 120 ‘‘‘) 121 sqlite3_commit(sql_conn) 122 sqlite3_close(sql_conn) 123 124 125 def sqlite3_create_tables(): 126 print "sqlite3_create_tables" 127 sql_conn = sqlite3_conn(sqlite3_db_file) 128 sql_cursor = sql_conn.cursor() 129 sql_cursor.execute(‘‘‘CREATE TABLE "main"."weixin_token" ( 130 "id" INTEGER , 131 "access_token" TEXT, 132 "expires_in" TEXT, 133 "expires_on" TEXT, 134 "is_expired" INTEGER 135 ) 136 ; 137 ‘‘‘) 138 sql_cursor.execute(‘‘‘CREATE TABLE "main"."weixin_account" ( 139 "id" INTEGER, 140 "name" TEXT, 141 "corpid" TEXT, 142 "secret" TEXT, 143 "current" INTEGER 144 ) 145 ; 146 ‘‘‘) 147 sqlite3_commit(sql_conn) 148 sqlite3_close(sql_conn) 149 150 151 def sqlite3_set_credential(corpid, secret): 152 try: 153 sql_conn = sqlite3_conn(sqlite3_db_file) 154 sql_cursor = sql_conn.cursor() 155 sql_cursor.execute(‘‘‘INSERT INTO "weixin_account" ("id", "name", "corpid", "secret", "current") VALUES 156 (1, 157 ‘odbp‘, 158 ?, 159 ?, 160 1) 161 ‘‘‘, (corpid, secret)) 162 sqlite3_commit(sql_conn) 163 sqlite3_close(sql_conn) 164 except sqlite3.Error: 165 sqlite3_create_table_account() 166 sqlite3_set_credential(corpid, secret) 167 168 169 def sqlite3_set_token(access_token, expires_in, expires_on, is_expired): 170 try: 171 sql_conn = sqlite3_conn(sqlite3_db_file) 172 sql_cursor = sql_conn.cursor() 173 sql_cursor.execute(‘‘‘INSERT INTO "weixin_token" 174 ("id", "access_token", "expires_in", "expires_on", "is_expired") VALUES 175 ( 176 1, 177 ?, 178 ?, 179 ?, 180 ? 181 ) 182 ‘‘‘, (access_token, expires_in, expires_on, is_expired)) 183 sqlite3_commit(sql_conn) 184 sqlite3_close(sql_conn) 185 except sqlite3.Error: 186 sqlite3_create_table_token() 187 sqlite3_set_token(access_token, expires_in, expires_on, is_expired) 188 189 190 def sqlite3_get_credential(): 191 try: 192 sql_conn = sqlite3_conn(sqlite3_db_file) 193 sql_cursor = sql_conn.cursor() 194 credential = sql_cursor.execute(‘‘‘SELECT "corpid", "secret" FROM weixin_account WHERE current == 1;‘‘‘) 195 result = credential.fetchall() 196 sqlite3_close(sql_conn) 197 except sqlite3.Error: 198 sqlite3_set_credential(weixin_qy_CorpID, weixin_qy_Secret) 199 return sqlite3_get_credential() 200 else: 201 if result is not None and len(result) != 0: 202 return result 203 else: 204 print "unrecoverable problem, please alter to %s" % AUTHOR_MAIL 205 sys.exit(1) 206 207 208 def sqlite3_get_token(): 209 try: 210 sql_conn = sqlite3_conn(sqlite3_db_file) 211 sql_cursor = sql_conn.cursor() 212 credential = sql_cursor.execute( 213 ‘‘‘SELECT "access_token", "expires_on" FROM weixin_token WHERE "is_expired" == 1 ;‘‘‘) 214 result = credential.fetchall() 215 sqlite3_close(sql_conn) 216 except sqlite3.Error: 217 info = sys.exc_info() 218 print info[0], ":", info[1] 219 else: 220 if result is not None and len(result) != 0: 221 return result 222 else: 223 # print "unrecoverable problem, please alter to %s" % AUTHOR_MAIL 224 # sys.exit(1) 225 return None 226 227 228 def sqlite3_update_token(access_token, expires_on): 229 sql_conn = sqlite3_conn(sqlite3_db_file) 230 sql_cursor = sql_conn.cursor() 231 sql_cursor.execute(‘‘‘UPDATE "weixin_token" SET 232 access_token=?, 233 expires_on=? 234 WHERE _ROWID_ = 1;‘‘‘, (access_token, expires_on) 235 ) 236 sqlite3_commit(sql_conn) 237 sqlite3_close(sql_conn) 238 239 240 class WeiXinTokenClass(object): 241 def __init__(self): 242 self.__corpid = None 243 self.__corpsecret = None 244 self.__use_persistence = True 245 246 self.__access_token = None 247 self.__expires_in = None 248 self.__expires_on = None 249 self.__is_expired = None 250 251 if self.__use_persistence: 252 self.__corpid = sqlite3_get_credential()[0][0] 253 self.__corpsecret = sqlite3_get_credential()[0][1] 254 else: 255 self.__corpid = weixin_qy_CorpID 256 self.__corpsecret = weixin_qy_Secret 257 258 def __get_token_from_weixin_qy_api(self): 259 parameters = { 260 "corpid": self.__corpid, 261 "corpsecret": self.__corpsecret 262 } 263 url_parameters = urllib.urlencode(parameters) 264 token_url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken?" 265 url = token_url + url_parameters 266 response = urllib2.urlopen(url) 267 result = response.read() 268 token_json = json.loads(result) 269 if token_json[‘access_token‘] is not None: 270 get_time_now = datetime.datetime.now() 271 # TODO(Guodong Ding) token will expired ahead of time or not expired after the time 272 expire_time = get_time_now + datetime.timedelta(seconds=token_json[‘expires_in‘]) 273 token_json[‘expires_on‘] = str(expire_time) 274 self.__access_token = token_json[‘access_token‘] 275 self.__expires_in = token_json[‘expires_in‘] 276 self.__expires_on = token_json[‘expires_on‘] 277 self.__is_expired = 1 278 279 try: 280 token_result_set = sqlite3_get_token() 281 except sqlite3.Error: 282 token_result_set = None 283 if token_result_set is None and len(token_result_set) == 0: 284 sqlite3_set_token(self.__access_token, self.__expires_in, self.__expires_on, self.__is_expired) 285 else: 286 if self.__is_token_expired() is True: 287 sqlite3_update_token(self.__access_token, self.__expires_on) 288 else: 289 debug("pass") 290 return 291 else: 292 if token_json[‘errcode‘] is not None: 293 print "errcode is: %s" % token_json[‘errcode‘] 294 print "errmsg is: %s" % token_json[‘errmsg‘] 295 else: 296 print result 297 298 def __get_token_from_persistence_storage(self): 299 try: 300 token_result_set = sqlite3_get_token() 301 except sqlite3.Error: 302 self.__get_token_from_weixin_qy_api() 303 finally: 304 if token_result_set is None: 305 self.__get_token_from_weixin_qy_api() 306 token_result_set = sqlite3_get_token() 307 access_token = token_result_set[0][0] 308 expire_time = token_result_set[0][1] 309 else: 310 access_token = token_result_set[0][0] 311 expire_time = token_result_set[0][1] 312 expire_time = datetime.datetime.strptime(expire_time, ‘%Y-%m-%d %H:%M:%S.%f‘) 313 now_time = datetime.datetime.now() 314 if now_time < expire_time: 315 # print "The token is %s" % access_token 316 # print "The token will expire on %s" % expire_time 317 return access_token 318 else: 319 self.__get_token_from_weixin_qy_api() 320 return self.__get_token_from_persistence_storage() 321 322 @staticmethod 323 def __is_token_expired(): 324 try: 325 token_result_set = sqlite3_get_token() 326 except sqlite3.Error as e: 327 print e 328 sys.exit(1) 329 expire_time = token_result_set[0][1] 330 expire_time = datetime.datetime.strptime(expire_time, ‘%Y-%m-%d %H:%M:%S.%f‘) 331 now_time = datetime.datetime.now() 332 if now_time < expire_time: 333 return False 334 else: 335 return True 336 337 def get(self): 338 return self.__get_token_from_persistence_storage()
Python实现通过微信企业号发送文本消息的Class
编程要点和调用方法：
支持发送中文，核心语句“payload = json.dumps(self.data, encoding=‘utf-8‘, ensure_ascii=False)”，关键字“python json 中文”
这个Class只有一个公共方法send()。
使用方法：import这个class，然后调用send方法即可，方法参数只需要两个，给谁（多UserID用"|"隔开），内容是什么，例如：
1 import odbp_sendMessage 2 msg = odbp_sendMessage.WeiXinSendMsgClass() 3 msg.send("dingguodong", "Python 大魔王！") 4 5 6 #!/usr/bin/python 7 # encoding: utf-8 8 # -*- coding: utf8 -*- 9 """ 10 Created by PyCharm. 11 File: LinuxBashShellScriptForOps:odbp_sendMessage.py 12 User: Guodong 13 Create Date: 2016/8/12 14 Create Time: 14:49 15 """ 16 17 import odbp_getToken 18 19 20 class WeiXinSendMsgClass(object): 21 def __init__(self): 22 self.access_token = odbp_getToken.WeiXinTokenClass().get() 23 self.to_user = "" 24 self.to_party = "" 25 self.to_tag = "" 26 self.msg_type = "text" 27 self.agent_id = 2 28 self.content = "" 29 self.safe = 0 30 31 self.data = { 32 "touser": self.to_user, 33 "toparty": self.to_party, 34 "totag": self.to_tag, 35 "msgtype": self.msg_type, 36 "agentid": self.agent_id, 37 "text": { 38 "content": self.content 39 }, 40 "safe": self.safe 41 } 42 43 def send(self, to_user, content): 44 if to_user is not None and content is not None: 45 self.data[‘touser‘] = to_user 46 self.data[‘text‘][‘content‘] = content 47 else: 48 print 49 raise RuntimeError 50 import requests 51 import json 52 53 url = "https://qyapi.weixin.qq.com/cgi-bin/message/send" 54 55 querystring = {"access_token": self.access_token} 56 57 payload = json.dumps(self.data, encoding=‘utf-8‘, ensure_ascii=False) 58 59 headers = { 60 ‘content-type‘: "application/json", 61 ‘cache-control‘: "no-cache", 62 } 63 64 response = requests.request("POST", url, data=payload, headers=headers, params=querystring) 65 66 return_content = json.loads(response.content) 67 if return_content["errcode"] == 0 and return_content["errmsg"] == "ok": 68 print "Send successfully! %s " % return_content 69 else: 70 print "Send failed! %s " % return_content
python调用mongodb发送微信企业号
python2.x
注意：data变量里， agent_id为刚刚创建的应用id（可在web页面看到）
toparty即为目标部门，或者可以用touser,totag指定目标账户
比较简单的调用，已实测，可以使用。
1 #coding:utf-8 2 import sys 3 import requests 4 import json 5 from pymongo import MongoClient 6 reload(sys) 7 sys.setdefaultencoding(‘utf-8‘) 8 class Weixin(object): 9 def __init__(self, corp_id, corp_secret): 10 self.token_url = ‘https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=%s&corpsecret=%s‘ %(corp_id, corp_secret) 11 self.send_url = ‘https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=‘ 12 def get_token(self): 13 try: 14 r = requests.get(self.token_url, timeout=10) 15 except Exception as e: 16 print e 17 sys.exit(1) 18 if r.status_code == requests.codes.ok: 19 data = r.json() 20 if data.get(‘errcode‘): 21 print data[‘errmsg‘] 22 sys.exit(1) 23 return data[‘access_token‘] 24 else: 25 print r.status_code 26 sys.exit(1) 27 def send(self,message): 28 url = self.send_url + self.get_token() 29 data = { 30 "touser": "hequan2011", 31 "msgtype": "text", 32 "agentid": "0", 33 "text": { 34 "content": message 35 }, 36 "safe":"0" 37 } 38 send_data = json.dumps(data,ensure_ascii=False) 39 try: 40 r = requests.post(url, send_data) 41 except Exception, e: 42 print e 43 sys.exit(1) 44 if r.status_code == requests.codes.ok: 45 print r.json() 46 else: 47 print r.code 48 sys.exit(1) 49 50 51 corpid = ‘xxxxxxxxxxx‘ 52 corpsecret = ‘xxxxxxxxxxxxxxxxx‘ 53 client = MongoClient(‘mongodb://user:[email protected]:27017/‘) 54 55 db = client.ku 56 collection = db.biao 57 a = [] 58 for data in collection.find(): 59 a.append(data) 60 l = a[0] 61 g = l 62 z = str(g["name"]) 63 z1 = int(g["jg"]) 64 print z 65 66 67 msg = "1:{0} 2:{1} ".format(z,z1) 68 69 w = Weixin(corpid,corpsecret) 70 w.send(msg)
ZABBIX 微信报警 插件
1 #!/usr/bin/env python 2 # -*- coding:utf-8 -*- 3 # __author__ = ‘懒懒的天空‘ 4 5 import requests 6 import sys 7 import json 8 from conf.INIFILES import read_config, write_config 9 import os 10 import datetime 11 from conf.BLog import Log 12 reload(sys) 13 sys.setdefaultencoding(‘utf-8‘) 14 15 16 class WeiXin(object): 17 def __init__(self, corpid, corpsecret): # 初始化的时候需要获取corpid和corpsecret，需要从管理后台获取 18 self.__params = { 19 ‘corpid‘: corpid, 20 ‘corpsecret‘: corpsecret 21 } 22 23 self.url_get_token = ‘https://qyapi.weixin.qq.com/cgi-bin/gettoken‘ 24 self.url_send = ‘https://qyapi.weixin.qq.com/cgi-bin/message/send?‘ 25 26 self.__token = self.__get_token() 27 self.__token_params = { 28 ‘access_token‘: self.__token 29 } 30 31 def __raise_error(self, res): 32 raise Exception(‘error code: %s,error message: %s‘ % (res.json()[‘errcode‘], res.json()[‘errmsg‘])) 33 global senderr 34 sendstatus = False 35 senderr = ‘error code: %s,error message: %s‘ % (res.json()[‘errcode‘], res.json()[‘errmsg‘]) 36 37 def __get_token(self): 38 # print self.url_get_token 39 headers = {‘content-type‘: ‘application/json‘} 40 res = requests.get(self.url_get_token, headers=headers, params=self.__params) 41 42 try: 43 return res.json()[‘access_token‘] 44 except: 45 self.__raise_error(res.content) 46 47 48 def send_message(self, agentid, messages, userid=‘‘, toparty=‘‘): 49 payload = { 50 ‘touser‘: userid, 51 ‘toparty‘: toparty, 52 ‘agentid‘: agentid, 53 "msgtype": "news", 54 "news": messages 55 } 56 headers = {‘content-type‘: ‘application/json‘} 57 data = json.dumps(payload, ensure_ascii=False).encode(‘utf-8‘) 58 params = self.__token_params 59 res = requests.post(self.url_send, headers=headers, params=params, data=data) 60 try: 61 return res.json() 62 except: 63 self.__raise_error(res) 64 65 66 def main(send_to, subject, content): 67 try: 68 global sendstatus 69 global senderr 70 data = ‘‘ 71 messages = {} 72 body = {} 73 config_file_path = get_path() 74 CorpID = read_config(config_file_path, ‘wei‘, "CorpID") 75 CorpSecret = read_config(config_file_path, ‘wei‘, "CorpSecret") 76 agentid = read_config(config_file_path, ‘wei‘, "agentid") 77 web = read_config(config_file_path, ‘wei‘, "web") 78 content = json.loads(content) 79 messages["message_url"] = web 80 body["url"] = web + "history.php?action=showgraph&itemids[]=" + content[u‘监控ID‘] 81 warn_message = ‘‘ 82 if content[u‘当前状态‘] == ‘PROBLEM‘: 83 body["title"] = "服务器故障" 84 warn_message += subject + ‘ ‘ 85 warn_message += ‘详情： ‘ 86 warn_message += ‘告警等级：‘+ content[u‘告警等级‘] + ‘ ‘ 87 warn_message += ‘告警时间：‘+ content[u‘告警时间‘] + ‘ ‘ 88 warn_message += ‘告警地址：‘+ content[u‘告警地址‘] + ‘ ‘ 89 warn_message += ‘持续时间：‘+ content[u‘持续时间‘] + ‘ ‘ 90 warn_message += ‘监控项目：‘+ content[u‘监控项目‘] + ‘ ‘ 91 warn_message += content[u‘告警主机‘] + ‘故障(‘ + content[u‘事件ID‘]+ ‘)‘ 92 else: 93 body["title"] = "服务器恢复" 94 warn_message += subject + ‘ ‘ 95 warn_message += ‘详情： ‘ 96 warn_message += ‘告警等级：‘+ content[u‘告警等级‘] + ‘ ‘ 97 warn_message += ‘恢复时间：‘+ content[u‘恢复时间‘] + ‘ ‘ 98 warn_message += ‘告警地址：‘+ content[u‘告警地址‘] + ‘ ‘ 99 warn_message += ‘持续时间：‘+ content[u‘持续时间‘] + ‘ ‘ 100 warn_message += ‘监控项目：‘+ content[u‘监控项目‘] + ‘ ‘ 101 warn_message += content[u‘告警主机‘] + ‘恢复(‘ + content[u‘事件ID‘]+ ‘)‘ 102 body[‘description‘] = warn_message 103 data = [] 104 data.append(body) 105 messages[‘articles‘] = data 106 wx = WeiXin(CorpID, CorpSecret) 107 data = wx.send_message(toparty=send_to, agentid=agentid, messages=messages) 108 sendstatus = True 109 except Exception, e: 110 senderr = str(e) 111 sendstatus = False 112 logwrite(sendstatus, data) 113 114 115 def get_path(): 116 path = os.path.dirname(os.path.abspath(sys.argv[0])) 117 config_path = path + ‘/config.ini‘ 118 return config_path 119 120 121 def logwrite(sendstatus, content): 122 logpath = ‘/var/log/zabbix/weixin‘ 123 if not sendstatus: 124 content = senderr 125 t = datetime.datetime.now() 126 daytime = t.strftime(‘%Y-%m-%d‘) 127 daylogfile = logpath+‘/‘+str(daytime)+‘.log‘ 128 logger = Log(daylogfile, level="info", is_console=False, mbs=5, count=5) 129 os.system(‘chown zabbix.zabbix {0}‘.format(daylogfile)) 130 logger.info(content) 131 132 if __name__ == "__main__": 133 if len(sys.argv) > 1: 134 send_to = sys.argv[1] 135 subject = sys.argv[2] 136 content = sys.argv[3] 137 logwrite(True, content) 138 main(send_to, subject, content)
python实现微信企业号的文本消息推送
1 #!/usr/bin/python 2 # _*_coding:utf-8 _*_ 3 import urllib2 4 import json 5 import sys 6 reload(sys) 7 sys.setdefaultencoding(‘utf-8‘) 8 def gettoken(corpid, corpsecret): 9 gettoken_url = ‘https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=‘ + corpid + ‘&corpsecret=‘ + corpsecret 10 try: 11 token_file = urllib2.urlopen(gettoken_url) 12 except urllib2.HTTPError as e: 13 print e.code 14 print e.read().decode("utf8") 15 sys.exit() 16 token_data = token_file.read().decode(‘utf-8‘) 17 token_json = json.loads(token_data) 18 token_json.keys() 19 token = token_json[‘access_token‘] 20 return token 21 def senddata(access_token, user, party, agent, subject, content): 22 send_url = ‘https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=‘ + access_token 23 send_values = "{"touser":"" + user + "","toparty":"" + party + "","totag":"","msgtype":"text","agentid":"" + agent + "","text":{"content":"" + subject + " " + content + ""},"safe":"0"}" 24 send_request = urllib2.Request(send_url, send_values) 25 response = json.loads(urllib2.urlopen(send_request).read()) 26 print str(response) 27 28 29 if __name__ == ‘__main__‘: 30 user = str(sys.argv[1]) # 参数1:发送给用户的账号,必须关注企业号,并对企业号有发消息权限 31 party = str(sys.argv[2]) # 参数2:发送给组的id号,必须对企业号有权限 32 agent = str(sys.argv[3]) # 参数3:企业号中的应用id 33 subject = str(sys.argv[4]) # 参数4:标题【消息内容的一部分】 34 content = str(sys.argv[5]) # 参数5:文本具体内容 35 corpid = ‘CorpID‘ # CorpID是企业号的标识 36 corpsecret = ‘corpsecretSecret‘ # corpsecretSecret是管理组凭证密钥 37 try: 38 accesstoken = gettoken(corpid, corpsecret) 39 senddata(accesstoken, user, party, agent, subject, content) 40 except Exception, e: 41 print str(e) + "Error Please Check "corpid" or "corpsecret" Config"
Nagios调用Python程序控制微信公众平台发布报警信息
1 vim Notify-host-by-weixin-party.py 2 3 import urllib.request 4 import json 5 import sys 6 #以上是导入模块 7 #创建获取AccessToken的方法 8 def gettoken(corp_id,corp_secret): 9 gettoken_url = ‘https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=‘ + corp_id + ‘&corpsecret=‘ + corp_secret 10 try: 11 token_file = urllib.request.urlopen(gettoken_url) 12 except urllib.error.HTTPError as e: 13 print(e.code) 14 print(e.read().decode("utf8")) 15 token_data = token_file.read().decode(‘utf-8‘) 16 token_json = json.loads(token_data) 17 token_json.keys() 18 token = token_json[‘access_token‘] 19 return token 20 #这里是发送消息的方法 21 def senddata(access_token,notify_str): 22 send_url = ‘https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=‘ + access_token 23 #我传入的参数是一段字符串每个信息用separator连起来，只要再用字符串的split("separator")方法分开信息就可以了。 24 notifydata = notify_str.split("separator") 25 party = notifydata[0] 26 cationtype = notifydata[1] 27 name = notifydata[2] 28 state = notifydata[3] 29 address = notifydata[4] 30 output = notifydata[5] 31 datatime = notifydata[6] 32 # content = ‘[擦汗]Host Notification[擦汗] 类型: ‘ + cationtype + ‘ 主机名: ‘ + name + ‘ 状态: ‘ + state + ‘ IP地址: ‘ + address + ‘ 摘要: ‘ + output + ‘ 时间: ‘ + datatime + ‘ ‘ 33 if cationtype == "RECOVERY": 34 content = ‘[嘘]‘ + address + ‘ is ‘ + state + ‘[嘘] IP地址: ‘ + address + ‘ 主要用途: ‘ + name + ‘ 当前状态: ‘ + state + ‘ 日志摘要: ‘ + output + ‘ 检测时间: ‘ + datatime + ‘ ‘ 35 else: 36 content = ‘[擦汗]‘ + address + ‘ is ‘ + state + ‘[擦汗] IP地址: ‘ + address + ‘ 主要用途: ‘ + name + ‘ 当前状态: ‘ + state + ‘ 日志摘要: ‘ + output + ‘ 检测时间: ‘ + datatime + ‘ ‘ 37 send_values = { 38 "toparty":party, 39 "totag":"2", 40 "msgtype":"text", 41 "agentid":"15", 42 "text":{ 43 "content":content 44 }, 45 "safe":"0" 46 } 47 send_data = json.dumps(send_values, ensure_ascii=False).encode(encoding=‘UTF8‘) 48 #设置为非ascii解析，使其支持中文 49 send_request = urllib.request.Request(send_url, send_data) 50 response = urllib.request.urlopen(send_request) 51 #这个是返回微信公共平台的信息，调试时比较有用 52 msg = response.read() 53 return msg 54 default_encoding = ‘utf-8‘ 55 if sys.getdefaultencoding() != default_encoding: 56 reload(sys) 57 sys.setdefaultencoding(default_encoding) 58 #我编辑的脚本是要获取nagios传入的一段参数的（字符串），下面这条代码是获取执行脚本后获取的第一个参数（经测试nagios只能传入一个参进python，所以把所有包括用户名跟报警主机报警信息放进一个字符串里） 59 notifystr = str(sys.argv[1]) 60 corpid = ‘wxb6162862801114c9da602‘ 61 corpsecret = ‘2nCsNcHxepBCV4U9Lcf-23By1RGzU1Zs422tdJpKTQzqjQ1b26IFxP76ydG2rKkchGN6E‘ 62 accesstoken = gettoken(corpid,corpsecret) 63 msg = senddata(accesstoken,notifystr) 64 print(msg) 65 [[email protected] python]# vim Notify-service-by-weixin-party.py 66 import urllib.request 67 import json 68 import sys 69 70 def gettoken(corp_id,corp_secret): 71 gettoken_url = ‘https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=‘ + corp_id + ‘&corpsecret=‘ + corp_secret 72 try: 73 token_file = urllib.request.urlopen(gettoken_url) 74 except urllib.error.HTTPError as e: 75 print(e.code) 76 print(e.read().decode("utf8")) 77 token_data = token_file.read().decode(‘utf-8‘) 78 token_json = json.loads(token_data) 79 token_json.keys() 80 token = token_json[‘access_token‘] 81 return token 82 83 def senddata(access_token,notify_str): 84 send_url = ‘https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=‘ + access_token 85 notifydata = notify_str.split("separator") 86 party = notifydata[0] 87 cationtype = notifydata[1] 88 desc = notifydata[2] 89 alias = notifydata[3] 90 address = notifydata[4] 91 state = notifydata[5] 92 datatime = notifydata[6] 93 output = notifydata[7] 94 # content =‘[擦汗]Service Notification [擦汗] 类型: ‘ + cationtype + ‘ 服务名: ‘ + desc + ‘ 主机名: ‘ + alias + ‘ IP址: ‘ + address + ‘ 状态: ‘ + state + ‘ 时间: ‘ + datatime + ‘ 摘要: ‘ + output + ‘ ‘ 95 if cationtype == "RECOVERY": 96 content =‘[鼓掌]‘ + desc + ‘ is ‘ + state + ‘[鼓掌] IP地址: ‘ + address + ‘ 主要用途: ‘ + alias + ‘ 服务状态: ‘ + desc + ‘ is ‘ + state + ‘ 检测时间: ‘ + datatime + ‘ 日志摘要: ‘ + output + ‘ ‘ 97 else: 98 content =‘[擦汗]‘ + desc + ‘ is ‘ + state + ‘[擦汗] IP地址: ‘ + address + ‘ 主要用途: ‘ + alias + ‘ 服务状态: ‘ + desc + ‘ is ‘ + state + ‘ 检测时间: ‘ + datatime + ‘ 日志摘要: ‘ + output + ‘ ‘ 99 send_values = { 100 "toparty":party, 101 "totag":"2", 102 "msgtype":"text", 103 "agentid":"15", 104 "text":{ 105 "content":content 106 }, 107 "safe":"0" 108 } 109 send_data = json.dumps(send_values, ensure_ascii=False).encode(encoding=‘UTF8‘) 110 send_request = urllib.request.Request(send_url, send_data) 111 response = urllib.request.urlopen(send_request) 112 msg = response.read() 113 return msg 114 115 116 default_encoding = ‘utf-8‘ 117 if sys.getdefaultencoding() != default_encoding: 118 reload(sys) 119 sys.setdefaultencoding(default_encoding) 120 notifystr = str(sys.argv[1]) 121 corpid = ‘wxb616286d28ds01114c9da602‘ 122 corpsecret = ‘2nCsNcHxepBCdtgV4U9Lcf-23By1RGzUgh1Zs422tdJpKTQzqjQ1b26IFxP76ydG2rKkchGN6E‘ 123 accesstoken = gettoken(corpid,corpsecret) 124 msg = senddata(accesstoken,notifystr) 125 print(msg)
shell和Python调用企业微信服务号进行报警
1 #!/bin/bash 2 corpid="wxd6b3" 3 corpsecret="aJTaPaGjW" 4 access_token=`curl -s "https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=$corpid&corpsecret=$corpsecret" |jq ‘.access_token‘ | awk -F‘"‘ ‘{print $2}‘` 5 6 curl -l -H "Content-type: application/json" -X POST -d ‘{"touser":"@all","msgtype":"text","toparty":"14","agentid":"14","text":{"content":"测试"} , "safe":"0"}‘ "
Python脚本如下：
1 # coding:utf-8 2 import sys 3 import urllib2 4 import time 5 import json 6 import requests 7 reload(sys) 8 sys.setdefaultencoding(‘utf-8‘) 9 #title = sys.argv[2] # 位置参数获取title 适用于zabbix 10 #content = sys.argv[3] # 位置参数获取content 适用于zabbix 11 title = "title 测试" # 位置参数获取title 适用于zabbix 12 content = "content 测试" # 位置参数获取content 适用于zabbix 13 class Token(object): 14 # 获取token 15 def __init__(self, corpid, corpsecret): 16 self.baseurl = ‘https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={0}&corpsecret={1}‘.format( 17 corpid, corpsecret) 18 self.expire_time = sys.maxint 19 def get_token(self): 20 if self.expire_time > time.time(): 21 request = urllib2.Request(self.baseurl) 22 response = urllib2.urlopen(request) 23 ret = response.read().strip() 24 ret = json.loads(ret) 25 if ‘errcode‘ in ret.keys(): 26 print >> ret[‘errmsg‘], sys.stderr 27 sys.exit(1) 28 self.expire_time = time.time() + ret[‘expires_in‘] 29 self.access_token = ret[‘access_token‘] 30 return self.access_token 31 def send_msg(title, content): 32 # 发送消息 33 corpid = "" # 填写自己应用的 34 corpsecret = "" # 填写自己应用的 35 qs_token = Token(corpid=corpid, corpsecret=corpsecret).get_token() 36 url = "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={0}".format( 37 qs_token) 38 payload = { 39 "touser": "@all", 40 "msgtype": "text", 41 "agentid": "14", 42 "text": { 43 "content": "标题:{0} 内容:{1}".format(title, content) 44 }, 45 "safe": "0" 46 } 47 ret = requests.post(url, data=json.dumps(payload, ensure_ascii=False)) 48 print ret.json() 49 if __name__ == ‘__main__‘: 50 # print title, content 51 send_msg(title, content)
python利用微信订阅号报警
1 # coding=utf-8 2 import urllib 3 import urllib2 4 import cookielib 5 import json 6 import sys 7 data={‘username‘:‘yaokuaile-99‘, 8 ‘pwd‘:‘f4bb2d8abe7a799ad62495a912ae3363‘, 9 ‘imgcode‘:‘‘, 10 ‘f‘:‘json‘ 11 } 12 cj=cookielib.LWPCookieJar() 13 opener=urllib2.build_opener(urllib2.HTTPCookieProcessor(cj)) 14 urllib2.install_opener(opener) 15 def getToken(): 16 headers = {‘Accept‘: ‘application/json, text/javascript, */*; q=0.01‘, 17 ‘Accept-Encoding‘: ‘gzip,deflate,sdch‘, 18 ‘Accept-Language‘: ‘zh-CN,zh;q=0.8‘, 19 ‘Connection‘: ‘keep-alive‘, 20 ‘Content-Type‘: ‘application/x-www-form-urlencoded; charset=UTF-8‘, 21 ‘Content-Length‘: ‘74‘, 22 ‘Content-Type‘: ‘application/x-www-form-urlencoded; charset=UTF-8‘, 23 ‘Host‘: ‘mp.weixin.qq.com‘, 24 ‘Origin‘: ‘https://mp.weixin.qq.com‘, 25 ‘Referer‘: ‘https://mp.weixin.qq.com/cgi-bin/loginpage?t=wxm2-login&lang=zh_CN‘, 26 ‘User-Agent‘: ‘Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/33.0.1750.154 Safari/537.36‘, 27 ‘X-Requested-With‘: ‘XMLHttpRequest‘, 28 } 29 req = urllib2.Request(‘https://mp.weixin.qq.com/cgi-bin/login?lang=zh_CN‘,urllib.urlencode(data),headers) 30 ret = urllib2.urlopen(req) 31 retread= ret.read() 32 token = json.loads(retread) 33 token=token[‘redirect_url‘][44:] 34 return token 35 ### send msg 36 def sendWeixin(msg,token,tofakeid): 37 msg = msg 38 token = token 39 data1 = {‘type‘:‘1‘,‘content‘:msg,‘imgcode‘:‘‘,‘imgcode‘:‘‘,‘tofakeid‘:tofakeid,‘f‘:‘json‘,‘token‘:token,‘ajax‘:‘1‘} 40 headers = {‘Accept‘:‘*/*‘, 41 ‘Accept-Encoding‘: ‘gzip,deflate,sdch‘, 42 ‘Accept-Language‘: ‘zh-CN,zh;q=0.8‘, 43 ‘Connection‘: ‘keep-alive‘, 44 ‘Content-Type‘: ‘application/x-www-form-urlencoded; charset=UTF-8‘, 45 ‘Host‘: ‘mp.weixin.qq.com‘, 46 ‘Origin‘: ‘https://mp.weixin.qq.com‘, 47 ‘Referer‘: ‘https://mp.weixin.qq.com/cgi-bin/singlemsgpage?msgid=&source=&count=20&t=wxm-singlechat&fromfakeid=150890&token=%s&lang=zh_CN‘, 48 ‘User-Agent‘: ‘Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/33.0.1750.154 Safari/537.36‘, 49 ‘X-Requested-With‘:‘XMLHttpRequest‘, 50 } 51 req2 = urllib2.Request(‘https://mp.weixin.qq.com/cgi-bin/singlesend?t=ajax-response&f=json&lang=zh_CN‘,urllib.urlencode(data1),headers) 52 ret2=urllib2.urlopen(req2) 53 54 if __name__==‘__main__‘: 55 ‘‘‘ 56 useage: ./send_wx.py msg 57 ‘‘‘ 58 token = getToken() 59 msg = sys.argv[1:] 60 msg = ‘ ‘.join(msg) 61 tofakeid = ‘2443746922‘ 62 sendWeixin(msg, token, tofakeid)
1 #!/usr/bin/env python 2 #-*- coding:utf-8 -*- 3 4 import urllib2,json 5 import datetime,time 6 from config import * 7 import sys 8 reload(sys) 9 sys.setdefaultencoding("utf-8") 10 11 12 13 class WechatPush(): 14 15 def __init__(self,appid,secrect,file_name): 16 # 传入appid 17 self.appid = appid 18 # 传入密码 19 self.secrect = secrect 20 # 传入记录token和过期时间的文件名 21 self.file_name=file_name 22 23 def build_timestamp(self,interval): 24 # 传入时间间隔,得到指定interval后的时间 格式为"2015-07-01 14:41:40" 25 now = datetime.datetime.now() 26 delta = datetime.timedelta(seconds=interval) 27 now_interval=now + delta 28 return now_interval.strftime(‘%Y-%m-%d %H:%M:%S‘) 29 30 31 def check_token_expires(self): 32 # 判断token是否过期 33 with open(self.file_name,‘r‘) as f: 34 line=f.read() 35 if len(line)>0: 36 expires_time=line.split(",")[1] 37 token=line.split(",")[0] 38 else: 39 return "","true" 40 curr_time=time.strftime(‘%Y-%m-%d %H:%M:%S‘) 41 # 如果过期返回false 42 if curr_time>expires_time: 43 return token,"false" 44 # 没过期返回true 45 else: 46 return token,"true" 47 48 def getToken(self): 49 # 获取accessToken 50 url = ‘https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=‘+self.appid + "&secret="+self.secrect 51 try: 52 f = urllib2.urlopen(url) 53 s = f.read() 54 # 读取json数据 55 j = json.loads(s) 56 j.keys() 57 # 从json中获取token 58 token = j[‘access_token‘] 59 # 从json中获取过期时长 60 expires_in =j[‘expires_in‘] 61 # 将得到的过期时长减去300秒然后与当前时间做相加计算然后写入到过期文件 62 write_expires=self.build_timestamp(int(expires_in-300)) 63 content="%s,%s" % (token,write_expires) 64 with open(self.file_name,‘w‘) as f: 65 f.write(content) 66 except Exception,e: 67 print e 68 69 return token 70 71 72 def post_data(self,url,para_dct): 73 """触发post请求微信发送最终的模板消息""" 74 para_data = para_dct 75 f = urllib2.urlopen(url,para_data) 76 content = f.read() 77 return content 78 79 80 81 def do_push(self,touser,template_id,url,topcolor,data): 82 ‘‘‘推送消息 ‘‘‘ 83 #获取存入到过期文件中的token,同时判断是否过期 84 token,if_token_expires=self.check_token_expires() 85 #如果过期了就重新获取token 86 if if_token_expires=="false": 87 token=self.getToken() 88 # 背景色设置,貌似不生效 89 if topcolor.strip()==‘‘: 90 topcolor = "#7B68EE" 91 #最红post的求情数据 92 dict_arr = {‘touser‘: touser, ‘template_id‘:template_id, ‘url‘:url, ‘topcolor‘:topcolor,‘data‘:data} 93 json_template = json.dumps(dict_arr) 94 requst_url = "https://api.weixin.qq.com/cgi-bin/message/template/send?access_token="+token 95 content = self.post_data(requst_url,json_template) 96 #读取json数据 97 j = json.loads(content) 98 j.keys() 99 errcode = j[‘errcode‘] 100 errmsg = j[‘errmsg‘] 101 #print errmsg 102 103 104 105 if __name__ == "__main__": 106 107 def alarm(title,hostname,timestap,level,message,state,tail): 108 """报警函数""" 109 color="#FF0000" 110 data={"first":{"value":title},"keyword1":{"value":hostname,"color":color},"keyword2":{"value":timestap,"color":color},"keyword3":{"value":level,"color":color},"keyword4":{"value":message,"color":color},"keyword5":{"value":state,"color":color},"remark":{"value":tail}} 111 return data 112 113 def recover(title,message,alarm_time,recover_time,continue_time,tail): 114 """恢复函数""" 115 re_color="#228B22" 116 data={"first":{"value":title},"content":{"value":message,"color":re_color},"occurtime":{"value":alarm_time,"color":re_color},"recovertime":{"value":recover_time,"color":re_color},"lasttime":{"value":continue_time,"color":re_color},"remark":{"value":tail}} 117 return data 118 119 120 121 122 # data=alarm("测试的报警消息","8.8.8.8",time.ctime(),"最高级别","然并卵","挂了","大傻路赶紧处理") 123 124 # 实例化类 125 webchart=WechatPush(appid,secrect,file_name) 126 url="http://www.xiaoniu88.com" 127 128 129 print len(sys.argv) 130 # 发送报警消息 131 if len(sys.argv) == 9: 132 title=sys.argv[1] 133 hostname=sys.argv[2] 134 timestap=sys.argv[3] 135 level=sys.argv[4] 136 message=sys.argv[5] 137 state=sys.argv[6] 138 tail=sys.argv[7] 139 140 print "sys.argv[1]"+sys.argv[1] 141 print "sys.argv[2]"+sys.argv[2] 142 print "sys.argv[3]"+sys.argv[3] 143 print "sys.argv[4]"+sys.argv[4] 144 print "sys.argv[5]"+sys.argv[5] 145 print "sys.argv[6]"+sys.argv[6] 146 print "sys.argv[7]"+sys.argv[7] 147 print "sys.argv[8]"+sys.argv[8] 148 149 with open("/etc/zabbix/moniter_scripts/test.log",‘a+‘) as f: 150 f.write(title+" ") 151 f.write(hostname+" ") 152 f.write(timestap+" ") 153 f.write(level+" ") 154 f.write(message+" ") 155 f.write(state+" ") 156 f.write(tail+" ") 157 f.write("%s_%s" % ("group",sys.argv[8])+" ") 158 159 160 data=alarm(title,hostname,timestap,level,message,state,tail) 161 group_name="%s_%s" % ("group",sys.argv[8]) 162 for touser in eval("%s_%s" % ("group",sys.argv[8])): 163 webchart.do_push(touser,alarm_id,url,"",data) 164 165 for touser in group_super: 166 webchart.do_push(touser,alarm_id,url,"",data) 167 #发送恢复消息 168 elif len(sys.argv) == 8: 169 title=sys.argv[1] 170 message=sys.argv[2] 171 alarm_time=sys.argv[3] 172 recover_time=sys.argv[4] 173 continue_time=sys.argv[5] 174 tail=sys.argv[6] 175 176 177 print "sys.argv[1]"+sys.argv[1] 178 print "sys.argv[2]"+sys.argv[2] 179 print "sys.argv[3]"+sys.argv[3] 180 print "sys.argv[4]"+sys.argv[4] 181 print "sys.argv[5]"+sys.argv[5] 182 print "sys.argv[6]"+sys.argv[6] 183 print "sys.argv[7]"+sys.argv[7] 184 185 data=recover(title,message,alarm_time,recover_time,continue_time,tail) 186 187 for touser in eval("%s_%s" % ("group",sys.argv[7])): 188 webchart.do_push(touser,recover_id,url,"",data) 189 190 for touser in group_super: 191 webchart.do_push(touser,recover_id,url,"",data)
zabbix使用微信报警python脚本
1 #!/usr/bin/python 2 # coding: utf-8 3 #python2将zabbix报警信息发送到微信。 4 #by linwangyi #2016-01-18 5 import urllib,urllib2 6 import json 7 import sys 8 9 def gettoken(corpid,corpsecret): 10 gettoken_url = ‘https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=‘ + corpid + ‘&corpsecret=‘ + corpsecret 11 try: 12 token_file = urllib2.urlopen(gettoken_url) 13 except urllib2.HTTPError as e: 14 print e.code 15 print e.read().decode("utf8") 16 sys.exit() 17 token_data = token_file.read().decode(‘utf-8‘) 18 token_json = json.loads(token_data) 19 token_json.keys() 20 token = token_json[‘access_token‘] 21 return token 22 23 def senddata(access_token,user,content): 24 send_url = ‘https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=‘ + access_token 25 send_values = { 26 "touser":user, #企业号中的用户帐号，在zabbix用户Media中配置，如果配置不正常，将按部门发送。 27 "toparty":"1", #企业号中的部门id 28 "msgtype":"text", #企业号中的应用id，消息类型。 29 "agentid":"1", 30 "text":{ 31 "content":content 32 }, 33 "safe":"0" 34 } 35 send_data = json.dumps(send_values, ensure_ascii=False) 36 send_request = urllib2.Request(send_url, send_data) 37 response = json.loads(urllib2.urlopen(send_request).read()) 38 print str(response) 39 40 if __name__ == ‘__main__‘: 41 user = str(sys.argv[1]) #zabbix传过来的第一个参数 42 content = str(sys.argv[3]) #zabbix传过来的第三个参数 43 corpid = ‘XXXX‘ #CorpID是企业号的标识 44 corpsecret = ‘XXXXX‘ #corpsecretSecret是管理组凭证密钥 45 accesstoken = gettoken(corpid,corpsecret) 46 senddata(accesstoken,user,content) 47 48 49 50 51 52 #!/usr/bin/python3 53 # coding: utf-8 54 #python3将zabbix报警信息发送到微信。 55 #by http://sunday208.blog.51cto.com/ #2016-01-18 56 import urllib.request 57 import json 58 import sys 59 60 def gettoken(corp_id,corp_secret): 61 gettoken_url = ‘https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=‘ + corp_id + ‘&corpsecret=‘ + corp_secret 62 try: 63 token_file = urllib.request.urlopen(gettoken_url) 64 except urllib.error.HTTPError as e: 65 print(e.code) 66 print(e.read().decode("utf8")) 67 token_data = token_file.read().decode(‘utf-8‘) 68 token_json = json.loads(token_data) 69 token_json.keys() 70 token = token_json[‘access_token‘] 71 return token 72 73 def senddata(access_token,user,content): 74 try: 75 send_url = ‘https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=‘ + access_token 76 send_values = { 77 "touser":user, #企业号中的用户帐号，在zabbix用户Media中配置，如果配置不正常，将按部门发送。 78 "toparty":"1", #企业号中的部门id 79 "msgtype":"text", 80 "agentid":"1", #企业号中的应用id，消息类型。 81 "text":{ 82 "content":content 83 }, 84 "safe":"0" 85 } 86 send_data = json.dumps(send_values, ensure_ascii=False).encode(encoding=‘UTF8‘) 87 send_request = urllib.request.Request(send_url, send_data) 88 response = urllib.request.urlopen(send_request) 89 msg = response.read() 90 print("returned value : " + str(msg)) 91 except: 92 print("returned value : " + str(msg)) 93 94 default_encoding = ‘utf-8‘ 95 if sys.getdefaultencoding() != default_encoding: 96 reload(sys) 97 sys.setdefaultencoding(default_encoding) 98 user = str(sys.argv[1]) #zabbix传过来的第一个参数 99 content = str(sys.argv[3]) #zabbix传过来的第三个参数 100 corpid = ‘XXXX‘ #CorpID是企业号的标识 101 corpsecret = ‘XXXXX‘ #corpsecretSecret是管理组凭证密钥 102 accesstoken = gettoken(corpid,corpsecret) 103 senddata(accesstoken,user,content)
以上是关于python程序猿：利用微信公众号实现报警功能的主要内容，如果未能解决你的问题，请参考以下文章