Python源码转易语言实现钉钉stream机qi人（求助）

renjinyu · 发表于 2025-8-18 16:53:02

# !/usr/bin/env python

import argparse
import logging
from dingtalk_stream import AckMessage
import dingtalk_stream
from alibabacloud_dingtalk.oauth2_1_0.client import Client as dingtalkoauth2_1_0Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_dingtalk.oauth2_1_0 import models as dingtalkoauth_2__1__0_models
from alibabacloud_dingtalk.robot_1_0.client import Client as dingtalkrobot_1_0Client
from alibabacloud_dingtalk.robot_1_0 import models as dingtalkrobot__1__0_models
from alibabacloud_tea_util import models as util_models
import time

_token_cache = {"token": None, "expire": 0}

def setup_logger():
logger = logging.getLogger()
handler = logging.StreamHandler()
handler.setFormatter(
      logging.Formatter('%(asctime)s %(name)-8s %(levelname)-8s %(message)s [%(filename)s:%(lineno)d]'))
logger.addHandler(handler)
logger.setLevel(logging.INFO)
return logger

def define_options():
parser = argparse.ArgumentParser()
parser.add_argument(
      '--client_id', dest='client_id', required=True,
      help='app_key or suite_key from https://open-dev.digntalk.com'
)
parser.add_argument(
      '--client_secret', dest='client_secret', required=True,
      help='app_secret or suite_secret from https://open-dev.digntalk.com'
)
parser.add_argument(
      '--robot_code', dest='robot_code', required=True,
      help='robot_code from https://open-dev.dingtalk.com'
)
options = parser.parse_args()
return options

def get_token(options):
"""
使用钉钉SDK获取access_token，带本地缓存，2小时有效，提前200秒刷新。
:param options: 命令行参数对象，包含 client_id, client_secret
:return: access_token字符串，获取失败返回None
"""
now = time.time()
if _token_cache["token"] and now < _token_cache["expire"]:
      return _token_cache["token"]
config = open_api_models.Config()
config.protocol = 'https'
config.region_id = 'central'
client = dingtalkoauth2_1_0Client(config)
get_access_token_request = dingtalkoauth_2__1__0_models.GetAccessTokenRequest(
      app_key=options.client_id,
      app_secret=options.client_secret
)
try:
      response = client.get_access_token(get_access_token_request)
      token = getattr(response.body, "access_token", None)
      expire_in = getattr(response.body, "expire_in", 7200)
      if token:
         _token_cache["token"] = token
         _token_cache["expire"] = now + expire_in - 200  # 提前200秒刷新
      return token
except Exception as err:
      print(f"获取token失败: {err}")
      return None

def send_robot_group_message(access_token, open_conversation_id, options):
"""
使用钉钉SDK发送群消息
:param access_token: 已获取的 access_token
:param open_conversation_id: 群会话ID
:param options: 命令行参数对象，包含 robot_code
:return: 发送结果或 None
"""
robot_code = options.robot_code
msg_param = '{"content":"python-getting-start say：hello"}'
msg_key = 'sampleText'
config = open_api_models.Config()
config.protocol = 'https'
config.region_id = 'central'
client = dingtalkrobot_1_0Client(config)
org_group_send_headers = dingtalkrobot__1__0_models.OrgGroupSendHeaders()
org_group_send_headers.x_acs_dingtalk_access_token = access_token
org_group_send_request = dingtalkrobot__1__0_models.OrgGroupSendRequest(
      msg_param=msg_param,
      msg_key=msg_key,
      open_conversation_id=open_conversation_id,
      robot_code=robot_code
)
try:
      response = client.org_group_send_with_options(
         org_group_send_request,
         org_group_send_headers,
         util_models.RuntimeOptions()
      )
      print("消息发送成功，返回：", response)
      return response
except Exception as err:
      print(f"发送群消息失败: {err}")
      return None

class EchoTextHandler(dingtalk_stream.ChatbotHandler):
def __init__(self, logger: logging.Logger = None, options=None):
      super(dingtalk_stream.ChatbotHandler, self).__init__()
      self.logger = logger
      self.options = options

async def process(self, callback: dingtalk_stream.CallbackMessage):
      incoming_message = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
      open_conversation_id = incoming_message.conversation_id
      access_token = get_token(self.options)
      if access_token:
         send_robot_group_message(access_token, open_conversation_id, self.options)
      else:
         print("access_token 获取失败")
      return AckMessage.STATUS_OK, 'OK'

def main():
logger = setup_logger()
options = define_options()
credential = dingtalk_stream.Credential(options.client_id, options.client_secret)
client = dingtalk_stream.DingTalkStreamClient(credential)
client.register_callback_handler(
      dingtalk_stream.chatbot.ChatbotMessage.TOPIC,
      EchoTextHandler(logger, options)
)
client.start_forever()

if __name__ == '__main__':
main()

有大佬可以转的吗

小李吃不饱 · 发表于 2025-8-18 17:19:03

窗口程序集名	保留	保留	备注
程序集1
变量名	类型	数组	备注
token_cache	文本型	2	[0]为token，[1]为过期时间
logger	日志类
client_id	文本型
client_secret	文本型
robot_code	文本型

子程序名	返回值类型	公开	备注
_启动子程序	整数型		本子程序在程序启动后最先执行

变量名	类型	静态	数组	备注
命令行参数	文本型		0
i	整数型
参数名	文本型
参数值	文本型

' 初始化日志
logger.初始化 ()
logger.设置日志级别 (日志级别_信息)
' 解析命令行参数
命令行参数＝取命令行参数 ()

计次循环首 (取数组成员数 (命令行参数), i)

如果真 (取文本左边 (命令行参数 [i], 2) ＝ "--")

参数名＝取文本中间 (命令行参数 [i], 3, 取文本长度 (命令行参数 [i]) － 2)

如果真 (i ＜取数组成员数 (命令行参数))

参数值＝命令行参数 [i ＋ 1]

如果 (参数名＝ "client_id")

client_id ＝参数值

.否则如果 (参数名＝ "client_secret")

client_secret ＝参数值

.否则如果 (参数名＝ "robot_code")

robot_code ＝参数值

logger.写日志 (日志级别_警告, "未知参数: " ＋参数名)

计次循环尾 ()
' 检查必要参数

如果 (client_id ＝ "" 或 client_secret ＝ "" 或 robot_code ＝ "")

logger.写日志 (日志级别_错误, "缺少必要参数，请提供client_id、client_secret和robot_code")

返回 (1)

' 初始化token缓存
token_cache [1] ＝ "0"
' 启动钉钉消息监听（这里使用模拟，实际需要根据钉钉stream协议实现）
logger.写日志 (日志级别_信息, "钉钉机qi人程序启动成功")
模拟消息监听 ()
返回 (0)

子程序名	返回值类型	公开	备注
模拟消息监听

' 注意：这里只是模拟消息监听，实际需要根据钉钉stream协议实现WebSocket连接

变量名	类型	静态	数组	备注
模拟会话ID	文本型

模拟会话ID ＝ "模拟群会话ID_123456"
logger.写日志 (日志级别_信息, "开始监听消息...")
' 模拟接收到消息

计次循环首 (5, )

处理接收到的消息 (模拟会话ID)

延迟 (5000)

计次循环尾 ()

子程序名	返回值类型	公开	备注
处理接收到的消息
参数名	类型	参考	可空	数组	备注
open_conversation_id	文本型

变量名	类型	静态	数组	备注
access_token	文本型

logger.写日志 (日志级别_信息, "收到消息，会话ID: " ＋ open_conversation_id)
access_token ＝获取Token ()

如果 (access_token ≠ "")

发送机qi人群消息 (access_token, open_conversation_id)

logger.写日志 (日志级别_错误, "获取access_token失败")

子程序名	返回值类型	公开	备注
获取Token	文本型

变量名	类型	静态	数组	备注
现在时间	整数型
过期时间	整数型
响应文本	文本型
json解析	类_json解析
状态码	整数型
token	文本型
过期时间秒	整数型

' 检查缓存
现在时间＝取现行时间戳 ()
过期时间＝到整数 (token_cache [1])

如果 (token_cache [0] ≠ "" 且现在时间＜过期时间)

返回 (token_cache [0])

' 缓存失效，重新获取
logger.写日志 (日志级别_信息, "获取新的access_token...")
响应文本＝网页_访问S ("https://api.dingtalk.com/v1.0/oauth2/accessToken", , "grant_type=client_credentials&client_id=" ＋编码_URL编码 (client_id) ＋ "&client_secret=" ＋编码_URL编码 (client_secret), , , 状态码)

如果 (状态码 ≠ 200)

logger.写日志 (日志级别_错误, "获取token失败，状态码: " ＋到文本 (状态码) ＋ "，响应: " ＋响应文本)

返回 ("")

json解析.解析 (响应文本)
token ＝ json解析.取属性值 (“accessToken”)
过期时间秒＝到整数 (json解析.取属性值 (“expireIn”))

如果 (token ≠ "")

' 提前200秒刷新

token_cache [0] ＝ token

token_cache [1] ＝到文本 (现在时间＋过期时间秒－ 200)

logger.写日志 (日志级别_信息, "获取access_token成功")

返回 (token)

logger.写日志 (日志级别_错误, "解析token失败，响应: " ＋响应文本)

返回 ("")

子程序名	返回值类型	公开	备注
发送机qi人群消息
参数名	类型	参考	可空	数组	备注
access_token	文本型
open_conversation_id	文本型

变量名	类型	静态	数组	备注
响应文本	文本型
状态码	整数型
请求头	文本型
请求数据	文本型
json解析	类_json解析

logger.写日志 (日志级别_信息, "发送群消息到会话: " ＋ open_conversation_id)
' 构建请求头
请求头＝ "Content-Type: application/json" ＋ #换行符＋ "x-acs-dingtalk-access-token: " ＋ access_token
' 构建请求数据
json解析.清空 ()
json解析.添加键值对 (“msgParam”, “{"content":"易语言机qi人 say：hello"}”)
json解析.添加键值对 (“msgKey”, “sampleText”)
json解析.添加键值对 (“openConversationId”, open_conversation_id)
json解析.添加键值对 (“robotCode”, robot_code)
请求数据＝ json解析.转换为文本 ()
' 发送请求
响应文本＝网页_访问S (“https://api.dingtalk.com/v1.0/robot/groupMessages/send”, 1, 请求数据, 请求头, , 状态码)

如果 (状态码＝ 200)

logger.写日志 (日志级别_信息, "消息发送成功，响应: " ＋响应文本)

logger.写日志 (日志级别_错误, "消息发送失败，状态码: " ＋到文本 (状态码) ＋ "，响应: " ＋响应文本)

窗口程序集名	保留	保留	备注
日志类
变量名	类型	数组	备注
日志级别	整数型

子程序名	返回值类型	公开	备注
初始化

日志级别＝日志级别_信息

子程序名	返回值类型	公开	备注
设置日志级别
参数名	类型	参考	可空	数组	备注
级别	整数型

日志级别＝级别

子程序名	返回值类型	公开	备注
写日志
参数名	类型	参考	可空	数组	备注
级别	整数型
内容	文本型

变量名	类型	静态	数组	备注
时间字符串	文本型
级别字符串	文本型

如果 (级别＞日志级别)

返回

时间字符串＝取时间字符串 (取现行时间 (), #日期时间格式 )

判断 (级别＝日志级别_调试)

级别字符串＝ "DEBUG"

判断 (级别＝日志级别_信息)

级别字符串＝ "INFO"

判断 (级别＝日志级别_警告)

级别字符串＝ "WARNING"

判断 (级别＝日志级别_错误)

级别字符串＝ "ERROR"

判断 (级别＝日志级别_致命)

级别字符串＝ "FATAL"

级别字符串＝ "UNKNOWN"

输出调试文本 (时间字符串＋ " [" ＋级别字符串＋ "] " ＋内容)

常量名称	常量值	公开	备注
日志级别_调试	0
日志级别_信息	1
日志级别_警告	2
日志级别_错误	3
日志级别_致命	4

i支持库列表	支持库注释
internet	互联网支持库
dp1	数据操作支持库一
eJSON	(未知支持库)

.版本 2<br />
.支持库 internet<br />
.支持库 dp1<br />
.支持库 eJSON<br />
<br />
.程序集 程序集1<br />
.程序集变量 token_cache, 文本型, , "2", [0]为token，[1]为过期时间<br />
.程序集变量 logger, 日志类<br />
.程序集变量 client_id, 文本型<br />
.程序集变量 client_secret, 文本型<br />
.程序集变量 robot_code, 文本型<br />
<br />
.子程序 _启动子程序, 整数型, , 本子程序在程序启动后最先执行<br />
.局部变量 命令行参数, 文本型, , "0"<br />
.局部变量 i, 整数型<br />
.局部变量 参数名, 文本型<br />
.局部变量 参数值, 文本型<br />
<br />
' 初始化日志<br />
logger.初始化 ()<br />
logger.设置日志级别 (日志级别_信息)<br />
<br />
' 解析命令行参数<br />
命令行参数 ＝ 取命令行参数 ()<br />
.计次循环首 (取数组成员数 (命令行参数), i)<br />
    .如果真 (取文本左边 (命令行参数 <i>, 2) ＝ "--")<br />
        参数名 ＝ 取文本中间 (命令行参数 <i>, 3, 取文本长度 (命令行参数 <i>) － 2)<br />
        .如果真 (i ＜ 取数组成员数 (命令行参数))<br />
            参数值 ＝ 命令行参数 [i ＋ 1]<br />
            .如果 (参数名 ＝ "client_id")<br />
                client_id ＝ 参数值<br />
            .否则如果 (参数名 ＝ "client_secret")<br />
                client_secret ＝ 参数值<br />
            .否则如果 (参数名 ＝ "robot_code")<br />
                robot_code ＝ 参数值<br />
            .否则<br />
                logger.写日志 (日志级别_警告, "未知参数: " ＋ 参数名)<br />
            .如果结束<br />
        .如果真结束<br />
    .如果真结束<br />
.计次循环尾 ()<br />
<br />
' 检查必要参数<br />
.如果 (client_id ＝ "" 或 client_secret ＝ "" 或 robot_code ＝ "")<br />
    logger.写日志 (日志级别_错误, "缺少必要参数，请提供client_id、client_secret和robot_code")<br />
    返回 (1)<br />
.如果结束<br />
<br />
' 初始化token缓存<br />
token_cache [1] ＝ "0"<br />
<br />
' 启动钉钉消息监听（这里使用模拟，实际需要根据钉钉stream协议实现）<br />
logger.写日志 (日志级别_信息, "钉钉机qi人程序启动成功")<br />
模拟消息监听 ()<br />
<br />
返回 (0)<br />
<br />
.子程序 模拟消息监听<br />
' 注意：这里只是模拟消息监听，实际需要根据钉钉stream协议实现WebSocket连接<br />
.局部变量 模拟会话ID, 文本型<br />
<br />
模拟会话ID ＝ "模拟群会话ID_123456"<br />
logger.写日志 (日志级别_信息, "开始监听消息...")<br />
<br />
' 模拟接收到消息<br />
.计次循环首 (5, )<br />
    处理接收到的消息 (模拟会话ID)<br />
    延迟 (5000)<br />
.计次循环尾 ()<br />
<br />
.子程序 处理接收到的消息<br />
.参数 open_conversation_id, 文本型<br />
.局部变量 access_token, 文本型<br />
<br />
logger.写日志 (日志级别_信息, "收到消息，会话ID: " ＋ open_conversation_id)<br />
access_token ＝ 获取Token ()<br />
.如果 (access_token ≠ "")<br />
    发送机qi人群消息 (access_token, open_conversation_id)<br />
.否则<br />
    logger.写日志 (日志级别_错误, "获取access_token失败")<br />
.如果结束<br />
<br />
.子程序 获取Token, 文本型<br />
.局部变量 现在时间, 整数型<br />
.局部变量 过期时间, 整数型<br />
.局部变量 响应文本, 文本型<br />
.局部变量 json解析, 类_json解析<br />
.局部变量 状态码, 整数型<br />
.局部变量 token, 文本型<br />
.局部变量 过期时间秒, 整数型<br />
<br />
' 检查缓存<br />
现在时间 ＝ 取现行时间戳 ()<br />
过期时间 ＝ 到整数 (token_cache [1])<br />
.如果 (token_cache [0] ≠ "" 且 现在时间 ＜ 过期时间)<br />
    返回 (token_cache [0])<br />
.如果结束<br />
<br />
' 缓存失效，重新获取<br />
logger.写日志 (日志级别_信息, "获取新的access_token...")<br />
响应文本 ＝ 网页_访问S ("https://api.dingtalk.com/v1.0/oauth2/accessToken", , "grant_type=client_credentials&client_id=" ＋ 编码_URL编码 (client_id) ＋ "&client_secret=" ＋ 编码_URL编码 (client_secret), , , 状态码)<br />
<br />
.如果 (状态码 ≠ 200)<br />
    logger.写日志 (日志级别_错误, "获取token失败，状态码: " ＋ 到文本 (状态码) ＋ "，响应: " ＋ 响应文本)<br />
    返回 ("")<br />
.如果结束<br />
<br />
json解析.解析 (响应文本)<br />
token ＝ json解析.取属性值 (“accessToken”)<br />
过期时间秒 ＝ 到整数 (json解析.取属性值 (“expireIn”))<br />
<br />
.如果 (token ≠ "")<br />
    ' 提前200秒刷新<br />
    token_cache [0] ＝ token<br />
    token_cache [1] ＝ 到文本 (现在时间 ＋ 过期时间秒 － 200)<br />
    logger.写日志 (日志级别_信息, "获取access_token成功")<br />
    返回 (token)<br />
.否则<br />
    logger.写日志 (日志级别_错误, "解析token失败，响应: " ＋ 响应文本)<br />
    返回 ("")<br />
.如果结束<br />
<br />
.子程序 发送机qi人群消息<br />
.参数 access_token, 文本型<br />
.参数 open_conversation_id, 文本型<br />
.局部变量 响应文本, 文本型<br />
.局部变量 状态码, 整数型<br />
.局部变量 请求头, 文本型<br />
.局部变量 请求数据, 文本型<br />
.局部变量 json解析, 类_json解析<br />
<br />
logger.写日志 (日志级别_信息, "发送群消息到会话: " ＋ open_conversation_id)<br />
<br />
' 构建请求头<br />
请求头 ＝ "Content-Type: application/json" ＋ #换行符 ＋ "x-acs-dingtalk-access-token: " ＋ access_token<br />
<br />
' 构建请求数据<br />
json解析.清空 ()<br />
json解析.添加键值对 (“msgParam”, “{"content":"易语言机qi人 say：hello"}”)<br />
json解析.添加键值对 (“msgKey”, “sampleText”)<br />
json解析.添加键值对 (“openConversationId”, open_conversation_id)<br />
json解析.添加键值对 (“robotCode”, robot_code)<br />
请求数据 ＝ json解析.转换为文本 ()<br />
<br />
' 发送请求<br />
响应文本 ＝ 网页_访问S (“https://api.dingtalk.com/v1.0/robot/groupMessages/send”, 1, 请求数据, 请求头, , 状态码)<br />
<br />
.如果 (状态码 ＝ 200)<br />
    logger.写日志 (日志级别_信息, "消息发送成功，响应: " ＋ 响应文本)<br />
.否则<br />
    logger.写日志 (日志级别_错误, "消息发送失败，状态码: " ＋ 到文本 (状态码) ＋ "，响应: " ＋ 响应文本)<br />
.如果结束<br />
<br />
.程序集 日志类<br />
.程序集变量 日志级别, 整数型<br />
<br />
.子程序 初始化<br />
日志级别 ＝ 日志级别_信息<br />
<br />
.子程序 设置日志级别<br />
.参数 级别, 整数型<br />
日志级别 ＝ 级别<br />
<br />
.子程序 写日志<br />
.参数 级别, 整数型<br />
.参数 内容, 文本型<br />
.局部变量 时间字符串, 文本型<br />
.局部变量 级别字符串, 文本型<br />
<br />
.如果 (级别 ＞ 日志级别)<br />
    返回<br />
.如果结束<br />
<br />
时间字符串 ＝ 取时间字符串 (取现行时间 (), #日期时间格式)<br />
.判断 (级别 ＝ 日志级别_调试)<br />
    级别字符串 ＝ "DEBUG"<br />
.判断 (级别 ＝ 日志级别_信息)<br />
    级别字符串 ＝ "INFO"<br />
.判断 (级别 ＝ 日志级别_警告)<br />
    级别字符串 ＝ "WARNING"<br />
.判断 (级别 ＝ 日志级别_错误)<br />
    级别字符串 ＝ "ERROR"<br />
.判断 (级别 ＝ 日志级别_致命)<br />
    级别字符串 ＝ "FATAL"<br />
.默认<br />
    级别字符串 ＝ "UNKNOWN"<br />
.判断结束<br />
<br />
输出调试文本 (时间字符串 ＋ " [" ＋ 级别字符串 ＋ "] " ＋ 内容)<br />
<br />
.常量 日志级别_调试, "0"<br />
.常量 日志级别_信息, "1"<br />
.常量 日志级别_警告, "2"<br />
.常量 日志级别_错误, "3"<br />
.常量 日志级别_致命, "4"

huayuanderen · 发表于 2025-8-18 17:45:04

子程序名	返回值类型	公开	备注
_启动子程序	整数型

日志初始化 ()
启动线程 (&主逻辑, , )
返回 (0)
' ================== 日志初始化 ==================

子程序名	返回值类型	公开	备注
日志初始化

输出调试文本 (“[INFO] 日志系统已初始化”)
' ================== 主逻辑 ==================

子程序名	返回值类型	公开	备注
主逻辑

变量名	类型	静态	数组	备注
client_id	文本型
client_secret	文本型
robot_code	文本型
access_token	文本型

' 替换为你的钉钉开放平台凭证
client_id ＝ “your_client_id”
client_secret ＝ “your_client_secret”
robot_code ＝ “your_robot_code”

判断循环首 (真) ' 保持长连接

access_token ＝获取Token (client_id, client_secret)

如果 (access_token ≠ “”)

输出调试文本 (“[INFO] 获取Token成功: ” ＋ access_token)

' 模拟接收消息并回复（实际需对接钉钉Stream API）

发送群消息 (access_token, “test_conversation_id”, robot_code, “易语言机qi人说：你好！”)

输出调试文本 (“[ERROR] 获取Token失败”)

延迟 (5000) ' 5秒轮询一次

判断循环尾 ()
' ================== 获取Token ==================

子程序名	返回值类型	公开	备注
获取Token	文本型
参数名	类型	参考	可空	数组	备注
client_id	文本型
client_secret	文本型

变量名	类型	静态	数组	备注
当前时间	整数型
请求地址	文本型
提交数据	文本型
协议头	文本型
返回文本	文本型
HTTP响应	整数型
token	文本型

当前时间＝取启动时间 () ÷ 1000 ' 转换为秒
' 检查缓存是否有效

如果 (_token缓存 ≠ “” 且当前时间＜ _token过期时间)

返回 (_token缓存)

请求地址＝ “https://api.dingtalk.com/v1.0/oauth2/accessToken”

提交数据＝ “{”clientId“:”” ＋ client_id ＋ “”,”clientSecret“:”” ＋ client_secret ＋ “”}”

协议头＝ “Content-Type: application/json”

HTTP响应＝网页_访问 (请求地址, 1, 提交数据, , , 协议头, , , , , , , )

返回文本＝编码_Utf8到Ansi (HTTP响应)

token ＝文本_取出中间文本 (返回文本, “access_token””:”“, ”””)

如果 (token ≠ “”)

_token缓存＝ token

_token过期时间＝当前时间＋ 7000 ' 提前200秒刷新（7200-200）

返回 (token)

返回 (“”)

' ================== 发送群消息 ==================

子程序名	返回值类型	公开	备注
发送群消息	逻辑型
参数名	类型	参考	可空	数组	备注
access_token	文本型
open_conversation_id	文本型
robot_code	文本型
消息内容	文本型

变量名	类型	静态	数组	备注
请求地址	文本型
提交数据	文本型
协议头	文本型
HTTP响应	整数型
返回文本	文本型

请求地址＝ “https://api.dingtalk.com/v1.0/robot/groupMessages/send”
提交数据＝ “{”msgParam“: ”{”content“:”” ＋消息内容＋ “”}“,”msgKey“:”sampleText“,”openConversationId“:”” ＋ open_conversation_id ＋ “”,”robotCode“:”” ＋ robot_code ＋ “”}”
协议头＝ “Content-Type: application/json” ＋ #换行符＋ “x-acs-dingtalk-access-token: ” ＋ access_token
HTTP响应＝网页_访问 (请求地址, 1, 提交数据, , , 协议头, , , , , , , )
返回文本＝编码_Utf8到Ansi (HTTP响应)
输出调试文本 (“[DEBUG] 消息发送返回: ” ＋返回文本)
返回 (寻找文本 (返回文本, “success”, , 假) ≠ -1)

i支持库列表	支持库注释
EThread	多线程支持库
spec	特殊功能支持库

.版本 2<br />
.支持库 EThread<br />
.支持库 spec<br />
<br />
.子程序 _启动子程序, 整数型<br />
<br />
日志初始化 ()<br />
启动线程 (&主逻辑, , )<br />
返回 (0)<br />
<br />
' ================== 日志初始化 ==================<br />
<br />
.子程序 日志初始化<br />
<br />
输出调试文本 (“[INFO] 日志系统已初始化”)<br />
<br />
' ================== 主逻辑 ==================<br />
<br />
.子程序 主逻辑<br />
.局部变量 client_id, 文本型<br />
.局部变量 client_secret, 文本型<br />
.局部变量 robot_code, 文本型<br />
.局部变量 access_token, 文本型<br />
<br />
' 替换为你的钉钉开放平台凭证<br />
client_id ＝ “your_client_id”<br />
client_secret ＝ “your_client_secret”<br />
robot_code ＝ “your_robot_code”<br />
<br />
.判断循环首 (真)  ' 保持长连接<br />
    access_token ＝ 获取Token (client_id, client_secret)<br />
<br />
    .如果 (access_token ≠ “”)<br />
        输出调试文本 (“[INFO] 获取Token成功: ” ＋ access_token)<br />
        ' 模拟接收消息并回复（实际需对接钉钉Stream API）<br />
        发送群消息 (access_token, “test_conversation_id”, robot_code, “易语言机qi人说：你好！”)<br />
    .否则<br />
        输出调试文本 (“[ERROR] 获取Token失败”)<br />
    .如果结束<br />
<br />
    延迟 (5000)  ' 5秒轮询一次<br />
.判断循环尾 ()<br />
<br />
' ================== 获取Token ==================<br />
<br />
.子程序 获取Token, 文本型<br />
.参数 client_id, 文本型<br />
.参数 client_secret, 文本型<br />
.局部变量 当前时间, 整数型<br />
.局部变量 请求地址, 文本型<br />
.局部变量 提交数据, 文本型<br />
.局部变量 协议头, 文本型<br />
.局部变量 返回文本, 文本型<br />
.局部变量 HTTP响应, 整数型<br />
.局部变量 token, 文本型<br />
<br />
当前时间 ＝ 取启动时间 () ÷ 1000  ' 转换为秒<br />
<br />
' 检查缓存是否有效<br />
.如果 (_token缓存 ≠ “” 且 当前时间 ＜ _token过期时间)<br />
    返回 (_token缓存)<br />
.否则<br />
<br />
    请求地址 ＝ “https://api.dingtalk.com/v1.0/oauth2/accessToken”<br />
    提交数据 ＝ “{”clientId“:”” ＋ client_id ＋ “”,”clientSecret“:”” ＋ client_secret ＋ “”}”<br />
    协议头 ＝ “Content-Type: application/json”<br />
<br />
    HTTP响应 ＝ 网页_访问 (请求地址, 1, 提交数据, , , 协议头, , , , , , , )<br />
    返回文本 ＝ 编码_Utf8到Ansi (HTTP响应)<br />
<br />
    token ＝ 文本_取出中间文本 (返回文本, “access_token””:”“, ”””)<br />
<br />
    .如果 (token ≠ “”)<br />
        _token缓存 ＝ token<br />
        _token过期时间 ＝ 当前时间 ＋ 7000  ' 提前200秒刷新（7200-200）<br />
        返回 (token)<br />
    .否则<br />
        返回 (“”)<br />
    .如果结束<br />
<br />
.如果结束<br />
<br />
' ================== 发送群消息 ==================<br />
<br />
.子程序 发送群消息, 逻辑型<br />
.参数 access_token, 文本型<br />
.参数 open_conversation_id, 文本型<br />
.参数 robot_code, 文本型<br />
.参数 消息内容, 文本型<br />
.局部变量 请求地址, 文本型<br />
.局部变量 提交数据, 文本型<br />
.局部变量 协议头, 文本型<br />
.局部变量 HTTP响应, 整数型<br />
.局部变量 返回文本, 文本型<br />
<br />
请求地址 ＝ “https://api.dingtalk.com/v1.0/robot/groupMessages/send”<br />
提交数据 ＝ “{”msgParam“: ”{”content“:”” ＋ 消息内容 ＋ “”}“,”msgKey“:”sampleText“,”openConversationId“:”” ＋ open_conversation_id ＋ “”,”robotCode“:”” ＋ robot_code ＋ “”}”<br />
协议头 ＝ “Content-Type: application/json” ＋ #换行符 ＋ “x-acs-dingtalk-access-token: ” ＋ access_token<br />
<br />
HTTP响应 ＝ 网页_访问 (请求地址, 1, 提交数据, , , 协议头, , , , , , , )<br />
<br />
返回文本 ＝ 编码_Utf8到Ansi (HTTP响应)<br />
<br />
输出调试文本 (“[DEBUG] 消息发送返回: ” ＋ 返回文本)<br />
返回 (寻找文本 (返回文本, “success”, , 假) ≠ -1)

renjinyu · 发表于 2025-8-18 23:15:29

huayuanderen 发表于 2025-8-18 17:45
[e=0].版本 2
.支持库 EThread
.支持库 spec

现在是回调不了群里@消息

huayuanderen · 发表于 2025-8-19 01:06:33

什么是不是你代码有问题啊所以回调不了

renjinyu · 发表于 2025-8-19 08:28:17

huayuanderen 发表于 2025-8-19 01:06
什么是不是你代码有问题啊所以回调不了

从官网下的实例，用PYthon 试了可以用

huayuanderen · 发表于 2025-8-19 09:30:33

renjinyu 发表于 2025-8-19 08:28
从官网下的实例，用PYthon 试了可以用

那不清除啊，没弄过钉钉

		自动登录	找回密码
密码			注册

[易语言] Python源码转易语言实现钉钉stream机qi人（求助）

精易论坛 - 有你更精彩 /2