python实现钉钉文件上传发送，钉钉机器人接收信息

Z时代
2024-01-10
分类：综合

python

一、钉钉代码

import dingtalk.api
import requests
class DingDingAPI():
    def __init__(self, appkey, appsecret):
        self.appkey = appkey
        self.appsecret = appsecret
        self.access_token = self.get_token()
    # 获取token
    def get_token(self):
        params = {
            "appkey": self.appkey,
            "appsecret": self.appsecret
        }
        try:
            res = requests.get("https://oapi.dingtalk.com/gettoken", params=params)
            access_token = res.json().get("access_token")
            return access_token
        except Exception as e:
            print(e)
    # 发送消息
    def send_msg(self, chatid="", content=""):
        req = dingtalk.api.OapiChatSendRequest("https://oapi.dingtalk.com/chat/send")
        req.chatid = chatid
        req.text = {
            "content": content
        }
        req.msgtype = "text"
        resp = req.getResponse(self.access_token)
        return resp
    # 发送文件
    def send_file(self, chatid="", media_id=""):
        req = dingtalk.api.OapiChatSendRequest("https://oapi.dingtalk.com/chat/send")
        req.chatid = chatid
        req.file = {
            "media_id": media_id
        }
        req.msgtype = "file"
        resp = req.getResponse(self.access_token)
        return resp
    # 上传文件
    def upload_media(self,file_name="name.docx",file_path=""):
        req = dingtalk.api.OapiMediaUploadRequest("https://oapi.dingtalk.com/media/upload")
        req.type = "file"
        req.media = dingtalk.api.FileItem(file_name,open(file_path, 'rb'))
        resp = req.getResponse(self.access_token)
        return resp
    # 通过电话获取userid
    def get_by_mobile(self,mobile):
        req = dingtalk.api.OapiUserGetByMobileRequest("https://oapi.dingtalk.com/user/get_by_mobile")
        req.mobile = mobile
        try:
            resp = req.getResponse(self.access_token)
            return resp
        except Exception as e:
            print(e)
    # 通过userid获取信息,不能使用机器人appkey
    def get_user_msg(self,userid):
        req = dingtalk.api.OapiUserGetRequest("https://oapi.dingtalk.com/user/get")
        req.userid = userid
        try:
            resp = req.getResponse(self.access_token)
            print(resp)
            return resp
        except Exception as e:
            print(e)
if __name__ == '__main__':
    appkey = "*******"
    appsecret = "**********"
    dd=DingDingAPI(appkey,appsecret)

官方的封装模块dingtalk是python2.7的，我已改成3.0的

https://pan.baidu.com/s/1tpnTm17h4kVq81DK71wlDQ

密码：ivt4

钉钉的debug调试网址：https://wsdebug.dingtalk.com/，可以用来获取chatid

二、钉钉机器人

需要一个公网ip，搭建django服务器来接收钉钉服务器的回调

首先是注册一个钉钉机器人，重要的是将自己接收钉钉消息的服务器ip加路由填入到-‘消息接收地址’

#url部分代码
from django.conf.urls import url
from django.contrib import admin
from dingding import views
urlpatterns = [
    url(r'^admin/', admin.site.urls),
    url(r'^robot/', views.robot),
]

#views部分代码
from django.http import HttpResponse, JsonResponse
import json
import hmac
import hashlib
import base64
# 机器人的app_secret
app_secret = "************"
# Create your views here.
def robot(request):
    if request.method == "POST":
        HTTP_SIGN = request.META.get("HTTP_SIGN")
        HTTP_TIMESTAMP = request.META.get("HTTP_TIMESTAMP")
        res = json.loads(request.body)
        print(res)
        # 用户输入钉钉的信息
        content = res.get("text").get("content")
        string_to_sign = '{}\n{}'.format(HTTP_TIMESTAMP, app_secret)
        string_to_sign_enc = string_to_sign.encode('utf-8')
        hmac_code = hmac.new(app_secret.encode("utf-8"), string_to_sign_enc, digestmod=hashlib.sha256).digest()
        sign = base64.b64encode(hmac_code).decode("utf-8")
        print(sign)
        print(HTTP_SIGN)
        #验证签名是否为钉钉服务器发来的
        if sign == HTTP_SIGN:
            '''
            可以写一些执行逻辑，返回用户想要的信息，比如工资信息，可以去数据库查工资信息回给钉钉用户
            '''
            if "我的工资" in content:
                return JsonResponse(
                    {"msgtype": "text",
                     "text": {
                         "content": "您上月的工资为*****元"
                     }
                     }
                )
            return JsonResponse(
                {"msgtype": "text",
                 "text": {
                     "content": "谢谢使用此机器人，{}".format(content)
                 }
                 }
            )
        return JsonResponse({"error":"你没有权限访问此接口"})
    if request.method == "GET":
        return HttpResponse("hello")

三、修改发送文件不能打开的问题

import mimetypes
# 直接用requests包发送请求，官方包哪里传参数的时候有问题
def upload_media_v1(self, file_name="", file_path=""):
    files = {'type': (None, "file"), 'media': (file_name, open(file_path, 'rb'), mimetypes.guess_type(file_name)[0])}
    params = {
        "access_token": self.access_token,
        "type": "file"
    }
    re = requests.post("https://oapi.dingtalk.com/media/upload", files=files, params=params)
    return re.json()

还需要修改一处源码，不然不能发送中文文件，红色的注释，绿色的增加

四、增加一个钉钉webhook代码

import requests
import datetime
def send_msg(task_name,start_time,end_time):
    url = "https://oapi.dingtalk.com/robot/send?access_token=**************"
    data = {
        "msgtype": "markdown",
        "markdown": {
            "title": "定时任务运行结果！",
            "text": "### 您的任务已出错，请及时处理！\n\n"
                    "> **任务名称：** %s \n\n"
                    "> **开始时间：** %s \n\n"
                    "> **结束时间：** %s"%(task_name,start_time,end_time)
        },
        "at": {
            "atMobiles": [
                "手机号"
            ],
            "isAtAll": False
        }
    }
    res=requests.post(url=url,json=data)
    print(res.json())
task_name="T公司发票上传"
start_time=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
end_time=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
send_msg(task_name,start_time,end_time)