pip install requests dingtalk-sdk

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)

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,  # 應(yīng)用的 AgentId
        "userid_list": userid_list,  # 接收消息的用戶ID列表，用逗號(hào)分隔
        "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列表，用逗號(hào)分隔
content = "這是一條測(cè)試消息"
response = send_text_message(access_token, agent_id, userid_list, content)
print("Message Sent:", response)

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)

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()

# 替換為你的部門(mén)ID
department_id = "your_department_id"
user_details = get_department_user_details(access_token, department_id)
print("User Details:", user_details)

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,  # 審批模板的唯一標(biāo)識(shí)
        "originator_user_id": originator_user_id,  # 發(fā)起審批的用戶ID
        "dept_id": dept_id,  # 發(fā)起審批的部門(mén)ID
        "form_component_values": form_component_values  # 審批表單內(nèi)容
    }
    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)

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")

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

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

import os

# 從環(huán)境變量中讀取敏感信息
app_key = os.getenv("DINGTALK_APP_KEY")
app_secret = os.getenv("DINGTALK_APP_SECRET")

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

# 配置文件示例
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"
    }
}

# 根據(jù)環(huán)境選擇配置
env = "test"  # 或 "production"
app_key = config[env]["app_key"]
app_secret = config[env]["app_secret"]
agent_id = config[env]["agent_id"]