diff --git a/docs/channels/dingtalk/README.zh.md b/docs/channels/dingtalk/README.zh.md index 1e445d0b0..85bcf56e6 100644 --- a/docs/channels/dingtalk/README.zh.md +++ b/docs/channels/dingtalk/README.zh.md @@ -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 | diff --git a/pkg/channels/dingtalk/api.go b/pkg/channels/dingtalk/api.go new file mode 100644 index 000000000..d83471990 --- /dev/null +++ b/pkg/channels/dingtalk/api.go @@ -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) +} diff --git a/pkg/channels/dingtalk/dingtalk.go b/pkg/channels/dingtalk/dingtalk.go index c03122892..e99664c24 100644 --- a/pkg/channels/dingtalk/dingtalk.go +++ b/pkg/channels/dingtalk/dingtalk.go @@ -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, diff --git a/pkg/channels/dingtalk/token.go b/pkg/channels/dingtalk/token.go new file mode 100644 index 000000000..4ce2be481 --- /dev/null +++ b/pkg/channels/dingtalk/token.go @@ -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(), + }) + } + } + } + } + } +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 190341224..27e354d68 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -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 {