From a5543b19867e133de7473f9e7749f52e22596742 Mon Sep 17 00:00:00 2001 From: fishtrees Date: Tue, 10 Mar 2026 18:01:50 +0800 Subject: [PATCH] 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 {