fix: platform_silent记忆提取 + 群聊上下文整合 + 多QQ实例支持
- platform_silent模式接入Orchestrator记忆提取：被动观察群聊时提取值得记住的信息到对应命名空间 - post_chat后台思考注入平台观察：对话后思考也能看到群聊摘要 - QQ适配器：OneBot v11 self_id动态捕获、CQ图片URL提取、视觉+OCR并行处理 - Router解耦：ConfigName/PlatformName分离，支持多QQ实例独立连接 - 黑白名单功能：后端API + Ethend代理 + UI面板 - \n\n双换行断句：AI回复按双换行分割为多条消息按间隔发送 - @提及修复：bot自感知UID进行@检测 - 群聊上下文共享：channel-based userID避免记忆碎片化 - 消息日志显示处理后内容而非原始SSE数据 - platform-bridge Dockerfile + docker-compose.yml更新 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
+625
-128
@@ -5,9 +5,12 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
@@ -34,6 +37,13 @@ func main() {
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
// Blocklist settings.
|
||||
blocklistStore, err := config.NewBlocklistStore("platform_blocklist.json")
|
||||
if err != nil {
|
||||
fmt.Printf("FATAL: blocklist store: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
// Message logger.
|
||||
msgLogger, err := logging.NewLogger("logs")
|
||||
if err != nil {
|
||||
@@ -48,7 +58,7 @@ func main() {
|
||||
router := bridge.NewPlatformRouter(mapper, checker)
|
||||
|
||||
// Seed default identities from environment.
|
||||
seedIdentities(mapper)
|
||||
seedIdentities(mapper, configStore)
|
||||
|
||||
// Register platform adapters based on stored configs or defaults.
|
||||
adapters := createAdapters(cfg, configStore)
|
||||
@@ -64,7 +74,7 @@ func main() {
|
||||
Direction: "incoming",
|
||||
Platform: msg.Platform,
|
||||
ChannelID: msg.ChannelID,
|
||||
SenderID: msg.SenderID,
|
||||
SenderID: msg.OriginalSenderUID,
|
||||
SenderName: msg.SenderName,
|
||||
Content: msg.Content,
|
||||
ContentType: msg.ContentType,
|
||||
@@ -72,33 +82,122 @@ func main() {
|
||||
Success: true,
|
||||
})
|
||||
|
||||
response, err := forwardToAICore(cfg, msg)
|
||||
if err != nil {
|
||||
// Routing decisions.
|
||||
isAdmin := mapper.IsAdmin(msg.Platform, msg.OriginalSenderUID)
|
||||
isMentioned, mentionReason := detectAdminMention(msg, mapper, cfg)
|
||||
isBotMentioned := msg.BotUID != "" && containsString(msg.Mentions, msg.BotUID)
|
||||
isSilent := cfg.PlatformSilentEnabled && !isAdmin && !isBotMentioned
|
||||
|
||||
// Blocklist/whitelist check (admin always bypasses).
|
||||
if blocked := blocklistStore.IsBlocked(msg.ChannelType, msg.ChannelID, msg.OriginalSenderUID, isAdmin); blocked {
|
||||
msgLogger.Log(logging.LogEntry{
|
||||
Timestamp: time.Now(),
|
||||
Direction: "outgoing",
|
||||
Platform: msg.Platform,
|
||||
ChannelID: msg.ChannelID,
|
||||
SenderID: msg.SenderID,
|
||||
Success: false,
|
||||
Error: err.Error(),
|
||||
SenderID: msg.OriginalSenderUID,
|
||||
SenderName: "Cyrene",
|
||||
Content: "[blocked]",
|
||||
Success: true,
|
||||
})
|
||||
return nil, err
|
||||
return &bridge.UnifiedResponse{
|
||||
Messages: []bridge.ResponseMessage{
|
||||
{DisplayType: "silent", Content: "", FormatMode: "plain"},
|
||||
},
|
||||
Platform: msg.Platform,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Log outgoing.
|
||||
for _, rm := range response.Messages {
|
||||
var response *bridge.UnifiedResponse
|
||||
var routeErr error
|
||||
|
||||
// Extract image URLs for vision/OCR processing (admin + bot-mentioned + admin-mentioned only).
|
||||
imageURLs := getImageURLs(msg)
|
||||
includeImages := isAdmin || isBotMentioned || isMentioned
|
||||
|
||||
// For group chats, use a channel-based user ID to share context between admin and regular users.
|
||||
chatUserID := msg.SenderID
|
||||
if msg.ChannelType == "group" {
|
||||
chatUserID = fmt.Sprintf("platform_%s_group_%s", msg.Platform, msg.ChannelID)
|
||||
}
|
||||
groupSessionID := fmt.Sprintf("platform_%s_%s", msg.Platform, msg.ChannelID)
|
||||
|
||||
switch {
|
||||
case isAdmin:
|
||||
msg.RouteType = "normal"
|
||||
response, routeErr = forwardToAICore(cfg, msg, "text", chatUserID, groupSessionID, imageURLs)
|
||||
|
||||
case isBotMentioned:
|
||||
msg.RouteType = "normal"
|
||||
response, routeErr = forwardToAICore(cfg, msg, "text", chatUserID, groupSessionID, imageURLs)
|
||||
|
||||
case isMentioned:
|
||||
msg.RouteType = "admin_mention"
|
||||
enhancedContent := fmt.Sprintf(
|
||||
"[来自%s平台 %s 频道 %s 的用户 %s 说]\n%s\n\n[系统提示：此消息提及了管理员（原因：%s）。请参考以上消息内容判断是否需要关注。]",
|
||||
msg.Platform, msg.ChannelType, msg.ChannelID, msg.SenderName, msg.Content, mentionReason,
|
||||
)
|
||||
originalContent := msg.Content
|
||||
msg.Content = enhancedContent
|
||||
if includeImages {
|
||||
response, routeErr = forwardToAICore(cfg, msg, "text", "admin", "admin-session-main", imageURLs)
|
||||
} else {
|
||||
response, routeErr = forwardToAICore(cfg, msg, "text", "admin", "admin-session-main", nil)
|
||||
}
|
||||
msg.Content = originalContent
|
||||
|
||||
case isSilent:
|
||||
msg.RouteType = "silent"
|
||||
namespace := buildMemoryNamespace(msg.Platform, msg.ChannelType, msg.ChannelID)
|
||||
silentResponse, silentErr := forwardToAICore(cfg, msg, "platform_silent", namespace, namespace, nil)
|
||||
if silentErr != nil {
|
||||
msgLogger.Log(logging.LogEntry{
|
||||
Timestamp: time.Now(),
|
||||
Direction: "outgoing",
|
||||
Platform: msg.Platform,
|
||||
ChannelID: msg.ChannelID,
|
||||
SenderID: msg.OriginalSenderUID,
|
||||
Success: false,
|
||||
Error: silentErr.Error(),
|
||||
})
|
||||
return nil, silentErr
|
||||
}
|
||||
response = silentResponse
|
||||
routeErr = nil
|
||||
|
||||
default:
|
||||
msg.RouteType = "normal"
|
||||
response, routeErr = forwardToAICore(cfg, msg, "text", chatUserID, groupSessionID, nil)
|
||||
}
|
||||
|
||||
if routeErr != nil {
|
||||
msgLogger.Log(logging.LogEntry{
|
||||
Timestamp: time.Now(),
|
||||
Direction: "outgoing",
|
||||
Platform: msg.Platform,
|
||||
ChannelID: msg.ChannelID,
|
||||
SenderID: msg.SenderID,
|
||||
SenderName: "Cyrene",
|
||||
Content: rm.Content,
|
||||
ContentType: "text",
|
||||
Success: true,
|
||||
Timestamp: time.Now(),
|
||||
Direction: "outgoing",
|
||||
Platform: msg.Platform,
|
||||
ChannelID: msg.ChannelID,
|
||||
SenderID: msg.OriginalSenderUID,
|
||||
Success: false,
|
||||
Error: routeErr.Error(),
|
||||
})
|
||||
return nil, routeErr
|
||||
}
|
||||
|
||||
// Log outgoing messages (skip for silent route).
|
||||
if msg.RouteType != "silent" {
|
||||
for _, rm := range response.Messages {
|
||||
msgLogger.Log(logging.LogEntry{
|
||||
Timestamp: time.Now(),
|
||||
Direction: "outgoing",
|
||||
Platform: msg.Platform,
|
||||
ChannelID: msg.ChannelID,
|
||||
SenderID: msg.OriginalSenderUID,
|
||||
SenderName: "Cyrene",
|
||||
Content: rm.Content,
|
||||
ContentType: "text",
|
||||
Success: true,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
return response, nil
|
||||
@@ -121,31 +220,50 @@ func main() {
|
||||
|
||||
// Config and log handlers.
|
||||
ch := handler.NewConfigHandler(configStore, router)
|
||||
|
||||
// Hot-reload: on config save/delete, dynamically replace adapters.
|
||||
ch.SetOnConfigChanged(func(name, platform string, enabled bool, fields map[string]string) {
|
||||
if enabled {
|
||||
a := createSingleAdapter(cfg, platform, name, fields)
|
||||
if a == nil {
|
||||
fmt.Printf("WARN: cannot create adapter for %s (platform=%s)\n", name, platform)
|
||||
return
|
||||
}
|
||||
if err := router.ReplaceAdapter(a); err != nil {
|
||||
fmt.Printf("WARN: hot-reload connect %s failed: %v\n", name, err)
|
||||
} else {
|
||||
fmt.Printf("Platform adapter hot-reloaded: %s\n", name)
|
||||
}
|
||||
// Sync admin identities from config fields.
|
||||
syncAdminUIDs(mapper, platform, fields)
|
||||
// Restart QQ reader when QQ config changes.
|
||||
if platform == "qq" {
|
||||
startQQReaders(router)
|
||||
}
|
||||
} else {
|
||||
router.RemoveAdapter(name)
|
||||
fmt.Printf("Platform adapter removed: %s\n", name)
|
||||
// Cancel reader goroutines for removed adapter.
|
||||
if platform == "qq" {
|
||||
startQQReaders(router)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
ch.RegisterRoutes(mux)
|
||||
lh := handler.NewLogHandler(msgLogger)
|
||||
lh := handler.NewLogHandler(msgLogger, configStore)
|
||||
lh.RegisterRoutes(mux)
|
||||
|
||||
// Log WebSocket for real-time log streaming to ethend.
|
||||
logWS := handler.NewLogWSHub(msgLogger)
|
||||
mux.HandleFunc("/ws/logs", logWS.ServeWS)
|
||||
|
||||
// Blocklist settings.
|
||||
blh := handler.NewBlocklistHandler(blocklistStore)
|
||||
blh.RegisterRoutes(mux)
|
||||
|
||||
// Start QQ message reader loop.
|
||||
qq, _ := router.GetAdapter("qq")
|
||||
if qqa, ok := qq.(*qqadapter.Adapter); ok {
|
||||
qqMsgCh := make(chan *qqadapter.OBv11Message, 100)
|
||||
go qqa.ReadMessages(ctx, qqMsgCh)
|
||||
go func() {
|
||||
for msg := range qqMsgCh {
|
||||
response, err := router.RouteMessage("qq", msg)
|
||||
if err != nil {
|
||||
fmt.Printf("[qq] route error: %v\n", err)
|
||||
continue
|
||||
}
|
||||
msgs, err := router.SendResponse(response)
|
||||
if err != nil {
|
||||
fmt.Printf("[qq] send error: %v\n", err)
|
||||
continue
|
||||
}
|
||||
_ = msgs
|
||||
}
|
||||
}()
|
||||
}
|
||||
startQQReaders(router)
|
||||
|
||||
addr := ":" + cfg.Port
|
||||
srv := &http.Server{Addr: addr, Handler: mux}
|
||||
@@ -171,48 +289,121 @@ func main() {
|
||||
fmt.Println("Platform Bridge stopped")
|
||||
}
|
||||
|
||||
// createAdapters builds platform adapters, preferring stored configs over defaults.
|
||||
func createAdapters(cfg *config.Config, store *config.Store) []bridge.PlatformAdapter {
|
||||
allNames := []string{"qq", "telegram", "webhook", "wechat", "feishu", "discord"}
|
||||
var adapters []bridge.PlatformAdapter
|
||||
// qqReaderCancels maps adapter config name to its cancel function.
|
||||
var qqReaderCancels = make(map[string]context.CancelFunc)
|
||||
var qqReaderCancelsMu sync.Mutex
|
||||
|
||||
for _, name := range allNames {
|
||||
stored, _ := store.Get(name)
|
||||
if stored != nil && !stored.Enabled {
|
||||
fmt.Printf("Platform %s is disabled in config, skipping\n", name)
|
||||
// startQQReaders cancels any existing QQ readers and starts one per registered QQ adapter.
|
||||
func startQQReaders(router *bridge.PlatformRouter) {
|
||||
// Cancel all existing readers.
|
||||
qqReaderCancelsMu.Lock()
|
||||
for _, cancel := range qqReaderCancels {
|
||||
cancel()
|
||||
}
|
||||
qqReaderCancels = make(map[string]context.CancelFunc)
|
||||
qqReaderCancelsMu.Unlock()
|
||||
|
||||
for _, qqa := range router.GetAdaptersByPlatform("qq") {
|
||||
qqAdapter, ok := qqa.(*qqadapter.Adapter)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
adapterKey := qqAdapter.ConfigName()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
qqReaderCancelsMu.Lock()
|
||||
qqReaderCancels[adapterKey] = cancel
|
||||
qqReaderCancelsMu.Unlock()
|
||||
|
||||
var a bridge.PlatformAdapter
|
||||
fields := mergeFields(cfg, name, stored)
|
||||
adapter := qqAdapter // capture for goroutine
|
||||
qqMsgCh := make(chan *qqadapter.OBv11Message, 100)
|
||||
go adapter.ReadMessages(ctx, qqMsgCh)
|
||||
go func() {
|
||||
for msg := range qqMsgCh {
|
||||
response, err := router.RouteMessage(adapterKey, msg)
|
||||
if err != nil {
|
||||
fmt.Printf("[qq:%s] route error: %v\n", adapterKey, err)
|
||||
continue
|
||||
}
|
||||
if response != nil && len(response.Messages) > 0 && !hasOnlySilentMessages(response.Messages) {
|
||||
messageType := msg.MessageType
|
||||
userID := msg.UserID
|
||||
groupID := msg.GroupID
|
||||
// Filter non-empty messages.
|
||||
var toSend []bridge.ResponseMessage
|
||||
for _, rm := range response.Messages {
|
||||
if rm.Content != "" {
|
||||
toSend = append(toSend, rm)
|
||||
}
|
||||
}
|
||||
interval := time.Duration(adapter.SendIntervalMs()) * time.Millisecond
|
||||
if interval <= 0 {
|
||||
interval = 2 * time.Second
|
||||
}
|
||||
for i, rm := range toSend {
|
||||
if i > 0 && interval > 0 {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(interval):
|
||||
}
|
||||
}
|
||||
// Re-acquire current adapter for hot-reload safety.
|
||||
cur, err := router.GetAdapter(adapterKey)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
curQQ, ok := cur.(*qqadapter.Adapter)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
var sendErr error
|
||||
switch messageType {
|
||||
case "private":
|
||||
sendErr = curQQ.SendMessage("private", userID, 0, rm.Content)
|
||||
case "group":
|
||||
sendErr = curQQ.SendMessage("group", 0, groupID, rm.Content)
|
||||
}
|
||||
if sendErr != nil {
|
||||
fmt.Printf("[qq:%s] send msg error: %v\n", adapterKey, sendErr)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
switch name {
|
||||
case "qq":
|
||||
port := cfg.QQBotPort
|
||||
if p, ok := fields["bot_port"]; ok && p != "" {
|
||||
port = p
|
||||
}
|
||||
a = qqadapter.NewAdapter(port)
|
||||
case "telegram":
|
||||
token := cfg.TelegramToken
|
||||
if t, ok := fields["bot_token"]; ok && t != "" {
|
||||
token = t
|
||||
}
|
||||
webhookURL := cfg.TelegramWebhookURL
|
||||
if w, ok := fields["webhook_url"]; ok && w != "" {
|
||||
webhookURL = w
|
||||
}
|
||||
a = telegramadapter.NewAdapter(token, webhookURL)
|
||||
case "webhook":
|
||||
a = webhookadapter.NewAdapter("webhook")
|
||||
case "wechat":
|
||||
a = wechatstub.NewAdapter()
|
||||
case "feishu":
|
||||
a = feishustub.NewAdapter()
|
||||
case "discord":
|
||||
a = discordstub.NewAdapter()
|
||||
// createAdapters builds platform adapters from stored configs.
|
||||
func createAdapters(cfg *config.Config, store *config.Store) []bridge.PlatformAdapter {
|
||||
var adapters []bridge.PlatformAdapter
|
||||
|
||||
// Build adapters from stored configs. Each config is a separate adapter instance.
|
||||
seen := make(map[string]bool)
|
||||
for _, stored := range store.List() {
|
||||
if !stored.Enabled {
|
||||
fmt.Printf("Platform %s is disabled in config, skipping\n", stored.Name)
|
||||
continue
|
||||
}
|
||||
platform := stored.Platform
|
||||
if platform == "" {
|
||||
platform = stored.Name
|
||||
}
|
||||
fields := mergeFields(cfg, platform, &stored)
|
||||
a := createSingleAdapter(cfg, platform, stored.Name, fields)
|
||||
if a != nil {
|
||||
adapters = append(adapters, a)
|
||||
seen[stored.Name] = true
|
||||
}
|
||||
}
|
||||
|
||||
// Seed default adapters for platforms that have no stored config.
|
||||
defaultPlatforms := []string{"qq", "telegram", "webhook", "wechat", "feishu", "discord"}
|
||||
for _, name := range defaultPlatforms {
|
||||
if seen[name] {
|
||||
continue
|
||||
}
|
||||
fields := mergeFields(cfg, name, nil)
|
||||
a := createSingleAdapter(cfg, name, name, fields)
|
||||
if a != nil {
|
||||
adapters = append(adapters, a)
|
||||
}
|
||||
@@ -220,46 +411,132 @@ func createAdapters(cfg *config.Config, store *config.Store) []bridge.PlatformAd
|
||||
return adapters
|
||||
}
|
||||
|
||||
// createSingleAdapter creates one platform adapter from config fields.
|
||||
// platform is the base platform type ("qq", "telegram", etc.), configName is the instance key.
|
||||
func createSingleAdapter(cfg *config.Config, platform, configName string, fields map[string]string) bridge.PlatformAdapter {
|
||||
switch platform {
|
||||
case "qq":
|
||||
port := cfg.QQBotPort
|
||||
if p, ok := fields["bot_port"]; ok && p != "" {
|
||||
port = p
|
||||
}
|
||||
token := ""
|
||||
if t, ok := fields["access_token"]; ok {
|
||||
token = t
|
||||
}
|
||||
remoteURL := ""
|
||||
if r, ok := fields["remote_url"]; ok {
|
||||
remoteURL = r
|
||||
}
|
||||
mode := "server"
|
||||
if m, ok := fields["mode"]; ok && m != "" {
|
||||
mode = m
|
||||
} else if fields["remote_url"] != "" {
|
||||
mode = "client" // backward compat: old configs with remote_url
|
||||
}
|
||||
sendIntervalMs := cfg.MessageSendIntervalMs
|
||||
if s, ok := fields["send_interval_ms"]; ok && s != "" {
|
||||
if n := parseIntOr(s, cfg.MessageSendIntervalMs); n > 0 {
|
||||
sendIntervalMs = n
|
||||
}
|
||||
}
|
||||
return qqadapter.NewAdapter(configName, mode, port, token, remoteURL, sendIntervalMs)
|
||||
case "telegram":
|
||||
token := cfg.TelegramToken
|
||||
if t, ok := fields["bot_token"]; ok && t != "" {
|
||||
token = t
|
||||
}
|
||||
webhookURL := cfg.TelegramWebhookURL
|
||||
if w, ok := fields["webhook_url"]; ok && w != "" {
|
||||
webhookURL = w
|
||||
}
|
||||
return telegramadapter.NewAdapter(token, webhookURL)
|
||||
case "webhook":
|
||||
return webhookadapter.NewAdapter(configName)
|
||||
case "wechat":
|
||||
return wechatstub.NewAdapter()
|
||||
case "feishu":
|
||||
return feishustub.NewAdapter()
|
||||
case "discord":
|
||||
return discordstub.NewAdapter()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// mergeFields returns fields from stored config, falling back to env defaults.
|
||||
func mergeFields(cfg *config.Config, name string, stored *config.PlatformConfig) map[string]string {
|
||||
func mergeFields(cfg *config.Config, platform string, stored *config.PlatformConfig) map[string]string {
|
||||
fields := make(map[string]string)
|
||||
if stored != nil {
|
||||
for k, v := range stored.Fields {
|
||||
fields[k] = v
|
||||
}
|
||||
}
|
||||
// Apply env var defaults if fields are missing.
|
||||
if fields["bot_token"] == "" && cfg.TelegramToken != "" && name == "telegram" {
|
||||
if fields["bot_token"] == "" && cfg.TelegramToken != "" && platform == "telegram" {
|
||||
fields["bot_token"] = cfg.TelegramToken
|
||||
}
|
||||
if fields["webhook_url"] == "" && cfg.TelegramWebhookURL != "" && name == "telegram" {
|
||||
if fields["webhook_url"] == "" && cfg.TelegramWebhookURL != "" && platform == "telegram" {
|
||||
fields["webhook_url"] = cfg.TelegramWebhookURL
|
||||
}
|
||||
if fields["bot_port"] == "" && cfg.QQBotPort != "" && name == "qq" {
|
||||
if fields["bot_port"] == "" && cfg.QQBotPort != "" && platform == "qq" {
|
||||
fields["bot_port"] = cfg.QQBotPort
|
||||
}
|
||||
return fields
|
||||
}
|
||||
|
||||
// containsString checks whether a string slice contains a specific value.
|
||||
func containsString(list []string, val string) bool {
|
||||
for _, v := range list {
|
||||
if v == val {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// getImageURLs extracts image attachment URLs from a UnifiedMessage.
|
||||
func getImageURLs(msg *bridge.UnifiedMessage) []string {
|
||||
if len(msg.Attachments) == 0 {
|
||||
return nil
|
||||
}
|
||||
var urls []string
|
||||
for _, a := range msg.Attachments {
|
||||
if a.Type == "image" && a.URL != "" {
|
||||
urls = append(urls, a.URL)
|
||||
}
|
||||
}
|
||||
return urls
|
||||
}
|
||||
|
||||
// forwardToAICore sends a unified message to AI-Core's chat endpoint and returns the response.
|
||||
func forwardToAICore(cfg *config.Config, msg *bridge.UnifiedMessage) (*bridge.UnifiedResponse, error) {
|
||||
reqBody, _ := json.Marshal(map[string]interface{}{
|
||||
"user_id": msg.SenderID,
|
||||
"session_id": fmt.Sprintf("platform_%s_%s", msg.Platform, msg.ChannelID),
|
||||
// If images is non-empty, they are passed as URL strings for AI-Core to download and process.
|
||||
func forwardToAICore(cfg *config.Config, msg *bridge.UnifiedMessage, mode, userID, sessionID string, images []string) (*bridge.UnifiedResponse, error) {
|
||||
bodyMap := map[string]interface{}{
|
||||
"user_id": userID,
|
||||
"session_id": sessionID,
|
||||
"message": msg.Content,
|
||||
"mode": "text",
|
||||
"mode": mode,
|
||||
"routing": msg.RouteType,
|
||||
"source": map[string]string{
|
||||
"platform": msg.Platform,
|
||||
"channel_id": msg.ChannelID,
|
||||
"channel_type": msg.ChannelType,
|
||||
"sender_name": msg.SenderName,
|
||||
"platform": msg.Platform,
|
||||
"channel_id": msg.ChannelID,
|
||||
"channel_type": msg.ChannelType,
|
||||
"sender_name": msg.SenderName,
|
||||
"original_uid": msg.OriginalSenderUID,
|
||||
},
|
||||
})
|
||||
}
|
||||
if len(images) > 0 {
|
||||
bodyMap["images"] = images
|
||||
}
|
||||
reqBody, _ := json.Marshal(bodyMap)
|
||||
|
||||
url := cfg.AICoreURL + "/api/v1/chat"
|
||||
req, _ := http.NewRequest("POST", url, bytes.NewReader(reqBody))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Accept", "text/event-stream")
|
||||
if mode == "platform_silent" {
|
||||
req.Header.Set("Accept", "application/json")
|
||||
} else {
|
||||
req.Header.Set("Accept", "text/event-stream")
|
||||
}
|
||||
if cfg.InternalToken != "" {
|
||||
req.Header.Set("X-Internal-Token", cfg.InternalToken)
|
||||
}
|
||||
@@ -271,56 +548,276 @@ func forwardToAICore(cfg *config.Config, msg *bridge.UnifiedMessage) (*bridge.Un
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
// Silent mode: only check status, no reply expected.
|
||||
if mode == "platform_silent" {
|
||||
if resp.StatusCode >= 400 {
|
||||
return nil, fmt.Errorf("silent forward returned status %d", resp.StatusCode)
|
||||
}
|
||||
return &bridge.UnifiedResponse{
|
||||
Messages: []bridge.ResponseMessage{
|
||||
{DisplayType: "silent", Content: "", FormatMode: "plain"},
|
||||
},
|
||||
Platform: msg.Platform,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Try JSON first (non-streaming or already-complete response).
|
||||
var result struct {
|
||||
Content string `json:"content"`
|
||||
Error string `json:"error"`
|
||||
}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
|
||||
buf := new(bytes.Buffer)
|
||||
buf.ReadFrom(resp.Body)
|
||||
return &bridge.UnifiedResponse{
|
||||
Messages: []bridge.ResponseMessage{
|
||||
{DisplayType: "chat", Content: buf.String(), FormatMode: "plain"},
|
||||
},
|
||||
Platform: msg.Platform,
|
||||
}, nil
|
||||
bodyBytes, readErr := ioReadAll(resp.Body)
|
||||
if readErr != nil {
|
||||
return nil, fmt.Errorf("read ai-core response: %w", readErr)
|
||||
}
|
||||
|
||||
if result.Error != "" {
|
||||
return &bridge.UnifiedResponse{
|
||||
Messages: []bridge.ResponseMessage{
|
||||
{DisplayType: "system_info", Content: result.Error, FormatMode: "plain"},
|
||||
},
|
||||
Platform: msg.Platform,
|
||||
}, nil
|
||||
if json.Unmarshal(bodyBytes, &result) == nil {
|
||||
if result.Error != "" {
|
||||
return &bridge.UnifiedResponse{
|
||||
Messages: []bridge.ResponseMessage{
|
||||
{DisplayType: "system_info", Content: result.Error, FormatMode: "plain"},
|
||||
},
|
||||
Platform: msg.Platform,
|
||||
}, nil
|
||||
}
|
||||
if result.Content != "" {
|
||||
return &bridge.UnifiedResponse{
|
||||
Messages: splitContent(filterActions(result.Content)),
|
||||
Platform: msg.Platform,
|
||||
}, nil
|
||||
}
|
||||
}
|
||||
|
||||
// Not JSON — parse as SSE (text/event-stream).
|
||||
content := parseSSEAndAccumulate(string(bodyBytes))
|
||||
if content == "" {
|
||||
return nil, fmt.Errorf("ai-core returned empty response")
|
||||
}
|
||||
return &bridge.UnifiedResponse{
|
||||
Messages: []bridge.ResponseMessage{
|
||||
{DisplayType: "chat", Content: result.Content, FormatMode: "plain"},
|
||||
},
|
||||
Messages: splitContent(filterActions(content)),
|
||||
Platform: msg.Platform,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// seedIdentities loads default identity mappings.
|
||||
func seedIdentities(m *bridge.IdentityMapper) {
|
||||
if qqAdmin := os.Getenv("QQ_ADMIN_UID"); qqAdmin != "" {
|
||||
m.Register(permissions.PlatformIdentity{
|
||||
Platform: "qq",
|
||||
PlatformUID: qqAdmin,
|
||||
CyreneUser: "admin",
|
||||
Nickname: "开拓者",
|
||||
PermissionLevel: "admin",
|
||||
})
|
||||
// ioReadAll reads all bytes from a reader (replaces io.ReadAll for older Go compat).
|
||||
func ioReadAll(r io.Reader) ([]byte, error) {
|
||||
buf := new(bytes.Buffer)
|
||||
_, err := buf.ReadFrom(r)
|
||||
return buf.Bytes(), err
|
||||
}
|
||||
|
||||
// parseSSEAndAccumulate parses a Server-Sent Events stream and accumulates the full text.
|
||||
// It handles both "delta" chunks and the final "segments" structure.
|
||||
func parseSSEAndAccumulate(body string) string {
|
||||
lines := strings.Split(body, "\n")
|
||||
var deltas []string
|
||||
var finalText string
|
||||
|
||||
for _, line := range lines {
|
||||
line = strings.TrimSpace(line)
|
||||
if line == "" || line == "[DONE]" {
|
||||
continue
|
||||
}
|
||||
if !strings.HasPrefix(line, "data: ") {
|
||||
continue
|
||||
}
|
||||
payload := strings.TrimPrefix(line, "data: ")
|
||||
|
||||
var chunk map[string]interface{}
|
||||
if err := json.Unmarshal([]byte(payload), &chunk); err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
// Final segment: use the full text if available.
|
||||
if done, _ := chunk["done"].(bool); done {
|
||||
if segs, ok := chunk["segments"].([]interface{}); ok && len(segs) > 0 {
|
||||
if seg, ok := segs[0].(map[string]interface{}); ok {
|
||||
if t, ok := seg["text"].(string); ok && t != "" {
|
||||
finalText = t
|
||||
}
|
||||
}
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
// Accumulate deltas.
|
||||
if delta, ok := chunk["delta"].(string); ok && delta != "" {
|
||||
deltas = append(deltas, delta)
|
||||
}
|
||||
// Also accumulate content if present (some APIs send full content).
|
||||
if content, ok := chunk["content"].(string); ok && content != "" {
|
||||
deltas = append(deltas, content)
|
||||
}
|
||||
}
|
||||
if tgAdmin := os.Getenv("TELEGRAM_ADMIN_UID"); tgAdmin != "" {
|
||||
m.Register(permissions.PlatformIdentity{
|
||||
Platform: "telegram",
|
||||
PlatformUID: tgAdmin,
|
||||
CyreneUser: "admin",
|
||||
Nickname: "开拓者",
|
||||
PermissionLevel: "admin",
|
||||
})
|
||||
|
||||
if finalText != "" {
|
||||
return finalText
|
||||
}
|
||||
return strings.Join(deltas, "")
|
||||
}
|
||||
|
||||
// splitContent splits text by \n\n into multiple ResponseMessage segments.
|
||||
// Non-empty segments are each wrapped as a chat message; empty input returns a single empty message.
|
||||
func splitContent(text string) []bridge.ResponseMessage {
|
||||
parts := strings.Split(text, "\n\n")
|
||||
var msgs []bridge.ResponseMessage
|
||||
for _, part := range parts {
|
||||
part = strings.TrimSpace(part)
|
||||
if part != "" {
|
||||
msgs = append(msgs, bridge.ResponseMessage{
|
||||
DisplayType: "chat",
|
||||
Content: part,
|
||||
FormatMode: "plain",
|
||||
})
|
||||
}
|
||||
}
|
||||
if len(msgs) == 0 {
|
||||
return []bridge.ResponseMessage{
|
||||
{DisplayType: "chat", Content: text, FormatMode: "plain"},
|
||||
}
|
||||
}
|
||||
return msgs
|
||||
}
|
||||
|
||||
// filterActions removes <action>...</action> tags and their content from text.
|
||||
// Also handles unescaped <action> tags from SSE (e.g. <action>).
|
||||
func filterActions(text string) string {
|
||||
// Remove standard <action>...</action> tags.
|
||||
for {
|
||||
start := strings.Index(text, "<action>")
|
||||
if start == -1 {
|
||||
break
|
||||
}
|
||||
end := strings.Index(text[start:], "</action>")
|
||||
if end == -1 {
|
||||
// Unclosed action tag — just remove the opening tag.
|
||||
text = text[:start] + text[start+len("<action>"):]
|
||||
continue
|
||||
}
|
||||
text = text[:start] + text[start+end+len("</action>"):]
|
||||
}
|
||||
|
||||
// Remove SSE-escaped action tags (<action>...</action>).
|
||||
for {
|
||||
start := strings.Index(text, `<action>`)
|
||||
if start == -1 {
|
||||
break
|
||||
}
|
||||
end := strings.Index(text[start:], `</action>`)
|
||||
if end == -1 {
|
||||
text = text[:start] + text[start+len(`<action>`):]
|
||||
continue
|
||||
}
|
||||
text = text[:start] + text[start+end+len(`</action>`):]
|
||||
}
|
||||
|
||||
return strings.TrimSpace(text)
|
||||
}
|
||||
|
||||
// buildMemoryNamespace creates a memory-isolated user_id for a platform channel.
|
||||
func buildMemoryNamespace(platform, channelType, channelID string) string {
|
||||
return fmt.Sprintf("platform_%s_%s_%s", platform, channelType, channelID)
|
||||
}
|
||||
|
||||
// detectAdminMention checks whether a message mentions the admin.
|
||||
func detectAdminMention(msg *bridge.UnifiedMessage, mapper *bridge.IdentityMapper, cfg *config.Config) (bool, string) {
|
||||
for _, mentionUID := range msg.Mentions {
|
||||
if mapper.IsAdmin(msg.Platform, mentionUID) {
|
||||
return true, fmt.Sprintf("@mention of admin UID %s", mentionUID)
|
||||
}
|
||||
}
|
||||
lowerContent := strings.ToLower(msg.Content)
|
||||
for _, nickname := range cfg.AdminNicknames {
|
||||
if strings.Contains(msg.Content, nickname) {
|
||||
return true, fmt.Sprintf("admin nickname mention: %s", nickname)
|
||||
}
|
||||
}
|
||||
for _, kw := range cfg.AdminMentionKeywords {
|
||||
if strings.Contains(lowerContent, strings.ToLower(kw)) {
|
||||
return true, fmt.Sprintf("admin keyword mention: %s", kw)
|
||||
}
|
||||
}
|
||||
return false, ""
|
||||
}
|
||||
|
||||
// hasOnlySilentMessages checks if all response messages are silent (no reply needed).
|
||||
func hasOnlySilentMessages(messages []bridge.ResponseMessage) bool {
|
||||
for _, m := range messages {
|
||||
if m.DisplayType != "silent" && m.Content != "" {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func parseIntOr(s string, defaultVal int) int {
|
||||
n := 0
|
||||
for _, c := range s {
|
||||
if c >= '0' && c <= '9' {
|
||||
n = n*10 + int(c-'0')
|
||||
} else {
|
||||
return defaultVal
|
||||
}
|
||||
}
|
||||
if n == 0 && s != "0" {
|
||||
return defaultVal
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// seedIdentities loads default identity mappings from env vars and stored platform configs.
|
||||
func seedIdentities(m *bridge.IdentityMapper, store *config.Store) {
|
||||
// From environment variables.
|
||||
for _, entry := range []struct{ envKey, platform string }{
|
||||
{"QQ_ADMIN_UID", "qq"},
|
||||
{"TELEGRAM_ADMIN_UID", "telegram"},
|
||||
} {
|
||||
if raw := os.Getenv(entry.envKey); raw != "" {
|
||||
for _, uid := range strings.Split(raw, ",") {
|
||||
uid = strings.TrimSpace(uid)
|
||||
if uid == "" {
|
||||
continue
|
||||
}
|
||||
m.Register(permissions.PlatformIdentity{
|
||||
Platform: entry.platform,
|
||||
PlatformUID: uid,
|
||||
CyreneUser: "admin",
|
||||
Nickname: "开拓者",
|
||||
PermissionLevel: "admin",
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// From stored platform configs (admin_uids field).
|
||||
for _, name := range []string{"qq", "telegram", "webhook", "wechat", "feishu", "discord"} {
|
||||
stored, _ := store.Get(name)
|
||||
if stored == nil {
|
||||
continue
|
||||
}
|
||||
syncAdminUIDs(m, name, stored.Fields)
|
||||
}
|
||||
}
|
||||
|
||||
// syncAdminUIDs registers admin identities from a platform config's admin_uids field.
|
||||
// Comma-separated list of platform UIDs.
|
||||
func syncAdminUIDs(m *bridge.IdentityMapper, platform string, fields map[string]string) {
|
||||
raw, ok := fields["admin_uids"]
|
||||
if !ok || raw == "" {
|
||||
return
|
||||
}
|
||||
for _, uid := range strings.Split(raw, ",") {
|
||||
uid = strings.TrimSpace(uid)
|
||||
if uid == "" {
|
||||
continue
|
||||
}
|
||||
m.Register(permissions.PlatformIdentity{
|
||||
Platform: platform,
|
||||
PlatformUID: uid,
|
||||
CyreneUser: "admin",
|
||||
Nickname: "开拓者",
|
||||
PermissionLevel: "admin",
|
||||
})
|
||||
}
|
||||
fmt.Printf("Synced admin identities for %s from config: %s\n", platform, raw)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user