用Python實現阿裏釘釘機器人讀取數據庫內容自動發群通知
阿新 • • 發佈：2018-09-30
media rep sql查詢 docs ken 由於 false n! 效果
最近想把一些預警數據信息按照一定的要求自動發送到移動端APP，最終把目標放在了騰訊的微信和阿裏的釘釘軟件上，由於剛開始學習python，於是編程工具想用python來實現。微信使用群體最廣，通過一天的研究用itchat庫已經實現，但由於itchat需要用web微信方式登錄，發現微信對新註冊的用戶關閉了web微信功能，於是考慮用備選方案阿裏釘釘來實現，其實阿裏釘釘雖然沒有微信用戶群體龐大，但是在企業應用方面比微信強大了太多，很多企業已經開始開始用釘釘作為內部溝通工具。
一、工作準備
1、Python工作環境搭建（略）
需要用到兩個模塊
# pip install apscheduler
# pip install pymysql
2、申請阿裏釘釘webhok，首先在群設置裏面添加自定義機器人
找到webhook地址復制備用。
二、編寫代碼如下
from datetime import datetime import json import urllib.request import pymysql as pms from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.schedulers.background import BackgroundScheduler import time import os#Mac下關閉ssl驗證用到以下模塊 import ssl ‘‘‘ ---------------------------------------------- # 需要CMD命令下安裝以下支持庫： # pip install apscheduler # pip install pymysql # By wzy 2018-9-28 ---------------------------------------------- ‘‘‘ #Mac下關閉ssl驗證，不然會報錯 ssl._create_default_https_context = ssl._create_unverified_context#你的釘釘機器人url global myurl my_url = "https://oapi.dingtalk.com/robot/send?access_token=XXXXXXXXXXXX" def send_request(url, datas): #傳入url和內容發送請求 # 構建一下請求頭部 header = { "Content-Type": "application/json", "Charset": "UTF-8" } sendData = json.dumps(datas) # 將字典類型數據轉化為json格式 sendDatas = sendData.encode("utf-8") # python3的Request要求data為byte類型 # 發送請求 request = urllib.request.Request(url=url, data=sendDatas, headers=header) # 將請求發回的數據構建成為文件格式 opener = urllib.request.urlopen(request) # 打印返回的結果 print(opener.read()) def get_mysqldatas(sql): # 一個傳入sql導出數據的函數，實例為MySQL需要先安裝pymysql庫，cmd窗口命令：pip install pymysql # 跟數據庫建立連接 conn = pms.connect(host=‘服務器地址‘, user=‘用戶名‘, passwd=‘密碼‘, database=‘數據庫‘, port=3306, charset="utf8") # 使用 cursor() 方法創建一個遊標對象 cur = conn.cursor() # 使用 execute() 方法執行 SQL cur.execute(sql) # 獲取所需要的數據 datas = cur.fetchall() # 關閉連接 cur.close() # 返回所需的數據 return datas def get_ddmodel_datas(type): #返回釘釘模型數據，1:文本；2:markdown所有人；3:markdown帶圖片，@接收人；4:link類型 if type == 1: my_data = { "msgtype": "text", "text": { "content": " " }, "at": { "atMobiles": [ "188XXXXXXX" ], "isAtAll": False } } elif type == 2: my_data = { "msgtype": "markdown", "markdown": {"title": " ", "text": " " }, "at": { "isAtAll": True } } elif type == 3: my_data = { "msgtype": "markdown", "markdown": {"title":" ", "text":" " }, "at": { "atMobiles": [ "188XXXXXXXX" ], "isAtAll": False } } elif type == 4: my_data = { "msgtype": "link", "link": { "text":" ", "title": " ", "picUrl": "", "messageUrl": " " } } return my_data def main(): print(‘Main! The time is: %s‘ % datetime.now()) #按照釘釘給的數據格式設計請求內容 鏈接https://open-doc.dingtalk.com/docs/doc.htm?spm=a219a.7629140.0.0.p7hJKp&treeId=257&articleId=105735&docType=1 #調用釘釘機器人全局變量myurl global myurl #1.Text類型群發消息 #合並標題和數據 My_content = "hello, @188XXXXXXXX 這是一個測試消息" my_data = get_ddmodel_datas(1) #把文本內容寫入請求格式中 my_data["text"]["content"] = My_content send_request(my_url, my_data) #2.Markdown類型群發消息（MySQL查詢結果發送） #獲取sql數據 sql = "SELECT branch_no,count(*) from wzy_customer_user group by branch_no order by branch_no" my_mydata = get_mysqldatas(sql) str1 = ‘\t\n\r‘ seq = [] for i in range(len(my_mydata)): seq.append(str(my_mydata[i])) data = str1.join(seq) data = data.replace(‘\‘‘,‘‘) data = data.replace(‘(‘,‘‘) data = data.replace(‘)‘,‘‘) data = data.replace(‘,‘,‘\t‘) print(data) Mytitle = "#### XXX報表\r\n單位\t數量\t\n\r %s" my_Mytitle = Mytitle.join(‘\t\n‘) % data my_data = get_ddmodel_datas(2) my_data["markdown"]["title"] ="XXXX 通報" my_data["markdown"]["text"] = my_Mytitle send_request(my_url, my_data) #3.Markdown（帶圖片@對象） my_data = get_ddmodel_datas(3) my_data["markdown"]["title"] = "系統預警" my_data["markdown"]["text"] = "#### 系統預警內容 \n > @188XXXXXXXX \n\n > \n > ###### 20點00分發布 [詳情](http://www.baidu.cn/)" send_request(my_url, my_data) #4.Link類型群發消息 my_data = get_ddmodel_datas(4) my_data["link"]["text"] = "群機器人是釘釘群的高級擴展功能。群機器人可以將第三方服務的信息聚合到群聊中，實現自動化的信息同步。 " my_data["link"]["title"] = "自定義機器人協議" my_data["link"]["messageUrl"] = "https://open-doc.dingtalk.com/docs/doc.htm?spm=a219a.7629140.0.0.Rqyvqo&treeId=257&articleId=105735&docType=1" send_request(my_url, my_data) if __name__ == "__main__": #定時執行任務，需要先安裝apscheduler庫，cmd窗口命令：pip install apscheduler #隨腳本執行 #scheduler = BlockingScheduler() #後臺執行 scheduler = BackgroundScheduler() #每隔20秒執行一次 scheduler.add_job(main, ‘interval‘, seconds=20) ‘‘‘ ***定時執行示例*** #固定時間執行一次 #sched.add_job(main, ‘cron‘, year=2018, month=9, day=28, hour=15, minute=40, second=30) #表示2017年3月22日17時19分07秒執行該程序 scheduler.add_job(my_job, ‘cron‘, year=2017,month = 03,day = 22,hour = 17,minute = 19,second = 07) #表示任務在6,7,8,11,12月份的第三個星期五的00:00,01:00,02:00,03:00 執行該程序 scheduler.add_job(my_job, ‘cron‘, month=‘6-8,11-12‘, day=‘3rd fri‘, hour=‘0-3‘) #表示從星期一到星期五5:30（AM）直到2014-05-30 00:00:00 scheduler.add_job(my_job(), ‘cron‘, day_of_week=‘mon-fri‘, hour=5, minute=30,end_date=‘2014-05-30‘) #表示每5秒執行該程序一次，相當於interval 間隔調度中seconds = 5 scheduler.add_job(my_job, ‘cron‘,second = ‘*/5‘) ‘‘‘ scheduler.start() print(‘Press Ctrl+{0} to exit‘.format(‘Break‘ if os.name == ‘nt‘ else ‘C‘)) try: # 其他任務是獨立的線程執行 while True: pass #time.sleep(60) #print(‘進程正在執行!‘) except (KeyboardInterrupt, SystemExit): #終止任務 scheduler.shutdown() print(‘Exit The Job!‘)
三、測試運行效果
用Python實現阿裏釘釘機器人讀取數據庫內容自動發群通知