使用Python实现钉钉Stream模式服务开发及内部程序通信

1、什么是Stream模式

Stream 模式是钉钉开放平台提供的一种集成方式，它可以监听机器人回调、事件订阅回调和注册卡片回调。使用 Stream 模式接入，钉钉开放平台将通过 Websocket 连接与应用程序通讯，Stream 模式将极大降低接入门槛和资源依赖，不需要公网服务器、IP、域名等资源，只需集成钉钉开放平台 SDK 即可。

2、Stream模式原理

在 Stream 模式下，开发者的应用程序通过集成 SDK 的方式与钉钉开放平台建立一条 WebSocket 连接，建立连接过程中开放平台将对连接进行鉴权。当有卡片回调发生时，开放平台将通过 WebSocket 连接将数据通知到开发者的应用程序。开发者的应用程序可以接收到这些数据并进行相应处理，从而实现与钉钉开放平台的实时通信，参考下图所示

在这里插入图片描述

3、Stream模式优势

在钉钉开放平台向应用程序发送请求的场景中，大部分都是采用 Webhook （注册公网 HTTPS 服务）的方式，包括卡片回调，使用 Webhook 方式开发过程中会遇到较多的问题，包括

申请公网域名和TLS证书
申请公网IP并部署接入网关
部署应用防火墙并配置白名单
独立处理请求的鉴权，以及加解密处理
搭建内网穿透环境进行本地开发调试

4、stream 服务开发（python）

直接上源码

from flask import Flask, request, jsonify
import threading
import dingtalk_stream
from dingtalk_stream import AckMessage
import requests

app = Flask(__name__)

# 钉钉应用的App Key和App Secret
APP_KEY = 'xxxxxx'
APP_SECRET = 'xxxxxxxxxxxxx'

# 定义事件处理器
class MyEventHandler(dingtalk_stream.EventHandler):
    async def process(self, event: dingtalk_stream.EventMessage):
        print(event.headers.event_type, event.headers.event_id, event.headers.event_born_time, event.data)
        # 处理接收到的事件，并发送到WordPress
        send_data_to_wordpress_thread = threading.Thread(target=send_data_to_wordpress,args=(event.data,))
        send_data_to_wordpress_thread.start()
        return AckMessage.STATUS_OK, 'OK'

class MyCallbackHandler(dingtalk_stream.CallbackHandler):
    async def process(self, message: dingtalk_stream.CallbackMessage):
        print(message.headers.topic, message.data)
        # 处理接收到的回调消息，并发送到WordPress
        send_data_to_wordpress_thread = threading.Thread(target=send_data_to_wordpress,args=(message.data,))
        send_data_to_wordpress_thread.start()
        return AckMessage.STATUS_OK, 'OK'

# 启动钉钉Stream客户端
def start_dingtalk_client():
    credential = dingtalk_stream.Credential(APP_KEY, APP_SECRET)
    client = dingtalk_stream.DingTalkStreamClient(credential)
    client.register_all_event_handler(MyEventHandler())
    client.register_callback_handler(dingtalk_stream.ChatbotMessage.TOPIC, MyCallbackHandler())
    client.start_forever()

# 发送数据到wordpress 
def send_data_to_wordpress(data, allowed_origin=None):
    url = 'http://10.11.11.111:1111/dingtalk/data'  # 网站服务接口
    headers = {'Content-Type': 'application/json'}
    
    # 添加允许跨域的请求头
    if allowed_origin:
        headers['Access-Control-Allow-Origin'] = allowed_origin
    
    requests.post(url, json=data, headers=headers)
    

# 启动钉钉Stream客户端线程
thread = threading.Thread(target=start_dingtalk_client)
thread.start()

# Flask路由，用于健康检查或其他用途
@app.route('/')
def index():
    return 'DingTalk Stream client is running.'

if __name__ == '__main__':
    app.run(port=5000)