From 232abb4922ec67312a539cbc4751de5f6eba8504 Mon Sep 17 00:00:00 2001 From: fishtrees Date: Wed, 4 Mar 2026 18:25:50 +0800 Subject: [PATCH 1/4] fix: add health monitoring and auto-recovery for dingtalk stream connection --- pkg/channels/dingtalk/dingtalk.go | 102 +++++++++++++++++++++++++++++- 1 file changed, 99 insertions(+), 3 deletions(-) diff --git a/pkg/channels/dingtalk/dingtalk.go b/pkg/channels/dingtalk/dingtalk.go index 8642ad362..89ddbbc50 100644 --- a/pkg/channels/dingtalk/dingtalk.go +++ b/pkg/channels/dingtalk/dingtalk.go @@ -7,6 +7,7 @@ import ( "context" "fmt" "sync" + "time" "github.com/open-dingtalk/dingtalk-stream-sdk-go/chatbot" "github.com/open-dingtalk/dingtalk-stream-sdk-go/client" @@ -19,6 +20,14 @@ 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 +) + // DingTalkChannel implements the Channel interface for DingTalk (钉钉) // It uses WebSocket for receiving messages via stream mode and API for sending type DingTalkChannel struct { @@ -31,6 +40,9 @@ type DingTalkChannel struct { cancel context.CancelFunc // Map to store session webhooks for each chat sessionWebhooks sync.Map // chatID -> sessionWebhook + // Health monitoring + lastMessageTime time.Time + mu sync.RWMutex } // NewDingTalkChannel creates a new DingTalk channel instance @@ -58,7 +70,23 @@ 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 + // Start the stream client + if err := c.startStreamClient(); err != nil { + return err + } + + // Start health monitoring goroutine + go c.healthMonitor() + + c.SetRunning(true) + logger.InfoC("dingtalk", "DingTalk channel started (Stream Mode)") + return nil +} + +// startStreamClient creates and starts the stream client +func (c *DingTalkChannel) startStreamClient() error { // Create credential config cred := client.NewAppCredentialConfig(c.clientID, c.clientSecret) @@ -68,7 +96,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 @@ -76,11 +104,76 @@ 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 %v", silenceDuration) + + if silenceDuration >= maxSilenceDuration { + logger.InfoCF("dingtalk", "Connection appears stale (no messages for %v), triggering recovery", 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") + c.mu.Lock() + c.lastMessageTime = time.Now() + c.mu.Unlock() + return + } + + logger.WarnCF("dingtalk", "Recovery failed: %v, retrying in %v", err, 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...") @@ -131,6 +224,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 == "" { From 3f96ccaec9f6a9cf391a52ac27ddff2a61763a05 Mon Sep 17 00:00:00 2001 From: fishtrees Date: Mon, 9 Mar 2026 16:04:27 +0800 Subject: [PATCH 2/4] refactor(dingtalk): improve logging with structured fields Convert format string logging to structured key-value logging for better log parsing and analysis in health check and recovery functions. Co-Authored-By: Claude Opus 4.6 --- pkg/channels/dingtalk/dingtalk.go | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/pkg/channels/dingtalk/dingtalk.go b/pkg/channels/dingtalk/dingtalk.go index 89ddbbc50..2f8fe7767 100644 --- a/pkg/channels/dingtalk/dingtalk.go +++ b/pkg/channels/dingtalk/dingtalk.go @@ -128,10 +128,14 @@ func (c *DingTalkChannel) checkAndRecover() { silenceDuration := time.Since(c.lastMessageTime) c.mu.RUnlock() - logger.DebugCF("dingtalk", "Health check: silence duration %v", silenceDuration) + logger.DebugCF("dingtalk", "Health check: silence duration", map[string]any{"duration": silenceDuration}) if silenceDuration >= maxSilenceDuration { - logger.InfoCF("dingtalk", "Connection appears stale (no messages for %v), triggering recovery", silenceDuration) + logger.InfoCF( + "dingtalk", + "Connection appears stale (no messages for configured duration), triggering recovery", + map[string]any{"silenceDuration": silenceDuration}, + ) c.recoverConnection() } } @@ -155,14 +159,18 @@ func (c *DingTalkChannel) recoverConnection() { err := c.startStreamClient() if err == nil { - logger.InfoCF("dingtalk", "Connection recovered successfully") + 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: %v, retrying in %v", err, recoveryRetryDelay) + logger.WarnCF( + "dingtalk", + "Recovery failed, retrying", + map[string]any{"error": err, "retryDelay": recoveryRetryDelay}, + ) time.Sleep(recoveryRetryDelay) } } From a5543b19867e133de7473f9e7749f52e22596742 Mon Sep 17 00:00:00 2001 From: fishtrees Date: Tue, 10 Mar 2026 18:01:50 +0800 Subject: [PATCH 3/4] feat(dingtalk): add proactive messaging support via robot API Enable DingTalk bot to send messages proactively without requiring user to initiate conversation first. This allows cron jobs and other automated tasks to send notifications. - Add token management with auto-refresh (token.go) - Add robot API for group/direct messaging (api.go) - Modify Send() to try session_webhook first, fallback to robot API - Store complete session info including staffId and conversationId - Auto-detect message type based on chatID format (cid prefix = group) Configuration: set proactive_send: true in dingtalk channel config. For direct messages, use staffId in the "to" field of cron jobs. Co-Authored-By: Claude Opus 4.6 --- pkg/channels/dingtalk/api.go | 151 ++++++++++++++++++++++++++++++ pkg/channels/dingtalk/dingtalk.go | 107 +++++++++++++++++---- pkg/channels/dingtalk/token.go | 139 +++++++++++++++++++++++++++ pkg/config/config.go | 1 + 4 files changed, 379 insertions(+), 19 deletions(-) create mode 100644 pkg/channels/dingtalk/api.go create mode 100644 pkg/channels/dingtalk/token.go 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 2f8fe7767..26ac4af15 100644 --- a/pkg/channels/dingtalk/dingtalk.go +++ b/pkg/channels/dingtalk/dingtalk.go @@ -6,6 +6,7 @@ package dingtalk import ( "context" "fmt" + "net/http" "sync" "time" @@ -28,6 +29,16 @@ const ( 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 { @@ -38,8 +49,13 @@ 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 @@ -72,6 +88,19 @@ func (c *DingTalkChannel) Start(ctx context.Context) error { 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 @@ -81,7 +110,9 @@ func (c *DingTalkChannel) Start(ctx context.Context) error { go c.healthMonitor() c.SetRunning(true) - logger.InfoC("dingtalk", "DingTalk channel started (Stream Mode)") + logger.InfoCF("dingtalk", "DingTalk channel started", map[string]any{ + "proactive_send": c.config.ProactiveSend, + }) return nil } @@ -199,30 +230,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 @@ -258,8 +319,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 b3ad050b7..1ce9bf6f8 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -322,6 +322,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 { From 0652258c731181b2ac52bd4c31064996cef2d3f8 Mon Sep 17 00:00:00 2001 From: fishtrees Date: Thu, 12 Mar 2026 15:42:14 +0800 Subject: [PATCH 4/4] docs(dingtalk): update documentation with new features - Add proactive_send configuration option - Add group_trigger configuration documentation - Document health monitoring and auto-recovery mechanism - Add structured logging examples - Add environment variables reference table Co-Authored-By: Claude Opus 4.6 --- docs/channels/dingtalk/README.zh.md | 76 ++++++++++++++++++++++++++--- 1 file changed, 68 insertions(+), 8 deletions(-) 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 |