钉钉机器人消息推送类型案例

server/2024/12/14 7:31:34/

CSDN 目录展示

目录

  • 钉钉机器人消息推送
      • 1- 文本text类型
        • 推送代码案例
        • 推送结果
        • 参数说明
      • 2- 链接Link类型
        • 推送代码案例
        • 推送结果
        • 参数说明
      • 3- Markdown类型
        • 推送代码案例1
        • 推送结果1
        • 推送代码案例2
        • 推送结果2
        • 推送代码案例2 (版本2)
        • 推送结果2(版本2)
        • 参数说明
      • 4- 整体跳转ActionCard类型
        • 推送代码案例
        • 推送结果
        • 参数说明
      • 5- 推送文件链接供下载
        • 推送背景
        • 推送代码案例
        • 推送结果
      • 结语:
      • end
    • end

钉钉机器人消息推送

1- 文本text类型

推送代码案例
python">import json
from datetime import datetime
import requests now_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(now_time)# 接口地址和token信息
url = 'https://oapi.dingtalk.com/robot/send?access_token=*************'
keywords = 'DataWorks'# 钉钉机器人,发送消息
def dd_robot(url:str, content: str):HEADERS = {"Content-Type": "application/json;charset=utf-8"}#content里面要设置关键字data_info = {"msgtype": "text","text": {"content": content}#这是配置需要@的人#,"at": {"atMobiles": ["15xxxxxx06",'18xxxxxx1'], "atUserIds": ["user123"], "isAtAll": False} }value = json.dumps(data_info)response = requests.post(url,data=value,headers=HEADERS)if response.json()['errmsg']!='ok':print(response.text)# 主函数
if __name__ == '__main__':# 编辑需要推送的文档信息content = keywords + "(DQC)校验告警: \n  "content = content + "【测试时间】：" + now_time + " \n  "content = content + "【测试内容】：" + "测试钉钉推送text文档信息;"# 调用推送函数, 推送消息到钉钉群dd_robot(url, content)
推送结果
DataWorks(DQC)校验告警: 【测试时间】：2024-12-03 14:34:18 【测试内容】：测试钉钉推送text文档信息;

在这里插入图片描述

参数说明
python">{"at": {"atMobiles": ["180xxxxxx"],"atUserIds": ["user123"],"isAtAll": false},"text": {"content": "这里是需要推送的文本内容"},"msgtype": "text"
}
参数是否必填类型说明
msgtype是Stringtext。
text.content是String消息文本。
at.atMobiles否Array被@人的手机号。
at.atUserIds否Array被@人的用户userid。
at.isAtAll否Boolean@所有人是true，否则为false。
keywords是String钉钉机器人的关键词

2- 链接Link类型

推送代码案例
python">import json
from datetime import datetime
import requests now_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(now_time)# 接口地址和token信息
url = 'https://oapi.dingtalk.com/robot/send?access_token=**********'
keywords = 'DataWorks'# 钉钉机器人,发送消息
def dd_robot(url:str, text_content: str, title_content: str, pic_url: str, message_url: str):HEADERS = {"Content-Type": "application/json;charset=utf-8"}#content里面要设置关键字data_info = {"msgtype": "link","link": {"text": text_content,"title": title_content,"picUrl": pic_url,"messageUrl": message_url}}value = json.dumps(data_info)response = requests.post(url,data=value,headers=HEADERS)if response.json()['errmsg']!='ok':print(response.text)# 主函数
if __name__ == '__main__':# 编辑需要推送的信息title_content = keywords + ":这里是一个Link消息 \n"text_content = "SQL统计连续登陆3天的用户(连续活跃超3天用户)"pic_url = "https://img-blog.csdnimg.cn/img_convert/f63f1c575e3c48d9a3f7e4ba9bd460d2.png"message_url = "https://blog.csdn.net/Taerge0110/article/details/134536574?spm=1001.2014.3001.5502"# 调用推送函数, 推送消息到钉钉群dd_robot(url, text_content, title_content, pic_url, message_url)
推送结果
DataWorks:这里是一个Link消息 
SQL统计连续登陆3天的用户(连续活跃超3天用户)

在这里插入图片描述

参数说明
python">{"msgtype": "link", "link": {"text": "SQL统计连续登陆3天的用户(连续活跃超3天用户)", "title": "DataWorks:这里是一个Link消息", "picUrl": "https://img.alicdn.com/tfs/TB1NwmBEL9TBuNjy1zbXXXpepXa-2400-1218.png", "messageUrl": "https://open.dingtalk.com/document/"}
}
参数参数类型是否必填说明
msgtypeString是消息类型，此时固定为：link。
link.titleString是消息标题。如果太长只会部分展示。
link.textString是消息内容。如果太长只会部分展示。
link.messageUrlString是点击消息跳转的URL。
link.picUrlString否图片URL。
keywordsString是钉钉机器人的关键词

3- Markdown类型

推送代码案例1
python">import json
from datetime import datetime
import requests now_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(now_time)# 接口地址和token信息
url = 'https://oapi.dingtalk.com/robot/send?access_token=************'
keywords = 'DataWorks'# 钉钉机器人,发送消息
def dd_robot(url:str, text_content: str, title_content: str):HEADERS = {"Content-Type": "application/json;charset=utf-8"}#content里面要设置关键字data_info = {"msgtype": "markdown","markdown": {"title": title_content,"text": text_content,}#这是配置需要@的人,"at": {"atMobiles": ["15xxxxxx06"], "atUserIds": ["user123"], "isAtAll": False} }value = json.dumps(data_info)response = requests.post(url,data=value,headers=HEADERS)if response.json()['errmsg']!='ok':print(response.text)# 主函数
if __name__ == '__main__':# 编辑需要推送的信息title_content = keywords + ":百度百科 \n"text_content = "#### 百度百科 \n 百度公司推出的网络百科全书 \n ![screenshot](https://bkimg.cdn.bcebos.com/pic/a2cc7cd98d1001e93901c06c39446cec54e736d13937?x-bce-process=image/format,f_auto/quality,Q_70/resize,m_lfit,limit_1,w_536) \n [百度百科](https://baike.baidu.com/item/%E7%99%BE%E5%BA%A6%E7%99%BE%E7%A7%91?fromModule=lemma_search-box) \n @15xxxxxx06"# 调用推送函数, 推送消息到钉钉群dd_robot(url, text_content, title_content)
推送结果1
百度百科
百度公司推出的网络百科全书
[图片]
百度百科
@Johnathan

在这里插入图片描述

推送代码案例2
python">import json
from datetime import datetime
import requests 
from odps import ODPS
import pandas as pdnow_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(now_time)# 接口地址和token信息
url = 'https://oapi.dingtalk.com/robot/send?access_token=*************'
keywords = 'DataWorks'# 钉钉机器人,发送消息
def dd_robot(url:str, text_content: str, title_content: str):HEADERS = {"Content-Type": "application/json;charset=utf-8"}#content里面要设置关键字data_info = {"msgtype": "markdown","markdown": {"title": title_content,"text": text_content,}#这是配置需要@的人#,"at": {"atMobiles": ["15xxxxxx06","18xxxxxx1"], "atUserIds": ["user123"], "isAtAll": False} }value = json.dumps(data_info)response = requests.post(url,data=value,headers=HEADERS)if response.json()['errmsg']!='ok':print(response.text)# 执行SQL体
sql = """SELECT stat_day, COUNT(1) AS ctFROM st_bi_stock_sales_ratio_msku_hi WHERE pt = MAX_PT('st_bi_stock_sales_ratio_msku_hi')GROUP BY stat_dayORDER BY stat_day
"""# 主函数
if __name__ == '__main__':# 执行SQL并读取结果with odps.execute_sql(sql).open_reader() as reader:df = reader.to_pandas()  # 转换为Pandas DataFrame#print(df)# 转换为Markdown表格格式markdown_table = "| " + " | ".join(df.columns) + " |\n"markdown_table += "| " + " | ".join(["---"] * len(df.columns)) + " |\n"for row in df.itertuples(index=False):markdown_table += "| " + " | ".join(map(str, row)) + " |\n"# 打印Markdown格式表格#print(markdown_table)# 编辑需要推送的信息title_content = keywords + ":数据量查看 \n"text_content = "#### 库销比统计 \n 库销比统计近4天数据量(" + now_time + ") \n"text_content += markdown_table + "\n"text_content += "By Johnathan \n"# 调用推送函数, 推送消息到钉钉群dd_robot(url, text_content, title_content)
推送结果2
库销比统计
库销比统计近4天数据量(2024-12-04 16:47:41)
stat_day ct
2024-12-01 10603
2024-12-02 10612
2024-12-03 10595 
2024-12-04 10625
By Johnathan

在这里插入图片描述

推送代码案例2 (版本2)
python">import json
from datetime import datetime
import requests 
from odps import ODPS
import pandas as pd
from tabulate import tabulatenow_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(now_time)# 接口地址和token信息
url = 'https://oapi.dingtalk.com/robot/send?access_token=**********'
keywords = 'DataWorks'# 钉钉机器人,发送消息
def dd_robot(url:str, text_content: str, title_content: str):HEADERS = {"Content-Type": "application/json;charset=utf-8"}#content里面要设置关键字data_info = {"msgtype": "markdown","markdown": {"title": title_content,"text": text_content,}#这是配置需要@的人#,"at": {"atMobiles": ["15xxxxxx06","18xxxxxx1"], "atUserIds": ["user123"], "isAtAll": False} }value = json.dumps(data_info)response = requests.post(url,data=value,headers=HEADERS)if response.json()['errmsg']!='ok':print(response.text)# 执行SQL体
sql = """SELECT stat_day, COUNT(1) AS ctFROM st_bi_stock_sales_ratio_msku_hi WHERE pt = '2024121312'GROUP BY stat_dayORDER BY stat_day
"""# 主函数
if __name__ == '__main__':# 执行SQL并读取结果with odps.execute_sql(sql).open_reader() as reader:df = reader.to_pandas()  # 转换为Pandas DataFrame#print(df)# 转换为Markdown格式markdown_table = df.to_markdown()# 输出Markdown表格print(markdown_table)# 编辑需要推送的信息title_content = keywords + ":数据量查看 \n"text_content = "#### 库销比统计 \n 库销比统计近4天数据量(" + now_time + ") \n"text_content += markdown_table + "\n"text_content += "By Johnathan \n"# 调用推送函数, 推送消息到钉钉群dd_robot(url, text_content, title_content)
推送结果2(版本2)

在这里插入图片描述

参数说明
python">{"msgtype": "markdown","markdown": {"title":"百度百科","text": "#### 百度百科 \n 百度公司推出的网络百科全书 \n ![screenshot](https://bkimg.cdn.bcebos.com/pic/a2cc7cd98d1001e93901c06c39446cec54e736d13937?x-bce-process=image/format,f_auto/quality,Q_70/resize,m_lfit,limit_1,w_536) \n [百度百科](https://baike.baidu.com/item/%E7%99%BE%E5%BA%A6%E7%99%BE%E7%A7%91?fromModule=lemma_search-box) \n @15xxxxxx06"},"at": {"atMobiles": ["15xxxxxx06"],"atUserIds": ["user123"],"isAtAll": false}}
参数是否必填类型说明
msgtype是Stringmarkdown。
markdown.title是String首屏会话透出的展示内容。
markdown.text是StringMarkdown格式的消息内容。
at.atMobiles否Array被@人的手机号。消息内容text内要带上"@手机号"，跟atMobiles参数结合使用，才有@效果。
at.atUserIds否Array被@人的用户userid。
at.isAtAll否Boolean@所有人是true，否则为false。
keywords是String钉钉机器人的关键词

目前只支持Markdown语法的子集，支持的元素如下：

标题
# 一级标题
## 二级标题
### 三级标题
#### 四级标题
##### 五级标题
###### 六级标题引用
> A man who stands for nothing will fall for anything.文字加粗、斜体
**bold**
*italic*链接
[this is a link](https://www.dingtalk.com/)图片
![](http://name.com/pic.jpg)无序列表
- item1
- item2有序列表
1. item1
2. item2

4- 整体跳转ActionCard类型

推送代码案例
python">import json
from datetime import datetime
import requests now_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(now_time)# 接口地址和token信息
url = 'https://oapi.dingtalk.com/robot/send?access_token=**********'
keywords = 'DataWorks'# 钉钉机器人,发送消息
def dd_robot(url:str, text_content: str, title_content: str, single_title: str, single_url: str):HEADERS = {"Content-Type": "application/json;charset=utf-8"}#content里面要设置关键字data_info = {"msgtype": "actionCard","actionCard": {"title": title_content,"text": text_content,"singleTitle" : single_title,"singleURL" : single_url}}value = json.dumps(data_info)response = requests.post(url,data=value,headers=HEADERS)if response.json()['errmsg']!='ok':print(response.text)# 主函数
if __name__ == '__main__':# 编辑需要推送的信息title_content = keywords + ": CASE WHEN用法 \n"text_content = "![screenshot](https://img-blog.csdnimg.cn/img_convert/f63f1c575e3c48d9a3f7e4ba9bd460d2.png) \n "text_content += "#### SQL中的 CASE WHEN用法详解 \n CASE WHEN 语句提供了一种在 SQL 查询中根据不同条件执行逻辑的灵活方法。它可用于简单的条件检查，也可用于复杂的逻辑操作。使用 CASE WHEN 可以使查询更具可读性，并且可以减少在应用程序代码中进行逻辑操作的需要。\n"single_title = "阅读全文"single_url = "https://blog.csdn.net/Taerge0110/article/details/136261423?spm=1001.2014.3001.5502"# 调用推送函数, 推送消息到钉钉群dd_robot(url, text_content, title_content, single_title, single_url)
推送结果

在这里插入图片描述

参数说明
python">{"msgtype": "actionCard","actionCard": {"title": "CASE WHEN用法", "text": "![screenshot](https://img-blog.csdnimg.cn/img_convert/f63f1c575e3c48d9a3f7e4ba9bd460d2.png) \n #### SQL中的 CASE WHEN用法详解 \n CASE WHEN 语句提供了一种在 SQL 查询中根据不同条件执行逻辑的灵活方法。它可用于简单的条件检查，也可用于复杂的逻辑操作。使用 CASE WHEN 可以使查询更具可读性，并且可以减少在应用程序代码中进行逻辑操作的需要。\n","singleTitle" : "阅读全文","singleURL" : "https://blog.csdn.net/Taerge0110/article/details/136261423?spm=1001.2014.3001.5502"}
}
参数是否必填类型说明
msgtype是StringactionCard。
actionCard.title是String首屏会话透出的展示内容。
actionCard.text是Stringmarkdown格式的消息内容。
actionCard.singleTitle是String单个按钮的标题。
actionCard.singleURL是String单个按钮的跳转链接。
keywords是String钉钉机器人的关键词

5- 推送文件链接供下载

推送背景
1. 需要推送的结果比较多, 不方便展示 ;
2. 推送结果需要保存下载 ;
推送代码案例

说明: 基于DataWorks, 使用Python 查询数据库的数据, 将查询结果保存到OSS, 将OSS文件链接 通过钉钉机器人的方式推送到钉钉群里

python">import json
from datetime import datetime
import requests 
from odps import ODPS
import pandas as pd
import oss2
from io import StringIOnow_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(now_time)# 接口地址和token信息
url = 'https://oapi.dingtalk.com/robot/send?access_token=*********'
keywords = 'DataWorks'# 阿里云 OSS 配置
access_key_id = "your-access-key-id"
access_key_secret = "your-access-key-secret"
bucket_name = "your-bucket-name"   
endpoint = "https://oss-cn-your-region.aliyuncs.com"  # 替换为你的OSS区域# 初始化 OSS 连接
auth = oss2.Auth(access_key_id, access_key_secret)
bucket = oss2.Bucket(auth, endpoint, bucket_name)# 钉钉机器人,发送消息
def dd_robot(url:str, text_content: str, title_content: str):HEADERS = {"Content-Type": "application/json;charset=utf-8"}#content里面要设置关键字data_info = {"msgtype": "markdown","markdown": {"title": title_content,"text": text_content,}#这是配置需要@的人#,"at": {"atMobiles": ["15xxxxxx06","18xxxxxx1"], "atUserIds": ["user123"], "isAtAll": False} }value = json.dumps(data_info)response = requests.post(url,data=value,headers=HEADERS)if response.json()['errmsg']!='ok':print(response.text)# 执行SQL体
sql = """
SELECT platform_id, platform_name, COUNT(1) AS ct  , SUM(wait_arrival_quantity) AS wait_arrival_quantity   -- 1.采购中, SUM(no_qc_quantity) AS no_qc_quantity         -- 2.待质检, SUM(doing_qc_quantity) AS doing_qc_quantity   -- 3.质检, SUM(inspected_quantity) AS inspected_quantity -- 已质检, SUM(usable_quantity) AS usable_quantity       -- 5.良品, SUM(testing_quantity) AS testing_quantity     -- 4.检测良品, SUM(occupy_quantity) AS occupy_quantity       -- 6.占用, SUM(imperfect_quantity) AS imperfect_quantity -- 7.次品, SUM(shipping_quantity) AS shipping_quantity   -- 8.发货中, SUM(allocation_quantity) AS allocation_quantity -- 9.调拨在途, SUM(repair_quantity) AS repair_quantity   -- 10.返修在途, SUM(interception_quantity) AS interception_quantity -- 截留在途
FROM dws_wh_scm_inventory_details_df
WHERE pt = '20241212' AND store_name IS NOT NULL 
GROUP BY platform_id, platform_name
"""# 主函数
if __name__ == '__main__':# 执行SQL并读取结果with odps.execute_sql(sql).open_reader() as reader:df = reader.to_pandas()  # 转换为Pandas DataFrame#print(df)# 将 DataFrame 转为 CSV 格式字符串csv_buffer = StringIO()df.to_csv(csv_buffer, index=False)csv_data = csv_buffer.getvalue()#print(csv_data)# 上传文件到 OSSoss_file_path = "data-test/***/scm_inventory.csv"try:bucket.put_object(oss_file_path, csv_data)print(f"文件已成功上传到 OSS: {oss_file_path}")except Exception as e:print(f"文件上传失败: {e}")# OSS 文件的对象键（路径）object_key = oss_file_path# 设置签名 URL 的有效期（单位：秒）expire_seconds = 3600  # 1小时# 生成临时访问链接res_url = bucket.sign_url('GET', object_key, expire_seconds)#print(f"OSS文件的访问地址（有效期{expire_seconds}秒）：\n{url_1}")# 编辑需要推送的信息title_content = keywords + ":数据量查看 \n"text_content = "#### 库存明细数据统计:  \n " text_content += "![screenshot](https://img-blog.csdnimg.cn/direct/03debdc1cb6941949c124770c8d2b24f.png) \n"  # 文档内容可自定义text_content += "[下载链接](" + res_url + ")  \n"text_content += "By Johnathan \n"# 调用推送函数, 推送消息到钉钉群dd_robot(url, text_content, title_content)
推送结果

在这里插入图片描述

在这里插入图片描述
在这里插入图片描述

结语:

持续更新

欢迎交流

end

end


http://www.ppmy.cn/server/150032.html

相关文章

DevOps系统设计和技术选型

DevOps系统设计和技术选型

命名是一件痛苦的事情，除非你不想要一个好名字。 我正在做的这个管理系统叫什么合适，或者是什么类型的系统，想去想来不知所措，后来想想这么小的东西纠结什么，先从小的细节一点点来，能用就行，就用…
阅读更多...
R语言——缺失值处理

R语言——缺失值处理

目录 缺失数据 1 R中的缺失值 2查看缺失值is.na() 3 去除缺失值 1 R中的缺失值 在R中，NA代表缺失值，NA是不可用（可能是0，可能是其他值，NA是未知的），notavailable的简称，用来存储…
阅读更多...
Typora教程

Typora教程

目录 一、下载安装 二、激活 1.激活 2.解决激活提示窗口 一、下载安装 去官网下载Typora安装，我的是1.9.5版本 二、激活 1.激活 根据路径找到Typora/resources/page-dist/static/js 使用记事本打开LicenseIndex文件，如下图： 按住快捷…
阅读更多...
Simdroid-EC：液冷仿真新星，助力新能源汽车电机控制器高效散热

Simdroid-EC：液冷仿真新星，助力新能源汽车电机控制器高效散热

近年来，新能源电动车的销量呈现出快速增长的态势。据统计，2024 年1-10月中国新能源汽车销量达728万辆，同比增长37.8%。 电机控制器在新能源汽车中对于保障动力和安全性能扮演着至关重要的角色，其核心部件IGBT（绝缘栅双…
阅读更多...
基于Qwen2-VL模型针对LaTeX OCR任务进行微调训练 - 多图推理

基于Qwen2-VL模型针对LaTeX OCR任务进行微调训练 - 多图推理

基于Qwen2-VL模型针对LaTeX OCR任务进行微调训练 - 多图推理 flyfish 基于Qwen2-VL模型针对LaTeX_OCR任务进行微调训练_-_LoRA配置如何写 基于Qwen2-VL模型针对LaTeX_OCR任务进行微调训练_-_单图推理 基于Qwen2-VL模型针对LaTeX_OCR任务进行微调训练_-_原模型_单图推理 基于Q…
阅读更多...
webform自动提交的情况

webform自动提交的情况

webform按回车自动提交的情况 当页面只有一个textbox控件的时候, <asp:TextBox runat"server" ID"strStandardName" />当页面有button按钮的时候 <asp:Button ID"Button1" runat"server" Text"Button" OnClick&q…
阅读更多...
使用秘钥登录服务器

使用秘钥登录服务器

在我们测试或生产环境中&#xff0c;为了服务器安全性&#xff0c;有时可能需要以 SSH 密钥的方式登录服务器&#xff0c;接下来&#xff0c;将演示如何通过 SSH 私钥的方式来远程服务器。 一、远程服务器生成密钥对 1、首先在目标远程服务器下生成 SSH 密钥对 ssh-keygen然…
阅读更多...
Linux下禁止root远程登录访问

Linux下禁止root远程登录访问

开始讲故事 Long long ago&#xff0c; Linux远程访问方式有telnet、ssh两种协议&#xff1b;有人可能还会说vnc和rdp协议方式&#xff0c;后面这两种主要是可视化桌面场景下的&#xff0c;并非主流。 时过境迁&#xff0c;telnet因安全性低逐渐被禁用淘汰&#xff0c;最后就…
阅读更多...
最新文章

Copyright @ 2022~2023