python利用微信公众号报警代码

布满荆棘的人生 2022-09-27 15:25 224阅读 0赞

微信公众号共有三种，服务号、订阅号、企业号。它们在获取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。

\#!/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:

The error leading to this problem was:

It's possible that this database is broken or permission denied.

If you cannot solve this problem yourself, please mail to:

""" % (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',

''', (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

(

)

''', (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实现通过微信企业号发送文本消息的Class

编程要点和调用方法：

支持发送中文，核心语句“payload = json.dumps(self.data, encoding='utf-8', ensure\_ascii=False)”，关键字“python json 中文”

这个Class只有一个公共方法send()。

使用方法：import这个class，然后调用send方法即可，方法参数只需要两个，给谁（多UserID用"|"隔开），内容是什么，例如：

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:

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调用mongodb发送微信企业号

python2.x

注意：data变量里， agent\_id为刚刚创建的应用id（可在web页面看到）

toparty即为目标部门，或者可以用touser,totag指定目标账户

比较简单的调用，已实测，可以使用。

\#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\}\\n 2:\{1\}\\n".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'监控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\_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实现微信企业号的文本消息推送

\#!/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 + "\\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.request

import json

import 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 msg

default\_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.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 \[擦汗\]\\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 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调用企业微信服务号进行报警

\#!/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\]   \# 位置参数获取title 适用于zabbix

\#content = sys.argv\[3\] \# 位置参数获取content 适用于zabbix

title = "title 测试"   \# 位置参数获取title 适用于zabbix

content = "content 测试"  \# 位置参数获取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):

\# 发送消息

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利用微信订阅号报警

\# 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 = '\\n'.join(msg)

tofakeid = '2443746922'

sendWeixin(msg, token, tofakeid)

使用python实现微信模板消息

\#!/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数据

j = json.loads(s)

j.keys()

\# 从json中获取token

token = j\['access\_token'\]

\# 从json中获取过期时长

expires\_in =j\['expires\_in'\]

\# 将得到的过期时长减去300秒然后与当前时间做相加计算然后写入到过期文件

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):

"""触发post请求微信发送最终的模板消息"""

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()

\# 背景色设置,貌似不生效

if topcolor.strip()=='':

topcolor = "\#7B68EE"

\#最红post的求情数据

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数据

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):

"""报警函数"""

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):

"""恢复函数"""

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)

\# 发送报警消息

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+"\\n")

f.write(hostname+"\\n")

f.write(timestap+"\\n")

f.write(level+"\\n")

f.write(message+"\\n")

f.write(state+"\\n")

f.write(tail+"\\n")

f.write("%s\_%s" % ("group",sys.argv\[8\])+"\\n")

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)

\#发送恢复消息

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报警信息发送到微信。

\#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,    \#企业号中的用户帐号，在zabbix用户Media中配置，如果配置不正常，将按部门发送。

"toparty":"1",    \#企业号中的部门id

"msgtype":"text",  \#企业号中的应用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传过来的第一个参数

content = str(sys.argv\[3\])  \#zabbix传过来的第三个参数

corpid = 'XXXX'   \#CorpID是企业号的标识

corpsecret = 'XXXXX'  \#corpsecretSecret是管理组凭证密钥

accesstoken = gettoken(corpid,corpsecret)

senddata(accesstoken,user,content)

\#!/usr/bin/python3

\# coding: utf-8

\#python3将zabbix报警信息发送到微信。

\#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,  \#企业号中的用户帐号，在zabbix用户Media中配置，如果配置不正常，将按部门发送。

"toparty":"1",  \#企业号中的部门id

"msgtype":"text",

"agentid":"1",   \#企业号中的应用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传过来的第一个参数

content = str(sys.argv\[3\])  \#zabbix传过来的第三个参数

corpid = 'XXXX'   \#CorpID是企业号的标识

corpsecret = 'XXXXX'  \#corpsecretSecret是管理组凭证密钥

accesstoken = gettoken(corpid,corpsecret)

senddata(accesstoken,user,content)

python利用微信公众号报警代码

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关利用Python实现微信公众号文章的自动发布

相关微信公众平台报警排查指引

相关利用微信企业号发送报警信息

相关 python利用微信公众号报警代码

相关【微信公众号】微信公众号开发——Werobot入门指南

相关微信公众号开发（四）代码调试

相关微信公众号

相关微信公众号开发

相关微信公众号开发——微信获取、操作公众号文章

相关微信公众号核心代码实现