diff --git a/config/config.yaml b/config/config.yaml index 067437a7..0c1f0ac0 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -141,12 +141,8 @@ conversation: - Do not copy or reference any content from examples; ensure the summary is entirely based on the user's new article - Make every effort to extract key points and summarize for any text, regardless of length or complexity - ## CRITICAL: Language Rule - - Write the summary in the SAME LANGUAGE as the original document - - If the document is in Korean, write the summary in Korean - - If the document is in English, write the summary in English - - If the document is in Chinese, write the summary in Chinese - - NEVER translate the content to a different language + ## Requirements: + - Use the Chinese language for all outputs ## The following is the article information provided by the user: diff --git a/frontend/src/api/agent/index.ts b/frontend/src/api/agent/index.ts index 8a226a73..10b96c6f 100644 --- a/frontend/src/api/agent/index.ts +++ b/frontend/src/api/agent/index.ts @@ -187,6 +187,7 @@ export interface IMChannel { enabled: boolean; mode: 'webhook' | 'websocket'; output_mode: 'stream' | 'full'; + knowledge_base_id?: string; credentials: Record; created_at?: string; updated_at?: string; diff --git a/frontend/src/components/IMChannelPanel.vue b/frontend/src/components/IMChannelPanel.vue index 2e54e3a9..d4b5da79 100644 --- a/frontend/src/components/IMChannelPanel.vue +++ b/frontend/src/components/IMChannelPanel.vue @@ -116,6 +116,20 @@ + +
+ + + + +

{{ $t('agentEditor.im.fileKnowledgeBaseHint') }}

+
+
@@ -200,6 +214,7 @@ import { ref, onMounted } from 'vue'; import { useI18n } from 'vue-i18n'; import { MessagePlugin } from 'tdesign-vue-next'; import { listIMChannels, createIMChannel, updateIMChannel, deleteIMChannel, toggleIMChannel } from '@/api/agent'; +import { listKnowledgeBases } from '@/api/knowledge-base'; import type { IMChannel } from '@/api/agent'; const { t } = useI18n(); @@ -213,6 +228,9 @@ const loading = ref(false); const showCreateDialog = ref(false); const editingChannel = ref(null); +// Knowledge base options for file-to-KB feature +const knowledgeBases = ref<{ id: string; name: string }[]>([]); + const defaultCredentials = (): Record => ({}); const formData = ref({ @@ -220,14 +238,19 @@ const formData = ref({ name: '', mode: 'websocket' as 'webhook' | 'websocket', output_mode: 'stream' as 'stream' | 'full', + knowledge_base_id: '', credentials: defaultCredentials(), }); async function loadChannels() { loading.value = true; try { - const res = await listIMChannels(props.agentId); - channels.value = res.data || []; + const [channelRes, kbRes] = await Promise.all([ + listIMChannels(props.agentId), + listKnowledgeBases(), + ]); + channels.value = channelRes.data || []; + knowledgeBases.value = (kbRes.data || []).map((kb: any) => ({ id: kb.id, name: kb.name })); } catch { channels.value = []; } finally { @@ -256,6 +279,7 @@ function editChannel(channel: IMChannel) { name: channel.name, mode: channel.mode, output_mode: channel.output_mode, + knowledge_base_id: channel.knowledge_base_id || '', credentials: { ...channel.credentials }, }; showCreateDialog.value = true; @@ -268,6 +292,7 @@ function resetForm() { name: '', mode: 'websocket', output_mode: 'stream', + knowledge_base_id: '', credentials: defaultCredentials(), }; } @@ -279,6 +304,7 @@ async function handleSave() { name: formData.value.name, mode: formData.value.mode, output_mode: formData.value.output_mode, + knowledge_base_id: formData.value.knowledge_base_id, credentials: formData.value.credentials, }); MessagePlugin.success(t('common.updateSuccess')); @@ -288,6 +314,7 @@ async function handleSave() { name: formData.value.name, mode: formData.value.mode, output_mode: formData.value.output_mode, + knowledge_base_id: formData.value.knowledge_base_id, credentials: formData.value.credentials, }); MessagePlugin.success(t('common.createSuccess')); @@ -296,7 +323,8 @@ async function handleSave() { resetForm(); await loadChannels(); } catch (e: any) { - MessagePlugin.error(e?.message || t('common.operationFailed')); + const msg = e?.message || (typeof e?.error === 'string' ? e.error : null) || t('common.operationFailed'); + MessagePlugin.error(msg); } } diff --git a/frontend/src/i18n/locales/en-US.ts b/frontend/src/i18n/locales/en-US.ts index 1f6174ff..7a8a37a6 100755 --- a/frontend/src/i18n/locales/en-US.ts +++ b/frontend/src/i18n/locales/en-US.ts @@ -3275,6 +3275,9 @@ export default { feishuConsole: 'Feishu Open Platform', modeHint: 'WebSocket is recommended for easier setup', consoleTip: 'to get credentials', + fileKnowledgeBase: 'File Storage Knowledge Base', + fileKnowledgeBasePlaceholder: 'Select a knowledge base (optional)', + fileKnowledgeBaseHint: 'When configured, files sent by users will be automatically saved to this knowledge base', }, mcp: { label: 'MCP Services', diff --git a/frontend/src/i18n/locales/ko-KR.ts b/frontend/src/i18n/locales/ko-KR.ts index ea538ad0..2515a251 100755 --- a/frontend/src/i18n/locales/ko-KR.ts +++ b/frontend/src/i18n/locales/ko-KR.ts @@ -3305,6 +3305,9 @@ export default { feishuConsole: "Feishu 개발 플랫폼", modeHint: "WebSocket 방식이 설정이 더 간편하여 권장됩니다", consoleTip: "자격 증명 정보를 가져오세요", + fileKnowledgeBase: "파일 저장 지식 베이스", + fileKnowledgeBasePlaceholder: "지식 베이스 선택 (선택 사항)", + fileKnowledgeBaseHint: "설정 시 사용자가 보낸 파일이 자동으로 해당 지식 베이스에 저장됩니다", }, tools: { thinking: '사고', diff --git a/frontend/src/i18n/locales/ru-RU.ts b/frontend/src/i18n/locales/ru-RU.ts index 8f831a9c..bfb67fb0 100755 --- a/frontend/src/i18n/locales/ru-RU.ts +++ b/frontend/src/i18n/locales/ru-RU.ts @@ -2858,6 +2858,9 @@ export default { feishuConsole: 'Платформа Feishu', modeHint: 'Рекомендуется WebSocket — проще настроить', consoleTip: 'для получения учётных данных', + fileKnowledgeBase: 'База знаний для файлов', + fileKnowledgeBasePlaceholder: 'Выберите базу знаний (необязательно)', + fileKnowledgeBaseHint: 'При настройке файлы, отправленные пользователями, автоматически сохраняются в эту базу знаний', }, mcp: { label: 'MCP-сервисы', diff --git a/frontend/src/i18n/locales/zh-CN.ts b/frontend/src/i18n/locales/zh-CN.ts index 5d667025..66bce2ff 100755 --- a/frontend/src/i18n/locales/zh-CN.ts +++ b/frontend/src/i18n/locales/zh-CN.ts @@ -3250,6 +3250,9 @@ export default { feishuConsole: "飞书开放平台", modeHint: "推荐使用 WebSocket 方式接入，配置更简单", consoleTip: "前往获取凭证信息", + fileKnowledgeBase: "文件保存知识库", + fileKnowledgeBasePlaceholder: "选择知识库（可选）", + fileKnowledgeBaseHint: "配置后，用户发送的文件将自动保存到该知识库中", }, tools: { thinking: "思考", diff --git a/internal/agent/engine.go b/internal/agent/engine.go index 1ef64f41..b5fc0964 100644 --- a/internal/agent/engine.go +++ b/internal/agent/engine.go @@ -18,6 +18,12 @@ import ( "github.com/google/uuid" ) +const ( + // llmPerCallTimeout is the maximum time allowed for a single LLM call (stream initiation + full response). + // This prevents a single slow call from consuming the entire pipeline deadline. + llmPerCallTimeout = 120 * time.Second +) + // generateEventID generates a unique event ID with type suffix for better traceability func generateEventID(suffix string) string { return fmt.Sprintf("%s-%s", uuid.New().String()[:8], suffix) @@ -719,7 +725,14 @@ func (e *AgentEngine) streamLLMToEventBus( ) (string, []types.LLMToolCall, error) { logger.Debugf(ctx, "[Agent][Stream] Starting LLM stream with %d messages", len(messages)) - stream, err := e.chatModel.ChatStream(ctx, messages, opts) + // Create a per-call timeout context so that a single LLM call gets a + // guaranteed time window even when the parent context's deadline is almost + // exhausted after previous iterations. The shorter of the two deadlines wins, + // so the parent pipeline timeout is still respected. + llmCtx, llmCancel := context.WithTimeout(ctx, llmPerCallTimeout) + defer llmCancel() + + stream, err := e.chatModel.ChatStream(llmCtx, messages, opts) if err != nil { logger.Errorf(ctx, "[Agent][Stream] Failed to start LLM stream: %v", err) return "", nil, err diff --git a/internal/application/service/image_multimodal.go b/internal/application/service/image_multimodal.go index 8530af6e..4987da84 100644 --- a/internal/application/service/image_multimodal.go +++ b/internal/application/service/image_multimodal.go @@ -29,7 +29,7 @@ const ( "4. Organize content in the original reading order\n" + "5. Only output extracted text content, do not add any HTML tags\n" + "If there is no recognizable text content in the image, reply: No text content." - vlmCaptionPrompt = "Provide a brief and concise description of the main content of the image" + vlmCaptionPrompt = "Provide a brief and concise description of the main content of the image in Chinese" ) // ImageMultimodalService handles image:multimodal asynq tasks. diff --git a/internal/handler/im.go b/internal/handler/im.go index ec5f7942..cd3e00f1 100644 --- a/internal/handler/im.go +++ b/internal/handler/im.go @@ -3,6 +3,7 @@ package handler import ( "context" "net/http" + "strings" "github.com/Tencent/WeKnora/internal/im" "github.com/Tencent/WeKnora/internal/logger" @@ -39,12 +40,13 @@ func (h *IMHandler) CreateIMChannel(c *gin.Context) { } var req struct { - Platform string `json:"platform" binding:"required"` - Name string `json:"name"` - Mode string `json:"mode"` - OutputMode string `json:"output_mode"` - Credentials types.JSON `json:"credentials"` - Enabled *bool `json:"enabled"` + Platform string `json:"platform" binding:"required"` + Name string `json:"name"` + Mode string `json:"mode"` + OutputMode string `json:"output_mode"` + KnowledgeBaseID string `json:"knowledge_base_id"` + Credentials types.JSON `json:"credentials"` + Enabled *bool `json:"enabled"` } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) @@ -57,14 +59,15 @@ func (h *IMHandler) CreateIMChannel(c *gin.Context) { } channel := &im.IMChannel{ - TenantID: tenantID, - AgentID: agentID, - Platform: req.Platform, - Name: req.Name, - Mode: req.Mode, - OutputMode: req.OutputMode, - Credentials: req.Credentials, - Enabled: true, + TenantID: tenantID, + AgentID: agentID, + Platform: req.Platform, + Name: req.Name, + Mode: req.Mode, + OutputMode: req.OutputMode, + KnowledgeBaseID: req.KnowledgeBaseID, + Credentials: req.Credentials, + Enabled: true, } if req.Enabled != nil { channel.Enabled = *req.Enabled @@ -81,6 +84,10 @@ func (h *IMHandler) CreateIMChannel(c *gin.Context) { if err := h.imService.CreateChannel(channel); err != nil { logger.Errorf(c.Request.Context(), "[IM] Create channel failed: %v", err) + if strings.HasPrefix(err.Error(), "duplicate_bot:") { + c.JSON(http.StatusConflict, gin.H{"error": strings.TrimPrefix(err.Error(), "duplicate_bot: ")}) + return + } c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create channel"}) return } @@ -96,7 +103,13 @@ func (h *IMHandler) ListIMChannels(c *gin.Context) { return } - channels, err := h.imService.ListChannelsByAgent(agentID) + tenantID, ok := c.Request.Context().Value(types.TenantIDContextKey).(uint64) + if !ok { + c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"}) + return + } + + channels, err := h.imService.ListChannelsByAgent(agentID, tenantID) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list channels"}) return @@ -113,18 +126,25 @@ func (h *IMHandler) UpdateIMChannel(c *gin.Context) { return } - channel, err := h.imService.GetChannelByID(channelID) + tenantID, ok := c.Request.Context().Value(types.TenantIDContextKey).(uint64) + if !ok { + c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"}) + return + } + + channel, err := h.imService.GetChannelByIDAndTenant(channelID, tenantID) if err != nil { c.JSON(http.StatusNotFound, gin.H{"error": "channel not found"}) return } var req struct { - Name *string `json:"name"` - Mode *string `json:"mode"` - OutputMode *string `json:"output_mode"` - Credentials types.JSON `json:"credentials"` - Enabled *bool `json:"enabled"` + Name *string `json:"name"` + Mode *string `json:"mode"` + OutputMode *string `json:"output_mode"` + KnowledgeBaseID *string `json:"knowledge_base_id"` + Credentials types.JSON `json:"credentials"` + Enabled *bool `json:"enabled"` } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) @@ -140,6 +160,9 @@ func (h *IMHandler) UpdateIMChannel(c *gin.Context) { if req.OutputMode != nil { channel.OutputMode = *req.OutputMode } + if req.KnowledgeBaseID != nil { + channel.KnowledgeBaseID = *req.KnowledgeBaseID + } if req.Credentials != nil { channel.Credentials = req.Credentials } @@ -149,6 +172,10 @@ func (h *IMHandler) UpdateIMChannel(c *gin.Context) { if err := h.imService.UpdateChannel(channel); err != nil { logger.Errorf(c.Request.Context(), "[IM] Update channel failed: %v", err) + if strings.HasPrefix(err.Error(), "duplicate_bot:") { + c.JSON(http.StatusConflict, gin.H{"error": strings.TrimPrefix(err.Error(), "duplicate_bot: ")}) + return + } c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update channel"}) return } @@ -164,7 +191,13 @@ func (h *IMHandler) DeleteIMChannel(c *gin.Context) { return } - if err := h.imService.DeleteChannel(channelID); err != nil { + tenantID, ok := c.Request.Context().Value(types.TenantIDContextKey).(uint64) + if !ok { + c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"}) + return + } + + if err := h.imService.DeleteChannel(channelID, tenantID); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete channel"}) return } @@ -180,7 +213,13 @@ func (h *IMHandler) ToggleIMChannel(c *gin.Context) { return } - channel, err := h.imService.ToggleChannel(channelID) + tenantID, ok := c.Request.Context().Value(types.TenantIDContextKey).(uint64) + if !ok { + c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"}) + return + } + + channel, err := h.imService.ToggleChannel(channelID, tenantID) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to toggle channel"}) return diff --git a/internal/im/adapter.go b/internal/im/adapter.go index d23fd89d..374a685d 100644 --- a/internal/im/adapter.go +++ b/internal/im/adapter.go @@ -2,6 +2,7 @@ package im import ( "context" + "io" "github.com/gin-gonic/gin" ) @@ -14,10 +15,21 @@ const ( PlatformFeishu Platform = "feishu" ) +// MessageType identifies the kind of IM message. +type MessageType string + +const ( + MessageTypeText MessageType = "text" + MessageTypeFile MessageType = "file" + MessageTypeImage MessageType = "image" +) + // IncomingMessage is the unified message parsed from an IM callback. type IncomingMessage struct { // Platform identifies which IM platform the message comes from. Platform Platform + // MessageType is "text" (default) or "file". + MessageType MessageType // UserID is the IM-platform user identifier. UserID string // UserName is the display name of the user (optional). @@ -26,10 +38,16 @@ type IncomingMessage struct { ChatID string // ChatType distinguishes direct message from group chat. ChatType ChatType - // Content is the text content of the message. + // Content is the text content of the message (empty for file messages). Content string // MessageID is the IM-platform message identifier (for dedup). MessageID string + // FileKey is the platform file identifier (for file messages). + FileKey string + // FileName is the original file name (for file messages). + FileName string + // FileSize is the file size in bytes (for file messages, optional). + FileSize int64 // Extra holds platform-specific fields (e.g., WeCom stream ID). Extra map[string]string } @@ -89,3 +107,13 @@ type StreamSender interface { // EndStream finalizes a streaming reply. EndStream(ctx context.Context, incoming *IncomingMessage, streamID string) error } + +// FileDownloader is an optional interface that adapters can implement to support +// downloading file attachments from the IM platform. When the adapter implements +// this interface and the IM channel has a knowledge_base_id configured, file +// messages will be downloaded and saved to the specified knowledge base. +type FileDownloader interface { + // DownloadFile downloads a file resource from the IM platform. + // Returns the file content reader, the resolved file name, and any error. + DownloadFile(ctx context.Context, msg *IncomingMessage) (io.ReadCloser, string, error) +} diff --git a/internal/im/feishu/adapter.go b/internal/im/feishu/adapter.go index 7b1ba5ed..b24d73b3 100644 --- a/internal/im/feishu/adapter.go +++ b/internal/im/feishu/adapter.go @@ -29,8 +29,9 @@ import ( "github.com/gin-gonic/gin" ) -// Compile-time check that Adapter implements im.StreamSender. +// Compile-time check that Adapter implements im.StreamSender and im.FileDownloader. var _ im.StreamSender = (*Adapter)(nil) +var _ im.FileDownloader = (*Adapter)(nil) var httpClient = &http.Client{Timeout: 10 * time.Second} @@ -223,19 +224,6 @@ func (a *Adapter) ParseCallback(c *gin.Context) (*im.IncomingMessage, error) { } msg := eventBody.Event.Message - if msg.MessageType != "text" { - logger.Infof(c.Request.Context(), "[Feishu] Ignoring non-text message type: %s", msg.MessageType) - return nil, nil - } - - // Parse text content - var textContent struct { - Text string `json:"text"` - } - if err := json.Unmarshal([]byte(msg.Content), &textContent); err != nil { - return nil, fmt.Errorf("unmarshal text content: %w", err) - } - // Determine chat type chatType := im.ChatTypeDirect chatID := "" @@ -250,28 +238,146 @@ func (a *Adapter) ParseCallback(c *gin.Context) (*im.IncomingMessage, error) { openID = eventBody.Event.Sender.SenderID.OpenID } - // Strip @bot mention from group messages - content := textContent.Text - if chatType == im.ChatTypeGroup { - // Feishu @mentions are in the format @_user_xxx - for strings.HasPrefix(content, "@_user_") { - idx := strings.Index(content, " ") - if idx >= 0 { - content = content[idx+1:] - } else { - break + switch msg.MessageType { + case "text": + // Parse text content + var textContent struct { + Text string `json:"text"` + } + if err := json.Unmarshal([]byte(msg.Content), &textContent); err != nil { + return nil, fmt.Errorf("unmarshal text content: %w", err) + } + + // Strip @bot mention from group messages + content := textContent.Text + if chatType == im.ChatTypeGroup { + for strings.HasPrefix(content, "@_user_") { + idx := strings.Index(content, " ") + if idx >= 0 { + content = content[idx+1:] + } else { + break + } } } - } - return &im.IncomingMessage{ - Platform: im.PlatformFeishu, - UserID: openID, - ChatID: chatID, - ChatType: chatType, - Content: strings.TrimSpace(content), - MessageID: msg.MessageID, - }, nil + return &im.IncomingMessage{ + Platform: im.PlatformFeishu, + MessageType: im.MessageTypeText, + UserID: openID, + ChatID: chatID, + ChatType: chatType, + Content: strings.TrimSpace(content), + MessageID: msg.MessageID, + }, nil + + case "file": + var fileContent struct { + FileKey string `json:"file_key"` + FileName string `json:"file_name"` + } + if err := json.Unmarshal([]byte(msg.Content), &fileContent); err != nil { + return nil, fmt.Errorf("unmarshal file content: %w", err) + } + if fileContent.FileKey == "" { + return nil, nil + } + return &im.IncomingMessage{ + Platform: im.PlatformFeishu, + MessageType: im.MessageTypeFile, + UserID: openID, + ChatID: chatID, + ChatType: chatType, + MessageID: msg.MessageID, + FileKey: fileContent.FileKey, + FileName: fileContent.FileName, + }, nil + + case "image": + var imageContent struct { + ImageKey string `json:"image_key"` + } + if err := json.Unmarshal([]byte(msg.Content), &imageContent); err != nil { + return nil, fmt.Errorf("unmarshal image content: %w", err) + } + if imageContent.ImageKey == "" { + return nil, nil + } + return &im.IncomingMessage{ + Platform: im.PlatformFeishu, + MessageType: im.MessageTypeImage, + UserID: openID, + ChatID: chatID, + ChatType: chatType, + MessageID: msg.MessageID, + FileKey: imageContent.ImageKey, + FileName: imageContent.ImageKey + ".png", + }, nil + + case "post": + // Rich text: extract plain text for QA + var postContent struct { + Title string `json:"title"` + Content [][]json.RawMessage `json:"content"` + } + if err := json.Unmarshal([]byte(msg.Content), &postContent); err != nil { + return nil, fmt.Errorf("unmarshal post content: %w", err) + } + + var textParts []string + if postContent.Title != "" { + textParts = append(textParts, postContent.Title) + } + for _, line := range postContent.Content { + var lineText strings.Builder + for _, elem := range line { + var tag struct { + Tag string `json:"tag"` + Text string `json:"text"` + } + if err := json.Unmarshal(elem, &tag); err != nil { + continue + } + switch tag.Tag { + case "text", "a": + lineText.WriteString(tag.Text) + } + } + if t := strings.TrimSpace(lineText.String()); t != "" { + textParts = append(textParts, t) + } + } + + content := strings.Join(textParts, "\n") + if chatType == im.ChatTypeGroup { + for strings.HasPrefix(content, "@_user_") { + idx := strings.Index(content, " ") + if idx >= 0 { + content = content[idx+1:] + } else { + break + } + } + } + content = strings.TrimSpace(content) + if content == "" { + return nil, nil + } + + return &im.IncomingMessage{ + Platform: im.PlatformFeishu, + MessageType: im.MessageTypeText, + UserID: openID, + ChatID: chatID, + ChatType: chatType, + Content: content, + MessageID: msg.MessageID, + }, nil + + default: + logger.Infof(c.Request.Context(), "[Feishu] Ignoring unsupported message type: %s", msg.MessageType) + return nil, nil + } } // SendReply sends a reply message via Feishu API. @@ -327,6 +433,63 @@ func (a *Adapter) SendReply(ctx context.Context, incoming *im.IncomingMessage, r return nil } +// ────────────────────────────────────────────────────────────────────── +// File download support via Feishu GetMessageResource API +// ────────────────────────────────────────────────────────────────────── + +// DownloadFile downloads a file or image attachment from a Feishu message. +// Uses the GetMessageResource API: GET /open-apis/im/v1/messages/:message_id/resources/:file_key?type={file|image} +func (a *Adapter) DownloadFile(ctx context.Context, msg *im.IncomingMessage) (io.ReadCloser, string, error) { + if msg.FileKey == "" || msg.MessageID == "" { + return nil, "", fmt.Errorf("file_key and message_id are required") + } + + accessToken, err := a.getTenantAccessToken(ctx) + if err != nil { + return nil, "", fmt.Errorf("get access token: %w", err) + } + + // Determine resource type based on message type + resourceType := "file" + if msg.MessageType == im.MessageTypeImage { + resourceType = "image" + } + + apiURL := fmt.Sprintf("https://open.feishu.cn/open-apis/im/v1/messages/%s/resources/%s?type=%s", + msg.MessageID, msg.FileKey, resourceType) + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, apiURL, nil) + if err != nil { + return nil, "", fmt.Errorf("create request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+accessToken) + + resp, err := httpClient.Do(req) + if err != nil { + return nil, "", fmt.Errorf("download file: %w", err) + } + + if resp.StatusCode != http.StatusOK { + resp.Body.Close() + return nil, "", fmt.Errorf("download file failed: status=%d", resp.StatusCode) + } + + // Use the original file name from the message, or extract from Content-Disposition + fileName := msg.FileName + if fileName == "" { + if cd := resp.Header.Get("Content-Disposition"); cd != "" { + if idx := strings.Index(cd, "filename="); idx >= 0 { + fileName = strings.Trim(cd[idx+len("filename="):], "\" ") + } + } + } + if fileName == "" { + fileName = msg.FileKey + } + + return resp.Body, fileName, nil +} + // ────────────────────────────────────────────────────────────────────── // Feishu CardKit v1 streaming implementation (official best practice) // diff --git a/internal/im/feishu/longconn.go b/internal/im/feishu/longconn.go index aec48f57..1f70bf66 100644 --- a/internal/im/feishu/longconn.go +++ b/internal/im/feishu/longconn.go @@ -81,27 +81,18 @@ func (l *feishuLoggerAdapter) Error(ctx context.Context, args ...interface{}) { } // convertEvent converts a Feishu SDK event to a unified IncomingMessage. -// Returns nil for non-text messages. +// Supports text and file messages. Returns nil for unsupported types. func convertEvent(event *larkim.P2MessageReceiveV1) *im.IncomingMessage { if event == nil || event.Event == nil || event.Event.Message == nil { return nil } msg := event.Event.Message - if msg.MessageType == nil || *msg.MessageType != "text" { + if msg.MessageType == nil { return nil } - // Parse text content from JSON: {"text": "..."} - var textContent struct { - Text string `json:"text"` - } - if msg.Content == nil { - return nil - } - if err := json.Unmarshal([]byte(*msg.Content), &textContent); err != nil { - return nil - } + msgType := *msg.MessageType // Sender info openID := "" @@ -125,7 +116,32 @@ func convertEvent(event *larkim.P2MessageReceiveV1) *im.IncomingMessage { messageID = *msg.MessageId } - // Strip @bot mention from group messages + switch msgType { + case "text": + return convertTextEvent(msg, openID, chatID, chatType, messageID) + case "file": + return convertFileEvent(msg, openID, chatID, chatType, messageID) + case "image": + return convertImageEvent(msg, openID, chatID, chatType, messageID) + case "post": + return convertPostEvent(msg, openID, chatID, chatType, messageID) + default: + return nil + } +} + +// convertTextEvent handles text message type. +func convertTextEvent(msg *larkim.EventMessage, openID, chatID string, chatType im.ChatType, messageID string) *im.IncomingMessage { + var textContent struct { + Text string `json:"text"` + } + if msg.Content == nil { + return nil + } + if err := json.Unmarshal([]byte(*msg.Content), &textContent); err != nil { + return nil + } + content := textContent.Text if chatType == im.ChatTypeGroup { for strings.HasPrefix(content, "@_user_") { @@ -139,11 +155,139 @@ func convertEvent(event *larkim.P2MessageReceiveV1) *im.IncomingMessage { } return &im.IncomingMessage{ - Platform: im.PlatformFeishu, - UserID: openID, - ChatID: chatID, - ChatType: chatType, - Content: strings.TrimSpace(content), - MessageID: messageID, + Platform: im.PlatformFeishu, + MessageType: im.MessageTypeText, + UserID: openID, + ChatID: chatID, + ChatType: chatType, + Content: strings.TrimSpace(content), + MessageID: messageID, + } +} + +// convertFileEvent handles file message type. +func convertFileEvent(msg *larkim.EventMessage, openID, chatID string, chatType im.ChatType, messageID string) *im.IncomingMessage { + if msg.Content == nil { + return nil + } + var fileContent struct { + FileKey string `json:"file_key"` + FileName string `json:"file_name"` + } + if err := json.Unmarshal([]byte(*msg.Content), &fileContent); err != nil { + return nil + } + if fileContent.FileKey == "" { + return nil + } + + return &im.IncomingMessage{ + Platform: im.PlatformFeishu, + MessageType: im.MessageTypeFile, + UserID: openID, + ChatID: chatID, + ChatType: chatType, + MessageID: messageID, + FileKey: fileContent.FileKey, + FileName: fileContent.FileName, + } +} + +// convertImageEvent handles image message type. +// Downloads via GetMessageResource API with type=image. +func convertImageEvent(msg *larkim.EventMessage, openID, chatID string, chatType im.ChatType, messageID string) *im.IncomingMessage { + if msg.Content == nil { + return nil + } + var imageContent struct { + ImageKey string `json:"image_key"` + } + if err := json.Unmarshal([]byte(*msg.Content), &imageContent); err != nil { + return nil + } + if imageContent.ImageKey == "" { + return nil + } + + return &im.IncomingMessage{ + Platform: im.PlatformFeishu, + MessageType: im.MessageTypeImage, + UserID: openID, + ChatID: chatID, + ChatType: chatType, + MessageID: messageID, + FileKey: imageContent.ImageKey, + FileName: imageContent.ImageKey + ".png", + } +} + +// convertPostEvent handles rich-text (post) message type. +// Extracts all plain text content and treats it as a text query for QA. +func convertPostEvent(msg *larkim.EventMessage, openID, chatID string, chatType im.ChatType, messageID string) *im.IncomingMessage { + if msg.Content == nil { + return nil + } + + // Post content structure: {"title":"...", "content":[[{"tag":"text","text":"..."},{"tag":"a","href":"...","text":"..."}]]} + var postContent struct { + Title string `json:"title"` + Content [][]json.RawMessage `json:"content"` + } + if err := json.Unmarshal([]byte(*msg.Content), &postContent); err != nil { + return nil + } + + var textParts []string + if postContent.Title != "" { + textParts = append(textParts, postContent.Title) + } + + for _, line := range postContent.Content { + var lineText strings.Builder + for _, elem := range line { + var tag struct { + Tag string `json:"tag"` + Text string `json:"text"` + } + if err := json.Unmarshal(elem, &tag); err != nil { + continue + } + switch tag.Tag { + case "text", "a": + lineText.WriteString(tag.Text) + case "at": + // Skip @mentions + } + } + if t := strings.TrimSpace(lineText.String()); t != "" { + textParts = append(textParts, t) + } + } + + content := strings.Join(textParts, "\n") + if chatType == im.ChatTypeGroup { + for strings.HasPrefix(content, "@_user_") { + idx := strings.Index(content, " ") + if idx >= 0 { + content = content[idx+1:] + } else { + break + } + } + } + + content = strings.TrimSpace(content) + if content == "" { + return nil + } + + return &im.IncomingMessage{ + Platform: im.PlatformFeishu, + MessageType: im.MessageTypeText, + UserID: openID, + ChatID: chatID, + ChatType: chatType, + Content: content, + MessageID: messageID, } } diff --git a/internal/im/service.go b/internal/im/service.go index ff521c2d..55af43f5 100644 --- a/internal/im/service.go +++ b/internal/im/service.go @@ -1,14 +1,19 @@ package im import ( + "bytes" "context" "fmt" + "io" + "mime/multipart" + "net/textproto" "strings" "sync" "time" "github.com/Tencent/WeKnora/internal/event" "github.com/Tencent/WeKnora/internal/logger" + "github.com/Tencent/WeKnora/internal/models/chat" "github.com/Tencent/WeKnora/internal/types" "github.com/Tencent/WeKnora/internal/types/interfaces" "github.com/google/uuid" @@ -52,6 +57,12 @@ type Service struct { tenantService interfaces.TenantService agentService interfaces.CustomAgentService + // knowledgeService is used for saving IM file messages to knowledge bases. + knowledgeService interfaces.KnowledgeService + + // modelService is used to obtain the chat model for generating smart notification replies. + modelService interfaces.ModelService + // channels maps channel ID -> running channel state channels map[string]*channelState mu sync.RWMutex @@ -90,6 +101,8 @@ func NewService( messageService interfaces.MessageService, tenantService interfaces.TenantService, agentService interfaces.CustomAgentService, + knowledgeService interfaces.KnowledgeService, + modelService interfaces.ModelService, ) *Service { s := &Service{ db: db, @@ -97,6 +110,8 @@ func NewService( messageService: messageService, tenantService: tenantService, agentService: agentService, + knowledgeService: knowledgeService, + modelService: modelService, channels: make(map[string]*channelState), adapterFactories: make(map[string]AdapterFactory), stopCh: make(chan struct{}), @@ -242,6 +257,15 @@ func (s *Service) GetChannelByID(channelID string) (*IMChannel, error) { return &ch, nil } +// GetChannelByIDAndTenant loads a channel from the database, scoped to a specific tenant. +func (s *Service) GetChannelByIDAndTenant(channelID string, tenantID uint64) (*IMChannel, error) { + var ch IMChannel + if err := s.db.Where("id = ? AND tenant_id = ? AND deleted_at IS NULL", channelID, tenantID).First(&ch).Error; err != nil { + return nil, err + } + return &ch, nil +} + // HandleMessage processes an incoming IM message end-to-end using channel config. func (s *Service) HandleMessage(ctx context.Context, msg *IncomingMessage, channelID string) error { // Dedup: skip if this message was already processed (IM platforms may retry) @@ -280,8 +304,17 @@ func (s *Service) HandleMessage(ctx context.Context, msg *IncomingMessage, chann tenantID := channel.TenantID agentID := channel.AgentID - logger.Infof(ctx, "[IM] HandleMessage: channel=%s platform=%s user=%s chat=%s content_len=%d", - channelID, msg.Platform, msg.UserID, msg.ChatID, len(msg.Content)) + logger.Infof(ctx, "[IM] HandleMessage: channel=%s platform=%s user=%s chat=%s msgtype=%s content_len=%d", + channelID, msg.Platform, msg.UserID, msg.ChatID, msg.MessageType, len(msg.Content)) + logger.Debugf(ctx, "[IM] HandleMessage detail: msgid=%s filekey=%s filename=%s", + msg.MessageID, msg.FileKey, msg.FileName) + + // ── File/Image message shortcut ── + // If the message is a file or image and the channel has a knowledge_base_id configured, + // handle it separately without entering the QA pipeline. + if (msg.MessageType == MessageTypeFile || msg.MessageType == MessageTypeImage) && channel.KnowledgeBaseID != "" { + return s.handleFileMessage(ctx, msg, adapter, channel) + } // 1. Get tenant tenant, err := s.tenantService.GetTenantByID(ctx, tenantID) @@ -410,6 +443,71 @@ func (s *Service) resolveSession(ctx context.Context, msg *IncomingMessage, tena return &cs, nil } +// ── Agent tool call progress formatting ────────────────────────────── +// These helpers format tool-call / tool-result events as Markdown text +// that is injected into the streaming reply so IM users can see the +// agent's reasoning process in real-time. +// ───────────────────────────────────────────────────────────────────── + +// toolDisplayNames maps internal tool function names to user-friendly labels. +var toolDisplayNames = map[string]string{ + "thinking": "深度思考", + "todo_write": "制定计划", + "knowledge_search": "知识库检索", + "grep_chunks": "关键词搜索", + "list_knowledge_chunks": "查看文档分块", + "query_knowledge_graph": "查询知识图谱", + "get_document_info": "获取文档信息", + "database_query": "查询数据库", + "data_analysis": "数据分析", + "data_schema": "查看数据元信息", + "web_search": "网络搜索", + "web_fetch": "网页阅读", + "read_skill": "读取技能", + "execute_skill_script": "执行技能脚本", + "final_answer": "生成回答", +} + +// internalToolNames lists tools whose execution should NOT be displayed in IM +// messages because they are internal reasoning aids (thinking, planning) rather +// than user-facing actions. +var internalToolNames = map[string]bool{ + "thinking": true, + "todo_write": true, +} + +// friendlyToolName returns a human-readable name for a tool. +func friendlyToolName(toolName string) string { + if display, ok := toolDisplayNames[toolName]; ok { + return display + } + return toolName +} + +// isToolVisibleToUser returns true if the tool's execution progress should be +// displayed to the IM user. Internal reasoning tools (thinking, planning) and +// the final_answer pseudo-tool are hidden. +func isToolVisibleToUser(toolName string) bool { + if toolName == "final_answer" { + return false + } + return !internalToolNames[toolName] +} + +// formatToolCallStart returns a Markdown snippet announcing that a tool is being invoked. +func formatToolCallStart(toolName string) string { + return fmt.Sprintf("\n\n> 🔧 正在调用工具: **%s** ...\n\n", friendlyToolName(toolName)) +} + +// formatToolCallResult returns a Markdown snippet summarising a tool result. +func formatToolCallResult(toolName string, success bool, durationMs int64) string { + friendly := friendlyToolName(toolName) + if success { + return fmt.Sprintf("> ✅ **%s** 完成 (%.1fs)\n\n", friendly, float64(durationMs)/1000) + } + return fmt.Sprintf("> ⚠️ **%s** 失败 (%.1fs)\n\n", friendly, float64(durationMs)/1000) +} + // handleMessageStream runs the QA pipeline and streams answer chunks to the IM platform // in real-time via the StreamSender interface. Chunks are batched at streamFlushInterval // to avoid API rate-limiting. @@ -435,6 +533,8 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage, done = make(chan struct{}) closeOnce sync.Once inThinking bool // tracks whether we're inside a block + hasToolCalls bool // whether any tool call events have been received + answerStarted bool // whether the final answer stream has begun ) closeDone := func() { closeOnce.Do(func() { close(done) }) } @@ -469,6 +569,12 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage, } if content != "" { + // Insert a separator before the first answer content when tool calls were displayed, + // so the user can clearly distinguish the tool-call progress from the actual answer. + if hasToolCalls && !answerStarted { + answerStarted = true + buf.WriteString("---\n\n") + } buf.WriteString(content) } bufMu.Unlock() @@ -492,6 +598,42 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage, return nil }) + // Subscribe to agent tool call events — show "calling tool X..." in real-time + eventBus.On(event.EventAgentToolCall, func(_ context.Context, evt event.Event) error { + data, ok := evt.Data.(event.AgentToolCallData) + if !ok { + return nil + } + if !isToolVisibleToUser(data.ToolName) { + return nil + } + snippet := formatToolCallStart(data.ToolName) + bufMu.Lock() + hasToolCalls = true + buf.WriteString(snippet) + bufMu.Unlock() + logger.Debugf(ctx, "[IM] Tool call streamed to IM: tool=%s", data.ToolName) + return nil + }) + + // Subscribe to agent tool result events — show "tool X done" in real-time + eventBus.On(event.EventAgentToolResult, func(_ context.Context, evt event.Event) error { + data, ok := evt.Data.(event.AgentToolResultData) + if !ok { + return nil + } + if !isToolVisibleToUser(data.ToolName) { + return nil + } + snippet := formatToolCallResult(data.ToolName, data.Success, data.Duration) + bufMu.Lock() + buf.WriteString(snippet) + bufMu.Unlock() + logger.Debugf(ctx, "[IM] Tool result streamed to IM: tool=%s success=%v duration=%dms", + data.ToolName, data.Success, data.Duration) + return nil + }) + // Determine whether to use agent mode useAgent := customAgent != nil && customAgent.IsAgentMode() requestID := uuid.New().String() @@ -733,10 +875,10 @@ func (s *Service) runQA(ctx context.Context, session *types.Session, query strin // ── CRUD operations for IM channels ── -// ListChannelsByAgent returns all channels for a given agent. -func (s *Service) ListChannelsByAgent(agentID string) ([]IMChannel, error) { +// ListChannelsByAgent returns all channels for a given agent within a tenant. +func (s *Service) ListChannelsByAgent(agentID string, tenantID uint64) ([]IMChannel, error) { var channels []IMChannel - if err := s.db.Where("agent_id = ? AND deleted_at IS NULL", agentID). + if err := s.db.Where("agent_id = ? AND tenant_id = ? AND deleted_at IS NULL", agentID, tenantID). Order("created_at DESC").Find(&channels).Error; err != nil { return nil, err } @@ -744,7 +886,11 @@ func (s *Service) ListChannelsByAgent(agentID string) ([]IMChannel, error) { } // CreateChannel creates a new IM channel and optionally starts it. +// Returns a duplicate_bot error if the bot identity is already used by another channel. func (s *Service) CreateChannel(channel *IMChannel) error { + if err := s.checkDuplicateBot(channel, ""); err != nil { + return err + } if err := s.db.Create(channel).Error; err != nil { return err } @@ -757,7 +903,11 @@ func (s *Service) CreateChannel(channel *IMChannel) error { } // UpdateChannel updates a channel and restarts it if needed. +// Returns a duplicate_bot error if the bot identity is already used by another channel. func (s *Service) UpdateChannel(channel *IMChannel) error { + if err := s.checkDuplicateBot(channel, channel.ID); err != nil { + return err + } if err := s.db.Save(channel).Error; err != nil { return err } @@ -771,16 +921,23 @@ func (s *Service) UpdateChannel(channel *IMChannel) error { return nil } -// DeleteChannel soft-deletes a channel and stops it. -func (s *Service) DeleteChannel(channelID string) error { +// DeleteChannel soft-deletes a channel and stops it. Only deletes if the channel belongs to the given tenant. +func (s *Service) DeleteChannel(channelID string, tenantID uint64) error { s.StopChannel(channelID) - return s.db.Where("id = ?", channelID).Delete(&IMChannel{}).Error + result := s.db.Where("id = ? AND tenant_id = ?", channelID, tenantID).Delete(&IMChannel{}) + if result.Error != nil { + return result.Error + } + if result.RowsAffected == 0 { + return fmt.Errorf("channel not found") + } + return nil } -// ToggleChannel enables or disables a channel. -func (s *Service) ToggleChannel(channelID string) (*IMChannel, error) { +// ToggleChannel enables or disables a channel. Only toggles if the channel belongs to the given tenant. +func (s *Service) ToggleChannel(channelID string, tenantID uint64) (*IMChannel, error) { var ch IMChannel - if err := s.db.Where("id = ? AND deleted_at IS NULL", channelID).First(&ch).Error; err != nil { + if err := s.db.Where("id = ? AND tenant_id = ? AND deleted_at IS NULL", channelID, tenantID).First(&ch).Error; err != nil { return nil, err } ch.Enabled = !ch.Enabled @@ -796,3 +953,524 @@ func (s *Service) ToggleChannel(channelID string) (*IMChannel, error) { } return &ch, nil } + +// checkDuplicateBot queries the bot_identity index to see if another active channel +// already uses the same bot. This is an O(1) index lookup, not a full table scan. +// The DB unique index on bot_identity serves as an additional safety net. +// excludeID is the channel's own ID (for updates); pass "" for new channels. +func (s *Service) checkDuplicateBot(channel *IMChannel, excludeID string) error { + // Compute bot_identity the same way the BeforeSave hook will + botKey := channel.computeBotIdentity() + if botKey == "" { + return nil + } + + var existing IMChannel + query := s.db.Where("bot_identity = ? AND deleted_at IS NULL", botKey) + if excludeID != "" { + query = query.Where("id != ?", excludeID) + } + if err := query.First(&existing).Error; err != nil { + if err == gorm.ErrRecordNotFound { + return nil // no conflict + } + return fmt.Errorf("check duplicate bot: %w", err) + } + return fmt.Errorf("duplicate_bot: this bot is already bound to channel %q (%s); each bot can only be connected to one channel", existing.Name, existing.ID) +} + +// ── File message handling ────────────────────────────────────────────── +// These methods handle file messages received via IM platforms. +// Files are downloaded from the IM platform, validated, and saved to the +// configured knowledge base asynchronously. The user receives a notification +// at the start and end of processing. +// ──────────────────────────────────────────────────────────────────────── + +// supportedKBFileExts is the set of file extensions that can be saved to a knowledge base. +var supportedKBFileExts = map[string]bool{ + "pdf": true, "txt": true, "docx": true, "doc": true, + "md": true, "markdown": true, + "png": true, "jpg": true, "jpeg": true, "gif": true, + "csv": true, "xlsx": true, "xls": true, + "pptx": true, "ppt": true, +} + +// handleFileMessage processes a file message by downloading it from the IM platform +// and saving it to the channel's configured knowledge base. Sends start/end +// notifications to the user via the adapter. +func (s *Service) handleFileMessage(ctx context.Context, msg *IncomingMessage, adapter Adapter, channel *IMChannel) error { + // Check if the adapter supports file downloading + downloader, ok := adapter.(FileDownloader) + if !ok { + logger.Infof(ctx, "[IM] Adapter for platform %s does not support file download, ignoring file message", msg.Platform) + return s.sendSmartReply(ctx, adapter, msg, channel, + "用户尝试发送文件，但当前平台暂不支持文件消息处理。", + "❌ 当前平台暂不支持文件消息处理。") + } + + // For image messages, ensure a proper file extension is present. + // IM platforms may only provide a hash/key as filename without extension. + if msg.MessageType == MessageTypeImage && fileExtension(msg.FileName) == "" { + msg.FileName = msg.FileName + ".png" + } + + // Validate file extension (pre-download). + // Some platforms (e.g. WeCom aibot) do not provide original filenames in the + // callback JSON — only a hash ID. For such cases we defer extension validation + // to after the file is downloaded, where the real name may be obtained from + // HTTP Content-Disposition or Content-Type headers. + ext := fileExtension(msg.FileName) + if ext != "" && !supportedKBFileExts[ext] { + logger.Infof(ctx, "[IM] Unsupported file type: %s (file=%s)", ext, msg.FileName) + return s.sendSmartReply(ctx, adapter, msg, channel, + fmt.Sprintf("用户上传了一个不支持的文件类型「%s」。目前支持的类型包括：PDF、Word、TXT、Markdown、Excel、CSV、PPT、图片。", ext), + fmt.Sprintf("❌ 不支持的文件类型「%s」。\n\n支持的类型：PDF、Word、TXT、Markdown、Excel、CSV、PPT、图片。", ext)) + } + + displayName := msg.FileName + if ext == "" { + displayName = "文件" + } + + // Send "processing started" notification (streaming) + if err := s.sendSmartReply(ctx, adapter, msg, channel, + fmt.Sprintf("用户发送了一个文件「%s」，系统正在处理并保存到知识库中，需要告知用户请稍候。", displayName), + fmt.Sprintf("📥 已收到%s，正在处理并保存到知识库，请稍候...", displayName)); err != nil { + logger.Warnf(ctx, "[IM] Failed to send file processing start notification: %v", err) + } + + // Process asynchronously to avoid blocking the message handler + go s.processFileToKnowledgeBase(context.WithoutCancel(ctx), msg, downloader, adapter, channel) + + return nil +} + +// processFileToKnowledgeBase is the async worker that downloads a file from the +// IM platform and creates a knowledge entry in the configured knowledge base. +func (s *Service) processFileToKnowledgeBase(ctx context.Context, msg *IncomingMessage, downloader FileDownloader, adapter Adapter, channel *IMChannel) { + kbID := channel.KnowledgeBaseID + tenantID := channel.TenantID + + // Build context with tenant info for the knowledge service + tenant, err := s.tenantService.GetTenantByID(ctx, tenantID) + if err != nil { + logger.Errorf(ctx, "[IM] Failed to get tenant %d for file processing: %v", tenantID, err) + s.sendFileResult(ctx, adapter, msg, msg.FileName, false, "获取租户信息失败", channel) + return + } + kbCtx := context.WithValue(ctx, types.TenantIDContextKey, tenantID) + kbCtx = context.WithValue(kbCtx, types.TenantInfoContextKey, tenant) + + // Download file from IM platform + reader, fileName, err := downloader.DownloadFile(ctx, msg) + if err != nil { + logger.Errorf(ctx, "[IM] Failed to download file from %s: %v", msg.Platform, err) + s.sendFileResult(ctx, adapter, msg, msg.FileName, false, "下载文件失败", channel) + return + } + defer reader.Close() + + logger.Debugf(ctx, "[IM] Downloaded file: original_name=%s resolved_name=%s", msg.FileName, fileName) + + // Post-download extension validation: if the pre-download name had no extension + // (e.g. WeCom file messages only provide a hash), check the resolved name now. + ext := fileExtension(fileName) + if !supportedKBFileExts[ext] { + logger.Infof(ctx, "[IM] Unsupported file type after download: %s (file=%s)", ext, fileName) + s.sendFileResult(ctx, adapter, msg, fileName, false, + fmt.Sprintf("不支持的文件类型「%s」。支持：PDF、Word、TXT、Markdown、Excel、CSV、PPT、图片", ext), channel) + return + } + + // Read file content into memory for multipart upload + content, err := io.ReadAll(reader) + if err != nil { + logger.Errorf(ctx, "[IM] Failed to read file content: %v", err) + s.sendFileResult(ctx, adapter, msg, fileName, false, "读取文件内容失败", channel) + return + } + + // Create a multipart.FileHeader compatible wrapper + fh := newInMemoryFileHeader(fileName, content) + + // Create knowledge entry via the knowledge service + knowledge, err := s.knowledgeService.CreateKnowledgeFromFile(kbCtx, kbID, fh, nil, nil, "", "") + if err != nil { + errMsg := err.Error() + // Check for duplicate file + if strings.Contains(errMsg, "duplicate") || strings.Contains(errMsg, "already exists") { + logger.Infof(ctx, "[IM] File already exists in knowledge base: %s", fileName) + s.sendFileResult(ctx, adapter, msg, fileName, false, "文件已存在于知识库中", channel) + return + } + logger.Errorf(ctx, "[IM] Failed to create knowledge from file: %v", err) + s.sendFileResult(ctx, adapter, msg, fileName, false, "保存到知识库失败", channel) + return + } + + logger.Infof(ctx, "[IM] File saved to knowledge base: kb=%s knowledge=%s file=%s", kbID, knowledge.ID, fileName) + s.sendFileResult(ctx, adapter, msg, fileName, true, "", channel) + + // Start a background watcher to send the document summary once Asynq + // finishes parsing + summary generation. This is intentionally decoupled + // from the Asynq task pipeline to avoid modifying any existing logic. + go s.watchAndSendSummary(ctx, kbCtx, adapter, msg, knowledge.ID, fileName, channel) +} + +// sendFileResult sends a notification about the file processing result. +// It uses sendSmartReply to generate a friendly, streaming reply via the channel's LLM. +// Falls back to a static template if the LLM is unavailable. +func (s *Service) sendFileResult(ctx context.Context, adapter Adapter, msg *IncomingMessage, fileName string, success bool, errDetail string, channel *IMChannel) { + var fallback string + if success { + fallback = fmt.Sprintf("✅ 文件「%s」已保存到知识库，正在解析中，完成后会通知你～", fileName) + } else { + fallback = fmt.Sprintf("❌ 文件「%s」处理失败：%s", fileName, errDetail) + } + + var situation string + if success { + situation = fmt.Sprintf("用户上传的文件「%s」已成功保存到知识库，但还需要后台解析文档内容（这需要一些时间）。请告知用户文件已收到，正在解析处理中，解析完成后会自动推送结果。", fileName) + } else { + situation = fmt.Sprintf("用户上传的文件「%s」处理失败，原因：%s。", fileName, errDetail) + } + + if err := s.sendSmartReply(ctx, adapter, msg, channel, situation, fallback); err != nil { + logger.Warnf(ctx, "[IM] Failed to send file result notification: %v", err) + } +} + +// smartReplySystemPrompt is the system prompt used for generating smart notification replies. +const smartReplySystemPrompt = "你是一个专业的 IM 机器人助手。请根据以下事件情况，生成一条简洁、清晰的通知消息。" + + "要求：1) 可适当使用 emoji 但不要过多；2) 语气专业平等，像同事之间对话，不要谄媚讨好，不要用「啦」「哦」「呢」「哟」等撒娇语气词；" + + "3) 直接输出消息内容，不要加任何额外解释；" + + "4) 如果事件中包含摘要或详细内容，请用 Markdown 格式结构化展示（使用标题、列表、加粗等），完整呈现，不要删减或概括；如果是简单通知，则控制在 2-3 句话以内。" + +// sendSmartReply generates a notification message using the channel's LLM and sends it +// to the user. If the adapter supports streaming (StreamSender), it streams the reply +// in real-time for a better user experience. Otherwise, it falls back to non-streaming. +// If the LLM is unavailable or fails, it sends the provided fallback text. +func (s *Service) sendSmartReply(ctx context.Context, adapter Adapter, msg *IncomingMessage, channel *IMChannel, situation string, fallback string) error { + chatModel := s.getChatModelForChannel(ctx, channel) + if chatModel == nil { + return adapter.SendReply(ctx, msg, &ReplyMessage{Content: fallback, IsFinal: true}) + } + + // If the adapter supports streaming, use stream mode + if streamer, ok := adapter.(StreamSender); ok { + if err := s.streamSmartReply(ctx, chatModel, streamer, msg, situation); err == nil { + return nil + } + // Stream failed — fall through to non-streaming + logger.Warnf(ctx, "[IM] Stream smart reply failed, falling back to non-streaming") + } + + // Non-streaming fallback + content := s.generateSmartReply(ctx, chatModel, situation, fallback) + return adapter.SendReply(ctx, msg, &ReplyMessage{Content: content, IsFinal: true}) +} + +// streamSmartReply uses ChatStream to generate and stream a notification reply in real-time. +func (s *Service) streamSmartReply(ctx context.Context, chatModel chat.Chat, streamer StreamSender, msg *IncomingMessage, situation string) error { + messages := []chat.Message{ + {Role: "system", Content: smartReplySystemPrompt}, + {Role: "user", Content: situation}, + } + + timeoutCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + + streamCh, err := chatModel.ChatStream(timeoutCtx, messages, &chat.ChatOptions{ + Temperature: 0.7, + MaxTokens: 800, + }) + if err != nil { + logger.Warnf(ctx, "[IM] ChatStream failed for smart reply: %v", err) + return err + } + + // Start the stream on the IM platform + streamID, err := streamer.StartStream(ctx, msg) + if err != nil { + logger.Warnf(ctx, "[IM] StartStream failed for smart reply: %v", err) + return err + } + + // Flush loop with batching (same pattern as handleMessageStream) + var ( + bufMu sync.Mutex + buf strings.Builder + done = make(chan struct{}) + ) + + go func() { + defer close(done) + for resp := range streamCh { + if resp.Content != "" { + bufMu.Lock() + buf.WriteString(resp.Content) + bufMu.Unlock() + } + } + }() + + ticker := time.NewTicker(streamFlushInterval) + defer ticker.Stop() + + flush := func() { + bufMu.Lock() + chunk := buf.String() + buf.Reset() + bufMu.Unlock() + + if chunk != "" { + if err := streamer.SendStreamChunk(ctx, msg, streamID, chunk); err != nil { + logger.Warnf(ctx, "[IM] SendStreamChunk failed for smart reply: %v", err) + } + } + } + +loop: + for { + select { + case <-ticker.C: + flush() + case <-done: + break loop + case <-timeoutCtx.Done(): + break loop + } + } + + // Final flush + flush() + + // End the stream + if err := streamer.EndStream(ctx, msg, streamID); err != nil { + logger.Warnf(ctx, "[IM] EndStream failed for smart reply: %v", err) + } + + return nil +} + +// generateSmartReply uses the channel's agent LLM to produce a natural-language +// notification message for the given situation (non-streaming). +// If the call fails, it returns the provided fallback text. +func (s *Service) generateSmartReply(ctx context.Context, chatModel chat.Chat, situation string, fallback string) string { + messages := []chat.Message{ + {Role: "system", Content: smartReplySystemPrompt}, + {Role: "user", Content: situation}, + } + + timeoutCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + + resp, err := chatModel.Chat(timeoutCtx, messages, &chat.ChatOptions{ + Temperature: 0.7, + MaxTokens: 800, + }) + if err != nil { + logger.Warnf(ctx, "[IM] Smart reply generation failed, using fallback: %v", err) + return fallback + } + + reply := strings.TrimSpace(resp.Content) + if reply == "" { + return fallback + } + return reply +} + +// getChatModelForChannel resolves the chat.Chat instance configured on the +// channel's agent. Returns nil if the model cannot be resolved. +func (s *Service) getChatModelForChannel(ctx context.Context, channel *IMChannel) chat.Chat { + if channel == nil || channel.AgentID == "" { + return nil + } + + // Ensure the context carries tenant ID — some call sites (e.g. handleFileMessage) + // may invoke this before the tenant has been injected into ctx. + if _, ok := types.TenantIDFromContext(ctx); !ok && channel.TenantID != 0 { + ctx = context.WithValue(ctx, types.TenantIDContextKey, channel.TenantID) + } + + agent, err := s.agentService.GetAgentByID(ctx, channel.AgentID) + if err != nil || agent == nil { + logger.Debugf(ctx, "[IM] Cannot get agent %s for smart reply: %v", channel.AgentID, err) + return nil + } + + modelID := agent.Config.ModelID + if modelID == "" { + return nil + } + + chatModel, err := s.modelService.GetChatModel(ctx, modelID) + if err != nil { + logger.Debugf(ctx, "[IM] Cannot get chat model %s for smart reply: %v", modelID, err) + return nil + } + return chatModel +} + +// watchAndSendSummary polls the knowledge record until document parsing (and +// optionally summary generation) completes, then sends the result back to the +// IM user. This runs as a fire-and-forget goroutine, completely decoupled from +// the Asynq worker pipeline. +func (s *Service) watchAndSendSummary( + ctx context.Context, + kbCtx context.Context, + adapter Adapter, + msg *IncomingMessage, + knowledgeID string, + fileName string, + channel *IMChannel, +) { + const ( + pollInterval = 5 * time.Second + maxWait = 10 * time.Minute // give up after 10 minutes + ) + + deadline := time.Now().Add(maxWait) + ticker := time.NewTicker(pollInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if time.Now().After(deadline) { + logger.Infof(ctx, "[IM] Summary watcher timed out for knowledge %s", knowledgeID) + return + } + + knowledge, err := s.knowledgeService.GetKnowledgeByID(kbCtx, knowledgeID) + if err != nil { + logger.Warnf(ctx, "[IM] Summary watcher: failed to get knowledge %s: %v", knowledgeID, err) + return + } + + switch knowledge.ParseStatus { + case types.ParseStatusFailed: + // Parsing failed — notify user and stop watching + errMsg := knowledge.ErrorMessage + if errMsg == "" { + errMsg = "文档解析失败" + } + _ = s.sendSmartReply(ctx, adapter, msg, channel, + fmt.Sprintf("用户之前上传的文件「%s」解析失败了，错误原因：%s。请安慰用户并建议重试。", fileName, errMsg), + fmt.Sprintf("⚠️ 文件「%s」解析失败：%s", fileName, errMsg)) + return + + case types.ParseStatusCompleted: + // Parsing done. If summary generation is in progress, wait for it. + switch knowledge.SummaryStatus { + case types.SummaryStatusNone, "": + // No summary task configured. For image files the VLM caption + // is stored in Description by finalizeImageKnowledge, so we + // still show it if present. + if knowledge.Description != "" && knowledge.Description != fileName { + _ = s.sendSmartReply(ctx, adapter, msg, channel, + fmt.Sprintf("用户之前上传的文件「%s」已解析完成。以下是文件的完整摘要内容：\n%s\n\n请生成一条通知消息，包含：1) 告知文件已解析完成；2) 用 Markdown 格式（标题、列表、加粗等）结构化展示上述摘要内容，不要删减或概括；3) 提示用户可以针对该文件提问。", fileName, knowledge.Description), + fmt.Sprintf("📄 文件「%s」已解析完成。\n\n**摘要：**\n\n%s\n\n---\n可以针对该文件进行提问。", fileName, knowledge.Description)) + } else { + _ = s.sendSmartReply(ctx, adapter, msg, channel, + fmt.Sprintf("用户之前上传的文件「%s」已解析完成，现在可以开始针对该文件进行提问了。", fileName), + fmt.Sprintf("📄 文件「%s」已解析完成，可以开始提问了！", fileName)) + } + return + + case types.SummaryStatusCompleted: + // Summary is ready — send it + s.sendSummaryNotification(ctx, adapter, msg, knowledge, fileName, channel) + return + + case types.SummaryStatusFailed: + _ = s.sendSmartReply(ctx, adapter, msg, channel, + fmt.Sprintf("用户之前上传的文件「%s」已解析完成，但摘要生成失败了。不过文件已可用于提问。", fileName), + fmt.Sprintf("📄 文件「%s」已解析完成，可以开始提问了！（摘要生成失败）", fileName)) + return + + default: + // Still generating summary — keep polling + } + + default: + // Still parsing — keep polling + } + } + } +} + +// sendSummaryNotification retrieves the summary chunk for a knowledge entry +// and sends it as a message to the IM user. +func (s *Service) sendSummaryNotification( + ctx context.Context, + adapter Adapter, + msg *IncomingMessage, + knowledge *types.Knowledge, + fileName string, + channel *IMChannel, +) { + // The summary is stored in the knowledge's Description field or as a + // ChunkTypeSummary chunk. We use Description first (populated by the + // summary generation task), falling back to a generic notice. + summary := knowledge.Description + if summary == "" { + summary = knowledge.Title + } + + var situation, fallback string + if summary != "" && summary != fileName { + situation = fmt.Sprintf("用户之前上传的文件「%s」已解析完成。以下是文件的完整摘要内容：\n%s\n\n请生成一条通知消息，包含：1) 告知文件已解析完成；2) 用 Markdown 格式（标题、列表、加粗等）结构化展示上述摘要内容，不要删减或概括；3) 提示用户可以针对该文件提问。", fileName, summary) + fallback = fmt.Sprintf("📄 文件「%s」已解析完成。\n\n**摘要：**\n\n%s\n\n---\n可以针对该文件进行提问。", fileName, summary) + } else { + situation = fmt.Sprintf("用户之前上传的文件「%s」已解析完成，现在可以开始针对该文件进行提问了。", fileName) + fallback = fmt.Sprintf("📄 文件「%s」已解析完成，可以开始提问了！", fileName) + } + + if err := s.sendSmartReply(ctx, adapter, msg, channel, situation, fallback); err != nil { + logger.Warnf(ctx, "[IM] Failed to send summary notification: %v", err) + } +} + +// fileExtension extracts the lowercase file extension from a filename. +func fileExtension(filename string) string { + parts := strings.Split(filename, ".") + if len(parts) < 2 { + return "" + } + return strings.ToLower(parts[len(parts)-1]) +} + +// newInMemoryFileHeader wraps in-memory file content as a *multipart.FileHeader +// so it can be passed to CreateKnowledgeFromFile which expects a multipart upload. +func newInMemoryFileHeader(filename string, data []byte) *multipart.FileHeader { + body := &bytes.Buffer{} + writer := multipart.NewWriter(body) + + h := make(textproto.MIMEHeader) + h.Set("Content-Disposition", fmt.Sprintf(`form-data; name="file"; filename="%s"`, filename)) + h.Set("Content-Type", "application/octet-stream") + + part, err := writer.CreatePart(h) + if err != nil { + // Fallback: return a minimal FileHeader + return &multipart.FileHeader{Filename: filename, Size: int64(len(data))} + } + _, _ = part.Write(data) + _ = writer.Close() + + // Parse the multipart body to extract the FileHeader + reader := multipart.NewReader(body, writer.Boundary()) + form, err := reader.ReadForm(int64(len(data)) + 1024) + if err != nil || form == nil { + return &multipart.FileHeader{Filename: filename, Size: int64(len(data))} + } + files := form.File["file"] + if len(files) == 0 { + return &multipart.FileHeader{Filename: filename, Size: int64(len(data))} + } + return files[0] +} diff --git a/internal/im/types.go b/internal/im/types.go index 4a6804da..d1abe413 100644 --- a/internal/im/types.go +++ b/internal/im/types.go @@ -1,6 +1,8 @@ package im import ( + "encoding/json" + "fmt" "time" "github.com/Tencent/WeKnora/internal/types" @@ -18,8 +20,10 @@ type IMChannel struct { Name string `json:"name" gorm:"type:varchar(255);not null;default:''"` Enabled bool `json:"enabled" gorm:"not null;default:true"` Mode string `json:"mode" gorm:"type:varchar(20);not null;default:'websocket'"` - OutputMode string `json:"output_mode" gorm:"type:varchar(20);not null;default:'stream'"` - Credentials types.JSON `json:"credentials" gorm:"type:jsonb;not null;default:'{}'"` + OutputMode string `json:"output_mode" gorm:"type:varchar(20);not null;default:'stream'"` + KnowledgeBaseID string `json:"knowledge_base_id" gorm:"type:varchar(36);default:''"` + BotIdentity string `json:"bot_identity" gorm:"type:varchar(255);not null;default:'';uniqueIndex:idx_im_channels_bot_identity,where:deleted_at IS NULL AND bot_identity != ''"` + Credentials types.JSON `json:"credentials" gorm:"type:jsonb;not null;default:'{}'"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` DeletedAt gorm.DeletedAt `json:"deleted_at" gorm:"index"` @@ -39,9 +43,58 @@ func (ch *IMChannel) BeforeCreate(tx *gorm.DB) error { if ch.OutputMode == "" { ch.OutputMode = "stream" } + ch.BotIdentity = ch.computeBotIdentity() return nil } +// BeforeSave ensures bot_identity is recomputed on every save (create + update). +func (ch *IMChannel) BeforeSave(tx *gorm.DB) error { + ch.BotIdentity = ch.computeBotIdentity() + return nil +} + +// computeBotIdentity derives a unique bot identity string from the channel's +// platform, mode, and credentials. Returns "" if no identity can be extracted. +func (ch *IMChannel) computeBotIdentity() string { + creds := make(map[string]interface{}) + if err := json.Unmarshal([]byte(ch.Credentials), &creds); err != nil { + return "" + } + + str := func(key string) string { + if v, ok := creds[key]; ok { + switch val := v.(type) { + case string: + return val + case float64: + return fmt.Sprintf("%.0f", val) + } + } + return "" + } + + switch ch.Platform { + case "wecom": + switch ch.Mode { + case "websocket": + if botID := str("bot_id"); botID != "" { + return "wecom:ws:" + botID + } + case "webhook": + corpID := str("corp_id") + agentID := str("corp_agent_id") + if corpID != "" && agentID != "" { + return "wecom:wh:" + corpID + ":" + agentID + } + } + case "feishu": + if appID := str("app_id"); appID != "" { + return "feishu:" + appID + } + } + return "" +} + // ChannelSession maps an IM channel (user+chat combination) to a WeKnora session. // This allows the IM integration to maintain conversation continuity. type ChannelSession struct { diff --git a/internal/im/wecom/longconn.go b/internal/im/wecom/longconn.go index b7bd35a9..64bfde50 100644 --- a/internal/im/wecom/longconn.go +++ b/internal/im/wecom/longconn.go @@ -51,12 +51,14 @@ type wsFrame struct { } // botMessage is the body of an aibot_msg_callback frame. +// Supports text, image, file, voice, and mixed message types. +// Reference: https://developer.work.weixin.qq.com/document/path/100719 type botMessage struct { MsgID string `json:"msgid"` AiBotID string `json:"aibotid"` ChatID string `json:"chatid"` ChatType string `json:"chattype"` // "single" or "group" - MsgType string `json:"msgtype"` // "text", "image", "event", ... + MsgType string `json:"msgtype"` // "text", "image", "file", "voice", "video", "mixed", "stream" CreateTime int64 `json:"create_time"` From struct { UserID string `json:"userid"` @@ -64,6 +66,37 @@ type botMessage struct { Text struct { Content string `json:"content"` } `json:"text"` + Image struct { + URL string `json:"url"` // encrypted download URL, valid for 5 minutes + AESKey string `json:"aeskey"` // per-message AES key for decrypting downloaded content + } `json:"image"` + File struct { + URL string `json:"url"` // encrypted download URL, valid for 5 minutes + AESKey string `json:"aeskey"` // per-message AES key for decrypting downloaded content + } `json:"file"` + Voice struct { + Content string `json:"content"` // speech-to-text result + } `json:"voice"` + Video struct { + URL string `json:"url"` // encrypted download URL, valid for 5 minutes + AESKey string `json:"aeskey"` // per-message AES key for decrypting downloaded content + } `json:"video"` + Mixed struct { + MsgItem []botMixedItem `json:"msg_item"` + } `json:"mixed"` + Quote *botMessage `json:"quote,omitempty"` // quoted message (optional) +} + +// botMixedItem is one element in a mixed (text+image) message. +type botMixedItem struct { + MsgType string `json:"msgtype"` // "text" or "image" + Text struct { + Content string `json:"content"` + } `json:"text"` + Image struct { + URL string `json:"url"` + AESKey string `json:"aeskey"` + } `json:"image"` } // streamReplyBody is the body for a streaming text reply. @@ -380,17 +413,18 @@ func (c *LongConnClient) heartbeatLoop(ctx context.Context) { } func (c *LongConnClient) handleCallback(ctx context.Context, frame wsFrame) { + // Log raw message body for debugging + logger.Debugf(ctx, "[WeCom] Raw callback body: %s", string(frame.Body)) + var msg botMessage if err := json.Unmarshal(frame.Body, &msg); err != nil { logger.Warnf(ctx, "[WeCom] Failed to unmarshal callback body: %v", err) return } - // Only handle text messages for now - if msg.MsgType != "text" { - logger.Infof(ctx, "[WeCom] Ignoring non-text message type: %s", msg.MsgType) - return - } + logger.Debugf(ctx, "[WeCom] Parsed message: msgid=%s msgtype=%s from=%s chattype=%s text=%q image_url=%q file_url=%q voice=%q mixed_items=%d", + msg.MsgID, msg.MsgType, msg.From.UserID, msg.ChatType, + msg.Text.Content, msg.Image.URL, msg.File.URL, msg.Voice.Content, len(msg.Mixed.MsgItem)) chatType := im.ChatTypeDirect chatID := "" @@ -405,15 +439,87 @@ func (c *LongConnClient) handleCallback(ctx context.Context, frame wsFrame) { reqID = frame.Headers["req_id"] } - incoming := &im.IncomingMessage{ - Platform: im.PlatformWeCom, - UserID: msg.From.UserID, - UserName: msg.From.UserID, - ChatID: chatID, - ChatType: chatType, - Content: strings.TrimSpace(msg.Text.Content), - MessageID: msg.MsgID, - Extra: map[string]string{"req_id": reqID}, + var incoming *im.IncomingMessage + + switch msg.MsgType { + case "text": + incoming = &im.IncomingMessage{ + Platform: im.PlatformWeCom, + MessageType: im.MessageTypeText, + UserID: msg.From.UserID, + UserName: msg.From.UserID, + ChatID: chatID, + ChatType: chatType, + Content: strings.TrimSpace(msg.Text.Content), + MessageID: msg.MsgID, + Extra: map[string]string{"req_id": reqID}, + } + + case "voice": + // WeCom returns speech-to-text content directly — treat as text query + if msg.Voice.Content == "" { + logger.Infof(ctx, "[WeCom] Ignoring voice message with empty content") + return + } + incoming = &im.IncomingMessage{ + Platform: im.PlatformWeCom, + MessageType: im.MessageTypeText, + UserID: msg.From.UserID, + UserName: msg.From.UserID, + ChatID: chatID, + ChatType: chatType, + Content: strings.TrimSpace(msg.Voice.Content), + MessageID: msg.MsgID, + Extra: map[string]string{"req_id": reqID}, + } + + case "image": + if msg.Image.URL == "" { + logger.Infof(ctx, "[WeCom] Ignoring image message with empty URL") + return + } + incoming = &im.IncomingMessage{ + Platform: im.PlatformWeCom, + MessageType: im.MessageTypeImage, + UserID: msg.From.UserID, + UserName: msg.From.UserID, + ChatID: chatID, + ChatType: chatType, + MessageID: msg.MsgID, + FileKey: msg.Image.URL, // store encrypted URL in FileKey + FileName: msg.MsgID + ".png", + Extra: map[string]string{"req_id": reqID, "aes_key": msg.Image.AESKey}, + } + + case "file": + if msg.File.URL == "" { + logger.Infof(ctx, "[WeCom] Ignoring file message with empty URL") + return + } + incoming = &im.IncomingMessage{ + Platform: im.PlatformWeCom, + MessageType: im.MessageTypeFile, + UserID: msg.From.UserID, + UserName: msg.From.UserID, + ChatID: chatID, + ChatType: chatType, + MessageID: msg.MsgID, + FileKey: msg.File.URL, // store encrypted URL in FileKey + FileName: msg.MsgID, // WeCom doesn't provide file name directly + Extra: map[string]string{"req_id": reqID, "aes_key": msg.File.AESKey}, + } + + case "mixed": + // Extract text parts for QA content, and detect if any images are present + incoming = convertMixedMessage(&msg, chatID, chatType, reqID) + if incoming == nil { + logger.Infof(ctx, "[WeCom] Ignoring empty mixed message") + return + } + + default: + logger.Infof(ctx, "[WeCom] Ignoring unsupported message type: %s", msg.MsgType) + return } if err := c.handler(ctx, incoming); err != nil { @@ -421,6 +527,61 @@ func (c *LongConnClient) handleCallback(ctx context.Context, frame wsFrame) { } } +// convertMixedMessage converts a WeCom mixed (text+image) message. +// Extracts all text content for QA; if there's only images, treat as image message. +func convertMixedMessage(msg *botMessage, chatID string, chatType im.ChatType, reqID string) *im.IncomingMessage { + var textParts []string + var firstImageURL string + var firstImageAESKey string + + for _, item := range msg.Mixed.MsgItem { + switch item.MsgType { + case "text": + if t := strings.TrimSpace(item.Text.Content); t != "" { + textParts = append(textParts, t) + } + case "image": + if firstImageURL == "" && item.Image.URL != "" { + firstImageURL = item.Image.URL + firstImageAESKey = item.Image.AESKey + } + } + } + + // If there's text content, treat as text message (QA query) + if len(textParts) > 0 { + return &im.IncomingMessage{ + Platform: im.PlatformWeCom, + MessageType: im.MessageTypeText, + UserID: msg.From.UserID, + UserName: msg.From.UserID, + ChatID: chatID, + ChatType: chatType, + Content: strings.Join(textParts, "\n"), + MessageID: msg.MsgID, + Extra: map[string]string{"req_id": reqID}, + } + } + + // Only images, treat as image message (save to KB) + if firstImageURL != "" { + return &im.IncomingMessage{ + Platform: im.PlatformWeCom, + MessageType: im.MessageTypeImage, + UserID: msg.From.UserID, + UserName: msg.From.UserID, + ChatID: chatID, + ChatType: chatType, + MessageID: msg.MsgID, + FileKey: firstImageURL, + FileName: msg.MsgID + ".png", + Extra: map[string]string{"req_id": reqID, "aes_key": firstImageAESKey}, + } + } + + return nil +} + func (c *LongConnClient) writeJSON(v interface{}) error { c.mu.Lock() defer c.mu.Unlock() diff --git a/internal/im/wecom/webhook_adapter.go b/internal/im/wecom/webhook_adapter.go index e0f4aaf1..1472d5cf 100644 --- a/internal/im/wecom/webhook_adapter.go +++ b/internal/im/wecom/webhook_adapter.go @@ -22,7 +22,10 @@ import ( "encoding/xml" "fmt" "io" + "mime" "net/http" + "net/url" + "path" "sort" "strings" "sync" @@ -33,7 +36,7 @@ import ( "github.com/gin-gonic/gin" ) -var httpClient = &http.Client{Timeout: 10 * time.Second} +var httpClient = &http.Client{Timeout: 30 * time.Second} // WebhookAdapter implements im.Adapter for WeCom in webhook (self-built app callback) mode. // Messages arrive via HTTP callback; replies are sent via the WeCom REST API. @@ -51,6 +54,9 @@ type WebhookAdapter struct { tokenExpAt time.Time } +// Compile-time check that WebhookAdapter implements im.FileDownloader. +var _ im.FileDownloader = (*WebhookAdapter)(nil) + // NewWebhookAdapter creates a new WeCom webhook adapter. func NewWebhookAdapter(corpID, agentSecret, token, encodingAESKey string, corpAgentID int) (*WebhookAdapter, error) { // Decode the AES key from base64 @@ -146,37 +152,63 @@ func (a *WebhookAdapter) ParseCallback(c *gin.Context) (*im.IncomingMessage, err return nil, fmt.Errorf("decrypt message: %w", err) } + // Log raw decrypted message for debugging + logger.Debugf(c.Request.Context(), "[WeCom] Raw decrypted callback: %s", string(decrypted)) + var msg wecomMessage if err := xml.Unmarshal(decrypted, &msg); err != nil { return nil, fmt.Errorf("unmarshal decrypted message: %w", err) } - // Only handle text messages - if msg.MsgType != "text" { - logger.Infof(c.Request.Context(), "[WeCom] Ignoring non-text message type: %s", msg.MsgType) - return nil, nil - } + logger.Debugf(c.Request.Context(), "[WeCom] Parsed webhook message: msgid=%s msgtype=%s from=%s content=%q picurl=%q mediaid=%q", + msg.MsgID, msg.MsgType, msg.FromUserName, msg.Content, msg.PicUrl, msg.MediaId) + // Determine chat type chatType := im.ChatTypeDirect chatID := "" - content := msg.Content - - // Check if this is a group message (has ChatId field in WeCom smart bot callback) - // In group chats, the content may contain @bot mention prefix that should be stripped if msg.ChatID != "" { chatType = im.ChatTypeGroup chatID = msg.ChatID } - return &im.IncomingMessage{ - Platform: im.PlatformWeCom, - UserID: msg.FromUserName, - UserName: msg.FromUserName, - ChatID: chatID, - ChatType: chatType, - Content: strings.TrimSpace(content), - MessageID: msg.MsgID, - }, nil + switch msg.MsgType { + case "text": + return &im.IncomingMessage{ + Platform: im.PlatformWeCom, + MessageType: im.MessageTypeText, + UserID: msg.FromUserName, + UserName: msg.FromUserName, + ChatID: chatID, + ChatType: chatType, + Content: strings.TrimSpace(msg.Content), + MessageID: msg.MsgID, + }, nil + + case "image": + // Image via webhook: has PicUrl (direct download) and MediaId + if msg.PicUrl == "" && msg.MediaId == "" { + return nil, nil + } + fileKey := msg.PicUrl + if fileKey == "" { + fileKey = msg.MediaId + } + return &im.IncomingMessage{ + Platform: im.PlatformWeCom, + MessageType: im.MessageTypeImage, + UserID: msg.FromUserName, + UserName: msg.FromUserName, + ChatID: chatID, + ChatType: chatType, + MessageID: msg.MsgID, + FileKey: fileKey, + FileName: msg.MsgID + ".png", + }, nil + + default: + logger.Infof(c.Request.Context(), "[WeCom] Ignoring unsupported message type: %s", msg.MsgType) + return nil, nil + } } // SendReply sends a reply message via WeCom API. @@ -357,14 +389,160 @@ type callbackRequestBody struct { } // wecomMessage is the decrypted WeCom message structure. +// Supports text, image, voice, video, location, and link message types. +// Reference: https://developer.work.weixin.qq.com/document/path/90375 type wecomMessage struct { XMLName xml.Name `xml:"xml"` ToUserName string `xml:"ToUserName"` FromUserName string `xml:"FromUserName"` CreateTime int64 `xml:"CreateTime"` MsgType string `xml:"MsgType"` - Content string `xml:"Content"` + Content string `xml:"Content"` // text + PicUrl string `xml:"PicUrl"` // image: download URL + MediaId string `xml:"MediaId"` // image/voice/video: media ID for download + Format string `xml:"Format"` // voice: audio format (amr/speex) + ThumbMediaId string `xml:"ThumbMediaId"` // video: thumbnail media ID MsgID string `xml:"MsgId"` AgentID string `xml:"AgentID"` ChatID string `xml:"ChatId"` } + +// ────────────────────────────────────────────────────────────────────── +// File download support for WeCom webhook mode +// ────────────────────────────────────────────────────────────────────── + +// DownloadFile downloads a file/image from WeCom. +// For webhook mode, images come with MediaId (temporary media) which can be +// downloaded via the GetMedia API, or PicUrl for direct download. +func (a *WebhookAdapter) DownloadFile(ctx context.Context, msg *im.IncomingMessage) (io.ReadCloser, string, error) { + if msg.FileKey == "" { + return nil, "", fmt.Errorf("no file key (URL or media_id) in message") + } + + fileName := msg.FileName + if fileName == "" { + fileName = msg.FileKey + } + + // If FileKey looks like a URL, download directly + if strings.HasPrefix(msg.FileKey, "http://") || strings.HasPrefix(msg.FileKey, "https://") { + return downloadFromURL(ctx, msg.FileKey, fileName) + } + + // Otherwise treat as media_id, download via temporary media API + accessToken, err := a.getAccessToken(ctx) + if err != nil { + return nil, "", fmt.Errorf("get access token: %w", err) + } + + apiURL := fmt.Sprintf("https://qyapi.weixin.qq.com/cgi-bin/media/get?access_token=%s&media_id=%s", + accessToken, msg.FileKey) + return downloadFromURL(ctx, apiURL, fileName) +} + +// downloadFromURL performs a GET request and returns the response body. +// It tries to resolve the real filename from HTTP response headers: +// 1. Content-Disposition: attachment; filename="xxx.pdf" +// 2. Content-Type → extension mapping (fallback for platforms like WeCom that +// don't provide the original filename in the callback JSON) +func downloadFromURL(ctx context.Context, rawURL, fileName string) (io.ReadCloser, string, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil) + if err != nil { + return nil, "", fmt.Errorf("create request: %w", err) + } + + resp, err := httpClient.Do(req) + if err != nil { + return nil, "", fmt.Errorf("download: %w", err) + } + + if resp.StatusCode != http.StatusOK { + resp.Body.Close() + return nil, "", fmt.Errorf("download failed: status=%d", resp.StatusCode) + } + + logger.Debugf(ctx, "[WeCom] Download response: status=%d content-type=%s content-disposition=%s", + resp.StatusCode, resp.Header.Get("Content-Type"), resp.Header.Get("Content-Disposition")) + + // Try to extract filename from Content-Disposition header. + // Supports both standard filename and RFC 5987 filename* parameters. + if cd := resp.Header.Get("Content-Disposition"); cd != "" { + if _, params, err := mime.ParseMediaType(cd); err == nil { + // Prefer filename* (RFC 5987, already decoded by mime.ParseMediaType) + if fn := params["filename"]; fn != "" { + fileName = fn + } + } else { + // Fallback: manual extraction for malformed headers + if idx := strings.Index(cd, "filename="); idx >= 0 { + extracted := strings.Trim(cd[idx+len("filename="):], "\" ") + if extracted != "" { + fileName = extracted + } + } + } + } + + // URL-decode the filename if it contains percent-encoded characters. + // Some servers (e.g. WeCom COS) return URL-encoded Chinese filenames. + if strings.Contains(fileName, "%") { + if decoded, err := url.QueryUnescape(fileName); err == nil && decoded != "" { + fileName = decoded + } + } + + // Also try to extract a meaningful filename from the URL path itself, + // in case Content-Disposition is missing but the URL contains the real name. + if !strings.Contains(fileName, ".") { + if u, err := url.Parse(rawURL); err == nil { + base := path.Base(u.Path) + if base != "" && base != "." && base != "/" && strings.Contains(base, ".") { + // URL-decode the path component as well + if decoded, err := url.QueryUnescape(base); err == nil { + fileName = decoded + } else { + fileName = base + } + } + } + } + + // If filename still has no extension, try to infer from Content-Type. + // This handles platforms (e.g. WeCom aibot) where the callback only provides + // a hash ID as the filename without any extension. + if !strings.Contains(fileName, ".") { + if ext := contentTypeToExt(resp.Header.Get("Content-Type")); ext != "" { + fileName = fileName + "." + ext + } + } + + return resp.Body, fileName, nil +} + +// contentTypeToExt maps common Content-Type values to file extensions. +func contentTypeToExt(ct string) string { + // Normalize: take only the media type, ignore parameters like charset + if idx := strings.Index(ct, ";"); idx >= 0 { + ct = strings.TrimSpace(ct[:idx]) + } + ct = strings.ToLower(ct) + + mapping := map[string]string{ + "application/pdf": "pdf", + "application/msword": "doc", + "application/vnd.openxmlformats-officedocument.wordprocessingml.document": "docx", + "application/vnd.ms-excel": "xls", + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": "xlsx", + "application/vnd.ms-powerpoint": "ppt", + "application/vnd.openxmlformats-officedocument.presentationml.presentation": "pptx", + "text/plain": "txt", + "text/markdown": "md", + "text/csv": "csv", + "image/png": "png", + "image/jpeg": "jpg", + "image/gif": "gif", + "image/webp": "webp", + } + + return mapping[ct] +} diff --git a/internal/im/wecom/ws_adapter.go b/internal/im/wecom/ws_adapter.go index 4040b9e6..4cf686e9 100644 --- a/internal/im/wecom/ws_adapter.go +++ b/internal/im/wecom/ws_adapter.go @@ -1,17 +1,24 @@ package wecom import ( + "bytes" "context" + "crypto/aes" + "crypto/cipher" + "encoding/base64" "fmt" + "io" "github.com/Tencent/WeKnora/internal/im" + "github.com/Tencent/WeKnora/internal/logger" "github.com/gin-gonic/gin" ) // Compile-time checks. var ( - _ im.Adapter = (*WSAdapter)(nil) - _ im.StreamSender = (*WSAdapter)(nil) + _ im.Adapter = (*WSAdapter)(nil) + _ im.StreamSender = (*WSAdapter)(nil) + _ im.FileDownloader = (*WSAdapter)(nil) ) // WSAdapter implements im.Adapter and im.StreamSender for WeCom in WebSocket @@ -60,3 +67,107 @@ func (a *WSAdapter) SendStreamChunk(ctx context.Context, incoming *im.IncomingMe func (a *WSAdapter) EndStream(ctx context.Context, incoming *im.IncomingMessage, streamID string) error { return a.client.EndStream(ctx, incoming, streamID) } + +// ── FileDownloader implementation ── +// WeCom aibot provides AES-256-CBC encrypted URLs for image/file/video messages. +// Each message carries its own aeskey for decryption. + +func (a *WSAdapter) DownloadFile(ctx context.Context, msg *im.IncomingMessage) (io.ReadCloser, string, error) { + if msg.FileKey == "" { + return nil, "", fmt.Errorf("no file URL in message") + } + fileName := msg.FileName + if fileName == "" { + fileName = msg.FileKey + } + + // Download the (encrypted) file content + reader, fileName, err := downloadFromURL(ctx, msg.FileKey, fileName) + if err != nil { + return nil, "", err + } + + // If an AES key is provided, the downloaded content is AES-256-CBC encrypted + // and must be decrypted before use. This is the case for WeCom aibot long + // connection mode where each file/image message carries a per-message aeskey. + aesKeyB64 := msg.Extra["aes_key"] + if aesKeyB64 == "" { + // No encryption — return raw content (e.g. webhook mode uses media API) + return reader, fileName, nil + } + + // Read all encrypted content + encryptedData, err := io.ReadAll(reader) + reader.Close() + if err != nil { + return nil, "", fmt.Errorf("read encrypted file: %w", err) + } + + logger.Debugf(ctx, "[WeCom] Decrypting file: name=%s encrypted_size=%d aes_key_len=%d", + fileName, len(encryptedData), len(aesKeyB64)) + + // Decrypt + decrypted, err := decryptAESCBC(encryptedData, aesKeyB64) + if err != nil { + return nil, "", fmt.Errorf("decrypt file: %w", err) + } + + logger.Debugf(ctx, "[WeCom] File decrypted: name=%s decrypted_size=%d", fileName, len(decrypted)) + + return io.NopCloser(bytes.NewReader(decrypted)), fileName, nil +} + +// decryptAESCBC decrypts data encrypted with AES-256-CBC using PKCS#7 padding. +// The aesKeyB64 is the base64-encoded AES key provided per-message by WeCom. +// IV is the first 16 bytes of the decoded AES key. +func decryptAESCBC(ciphertext []byte, aesKeyB64 string) ([]byte, error) { + // WeCom's per-message aeskey is base64-encoded (43 chars → 32 bytes after decode) + aesKey, err := base64.StdEncoding.DecodeString(aesKeyB64 + "=") + if err != nil { + // Try without padding + aesKey, err = base64.RawStdEncoding.DecodeString(aesKeyB64) + if err != nil { + return nil, fmt.Errorf("base64 decode aes key: %w", err) + } + } + + if len(aesKey) < 16 { + return nil, fmt.Errorf("aes key too short: %d bytes", len(aesKey)) + } + + block, err := aes.NewCipher(aesKey) + if err != nil { + return nil, fmt.Errorf("new aes cipher: %w", err) + } + + if len(ciphertext) < aes.BlockSize { + return nil, fmt.Errorf("ciphertext too short: %d bytes", len(ciphertext)) + } + if len(ciphertext)%aes.BlockSize != 0 { + return nil, fmt.Errorf("ciphertext not a multiple of block size: %d bytes", len(ciphertext)) + } + + // IV = first 16 bytes of the AES key + iv := aesKey[:aes.BlockSize] + mode := cipher.NewCBCDecrypter(block, iv) + plaintext := make([]byte, len(ciphertext)) + mode.CryptBlocks(plaintext, ciphertext) + + // Remove PKCS#7 padding + if len(plaintext) == 0 { + return nil, fmt.Errorf("empty plaintext after decryption") + } + padLen := int(plaintext[len(plaintext)-1]) + if padLen > aes.BlockSize || padLen == 0 || padLen > len(plaintext) { + // No valid PKCS#7 padding — return as-is (some implementations may not pad) + return plaintext, nil + } + // Verify padding bytes + for i := 0; i < padLen; i++ { + if plaintext[len(plaintext)-1-i] != byte(padLen) { + // Invalid padding — return as-is + return plaintext, nil + } + } + return plaintext[:len(plaintext)-padLen], nil +} diff --git a/migrations/versioned/000023_im_channel_kb_id.down.sql b/migrations/versioned/000023_im_channel_kb_id.down.sql new file mode 100644 index 00000000..f3a70a85 --- /dev/null +++ b/migrations/versioned/000023_im_channel_kb_id.down.sql @@ -0,0 +1 @@ +ALTER TABLE im_channels DROP COLUMN IF EXISTS knowledge_base_id; diff --git a/migrations/versioned/000023_im_channel_kb_id.up.sql b/migrations/versioned/000023_im_channel_kb_id.up.sql new file mode 100644 index 00000000..208306fb --- /dev/null +++ b/migrations/versioned/000023_im_channel_kb_id.up.sql @@ -0,0 +1,3 @@ +-- Add knowledge_base_id column to im_channels table. +-- When set, file messages received on this channel will be saved to the specified knowledge base. +ALTER TABLE im_channels ADD COLUMN IF NOT EXISTS knowledge_base_id VARCHAR(36) DEFAULT ''; diff --git a/migrations/versioned/000024_im_channel_bot_identity.down.sql b/migrations/versioned/000024_im_channel_bot_identity.down.sql new file mode 100644 index 00000000..e2588a46 --- /dev/null +++ b/migrations/versioned/000024_im_channel_bot_identity.down.sql @@ -0,0 +1,2 @@ +DROP INDEX IF EXISTS idx_im_channels_bot_identity; +ALTER TABLE im_channels DROP COLUMN IF EXISTS bot_identity; diff --git a/migrations/versioned/000024_im_channel_bot_identity.up.sql b/migrations/versioned/000024_im_channel_bot_identity.up.sql new file mode 100644 index 00000000..409bde91 --- /dev/null +++ b/migrations/versioned/000024_im_channel_bot_identity.up.sql @@ -0,0 +1,17 @@ +-- Migration: 000024_im_channel_bot_identity +-- Description: Add bot_identity column to im_channels for duplicate bot prevention. +-- bot_identity is a computed unique key derived from credentials, ensuring each bot +-- can only be connected to one active (non-deleted) channel. +DO $$ BEGIN RAISE NOTICE '[Migration 000024] Adding bot_identity column to im_channels'; END $$; + +ALTER TABLE im_channels ADD COLUMN IF NOT EXISTS bot_identity VARCHAR(255) NOT NULL DEFAULT ''; + +-- Partial unique index: only enforce uniqueness for non-deleted rows with a non-empty bot_identity. +-- Empty bot_identity (unknown credential format) is excluded from the constraint. +CREATE UNIQUE INDEX IF NOT EXISTS idx_im_channels_bot_identity + ON im_channels (bot_identity) + WHERE deleted_at IS NULL AND bot_identity != ''; + +COMMENT ON COLUMN im_channels.bot_identity IS 'Unique bot identity derived from credentials (e.g. wecom:ws:{bot_id}, feishu:{app_id}). Used to prevent duplicate bot bindings.'; + +DO $$ BEGIN RAISE NOTICE '[Migration 000024] bot_identity column and unique index created successfully'; END $$;