533 lines
14 KiB
Go
533 lines
14 KiB
Go
package biz
|
||
|
||
import (
|
||
"ai_scheduler/internal/config"
|
||
"ai_scheduler/internal/data/mongo_model"
|
||
"ai_scheduler/internal/entitys"
|
||
"ai_scheduler/internal/pkg"
|
||
"ai_scheduler/internal/pkg/wx"
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"go.mongodb.org/mongo-driver/bson"
|
||
"go.mongodb.org/mongo-driver/mongo/options"
|
||
|
||
"github.com/gofiber/fiber/v2/log"
|
||
)
|
||
|
||
// WsBroadcaster WebSocket 广播接口（由 services/advice.WxWsHub 实现）
|
||
type WsBroadcaster interface {
|
||
Broadcast(appId string, event string, data interface{})
|
||
}
|
||
|
||
// AdviceWxBiz 个人微信消息回调接入与消息流水
|
||
// 上游（WTAPI）消息通过 login/setCallback 配置的回调地址推送到本服务，
|
||
// 本模块负责容错解析回调报文、记录消息流水、更新客户互动时间。
|
||
type AdviceWxBiz struct {
|
||
cfg *config.Config
|
||
mongo *pkg.Mongo
|
||
wxMsgMongo *mongo_model.AdvicerWxMsgMongo
|
||
clientBiz *AdviceClientBiz
|
||
strategyBiz *AdviceStrategyBiz
|
||
wsBroadcaster WsBroadcaster
|
||
}
|
||
|
||
func NewAdviceWxBiz(
|
||
cfg *config.Config,
|
||
mongo *pkg.Mongo,
|
||
wxMsgMongo *mongo_model.AdvicerWxMsgMongo,
|
||
clientBiz *AdviceClientBiz,
|
||
strategyBiz *AdviceStrategyBiz,
|
||
wsBroadcaster WsBroadcaster,
|
||
) *AdviceWxBiz {
|
||
return &AdviceWxBiz{
|
||
cfg: cfg,
|
||
mongo: mongo,
|
||
wxMsgMongo: wxMsgMongo,
|
||
clientBiz: clientBiz,
|
||
strategyBiz: strategyBiz,
|
||
wsBroadcaster: wsBroadcaster,
|
||
}
|
||
}
|
||
|
||
// ==================== 回调处理 ====================
|
||
|
||
// HandleCallback 处理上游推送的回调报文（支持单条/数组/字符串包裹 JSON）
|
||
func (a *AdviceWxBiz) HandleCallback(ctx context.Context, body []byte) (err error) {
|
||
body = bytes.TrimSpace(body)
|
||
if len(body) == 0 {
|
||
return nil
|
||
}
|
||
// 数组报文：逐条处理，单条失败不影响其他
|
||
if body[0] == '[' {
|
||
var arr []json.RawMessage
|
||
if e := json.Unmarshal(body, &arr); e == nil {
|
||
for _, item := range arr {
|
||
if e := a.handleOneEvent(ctx, item); e != nil {
|
||
err = e
|
||
}
|
||
}
|
||
return err
|
||
}
|
||
}
|
||
return a.handleOneEvent(ctx, body)
|
||
}
|
||
|
||
// handleOneEvent 处理单条回调事件（基于 callback.md 文档的类型化解析）
|
||
func (a *AdviceWxBiz) handleOneEvent(ctx context.Context, body []byte) error {
|
||
event, parseErr := wx.ParseCallbackEvent(body)
|
||
if parseErr != nil {
|
||
return fmt.Errorf("解析回调报文失败: %w", parseErr)
|
||
}
|
||
|
||
// 按 TypeName 分发
|
||
isAddMsg, isModContacts, isDelContacts, isOffline, isFinder := wx.DispatchCallbackEvent(event)
|
||
|
||
// 联系人变动/离线/视频号事件不处理（仅处理 AddMsg）
|
||
if !isAddMsg {
|
||
_ = isModContacts
|
||
_ = isDelContacts
|
||
_ = isOffline
|
||
_ = isFinder
|
||
return nil
|
||
}
|
||
|
||
if event.Data == nil {
|
||
return nil
|
||
}
|
||
|
||
// 判定消息方向与对话方 wxid
|
||
direction := mongo_model.WxMsgDirectionCustomer
|
||
peerWxid := event.PeerWxid()
|
||
|
||
if event.IsSelfMsg() {
|
||
direction = mongo_model.WxMsgDirectionSelf
|
||
} else if len(event.Wxid) == 0 && a.clientBiz != nil {
|
||
// 报文未提供登录微信标识：借助客户绑定库判定
|
||
from := event.Data.FromUserName.Val()
|
||
to := event.Data.ToUserName.Val()
|
||
if _, found, _ := a.clientBiz.FindByWxid(ctx, from); !found {
|
||
if _, foundTo, _ := a.clientBiz.FindByWxid(ctx, to); foundTo {
|
||
direction = mongo_model.WxMsgDirectionSelf
|
||
peerWxid = to
|
||
}
|
||
}
|
||
}
|
||
if len(peerWxid) == 0 {
|
||
peerWxid = event.Data.FromUserName.Val()
|
||
}
|
||
|
||
// 群消息：提取真实发送者
|
||
content := event.Data.Content.Val()
|
||
var senderWxid string
|
||
if event.IsGroupMsg() && direction == mongo_model.WxMsgDirectionCustomer {
|
||
realSender := event.GroupRealSender()
|
||
if realSender != "" {
|
||
senderWxid = realSender
|
||
// 群消息 Content 格式为 "wxid_xxx:\n消息内容"，去掉前缀取正文
|
||
if idx := strings.Index(content, ":\n"); idx > 0 {
|
||
content = content[idx+2:]
|
||
}
|
||
}
|
||
}
|
||
|
||
msgType := event.Data.MsgType
|
||
newMsgId := event.Data.NewMsgId.String()
|
||
msgId := event.Data.MsgId.String()
|
||
|
||
// 过滤：公众号消息（gh_ 开头）不存储不展示
|
||
if strings.HasPrefix(peerWxid, "gh_") {
|
||
return nil
|
||
}
|
||
// 过滤：msgType 为 other 的不存储不展示
|
||
if wx.MsgTypeCategory(msgType) == "other" {
|
||
return nil
|
||
}
|
||
|
||
// 媒体消息存原始 XML（去前缀后的），供前端调用下载 API
|
||
var mediaXml string
|
||
if wx.IsMediaMsg(msgType) {
|
||
mediaXml = content
|
||
}
|
||
|
||
record := &mongo_model.AdvicerWxMsgMongo{
|
||
AppId: event.AppId,
|
||
Wxid: peerWxid,
|
||
SelfWxid: event.Wxid,
|
||
SenderWxid: senderWxid,
|
||
Direction: direction,
|
||
MsgType: wx.MsgTypeCategory(msgType),
|
||
Content: content,
|
||
MediaXml: mediaXml,
|
||
MsgId: msgId,
|
||
NewMsgId: newMsgId,
|
||
Source: mongo_model.WxMsgSourceCallback,
|
||
Raw: string(body),
|
||
CreateAt: callbackCreateTime(event.Data.CreateTime),
|
||
}
|
||
|
||
// 消息去重：使用 AppId + NewMsgId（文档推荐去重键）
|
||
dedupKey := event.DedupKey()
|
||
if len(dedupKey) > 1 {
|
||
count, e := a.mongo.Co(a.wxMsgMongo).CountDocuments(ctx, bson.M{"newMsgId": newMsgId, "appId": event.AppId})
|
||
if e != nil {
|
||
return fmt.Errorf("消息去重查询失败: %w", e)
|
||
}
|
||
if count > 0 {
|
||
return nil
|
||
}
|
||
} else if len(msgId) != 0 {
|
||
// 回退：旧格式无 NewMsgId，使用 MsgId 去重
|
||
count, e := a.mongo.Co(a.wxMsgMongo).CountDocuments(ctx, bson.M{"msgId": msgId})
|
||
if e != nil {
|
||
return fmt.Errorf("消息去重查询失败: %w", e)
|
||
}
|
||
if count > 0 {
|
||
return nil
|
||
}
|
||
}
|
||
|
||
if _, e := a.mongo.Co(a.wxMsgMongo).InsertOne(ctx, record); e != nil {
|
||
return fmt.Errorf("消息流水落库失败: %w", e)
|
||
}
|
||
|
||
// WebSocket 实时广播：向对应 appId 的已连接前端推送新消息事件
|
||
if a.wsBroadcaster != nil {
|
||
log.Infof("[WS] broadcasting new_message appId=%s wxid=%s", event.AppId, peerWxid)
|
||
a.wsBroadcaster.Broadcast(event.AppId, "new_message", map[string]interface{}{
|
||
"appId": event.AppId,
|
||
"wxid": peerWxid,
|
||
"senderWxid": senderWxid,
|
||
"direction": string(direction),
|
||
"msgType": wx.MsgTypeCategory(msgType),
|
||
"content": content,
|
||
"mediaXml": mediaXml,
|
||
"msgId": msgId,
|
||
"createAt": record.CreateAt.Unix(),
|
||
})
|
||
}
|
||
|
||
// 已绑定客户则同步互动时间与消息数
|
||
if a.clientBiz != nil && len(peerWxid) != 0 {
|
||
_ = a.clientBiz.TouchInteraction(ctx, peerWxid, direction)
|
||
}
|
||
|
||
// 智能策略：客户文本消息异步触发 AI 回复决策
|
||
if direction == mongo_model.WxMsgDirectionCustomer && len(peerWxid) != 0 && wx.IsTextMsg(msgType) && a.strategyBiz != nil {
|
||
wxid, msgContent := peerWxid, content
|
||
go func() {
|
||
sctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||
defer cancel()
|
||
a.strategyBiz.OnCustomerMessage(sctx, wxid, msgContent)
|
||
}()
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// callbackCreateTime 将回调时间戳转为 time.Time（兼容秒/毫秒，缺省取当前时间）
|
||
func callbackCreateTime(ts int64) time.Time {
|
||
if ts <= 0 {
|
||
return time.Now()
|
||
}
|
||
if ts > 1e12 { // 毫秒级
|
||
return time.UnixMilli(ts)
|
||
}
|
||
return time.Unix(ts, 0)
|
||
}
|
||
|
||
// VerifyCallbackToken 校验回调请求合法性。
|
||
// 上游回调若携带 token（query 或 body 字段）则与配置比对；未携带时放行。
|
||
func (a *AdviceWxBiz) VerifyCallbackToken(query map[string]string, body []byte) bool {
|
||
expect := a.cfg.Advicer.CallbackToken
|
||
if len(expect) == 0 {
|
||
return true
|
||
}
|
||
got := query["token"]
|
||
if got == "" {
|
||
// 从 body 中提取 token 字段（兼容多种命名）
|
||
m := toMap(body)
|
||
if m != nil {
|
||
for _, k := range []string{"token", "Token", "verifyToken", "verify_token"} {
|
||
if v, ok := m[k]; ok {
|
||
if s, ok := v.(string); ok && s != "" {
|
||
got = s
|
||
break
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
if got == "" {
|
||
return true
|
||
}
|
||
return got == expect
|
||
}
|
||
|
||
// SetCallback 调用上游接口配置消息回调地址（login/setCallback）
|
||
func (a *AdviceWxBiz) SetCallback(ctx context.Context, req *entitys.AdvicerWxCallbackReq) (err error) {
|
||
token := a.cfg.Advicer.WxToken
|
||
if len(token) == 0 {
|
||
return errors.New("未配置 advicer.wx_token，无法调用上游接口")
|
||
}
|
||
if len(req.CallbackUrl) == 0 {
|
||
return errors.New("回调地址不能为空")
|
||
}
|
||
cbToken := a.cfg.Advicer.CallbackToken
|
||
if len(cbToken) == 0 {
|
||
cbToken = token
|
||
}
|
||
var res wx.SetCallbackResData
|
||
if err = wx.Request(ctx, token, wx.SetCallback, wx.SetCallbackReq{
|
||
Token: cbToken,
|
||
CallbackUrl: req.CallbackUrl,
|
||
}, &res); err != nil {
|
||
return fmt.Errorf("设置回调失败: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ==================== 消息流水查询 ====================
|
||
|
||
// MsgList 查询微信消息流水（按时间倒序）
|
||
func (a *AdviceWxBiz) MsgList(ctx context.Context, param *entitys.AdvicerWxMsgListReq) (list []mongo_model.AdvicerWxMsgItem, err error) {
|
||
filter := bson.M{}
|
||
if len(param.AppId) != 0 {
|
||
filter["appId"] = param.AppId
|
||
}
|
||
if len(param.Wxid) != 0 {
|
||
filter["wxid"] = param.Wxid
|
||
}
|
||
if len(param.Direction) != 0 {
|
||
filter["direction"] = param.Direction
|
||
}
|
||
timeCond := bson.M{}
|
||
if len(param.StartAt) != 0 {
|
||
if t, e := parseTimeStr(param.StartAt); e == nil {
|
||
timeCond["$gte"] = t
|
||
}
|
||
}
|
||
if len(param.EndAt) != 0 {
|
||
if t, e := parseTimeStr(param.EndAt); e == nil {
|
||
timeCond["$lte"] = t
|
||
}
|
||
}
|
||
if len(timeCond) != 0 {
|
||
filter["createAt"] = timeCond
|
||
}
|
||
|
||
opts := options.Find().SetSort(bson.D{{Key: "createAt", Value: -1}})
|
||
if param.PageSize > 0 {
|
||
page := param.Page
|
||
if page < 1 {
|
||
page = 1
|
||
}
|
||
size := int64(param.PageSize)
|
||
opts.SetSkip(int64(page-1) * size).SetLimit(size)
|
||
} else {
|
||
// 未分页时限制最大返回条数，避免全量拉取
|
||
opts.SetLimit(200)
|
||
}
|
||
|
||
cursor, err := a.mongo.Co(a.wxMsgMongo).Find(ctx, filter, opts)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
for cursor.Next(ctx) {
|
||
var item mongo_model.AdvicerWxMsgItem
|
||
if err = cursor.Decode(&item); err != nil {
|
||
return nil, err
|
||
}
|
||
list = append(list, item)
|
||
}
|
||
if err = cursor.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
return list, nil
|
||
}
|
||
|
||
// ==================== 会话列表与消息记录 ====================
|
||
|
||
// ConversationList 查询会话列表（按最新消息时间倒序，聚合每个 wxid 的最新一条消息）
|
||
func (a *AdviceWxBiz) ConversationList(ctx context.Context, param *entitys.AdvicerWxConversationListReq) ([]entitys.AdvicerWxConversationItem, error) {
|
||
filter := bson.M{}
|
||
if len(param.SelfWxid) != 0 {
|
||
filter["selfWxid"] = param.SelfWxid
|
||
}
|
||
page := param.Page
|
||
if page < 1 {
|
||
page = 1
|
||
}
|
||
size := param.PageSize
|
||
if size <= 0 {
|
||
size = 50
|
||
}
|
||
if size > 200 {
|
||
size = 200
|
||
}
|
||
|
||
pipeline := []bson.M{
|
||
{"$match": filter},
|
||
{"$sort": bson.M{"createAt": -1}},
|
||
{"$group": bson.M{
|
||
"_id": "$wxid",
|
||
"lastMsg": bson.M{"$first": "$content"},
|
||
"lastMsgType": bson.M{"$first": "$msgType"},
|
||
"lastMsgTime": bson.M{"$first": "$createAt"},
|
||
"lastDirection": bson.M{"$first": "$direction"},
|
||
"unreadCount": bson.M{"$sum": bson.M{
|
||
"$cond": []interface{}{
|
||
bson.M{"$and": []interface{}{
|
||
bson.M{"$eq": []interface{}{"$direction", mongo_model.WxMsgDirectionCustomer}},
|
||
bson.M{"$eq": []interface{}{"$read", false}},
|
||
}},
|
||
1, 0,
|
||
},
|
||
}},
|
||
}},
|
||
{"$sort": bson.M{"lastMsgTime": -1}},
|
||
{"$skip": int64(page-1) * int64(size)},
|
||
{"$limit": int64(size)},
|
||
}
|
||
|
||
cursor, err := a.mongo.Co(a.wxMsgMongo).Aggregate(ctx, pipeline)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var list []entitys.AdvicerWxConversationItem
|
||
for cursor.Next(ctx) {
|
||
var row struct {
|
||
Wxid string `bson:"_id"`
|
||
LastMsg string `bson:"lastMsg"`
|
||
LastMsgType string `bson:"lastMsgType"`
|
||
LastMsgTime time.Time `bson:"lastMsgTime"`
|
||
LastDirection string `bson:"lastDirection"`
|
||
UnreadCount int `bson:"unreadCount"`
|
||
}
|
||
if err = cursor.Decode(&row); err != nil {
|
||
return nil, err
|
||
}
|
||
// 截断过长消息
|
||
preview := row.LastMsg
|
||
if len([]rune(preview)) > 40 {
|
||
preview = string([]rune(preview)[:40]) + "…"
|
||
}
|
||
if row.LastMsgType != "text" {
|
||
preview = "[" + row.LastMsgType + "]"
|
||
}
|
||
list = append(list, entitys.AdvicerWxConversationItem{
|
||
Wxid: row.Wxid,
|
||
LastMsg: preview,
|
||
LastMsgType: row.LastMsgType,
|
||
LastMsgTime: row.LastMsgTime.Unix(),
|
||
LastDirection: row.LastDirection,
|
||
UnreadCount: row.UnreadCount,
|
||
IsGroup: strings.HasSuffix(row.Wxid, "@chatroom"),
|
||
})
|
||
}
|
||
if err = cursor.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
return list, nil
|
||
}
|
||
|
||
// MarkAsRead 标记某个会话中所有客户消息为已读
|
||
func (a *AdviceWxBiz) MarkAsRead(ctx context.Context, param *entitys.AdvicerWxMarkAsReadReq) (int64, error) {
|
||
if len(param.Wxid) == 0 {
|
||
return 0, nil
|
||
}
|
||
filter := bson.M{
|
||
"wxid": param.Wxid,
|
||
"direction": mongo_model.WxMsgDirectionCustomer,
|
||
"read": false,
|
||
}
|
||
if len(param.SelfWxid) != 0 {
|
||
filter["selfWxid"] = param.SelfWxid
|
||
}
|
||
update := bson.M{"$set": bson.M{"read": true}}
|
||
result, err := a.mongo.Co(a.wxMsgMongo).UpdateMany(ctx, filter, update)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return result.ModifiedCount, nil
|
||
}
|
||
|
||
// ConversationMsgs 查询某个会话的消息记录（按时间正序，用于聊天展示）
|
||
func (a *AdviceWxBiz) ConversationMsgs(ctx context.Context, param *entitys.AdvicerWxConversationMsgsReq) ([]mongo_model.AdvicerWxMsgItem, error) {
|
||
filter := bson.M{}
|
||
if len(param.SelfWxid) != 0 {
|
||
filter["selfWxid"] = param.SelfWxid
|
||
}
|
||
if len(param.Wxid) != 0 {
|
||
filter["wxid"] = param.Wxid
|
||
}
|
||
opts := options.Find().SetSort(bson.D{{Key: "createAt", Value: 1}})
|
||
if param.PageSize > 0 {
|
||
page := param.Page
|
||
if page < 1 {
|
||
page = 1
|
||
}
|
||
size := int64(param.PageSize)
|
||
if size > 500 {
|
||
size = 500
|
||
}
|
||
opts.SetSkip(int64(page-1) * size).SetLimit(size)
|
||
} else {
|
||
opts.SetLimit(200)
|
||
}
|
||
|
||
cursor, err := a.mongo.Co(a.wxMsgMongo).Find(ctx, filter, opts)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var list []mongo_model.AdvicerWxMsgItem
|
||
for cursor.Next(ctx) {
|
||
var item mongo_model.AdvicerWxMsgItem
|
||
if err = cursor.Decode(&item); err != nil {
|
||
return nil, err
|
||
}
|
||
list = append(list, item)
|
||
}
|
||
if err = cursor.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
return list, nil
|
||
}
|
||
|
||
// ==================== 辅助工具 ====================
|
||
|
||
// toMap 将 JSON 解析为 map（数字保留原文，避免大 MsgId 精度丢失）
|
||
func toMap(b []byte) map[string]interface{} {
|
||
dec := json.NewDecoder(bytes.NewReader(b))
|
||
dec.UseNumber()
|
||
var m map[string]interface{}
|
||
if err := dec.Decode(&m); err != nil {
|
||
return nil
|
||
}
|
||
return m
|
||
}
|
||
|
||
// parseTimeStr 容错解析时间字符串（支持多种常见格式）
|
||
func parseTimeStr(s string) (time.Time, error) {
|
||
s = strings.TrimSpace(s)
|
||
if s == "" {
|
||
return time.Time{}, errors.New("空时间")
|
||
}
|
||
layouts := []string{
|
||
time.RFC3339,
|
||
"2006-01-02 15:04:05",
|
||
"2006-01-02 15:04",
|
||
"2006-01-02",
|
||
}
|
||
for _, l := range layouts {
|
||
if t, err := time.ParseInLocation(l, s, time.Local); err == nil {
|
||
return t, nil
|
||
}
|
||
}
|
||
return time.Time{}, fmt.Errorf("无法解析时间: %s", s)
|
||
}
|