其中訂閱號比較坑,它的AccessToken是需定時刷新,重復(fù)獲取將導(dǎo)致上次獲取的AccessToken失效。
而企業(yè)號就比較好,AccessToken有效期同樣為7200秒,但有效期內(nèi)重復(fù)獲取返回相同結(jié)果。
為兼容這兩種方式,因此按照訂閱號的方式處理。
處理辦法與接口文檔中的要求相同:
為了保密appsecrect,第三方需要一個access_token獲取和刷新的中控服務(wù)器。
而其他業(yè)務(wù)邏輯服務(wù)器所使用的access_token均來自于該中控服務(wù)器,不應(yīng)該各自去刷新,否則會造成access_token覆蓋而影響業(yè)務(wù)。
下面的代碼以企業(yè)號為例,將access_token儲存在sqlite3數(shù)據(jù)庫中,相比儲存在文本中,放在數(shù)
據(jù)庫里,可以為后期存放其他數(shù)據(jù)提供向后兼容。如果放在文本中,則不如放在數(shù)據(jù)庫中靈活。
設(shè)計思路和使用方法:
自動創(chuàng)建sqlite3數(shù)據(jù)庫,包括表結(jié)構(gòu)和數(shù)據(jù),并能在數(shù)據(jù)庫表結(jié)構(gòu)不存在或者數(shù)據(jù)不存在或遭刪除的情況下,創(chuàng)建新的可用的數(shù)據(jù)
盡可能的保證Class中每一個可執(zhí)行的函數(shù)單獨調(diào)用都能成功。
Class中只將真正能被用到的方法和變量設(shè)置為public的。
使用時只需要修改此文件中的weixin_qy_CorpID和weixin_qy_Secret改成自己的,并import此文件,使
用WeiXinTokenClass().get()方法即可得到access_token。
#!/usr/bin/python # encoding: utf-8 # -*- coding: utf8 -*- import os import sqlite3 import sys import urllib import urllib2 import json import datetime # import time enable_debug = True def debug(msg, code=None): if enable_debug: if code is None: print "message: %s" % msg else: print "message: %s, code: %s " % (msg, code) AUTHOR_MAIL = "uberurey_ups@163.com" weixin_qy_CorpID = "your_corpid" weixin_qy_Secret = "your_secret" # Build paths inside the project like this: os.path.join(BASE_DIR, ...) BASE_DIR = os.path.dirname(os.path.abspath(__file__)) # Database # https://docs.djangoproject.com/en/1.9/ref/settings/#databases DATABASES = { 'default': { 'ENGINE': 'db.backends.sqlite3', 'NAME': os.path.join(BASE_DIR, '.odbp_db.sqlite3'), } } sqlite3_db_file = str(DATABASES['default']['NAME']) def sqlite3_conn(database): try: conn = sqlite3.connect(database) except sqlite3.Error: print >> sys.stderr, """ There was a problem connecting to Database: %s The error leading to this problem was: %s It's possible that this database is broken or permission denied. If you cannot solve this problem yourself, please mail to: %s """ % (database, sys.exc_value, AUTHOR_MAIL) sys.exit(1) else: return conn def sqlite3_commit(conn): return conn.commit() def sqlite3_close(conn): return conn.close() def sqlite3_execute(database, sql): try: sql_conn = sqlite3_conn(database) sql_cursor = sql_conn.cursor() sql_cursor.execute(sql) sql_conn.commit() sql_conn.close() except sqlite3.Error as e: print e sys.exit(1) def sqlite3_create_table_token(): sql_conn = sqlite3_conn(sqlite3_db_file) sql_cursor = sql_conn.cursor() sql_cursor.execute('''CREATE TABLE "main"."weixin_token" ( "id" INTEGER , "access_token" TEXT, "expires_in" TEXT, "expires_on" TEXT, "is_expired" INTEGER ) ; ''') sqlite3_commit(sql_conn) sqlite3_close(sql_conn) def sqlite3_create_table_account(): sql_conn = sqlite3_conn(sqlite3_db_file) sql_cursor = sql_conn.cursor() sql_cursor.execute('''CREATE TABLE "main"."weixin_account" ( "id" INTEGER, "name" TEXT, "corpid" TEXT, "secret" TEXT, "current" INTEGER ) ; ''') sqlite3_commit(sql_conn) sqlite3_close(sql_conn) def sqlite3_create_tables(): print "sqlite3_create_tables" sql_conn = sqlite3_conn(sqlite3_db_file) sql_cursor = sql_conn.cursor() sql_cursor.execute('''CREATE TABLE "main"."weixin_token" ( "id" INTEGER , "access_token" TEXT, "expires_in" TEXT, "expires_on" TEXT, "is_expired" INTEGER ) ; ''') sql_cursor.execute('''CREATE TABLE "main"."weixin_account" ( "id" INTEGER, "name" TEXT, "corpid" TEXT, "secret" TEXT, "current" INTEGER ) ; ''') sqlite3_commit(sql_conn) sqlite3_close(sql_conn) def sqlite3_set_credential(corpid, secret): try: sql_conn = sqlite3_conn(sqlite3_db_file) sql_cursor = sql_conn.cursor() sql_cursor.execute('''INSERT INTO "weixin_account" ("id", "name", "corpid", "secret", "current") VALUES (1, 'odbp', ?, ?, 1) ''', (corpid, secret)) sqlite3_commit(sql_conn) sqlite3_close(sql_conn) except sqlite3.Error: sqlite3_create_table_account() sqlite3_set_credential(corpid, secret) def sqlite3_set_token(access_token, expires_in, expires_on, is_expired): try: sql_conn = sqlite3_conn(sqlite3_db_file) sql_cursor = sql_conn.cursor() sql_cursor.execute('''INSERT INTO "weixin_token" ("id", "access_token", "expires_in", "expires_on", "is_expired") VALUES ( 1, ?, ?, ?, ? ) ''', (access_token, expires_in, expires_on, is_expired)) sqlite3_commit(sql_conn) sqlite3_close(sql_conn) except sqlite3.Error: sqlite3_create_table_token() sqlite3_set_token(access_token, expires_in, expires_on, is_expired) def sqlite3_get_credential(): try: sql_conn = sqlite3_conn(sqlite3_db_file) sql_cursor = sql_conn.cursor() credential = sql_cursor.execute('''SELECT "corpid", "secret" FROM weixin_account WHERE current == 1;''') result = credential.fetchall() sqlite3_close(sql_conn) except sqlite3.Error: sqlite3_set_credential(weixin_qy_CorpID, weixin_qy_Secret) return sqlite3_get_credential() else: if result is not None and len(result) != 0: return result else: print "unrecoverable problem, please alter to %s" % AUTHOR_MAIL sys.exit(1) def sqlite3_get_token(): try: sql_conn = sqlite3_conn(sqlite3_db_file) sql_cursor = sql_conn.cursor() credential = sql_cursor.execute( '''SELECT "access_token", "expires_on" FROM weixin_token WHERE "is_expired" == 1 ;''') result = credential.fetchall() sqlite3_close(sql_conn) except sqlite3.Error: info = sys.exc_info() print info[0], ":", info[1] else: if result is not None and len(result) != 0: return result else: # print "unrecoverable problem, please alter to %s" % AUTHOR_MAIL # sys.exit(1) return None def sqlite3_update_token(access_token, expires_on): sql_conn = sqlite3_conn(sqlite3_db_file) sql_cursor = sql_conn.cursor() sql_cursor.execute('''UPDATE "weixin_token" SET access_token=?, expires_on=? WHERE _ROWID_ = 1;''', (access_token, expires_on) ) sqlite3_commit(sql_conn) sqlite3_close(sql_conn) class WeiXinTokenClass(object): def __init__(self): self.__corpid = None self.__corpsecret = None self.__use_persistence = True self.__access_token = None self.__expires_in = None self.__expires_on = None self.__is_expired = None if self.__use_persistence: self.__corpid = sqlite3_get_credential()[0][0] self.__corpsecret = sqlite3_get_credential()[0][1] else: self.__corpid = weixin_qy_CorpID self.__corpsecret = weixin_qy_Secret def __get_token_from_weixin_qy_api(self): parameters = { "corpid": self.__corpid, "corpsecret": self.__corpsecret } url_parameters = urllib.urlencode(parameters) token_url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken?" url = token_url + url_parameters response = urllib2.urlopen(url) result = response.read() token_json = json.loads(result) if token_json['access_token'] is not None: get_time_now = datetime.datetime.now() # TODO(Guodong Ding) token will expired ahead of time or not expired after the time expire_time = get_time_now + datetime.timedelta(seconds=token_json['expires_in']) token_json['expires_on'] = str(expire_time) self.__access_token = token_json['access_token'] self.__expires_in = token_json['expires_in'] self.__expires_on = token_json['expires_on'] self.__is_expired = 1 try: token_result_set = sqlite3_get_token() except sqlite3.Error: token_result_set = None if token_result_set is None and len(token_result_set) == 0: sqlite3_set_token(self.__access_token, self.__expires_in, self.__expires_on, self.__is_expired) else: if self.__is_token_expired() is True: sqlite3_update_token(self.__access_token, self.__expires_on) else: debug("pass") return else: if token_json['errcode'] is not None: print "errcode is: %s" % token_json['errcode'] print "errmsg is: %s" % token_json['errmsg'] else: print result def __get_token_from_persistence_storage(self): try: token_result_set = sqlite3_get_token() except sqlite3.Error: self.__get_token_from_weixin_qy_api() finally: if token_result_set is None: self.__get_token_from_weixin_qy_api() token_result_set = sqlite3_get_token() access_token = token_result_set[0][0] expire_time = token_result_set[0][1] else: access_token = token_result_set[0][0] expire_time = token_result_set[0][1] expire_time = datetime.datetime.strptime(expire_time, '%Y-%m-%d %H:%M:%S.%f') now_time = datetime.datetime.now() if now_time < expire_time: # print "The token is %s" % access_token # print "The token will expire on %s" % expire_time return access_token else: self.__get_token_from_weixin_qy_api() return self.__get_token_from_persistence_storage() @staticmethod def __is_token_expired(): try: token_result_set = sqlite3_get_token() except sqlite3.Error as e: print e sys.exit(1) expire_time = token_result_set[0][1] expire_time = datetime.datetime.strptime(expire_time, '%Y-%m-%d %H:%M:%S.%f') now_time = datetime.datetime.now() if now_time < expire_time: return False else: return True def get(self): return self.__get_token_from_persistence_storage() Python實現(xiàn)通過微信企業(yè)號發(fā)送文本消息的Class 編程要點和調(diào)用方法: 支持發(fā)送中文,核心語句“payload = json.dumps(self.data, encoding='utf-8', ensure_ascii=False)”,關(guān)鍵字“python json 中文” 這個Class只有一個公共方法send()。 使用方法:import這個class,然后調(diào)用send方法即可,方法參數(shù)只需要兩個,給誰(多UserID用"|"隔開),內(nèi)容是什么,例如: import odbp_sendMessage msg = odbp_sendMessage.WeiXinSendMsgClass() msg.send("dingguodong", "Python 大魔王!") #!/usr/bin/python # encoding: utf-8 # -*- coding: utf8 -*- """ Created by PyCharm. File: LinuxBashShellScriptForOps:odbp_sendMessage.py User: Guodong Create Date: 2016/8/12 Create Time: 14:49 """ import odbp_getToken class WeiXinSendMsgClass(object): def __init__(self): self.access_token = odbp_getToken.WeiXinTokenClass().get() self.to_user = "" self.to_party = "" self.to_tag = "" self.msg_type = "text" self.agent_id = 2 self.content = "" self.safe = 0 self.data = { "touser": self.to_user, "toparty": self.to_party, "totag": self.to_tag, "msgtype": self.msg_type, "agentid": self.agent_id, "text": { "content": self.content }, "safe": self.safe } def send(self, to_user, content): if to_user is not None and content is not None: self.data['touser'] = to_user self.data['text']['content'] = content else: print raise RuntimeError import requests import json url = "https://qyapi.weixin.qq.com/cgi-bin/message/send" querystring = {"access_token": self.access_token} payload = json.dumps(self.data, encoding='utf-8', ensure_ascii=False) headers = { 'content-type': "application/json", 'cache-control': "no-cache", } response = requests.request("POST", url, data=payload, headers=headers, params=querystring) return_content = json.loads(response.content) if return_content["errcode"] == 0 and return_content["errmsg"] == "ok": print "Send successfully! %s " % return_content else: print "Send failed! %s " % return_content python調(diào)用mongodb發(fā)送微信企業(yè)號 python2.x 注意:data變量里, agent_id為剛剛創(chuàng)建的應(yīng)用id(可在web頁面看到) toparty即為目標部門,或者可以用touser,totag指定目標賬戶 比較簡單的調(diào)用,已實測,可以使用。 #coding:utf-8 import sys import requests import json from pymongo import MongoClient reload(sys) sys.setdefaultencoding('utf-8') class Weixin(object): def __init__(self, corp_id, corp_secret): self.token_url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=%s&corpsecret=%s' %(corp_id, corp_secret) self.send_url = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=' def get_token(self): try: r = requests.get(self.token_url, timeout=10) except Exception as e: print e sys.exit(1) if r.status_code == requests.codes.ok: data = r.json() if data.get('errcode'): print data['errmsg'] sys.exit(1) return data['access_token'] else: print r.status_code sys.exit(1) def send(self,message): url = self.send_url + self.get_token() data = { "touser": "hequan2011", "msgtype": "text", "agentid": "0", "text": { "content": message }, "safe":"0" } send_data = json.dumps(data,ensure_ascii=False) try: r = requests.post(url, send_data) except Exception, e: print e sys.exit(1) if r.status_code == requests.codes.ok: print r.json() else: print r.code sys.exit(1) corpid = 'xxxxxxxxxxx' corpsecret = 'xxxxxxxxxxxxxxxxx' client = MongoClient('mongodb://user:password@127.0.0.1:27017/') db = client.ku collection = db.biao a = [] for data in collection.find(): a.append(data) l = a[0] g = l z = str(g["name"]) z1 = int(g["jg"]) print z msg = "1:{0} 2:{1} ".format(z,z1) w = Weixin(corpid,corpsecret) w.send(msg) ZABBIX 微信報警 插件 #!/usr/bin/env python # -*- coding:utf-8 -*- # __author__ = '懶懶的天空' import requests import sys import json from conf.INIFILES import read_config, write_config import os import datetime from conf.BLog import Log reload(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'監(jiān)控ID'] warn_message = '' if content[u'當(dāng)前狀態(tài)'] == 'PROBLEM': body["title"] = "服務(wù)器故障" warn_message += subject + ' ' warn_message += '詳情: ' warn_message += '告警等級:'+ content[u'告警等級'] + ' ' warn_message += '告警時間:'+ content[u'告警時間'] + ' ' warn_message += '告警地址:'+ content[u'告警地址'] + ' ' warn_message += '持續(xù)時間:'+ content[u'持續(xù)時間'] + ' ' warn_message += '監(jiān)控項目:'+ content[u'監(jiān)控項目'] + ' ' warn_message += content[u'告警主機'] + '故障(' + content[u'事件ID']+ ')' else: body["title"] = "服務(wù)器恢復(fù)" warn_message += subject + ' ' warn_message += '詳情: ' warn_message += '告警等級:'+ content[u'告警等級'] + ' ' warn_message += '恢復(fù)時間:'+ content[u'恢復(fù)時間'] + ' ' warn_message += '告警地址:'+ content[u'告警地址'] + ' ' warn_message += '持續(xù)時間:'+ content[u'持續(xù)時間'] + ' ' warn_message += '監(jiān)控項目:'+ content[u'監(jiān)控項目'] + ' ' warn_message += content[u'告警主機'] + '恢復(fù)(' + 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_path def 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實現(xiàn)微信企業(yè)號的文本消息推送 #!/usr/bin/python # _*_coding:utf-8 _*_ import urllib2 import json import sys reload(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 token def 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 + " " + 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]) # 參數(shù)1:發(fā)送給用戶的賬號,必須關(guān)注企業(yè)號,并對企業(yè)號有發(fā)消息權(quán)限 party = str(sys.argv[2]) # 參數(shù)2:發(fā)送給組的id號,必須對企業(yè)號有權(quán)限 agent = str(sys.argv[3]) # 參數(shù)3:企業(yè)號中的應(yīng)用id subject = str(sys.argv[4]) # 參數(shù)4:標題【消息內(nèi)容的一部分】 content = str(sys.argv[5]) # 參數(shù)5:文本具體內(nèi)容 corpid = 'CorpID' # CorpID是企業(yè)號的標識 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調(diào)用Python程序控制微信公眾平臺發(fā)布報警信息 vim Notify-host-by-weixin-party.py import urllib.request import json import sys #以上是導(dǎo)入模塊 #創(chuàng)建獲取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 #這里是發(fā)送消息的方法 def senddata(access_token,notify_str): send_url = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=' + access_token #我傳入的參數(shù)是一段字符串每個信息用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[擦汗] 類型: ' + cationtype + ' 主機名: ' + name + ' 狀態(tài): ' + state + ' IP地址: ' + address + ' 摘要: ' + output + ' 時間: ' + datatime + ' ' if cationtype == "RECOVERY": content = '[噓]' + address + ' is ' + state + '[噓] IP地址: ' + address + ' 主要用途: ' + name + ' 當(dāng)前狀態(tài): ' + state + ' 日志摘要: ' + output + ' 檢測時間: ' + datatime + ' ' else: content = '[擦汗]' + address + ' is ' + state + '[擦汗] IP地址: ' + address + ' 主要用途: ' + name + ' 當(dāng)前狀態(tài): ' + state + ' 日志摘要: ' + output + ' 檢測時間: ' + datatime + ' ' 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') #設(shè)置為非ascii解析,使其支持中文 send_request = urllib.request.Request(send_url, send_data) response = urllib.request.urlopen(send_request) #這個是返回微信公共平臺的信息,調(diào)試時比較有用 msg = response.read() return msg default_encoding = 'utf-8' if sys.getdefaultencoding() != default_encoding: reload(sys) sys.setdefaultencoding(default_encoding) #我編輯的腳本是要獲取nagios傳入的一段參數(shù)的(字符串),下面這條代碼是獲取執(zhí)行腳本后獲取的第一個參數(shù)(經(jīng)測試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.request import json import sys 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 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 [擦汗] 類型: ' + cationtype + ' 服務(wù)名: ' + desc + ' 主機名: ' + alias + ' IP址: ' + address + ' 狀態(tài): ' + state + ' 時間: ' + datatime + ' 摘要: ' + output + ' ' if cationtype == "RECOVERY": content ='[鼓掌]' + desc + ' is ' + state + '[鼓掌] IP地址: ' + address + ' 主要用途: ' + alias + ' 服務(wù)狀態(tài): ' + desc + ' is ' + state + ' 檢測時間: ' + datatime + ' 日志摘要: ' + output + ' ' else: content ='[擦汗]' + desc + ' is ' + state + '[擦汗] IP地址: ' + address + ' 主要用途: ' + alias + ' 服務(wù)狀態(tài): ' + desc + ' is ' + state + ' 檢測時間: ' + datatime + ' 日志摘要: ' + output + ' ' 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 msg default_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調(diào)用企業(yè)微信服務(wù)號進行報警 #!/bin/bash corpid="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-8 import sys import urllib2 import time import json import requests reload(sys) sys.setdefaultencoding('utf-8') #title = sys.argv[2] # 位置參數(shù)獲取title 適用于zabbix #content = sys.argv[3] # 位置參數(shù)獲取content 適用于zabbix title = "title 測試" # 位置參數(shù)獲取title 適用于zabbix content = "content 測試" # 位置參數(shù)獲取content 適用于zabbix class 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_token def send_msg(title, content): # 發(fā)送消息 corpid = "" # 填寫自己應(yīng)用的 corpsecret = "" # 填寫自己應(yīng)用的 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èi)容:{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利用微信訂閱號報警 # coding=utf-8 import urllib import urllib2 import cookielib import json import sys data={'username':'yaokuaile-99', 'pwd':'f4bb2d8abe7a799ad62495a912ae3363', 'imgcode':'', 'f':'json' } cj=cookielib.LWPCookieJar() opener=urllib2.build_opener(urllib2.HTTPCookieProcessor(cj)) urllib2.install_opener(opener) def getToken(): headers = {'Accept': 'application/json, text/javascript, */*; q=0.01', 'Accept-Encoding': 'gzip,deflate,sdch', 'Accept-Language': 'zh-CN,zh;q=0.8', 'Connection': 'keep-alive', 'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8', 'Content-Length': '74', 'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8', 'Host': 'mp.weixin.qq.com', 'Origin': 'https://mp.weixin.qq.com', 'Referer': 'https://mp.weixin.qq.com/cgi-bin/loginpage?t=wxm2-login&lang=zh_CN', 'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/33.0.1750.154 Safari/537.36', 'X-Requested-With': 'XMLHttpRequest', } req = urllib2.Request('https://mp.weixin.qq.com/cgi-bin/login?lang=zh_CN',urllib.urlencode(data),headers) ret = urllib2.urlopen(req) retread= ret.read() token = json.loads(retread) token=token['redirect_url'][44:] return token ### send msg def sendWeixin(msg,token,tofakeid): msg = msg token = token data1 = {'type':'1','content':msg,'imgcode':'','imgcode':'','tofakeid':tofakeid,'f':'json','token':token,'ajax':'1'} headers = {'Accept':'*/*', 'Accept-Encoding': 'gzip,deflate,sdch', 'Accept-Language': 'zh-CN,zh;q=0.8', 'Connection': 'keep-alive', 'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8', 'Host': 'mp.weixin.qq.com', 'Origin': 'https://mp.weixin.qq.com', 'Referer': 'https://mp.weixin.qq.com/cgi-bin/singlemsgpage?msgid=&source=&count=20&t=wxm-singlechat&fromfakeid=150890&token=%s&lang=zh_CN', 'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/33.0.1750.154 Safari/537.36', 'X-Requested-With':'XMLHttpRequest', } req2 = urllib2.Request('https://mp.weixin.qq.com/cgi-bin/singlesend?t=ajax-response&f=json&lang=zh_CN',urllib.urlencode(data1),headers) ret2=urllib2.urlopen(req2) if __name__=='__main__': ''' useage: ./send_wx.py msg ''' token = getToken() msg = sys.argv[1:] msg = ' '.join(msg) tofakeid = '2443746922' sendWeixin(msg, token, tofakeid) 使用python實現(xiàn)微信模板消息 #!/usr/bin/env python #-*- coding:utf-8 -*- import urllib2,json import datetime,time from config import * import sys reload(sys) sys.setdefaultencoding("utf-8") class WechatPush(): def __init__(self,appid,secrect,file_name): # 傳入appid self.appid = appid # 傳入密碼 self.secrect = secrect # 傳入記錄token和過期時間的文件名 self.file_name=file_name def build_timestamp(self,interval): # 傳入時間間隔,得到指定interval后的時間 格式為"2015-07-01 14:41:40" now = datetime.datetime.now() delta = datetime.timedelta(seconds=interval) now_interval=now + delta return now_interval.strftime('%Y-%m-%d %H:%M:%S') def check_token_expires(self): # 判斷token是否過期 with open(self.file_name,'r') as f: line=f.read() if len(line)>0: expires_time=line.split(",")[1] token=line.split(",")[0] else: return "","true" curr_time=time.strftime('%Y-%m-%d %H:%M:%S') # 如果過期返回false if curr_time>expires_time: return token,"false" # 沒過期返回true else: return token,"true" def getToken(self): # 獲取accessToken url = 'https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid='+self.appid + "&secret="+self.secrect try: f = urllib2.urlopen(url) s = f.read() # 讀取json數(shù)據(jù) j = json.loads(s) j.keys() # 從json中獲取token token = j['access_token'] # 從json中獲取過期時長 expires_in =j['expires_in'] # 將得到的過期時長減去300秒然后與當(dāng)前時間做相加計算然后寫入到過期文件 write_expires=self.build_timestamp(int(expires_in-300)) content="%s,%s" % (token,write_expires) with open(self.file_name,'w') as f: f.write(content) except Exception,e: print e return token def post_data(self,url,para_dct): """觸發(fā)post請求微信發(fā)送最終的模板消息""" para_data = para_dct f = urllib2.urlopen(url,para_data) content = f.read() return content def do_push(self,touser,template_id,url,topcolor,data): '''推送消息 ''' #獲取存入到過期文件中的token,同時判斷是否過期 token,if_token_expires=self.check_token_expires() #如果過期了就重新獲取token if if_token_expires=="false": token=self.getToken() # 背景色設(shè)置,貌似不生效 if topcolor.strip()=='': topcolor = "#7B68EE" #最紅post的求情數(shù)據(jù) dict_arr = {'touser': touser, 'template_id':template_id, 'url':url, 'topcolor':topcolor,'data':data} json_template = json.dumps(dict_arr) requst_url = "https://api.weixin.qq.com/cgi-bin/message/template/send?access_token="+token content = self.post_data(requst_url,json_template) #讀取json數(shù)據(jù) j = json.loads(content) j.keys() errcode = j['errcode'] errmsg = j['errmsg'] #print errmsg if __name__ == "__main__": def alarm(title,hostname,timestap,level,message,state,tail): """報警函數(shù)""" color="#FF0000" 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}} return data def recover(title,message,alarm_time,recover_time,continue_time,tail): """恢復(fù)函數(shù)""" re_color="#228B22" 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}} return data # data=alarm("測試的報警消息","8.8.8.8",time.ctime(),"最高級別","然并卵","掛了","大傻路趕緊處理") # 實例化類 webchart=WechatPush(appid,secrect,file_name) url="http://www.xiaoniu88.com" print len(sys.argv) # 發(fā)送報警消息 if len(sys.argv) == 9: title=sys.argv[1] hostname=sys.argv[2] timestap=sys.argv[3] level=sys.argv[4] message=sys.argv[5] state=sys.argv[6] tail=sys.argv[7] print "sys.argv[1]"+sys.argv[1] print "sys.argv[2]"+sys.argv[2] print "sys.argv[3]"+sys.argv[3] print "sys.argv[4]"+sys.argv[4] print "sys.argv[5]"+sys.argv[5] print "sys.argv[6]"+sys.argv[6] print "sys.argv[7]"+sys.argv[7] print "sys.argv[8]"+sys.argv[8] with open("/etc/zabbix/moniter_scripts/test.log",'a+') as f: f.write(title+" ") f.write(hostname+" ") f.write(timestap+" ") f.write(level+" ") f.write(message+" ") f.write(state+" ") f.write(tail+" ") f.write("%s_%s" % ("group",sys.argv[8])+" ") data=alarm(title,hostname,timestap,level,message,state,tail) group_name="%s_%s" % ("group",sys.argv[8]) for touser in eval("%s_%s" % ("group",sys.argv[8])): webchart.do_push(touser,alarm_id,url,"",data) for touser in group_super: webchart.do_push(touser,alarm_id,url,"",data) #發(fā)送恢復(fù)消息 elif len(sys.argv) == 8: title=sys.argv[1] message=sys.argv[2] alarm_time=sys.argv[3] recover_time=sys.argv[4] continue_time=sys.argv[5] tail=sys.argv[6] print "sys.argv[1]"+sys.argv[1] print "sys.argv[2]"+sys.argv[2] print "sys.argv[3]"+sys.argv[3] print "sys.argv[4]"+sys.argv[4] print "sys.argv[5]"+sys.argv[5] print "sys.argv[6]"+sys.argv[6] print "sys.argv[7]"+sys.argv[7] data=recover(title,message,alarm_time,recover_time,continue_time,tail) for touser in eval("%s_%s" % ("group",sys.argv[7])): webchart.do_push(touser,recover_id,url,"",data) for touser in group_super: webchart.do_push(touser,recover_id,url,"",data) zabbix使用微信報警python腳本 #!/usr/bin/python # coding: utf-8 #python2將zabbix報警信息發(fā)送到微信。 #by linwangyi #2016-01-18 import urllib,urllib2 import json import sys 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 token def senddata(access_token,user,content): send_url = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=' + access_token send_values = { "touser":user, #企業(yè)號中的用戶帳號,在zabbix用戶Media中配置,如果配置不正常,將按部門發(fā)送。 "toparty":"1", #企業(yè)號中的部門id "msgtype":"text", #企業(yè)號中的應(yīng)用id,消息類型。 "agentid":"1", "text":{ "content":content }, "safe":"0" } send_data = json.dumps(send_values, ensure_ascii=False) send_request = urllib2.Request(send_url, send_data) response = json.loads(urllib2.urlopen(send_request).read()) print str(response) if __name__ == '__main__': user = str(sys.argv[1]) #zabbix傳過來的第一個參數(shù) content = str(sys.argv[3]) #zabbix傳過來的第三個參數(shù) corpid = 'XXXX' #CorpID是企業(yè)號的標識 corpsecret = 'XXXXX' #corpsecretSecret是管理組憑證密鑰 accesstoken = gettoken(corpid,corpsecret) senddata(accesstoken,user,content) #!/usr/bin/python3 # coding: utf-8 #python3將zabbix報警信息發(fā)送到微信。 #by http://sunday208.blog.51cto.com/ #2016-01-18 import urllib.request import json import sys 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,user,content): try: send_url = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=' + access_token send_values = { "touser":user, #企業(yè)號中的用戶帳號,在zabbix用戶Media中配置,如果配置不正常,將按部門發(fā)送。 "toparty":"1", #企業(yè)號中的部門id "msgtype":"text", "agentid":"1", #企業(yè)號中的應(yīng)用id,消息類型。 "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() print("returned value : " + str(msg)) except: print("returned value : " + str(msg)) default_encoding = 'utf-8' if sys.getdefaultencoding() != default_encoding: reload(sys) sys.setdefaultencoding(default_encoding) user = str(sys.argv[1]) #zabbix傳過來的第一個參數(shù) content = str(sys.argv[3]) #zabbix傳過來的第三個參數(shù) corpid = 'XXXX' #CorpID是企業(yè)號的標識 corpsecret = 'XXXXX' #corpsecretSecret是管理組憑證密鑰 accesstoken = gettoken(corpid,corpsecret) senddata(accesstoken,user,content)
聲明:本網(wǎng)頁內(nèi)容旨在傳播知識,若有侵權(quán)等問題請及時與本網(wǎng)聯(lián)系,我們將在第一時間刪除處理。TEL:177 7030 7066 E-MAIL:11247931@qq.com
本文如未解决您的问题请添加抖音号:51dongshi(抖音搜索懂视),直接咨询即可。