1、在钉钉创建自定义机器人webhook 点击“复制”按钮，即可获得这个机器人对应的Webhook地址，其格式如下： https://oapi.dingtalk.com/robot/send?access_token=xxxxxxxx
2、安装python依赖包
pip install pyinotify pip install requests3、准备 py_inotify.py，监控php慢日志，如果慢日志有变动，将变动内容发送到钉钉
# !/usr/bin/python #coding:utf-8 import sys, os, pyinotify import impose import datetime import dingding notifier = None monfile = None lastsize = 0 wm = None wd = 0 def roll_file(filename): global lastsize fd = os.open(filename, os.O_RDONLY) try: newsize = os.fstat(fd).st_size if newsize <= lastsize: return os.lseek(fd, lastsize, os.SEEK_SET) while True: data = os.read(fd, 4096) if not data: break # sys.stdout.write(data) impose_number = impose.number(data) if impose_number: dingding.dingmessage(data) else: print(datetime.datetime.now().strftime('[%Y-%m-%d %H:%M:%S] ') + "超过调用次数了") sys.stdout.flush() pos = os.lseek(fd, 0, os.SEEK_CUR) lastsize = pos if pos != lastsize else newsize finally: os.close(fd) class EventHandler(pyinotify.ProcessEvent): def process_IN_CREATE(self, event): if monfile == event.pathname: global wd wd = wm.add_watch(monfile, pyinotify.IN_MODIFY).values()[0] roll_file(monfile) def process_IN_DELETE(self, event): global wd, lastsize if monfile == event.pathname: if wd > 0: try: wm.rm_watch(wd, quiet=False) except pyinotify.WatchManagerError: pass wd = 0 lastsize = 0 def process_IN_MOVED_FROM(self, event): self.process_IN_DELETE(event) def process_IN_MOVED_TO(self, event): self.process_IN_DELETE(event) self.process_IN_CREATE(event) def process_IN_MODIFY(self, event): roll_file(monfile) def main(): global notifier, lastsize, wm, wd, monfile monfile = "./a.txt" #print "path={0}".format(monfile) lastsize = os.stat(monfile).st_size wm = pyinotify.WatchManager() notifier = pyinotify.Notifier(wm, EventHandler()) wd = wm.add_watch(monfile, pyinotify.IN_MODIFY).values()[0] wm.add_watch(os.path.dirname(monfile), pyinotify.IN_DELETE | pyinotify.IN_CREATE | pyinotify.IN_MOVED_FROM | pyinotify.IN_MOVED_TO) #print "watching {0} ...".format(monfile) while True: notifier.process_events() if notifier.check_events(): notifier.read_events() if __name__ == "__main__": try: main() finally: if notifier: notifier.stop()4、在py_inotify.py同级目录创建dingding.py
#coding:utf-8 import requests import json import sys import datetime reload(sys) sys.setdefaultencoding('utf-8') def dingmessage(message): # 请求的URL，WebHook地址 webhook = "https://oapi.dingtalk.com/robot/send?access_token=xxxxxxxxxxxxxxx" #构建请求头部 header = { "Content-Type": "application/json", "Charset": "UTF-8" } #构建请求数据 tex = message message ={ "msgtype": "text", "text": { "content": tex }, "at": { "isAtAll": True } } #对请求的数据进行json封装 message_json = json.dumps(message) #发送请求 info = requests.post(url=webhook,data=message_json,headers=header) #打印返回的结果 print(datetime.datetime.now().strftime('[%Y-%m-%d %H:%M:%S] ') + info.text)5、调用钉钉次数限制，在py_inotify.py同级目录创建impose.py
# -*- coding:utf-8 -*- import time import datetime def stat_called_time(func): cache = {} limit_times = [18] def _called_time(*args, **kwargs): key = func.__name__ #判断是否存在缓存 if key in cache.keys(): [call_times, updatetime] = cache[key] #过期时间固定为60秒 if time.time() - updatetime < 60: cache[key][0] += 1 else: cache[key] = [1, time.time()] else: call_times = 1 cache[key] = [call_times, time.time()] # print('调用次数: %s' % cache[key][0]) # print('限制次数: %s' % limit_times[0]) if cache[key][0] <= limit_times[0]: data = func(*args, **kwargs) cache[key][1] = time.time() return True else: print(datetime.datetime.now().strftime('[%Y-%m-%d %H:%M:%S] ') + "超过调用次数了") # return None return False return _called_time @stat_called_time def number(data): return data6、测试将文件变动内容发送到钉钉机器人： python py_inotify.py 钉钉返回结果： {“errcode”:0,“errmsg”:“ok”} 打开另一个终端： echo “test dingding” >> a.txt 钉钉接受到消息通知：
