新增自动流转到会话池功能

.gitignore

@@ -1,2 +1,3 @@
 /callback
 /logs
+/etc/callback.yaml

etc/callback.yaml

@@ -22,8 +22,13 @@ Redis:
   Tls: false
 
 Wxwork:
+#  企业ID
   Corpid: "wwa0779f3625818a4e"
+#  新建应用密钥
   Corpsecret: "bDdsGUmT3TbEAaUtPzEB4Fd2kiCIqX8bFcZ4Oy3F3jM"
+#  token
   Token: "8GLSPY3N6mwpWylT7PzHi"
+#  开发者绑定ID
   ReceiverId: "wwa0779f3625818a4e"
+#  aes密钥
   EncodingAeskey: "dkobxiOsG6UMg9SXby1LPc2L9sKoDF9mcq2QK78YwWV"

etc/callback.yaml.bak

@@ -0,0 +1,34 @@
+Name: &Name callback
+Host: 0.0.0.0
+Port: 8888
+
+Log:
+  ServiceName: *Name
+  Mode: "file"
+  Rotation: "daily"
+  Encoding: "plain"
+
+Auth:
+  AccessSecret: "i9hb5_ep"
+  AccessExpire: 3600
+
+Mysql:
+  Datasource: kftest:123456@tcp(10.8.230.17:3308)/wechat_cs_online?charset=utf8mb4&parseTime=true&loc=Asia%2FShanghai
+
+Redis:
+  Host: 10.8.230.17:6379
+  Pass:
+  Type: node
+  Tls: false
+
+Wxwork:
+#  企业ID
+  Corpid: "wwa0779f3625818a4e"
+#  新建应用密钥
+  Corpsecret: "bDdsGUmT3TbEAaUtPzEB4Fd2kiCIqX8bFcZ4Oy3F3jM"
+#  token
+  Token: "8GLSPY3N6mwpWylT7PzHi"
+#  开发者绑定ID
+  ReceiverId: "wwa0779f3625818a4e"
+#  aes密钥
+  EncodingAeskey: "dkobxiOsG6UMg9SXby1LPc2L9sKoDF9mcq2QK78YwWV"

internal/logic/callback/callbackmsglogic.go

@@ -8,6 +8,7 @@ import (
 	"encoding/xml"
 	"fmt"
 	"github.com/elliotchance/pie/pie"
+	"github.com/valyala/fastjson"
 	"io/ioutil"
 	"net/http"
 	"strconv"
@@ -61,10 +62,10 @@ func (l *CallbackMsgLogic) CallbackMsg(req *types.CallbackMsgRequest, w http.Res
 		//事件处理
 		if ct.MsgType == types.Event && ct.Event == types.KfMsgOrEvent {
 			//新消息事件
-			go kfMsgOrEventHandle(l.svcCtx, ct)
+			go KfMsgOrEventHandle(l.svcCtx, ct)
 		} else if ct.MsgType == types.Event && ct.Event == types.KfAccountAuthChange {
 			//客服账号变动事件
-			go kfAccountAuthChangeHandle(l.svcCtx, ct)
+			go KfAccountAuthChangeHandle(l.svcCtx, ct)
 		}
 	}
 	//加密返回
@@ -77,8 +78,8 @@ func (l *CallbackMsgLogic) CallbackMsg(req *types.CallbackMsgRequest, w http.Res
 	return
 }
 
-// 保存新消息或者事件
-func kfMsgOrEventHandle(svcCtx *svc.ServiceContext, ct types.EventContent) {
+// KfMsgOrEventHandle 新消息或者事件上报处理
+func KfMsgOrEventHandle(svcCtx *svc.ServiceContext, ct types.EventContent) {
 	//获取分布式锁
 	rl := lock.NewRedisLock(ct.OpenKfId, svcCtx.Redis)
 	if err := rl.Lock(); err != nil {
@@ -91,7 +92,7 @@ func kfMsgOrEventHandle(svcCtx *svc.ServiceContext, ct types.EventContent) {
 		cbService, err = svcCtx.CbServiceModel.GetServiceByOpenKfid(ct.OpenKfId)
 		cursor = cbService.NextCursor
 		if err != nil {
-			go kfAccountAuthChangeHandle(svcCtx, ct)
+			go KfAccountAuthChangeHandle(svcCtx, ct)
 		}
 	}
 
@@ -129,6 +130,14 @@ func kfMsgOrEventHandle(svcCtx *svc.ServiceContext, ct types.EventContent) {
 			if err != nil {
 				logx.Error("Insert cb_msg fail ", err)
 			}
+			//消息额外处理
+			if cbMsg.Msgtype == types.MsgTypeText {
+				msgTypeTextHandle(svcCtx, cbMsg)
+			}
+			//事件消息额外处理
+			if cbMsg.Msgtype == types.MsgTypeEvent {
+				msgTypeEventHandle(svcCtx, cbMsg)
+			}
 		}
 		go updateCustomerList(svcCtx, userIds)
 		if more == 0 {
@@ -138,8 +147,8 @@ func kfMsgOrEventHandle(svcCtx *svc.ServiceContext, ct types.EventContent) {
 	return
 }
 
-// 更新客服账号列表、接待人员列表
-func kfAccountAuthChangeHandle(svcCtx *svc.ServiceContext, ct types.EventContent) {
+// KfAccountAuthChangeHandle 管理员权限修改事件上报处理
+func KfAccountAuthChangeHandle(svcCtx *svc.ServiceContext, ct types.EventContent) {
 	err, list := svcCtx.WxApi.GetServiceList()
 	if err != nil {
 		logx.Error("GetServiceList fail ", err)
@@ -168,20 +177,22 @@ func kfAccountAuthChangeHandle(svcCtx *svc.ServiceContext, ct types.EventContent
 			logx.Error("GetServicerList fail ", err)
 		}
 		for _, a := range accounts {
-			cbServicer, err := svcCtx.CbServicerModel.GetServicer(s.OpenKfid, a.Userid)
+			err := svcCtx.CbServicerModel.DeleteServicerByOpenKfid(s.OpenKfid)
 			if err != nil {
-				cbServicer.OpenKfid = s.OpenKfid
-				cbServicer.Corpid = svcCtx.Config.Wxwork.Corpid
-				cbServicer.Userid = a.Userid
-				cbServicer.Status = a.Status
-				cbServicer.DepartmentId = a.DepartmentId
-				now := time.Now()
-				cbServicer.CreatedAt = now
-				cbServicer.UpdatedAt = now
-				_, err := svcCtx.CbServicerModel.Insert(nil, cbServicer)
-				if err != nil {
-					logx.Error("Insert cb_servicer fail ", err)
-				}
+				logx.Error("DeleteServicerByOpenKfid fail ", err)
+			}
+			cbServicer := &model.CbServicer{}
+			cbServicer.OpenKfid = s.OpenKfid
+			cbServicer.Corpid = svcCtx.Config.Wxwork.Corpid
+			cbServicer.Userid = a.Userid
+			cbServicer.Status = a.Status
+			cbServicer.DepartmentId = a.DepartmentId
+			now := time.Now()
+			cbServicer.CreatedAt = now
+			cbServicer.UpdatedAt = now
+			_, err = svcCtx.CbServicerModel.Insert(nil, cbServicer)
+			if err != nil {
+				logx.Error("Insert cb_servicer fail ", err)
 			}
 			//保存成员信息
 			staff, err := svcCtx.CbStaffModel.GetStaff(svcCtx.Config.Wxwork.Corpid, a.Userid)
@@ -240,3 +251,26 @@ func updateCustomerList(svcCtx *svc.ServiceContext, userIds pie.Strings) {
 		}
 	}
 }
+
+// 消息额外处理
+func msgTypeTextHandle(svcCtx *svc.ServiceContext, msg *model.CbMsg) {
+	return
+}
+
+// 事件消息额外处理
+func msgTypeEventHandle(svcCtx *svc.ServiceContext, msg *model.CbMsg) {
+	content, err := fastjson.Parse(msg.DataInfo)
+	if err != nil {
+		logx.Error("fastjson.Parse(cbMsg.DataInfo) fail ", err)
+	}
+	eventType := content.GetStringBytes("event_type")
+	switch string(eventType) {
+	case types.EventTypeEnterSession:
+		//进入会话事件,加入消息流转池
+		err := svcCtx.WxApi.UpdateSessionState(msg.OpenKfid, msg.ExternalUserid, "", 2)
+		if err != nil {
+			logx.Error("UpdateSessionState fail ", err)
+		}
+		return
+	}
+}

internal/types/const.go

@@ -4,5 +4,10 @@ const (
 	Event               = "event"
 	KfMsgOrEvent        = "kf_msg_or_event"
 	KfAccountAuthChange = "kf_account_auth_change"
-	SessionStatusChange = "session_status_change"
+
+	MsgTypeEvent = "event"
+	MsgTypeText  = "text"
+
+	EventTypeSessionStatusChange = "session_status_change"
+	EventTypeEnterSession        = "enter_session"
 )

model/cbservicermodel.go

@@ -13,6 +13,7 @@ type (
 	CbServicerModel interface {
 		cbServicerModel
 		GetServicer(openKfid, userid string) (d *CbServicer, err error)
+		DeleteServicerByOpenKfid(openKfid string) (err error)
 	}
 
 	customCbServicerModel struct {
@@ -34,3 +35,9 @@ func (m *customCbServicerModel) GetServicer(openKfid, userid string) (d *CbServi
 	d = &resp
 	return
 }
+
+func (m *customCbServicerModel) DeleteServicerByOpenKfid(openKfid string) (err error) {
+	query := fmt.Sprintf("delete from %s where `open_kfid` = ?", m.table)
+	_, err = m.conn.Exec(query, openKfid)
+	return err
+}

pkg/wxwork/wxapi/bean.go

@@ -90,3 +90,16 @@ type MsgListResp struct {
 	HasMore    int                      `json:"has_more"`
 	MsgList    []map[string]interface{} `json:"msg_list"`
 }
+
+type UpdateSessionStateParam struct {
+	OpenKfid       string `json:"open_kfid"`
+	ExternalUserid string `json:"external_userid"`
+	ServiceState   int    `json:"service_state"`
+	ServicerUserid string `json:"servicer_userid"`
+}
+
+type UpdateSessionStateResp struct {
+	Errcode int    `json:"errcode"`
+	Errmsg  string `json:"errmsg"`
+	MsgCode string `json:"msg_code"`
+}

pkg/wxwork/wxapi/wxapi.go

@@ -13,12 +13,14 @@ import (
 )
 
 const (
-	GetTokenUrl    = "https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=ID&corpsecret=SECRET"
-	GetServiceUrl  = "https://qyapi.weixin.qq.com/cgi-bin/kf/account/list?access_token=ACCESS_TOKEN"
-	GetServicerUrl = "https://qyapi.weixin.qq.com/cgi-bin/kf/servicer/list?access_token=ACCESS_TOKEN&open_kfid=OPEN_KFID"
-	GetStaffUrl    = "https://qyapi.weixin.qq.com/cgi-bin/user/get?access_token=ACCESS_TOKEN&userid=USERID"
-	GetCustomerUrl = "https://qyapi.weixin.qq.com/cgi-bin/kf/customer/batchget?access_token=ACCESS_TOKEN"
-	GetMsgUrl      = "https://qyapi.weixin.qq.com/cgi-bin/kf/sync_msg?access_token=ACCESS_TOKEN"
+	GetTokenUrl           = "https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=ID&corpsecret=SECRET"
+	GetServiceUrl         = "https://qyapi.weixin.qq.com/cgi-bin/kf/account/list?access_token=ACCESS_TOKEN"
+	GetServicerUrl        = "https://qyapi.weixin.qq.com/cgi-bin/kf/servicer/list?access_token=ACCESS_TOKEN&open_kfid=OPEN_KFID"
+	GetStaffUrl           = "https://qyapi.weixin.qq.com/cgi-bin/user/get?access_token=ACCESS_TOKEN&userid=USERID"
+	GetCustomerUrl        = "https://qyapi.weixin.qq.com/cgi-bin/kf/customer/batchget?access_token=ACCESS_TOKEN"
+	GetMsgUrl             = "https://qyapi.weixin.qq.com/cgi-bin/kf/sync_msg?access_token=ACCESS_TOKEN"
+	GetSessionStateUrl    = "https://qyapi.weixin.qq.com/cgi-bin/kf/service_state/get?access_token=ACCESS_TOKEN"
+	UpdateSessionStateUrl = "https://qyapi.weixin.qq.com/cgi-bin/kf/service_state/trans?access_token=ACCESS_TOKEN"
 )
 
 // 自定义封装企业微信的API接口
@@ -255,3 +257,38 @@ func (a *WxApi) GetMsgList(cursor, token, openKfid string) (err error, hasMore i
 	}
 	return
 }
+
+// UpdateSessionState 变更会话状态
+func (a *WxApi) UpdateSessionState(openKfid, externalUserid, servicerUserid string, serviceState int) (err error) {
+	err = a.refreshAccessToken()
+	if err != nil {
+		return
+	}
+	url := UpdateSessionStateUrl
+	url = strings.ReplaceAll(url, "ACCESS_TOKEN", a.accessToken)
+	body := UpdateSessionStateParam{
+		OpenKfid:       openKfid,
+		ExternalUserid: externalUserid,
+		ServiceState:   serviceState,
+		ServicerUserid: servicerUserid,
+	}
+	bts, _ := json.Marshal(body)
+	res, err := http.Post(url, "application/json", bytes.NewReader(bts))
+	if err != nil {
+		return
+	}
+	defer res.Body.Close()
+	bodyBts, err := ioutil.ReadAll(res.Body)
+	if err != nil {
+		return
+	}
+	var data UpdateSessionStateResp
+	err = json.Unmarshal(bodyBts, &data)
+	if err != nil {
+		return
+	}
+	if data.Errcode != 0 {
+		return errors.New(fmt.Sprintf("errcode: %v,errmsg: %v", data.Errcode, data.Errmsg))
+	}
+	return
+}

test/redis_test.go

@@ -21,3 +21,14 @@ func TestName(t *testing.T) {
 	}
 
 }
+func Test2(t *testing.T) {
+	var a = '好'
+	fmt.Printf("character %c,unicode %U binary %016b\n", a, a, a)
+	var b = "好"
+	for _, i := range []byte(b) {
+		fmt.Printf("binary %b\n", i)
+	}
+	for _, i := range b {
+		fmt.Printf("binary %b\n", i)
+	}
+}