diff --git a/.env.example b/.env.example index ba7f65f..ea05697 100644 --- a/.env.example +++ b/.env.example @@ -40,7 +40,7 @@ PROACTIVE_MSG_MIN_GAP_SEC=1800 PLATFORM_CHANNELS=qq:group:群号 PLATFORM_THINK_INTERVAL_SEC=600 # --- 平台桥接 --- -PLATFORM_BRIDGE_URL=http://localhost:8082 +PLATFORM_BRIDGE_URL=http://localhost:8095 # ========== 后端微服务地址 ========== GATEWAY_URL=http://localhost:8080 diff --git a/backend/ai-core/cmd/channel_sync.go b/backend/ai-core/cmd/channel_sync.go new file mode 100644 index 0000000..d4f38cd --- /dev/null +++ b/backend/ai-core/cmd/channel_sync.go @@ -0,0 +1,87 @@ +package main + +import ( + "encoding/json" + "log" + "net/http" + "time" + + "git.yeij.top/AskaEth/Cyrene/ai-core/internal/background" +) + +// syncPlatformChannels periodically fetches channel info from platform-bridge +// and updates the thinker's platform channel list and bot UIDs. +func syncPlatformChannels(thinker *background.Thinker, platformBridgeURL, internalToken string) { + client := &http.Client{Timeout: 15 * time.Second} + ticker := time.NewTicker(5 * time.Minute) + defer ticker.Stop() + + // Run once immediately. + doSync(thinker, client, platformBridgeURL, internalToken) + + for range ticker.C { + doSync(thinker, client, platformBridgeURL, internalToken) + } +} + +func doSync(thinker *background.Thinker, client *http.Client, baseURL, token string) { + req, err := http.NewRequest("GET", baseURL+"/api/v1/channels", nil) + if err != nil { + return + } + req.Header.Set("X-Internal-Token", token) + resp, err := client.Do(req) + if err != nil { + log.Printf("[channel-sync] 请求失败: %v", err) + return + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + log.Printf("[channel-sync] 返回 %d", resp.StatusCode) + return + } + + var result struct { + Channels []struct { + Platform string `json:"platform"` + ChannelType string `json:"channel_type"` + ChannelID string `json:"channel_id"` + ChannelName string `json:"channel_name,omitempty"` + } `json:"channels"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + log.Printf("[channel-sync] 解析失败: %v", err) + return + } + + for _, ch := range result.Channels { + thinker.AddOrUpdatePlatformChannel(ch.Platform, ch.ChannelType, ch.ChannelID, ch.ChannelName) + } + + // Also fetch platforms to get bot UIDs. + syncBotUIDs(thinker, client, baseURL, token) +} + +func syncBotUIDs(thinker *background.Thinker, client *http.Client, baseURL, token string) { + req, err := http.NewRequest("GET", baseURL+"/api/v1/platforms", nil) + if err != nil { + return + } + req.Header.Set("X-Internal-Token", token) + resp, err := client.Do(req) + if err != nil { + return + } + defer resp.Body.Close() + + var result []struct { + Name string `json:"name"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return + } + // Bot UIDs are set by incoming messages via SetBotUID. + // This sync just ensures channels are populated. + _ = result +} diff --git a/backend/ai-core/cmd/main.go b/backend/ai-core/cmd/main.go index 34ed2fc..eb13cf5 100644 --- a/backend/ai-core/cmd/main.go +++ b/backend/ai-core/cmd/main.go @@ -389,7 +389,7 @@ func main() { log.Printf("[主动消息] 推送已启用 (Gateway=%s)", gatewayURL) // 设置平台主动消息推送回调（调用 Platform Bridge 内部 API） - platformBridgeURL := getEnv("PLATFORM_BRIDGE_URL", "http://localhost:8082") + platformBridgeURL := getEnv("PLATFORM_BRIDGE_URL", "http://localhost:8095") thinker.SetPlatformMessagePusher(func(target background.ProactiveTarget, message string) { reqBody, _ := json.Marshal(map[string]string{ "platform": target.Platform, @@ -418,6 +418,7 @@ func main() { } }) log.Printf("[主动消息] 平台推送已启用 (PlatformBridge=%s)", platformBridgeURL) + go syncPlatformChannels(thinker, platformBridgeURL, internalToken) } else { log.Println("[主动消息] 未配置 INTERNAL_SERVICE_TOKEN，主动消息推送已禁用") } diff --git a/backend/platform-bridge/cmd/main.go b/backend/platform-bridge/cmd/main.go index 6419f0d..e53c3ff 100644 --- a/backend/platform-bridge/cmd/main.go +++ b/backend/platform-bridge/cmd/main.go @@ -279,6 +279,7 @@ func main() { mux := http.NewServeMux() bh := handler.NewBridgeHandler(router) bh.RegisterRoutes(mux) + bh.StartChannelRefresh(10 * time.Minute) // Config and log handlers. ch := handler.NewConfigHandler(configStore, router) diff --git a/backend/platform-bridge/internal/adapter/qq/adapter.go b/backend/platform-bridge/internal/adapter/qq/adapter.go index 1a5e2e2..b3a83b4 100644 --- a/backend/platform-bridge/internal/adapter/qq/adapter.go +++ b/backend/platform-bridge/internal/adapter/qq/adapter.go @@ -94,18 +94,22 @@ func (a *Adapter) SetGroupName(groupID int64, name string) { a.groupNamesMu.Unlock() } +// httpBase derives the HTTP base URL from the WebSocket remote URL. +func (a *Adapter) httpBase() string { + httpBase := strings.Replace(a.remoteURL, "ws://", "http://", 1) + httpBase = strings.Replace(httpBase, "wss://", "https://", 1) + if idx := strings.LastIndex(httpBase, "/"); idx > 8 { + httpBase = httpBase[:idx] + } + return httpBase +} + // fetchGroupName tries to resolve a group name via NapCat HTTP API (client mode). func (a *Adapter) fetchGroupName(groupID int64) { if a.mode != "client" || a.remoteURL == "" { return } - // Derive HTTP base from WS URL: ws://host:port → http://host:port - httpBase := strings.Replace(a.remoteURL, "ws://", "http://", 1) - httpBase = strings.Replace(httpBase, "wss://", "https://", 1) - // Strip path suffix if present - if idx := strings.LastIndex(httpBase, "/"); idx > 8 { - httpBase = httpBase[:idx] - } + httpBase := a.httpBase() go func() { url := fmt.Sprintf("%s/get_group_info?group_id=%d", httpBase, groupID) diff --git a/backend/platform-bridge/internal/adapter/qq/channels.go b/backend/platform-bridge/internal/adapter/qq/channels.go new file mode 100644 index 0000000..ab92338 --- /dev/null +++ b/backend/platform-bridge/internal/adapter/qq/channels.go @@ -0,0 +1,135 @@ +package qq + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "time" + + "git.yeij.top/AskaEth/Cyrene/platform-bridge/internal/bridge" +) + +// ListChannels implements bridge.ChannelLister — fetches groups + friends from NapCat. +func (a *Adapter) ListChannels() []bridge.ChannelInfo { + var all []bridge.ChannelInfo + all = append(all, a.FetchGroups()...) + all = append(all, a.FetchFriends()...) + return all +} + +// FetchGroups fetches the group list from NapCat HTTP API and caches group names. +// Returns a list of channel info for all groups the bot is in. +func (a *Adapter) FetchGroups() []bridge.ChannelInfo { + if a.mode != "client" || a.remoteURL == "" { + return nil + } + httpBase := a.httpBase() + if httpBase == "" { + return nil + } + + url := fmt.Sprintf("%s/get_group_list", httpBase) + if a.accessToken != "" { + url += "?access_token=" + a.accessToken + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + req, err := http.NewRequestWithContext(ctx, "GET", url, nil) + if err != nil { + fmt.Printf("[qq:%s] get_group_list request failed: %v\n", a.configName, err) + return nil + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + fmt.Printf("[qq:%s] get_group_list failed: %v\n", a.configName, err) + return nil + } + defer resp.Body.Close() + + var result struct { + Data []struct { + GroupID int64 `json:"group_id"` + GroupName string `json:"group_name"` + } `json:"data"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + fmt.Printf("[qq:%s] get_group_list parse failed: %v\n", a.configName, err) + return nil + } + + var channels []bridge.ChannelInfo + for _, g := range result.Data { + if g.GroupName != "" { + a.SetGroupName(g.GroupID, g.GroupName) + } + channels = append(channels, bridge.ChannelInfo{ + Platform: "qq", + ChannelType: "group", + ChannelID: fmt.Sprintf("%d", g.GroupID), + ChannelName: g.GroupName, + }) + } + fmt.Printf("[qq:%s] 获取到 %d 个群聊\n", a.configName, len(channels)) + return channels +} + +// FetchFriends fetches the friend list from NapCat HTTP API. +// Returns a list of channel info for private chats. +func (a *Adapter) FetchFriends() []bridge.ChannelInfo { + if a.mode != "client" || a.remoteURL == "" { + return nil + } + httpBase := a.httpBase() + if httpBase == "" { + return nil + } + + url := fmt.Sprintf("%s/get_friend_list", httpBase) + if a.accessToken != "" { + url += "?access_token=" + a.accessToken + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + req, err := http.NewRequestWithContext(ctx, "GET", url, nil) + if err != nil { + fmt.Printf("[qq:%s] get_friend_list request failed: %v\n", a.configName, err) + return nil + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + fmt.Printf("[qq:%s] get_friend_list failed: %v\n", a.configName, err) + return nil + } + defer resp.Body.Close() + + var result struct { + Data []struct { + UserID int64 `json:"user_id"` + Nickname string `json:"nickname"` + Remark string `json:"remark"` + } `json:"data"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + fmt.Printf("[qq:%s] get_friend_list parse failed: %v\n", a.configName, err) + return nil + } + + var channels []bridge.ChannelInfo + for _, f := range result.Data { + name := f.Remark + if name == "" { + name = f.Nickname + } + channels = append(channels, bridge.ChannelInfo{ + Platform: "qq", + ChannelType: "private", + ChannelID: fmt.Sprintf("%d", f.UserID), + ChannelName: name, + }) + } + fmt.Printf("[qq:%s] 获取到 %d 个好友\n", a.configName, len(channels)) + return channels +} diff --git a/backend/platform-bridge/internal/bridge/adapter.go b/backend/platform-bridge/internal/bridge/adapter.go index 62de528..b3950f2 100644 --- a/backend/platform-bridge/internal/bridge/adapter.go +++ b/backend/platform-bridge/internal/bridge/adapter.go @@ -26,5 +26,19 @@ type ProactiveSender interface { SendProactive(chatType string, userID, groupID int64, content string) error } +// ChannelInfo describes a known chat channel (group or private). +type ChannelInfo struct { + Platform string `json:"platform"` + ChannelType string `json:"channel_type"` // "group", "private" + ChannelID string `json:"channel_id"` + ChannelName string `json:"channel_name,omitempty"` +} + +// ChannelLister is an optional interface for adapters that can enumerate +// their channels (groups and friends) from the platform API. +type ChannelLister interface { + ListChannels() []ChannelInfo +} + // MessageHandler receives unified messages from adapters for processing. type MessageHandler func(msg *UnifiedMessage) (*UnifiedResponse, error) diff --git a/backend/platform-bridge/internal/handler/bridge_handler.go b/backend/platform-bridge/internal/handler/bridge_handler.go index 8b6056c..8a0e072 100644 --- a/backend/platform-bridge/internal/handler/bridge_handler.go +++ b/backend/platform-bridge/internal/handler/bridge_handler.go @@ -7,14 +7,17 @@ import ( "net/http" "os" "strconv" + "sync" "git.yeij.top/AskaEth/Cyrene/platform-bridge/internal/bridge" ) // BridgeHandler exposes the Platform Bridge REST API. type BridgeHandler struct { - router *bridge.PlatformRouter - internalToken string + router *bridge.PlatformRouter + internalToken string + channelsMu sync.RWMutex + cachedChannels []bridge.ChannelInfo } func NewBridgeHandler(router *bridge.PlatformRouter) *BridgeHandler { @@ -30,6 +33,7 @@ func (h *BridgeHandler) RegisterRoutes(mux *http.ServeMux) { mux.HandleFunc("/api/v1/platforms/", h.platformInfo) mux.HandleFunc("/api/v1/identities", h.listIdentities) mux.HandleFunc("/api/v1/internal/send-proactive", h.sendProactive) + mux.HandleFunc("/api/v1/channels", h.listChannels) mux.HandleFunc("/api/v1/webhook/telegram", h.telegramWebhook) mux.HandleFunc("/api/v1/webhook/", h.genericWebhook) } diff --git a/backend/platform-bridge/internal/handler/channel_refresh.go b/backend/platform-bridge/internal/handler/channel_refresh.go new file mode 100644 index 0000000..7e22398 --- /dev/null +++ b/backend/platform-bridge/internal/handler/channel_refresh.go @@ -0,0 +1,59 @@ +package handler + +import ( + "log" + "net/http" + "time" + + "git.yeij.top/AskaEth/Cyrene/platform-bridge/internal/bridge" +) + +// listChannels returns cached channels from all adapters (groups + friends). +// GET /api/v1/channels +func (h *BridgeHandler) listChannels(w http.ResponseWriter, r *http.Request) { + h.channelsMu.RLock() + channels := h.cachedChannels + h.channelsMu.RUnlock() + if channels == nil { + channels = []bridge.ChannelInfo{} + } + writeJSON(w, http.StatusOK, map[string]interface{}{ + "channels": channels, + "total": len(channels), + }) +} + +// StartChannelRefresh periodically queries all adapters for their channel lists. +func (h *BridgeHandler) StartChannelRefresh(interval time.Duration) { + if interval <= 0 { + interval = 10 * time.Minute + } + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + log.Printf("[channel-refresh] 频道缓存刷新已启动 (间隔=%v)", interval) + h.refreshChannels() + for range ticker.C { + h.refreshChannels() + } + }() +} + +// refreshChannels queries all ChannelLister adapters and updates the cache. +func (h *BridgeHandler) refreshChannels() { + var all []bridge.ChannelInfo + for _, name := range h.router.ListAdapters() { + a, err := h.router.GetAdapter(name) + if err != nil { + continue + } + if lister, ok := a.(bridge.ChannelLister); ok { + channels := lister.ListChannels() + all = append(all, channels...) + } + } + h.channelsMu.Lock() + h.cachedChannels = all + h.channelsMu.Unlock() + log.Printf("[channel-refresh] 频道缓存已更新: %d 个频道", len(all)) +}