Python实现钉钉自动化完整指南

2025-03-23 09:51:21 作者：老胖闲聊

本文详细介绍如何使用 Python 实现钉钉自动化,包括消息发送、部门管理、审批流程触发等功能,内容涵盖实现步骤、前置条件、依赖项以及注意事项,帮助快速上手并避免常见问题,需要的朋友可以参考下

实现步骤

1. 安装依赖项

在开始之前，需要安装必要的 Python 库。常用的库包括 requests 和 dingtalk-sdk。

pip install requests dingtalk-sdk

2. 获取钉钉开发者权限

在使用钉钉 API 之前，需要在钉钉开发者平台上创建一个应用，并获取 AppKey 和 AppSecret。

登录钉钉开发者平台。
创建一个企业内部应用或第三方应用。
获取应用的 AppKey 和 AppSecret。
设置应用的权限（例如：消息发送、通讯录管理、审批流等）。

3. 获取 Access Token

钉钉 API 的调用需要 Access Token，可以通过 AppKey 和 AppSecret 获取。

import requests

def get_access_token(app_key, app_secret):
    url = "https://oapi.dingtalk.com/gettoken"
    params = {
        "appkey": app_key,
        "appsecret": app_secret
    }
    response = requests.get(url, params=params)
    return response.json().get("access_token")

# 替换为你的 AppKey 和 AppSecret
app_key = "your_app_key"
app_secret = "your_app_secret"
access_token = get_access_token(app_key, app_secret)
print("Access Token:", access_token)

4. 发送消息

钉钉支持多种消息类型（如文本、链接、Markdown 等）。以下是一个发送文本消息的示例：

def send_text_message(access_token, agent_id, userid_list, content):
    url = "https://oapi.dingtalk.com/topapi/message/corpconversation/asyncsend_v2"
    headers = {
        "Content-Type": "application/json"
    }
    data = {
        "agent_id": agent_id,  # 应用的 AgentId
        "userid_list": userid_list,  # 接收消息的用户ID列表，用逗号分隔
        "msg": {
            "msgtype": "text",
            "text": {
                "content": content
            }
        }
    }
    params = {
        "access_token": access_token
    }
    response = requests.post(url, headers=headers, json=data, params=params)
    return response.json()

# 替换为你的 AgentId 和用户ID列表
agent_id = "your_agent_id"
userid_list = "user1,user2"  # 接收消息的用户ID列表，用逗号分隔
content = "这是一条测试消息"
response = send_text_message(access_token, agent_id, userid_list, content)
print("Message Sent:", response)

5. 获取部门列表

可以通过钉钉 API 获取公司内部的部门列表。

def get_department_list(access_token):
    url = "https://oapi.dingtalk.com/department/list"
    params = {
        "access_token": access_token
    }
    response = requests.get(url, params=params)
    return response.json()

department_list = get_department_list(access_token)
print("Department List:", department_list)

6. 获取部门用户详情

通过部门 ID 获取该部门下的用户详情。

def get_department_user_details(access_token, department_id):
    url = "https://oapi.dingtalk.com/user/listbypage"
    params = {
        "access_token": access_token,
        "department_id": department_id,
        "offset": 0,
        "size": 100
    }
    response = requests.get(url, params=params)
    return response.json()

# 替换为你的部门ID
department_id = "your_department_id"
user_details = get_department_user_details(access_token, department_id)
print("User Details:", user_details)

7. 处理审批流程

钉钉的审批流程可以通过 API 进行触发和查询。以下是一个触发审批流程的示例：

def trigger_approval_process(access_token, process_code, originator_user_id, dept_id, form_component_values):
    url = "https://oapi.dingtalk.com/topapi/processinstance/create"
    headers = {
        "Content-Type": "application/json"
    }
    data = {
        "process_code": process_code,  # 审批模板的唯一标识
        "originator_user_id": originator_user_id,  # 发起审批的用户ID
        "dept_id": dept_id,  # 发起审批的部门ID
        "form_component_values": form_component_values  # 审批表单内容
    }
    params = {
        "access_token": access_token
    }
    response = requests.post(url, headers=headers, json=data, params=params)
    return response.json()

# 替换为你的审批模板信息
process_code = "your_process_code"
originator_user_id = "your_user_id"
dept_id = "your_dept_id"
form_component_values = [
    {"name": "field1", "value": "value1"},
    {"name": "field2", "value": "value2"}
]
response = trigger_approval_process(access_token, process_code, originator_user_id, dept_id, form_component_values)
print("Approval Process Triggered:", response)

8. 错误处理

在实际使用中，可能会遇到网络问题或 API 调用频率限制等问题。建议在代码中加入错误处理机制。

import time

def safe_send_message(access_token, agent_id, userid_list, content, retries=3):
    for i in range(retries):
        try:
            response = send_text_message(access_token, agent_id, userid_list, content)
            if response.get("errcode") == 0:
                return response
            else:
                print(f"Error: {response.get('errmsg')}")
        except Exception as e:
            print(f"Exception: {e}")
        time.sleep(2)  # 等待2秒后重试
    return None

response = safe_send_message(access_token, agent_id, userid_list, content)
if response:
    print("Message Sent Successfully")
else:
    print("Failed to Send Message")

前置条件

钉钉开发者账号：
- 需要有一个钉钉开发者账号，并创建一个应用。
- 获取应用的 AppKey 和 AppSecret。
应用权限：
- 确保应用已开通相关权限（如消息发送、通讯录管理、审批流等）。
Python 环境：
- 确保已安装 Python 3.6 及以上版本。
- 安装必要的依赖库（如 requests 和 dingtalk-sdk）。

依赖项

以下是实现钉钉自动化所需的依赖项：

Python 库：
- requests：用于发送 HTTP 请求。
- dingtalk-sdk：钉钉官方提供的 Python SDK（可选）。
钉钉 API 权限：
- 消息发送权限。
- 通讯录管理权限。
- 审批流权限（如果需要处理审批流程）。
网络环境：
- 确保服务器或本地环境可以访问钉钉 API（https://oapi.dingtalk.com）。

注意事项

1. Access Token 的管理

有效期：钉钉的 Access Token 有效期为 7200 秒（2 小时），过期后需要重新获取。
缓存机制：建议将 Access Token 缓存起来（如存储在内存或 Redis 中），避免频繁调用获取接口。
刷新策略：在每次调用 API 前检查 Access Token 是否过期，如果过期则重新获取。

import time

# 缓存 Access Token
access_token_cache = {
    "token": None,
    "expires_at": 0
}

def get_access_token_with_cache(app_key, app_secret):
    now = time.time()
    if access_token_cache["token"] and access_token_cache["expires_at"] > now:
        return access_token_cache["token"]
    
    # 重新获取 Access Token
    access_token = get_access_token(app_key, app_secret)
    access_token_cache["token"] = access_token
    access_token_cache["expires_at"] = now + 7200  # 有效期 7200 秒
    return access_token

2. API 调用频率限制

频率限制：钉钉 API 对调用频率有限制，具体限制因接口而异。例如，消息发送接口的默认限制为 20 次/秒。
错误处理：如果触发频率限制，钉钉会返回错误码 43004，建议在代码中加入重试机制和延迟。

import time

def safe_send_message(access_token, agent_id, userid_list, content, retries=3):
    for i in range(retries):
        try:
            response = send_text_message(access_token, agent_id, userid_list, content)
            if response.get("errcode") == 0:
                return response
            elif response.get("errcode") == 43004:  # 频率限制
                print("Rate limit exceeded, retrying...")
                time.sleep(2)  # 等待 2 秒后重试
            else:
                print(f"Error: {response.get('errmsg')}")
        except Exception as e:
            print(f"Exception: {e}")
        time.sleep(1)  # 每次重试前等待 1 秒
    return None

3. 消息发送的权限和范围

用户权限：确保消息接收者（userid_list）在应用的可见范围内。
消息类型：钉钉支持多种消息类型（如文本、链接、Markdown 等），需根据需求选择合适的类型。
消息长度：文本消息的内容长度不能超过 5000 字符，否则会被截断。

4. 审批流程的配置

审批模板：在触发审批流程前，需在钉钉后台配置审批模板，并获取 process_code。
表单字段：审批表单的字段名称和值需与模板中的字段一致，否则会触发错误。
审批权限：确保发起审批的用户（originator_user_id）有权限发起该审批。

5. 数据安全性

敏感信息保护：避免在代码中硬编码敏感信息（如 AppKey、AppSecret），建议使用环境变量或配置文件。
HTTPS 加密：钉钉 API 仅支持 HTTPS 协议，确保代码使用 HTTPS 调用 API。

import os

# 从环境变量中读取敏感信息
app_key = os.getenv("DINGTALK_APP_KEY")
app_secret = os.getenv("DINGTALK_APP_SECRET")

6. 日志记录

日志输出：建议在代码中加入日志记录，方便排查问题和监控运行状态。
错误日志：记录 API 调用失败时的错误信息（如错误码、错误消息）。

import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

def send_text_message_with_logging(access_token, agent_id, userid_list, content):
    try:
        response = send_text_message(access_token, agent_id, userid_list, content)
        if response.get("errcode") == 0:
            logging.info("Message sent successfully")
        else:
            logging.error(f"Failed to send message: {response.get('errmsg')}")
        return response
    except Exception as e:
        logging.error(f"Exception occurred: {e}")
        return None

7. 测试环境与生产环境分离

测试环境：在开发阶段，建议使用钉钉的测试环境进行调试，避免影响生产数据。
环境切换：通过配置文件或环境变量区分测试环境和生产环境。

# 配置文件示例
config = {
    "test": {
        "app_key": "test_app_key",
        "app_secret": "test_app_secret",
        "agent_id": "test_agent_id"
    },
    "production": {
        "app_key": "prod_app_key",
        "app_secret": "prod_app_secret",
        "agent_id": "prod_agent_id"
    }
}

# 根据环境选择配置
env = "test"  # 或 "production"
app_key = config[env]["app_key"]
app_secret = config[env]["app_secret"]
agent_id = config[env]["agent_id"]

8. 钉钉 API 的版本兼容性

API 版本：钉钉 API 会不断更新，建议定期查看钉钉开放平台文档，确保使用的 API 版本是最新的。
废弃接口：如果使用旧版 API，需注意是否已被废弃，并及时迁移到新版 API。

9. 异步调用与回调

异步调用：部分钉钉 API 支持异步调用（如消息发送），需处理回调通知。
回调配置：在钉钉开发者平台配置回调地址，并实现回调接口以接收异步结果。

10. 性能优化

批量操作：如果需要对大量用户或部门进行操作，建议使用批量接口，减少 API 调用次数。
并发控制：在高并发场景下，需控制 API 调用频率，避免触发频率限制。

总结

通过以上步骤和注意事项，可以使用 Python 实现钉钉自动化，包括消息发送、部门管理、审批流程触发等功能。钉钉 API 功能强大，支持多种场景的自动化操作。可以根据具体需求，进一步探索和使用更多的 API 接口。

以上就是Python实现钉钉自动化完整指南的详细内容，更多关于Python钉钉自动化的资料请关注脚本之家其它相关文章！