Merge PR #1414
This commit is contained in:
OpenClaw-User
commit
711b58e7b3
5 changed files with 553 additions and 29 deletions
|
|
@ -11,23 +11,83 @@
|
|||
"enabled": true,
|
||||
"client_id": "YOUR_CLIENT_ID",
|
||||
"client_secret": "YOUR_CLIENT_SECRET",
|
||||
"allow_from": []
|
||||
"allow_from": [],
|
||||
"proactive_send": false,
|
||||
"group_trigger": {
|
||||
"mention_only": false,
|
||||
"prefixes": []
|
||||
},
|
||||
"reasoning_channel_id": ""
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
| 字段 | 类型 | 必填 | 描述 |
|
||||
| ------------- | ------ | ---- | -------------------------------- |
|
||||
| enabled | bool | 是 | 是否启用钉钉频道 |
|
||||
| client_id | string | 是 | 钉钉应用的 Client ID |
|
||||
| client_secret | string | 是 | 钉钉应用的 Client Secret |
|
||||
| allow_from | array | 否 | 用户ID白名单，空表示允许所有用户 |
|
||||
| 字段 | 类型 | 必填 | 描述 |
|
||||
| ------------------- | ------ | ---- | -------------------------------------------- |
|
||||
| enabled | bool | 是 | 是否启用钉钉频道 |
|
||||
| client_id | string | 是 | 钉钉应用的 Client ID |
|
||||
| client_secret | string | 是 | 钉钉应用的 Client Secret |
|
||||
| allow_from | array | 否 | 用户ID白名单，空表示允许所有用户 |
|
||||
| proactive_send | bool | 否 | 是否启用主动消息发送（通过机器人API） |
|
||||
| group_trigger | object | 否 | 群聊触发配置 |
|
||||
| reasoning_channel_id| string | 否 | 推理消息发送的目标频道ID |
|
||||
|
||||
### 群聊触发配置
|
||||
|
||||
| 字段 | 类型 | 描述 |
|
||||
| ----------- | -------- | ------------------------------------------ |
|
||||
| mention_only| bool | 是否仅在@提及时响应 |
|
||||
| prefixes | []string | 触发前缀列表，如 `["/ai", "机器人"]` |
|
||||
|
||||
## 功能特性
|
||||
|
||||
### 健康监控与自动恢复
|
||||
|
||||
钉钉频道内置了连接健康监控机制，确保流式连接的稳定性：
|
||||
|
||||
- **定期健康检查**：每 5 分钟检查一次连接状态
|
||||
- **自动恢复**：当连接超过 30 分钟未收到消息时，自动触发重连
|
||||
- **重试机制**：恢复失败时每 30 秒重试一次，直到成功
|
||||
|
||||
### 主动消息发送
|
||||
|
||||
启用 `proactive_send` 后，机器人可以主动向用户或群组发送消息：
|
||||
|
||||
- **智能路由**：优先使用会话 webhook 发送，失效时自动切换到机器人 API
|
||||
- **Token 自动刷新**：每 5 分钟自动刷新访问令牌，确保 API 可用性
|
||||
- **支持场景**：
|
||||
- 单聊消息：通过 `oToMessages/batchSend` API
|
||||
- 群聊消息：通过 `groupMessages/send` API
|
||||
|
||||
### 结构化日志
|
||||
|
||||
所有钉钉相关的日志均采用结构化格式输出，便于调试和监控：
|
||||
|
||||
```log
|
||||
INFO dingtalk: DingTalk channel started {"proactive_send": true}
|
||||
DEBUG dingtalk: Received message {"sender_nick": "张三", "sender_id": "user123", "preview": "你好"}
|
||||
INFO dingtalk: Connection recovered successfully {}
|
||||
```
|
||||
|
||||
## 设置流程
|
||||
|
||||
1. 前往 [钉钉开放平台](https://open.dingtalk.com/)
|
||||
2. 创建一个企业内部应用
|
||||
3. 从应用设置中获取 Client ID 和 Client Secret
|
||||
4. 配置OAuth和事件订阅(如需要)
|
||||
4. 配置 OAuth 和事件订阅（如需要）
|
||||
5. 将 Client ID 和 Client Secret 填入配置文件中
|
||||
6. 如需主动消息功能，设置 `proactive_send: true`
|
||||
|
||||
## 环境变量
|
||||
|
||||
所有配置项均可通过环境变量设置：
|
||||
|
||||
| 环境变量 | 对应配置项 |
|
||||
| --------------------------------------------- | ----------------------- |
|
||||
| PICOCLAW_CHANNELS_DINGTALK_ENABLED | enabled |
|
||||
| PICOCLAW_CHANNELS_DINGTALK_CLIENT_ID | client_id |
|
||||
| PICOCLAW_CHANNELS_DINGTALK_CLIENT_SECRET | client_secret |
|
||||
| PICOCLAW_CHANNELS_DINGTALK_ALLOW_FROM | allow_from |
|
||||
| PICOCLAW_CHANNELS_DINGTALK_PROACTIVE_SEND | proactive_send |
|
||||
| PICOCLAW_CHANNELS_DINGTALK_REASONING_CHANNEL_ID| reasoning_channel_id |
|
||||
|
|
|
|||
151
pkg/channels/dingtalk/api.go
Normal file
151
pkg/channels/dingtalk/api.go
Normal file
|
|
@ -0,0 +1,151 @@
|
|||
// PicoClaw - Ultra-lightweight personal AI agent
|
||||
// DingTalk robot API for proactive messaging
|
||||
|
||||
package dingtalk
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/channels"
|
||||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
)
|
||||
|
||||
// sendGroupMessage sends a message to a group chat via robot API
|
||||
func (c *DingTalkChannel) sendGroupMessage(ctx context.Context, accessToken, openConversationId, content string) error {
|
||||
url := fmt.Sprintf("%s/v1.0/robot/groupMessages/send", dingtalkAPIBase)
|
||||
|
||||
// Build msgParam for markdown message
|
||||
msgParam, _ := json.Marshal(map[string]string{
|
||||
"title": "PicoClaw",
|
||||
"text": content,
|
||||
})
|
||||
|
||||
body := map[string]any{
|
||||
"openConversationId": openConversationId,
|
||||
"robotCode": c.clientID,
|
||||
"msgKey": "sampleMarkdown",
|
||||
"msgParam": string(msgParam),
|
||||
}
|
||||
|
||||
return c.sendRobotAPIRequest(ctx, accessToken, url, body, "group")
|
||||
}
|
||||
|
||||
// sendOToMessage sends a message to a user via robot API (one-to-one)
|
||||
func (c *DingTalkChannel) sendOToMessage(ctx context.Context, accessToken, userID, content string) error {
|
||||
url := fmt.Sprintf("%s/v1.0/robot/oToMessages/batchSend", dingtalkAPIBase)
|
||||
|
||||
// Build msgParam for markdown message
|
||||
msgParam, _ := json.Marshal(map[string]string{
|
||||
"title": "PicoClaw",
|
||||
"text": content,
|
||||
})
|
||||
|
||||
body := map[string]any{
|
||||
"robotCode": c.clientID,
|
||||
"userIds": []string{userID},
|
||||
"msgKey": "sampleMarkdown",
|
||||
"msgParam": string(msgParam),
|
||||
}
|
||||
|
||||
return c.sendRobotAPIRequest(ctx, accessToken, url, body, "direct")
|
||||
}
|
||||
|
||||
// sendRobotAPIRequest sends a request to DingTalk robot API
|
||||
func (c *DingTalkChannel) sendRobotAPIRequest(
|
||||
ctx context.Context,
|
||||
accessToken, url string,
|
||||
body any,
|
||||
msgType string,
|
||||
) error {
|
||||
jsonData, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marshal request body failed: %w", err)
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(jsonData))
|
||||
if err != nil {
|
||||
return fmt.Errorf("create request failed: %w", err)
|
||||
}
|
||||
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("x-acs-dingtalk-access-token", accessToken)
|
||||
|
||||
resp, err := c.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return channels.ClassifyNetError(err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
respBody, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return fmt.Errorf("read response failed: %w", err)
|
||||
}
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
// Parse error response for better error message
|
||||
var errResp struct {
|
||||
Code string `json:"code"`
|
||||
Msg string `json:"message"`
|
||||
}
|
||||
if json.Unmarshal(respBody, &errResp) == nil && errResp.Msg != "" {
|
||||
return fmt.Errorf("dingtalk robot API error: %s (%s)", errResp.Msg, errResp.Code)
|
||||
}
|
||||
return channels.ClassifySendError(resp.StatusCode,
|
||||
fmt.Errorf("dingtalk robot API error (status %d): %s", resp.StatusCode, string(respBody)))
|
||||
}
|
||||
|
||||
logger.InfoCF("dingtalk", "Robot API message sent successfully", map[string]any{
|
||||
"type": msgType,
|
||||
"status": resp.StatusCode,
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// sendViaRobotAPI sends a message via the robot API
|
||||
// For proactive messaging: chatID is used directly as staffId (direct) or openConversationId (group)
|
||||
// Group IDs start with "cid", everything else is treated as direct message staffId
|
||||
func (c *DingTalkChannel) sendViaRobotAPI(ctx context.Context, chatID, content string) error {
|
||||
accessToken, err := c.ensureValidToken(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get access token: %w", err)
|
||||
}
|
||||
|
||||
// First, check if we have session info (from prior conversation)
|
||||
session := c.getSession(chatID)
|
||||
if session != nil {
|
||||
if session.ConversationType == "1" {
|
||||
// Direct message with session info
|
||||
if session.SenderStaffId != "" {
|
||||
return c.sendOToMessage(ctx, accessToken, session.SenderStaffId, content)
|
||||
}
|
||||
} else {
|
||||
// Group message with session info
|
||||
if session.OpenConversationId != "" {
|
||||
return c.sendGroupMessage(ctx, accessToken, session.OpenConversationId, content)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// No session info - detect type based on chatID format
|
||||
// Group IDs start with "cid" (openConversationId format)
|
||||
isGroup := len(chatID) > 3 && chatID[:3] == "cid"
|
||||
|
||||
if isGroup {
|
||||
logger.DebugCF("dingtalk", "Sending proactive group message", map[string]any{
|
||||
"open_conversation_id": chatID,
|
||||
})
|
||||
return c.sendGroupMessage(ctx, accessToken, chatID, content)
|
||||
}
|
||||
|
||||
// Direct message - chatID is the staffId
|
||||
logger.DebugCF("dingtalk", "Sending proactive direct message", map[string]any{
|
||||
"staff_id": chatID,
|
||||
})
|
||||
return c.sendOToMessage(ctx, accessToken, chatID, content)
|
||||
}
|
||||
|
|
@ -6,7 +6,9 @@ package dingtalk
|
|||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/open-dingtalk/dingtalk-stream-sdk-go/chatbot"
|
||||
"github.com/open-dingtalk/dingtalk-stream-sdk-go/client"
|
||||
|
|
@ -20,6 +22,24 @@ import (
|
|||
"github.com/sipeed/picoclaw/pkg/utils"
|
||||
)
|
||||
|
||||
// Health check constants
|
||||
const (
|
||||
healthCheckInterval = 5 * time.Minute // Check every 5 minutes
|
||||
maxSilenceDuration = 30 * time.Minute // Max time without messages before recovery
|
||||
recoveryDelay = 2 * time.Second // Delay before reconnecting
|
||||
recoveryRetryDelay = 30 * time.Second // Delay before retrying failed recovery
|
||||
)
|
||||
|
||||
// DingTalkSession stores complete session information for proactive messaging
|
||||
type DingTalkSession struct {
|
||||
SessionWebhook string `json:"session_webhook"`
|
||||
SessionWebhookExpiredTime int64 `json:"session_webhook_expired_time"` // milliseconds timestamp
|
||||
OpenConversationId string `json:"open_conversation_id"` // Group chat ID
|
||||
SenderStaffId string `json:"sender_staff_id"` // User ID for direct chat
|
||||
ConversationType string `json:"conversation_type"` // "1" = direct, else = group
|
||||
LastUpdated time.Time `json:"last_updated"`
|
||||
}
|
||||
|
||||
// DingTalkChannel implements the Channel interface for DingTalk (钉钉)
|
||||
// It uses WebSocket for receiving messages via stream mode and API for sending
|
||||
type DingTalkChannel struct {
|
||||
|
|
@ -30,8 +50,16 @@ type DingTalkChannel struct {
|
|||
streamClient *client.StreamClient
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
// Map to store session webhooks for each chat
|
||||
sessionWebhooks sync.Map // chatID -> sessionWebhook
|
||||
// Map to store session info for each chat
|
||||
sessions sync.Map // chatID -> *DingTalkSession
|
||||
// Token management for proactive messaging via robot API
|
||||
accessToken string
|
||||
tokenExpiry time.Time
|
||||
tokenMu sync.RWMutex
|
||||
httpClient *http.Client
|
||||
// Health monitoring
|
||||
lastMessageTime time.Time
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
// NewDingTalkChannel creates a new DingTalk channel instance
|
||||
|
|
@ -62,7 +90,38 @@ func (c *DingTalkChannel) Start(ctx context.Context) error {
|
|||
logger.InfoC("dingtalk", "Starting DingTalk channel (Stream Mode)...")
|
||||
|
||||
c.ctx, c.cancel = context.WithCancel(ctx)
|
||||
c.lastMessageTime = time.Now() // Initialize on start
|
||||
|
||||
// Initialize HTTP client for robot API
|
||||
c.httpClient = &http.Client{Timeout: 30 * time.Second}
|
||||
|
||||
// If proactive send is enabled, get initial token and start refresh loop
|
||||
if c.config.ProactiveSend {
|
||||
if err := c.refreshAccessToken(); err != nil {
|
||||
logger.WarnCF("dingtalk", "Failed to get initial access token, proactive send may not work", map[string]any{
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
go c.tokenRefreshLoop()
|
||||
}
|
||||
|
||||
// Start the stream client
|
||||
if err := c.startStreamClient(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Start health monitoring goroutine
|
||||
go c.healthMonitor()
|
||||
|
||||
c.SetRunning(true)
|
||||
logger.InfoCF("dingtalk", "DingTalk channel started", map[string]any{
|
||||
"proactive_send": c.config.ProactiveSend,
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
// startStreamClient creates and starts the stream client
|
||||
func (c *DingTalkChannel) startStreamClient() error {
|
||||
// Create credential config
|
||||
cred := client.NewAppCredentialConfig(c.clientID, c.clientSecret)
|
||||
|
||||
|
|
@ -72,7 +131,7 @@ func (c *DingTalkChannel) Start(ctx context.Context) error {
|
|||
client.WithAutoReconnect(true),
|
||||
)
|
||||
|
||||
// Register chatbot callback handler (IChatBotMessageHandler is a function type)
|
||||
// Register chatbot callback handler
|
||||
c.streamClient.RegisterChatBotCallbackRouter(c.onChatBotMessageReceived)
|
||||
|
||||
// Start the stream client
|
||||
|
|
@ -80,11 +139,84 @@ func (c *DingTalkChannel) Start(ctx context.Context) error {
|
|||
return fmt.Errorf("failed to start stream client: %w", err)
|
||||
}
|
||||
|
||||
c.SetRunning(true)
|
||||
logger.InfoC("dingtalk", "DingTalk channel started (Stream Mode)")
|
||||
return nil
|
||||
}
|
||||
|
||||
// healthMonitor periodically checks connection health and triggers recovery if needed
|
||||
func (c *DingTalkChannel) healthMonitor() {
|
||||
ticker := time.NewTicker(healthCheckInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-c.ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
c.checkAndRecover()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// checkAndRecover checks if connection is stale and triggers recovery
|
||||
func (c *DingTalkChannel) checkAndRecover() {
|
||||
c.mu.RLock()
|
||||
silenceDuration := time.Since(c.lastMessageTime)
|
||||
c.mu.RUnlock()
|
||||
|
||||
logger.DebugCF("dingtalk", "Health check: silence duration", map[string]any{"duration": silenceDuration})
|
||||
|
||||
if silenceDuration >= maxSilenceDuration {
|
||||
logger.InfoCF(
|
||||
"dingtalk",
|
||||
"Connection appears stale (no messages for configured duration), triggering recovery",
|
||||
map[string]any{"silenceDuration": silenceDuration},
|
||||
)
|
||||
c.recoverConnection()
|
||||
}
|
||||
}
|
||||
|
||||
// recoverConnection attempts to recover the stream connection
|
||||
func (c *DingTalkChannel) recoverConnection() {
|
||||
// Close old client
|
||||
if c.streamClient != nil {
|
||||
c.streamClient.Close()
|
||||
time.Sleep(recoveryDelay)
|
||||
}
|
||||
|
||||
// Attempt to reconnect
|
||||
for {
|
||||
select {
|
||||
case <-c.ctx.Done():
|
||||
logger.InfoC("dingtalk", "Recovery aborted: context cancelled")
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
err := c.startStreamClient()
|
||||
if err == nil {
|
||||
logger.InfoCF("dingtalk", "Connection recovered successfully", map[string]any{})
|
||||
c.mu.Lock()
|
||||
c.lastMessageTime = time.Now()
|
||||
c.mu.Unlock()
|
||||
return
|
||||
}
|
||||
|
||||
logger.WarnCF(
|
||||
"dingtalk",
|
||||
"Recovery failed, retrying",
|
||||
map[string]any{"error": err, "retryDelay": recoveryRetryDelay},
|
||||
)
|
||||
time.Sleep(recoveryRetryDelay)
|
||||
}
|
||||
}
|
||||
|
||||
// updateLastMessageTime updates the last message timestamp
|
||||
func (c *DingTalkChannel) updateLastMessageTime() {
|
||||
c.mu.Lock()
|
||||
c.lastMessageTime = time.Now()
|
||||
c.mu.Unlock()
|
||||
}
|
||||
|
||||
// Stop gracefully stops the DingTalk channel
|
||||
func (c *DingTalkChannel) Stop(ctx context.Context) error {
|
||||
logger.InfoC("dingtalk", "Stopping DingTalk channel...")
|
||||
|
|
@ -102,30 +234,60 @@ func (c *DingTalkChannel) Stop(ctx context.Context) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// Send sends a message to DingTalk via the chatbot reply API
|
||||
// Send sends a message to DingTalk
|
||||
// Priority: 1) session_webhook (if valid), 2) robot API (if proactive_send enabled)
|
||||
func (c *DingTalkChannel) Send(ctx context.Context, msg bus.OutboundMessage) error {
|
||||
if !c.IsRunning() {
|
||||
return channels.ErrNotRunning
|
||||
}
|
||||
|
||||
// Get session webhook from storage
|
||||
sessionWebhookRaw, ok := c.sessionWebhooks.Load(msg.ChatID)
|
||||
if !ok {
|
||||
return fmt.Errorf("no session_webhook found for chat %s, cannot send message", msg.ChatID)
|
||||
}
|
||||
|
||||
sessionWebhook, ok := sessionWebhookRaw.(string)
|
||||
if !ok {
|
||||
return fmt.Errorf("invalid session_webhook type for chat %s", msg.ChatID)
|
||||
}
|
||||
|
||||
logger.DebugCF("dingtalk", "Sending message", map[string]any{
|
||||
"chat_id": msg.ChatID,
|
||||
"preview": utils.Truncate(msg.Content, 100),
|
||||
})
|
||||
|
||||
// Use the session webhook to send the reply
|
||||
return c.SendDirectReply(ctx, sessionWebhook, msg.Content)
|
||||
// 1. Try session webhook first
|
||||
session := c.getSession(msg.ChatID)
|
||||
if session != nil && c.isSessionWebhookValid(session) {
|
||||
err := c.SendDirectReply(ctx, session.SessionWebhook, msg.Content)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
// Session webhook failed, log and try fallback
|
||||
logger.WarnCF("dingtalk", "Session webhook send failed, trying fallback", map[string]any{
|
||||
"error": err.Error(),
|
||||
"chat_id": msg.ChatID,
|
||||
})
|
||||
}
|
||||
|
||||
// 2. Fallback to robot API if proactive send is enabled
|
||||
if !c.config.ProactiveSend {
|
||||
return fmt.Errorf("no valid session_webhook for chat %s and proactive_send is disabled", msg.ChatID)
|
||||
}
|
||||
|
||||
return c.sendViaRobotAPI(ctx, msg.ChatID, msg.Content)
|
||||
}
|
||||
|
||||
// getSession retrieves session info for a chat
|
||||
func (c *DingTalkChannel) getSession(chatID string) *DingTalkSession {
|
||||
if v, ok := c.sessions.Load(chatID); ok {
|
||||
if session, ok := v.(*DingTalkSession); ok {
|
||||
return session
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// isSessionWebhookValid checks if the session webhook is still valid
|
||||
func (c *DingTalkChannel) isSessionWebhookValid(session *DingTalkSession) bool {
|
||||
if session.SessionWebhook == "" {
|
||||
return false
|
||||
}
|
||||
// Check expiry if available (SessionWebhookExpiredTime is in milliseconds)
|
||||
if session.SessionWebhookExpiredTime > 0 {
|
||||
return time.Now().UnixMilli() < session.SessionWebhookExpiredTime
|
||||
}
|
||||
return true // No expiry info, assume valid
|
||||
}
|
||||
|
||||
// onChatBotMessageReceived implements the IChatBotMessageHandler function signature
|
||||
|
|
@ -135,6 +297,9 @@ func (c *DingTalkChannel) onChatBotMessageReceived(
|
|||
ctx context.Context,
|
||||
data *chatbot.BotCallbackDataModel,
|
||||
) ([]byte, error) {
|
||||
// Update last message time for health monitoring
|
||||
c.updateLastMessageTime()
|
||||
|
||||
// Extract message content from Text field
|
||||
content := data.Text.Content
|
||||
if content == "" {
|
||||
|
|
@ -158,8 +323,16 @@ func (c *DingTalkChannel) onChatBotMessageReceived(
|
|||
chatID = data.ConversationId
|
||||
}
|
||||
|
||||
// Store the session webhook for this chat so we can reply later
|
||||
c.sessionWebhooks.Store(chatID, data.SessionWebhook)
|
||||
// Store complete session info for proactive messaging
|
||||
session := &DingTalkSession{
|
||||
SessionWebhook: data.SessionWebhook,
|
||||
SessionWebhookExpiredTime: data.SessionWebhookExpiredTime,
|
||||
OpenConversationId: data.ConversationId,
|
||||
SenderStaffId: data.SenderStaffId,
|
||||
ConversationType: data.ConversationType,
|
||||
LastUpdated: time.Now(),
|
||||
}
|
||||
c.sessions.Store(chatID, session)
|
||||
|
||||
metadata := map[string]string{
|
||||
"sender_name": senderNick,
|
||||
|
|
|
|||
139
pkg/channels/dingtalk/token.go
Normal file
139
pkg/channels/dingtalk/token.go
Normal file
|
|
@ -0,0 +1,139 @@
|
|||
// PicoClaw - Ultra-lightweight personal AI agent
|
||||
// DingTalk access token management
|
||||
|
||||
package dingtalk
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
)
|
||||
|
||||
const (
|
||||
dingtalkAPIBase = "https://api.dingtalk.com"
|
||||
tokenRefreshInterval = 5 * time.Minute
|
||||
tokenExpireBuffer = 5 * time.Minute // Refresh token 5 minutes before expiry
|
||||
)
|
||||
|
||||
// accessTokenResponse 钉钉 access_token 响应
|
||||
type accessTokenResponse struct {
|
||||
AccessToken string `json:"accessToken"`
|
||||
ExpireIn int64 `json:"expireIn"` // seconds
|
||||
}
|
||||
|
||||
// refreshAccessToken refreshes the access token from DingTalk API
|
||||
func (c *DingTalkChannel) refreshAccessToken() error {
|
||||
url := fmt.Sprintf("%s/v1.0/oauth2/accessToken", dingtalkAPIBase)
|
||||
|
||||
body := map[string]string{
|
||||
"appKey": c.clientID,
|
||||
"appSecret": c.clientSecret,
|
||||
}
|
||||
jsonData, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marshal request body failed: %w", err)
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(c.ctx, http.MethodPost, url, bytes.NewBuffer(jsonData))
|
||||
if err != nil {
|
||||
return fmt.Errorf("create request failed: %w", err)
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
resp, err := c.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("request failed: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
respBody, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return fmt.Errorf("read response failed: %w", err)
|
||||
}
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("dingtalk API error (status %d): %s", resp.StatusCode, string(respBody))
|
||||
}
|
||||
|
||||
var tokenResp accessTokenResponse
|
||||
if err := json.Unmarshal(respBody, &tokenResp); err != nil {
|
||||
return fmt.Errorf("parse response failed: %w", err)
|
||||
}
|
||||
|
||||
if tokenResp.AccessToken == "" {
|
||||
return fmt.Errorf("empty access token in response")
|
||||
}
|
||||
|
||||
c.tokenMu.Lock()
|
||||
c.accessToken = tokenResp.AccessToken
|
||||
c.tokenExpiry = time.Now().Add(time.Duration(tokenResp.ExpireIn) * time.Second)
|
||||
c.tokenMu.Unlock()
|
||||
|
||||
logger.DebugCF("dingtalk", "Access token refreshed successfully", map[string]any{
|
||||
"expires_in": tokenResp.ExpireIn,
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
// getAccessToken returns the current valid access token
|
||||
// Returns empty string if token is expired or about to expire
|
||||
func (c *DingTalkChannel) getAccessToken() string {
|
||||
c.tokenMu.RLock()
|
||||
defer c.tokenMu.RUnlock()
|
||||
|
||||
if c.accessToken == "" {
|
||||
return ""
|
||||
}
|
||||
|
||||
// Check if token is about to expire (within buffer time)
|
||||
if time.Now().After(c.tokenExpiry.Add(-tokenExpireBuffer)) {
|
||||
return "" // Token expired or about to expire
|
||||
}
|
||||
|
||||
return c.accessToken
|
||||
}
|
||||
|
||||
// ensureValidToken ensures we have a valid access token, refreshing if necessary
|
||||
func (c *DingTalkChannel) ensureValidToken(ctx context.Context) (string, error) {
|
||||
token := c.getAccessToken()
|
||||
if token != "" {
|
||||
return token, nil
|
||||
}
|
||||
|
||||
// Need to refresh
|
||||
if err := c.refreshAccessToken(); err != nil {
|
||||
return "", fmt.Errorf("failed to refresh access token: %w", err)
|
||||
}
|
||||
|
||||
return c.getAccessToken(), nil
|
||||
}
|
||||
|
||||
// tokenRefreshLoop runs periodically to refresh the access token
|
||||
func (c *DingTalkChannel) tokenRefreshLoop() {
|
||||
ticker := time.NewTicker(tokenRefreshInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-c.ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
if c.config.ProactiveSend {
|
||||
// Check if token needs refresh
|
||||
if c.getAccessToken() == "" {
|
||||
if err := c.refreshAccessToken(); err != nil {
|
||||
logger.ErrorCF("dingtalk", "Failed to refresh access token", map[string]any{
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -362,6 +362,7 @@ type DingTalkConfig struct {
|
|||
AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_DINGTALK_ALLOW_FROM"`
|
||||
GroupTrigger GroupTriggerConfig `json:"group_trigger,omitempty"`
|
||||
ReasoningChannelID string `json:"reasoning_channel_id" env:"PICOCLAW_CHANNELS_DINGTALK_REASONING_CHANNEL_ID"`
|
||||
ProactiveSend bool `json:"proactive_send" env:"PICOCLAW_CHANNELS_DINGTALK_PROACTIVE_SEND"`
|
||||
}
|
||||
|
||||
type SlackConfig struct {
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue