• 爬虫-接入钉钉机器人


    前言

    又间隔了好久都没发博客了,心血来潮来写写博客,今天就写一下用pthon实现钉钉群机器人自动发消息的功能吧,也就是通过去爬取数据,通过钉钉内置的群机器人去做一个自动通知的这么一个操作

    钉钉机器人相关文档

    链接: 钉钉群机器人开放文档首先大家可以了解一下钉钉机器人相关的一些规则以及相关API
    钉钉现在支持的消息类型如下:
    机器人可以发送哪些消息类型
    根据机器人可以发送的消息类型内容不同,分为基础消息、互动卡片消息。

    基础消息。基础消息支持发送以下内容,详情请查看基础消息类型图及数据格式。
    –文本
    –链接消息
    –ActionCard
    –FeedCard类型
    –Markdown类型等
    –互动卡片消息

    群聊机器人配置

    第一步:群聊中添加一个自定义通过Webhook接入的机器人
    在这里插入图片描述
    此处会有安全设置,建议进行一个安全设置的配置
    自定义关键词:设定后,只有包含关键词的消息内容才会被正常发送
    加密:加密,当相关的秘钥跟sign不对应时,无法发送消息
    IP地址段:设定后,只有来自IP地址范围内的请求才会被正常处理
    在这里插入图片描述
    加密的话,钉钉API文档已经给出了相关的加密代码了,这里就只贴python的加密代码了

    #python 3.8
    import time
    import hmac
    import hashlib
    import base64
    import urllib.parse
    
    timestamp = str(round(time.time() * 1000))
    secret = 'this is 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))
    print(timestamp)
    print(sign)
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16

    获取到相关的时间戳和sign后再通过接口进行请求即可
    下面展示一些 内联代码片

    // A code block
    var foo = 'bar';
    
    • 1
    • 2
    https://oapi.dingtalk.com/robot/send?access_token=XXXXXX&timestamp=XXX&sign=XXX
    
    • 1

    钉钉群机器人实战

    下面展示一些 内联代码片

    # !/usr/bin/env python3
    # -*- coding: utf-8 -*-
    """
    @Time : 2022/10/31 18:29
    @Author : D_FIRE
    @Email : dfire_liao@189.cn
    @File : 1.py
    @Software: PyCharm
    """
    import json
    import requests
    import time
    import hmac
    import hashlib
    import base64
    import urllib.parse
    import urllib3
    
    
    # 获取商品相关信息
    def get_itemmessage():
        url = 'http://xx.xx.xx.xx:xxxx/api/order/query'
        jsondata = dict(date="2022-03-02 10:20:46", data={
            "genTlogTicketId": 1,
            "attributes": [
                "items",
                "shipping",
                "payment",
                "express",
                "invoice",
                "channel",
                "channelInfo"
            ],
            "orderId": 111966168578820096,
            "pageParam": {
                "count": 1,
                "start": 0
            }
        }, appId="storewin_uat", sign="FA595084D34D3266D2E9C4881CAD1A00", certiId="storewin_uat")
        req = requests.post(url=url, json=jsondata)
        # 商品名称、单价、数量、总价
        itemname = req.json()["data"][0]["items"][0]["itemNameZht"]
        itemprice = req.json()["data"][0]["items"][0]["actualPrice"]
        itemqty = req.json()["data"][0]["items"][0]["orderQty"]
        allprice = req.json()["data"][0]["items"][0]["itemAmount"]
        if itemname:
            res = {
                "商品名称": itemname,
                "商品价格": itemprice,
                "商品数量": itemqty,
                "商品总价": allprice
            }
            yield res
    
    # 钉钉机器人验签方法
    timestamp = str(round(time.time() * 1000))
    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():
        urllib3.disable_warnings()
        # 请求的URL,WebHook地址
        webhook = f"https://oapi.dingtalk.com/robot/send?access_token={你的秘钥}×tamp={timestamp}&sign={sign}"
    
        # 构建请求头部
        header = {"Content-Type": "application/json", "Charset": "UTF-8"}
    
        for itemmessage in get_itemmessage():
            message = {
                "at": {
                    # 是否@所有人
                    "isAtAll": True
                },
                "text": {
                    # 发送信息
                    "content": f"爬取信息如下:{itemmessage}"
                },
                # 发送类型
                "msgtype": "text"
            }
        message_json = json.dumps(message)
        info = requests.post(url=webhook, data=message_json, headers=header, verify=False)  # 打印返回的结果
        print(info.text)
    
    
    if __name__ == "__main__":
        dingmessage()
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93

    成功接入效果如下:
    在这里插入图片描述
    如果想实现定时推送的话可以设置windows或者mac本机自带的定时任务计划程序去定时调用就好了

    结语

    钉钉公司用的少,现在想爬点东西下来每天定时推送,但是还没想到要去爬什么,如果看到这篇博客的你有啥建议的话可以留言哦。

  • 相关阅读:
    vue3中使用 tui-image-editor进行图片处理,并上传
    Linux查看指定端口是否被占用
    万字长文:从实践到原理说透Golang defer
    分布式微服务架构-一起学习吧之架构
    【论文理解】Batch Normalization论文中关于BN背景和减少内部协变量偏移的解读(论文第1、2节)
    关于go协程超时退出控制条件与方式分析
    软考知识点2
    记一次SpringBoot中Service层未注入排查
    电力社区电力故障,潜在风险如何避免?
    汽车Type-C接口:特点与要求解析
  • 原文地址:https://blog.csdn.net/DFireTesting/article/details/127620649