先導
工作中運維工作經常會遇到一些數據匯報,數據監控, 作為一個新人真心感覺這些數據沒有什么意思(當然也許是我菜),有句話怎么說,懶惰是人類進步的階梯.這里使用 python 連接數據把數據 傳到釘釘, 這樣可以進行數據監控 ,看看是哪家小可愛又 搞事情了 ╭(╯^╰)╮
釘釘接口
釘釘提供了群機器人接口等很多很多接口,網上 demo 一大堆.這里只制作簡單的表述
釘釘官網:
https://developers.dingtalk.com/document/app/overview-of-group-robots
可以給群里增加一個機器人，通過 @固定人 ，或者所有人，廣播等方式發送信息
創建釘釘機器人
首先你得建立個群
增加群機器人
完成必要的安全設置，勾選我已閱讀并同意《自定義機器人服務及免責條款》，然后單擊完成。
目前有 3 種安全設置方式，請根據需要選擇一種：
-
自定義關鍵詞：最多可以設置 10 個關鍵詞，消息中至少包含其中 1 個關鍵詞才可以發送成功。
例如添加了一個自定義關鍵詞：監控報警，則這個機器人所發送的消息，必須包含監控報警這個詞，才能發送成功。
-
加簽：
把timestamp+"\n"+密鑰當做簽名字符串，使用 HmacSHA256 算法計算簽名，然后進行 Base64 encode，最后再把簽名參數再進行 urlEncode，得到最終的簽名（需要使用 UTF-8 字符集）。
測試機器人
python 代碼版本
這里直接使用 加簽版本的,因為這種時間判定的才是最常用的
import requestsimport jsonimport timeimport hmacimport hashlibimport base64import urllib.parse
timestamp = str(round(time.time() * 1000)) secret = '你的secret'secret_enc = secret.encode('utf-8')string_to_sign = '{}\n{}'.format(timestamp, secret)string_to_sign_enc = string_to_sign.encode('utf-8')hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
def dingmessage():
webhook = "https://oapi.dingtalk.com/robot/send?access_token=你的token×tamp="+str(timestamp)+"&sign="+str(sign) print(webhook) header = { "Content-Type": "application/json", "Charset": "UTF-8"} tex = "能力越大，責任越大。" message ={
"msgtype": "text", "text": { "content": tex }, "at": {
"isAtAll": False }
} message_json = json.dumps(message) info = requests.post(url=webhook,data=message_json,headers=header) print(info.text)
if __name__=="__main__": dingmessage()
復制代碼
打印結果
如果顯示 ok 那么就是說數據傳輸成功了
這時候接入釘釘即可 查看數據
好了現在 py 已經能夠給釘釘發送信息了,那么怎么做監控呢
創建時間監控 實時發送信息
我這里以 apscheduler 框架 進行定時巡回
一、安裝 APScheduler
二、基本概念
APScheduler 有四大組件：
1、觸發器 triggers ：觸發器包含調度邏輯。每個作業都有自己的觸發器，用于確定下一個任務何時運行。除了初始配置之外，觸發器是完全無狀態的。有三種內建的 trigger:
（1）date: 特定的時間點觸發
（2）interval: 固定時間間隔觸發
(3）cron: 在特定時間周期性地觸發
2、任務儲存器 job stores：用于存放任務，把任務存放在內存（為默認 MemoryJobStore）或數據庫中。3、執行器 executors： 執行器是將任務提交到線程池或進程池中運行，當任務完成時，執行器通知調度器觸發相應的事件。
4、調度器 schedulers： 把上方三個組件作為參數，通過創建調度器實例來運行根據開發需求選擇相應的組件，下面是不同的調度器組件：BlockingScheduler 阻塞式調度器：適用于只跑調度器的程序。BackgroundScheduler 后臺調度器：適用于非阻塞的情況，調度器會在后臺獨立運行。AsyncIOScheduler AsyncIO 調度器，適用于應用使用 AsnycIO 的情況。GeventScheduler Gevent 調度器，適用于應用通過 Gevent 的情況。TornadoScheduler Tornado 調度器，適用于構建 Tornado 應用。TwistedScheduler Twisted 調度器，適用于構建 Twisted 應用。QtScheduler Qt 調度器，適用于構建 Qt 應用。
三、使用步驟
1、新建一個調度器 schedulers
2、添加調度任務
3、運行調度任務
比如寫 3 就是說 馮 3 觸發
*/3 就是說 每隔 3 單位觸發
假設觸發條件
我們這里判定如果當前時間 4 個數據加起來大于 12 那么 就發送消息
每分鐘發送一次
from apscheduler.schedulers.blocking import BlockingSchedulerimport datetimeimport pymssqlimport copyimport timeimport requestsimport jsonimport hmacimport hashlibimport base64import urllib.parse
timestamp = str(round(time.time() * 1000))secret = '你的secret'secret_enc = secret.encode('utf-8')string_to_sign = '{}\n{}'.format(timestamp, secret)string_to_sign_enc = string_to_sign.encode('utf-8')hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
def dingmessage(num):
webhook = "https://oapi.dingtalk.com/robot/send?access_token=你的token×tamp="+str(timestamp)+"&sign="+str(sign) print(webhook) header = { "Content-Type": "application/json", "Charset": "UTF-8"} tex = "能力越大，責任越大。"+str(num) message ={
"msgtype": "text", "text": { "content": tex }, "at": {
"isAtAll": False }
} message_json = json.dumps(message) info = requests.post(url=webhook,data=message_json,headers=header) print(info.text)
def job(): hour = datetime.datetime.now().hour min = datetime.datetime.now().minute f1 = str(hour)[0] f2 = str(hour)[1] f3 = str(min)[0] f4 = str(min)[1] print(f1) print(f2) print(f3) print(f4) print(int(f1) + int(f2) + int(f3) + int(f4)) if int(f1) + int(f2) + int(f3) + int(f4) > 12: dingmessage(int(f1) + int(f2) + int(f3) + int(f4) )
scheduler = BlockingScheduler()scheduler.add_job(job, 'cron', hour='1-23', minute='*/1',misfire_grace_time=3600)scheduler.start()
復制代碼
結果
運行結果
釘釘結果
如圖 23.18 分 2+3+1+8=14 >12 觸發條件 就給釘釘 發送信息了
致此一個見得 發送文字功能實現,有問題即可發送釘釘
還有一點就是釘釘一定要調成前臺允許顯示, 然后在錄音個對應的提示. 懶人必備,
浙公網安備 33010602011771號