From 09af247a7c41f46d511b8aeae043b5f3adfef3f9 Mon Sep 17 00:00:00 2001 From: Max Date: Mon, 23 Mar 2026 23:47:25 +0800 Subject: [PATCH] feat(robot): add Weixin integration and enhance existing adapters - Introduced Weixin integration support with new configuration options for WeChat iLink Bot. - Updated existing adapters (DingTalk, Discord, Feishu, Telegram) to include sender_id and app_id in message metadata for improved context handling. - Enhanced dispatcher logic to accommodate the new Weixin adapter and ensure proper initialization and shutdown processes. - Improved message handling across integrations to support typing indicators, providing a more interactive user experience. --- agent/robot/api/lifecycle.go | 2 + agent/robot/api/weixin_qrcode.go | 103 +++++ agent/robot/events/delivery.go | 10 +- .../events/integrations/dingtalk/dingtalk.go | 21 +- .../events/integrations/dingtalk/message.go | 2 + .../events/integrations/discord/discord.go | 3 +- .../events/integrations/discord/message.go | 2 + agent/robot/events/integrations/dispatcher.go | 79 +--- .../events/integrations/dispatcher_test.go | 12 +- .../events/integrations/feishu/feishu.go | 21 +- .../events/integrations/feishu/message.go | 2 + .../robot/events/integrations/feishu/reply.go | 4 + .../events/integrations/telegram/message.go | 2 + .../events/integrations/telegram/reply.go | 4 + .../events/integrations/telegram/telegram.go | 6 +- .../robot/events/integrations/weixin/dedup.go | 44 ++ .../events/integrations/weixin/message.go | 179 +++++++++ .../events/integrations/weixin/polling.go | 161 ++++++++ .../robot/events/integrations/weixin/reply.go | 305 ++++++++++++++ .../events/integrations/weixin/syncbuf.go | 44 ++ .../integrations/weixin/typing_cache.go | 43 ++ .../events/integrations/weixin/weixin.go | 130 ++++++ agent/robot/robot.go | 69 ---- agent/robot/types/config.go | 11 + integrations/feishu/message.go | 7 + integrations/telegram/message.go | 13 + integrations/weixin/bot.go | 376 ++++++++++++++++++ integrations/weixin/cdn.go | 126 ++++++ integrations/weixin/format.go | 226 +++++++++++ integrations/weixin/mime.go | 47 +++ integrations/weixin/qrcode.go | 75 ++++ integrations/weixin/types.go | 163 ++++++++ openapi/agent/robot/robot.go | 4 + openapi/agent/robot/weixin.go | 72 ++++ 34 files changed, 2213 insertions(+), 155 deletions(-) create mode 100644 agent/robot/api/weixin_qrcode.go create mode 100644 agent/robot/events/integrations/weixin/dedup.go create mode 100644 agent/robot/events/integrations/weixin/message.go create mode 100644 agent/robot/events/integrations/weixin/polling.go create mode 100644 agent/robot/events/integrations/weixin/reply.go create mode 100644 agent/robot/events/integrations/weixin/syncbuf.go create mode 100644 agent/robot/events/integrations/weixin/typing_cache.go create mode 100644 agent/robot/events/integrations/weixin/weixin.go create mode 100644 integrations/weixin/bot.go create mode 100644 integrations/weixin/cdn.go create mode 100644 integrations/weixin/format.go create mode 100644 integrations/weixin/mime.go create mode 100644 integrations/weixin/qrcode.go create mode 100644 integrations/weixin/types.go create mode 100644 openapi/agent/robot/weixin.go diff --git a/agent/robot/api/lifecycle.go b/agent/robot/api/lifecycle.go index 233896fd..98f66ec5 100644 --- a/agent/robot/api/lifecycle.go +++ b/agent/robot/api/lifecycle.go @@ -11,6 +11,7 @@ import ( dcadapter "github.com/yaoapp/yao/agent/robot/events/integrations/discord" fsadapter "github.com/yaoapp/yao/agent/robot/events/integrations/feishu" "github.com/yaoapp/yao/agent/robot/events/integrations/telegram" + weixinadapter "github.com/yaoapp/yao/agent/robot/events/integrations/weixin" "github.com/yaoapp/yao/agent/robot/logger" "github.com/yaoapp/yao/agent/robot/manager" "github.com/yaoapp/yao/agent/robot/types" @@ -65,6 +66,7 @@ func Start() error { "feishu": fsadapter.NewAdapter(), "dingtalk": dtadapter.NewAdapter(), "discord": dcadapter.NewAdapter(), + "weixin": weixinadapter.NewAdapter(), } globalDispatcher = integrations.NewDispatcher(globalManager.Cache(), adapters) if err := globalDispatcher.Start(context.Background()); err != nil { diff --git a/agent/robot/api/weixin_qrcode.go b/agent/robot/api/weixin_qrcode.go new file mode 100644 index 00000000..fa58fa4e --- /dev/null +++ b/agent/robot/api/weixin_qrcode.go @@ -0,0 +1,103 @@ +package api + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/google/uuid" + weixinapi "github.com/yaoapp/yao/integrations/weixin" +) + +const ( + qrSessionTTL = 5 * time.Minute + maxQRRefreshCount = 3 +) + +type qrSession struct { + qrcode string + apiHost string + startedAt time.Time + refreshes int +} + +var ( + qrSessions = make(map[string]*qrSession) + qrSessionsMu sync.Mutex +) + +// WeixinQRCodeCreate creates a new QR code session for WeChat login. +// Returns the session key and QR code URL. +func WeixinQRCodeCreate(apiHost string) (sessionKey, qrcodeURL, qrcodeImg string, err error) { + qrcode, qrcodeImgContent, err := weixinapi.GetQRCode(context.Background(), apiHost) + if err != nil { + return "", "", "", fmt.Errorf("get QR code: %w", err) + } + + sessionKey = uuid.New().String() + qrSessionsMu.Lock() + qrSessions[sessionKey] = &qrSession{ + qrcode: qrcode, + apiHost: apiHost, + startedAt: time.Now(), + } + qrSessionsMu.Unlock() + + return sessionKey, qrcode, qrcodeImgContent, nil +} + +// WeixinQRCodePoll polls the QR code status for a given session. +func WeixinQRCodePoll(sessionKey string) (status, botToken, accountID, baseURL, userID string, err error) { + qrSessionsMu.Lock() + session, ok := qrSessions[sessionKey] + if !ok { + qrSessionsMu.Unlock() + return "", "", "", "", "", fmt.Errorf("session not found: %s", sessionKey) + } + + if time.Since(session.startedAt) > qrSessionTTL { + if session.refreshes < maxQRRefreshCount { + session.refreshes++ + session.startedAt = time.Now() + apiHost := session.apiHost + qrSessionsMu.Unlock() + + newQR, _, refreshErr := weixinapi.GetQRCode(context.Background(), apiHost) + if refreshErr != nil { + qrSessionsMu.Lock() + delete(qrSessions, sessionKey) + qrSessionsMu.Unlock() + return "expired", "", "", "", "", nil + } + + qrSessionsMu.Lock() + if s, ok := qrSessions[sessionKey]; ok { + s.qrcode = newQR + } + qrSessionsMu.Unlock() + return "refreshed", "", "", "", "", nil + } + delete(qrSessions, sessionKey) + qrSessionsMu.Unlock() + return "expired", "", "", "", "", nil + } + + qrcode := session.qrcode + apiHost := session.apiHost + qrSessionsMu.Unlock() + + resp, err := weixinapi.PollQRStatus(context.Background(), apiHost, qrcode) + if err != nil { + return "wait", "", "", "", "", nil + } + + if resp.Status == "confirmed" { + qrSessionsMu.Lock() + delete(qrSessions, sessionKey) + qrSessionsMu.Unlock() + return resp.Status, resp.BotToken, resp.IlinkBotID, resp.BaseURL, resp.UserID, nil + } + + return resp.Status, "", "", "", "", nil +} diff --git a/agent/robot/events/delivery.go b/agent/robot/events/delivery.go index dd3a23aa..fe1bef9f 100644 --- a/agent/robot/events/delivery.go +++ b/agent/robot/events/delivery.go @@ -104,10 +104,14 @@ func (h *robotHandler) handleDelivery(ctx context.Context, ev *eventtypes.Event, for k, v := range payload.Extra { extra[k] = v } + senderID, _ := payload.Extra["sender_id"].(string) + appID, _ := payload.Extra["app_id"].(string) metadata := &MessageMetadata{ - Channel: channel, - ChatID: chatID, - Extra: extra, + Channel: channel, + ChatID: chatID, + SenderID: senderID, + AppID: appID, + Extra: extra, } if err := reply(ctx, msg, metadata); err != nil { log.Error("delivery handler: integration reply failed channel=%s execution=%s: %v", channel, payload.ExecutionID, err) diff --git a/agent/robot/events/integrations/dingtalk/dingtalk.go b/agent/robot/events/integrations/dingtalk/dingtalk.go index c14138af..fd48c36d 100644 --- a/agent/robot/events/integrations/dingtalk/dingtalk.go +++ b/agent/robot/events/integrations/dingtalk/dingtalk.go @@ -26,10 +26,11 @@ type Adapter struct { // botEntry holds the state for one robot's DingTalk integration. type botEntry struct { - robotID string - clientID string - bot *dtapi.Bot - cancelFn context.CancelFunc + robotID string + clientID string + clientSecret string + bot *dtapi.Bot + cancelFn context.CancelFunc } // NewAdapter creates a new DingTalk adapter. @@ -58,7 +59,8 @@ func (a *Adapter) Apply(ctx context.Context, robot *robottypes.Robot) { defer a.mu.Unlock() if existing, ok := a.bots[robot.MemberID]; ok { - if existing.clientID == dtConf.ClientID { + if existing.clientID == dtConf.ClientID && + existing.clientSecret == dtConf.ClientSecret { return } a.removeBotLocked(robot.MemberID) @@ -68,10 +70,11 @@ func (a *Adapter) Apply(ctx context.Context, robot *robottypes.Robot) { streamCtx, streamCancel := context.WithCancel(context.Background()) entry := &botEntry{ - robotID: robot.MemberID, - clientID: dtConf.ClientID, - bot: bot, - cancelFn: streamCancel, + robotID: robot.MemberID, + clientID: dtConf.ClientID, + clientSecret: dtConf.ClientSecret, + bot: bot, + cancelFn: streamCancel, } a.bots[robot.MemberID] = entry a.appIdx[dtConf.ClientID] = robot.MemberID diff --git a/agent/robot/events/integrations/dingtalk/message.go b/agent/robot/events/integrations/dingtalk/message.go index 6239ecb4..cd510254 100644 --- a/agent/robot/events/integrations/dingtalk/message.go +++ b/agent/robot/events/integrations/dingtalk/message.go @@ -62,6 +62,8 @@ func (a *Adapter) handleMessages(ctx context.Context, entry *botEntry, cms []*dt "session_webhook": lastCM.SessionWebhook, "conversation_type": lastCM.ConversationType, "dt_message_id": lastCM.MessageID, + "sender_id": lastCM.SenderID, + "app_id": entry.clientID, }, }, } diff --git a/agent/robot/events/integrations/discord/discord.go b/agent/robot/events/integrations/discord/discord.go index 0d1eaaec..beb123f8 100644 --- a/agent/robot/events/integrations/discord/discord.go +++ b/agent/robot/events/integrations/discord/discord.go @@ -58,7 +58,8 @@ func (a *Adapter) Apply(ctx context.Context, robot *robottypes.Robot) { defer a.mu.Unlock() if existing, ok := a.bots[robot.MemberID]; ok { - if existing.bot.Token() == dcConf.BotToken { + if existing.bot.Token() == dcConf.BotToken && + existing.appID == dcConf.AppID { return } a.removeBotLocked(robot.MemberID) diff --git a/agent/robot/events/integrations/discord/message.go b/agent/robot/events/integrations/discord/message.go index 8b83b4fd..42b69c59 100644 --- a/agent/robot/events/integrations/discord/message.go +++ b/agent/robot/events/integrations/discord/message.go @@ -67,6 +67,8 @@ func (a *Adapter) handleMessages(ctx context.Context, entry *botEntry, cms []*dc "discord_message_id": lastCM.MessageID, "guild_id": lastCM.GuildID, "is_dm": lastCM.IsDM, + "sender_id": lastCM.AuthorID, + "app_id": entry.appID, }, }, } diff --git a/agent/robot/events/integrations/dispatcher.go b/agent/robot/events/integrations/dispatcher.go index a277b024..88a05e84 100644 --- a/agent/robot/events/integrations/dispatcher.go +++ b/agent/robot/events/integrations/dispatcher.go @@ -4,8 +4,6 @@ import ( "context" "fmt" - "github.com/yaoapp/gou/model" - "github.com/yaoapp/kun/maps" agentcontext "github.com/yaoapp/yao/agent/context" robotcache "github.com/yaoapp/yao/agent/robot/cache" events "github.com/yaoapp/yao/agent/robot/events" @@ -22,6 +20,7 @@ type Adapter interface { Apply(ctx context.Context, robot *robottypes.Robot) Remove(ctx context.Context, robotID string) Reply(ctx context.Context, msg *agentcontext.Message, metadata *events.MessageMetadata) error + Shutdown() } // Dispatcher distributes Robot integration configs to platform adapters. @@ -48,7 +47,7 @@ func (d *Dispatcher) Start(ctx context.Context) error { events.RegisterReplyFunc(d.reply) - ch := make(chan *eventtypes.Event, 64) + ch := make(chan *eventtypes.Event, 256) d.subID = event.Subscribe("robot.config.*", ch) go d.watch(ctx, ch) @@ -81,76 +80,29 @@ func (d *Dispatcher) reply(ctx context.Context, msg *agentcontext.Message, metad return lastErr } -// Stop unsubscribes from events. +// Stop unsubscribes from events and shuts down all adapters. func (d *Dispatcher) Stop() { close(d.stopCh) if d.subID != "" { event.Unsubscribe(d.subID) } + for name, adapter := range d.adapters { + adapter.Shutdown() + log.Info("integration dispatcher: adapter %s shutdown", name) + } log.Info("integration dispatcher: stopped") } func (d *Dispatcher) loadAll(ctx context.Context) { - robots := d.loadIntegrationRobots() + robots := d.robotCache.ListAll() + count := 0 for _, robot := range robots { - d.robotCache.Add(robot) - d.apply(ctx, robot) + if robot.Config != nil && robot.Config.Integrations != nil && len(parseIntegrations(robot.Config.Integrations)) > 0 { + d.apply(ctx, robot) + count++ + } } - log.Info("integration dispatcher: initial load complete, %d robots with integrations", len(robots)) -} - -// loadIntegrationRobots queries all active robots that have a non-null -// robot_config (which may contain integrations). This is independent of -// autonomous_mode so non-autonomous robots with Telegram etc. are included. -func (d *Dispatcher) loadIntegrationRobots() []*robottypes.Robot { - m := model.Select("__yao.member") - fields := []interface{}{ - "id", "member_id", "team_id", "display_name", "bio", - "system_prompt", "robot_status", "autonomous_mode", - "robot_config", "robot_email", "agents", "mcp_servers", - "manager_id", "language_model", - } - - page := 1 - pageSize := 100 - var result []*robottypes.Robot - - for { - res, err := m.Paginate(model.QueryParam{ - Select: fields, - Wheres: []model.QueryWhere{ - {Column: "member_type", Value: "robot"}, - {Column: "status", Value: "active"}, - }, - }, page, pageSize) - if err != nil { - log.Error("loadIntegrationRobots: query failed page=%d: %v", page, err) - break - } - - data, ok := res.Get("data").([]maps.MapStr) - if !ok || len(data) == 0 { - break - } - - for _, record := range data { - robot, err := robottypes.NewRobotFromMap(map[string]interface{}(record)) - if err != nil { - continue - } - if robot.Config != nil && robot.Config.Integrations != nil && len(parseIntegrations(robot.Config.Integrations)) > 0 { - result = append(result, robot) - } - } - - total, _ := res.Get("total").(int) - if page*pageSize >= total { - break - } - page++ - } - - return result + log.Info("integration dispatcher: initial load complete, %d robots with integrations", count) } // apply parses which integrations the robot has configured, @@ -187,6 +139,9 @@ func parseIntegrations(intg *robottypes.Integrations) []string { if intg.Discord != nil { keys = append(keys, "discord") } + if intg.Weixin != nil { + keys = append(keys, "weixin") + } return keys } diff --git a/agent/robot/events/integrations/dispatcher_test.go b/agent/robot/events/integrations/dispatcher_test.go index 454a9bdd..717e5ed8 100644 --- a/agent/robot/events/integrations/dispatcher_test.go +++ b/agent/robot/events/integrations/dispatcher_test.go @@ -12,6 +12,7 @@ import ( robotcache "github.com/yaoapp/yao/agent/robot/cache" events "github.com/yaoapp/yao/agent/robot/events" robottypes "github.com/yaoapp/yao/agent/robot/types" + "github.com/yaoapp/yao/agent/testutils" "github.com/yaoapp/yao/event" eventtypes "github.com/yaoapp/yao/event/types" ) @@ -39,6 +40,8 @@ func (m *mockAdapter) Reply(ctx context.Context, msg *agentcontext.Message, meta return nil } +func (m *mockAdapter) Shutdown() {} + func (m *mockAdapter) getApplied() []*robottypes.Robot { m.mu.Lock() defer m.mu.Unlock() @@ -239,6 +242,13 @@ func TestConfigDeleted_TriggersRemove(t *testing.T) { } func TestConfigCreated_RobotNotInCache(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + setupEventBus(t) cache := robotcache.New() @@ -248,7 +258,7 @@ func TestConfigCreated_RobotNotInCache(t *testing.T) { require.NoError(t, d.Start(context.Background())) defer d.Stop() - // Push event but don't add robot to cache + // Push event but don't add robot to cache — triggers LoadByID DB fallback event.Push(context.Background(), events.RobotConfigCreated, events.RobotConfigPayload{ MemberID: "r-ghost", TeamID: "team1", }) diff --git a/agent/robot/events/integrations/feishu/feishu.go b/agent/robot/events/integrations/feishu/feishu.go index f0eb1a2f..12982259 100644 --- a/agent/robot/events/integrations/feishu/feishu.go +++ b/agent/robot/events/integrations/feishu/feishu.go @@ -26,10 +26,11 @@ type Adapter struct { // botEntry holds the state for one robot's Feishu integration. type botEntry struct { - robotID string - appID string - bot *fsapi.Bot - cancelFn context.CancelFunc // cancels the event subscription goroutine + robotID string + appID string + appSecret string + bot *fsapi.Bot + cancelFn context.CancelFunc // cancels the event subscription goroutine } // NewAdapter creates a new Feishu adapter. @@ -58,7 +59,8 @@ func (a *Adapter) Apply(ctx context.Context, robot *robottypes.Robot) { defer a.mu.Unlock() if existing, ok := a.bots[robot.MemberID]; ok { - if existing.appID == fsConf.AppID { + if existing.appID == fsConf.AppID && + existing.appSecret == fsConf.AppSecret { return } a.removeBotLocked(robot.MemberID) @@ -68,10 +70,11 @@ func (a *Adapter) Apply(ctx context.Context, robot *robottypes.Robot) { streamCtx, streamCancel := context.WithCancel(context.Background()) entry := &botEntry{ - robotID: robot.MemberID, - appID: fsConf.AppID, - bot: bot, - cancelFn: streamCancel, + robotID: robot.MemberID, + appID: fsConf.AppID, + appSecret: fsConf.AppSecret, + bot: bot, + cancelFn: streamCancel, } a.bots[robot.MemberID] = entry a.appIdx[fsConf.AppID] = robot.MemberID diff --git a/agent/robot/events/integrations/feishu/message.go b/agent/robot/events/integrations/feishu/message.go index 9ee5da86..9cd093c6 100644 --- a/agent/robot/events/integrations/feishu/message.go +++ b/agent/robot/events/integrations/feishu/message.go @@ -60,6 +60,8 @@ func (a *Adapter) handleMessages(ctx context.Context, entry *botEntry, cms []*fs Locale: events.NormalizeLocale(lastCM.LanguageCode), Extra: map[string]any{ "feishu_message_id": lastCM.MessageID, + "sender_id": lastCM.SenderID, + "app_id": entry.appID, }, }, } diff --git a/agent/robot/events/integrations/feishu/reply.go b/agent/robot/events/integrations/feishu/reply.go index 871cb4e9..977c4344 100644 --- a/agent/robot/events/integrations/feishu/reply.go +++ b/agent/robot/events/integrations/feishu/reply.go @@ -30,6 +30,10 @@ func (a *Adapter) Reply(ctx context.Context, msg *agentcontext.Message, metadata } } + if err := entry.bot.SendTyping(ctx, metadata.ChatID); err != nil { + log.Debug("feishu reply: send typing failed: %v", err) + } + return a.sendContent(ctx, entry, metadata.ChatID, replyToMsgID, msg.Content) } diff --git a/agent/robot/events/integrations/telegram/message.go b/agent/robot/events/integrations/telegram/message.go index 1a26cf68..00b83887 100644 --- a/agent/robot/events/integrations/telegram/message.go +++ b/agent/robot/events/integrations/telegram/message.go @@ -67,6 +67,8 @@ func (a *Adapter) handleMessages(ctx context.Context, entry *botEntry, cms []*tg Locale: events.NormalizeLocale(lastCM.LanguageCode), Extra: map[string]any{ "tg_message_id": lastCM.MessageID, + "sender_id": strconv.FormatInt(lastCM.SenderID, 10), + "app_id": entry.appID, }, }, } diff --git a/agent/robot/events/integrations/telegram/reply.go b/agent/robot/events/integrations/telegram/reply.go index 9643f15a..3809a93e 100644 --- a/agent/robot/events/integrations/telegram/reply.go +++ b/agent/robot/events/integrations/telegram/reply.go @@ -41,6 +41,10 @@ func (a *Adapter) Reply(ctx context.Context, msg *agentcontext.Message, metadata return fmt.Errorf("no bot registered for channel metadata (appID=%s)", metadata.AppID) } + if err := entry.bot.SendTyping(ctx, chatID); err != nil { + log.Debug("telegram reply: send typing failed: %v", err) + } + return a.sendContent(ctx, entry.bot, chatID, replyTo, msg.Content) } diff --git a/agent/robot/events/integrations/telegram/telegram.go b/agent/robot/events/integrations/telegram/telegram.go index 143d9f2e..92d690f2 100644 --- a/agent/robot/events/integrations/telegram/telegram.go +++ b/agent/robot/events/integrations/telegram/telegram.go @@ -30,6 +30,7 @@ type Adapter struct { type botEntry struct { robotID string appID string + host string bot *tgapi.Bot // bound to this robot's token offset int64 // polling offset } @@ -65,7 +66,9 @@ func (a *Adapter) Apply(ctx context.Context, robot *robottypes.Robot) { defer a.mu.Unlock() if existing, ok := a.bots[robot.MemberID]; ok { - if existing.bot.Token() == tgConf.BotToken { + if existing.bot.Token() == tgConf.BotToken && + existing.appID == tgConf.AppID && + existing.host == tgConf.Host { return } a.removeBotLocked(robot.MemberID) @@ -78,6 +81,7 @@ func (a *Adapter) Apply(ctx context.Context, robot *robottypes.Robot) { entry := &botEntry{ robotID: robot.MemberID, appID: tgConf.AppID, + host: tgConf.Host, bot: tgapi.NewBot(tgConf.BotToken, tgConf.WebhookSecret, opts...), } a.bots[robot.MemberID] = entry diff --git a/agent/robot/events/integrations/weixin/dedup.go b/agent/robot/events/integrations/weixin/dedup.go new file mode 100644 index 00000000..a8ae66fd --- /dev/null +++ b/agent/robot/events/integrations/weixin/dedup.go @@ -0,0 +1,44 @@ +package weixin + +import ( + "sync" + "time" +) + +const ( + dedupTTL = 24 * time.Hour + dedupCleanInterval = time.Hour +) + +type dedupStore struct { + m sync.Map +} + +func newDedupStore() *dedupStore { + return &dedupStore{} +} + +func (d *dedupStore) markSeen(key string) bool { + now := time.Now().Unix() + _, loaded := d.m.LoadOrStore(key, now) + return !loaded +} + +func (d *dedupStore) cleaner(stopCh <-chan struct{}) { + ticker := time.NewTicker(dedupCleanInterval) + defer ticker.Stop() + for { + select { + case <-stopCh: + return + case <-ticker.C: + cutoff := time.Now().Add(-dedupTTL).Unix() + d.m.Range(func(key, value any) bool { + if ts, ok := value.(int64); ok && ts < cutoff { + d.m.Delete(key) + } + return true + }) + } + } +} diff --git a/agent/robot/events/integrations/weixin/message.go b/agent/robot/events/integrations/weixin/message.go new file mode 100644 index 00000000..2e259634 --- /dev/null +++ b/agent/robot/events/integrations/weixin/message.go @@ -0,0 +1,179 @@ +package weixin + +import ( + "bytes" + "context" + "encoding/hex" + "fmt" + "mime/multipart" + "net/textproto" + "strings" + + "github.com/yaoapp/yao/attachment" + weixinapi "github.com/yaoapp/yao/integrations/weixin" +) + +type resolvedMedia struct { + Wrapper string + MimeType string + FileName string +} + +func convertMessage(ctx context.Context, bot *weixinapi.Bot, items []weixinapi.MsgItem, groups []string) (string, []resolvedMedia) { + var textBuf strings.Builder + var media []resolvedMedia + + for _, item := range items { + switch item.Type { + case weixinapi.ItemTypeText: + if item.TextItem != nil && item.TextItem.Text != "" { + text := item.TextItem.Text + if item.RefMsg != nil { + text = formatRefMessage(item.RefMsg, text) + } + textBuf.WriteString(text) + } + case weixinapi.ItemTypeVoice: + if item.VoiceItem != nil { + if item.VoiceItem.Text != "" { + textBuf.WriteString(item.VoiceItem.Text) + } else if item.VoiceItem.Media != nil && item.VoiceItem.Media.EncryptQueryParam != "" { + m := resolveVoice(ctx, bot, item.VoiceItem, groups) + if m != nil { + media = append(media, *m) + } + } + } + case weixinapi.ItemTypeImage: + if item.ImageItem != nil { + m := resolveImage(ctx, bot, item.ImageItem, groups) + if m != nil { + media = append(media, *m) + } + } + case weixinapi.ItemTypeFile: + if item.FileItem != nil && item.FileItem.Media != nil && item.FileItem.Media.EncryptQueryParam != "" { + m := resolveFile(ctx, bot, item.FileItem, groups) + if m != nil { + media = append(media, *m) + } + } + case weixinapi.ItemTypeVideo: + if item.VideoItem != nil && item.VideoItem.Media != nil && item.VideoItem.Media.EncryptQueryParam != "" { + m := resolveVideo(ctx, bot, item.VideoItem, groups) + if m != nil { + media = append(media, *m) + } + } + } + } + + return textBuf.String(), media +} + +func formatRefMessage(ref *weixinapi.RefMessage, text string) string { + if ref == nil || ref.MessageItem == nil { + return text + } + var refBody string + if ref.MessageItem.TextItem != nil { + refBody = ref.MessageItem.TextItem.Text + } + title := ref.Title + if title == "" && refBody == "" { + return text + } + return fmt.Sprintf("[引用: %s | %s]\n%s", title, refBody, text) +} + +func resolveImage(ctx context.Context, bot *weixinapi.Bot, img *weixinapi.ImageItem, groups []string) *resolvedMedia { + if img.AesKey != "" && img.Media != nil && img.Media.EncryptQueryParam != "" { + rawKey, err := hex.DecodeString(img.AesKey) + if err == nil { + data, err := weixinapi.DecryptFromRaw(bot.CDNBaseURL(), img.Media.EncryptQueryParam, rawKey) + if err == nil { + return storeMedia(ctx, data, "image/jpeg", "image.jpg", groups) + } + } + } + if img.Media != nil && img.Media.EncryptQueryParam != "" && img.Media.AesKey != "" { + data, err := weixinapi.DownloadAndDecrypt(bot.CDNBaseURL(), img.Media.EncryptQueryParam, img.Media.AesKey) + if err == nil { + return storeMedia(ctx, data, "image/jpeg", "image.jpg", groups) + } + } + return nil +} + +func resolveVoice(ctx context.Context, bot *weixinapi.Bot, voice *weixinapi.VoiceItem, groups []string) *resolvedMedia { + data, err := weixinapi.DownloadAndDecrypt(bot.CDNBaseURL(), voice.Media.EncryptQueryParam, voice.Media.AesKey) + if err != nil { + log.Error("weixin: voice decrypt failed: %v", err) + return nil + } + mime := "audio/mpeg" + ext := "mp3" + if voice.EncodeType == 6 { + mime = "audio/silk" + ext = "silk" + } + return storeMedia(ctx, data, mime, "voice."+ext, groups) +} + +func resolveFile(ctx context.Context, bot *weixinapi.Bot, file *weixinapi.FileItem, groups []string) *resolvedMedia { + data, err := weixinapi.DownloadAndDecrypt(bot.CDNBaseURL(), file.Media.EncryptQueryParam, file.Media.AesKey) + if err != nil { + log.Error("weixin: file decrypt failed: %v", err) + return nil + } + filename := file.FileName + if filename == "" { + filename = "file.bin" + } + mime := weixinapi.MimeFromFilename(filename) + return storeMedia(ctx, data, mime, filename, groups) +} + +func resolveVideo(ctx context.Context, bot *weixinapi.Bot, video *weixinapi.VideoItem, groups []string) *resolvedMedia { + data, err := weixinapi.DownloadAndDecrypt(bot.CDNBaseURL(), video.Media.EncryptQueryParam, video.Media.AesKey) + if err != nil { + log.Error("weixin: video decrypt failed: %v", err) + return nil + } + return storeMedia(ctx, data, "video/mp4", "video.mp4", groups) +} + +func storeMedia(ctx context.Context, data []byte, mimeType, filename string, groups []string) *resolvedMedia { + manager, exists := attachment.Managers["__yao.attachment"] + if !exists { + log.Error("weixin: __yao.attachment manager not found") + return nil + } + + fh := makeFileHeader(filename, mimeType, int64(len(data))) + reader := bytes.NewReader(data) + file, err := manager.Upload(ctx, fh, reader, attachment.UploadOption{Groups: groups}) + if err != nil { + log.Error("weixin: attachment upload failed: %v", err) + return nil + } + + return &resolvedMedia{ + Wrapper: fmt.Sprintf("__yao.attachment://%s", file.ID), + MimeType: mimeType, + FileName: filename, + } +} + +func makeFileHeader(filename, contentType string, size int64) *attachment.FileHeader { + hdr := make(textproto.MIMEHeader) + hdr.Set("Content-Disposition", fmt.Sprintf(`form-data; name="file"; filename="%s"`, filename)) + hdr.Set("Content-Type", contentType) + return &attachment.FileHeader{ + FileHeader: &multipart.FileHeader{ + Filename: filename, + Header: hdr, + Size: size, + }, + } +} diff --git a/agent/robot/events/integrations/weixin/polling.go b/agent/robot/events/integrations/weixin/polling.go new file mode 100644 index 00000000..f3eaf0a9 --- /dev/null +++ b/agent/robot/events/integrations/weixin/polling.go @@ -0,0 +1,161 @@ +package weixin + +import ( + "context" + "fmt" + "time" + + agentcontext "github.com/yaoapp/yao/agent/context" + events "github.com/yaoapp/yao/agent/robot/events" + "github.com/yaoapp/yao/event" + weixinapi "github.com/yaoapp/yao/integrations/weixin" +) + +const ( + maxConsecutiveFailures = 3 + backoffDuration = 30 * time.Second + retryDuration = 2 * time.Second + sessionPauseDuration = 30 * time.Minute + defaultTimeoutMs = 35_000 +) + +func (a *Adapter) pollLoop(ctx context.Context, entry *botEntry) { + syncBuf := loadSyncBuf(entry.accountID) + nextTimeoutMs := defaultTimeoutMs + failures := 0 + + for { + select { + case <-ctx.Done(): + return + default: + } + + resp, err := entry.bot.GetUpdates(ctx, syncBuf, nextTimeoutMs) + if err != nil { + if ctx.Err() != nil { + return + } + failures++ + if failures >= maxConsecutiveFailures { + failures = 0 + sleep(ctx, backoffDuration) + } else { + sleep(ctx, retryDuration) + } + continue + } + + if resp.ErrCode == weixinapi.SessionExpiredErrCode || resp.Ret == weixinapi.SessionExpiredErrCode { + log.Warn("weixin session expired, pausing %s robot=%s", sessionPauseDuration, entry.robotID) + failures = 0 + sleep(ctx, sessionPauseDuration) + continue + } + + isApiError := (resp.Ret != 0) || (resp.ErrCode != 0) + if isApiError { + failures++ + if failures >= maxConsecutiveFailures { + failures = 0 + sleep(ctx, backoffDuration) + } else { + sleep(ctx, retryDuration) + } + continue + } + failures = 0 + + if resp.LongPollingTimeoutMs > 0 { + nextTimeoutMs = resp.LongPollingTimeoutMs + } + + if resp.GetUpdatesBuf != "" && resp.GetUpdatesBuf != syncBuf { + syncBuf = resp.GetUpdatesBuf + saveSyncBuf(entry.accountID, syncBuf) + } + + for i := range resp.Msgs { + a.handleMessage(ctx, entry, &resp.Msgs[i]) + } + } +} + +func (a *Adapter) handleMessage(ctx context.Context, entry *botEntry, msg *weixinapi.WeixinMessage) { + var dedupKey string + switch { + case msg.MessageID != 0: + dedupKey = fmt.Sprintf("wx:%s:mid:%d", entry.robotID, msg.MessageID) + case msg.Seq != 0: + dedupKey = fmt.Sprintf("wx:%s:seq:%d", entry.robotID, msg.Seq) + default: + dedupKey = fmt.Sprintf("wx:%s:%s:%d", entry.robotID, msg.FromUserID, msg.CreateTimeMs) + } + if !a.dedup.markSeen(dedupKey) { + return + } + + log.Info("incoming msg from=%s context_token=%s", msg.FromUserID, msg.ContextToken) + + groups := []string{"weixin", entry.accountID} + content, mediaItems := convertMessage(ctx, entry.bot, msg.ItemList, groups) + + if content == "" && len(mediaItems) == 0 { + return + } + + var msgContent interface{} + if len(mediaItems) == 0 { + msgContent = content + } else { + parts := make([]interface{}, 0, 1+len(mediaItems)) + if content != "" { + parts = append(parts, map[string]interface{}{"type": "text", "text": content}) + } + for _, m := range mediaItems { + parts = append(parts, map[string]interface{}{ + "type": "file", + "file_url": m.Wrapper, + "mime_type": m.MimeType, + "file_name": m.FileName, + }) + } + msgContent = parts + } + + messageID := "" + if msg.MessageID != 0 { + messageID = fmt.Sprintf("%d", msg.MessageID) + } + + payload := events.MessagePayload{ + RobotID: entry.robotID, + Messages: []agentcontext.Message{ + {Role: agentcontext.RoleUser, Content: msgContent}, + }, + Metadata: &events.MessageMetadata{ + Channel: "weixin", + MessageID: messageID, + AppID: entry.accountID, + ChatID: msg.FromUserID, + SenderID: msg.FromUserID, + Locale: "zh-cn", + Extra: map[string]any{ + "context_token": msg.ContextToken, + "sender_id": msg.FromUserID, + "app_id": entry.accountID, + }, + }, + } + + if _, err := event.Push(ctx, events.Message, payload); err != nil { + log.Error("weixin adapter: event.Push failed robot=%s: %v", entry.robotID, err) + } +} + +func sleep(ctx context.Context, d time.Duration) { + select { + case <-ctx.Done(): + case <-time.After(d): + } +} diff --git a/agent/robot/events/integrations/weixin/reply.go b/agent/robot/events/integrations/weixin/reply.go new file mode 100644 index 00000000..0d4ba9a3 --- /dev/null +++ b/agent/robot/events/integrations/weixin/reply.go @@ -0,0 +1,305 @@ +package weixin + +import ( + "context" + "fmt" + "io" + "net/http" + "strings" + + agentcontext "github.com/yaoapp/yao/agent/context" + events "github.com/yaoapp/yao/agent/robot/events" + "github.com/yaoapp/yao/attachment" + weixinapi "github.com/yaoapp/yao/integrations/weixin" +) + +func (a *Adapter) Reply(ctx context.Context, msg *agentcontext.Message, metadata *events.MessageMetadata) error { + if msg == nil || metadata == nil { + return fmt.Errorf("weixin Reply: nil message or metadata") + } + + entry := a.resolveByAccountID(metadata.AppID) + if entry == nil { + a.mu.RLock() + for _, e := range a.bots { + entry = e + break + } + a.mu.RUnlock() + } + if entry == nil { + return fmt.Errorf("weixin Reply: no bot registered (appID=%s)", metadata.AppID) + } + + contextToken, _ := metadata.Extra["context_token"].(string) + toUserID := metadata.SenderID + if toUserID == "" { + toUserID = metadata.ChatID + } + + ticket := entry.ticketCache.Get(toUserID) + if ticket == "" { + if t, err := entry.bot.GetConfig(ctx, toUserID, contextToken); err == nil && t != "" { + ticket = t + entry.ticketCache.Set(toUserID, ticket) + } + } + + if ticket != "" { + _ = entry.bot.SendTyping(ctx, toUserID, ticket, 1) + } + + return a.sendContent(ctx, entry, toUserID, contextToken, msg.Content) +} + +func (a *Adapter) sendContent(ctx context.Context, entry *botEntry, toUserID, contextToken string, content interface{}) error { + switch c := content.(type) { + case string: + if strings.TrimSpace(c) == "" { + return nil + } + return entry.bot.SendMessage(ctx, toUserID, contextToken, weixinapi.FormatWeixinText(c)) + + case []interface{}: + return a.sendParts(ctx, entry, toUserID, contextToken, c) + + default: + parts, ok := toContentParts(content) + if ok { + return a.sendPartsTyped(ctx, entry, toUserID, contextToken, parts) + } + return entry.bot.SendMessage(ctx, toUserID, contextToken, weixinapi.FormatWeixinText(fmt.Sprintf("%v", content))) + } +} + +func (a *Adapter) sendParts(ctx context.Context, entry *botEntry, toUserID, contextToken string, parts []interface{}) error { + var textBuf strings.Builder + for _, part := range parts { + m, ok := part.(map[string]interface{}) + if !ok { + continue + } + partType, _ := m["type"].(string) + switch partType { + case "text": + if text, ok := m["text"].(string); ok { + textBuf.WriteString(text) + } + case "image_url": + if err := a.flushText(ctx, entry, toUserID, contextToken, &textBuf); err != nil { + return err + } + if imgMap, ok := m["image_url"].(map[string]interface{}); ok { + if url, ok := imgMap["url"].(string); ok { + if err := a.sendMediaFromURL(ctx, entry, toUserID, contextToken, url, "", "image"); err != nil { + log.Error("weixin reply: send image: %v", err) + } + } + } + case "file": + if err := a.flushText(ctx, entry, toUserID, contextToken, &textBuf); err != nil { + return err + } + fileURL, _ := m["file_url"].(string) + fileName, _ := m["file_name"].(string) + mimeType, _ := m["mime_type"].(string) + if fileURL == "" { + if fileMap, ok := m["file"].(map[string]interface{}); ok { + fileURL, _ = fileMap["url"].(string) + if fileName == "" { + fileName, _ = fileMap["filename"].(string) + } + } + } + if fileURL != "" { + mediaHint := detectMediaHint(mimeType, fileName) + if err := a.sendMediaFromURL(ctx, entry, toUserID, contextToken, fileURL, fileName, mediaHint); err != nil { + log.Error("weixin reply: send file: %v", err) + } + } + } + } + return a.flushText(ctx, entry, toUserID, contextToken, &textBuf) +} + +func (a *Adapter) sendPartsTyped(ctx context.Context, entry *botEntry, toUserID, contextToken string, parts []agentcontext.ContentPart) error { + var textBuf strings.Builder + for _, part := range parts { + switch part.Type { + case agentcontext.ContentText: + textBuf.WriteString(part.Text) + case agentcontext.ContentImageURL: + if err := a.flushText(ctx, entry, toUserID, contextToken, &textBuf); err != nil { + return err + } + if part.ImageURL != nil { + if err := a.sendMediaFromURL(ctx, entry, toUserID, contextToken, part.ImageURL.URL, "", "image"); err != nil { + log.Error("weixin reply: send image: %v", err) + } + } + case agentcontext.ContentFile: + if err := a.flushText(ctx, entry, toUserID, contextToken, &textBuf); err != nil { + return err + } + if part.File != nil { + mediaHint := detectMediaHint("", part.File.Filename) + if err := a.sendMediaFromURL(ctx, entry, toUserID, contextToken, part.File.URL, part.File.Filename, mediaHint); err != nil { + log.Error("weixin reply: send file: %v", err) + } + } + } + } + return a.flushText(ctx, entry, toUserID, contextToken, &textBuf) +} + +func (a *Adapter) flushText(ctx context.Context, entry *botEntry, toUserID, contextToken string, buf *strings.Builder) error { + if buf.Len() == 0 { + return nil + } + text := weixinapi.FormatWeixinText(buf.String()) + buf.Reset() + return entry.bot.SendMessage(ctx, toUserID, contextToken, text) +} + +func (a *Adapter) sendMediaFromURL(ctx context.Context, entry *botEntry, toUserID, contextToken, fileURL, fileName, mediaHint string) error { + log.Info("weixin sendMedia: to=%s url=%s fileName=%q hint=%s contextToken_len=%d", + toUserID, fileURL, fileName, mediaHint, len(contextToken)) + + var plaintext []byte + var contentType string + + if isWrapper(fileURL) { + managerName, fileID, err := parseWrapper(fileURL) + if err != nil { + return err + } + log.Info("weixin sendMedia: wrapper manager=%s fileID=%s", managerName, fileID) + manager, exists := attachment.Managers[managerName] + if !exists { + return fmt.Errorf("attachment manager %s not found", managerName) + } + resp, err := manager.Download(ctx, fileID) + if err != nil { + return fmt.Errorf("attachment download %s: %w", fileID, err) + } + defer resp.Reader.Close() + plaintext, err = io.ReadAll(resp.Reader) + if err != nil { + return fmt.Errorf("read attachment %s: %w", fileID, err) + } + contentType = resp.ContentType + if fileName == "" { + fileName = fileID + resp.Extension + } + log.Info("weixin sendMedia: attachment downloaded bytes=%d contentType=%q fileName=%q", len(plaintext), contentType, fileName) + } else if strings.HasPrefix(fileURL, "http") { + resp, err := http.Get(fileURL) //nolint:gosec + if err != nil { + return fmt.Errorf("download %s: %w", fileURL, err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("download %s: HTTP %d", fileURL, resp.StatusCode) + } + plaintext, err = io.ReadAll(resp.Body) + if err != nil { + return fmt.Errorf("read %s: %w", fileURL, err) + } + contentType = resp.Header.Get("Content-Type") + log.Info("weixin sendMedia: http downloaded bytes=%d contentType=%q", len(plaintext), contentType) + } else { + return fmt.Errorf("unsupported URL scheme: %s", fileURL) + } + + if mediaHint == "" { + mediaHint = detectMediaHint(contentType, fileName) + } + + var mediaType int + switch mediaHint { + case "image": + mediaType = weixinapi.UploadMediaImage + case "video": + mediaType = weixinapi.UploadMediaVideo + default: + mediaType = weixinapi.UploadMediaFile + } + + log.Info("weixin sendMedia: uploading media_type=%d mediaHint=%s bytes=%d to=%s", mediaType, mediaHint, len(plaintext), toUserID) + uploaded, err := entry.bot.UploadMedia(ctx, plaintext, toUserID, mediaType) + if err != nil { + log.Error("weixin UploadMedia failed: media_type=%d mediaHint=%s bytes=%d to=%s err=%v", mediaType, mediaHint, len(plaintext), toUserID, err) + fallbackText := fileURL + if fileName != "" { + fallbackText = fileName + "\n" + fileURL + } + return entry.bot.SendMessage(ctx, toUserID, contextToken, fallbackText) + } + + switch mediaHint { + case "image": + return entry.bot.SendImageMessage(ctx, toUserID, contextToken, uploaded) + case "video": + return entry.bot.SendVideoMessage(ctx, toUserID, contextToken, uploaded) + default: + if fileName == "" { + fileName = "file.bin" + } + return entry.bot.SendFileMessage(ctx, toUserID, contextToken, fileName, uploaded) + } +} + +func detectMediaHint(mimeType, fileName string) string { + lower := strings.ToLower(mimeType) + if strings.HasPrefix(lower, "image/") { + return "image" + } + if strings.HasPrefix(lower, "video/") { + return "video" + } + // TODO(weixin-voice): audio/* detected as "file" because iLink Bot voice + // playback is not yet functional. Switch to "voice" once supported. + if fileName != "" { + ext := strings.ToLower(fileName) + if strings.HasSuffix(ext, ".jpg") || strings.HasSuffix(ext, ".jpeg") || + strings.HasSuffix(ext, ".png") || strings.HasSuffix(ext, ".gif") || + strings.HasSuffix(ext, ".webp") || strings.HasSuffix(ext, ".bmp") { + return "image" + } + if strings.HasSuffix(ext, ".mp4") || strings.HasSuffix(ext, ".mov") || + strings.HasSuffix(ext, ".avi") || strings.HasSuffix(ext, ".webm") { + return "video" + } + } + return "file" +} + +func isWrapper(url string) bool { + return strings.Contains(url, "://") && !strings.HasPrefix(url, "http") +} + +func parseWrapper(wrapper string) (managerName, fileID string, err error) { + idx := strings.Index(wrapper, "://") + if idx < 0 { + return "", "", fmt.Errorf("invalid wrapper: %s", wrapper) + } + return wrapper[:idx], wrapper[idx+3:], nil +} + +func toContentParts(content interface{}) ([]agentcontext.ContentPart, bool) { + parts, ok := content.([]agentcontext.ContentPart) + return parts, ok +} + +func (a *Adapter) resolveByAccountID(accountID string) *botEntry { + if accountID == "" { + return nil + } + a.mu.RLock() + defer a.mu.RUnlock() + robotID, ok := a.accountIdx[accountID] + if !ok { + return nil + } + return a.bots[robotID] +} diff --git a/agent/robot/events/integrations/weixin/syncbuf.go b/agent/robot/events/integrations/weixin/syncbuf.go new file mode 100644 index 00000000..e8faba55 --- /dev/null +++ b/agent/robot/events/integrations/weixin/syncbuf.go @@ -0,0 +1,44 @@ +package weixin + +import ( + "encoding/json" + "os" + "path/filepath" + + "github.com/yaoapp/gou/application" +) + +type syncBufData struct { + GetUpdatesBuf string `json:"get_updates_buf"` +} + +func syncBufPath(accountID string) string { + root := application.App.Root() + return filepath.Join(root, "data", "weixin", accountID+".sync.json") +} + +func loadSyncBuf(accountID string) string { + p := syncBufPath(accountID) + data, err := os.ReadFile(p) + if err != nil { + return "" + } + var buf syncBufData + if err := json.Unmarshal(data, &buf); err != nil { + return "" + } + return buf.GetUpdatesBuf +} + +func saveSyncBuf(accountID, syncBuf string) { + p := syncBufPath(accountID) + dir := filepath.Dir(p) + if err := os.MkdirAll(dir, 0755); err != nil { + log.Error("weixin: mkdir for syncbuf: %v", err) + return + } + data, _ := json.Marshal(syncBufData{GetUpdatesBuf: syncBuf}) + if err := os.WriteFile(p, data, 0644); err != nil { + log.Error("weixin: write syncbuf: %v", err) + } +} diff --git a/agent/robot/events/integrations/weixin/typing_cache.go b/agent/robot/events/integrations/weixin/typing_cache.go new file mode 100644 index 00000000..8a66c90b --- /dev/null +++ b/agent/robot/events/integrations/weixin/typing_cache.go @@ -0,0 +1,43 @@ +package weixin + +import ( + "sync" + "time" +) + +const ticketTTL = 20 * time.Hour + +type ticketEntry struct { + ticket string + expiresAt time.Time +} + +type typingTicketCache struct { + mu sync.RWMutex + items map[string]*ticketEntry +} + +func newTypingTicketCache() *typingTicketCache { + return &typingTicketCache{ + items: make(map[string]*ticketEntry), + } +} + +func (c *typingTicketCache) Get(userID string) string { + c.mu.RLock() + defer c.mu.RUnlock() + entry, ok := c.items[userID] + if !ok || time.Now().After(entry.expiresAt) { + return "" + } + return entry.ticket +} + +func (c *typingTicketCache) Set(userID, ticket string) { + c.mu.Lock() + defer c.mu.Unlock() + c.items[userID] = &ticketEntry{ + ticket: ticket, + expiresAt: time.Now().Add(ticketTTL), + } +} diff --git a/agent/robot/events/integrations/weixin/weixin.go b/agent/robot/events/integrations/weixin/weixin.go new file mode 100644 index 00000000..2a2fdaba --- /dev/null +++ b/agent/robot/events/integrations/weixin/weixin.go @@ -0,0 +1,130 @@ +package weixin + +import ( + "context" + "sync" + + "github.com/yaoapp/yao/agent/robot/logger" + robottypes "github.com/yaoapp/yao/agent/robot/types" + weixinapi "github.com/yaoapp/yao/integrations/weixin" +) + +var log = logger.New("weixin") + +type Adapter struct { + mu sync.RWMutex + bots map[string]*botEntry + accountIdx map[string]string // accountID(ilink_bot_id) -> robotID + dedup *dedupStore + stopCh chan struct{} +} + +type botEntry struct { + robotID string + accountID string + bot *weixinapi.Bot + cancelFn context.CancelFunc + ticketCache *typingTicketCache +} + +func NewAdapter() *Adapter { + a := &Adapter{ + bots: make(map[string]*botEntry), + accountIdx: make(map[string]string), + dedup: newDedupStore(), + stopCh: make(chan struct{}), + } + go a.dedup.cleaner(a.stopCh) + return a +} + +func (a *Adapter) Apply(ctx context.Context, robot *robottypes.Robot) { + conf := extractConfig(robot) + if conf == nil || !conf.Enabled || conf.BotToken == "" { + a.removeBot(robot.MemberID) + return + } + + a.mu.Lock() + defer a.mu.Unlock() + + if existing, ok := a.bots[robot.MemberID]; ok { + if existing.bot.Token() == conf.BotToken && + existing.bot.BaseURL() == resolveBaseURL(conf) && + existing.bot.CDNBaseURL() == resolveCDNBaseURL(conf) { + return + } + a.removeBotLocked(robot.MemberID) + } + + pollCtx, cancel := context.WithCancel(context.Background()) + entry := &botEntry{ + robotID: robot.MemberID, + accountID: conf.AccountID, + bot: weixinapi.NewBot(conf.BotToken, resolveBaseURL(conf), resolveCDNBaseURL(conf)), + cancelFn: cancel, + ticketCache: newTypingTicketCache(), + } + a.bots[robot.MemberID] = entry + if conf.AccountID != "" { + a.accountIdx[conf.AccountID] = robot.MemberID + } + go a.pollLoop(pollCtx, entry) + + log.Info("weixin adapter: registered robot=%s accountID=%s", robot.MemberID, conf.AccountID) +} + +func (a *Adapter) Remove(ctx context.Context, robotID string) { + a.removeBot(robotID) +} + +func (a *Adapter) Shutdown() { + close(a.stopCh) + a.mu.Lock() + defer a.mu.Unlock() + for id := range a.bots { + a.removeBotLocked(id) + } +} + +func (a *Adapter) removeBot(robotID string) { + a.mu.Lock() + defer a.mu.Unlock() + a.removeBotLocked(robotID) +} + +func (a *Adapter) removeBotLocked(robotID string) { + entry, ok := a.bots[robotID] + if !ok { + return + } + entry.cancelFn() + if entry.accountID != "" { + delete(a.accountIdx, entry.accountID) + } + delete(a.bots, robotID) +} + +func resolveBaseURL(conf *robottypes.WeixinConfig) string { + if conf.APIHost != "" { + return conf.APIHost + } + if conf.BaseURL != "" { + return conf.BaseURL + } + return weixinapi.DefaultBaseURL() +} + +func resolveCDNBaseURL(conf *robottypes.WeixinConfig) string { + if conf.CDNBaseURL != "" { + return conf.CDNBaseURL + } + return weixinapi.DefaultCDNBaseURL() +} + +func extractConfig(robot *robottypes.Robot) *robottypes.WeixinConfig { + if robot.Config == nil || robot.Config.Integrations == nil { + return nil + } + return robot.Config.Integrations.Weixin +} diff --git a/agent/robot/robot.go b/agent/robot/robot.go index add079a0..8e1a8faa 100644 --- a/agent/robot/robot.go +++ b/agent/robot/robot.go @@ -1,70 +1 @@ package robot - -import ( - "context" - - "github.com/yaoapp/yao/agent/robot/cache" - "github.com/yaoapp/yao/agent/robot/dedup" - "github.com/yaoapp/yao/agent/robot/events/integrations" - "github.com/yaoapp/yao/agent/robot/events/integrations/telegram" - "github.com/yaoapp/yao/agent/robot/executor" - "github.com/yaoapp/yao/agent/robot/logger" - "github.com/yaoapp/yao/agent/robot/manager" - "github.com/yaoapp/yao/agent/robot/plan" - "github.com/yaoapp/yao/agent/robot/pool" - "github.com/yaoapp/yao/agent/robot/store" - robottypes "github.com/yaoapp/yao/agent/robot/types" -) - -var ( - log = logger.New("robot") - - globalManager *manager.Manager - globalCache *cache.Cache - globalPool *pool.Pool - globalDedup *dedup.Dedup - globalStore *store.Store - globalExecutor executor.Executor - globalPlan *plan.Plan - globalDispatcher *integrations.Dispatcher -) - -// Init initializes the robot agent system -func Init() error { - globalCache = cache.New() - globalDedup = dedup.New() - globalStore = store.New() - globalPool = pool.New() - globalExecutor = executor.New() - globalManager = manager.New() - globalPlan = plan.New() - - // Load robots into cache from database before starting dispatcher - rCtx := robottypes.NewContext(context.Background(), nil) - if err := globalCache.Load(rCtx); err != nil { - log.Warn("robot.Init: cache load failed (will rely on config events): %v", err) - } - - adapters := map[string]integrations.Adapter{ - "telegram": telegram.NewAdapter(), - } - globalDispatcher = integrations.NewDispatcher(globalCache, adapters) - if err := globalDispatcher.Start(context.Background()); err != nil { - return err - } - - return nil -} - -// Shutdown gracefully shuts down the robot agent system -func Shutdown() error { - if globalDispatcher != nil { - globalDispatcher.Stop() - } - return nil -} - -// Manager returns the global manager instance -func Manager() *manager.Manager { - return globalManager -} diff --git a/agent/robot/types/config.go b/agent/robot/types/config.go index 9ad5b52f..585e3193 100644 --- a/agent/robot/types/config.go +++ b/agent/robot/types/config.go @@ -28,6 +28,7 @@ type Integrations struct { Feishu *FeishuConfig `json:"feishu,omitempty"` DingTalk *DingTalkConfig `json:"dingtalk,omitempty"` Discord *DiscordConfig `json:"discord,omitempty"` + Weixin *WeixinConfig `json:"weixin,omitempty"` } // TelegramConfig holds Telegram Bot integration settings. @@ -273,6 +274,16 @@ type MCPConfig struct { Tools []string `json:"tools,omitempty"` // empty = all } +// WeixinConfig holds WeChat iLink Bot integration settings. +type WeixinConfig struct { + Enabled bool `json:"enabled"` + BotToken string `json:"bot_token"` + AccountID string `json:"account_id,omitempty"` // ilink_bot_id + APIHost string `json:"api_host,omitempty"` // custom API host + BaseURL string `json:"base_url,omitempty"` // alias for api_host + CDNBaseURL string `json:"cdn_base_url,omitempty"` // custom CDN base URL +} + // Event - event trigger config type Event struct { Type EventSource `json:"type"` // webhook | database diff --git a/integrations/feishu/message.go b/integrations/feishu/message.go index 76d65827..d70a6425 100644 --- a/integrations/feishu/message.go +++ b/integrations/feishu/message.go @@ -12,6 +12,13 @@ import ( "github.com/yaoapp/yao/attachment" ) +// SendTyping is a placeholder for typing indicator support. +// Feishu does not provide a public typing status API; this is a no-op +// so callers can use a uniform interface across all adapters. +func (b *Bot) SendTyping(ctx context.Context, chatID string) error { + return nil +} + // SendTextMessage sends a text message to a chat. func (b *Bot) SendTextMessage(ctx context.Context, chatID, text string) (string, error) { content, _ := json.Marshal(map[string]string{"text": text}) diff --git a/integrations/telegram/message.go b/integrations/telegram/message.go index c285eacc..fd75eaff 100644 --- a/integrations/telegram/message.go +++ b/integrations/telegram/message.go @@ -11,6 +11,19 @@ import ( "github.com/yaoapp/yao/attachment" ) +// SendTyping sends a "typing" chat action to indicate the bot is preparing a response. +func (b *Bot) SendTyping(ctx context.Context, chatID int64) error { + sdk, err := b.sdk() + if err != nil { + return err + } + _, err = sdk.SendChatAction(ctx, &bot.SendChatActionParams{ + ChatID: chatID, + Action: models.ChatActionTyping, + }) + return err +} + // SendMessage sends a message to a chat. If the text contains Markdown formatting, // it is automatically converted to Telegram-compatible HTML. func (b *Bot) SendMessage(ctx context.Context, chatID int64, text string, replyTo int64) error { diff --git a/integrations/weixin/bot.go b/integrations/weixin/bot.go new file mode 100644 index 00000000..680b4170 --- /dev/null +++ b/integrations/weixin/bot.go @@ -0,0 +1,376 @@ +package weixin + +import ( + "bytes" + "context" + "crypto/md5" + "crypto/rand" + "encoding/base64" + "encoding/binary" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strconv" + "strings" + "time" + + "github.com/yaoapp/kun/log" +) + +const defaultBaseURL = "https://ilinkai.weixin.qq.com" +const defaultCDNBaseURL = "https://novac2c.cdn.weixin.qq.com/c2c" +const channelVersion = "1.0.0" + +const ( + UploadMediaImage = 1 + UploadMediaVideo = 2 + UploadMediaFile = 3 + UploadMediaVoice = 4 +) + +const cdnUploadMaxRetries = 3 + +type Bot struct { + token string + baseURL string + cdnBaseURL string + httpClient *http.Client +} + +func NewBot(token, baseURL, cdnBaseURL string) *Bot { + if baseURL == "" { + baseURL = defaultBaseURL + } + if cdnBaseURL == "" { + cdnBaseURL = defaultCDNBaseURL + } + return &Bot{ + token: token, + baseURL: strings.TrimRight(baseURL, "/"), + cdnBaseURL: strings.TrimRight(cdnBaseURL, "/"), + httpClient: &http.Client{Timeout: 60 * time.Second}, + } +} + +func (b *Bot) Token() string { return b.token } +func (b *Bot) BaseURL() string { return b.baseURL } +func (b *Bot) CDNBaseURL() string { return b.cdnBaseURL } +func DefaultBaseURL() string { return defaultBaseURL } +func DefaultCDNBaseURL() string { return defaultCDNBaseURL } + +func (b *Bot) GetUpdates(ctx context.Context, syncBuf string, timeoutMs int) (*GetUpdatesResp, error) { + body, _ := json.Marshal(map[string]interface{}{ + "get_updates_buf": syncBuf, + "base_info": BaseInfo{ChannelVersion: channelVersion}, + }) + + reqCtx, cancel := context.WithTimeout(ctx, time.Duration(timeoutMs+5000)*time.Millisecond) + defer cancel() + + raw, err := b.post(reqCtx, "ilink/bot/getupdates", body) + if err != nil { + if reqCtx.Err() != nil { + return &GetUpdatesResp{GetUpdatesBuf: syncBuf}, nil + } + return nil, err + } + + var resp GetUpdatesResp + if err := json.Unmarshal(raw, &resp); err != nil { + return nil, fmt.Errorf("weixin GetUpdates unmarshal: %w", err) + } + return &resp, nil +} + +func (b *Bot) SendMessage(ctx context.Context, toUserID, contextToken, text string) error { + if contextToken == "" { + return fmt.Errorf("weixin SendMessage: contextToken is required for to=%s", toUserID) + } + clientID := randomClientID() + req := map[string]interface{}{ + "msg": map[string]interface{}{ + "from_user_id": "", + "to_user_id": toUserID, + "client_id": clientID, + "message_type": MessageTypeBot, + "message_state": MessageStateFinish, + "context_token": contextToken, + "item_list": []map[string]interface{}{ + { + "type": ItemTypeText, + "text_item": map[string]string{"text": text}, + }, + }, + }, + "base_info": BaseInfo{ChannelVersion: channelVersion}, + } + body, _ := json.Marshal(req) + _, err := b.post(ctx, "ilink/bot/sendmessage", body) + return err +} + +func (b *Bot) SendImageMessage(ctx context.Context, toUserID, contextToken string, uploaded *UploadedFileInfo) error { + return b.sendMediaMessage(ctx, toUserID, contextToken, MsgItem{ + Type: ItemTypeImage, + ImageItem: &ImageItem{ + Media: &CDNMedia{ + EncryptQueryParam: uploaded.DownloadParam, + AesKey: base64.StdEncoding.EncodeToString([]byte(uploaded.AesKeyHex)), + EncryptType: 1, + }, + MidSize: uploaded.FileSizeCiphertext, + }, + }) +} + +func (b *Bot) SendVideoMessage(ctx context.Context, toUserID, contextToken string, uploaded *UploadedFileInfo) error { + return b.sendMediaMessage(ctx, toUserID, contextToken, MsgItem{ + Type: ItemTypeVideo, + VideoItem: &VideoItem{ + Media: &CDNMedia{ + EncryptQueryParam: uploaded.DownloadParam, + AesKey: base64.StdEncoding.EncodeToString([]byte(uploaded.AesKeyHex)), + EncryptType: 1, + }, + VideoSize: uploaded.FileSizeCiphertext, + }, + }) +} + +func (b *Bot) SendFileMessage(ctx context.Context, toUserID, contextToken, fileName string, uploaded *UploadedFileInfo) error { + return b.sendMediaMessage(ctx, toUserID, contextToken, MsgItem{ + Type: ItemTypeFile, + FileItem: &FileItem{ + FileName: fileName, + Media: &CDNMedia{ + EncryptQueryParam: uploaded.DownloadParam, + AesKey: base64.StdEncoding.EncodeToString([]byte(uploaded.AesKeyHex)), + EncryptType: 1, + }, + Len: strconv.Itoa(uploaded.FileSize), + }, + }) +} + +// SendVoiceMessage sends a voice message with a bubble UI. +// TODO(weixin-voice): The voice bubble displays correctly (with playtime) but +// audio playback does not work — the WeChat client reports "message still +// downloading". This affects all formats tested (SILK, Speex, OGG, MP3) and +// even echoing back an inbound voice's CDN reference verbatim. The iLink Bot +// API likely does not yet fully support outbound voice playback. For now, +// callers should fall back to SendFileMessage for audio attachments until +// WeChat officially supports voice playback via iLink Bot. +func (b *Bot) SendVoiceMessage(ctx context.Context, toUserID, contextToken string, uploaded *UploadedFileInfo, playtimeMs, sampleRate int) error { + item := MsgItem{ + Type: ItemTypeVoice, + VoiceItem: &VoiceItem{ + Media: &CDNMedia{ + EncryptQueryParam: uploaded.DownloadParam, + AesKey: base64.StdEncoding.EncodeToString([]byte(uploaded.AesKeyHex)), + }, + PlayTime: playtimeMs, + SampleRate: sampleRate, + }, + } + return b.sendMediaMessage(ctx, toUserID, contextToken, item) +} + +func (b *Bot) sendMediaMessage(ctx context.Context, toUserID, contextToken string, item MsgItem) error { + if contextToken == "" { + return fmt.Errorf("weixin sendMediaMessage: contextToken is required for to=%s", toUserID) + } + clientID := randomClientID() + req := map[string]interface{}{ + "msg": map[string]interface{}{ + "from_user_id": "", + "to_user_id": toUserID, + "client_id": clientID, + "message_type": MessageTypeBot, + "message_state": MessageStateFinish, + "context_token": contextToken, + "item_list": []MsgItem{item}, + }, + "base_info": BaseInfo{ChannelVersion: channelVersion}, + } + body, _ := json.Marshal(req) + _, err := b.post(ctx, "ilink/bot/sendmessage", body) + return err +} + +func (b *Bot) UploadMedia(ctx context.Context, plaintext []byte, toUserID string, mediaType int) (*UploadedFileInfo, error) { + rawsize := len(plaintext) + hash := md5.Sum(plaintext) + rawfilemd5 := hex.EncodeToString(hash[:]) + filesize := aesEcbPaddedSize(rawsize) + + var filekeyBuf [16]byte + rand.Read(filekeyBuf[:]) + filekey := hex.EncodeToString(filekeyBuf[:]) + + var aeskeyBuf [16]byte + rand.Read(aeskeyBuf[:]) + aeskeyHex := hex.EncodeToString(aeskeyBuf[:]) + + uploadReq, _ := json.Marshal(map[string]interface{}{ + "filekey": filekey, + "media_type": mediaType, + "to_user_id": toUserID, + "rawsize": rawsize, + "rawfilemd5": rawfilemd5, + "filesize": filesize, + "no_need_thumb": true, + "aeskey": aeskeyHex, + "base_info": BaseInfo{ChannelVersion: channelVersion}, + }) + + log.Info("[weixin:upload] getuploadurl request: media_type=%d to_user_id=%s rawsize=%d filesize=%d filekey=%s md5=%s", + mediaType, toUserID, rawsize, filesize, filekey, rawfilemd5) + + raw, err := b.post(ctx, "ilink/bot/getuploadurl", uploadReq) + if err != nil { + return nil, fmt.Errorf("getUploadUrl: %w", err) + } + var uploadResp GetUploadUrlResp + if err := json.Unmarshal(raw, &uploadResp); err != nil { + return nil, fmt.Errorf("getUploadUrl unmarshal: %w (body: %s)", err, string(raw)) + } + + log.Info("[weixin:upload] getuploadurl response: ret=%d errcode=%d errmsg=%q upload_param_len=%d", + uploadResp.Ret, uploadResp.ErrCode, uploadResp.ErrMsg, len(uploadResp.UploadParam)) + + if uploadResp.Ret != 0 || uploadResp.ErrCode != 0 { + return nil, fmt.Errorf("getUploadUrl: ret=%d errcode=%d errmsg=%q media_type=%d to_user_id=%s rawsize=%d filesize=%d rawfilemd5=%s", + uploadResp.Ret, uploadResp.ErrCode, uploadResp.ErrMsg, mediaType, toUserID, rawsize, filesize, rawfilemd5) + } + if uploadResp.UploadParam == "" { + return nil, fmt.Errorf("getUploadUrl: empty upload_param (body: %s)", string(raw)) + } + + ciphertext := encryptAES128ECB(plaintext, aeskeyBuf[:]) + log.Info("[weixin:upload] CDN uploading: ciphertext_len=%d filekey=%s", len(ciphertext), filekey) + + downloadParam, err := b.uploadBufferToCDN(ctx, ciphertext, uploadResp.UploadParam, filekey) + if err != nil { + return nil, fmt.Errorf("CDN upload: %w", err) + } + log.Info("[weixin:upload] CDN success: download_param_len=%d", len(downloadParam)) + + return &UploadedFileInfo{ + Filekey: filekey, + DownloadParam: downloadParam, + AesKeyHex: aeskeyHex, + FileSize: rawsize, + FileSizeCiphertext: filesize, + }, nil +} + +func (b *Bot) uploadBufferToCDN(ctx context.Context, ciphertext []byte, uploadParam, filekey string) (string, error) { + cdnURL := b.cdnBaseURL + "/upload?encrypted_query_param=" + + url.QueryEscape(uploadParam) + "&filekey=" + url.QueryEscape(filekey) + + var lastErr error + for attempt := 1; attempt <= cdnUploadMaxRetries; attempt++ { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, cdnURL, bytes.NewReader(ciphertext)) + if err != nil { + return "", err + } + req.Header.Set("Content-Type", "application/octet-stream") + + resp, err := b.httpClient.Do(req) + if err != nil { + lastErr = err + continue + } + + if resp.StatusCode >= 400 && resp.StatusCode < 500 { + resp.Body.Close() + return "", fmt.Errorf("CDN upload client error %d", resp.StatusCode) + } + if resp.StatusCode != http.StatusOK { + resp.Body.Close() + lastErr = fmt.Errorf("CDN upload server error %d", resp.StatusCode) + continue + } + + downloadParam := resp.Header.Get("x-encrypted-param") + resp.Body.Close() + if downloadParam == "" { + lastErr = fmt.Errorf("CDN response missing x-encrypted-param") + continue + } + return downloadParam, nil + } + return "", fmt.Errorf("CDN upload failed after %d attempts: %w", cdnUploadMaxRetries, lastErr) +} + +func randomClientID() string { + var buf [8]byte + rand.Read(buf[:]) + return fmt.Sprintf("yao-weixin-%x", buf[:]) +} + +func (b *Bot) SendTyping(ctx context.Context, toUserID, typingTicket string, status int) error { + body, _ := json.Marshal(map[string]interface{}{ + "ilink_user_id": toUserID, + "typing_ticket": typingTicket, + "status": status, + "base_info": BaseInfo{ChannelVersion: channelVersion}, + }) + _, err := b.post(ctx, "ilink/bot/sendtyping", body) + return err +} + +func (b *Bot) GetConfig(ctx context.Context, ilinkUserID, contextToken string) (string, error) { + body, _ := json.Marshal(map[string]interface{}{ + "ilink_user_id": ilinkUserID, + "context_token": contextToken, + "base_info": BaseInfo{ChannelVersion: channelVersion}, + }) + raw, err := b.post(ctx, "ilink/bot/getconfig", body) + if err != nil { + return "", err + } + var resp GetConfigResp + if err := json.Unmarshal(raw, &resp); err != nil { + return "", fmt.Errorf("weixin GetConfig unmarshal: %w", err) + } + return resp.TypingTicket, nil +} + +func (b *Bot) post(ctx context.Context, endpoint string, body []byte) ([]byte, error) { + reqURL := b.baseURL + "/" + endpoint + req, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL, bytes.NewReader(body)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("AuthorizationType", HeaderAuthVal) + req.Header.Set("Authorization", "Bearer "+b.token) + req.Header.Set("Content-Length", strconv.Itoa(len(body))) + req.Header.Set("X-WECHAT-UIN", randomWechatUin()) + + resp, err := b.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("weixin %s: %w", endpoint, err) + } + defer resp.Body.Close() + + raw, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("weixin %s read body: %w", endpoint, err) + } + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("weixin %s HTTP %d: %s", endpoint, resp.StatusCode, string(raw)) + } + return raw, nil +} + +func randomWechatUin() string { + var buf [4]byte + rand.Read(buf[:]) + n := binary.BigEndian.Uint32(buf[:]) + return base64.StdEncoding.EncodeToString([]byte(strconv.FormatUint(uint64(n), 10))) +} diff --git a/integrations/weixin/cdn.go b/integrations/weixin/cdn.go new file mode 100644 index 00000000..3bcc392f --- /dev/null +++ b/integrations/weixin/cdn.go @@ -0,0 +1,126 @@ +package weixin + +import ( + "crypto/aes" + "encoding/base64" + "fmt" + "io" + "net/http" + "net/url" +) + +func buildCDNDownloadURL(cdnBaseURL, encryptedQueryParam string) string { + return cdnBaseURL + "/download?encrypted_query_param=" + url.QueryEscape(encryptedQueryParam) +} + +func parseAesKey(aesKeyBase64 string) ([]byte, error) { + decoded, err := base64.StdEncoding.DecodeString(aesKeyBase64) + if err != nil { + return nil, fmt.Errorf("parseAesKey: base64 decode: %w", err) + } + if len(decoded) == 16 { + return decoded, nil + } + if len(decoded) == 32 { + hexStr := string(decoded) + raw := make([]byte, 16) + for i := 0; i < 16; i++ { + var b byte + _, err := fmt.Sscanf(hexStr[i*2:i*2+2], "%02x", &b) + if err != nil { + return nil, fmt.Errorf("parseAesKey: hex parse: %w", err) + } + raw[i] = b + } + return raw, nil + } + return nil, fmt.Errorf("parseAesKey: unexpected decoded len=%d", len(decoded)) +} + +func decryptAES128ECB(ciphertext, key []byte) ([]byte, error) { + block, err := aes.NewCipher(key) + if err != nil { + return nil, err + } + bs := block.BlockSize() + if len(ciphertext)%bs != 0 { + return nil, fmt.Errorf("decryptAES128ECB: ciphertext len %d not multiple of block size", len(ciphertext)) + } + dst := make([]byte, len(ciphertext)) + for i := 0; i < len(ciphertext); i += bs { + block.Decrypt(dst[i:i+bs], ciphertext[i:i+bs]) + } + if len(dst) == 0 { + return dst, nil + } + padLen := int(dst[len(dst)-1]) + if padLen == 0 || padLen > bs { + return nil, fmt.Errorf("decryptAES128ECB: invalid PKCS7 padding %d", padLen) + } + return dst[:len(dst)-padLen], nil +} + +func downloadCDNBytes(cdnBaseURL, encryptedQueryParam string) ([]byte, error) { + u := buildCDNDownloadURL(cdnBaseURL, encryptedQueryParam) + resp, err := http.Get(u) //nolint:gosec + if err != nil { + return nil, fmt.Errorf("CDN download: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("CDN download HTTP %d", resp.StatusCode) + } + return io.ReadAll(resp.Body) +} + +func DownloadAndDecrypt(cdnBaseURL, encryptedQueryParam, aesKeyBase64 string) ([]byte, error) { + key, err := parseAesKey(aesKeyBase64) + if err != nil { + return nil, err + } + data, err := downloadCDNBytes(cdnBaseURL, encryptedQueryParam) + if err != nil { + return nil, err + } + return decryptAES128ECB(data, key) +} + +func DecryptFromRaw(cdnBaseURL, encryptedQueryParam string, rawKey []byte) ([]byte, error) { + data, err := downloadCDNBytes(cdnBaseURL, encryptedQueryParam) + if err != nil { + return nil, err + } + return decryptAES128ECB(data, rawKey) +} + +func DownloadPlain(cdnBaseURL, encryptedQueryParam string) ([]byte, error) { + return downloadCDNBytes(cdnBaseURL, encryptedQueryParam) +} + +func encryptAES128ECB(plaintext, key []byte) []byte { + block, err := aes.NewCipher(key) + if err != nil { + return nil + } + bs := block.BlockSize() + padLen := bs - (len(plaintext) % bs) + padded := make([]byte, len(plaintext)+padLen) + copy(padded, plaintext) + for i := len(plaintext); i < len(padded); i++ { + padded[i] = byte(padLen) + } + dst := make([]byte, len(padded)) + for i := 0; i < len(padded); i += bs { + block.Encrypt(dst[i:i+bs], padded[i:i+bs]) + } + return dst +} + +func aesEcbPaddedSize(plaintextSize int) int { + return ((plaintextSize + 1 + 15) / 16) * 16 +} + +func buildCDNUploadURL(cdnBaseURL, uploadParam, filekey string) string { + return cdnBaseURL + "/upload?encrypted_query_param=" + + url.QueryEscape(uploadParam) + "&filekey=" + url.QueryEscape(filekey) +} diff --git a/integrations/weixin/format.go b/integrations/weixin/format.go new file mode 100644 index 00000000..e4d45317 --- /dev/null +++ b/integrations/weixin/format.go @@ -0,0 +1,226 @@ +package weixin + +import ( + "regexp" + "strings" + "unicode/utf8" +) + +// FormatWeixinText converts standard Markdown to plain text suitable for +// WeChat iLink Bot's text_item. WeChat renders only plain text with clickable +// URLs and [text](url) style links. All other Markdown/HTML is stripped and +// gracefully degraded to readable plain text. +func FormatWeixinText(md string) string { + md = strings.ReplaceAll(md, "\r\n", "\n") + + var out strings.Builder + lines := strings.Split(md, "\n") + + inCodeBlock := false + var codeLines []string + + inTable := false + var tableRows [][]string + + for i := 0; i < len(lines); i++ { + line := lines[i] + + if strings.HasPrefix(line, "```") { + if !inCodeBlock { + inCodeBlock = true + codeLines = nil + } else { + inCodeBlock = false + for _, cl := range codeLines { + out.WriteString(" " + cl + "\n") + } + } + continue + } + if inCodeBlock { + codeLines = append(codeLines, line) + continue + } + + if wxIsTableRow(line) { + if !inTable { + inTable = true + tableRows = nil + } + if wxIsTableSep(line) { + continue + } + tableRows = append(tableRows, wxParseTableRow(line)) + continue + } + if inTable { + wxFlushTable(&out, tableRows) + inTable = false + tableRows = nil + } + + if line == "---" || line == "***" || line == "___" { + out.WriteString("——————\n") + continue + } + + if m := wxReHeading.FindStringSubmatch(line); m != nil { + out.WriteString("【" + wxFormatInline(m[2]) + "】\n") + continue + } + + if m := wxReBlockquote.FindStringSubmatch(line); m != nil { + out.WriteString("│ " + wxFormatInline(m[1]) + "\n") + continue + } + + if m := wxReUnorderedList.FindStringSubmatch(line); m != nil { + out.WriteString("• " + wxFormatInline(m[1]) + "\n") + continue + } + + if m := wxReOrderedList.FindStringSubmatch(line); m != nil { + out.WriteString(m[1] + ". " + wxFormatInline(m[2]) + "\n") + continue + } + + if m := wxReImage.FindStringSubmatch(line); m != nil { + alt := m[1] + url := m[2] + if alt != "" { + out.WriteString("[" + alt + "](" + url + ")\n") + } else { + out.WriteString(url + "\n") + } + continue + } + + out.WriteString(wxFormatInline(line) + "\n") + } + + if inCodeBlock && len(codeLines) > 0 { + for _, cl := range codeLines { + out.WriteString(" " + cl + "\n") + } + } + if inTable { + wxFlushTable(&out, tableRows) + } + + return strings.TrimRight(out.String(), "\n") +} + +var ( + wxReHeading = regexp.MustCompile(`^(#{1,6})\s+(.+)$`) + wxReBlockquote = regexp.MustCompile(`^>\s*(.*)$`) + wxReUnorderedList = regexp.MustCompile(`^[\s]*[-*+]\s+(.+)$`) + wxReOrderedList = regexp.MustCompile(`^[\s]*(\d+)[.)]\s+(.+)$`) + wxReImage = regexp.MustCompile(`^!\[([^\]]*)\]\(([^)]+)\)$`) + wxReTableRow = regexp.MustCompile(`^\|.*\|$`) + wxReTableSep = regexp.MustCompile(`^\|[\s\-:|]+\|$`) + + wxReBoldItalic = regexp.MustCompile(`\*\*\*(.+?)\*\*\*`) + wxReBold = regexp.MustCompile(`\*\*(.+?)\*\*`) + wxReBoldAlt = regexp.MustCompile(`__(.+?)__`) + wxReItalic = regexp.MustCompile(`(?:^|[^*])\*([^*]+?)\*(?:[^*]|$)`) + wxReStrikethrough = regexp.MustCompile(`~~(.+?)~~`) + wxReCode = regexp.MustCompile("`([^`]+)`") + wxReLink = regexp.MustCompile(`\[([^\]]+)\]\(([^)]+)\)`) + + wxReHTMLTag = regexp.MustCompile(`<[^>]+>`) +) + +// wxFormatInline strips inline Markdown/HTML formatting, keeping links in +// [text](url) form which WeChat renders as clickable. +func wxFormatInline(s string) string { + s = wxReLink.ReplaceAllString(s, "[$1]($2)") + s = wxReBoldItalic.ReplaceAllString(s, "$1") + s = wxReBold.ReplaceAllString(s, "$1") + s = wxReBoldAlt.ReplaceAllString(s, "$1") + s = wxReStrikethrough.ReplaceAllString(s, "$1") + s = wxReCode.ReplaceAllString(s, "$1") + s = wxReItalic.ReplaceAllStringFunc(s, func(match string) string { + m := wxReItalic.FindStringSubmatch(match) + if len(m) < 2 { + return match + } + prefix := "" + suffix := "" + if len(match) > 0 && match[0] != '*' { + prefix = string(match[0]) + } + if len(match) > 0 && match[len(match)-1] != '*' { + suffix = string(match[len(match)-1]) + } + return prefix + m[1] + suffix + }) + s = wxReHTMLTag.ReplaceAllString(s, "") + return s +} + +func wxIsTableRow(line string) bool { + return wxReTableRow.MatchString(strings.TrimSpace(line)) +} + +func wxIsTableSep(line string) bool { + return wxReTableSep.MatchString(strings.TrimSpace(line)) +} + +func wxParseTableRow(line string) []string { + line = strings.TrimSpace(line) + line = strings.TrimPrefix(line, "|") + line = strings.TrimSuffix(line, "|") + cells := strings.Split(line, "|") + for i := range cells { + cells[i] = strings.TrimSpace(cells[i]) + } + return cells +} + +func wxFlushTable(out *strings.Builder, rows [][]string) { + if len(rows) == 0 { + return + } + colWidths := make([]int, len(rows[0])) + for _, row := range rows { + for i, cell := range row { + if i < len(colWidths) && utf8.RuneCountInString(cell) > colWidths[i] { + colWidths[i] = utf8.RuneCountInString(cell) + } + } + } + for ri, row := range rows { + for ci, cell := range row { + if ci > 0 { + out.WriteString(" | ") + } + w := 0 + if ci < len(colWidths) { + w = colWidths[ci] + } + out.WriteString(wxPadRight(cell, w)) + } + out.WriteString("\n") + if ri == 0 && len(rows) > 1 { + for ci := range row { + if ci > 0 { + out.WriteString("-+-") + } + w := 0 + if ci < len(colWidths) { + w = colWidths[ci] + } + out.WriteString(strings.Repeat("-", w)) + } + out.WriteString("\n") + } + } +} + +func wxPadRight(s string, width int) string { + runes := utf8.RuneCountInString(s) + if runes >= width { + return s + } + return s + strings.Repeat(" ", width-runes) +} diff --git a/integrations/weixin/mime.go b/integrations/weixin/mime.go new file mode 100644 index 00000000..c2588289 --- /dev/null +++ b/integrations/weixin/mime.go @@ -0,0 +1,47 @@ +package weixin + +import ( + "path/filepath" + "strings" +) + +var mimeMap = map[string]string{ + ".jpg": "image/jpeg", + ".jpeg": "image/jpeg", + ".png": "image/png", + ".gif": "image/gif", + ".webp": "image/webp", + ".bmp": "image/bmp", + ".svg": "image/svg+xml", + ".mp3": "audio/mpeg", + ".wav": "audio/wav", + ".ogg": "audio/ogg", + ".silk": "audio/silk", + ".amr": "audio/amr", + ".mp4": "video/mp4", + ".mov": "video/quicktime", + ".avi": "video/x-msvideo", + ".webm": "video/webm", + ".pdf": "application/pdf", + ".doc": "application/msword", + ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + ".xls": "application/vnd.ms-excel", + ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + ".ppt": "application/vnd.ms-powerpoint", + ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation", + ".zip": "application/zip", + ".gz": "application/gzip", + ".tar": "application/x-tar", + ".txt": "text/plain", + ".csv": "text/csv", + ".json": "application/json", + ".xml": "application/xml", +} + +func MimeFromFilename(filename string) string { + ext := strings.ToLower(filepath.Ext(filename)) + if m, ok := mimeMap[ext]; ok { + return m + } + return "application/octet-stream" +} diff --git a/integrations/weixin/qrcode.go b/integrations/weixin/qrcode.go new file mode 100644 index 00000000..c4269a57 --- /dev/null +++ b/integrations/weixin/qrcode.go @@ -0,0 +1,75 @@ +package weixin + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" +) + +const DefaultBotType = "3" + +func GetQRCode(ctx context.Context, apiHost string) (qrcode, qrcodeImgURL string, err error) { + if apiHost == "" { + apiHost = defaultBaseURL + } + u := strings.TrimRight(apiHost, "/") + "/ilink/bot/get_bot_qrcode?bot_type=" + DefaultBotType + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) + if err != nil { + return "", "", err + } + + client := &http.Client{Timeout: 15 * time.Second} + resp, err := client.Do(req) + if err != nil { + return "", "", fmt.Errorf("GetQRCode: %w", err) + } + defer resp.Body.Close() + + raw, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK { + return "", "", fmt.Errorf("GetQRCode HTTP %d: %s", resp.StatusCode, string(raw)) + } + + var r QRCodeResp + if err := json.Unmarshal(raw, &r); err != nil { + return "", "", fmt.Errorf("GetQRCode unmarshal: %w", err) + } + return r.QRCode, r.QRCodeImgContent, nil +} + +func PollQRStatus(ctx context.Context, apiHost, qrcode string) (*QRStatusResp, error) { + if apiHost == "" { + apiHost = defaultBaseURL + } + u := fmt.Sprintf("%s/ilink/bot/get_qrcode_status?qrcode=%s", + strings.TrimRight(apiHost, "/"), qrcode) + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) + if err != nil { + return nil, err + } + req.Header.Set("iLink-App-ClientVersion", "1") + + client := &http.Client{Timeout: 35 * time.Second} + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("PollQRStatus: %w", err) + } + defer resp.Body.Close() + + raw, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("PollQRStatus HTTP %d: %s", resp.StatusCode, string(raw)) + } + + var r QRStatusResp + if err := json.Unmarshal(raw, &r); err != nil { + return nil, fmt.Errorf("PollQRStatus unmarshal: %w", err) + } + return &r, nil +} diff --git a/integrations/weixin/types.go b/integrations/weixin/types.go new file mode 100644 index 00000000..e47f4b80 --- /dev/null +++ b/integrations/weixin/types.go @@ -0,0 +1,163 @@ +package weixin + +const ( + HeaderAuthType = "AuthorizationType" + HeaderAuthVal = "ilink_bot_token" +) + +const SessionExpiredErrCode = -14 + +const ( + ItemTypeText = 1 + ItemTypeImage = 2 + ItemTypeVoice = 3 + ItemTypeFile = 4 + ItemTypeVideo = 5 +) + +const ( + MessageTypeNone = 0 + MessageTypeUser = 1 + MessageTypeBot = 2 +) + +const ( + MessageStateNew = 0 + MessageStateGenerating = 1 + MessageStateFinish = 2 +) + +const ( + TypingStatusTyping = 1 + TypingStatusCancel = 2 +) + +type BaseInfo struct { + ChannelVersion string `json:"channel_version,omitempty"` +} + +type TextItem struct { + Text string `json:"text,omitempty"` +} + +type CDNMedia struct { + EncryptQueryParam string `json:"encrypt_query_param"` + AesKey string `json:"aes_key"` + EncryptType int `json:"encrypt_type,omitempty"` +} + +type UploadedFileInfo struct { + Filekey string `json:"filekey"` + DownloadParam string `json:"download_encrypted_query_param"` + AesKeyHex string `json:"aeskey"` + FileSize int `json:"file_size"` + FileSizeCiphertext int `json:"file_size_ciphertext"` +} + +type GetUploadUrlResp struct { + Ret int `json:"ret"` + ErrCode int `json:"errcode"` + ErrMsg string `json:"errmsg,omitempty"` + UploadParam string `json:"upload_param"` + ThumbUploadParam string `json:"thumb_upload_param,omitempty"` +} + +type ImageItem struct { + AesKey string `json:"aeskey,omitempty"` + Media *CDNMedia `json:"media,omitempty"` + ThumbMedia *CDNMedia `json:"thumb_media,omitempty"` + MidSize int `json:"mid_size,omitempty"` + HdSize int `json:"hd_size,omitempty"` +} + +type VoiceItem struct { + Media *CDNMedia `json:"media,omitempty"` + EncodeType int `json:"encode_type,omitempty"` + SampleRate int `json:"sample_rate,omitempty"` + PlayTime int `json:"playtime,omitempty"` + Text string `json:"text,omitempty"` +} + +type FileItem struct { + FileName string `json:"file_name,omitempty"` + Media *CDNMedia `json:"media,omitempty"` + Len string `json:"len,omitempty"` +} + +type VideoItem struct { + Media *CDNMedia `json:"media,omitempty"` + ThumbMedia *CDNMedia `json:"thumb_media,omitempty"` + VideoSize int `json:"video_size,omitempty"` +} + +type RefMessage struct { + MessageItem *MsgItem `json:"message_item,omitempty"` + Title string `json:"title,omitempty"` +} + +type MsgItem struct { + Type int `json:"type"` + TextItem *TextItem `json:"text_item,omitempty"` + ImageItem *ImageItem `json:"image_item,omitempty"` + VoiceItem *VoiceItem `json:"voice_item,omitempty"` + FileItem *FileItem `json:"file_item,omitempty"` + VideoItem *VideoItem `json:"video_item,omitempty"` + RefMsg *RefMessage `json:"ref_msg,omitempty"` +} + +type WeixinMessage struct { + Seq int64 `json:"seq,omitempty"` + MessageID int64 `json:"message_id,omitempty"` + FromUserID string `json:"from_user_id"` + ToUserID string `json:"to_user_id"` + ClientID string `json:"client_id,omitempty"` + SessionID string `json:"session_id,omitempty"` + GroupID string `json:"group_id,omitempty"` + MessageType int `json:"message_type,omitempty"` + MessageState int `json:"message_state,omitempty"` + ContextToken string `json:"context_token"` + CreateTimeMs int64 `json:"create_time_ms"` + UpdateTimeMs int64 `json:"update_time_ms,omitempty"` + DeleteTimeMs int64 `json:"delete_time_ms,omitempty"` + ItemList []MsgItem `json:"item_list"` +} + +type GetUpdatesResp struct { + Ret int `json:"ret"` + ErrCode int `json:"errcode"` + ErrMsg string `json:"errmsg"` + Msgs []WeixinMessage `json:"msgs"` + GetUpdatesBuf string `json:"get_updates_buf"` + LongPollingTimeoutMs int `json:"longpolling_timeout_ms"` +} + +type SendMessageReq struct { + Msg *WeixinMessage `json:"msg"` + BaseInfo *BaseInfo `json:"base_info,omitempty"` +} + +type SendTypingReq struct { + IlinkUserID string `json:"ilink_user_id"` + TypingTicket string `json:"typing_ticket"` + Status int `json:"status"` + BaseInfo *BaseInfo `json:"base_info,omitempty"` +} + +type GetConfigResp struct { + Ret int `json:"ret"` + ErrMsg string `json:"errmsg"` + TypingTicket string `json:"typing_ticket"` +} + +type QRCodeResp struct { + QRCode string `json:"qrcode"` + QRCodeImgContent string `json:"qrcode_img_content"` +} + +type QRStatusResp struct { + Status string `json:"status"` + BotToken string `json:"bot_token"` + IlinkBotID string `json:"ilink_bot_id"` + BaseURL string `json:"baseurl"` + UserID string `json:"ilink_user_id"` +} diff --git a/openapi/agent/robot/robot.go b/openapi/agent/robot/robot.go index 05de8e57..99ba3c75 100644 --- a/openapi/agent/robot/robot.go +++ b/openapi/agent/robot/robot.go @@ -25,6 +25,10 @@ func Attach(group *gin.RouterGroup, oauth types.OAuth) { // Integration credential verification (must be before /:id to avoid conflict) group.POST("/integrations/verify", VerifyIntegration) // POST /robots/integrations/verify - Verify integration credentials + // WeChat iLink Bot QR code login + group.POST("/integrations/weixin/qrcode", CreateWeixinQRCode) // POST /robots/integrations/weixin/qrcode - Create QR session + group.GET("/integrations/weixin/qrcode/:session_key", PollWeixinQRCode) // GET /robots/integrations/weixin/qrcode/:session_key - Poll QR status + group.GET("/:id", GetRobot) // GET /robots/:id - Get robot details group.PUT("/:id", UpdateRobot) // PUT /robots/:id - Update robot group.DELETE("/:id", DeleteRobot) // DELETE /robots/:id - Delete robot diff --git a/openapi/agent/robot/weixin.go b/openapi/agent/robot/weixin.go new file mode 100644 index 00000000..c1378b74 --- /dev/null +++ b/openapi/agent/robot/weixin.go @@ -0,0 +1,72 @@ +package robot + +import ( + "os" + + "github.com/gin-gonic/gin" + api "github.com/yaoapp/yao/agent/robot/api" + "github.com/yaoapp/yao/openapi/response" +) + +type createWeixinQRCodeRequest struct { + APIHost string `json:"api_host"` +} + +// CreateWeixinQRCode handles POST /robots/integrations/weixin/qrcode +func CreateWeixinQRCode(c *gin.Context) { + var req createWeixinQRCodeRequest + _ = c.ShouldBindJSON(&req) + + apiHost := req.APIHost + if apiHost == "" { + apiHost = os.Getenv("YAO_WEIXIN_API_HOST") + } + if apiHost == "" { + apiHost = "https://ilinkai.weixin.qq.com" + } + + sessionKey, qrcodeURL, qrcodeImg, err := api.WeixinQRCodeCreate(apiHost) + if err != nil { + response.RespondWithError(c, response.StatusInternalServerError, &response.ErrorResponse{ + Code: response.ErrServerError.Code, + ErrorDescription: err.Error(), + }) + return + } + + response.RespondWithSuccess(c, response.StatusOK, gin.H{ + "session_key": sessionKey, + "qrcode_url": qrcodeURL, + "qrcode_img": qrcodeImg, + }) +} + +// PollWeixinQRCode handles GET /robots/integrations/weixin/qrcode/:session_key +func PollWeixinQRCode(c *gin.Context) { + sessionKey := c.Param("session_key") + if sessionKey == "" { + response.RespondWithError(c, response.StatusBadRequest, &response.ErrorResponse{ + Code: response.ErrInvalidRequest.Code, + ErrorDescription: "session_key is required", + }) + return + } + + status, botToken, accountID, baseURL, _, err := api.WeixinQRCodePoll(sessionKey) + if err != nil { + response.RespondWithError(c, response.StatusBadRequest, &response.ErrorResponse{ + Code: response.ErrInvalidRequest.Code, + ErrorDescription: err.Error(), + }) + return + } + + result := gin.H{"status": status} + if status == "confirmed" { + result["bot_token"] = botToken + result["account_id"] = accountID + result["base_url"] = baseURL + } + + response.RespondWithSuccess(c, response.StatusOK, result) +}