用Python實現阿里釘釘機器人讀取數據庫內容自動發群通知

最近想把一些預警數據信息按照一定的要求自動發送到移動端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 > ![screenshot](http://i01.lw.aliimg.com/media/lALPBbCc1ZhJGIvNAkzNBLA_1200_588.png)\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!')

 

三、測試運行效果