update

consume/canal_b2c_order_for_numerical_statement_consume.go

@@ -0,0 +1,299 @@
+package consume
+
+import (
+	"applet/app/db"
+	"applet/app/utils"
+	"applet/app/utils/cache"
+	"applet/app/utils/logx"
+	"applet/consume/md"
+	"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
+	db2 "code.fnuoos.com/go_rely_warehouse/zyos_model.git/src/models/official"
+	"code.fnuoos.com/go_rely_warehouse/zyos_model.git/src/models/official/model"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"github.com/shopspring/decimal"
+	"github.com/streadway/amqp"
+	"strings"
+	"time"
+)
+
+const ZhiOsB2cPlaceOrderNumOfPeopleHashMapCacheKey = "%s:zhiOs_b2c_place_order_num_of_people_hash_map_cache:%s" //下单人数缓存hashMap键
+
+func CanalB2cOrderForNumericalStatementConsume(queue md.MqQueue) {
+	ch, err := rabbit.Cfg.Pool.GetChannel()
+	if err != nil {
+		logx.Error(err)
+		return
+	}
+	defer ch.Release()
+	//1、将自己绑定到交换机上
+	ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey)
+	//2、取出数据进行消费
+	ch.Qos(100)
+	delivery := ch.Consume(queue.Name, false)
+
+	var res amqp.Delivery
+	var ok bool
+	for {
+		res, ok = <-delivery
+		if ok == true {
+			//fmt.Println(string(res.Body))
+			fmt.Println(">>>>>>>>>>>>>>>>CanalB2cOrderForNumericalStatementConsume<<<<<<<<<<<<<<<<<<<<<<<<<")
+			err = handleCanalB2cOrderForNumericalStatementTable(res.Body)
+			if err != nil {
+				fmt.Println("err ::: ", err)
+				utils.FilePutContents("CanalB2cOrderForNumericalStatementConsume_ERR", "[err]:"+err.Error())
+				_ = res.Reject(false)
+				////TODO::重新推回队列末尾，避免造成队列堵塞
+				//var msg *md.OneCirclesStructForSignIn
+				//json.Unmarshal(res.Body, &msg)
+				//ch.Publish(queue.ExchangeName, msg, queue.RoutKey)
+			} else {
+				_ = res.Ack(true)
+			}
+		} else {
+			panic(errors.New("error getting message"))
+		}
+	}
+	fmt.Println("get msg done")
+}
+
+func handleCanalB2cOrderForNumericalStatementTable(msg []byte) error {
+	//1、解析canal采集至mq中queue的数据结构体
+	var canalMsg *md.CanalB2cOrderMessage[md.CanalB2cOrder]
+	err := json.Unmarshal(msg, &canalMsg)
+	if err != nil {
+		return err
+	}
+	now := time.Now()
+
+	//2、获取masterId
+	masterId := utils.StrToInt(strings.Split(canalMsg.Database, "_")[1])
+
+	//TODO::日志记录
+	utils.FilePutContents("handleCanalB2cOrderForNumericalStatementTable_"+utils.IntToStr(masterId)+"_"+now.Format("2006-01-02"), string(msg))
+
+	//2、判断操作(insert | update)
+	if canalMsg.Type == md.CanalMsgInsertSqlType || canalMsg.Type == md.CanalMsgUpdateSqlType {
+		var isUpdate bool
+		//3、查找是否有数据
+		var ordDate string
+		for _, item := range canalMsg.Data {
+			ordDate = utils.TimeParseStd(item.CreateTime).Format("2006-01-02")
+		}
+
+		statistics, err := db2.GetMasterB2cOrderStatistics(db.Db, masterId, ordDate)
+		if err != nil {
+			return err
+		}
+		if statistics == nil && canalMsg.Type == md.CanalMsgInsertSqlType {
+			statistics = &model.MasterB2cOrderStatistics{
+				MasterId:              masterId,
+				PaymentTotal:          "",
+				OrderCount:            0,
+				LoseOrderCount:        0,
+				PlaceOrderNumOfPeople: 0,
+				EffectiveOrderCount:   0,
+				EffectiveCommission:   "",
+				ReceiveCommission:     "",
+				LoseCommission:        "",
+				AvgCommission:         "",
+				CustomerUnitPrice:     "",
+				Date:                  ordDate,
+				CreateAt:              now.Format("2006-01-02 15:04:05"),
+				UpdateAt:              now.Format("2006-01-02 15:04:05"),
+			}
+			_, err = db2.MasterB2cOrderStatisticsInsert(db.Db, statistics)
+			if err != nil {
+				return err
+			}
+		}
+		if statistics == nil {
+			return errors.New("过期订单数据不予处理")
+		}
+
+		paymentTotal := statistics.PaymentTotal
+		orderCount := statistics.OrderCount
+		loseOrderCount := statistics.LoseOrderCount
+		placeOrderNumOfPeople := statistics.PlaceOrderNumOfPeople
+		effectiveOrderCount := statistics.EffectiveOrderCount
+		effectiveCommission := statistics.EffectiveCommission
+		loseCommission := statistics.LoseCommission
+		avgCommission := statistics.AvgCommission
+		customerUnitPrice := statistics.CustomerUnitPrice
+		effectivePaymentTotal := statistics.EffectivePaymentTotal
+
+		if canalMsg.Type == md.CanalMsgInsertSqlType {
+			for _, item := range canalMsg.Data {
+				orderCount++
+				if item.State == "0" {
+					//未支付不统计
+					return errors.New("未支付不统计")
+				} else if item.State == "6" {
+					//失效
+					loseOrderCount++
+				} else {
+					effectiveOrderCount++
+				}
+
+				paymentTotal = utils.Float64ToStr(utils.StrToFloat64(paymentTotal) + utils.StrToFloat64(item.CostPrice))                        //付款金额
+				effectiveCommission = utils.Float64ToStr(utils.StrToFloat64(effectiveCommission) + utils.StrToFloat64(item.EstimateCommission)) //有效佣金(元)
+				effectivePaymentTotal = utils.Float64ToStr(utils.StrToFloat64(effectivePaymentTotal) + utils.StrToFloat64(item.CostPrice))      //有效付款金额(元)
+
+				if item.State != "6" {
+					cacheKey := fmt.Sprintf(ZhiOsB2cPlaceOrderNumOfPeopleHashMapCacheKey, utils.IntToStr(masterId), ordDate)
+					get, _ := cache.HGetString(cacheKey, item.Uid)
+					if get == "" {
+						placeOrderNumOfPeople++ //下单人数
+						cache.HSet(cacheKey, item.Uid, "1")
+					} else {
+						cache.HSet(cacheKey, item.Uid, utils.IntToStr(utils.StrToInt(get)+1))
+					}
+					cache.Expire(cacheKey, md.ZhiOsUserVisitIpAddressHashMapCacheTime)
+					paymentTotalValue, _ := decimal.NewFromString(paymentTotal)
+					if placeOrderNumOfPeople == 0 {
+						return errors.New("divider cannot be 0 in division operation")
+					}
+					placeOrderNumOfPeopleValue := decimal.NewFromInt(int64(placeOrderNumOfPeople))
+					customerUnitPrice = paymentTotalValue.Div(placeOrderNumOfPeopleValue).String()
+				}
+				//客单价
+				isUpdate = true
+			}
+		}
+
+		if canalMsg.Type == md.CanalMsgUpdateSqlType {
+			judgeSate := JudgeB2cOrdSate(*canalMsg)
+			if judgeSate > 0 {
+				if judgeSate == 2 {
+					//TODO::未收货失效
+					loseOrderCount++
+					effectiveOrderCount--
+					for _, item := range canalMsg.Data {
+						loseCommission = utils.Float64ToStr(utils.StrToFloat64(loseCommission) + utils.StrToFloat64(item.EstimateCommission))
+						effectiveCommission = utils.Float64ToStr(utils.StrToFloat64(effectiveCommission) - utils.StrToFloat64(item.EstimateCommission))
+						effectivePaymentTotal = utils.Float64ToStr(utils.StrToFloat64(effectivePaymentTotal) - utils.StrToFloat64(item.CostPrice))
+						isUpdate = true
+					}
+				}
+				if judgeSate == 3 {
+					//TODO::已收货失效
+					loseOrderCount++
+					effectiveOrderCount--
+					for _, item := range canalMsg.Data {
+						loseCommission = utils.Float64ToStr(utils.StrToFloat64(loseCommission) + utils.StrToFloat64(item.EstimateCommission))
+						effectiveCommission = utils.Float64ToStr(utils.StrToFloat64(effectiveCommission) - utils.StrToFloat64(item.EstimateCommission))
+						effectivePaymentTotal = utils.Float64ToStr(utils.StrToFloat64(effectivePaymentTotal) - utils.StrToFloat64(item.CostPrice))
+						isUpdate = true
+					}
+				}
+
+				if judgeSate == 1 || judgeSate == 3 {
+					//TODO::收货额外处理
+
+					//查找是否有数据
+					var ordConfirmAt string
+					for _, item := range canalMsg.Data {
+						ordConfirmAt = utils.TimeParseStd(item.FinishTime).Format("2006-01-02")
+					}
+					ordConfirmStatistics, err := db2.GetMasterB2cOrderStatistics(db.Db, masterId, ordConfirmAt)
+					if err != nil {
+						return err
+					}
+					if ordConfirmStatistics == nil {
+						return errors.New("过期收货订单数据不予处理")
+					}
+
+					if judgeSate == 1 {
+						//TODO::收货
+						for _, item := range canalMsg.Data {
+							ordConfirmStatistics.ReceiveCommission = utils.Float64ToStr(utils.StrToFloat64(ordConfirmStatistics.ReceiveCommission) + utils.StrToFloat64(item.EstimateCommission))
+						}
+					}
+					if judgeSate == 3 {
+						//TODO::已收货失效
+						for _, item := range canalMsg.Data {
+							ordConfirmStatistics.ReceiveCommission = utils.Float64ToStr(utils.StrToFloat64(ordConfirmStatistics.ReceiveCommission) - utils.StrToFloat64(item.EstimateCommission))
+						}
+					}
+					ordConfirmStatistics.UpdateAt = now.Format("2006-01-02 15:04:05")
+					_, err = db2.MasterB2cOrderStatisticsUpdate(db.Db, ordConfirmStatistics.Id, ordConfirmStatistics,
+						"receive_commission", "update_at")
+				}
+			}
+		}
+
+		if isUpdate {
+			statistics.PaymentTotal = paymentTotal
+			statistics.OrderCount = orderCount
+			statistics.LoseOrderCount = loseOrderCount
+			statistics.PlaceOrderNumOfPeople = placeOrderNumOfPeople
+			statistics.EffectiveOrderCount = effectiveOrderCount
+			statistics.EffectiveCommission = effectiveCommission
+			statistics.LoseCommission = loseCommission
+			statistics.AvgCommission = avgCommission
+			statistics.CustomerUnitPrice = customerUnitPrice
+			statistics.EffectivePaymentTotal = effectivePaymentTotal
+			statistics.UpdateAt = now.Format("2006-01-02 15:04:05")
+			_, err = db2.MasterB2cOrderStatisticsUpdate(db.Db, statistics.Id, statistics,
+				"payment_total", "order_count", "estimated_profit", "lose_order_count",
+				"place_order_num_of_people", "effective_order_count", "effective_commission", "lose_commission",
+				"avg_commission", "customer_unit_price", "effective_payment_total", "update_at")
+			if err != nil {
+				return err
+			}
+		}
+
+	}
+	return nil
+}
+
+// JudgeB2cOrdSate 处理订单状态(judgeSate[0:不需要处理 1:收货 2:未收货失效 3:已收货失效])
+func JudgeB2cOrdSate(message md.CanalB2cOrderMessage[md.CanalB2cOrder]) (judgeSate int) {
+	oldData := message.Old
+
+	//1、获取 旧的订单状态
+	var oldOrdState string
+	for _, item := range oldData {
+		if item.State != "" {
+			oldOrdState = item.State
+		}
+	}
+	if oldOrdState == "" {
+		return
+	}
+
+	//2、获取 新的订单状态
+	var nowOrdState string
+	for _, item := range message.Data {
+		if item.State != "" {
+			nowOrdState = item.State
+		}
+	}
+	if nowOrdState == "" {
+		return
+	}
+
+	if oldOrdState == nowOrdState {
+		return
+	}
+
+	//3、进行状态比较判断
+	if oldOrdState == "3" || oldOrdState == "4" || oldOrdState == "5" {
+		if nowOrdState == "6" {
+			//已收货失效
+			return 3
+		}
+	} else {
+		if nowOrdState == "6" {
+			//未收货失效
+			return 2
+		}
+		if nowOrdState == "3" || nowOrdState == "4" || nowOrdState == "5" {
+			//收货
+			return 1
+		}
+	}
+	return
+}

consume/canal_guide_order_for_numerical_statement_consume.go

@@ -135,8 +135,6 @@ func handleCanalGuideOrderForNumericalStatementTable(msg []byte) error {
 					effectiveOrderCount++
 				}
 
-				cacheKey := fmt.Sprintf(ZhiOsGuidePlaceOrderNumOfPeopleHashMapCacheKey, utils.IntToStr(masterId), ordDate)
-				get, _ := cache.HGetString(cacheKey, item.Uid)
 				paymentTotal = utils.Float64ToStr(utils.StrToFloat64(paymentTotal) + utils.StrToFloat64(item.PaidPrice))                   //付款金额
 				estimatedCommission = utils.Float64ToStr(utils.StrToFloat64(estimatedCommission) + utils.StrToFloat64(item.BenefitAll))    //预估佣金(元)
 				estimatedProfit = utils.Float64ToStr(utils.StrToFloat64(estimatedProfit) + utils.StrToFloat64(item.SysCommission))         //预估利润(元)
@@ -147,19 +145,23 @@ func handleCanalGuideOrderForNumericalStatementTable(msg []byte) error {
 				orderCountValue := decimal.NewFromInt(int64(orderCount))
 				avgCommission = estimatedCommissionValue.Div(orderCountValue).String() //平均佣金
 
-				if get == "" {
-					placeOrderNumOfPeople++ //下单人数
-					cache.HSet(cacheKey, item.Uid, "1")
-				} else {
-					cache.HSet(cacheKey, item.Uid, utils.IntToStr(utils.StrToInt(get)+1))
-				}
-				cache.Expire(cacheKey, md.ZhiOsUserVisitIpAddressHashMapCacheTime)
-				paymentTotalValue, _ := decimal.NewFromString(paymentTotal)
-				if placeOrderNumOfPeople == 0 {
-					return errors.New("divider cannot be 0 in division operation")
+				if item.State != "4" {
+					cacheKey := fmt.Sprintf(ZhiOsGuidePlaceOrderNumOfPeopleHashMapCacheKey, utils.IntToStr(masterId), ordDate)
+					get, _ := cache.HGetString(cacheKey, item.Uid)
+					if get == "" {
+						placeOrderNumOfPeople++ //下单人数
+						cache.HSet(cacheKey, item.Uid, "1")
+					} else {
+						cache.HSet(cacheKey, item.Uid, utils.IntToStr(utils.StrToInt(get)+1))
+					}
+					cache.Expire(cacheKey, md.ZhiOsUserVisitIpAddressHashMapCacheTime)
+					paymentTotalValue, _ := decimal.NewFromString(paymentTotal)
+					if placeOrderNumOfPeople == 0 {
+						return errors.New("divider cannot be 0 in division operation")
+					}
+					placeOrderNumOfPeopleValue := decimal.NewFromInt(int64(placeOrderNumOfPeople))
+					customerUnitPrice = paymentTotalValue.Div(placeOrderNumOfPeopleValue).String() //客单价
 				}
-				placeOrderNumOfPeopleValue := decimal.NewFromInt(int64(placeOrderNumOfPeople))
-				customerUnitPrice = paymentTotalValue.Div(placeOrderNumOfPeopleValue).String() //客单价
 				isUpdate = true
 			}
 		}

consume/canal_mall_order_for_numerical_statement_consume.go

@@ -1,282 +1,307 @@
 package consume
 
+import (
+	"applet/app/db"
+	"applet/app/utils"
+	"applet/app/utils/cache"
+	"applet/app/utils/logx"
+	"applet/consume/md"
+	"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
+	db2 "code.fnuoos.com/go_rely_warehouse/zyos_model.git/src/models/official"
+	"code.fnuoos.com/go_rely_warehouse/zyos_model.git/src/models/official/model"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"github.com/shopspring/decimal"
+	"github.com/streadway/amqp"
+	"strings"
+	"time"
+)
+
 const ZhiOsMallPlaceOrderNumOfPeopleHashMapCacheKey = "%s:zhiOs_mall_place_order_num_of_people_hash_map_cache:%s" //下单人数缓存hashMap键
 
-//func CanalMallOrderForNumericalStatementConsume(queue md.MqQueue) {
-//	ch, err := rabbit.Cfg.Pool.GetChannel()
-//	if err != nil {
-//		logx.Error(err)
-//		return
-//	}
-//	defer ch.Release()
-//	//1、将自己绑定到交换机上
-//	ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey)
-//	//2、取出数据进行消费
-//	ch.Qos(100)
-//	delivery := ch.Consume(queue.Name, false)
-//
-//	var res amqp.Delivery
-//	var ok bool
-//	for {
-//		res, ok = <-delivery
-//		if ok == true {
-//			//fmt.Println(string(res.Body))
-//			fmt.Println(">>>>>>>>>>>>>>>>CanalGuideOrderForNumericalStatementConsume<<<<<<<<<<<<<<<<<<<<<<<<<")
-//			err = handleCanalGuideOrderForNumericalStatementTable(res.Body)
-//			if err != nil {
-//				fmt.Println("err ::: ", err)
-//				utils.FilePutContents("CanalGuideOrderForNumericalStatementConsume_ERR", "[err]:"+err.Error())
-//				_ = res.Reject(false)
-//				////TODO::重新推回队列末尾，避免造成队列堵塞
-//				//var msg *md.OneCirclesStructForSignIn
-//				//json.Unmarshal(res.Body, &msg)
-//				//ch.Publish(queue.ExchangeName, msg, queue.RoutKey)
-//			} else {
-//				_ = res.Ack(true)
-//			}
-//		} else {
-//			panic(errors.New("error getting message"))
-//		}
-//	}
-//	fmt.Println("get msg done")
-//}
-//
-//func handleCanalMallOrderForNumericalStatementTable(msg []byte) error {
-//	//1、解析canal采集至mq中queue的数据结构体
-//	var canalMsg *md.CanalMallOrder
-//	err := json.Unmarshal(msg, &canalMsg)
-//	if err != nil {
-//		return err
-//	}
-//	now := time.Now()
-//
-//	//2、获取masterId
-//	masterId := utils.StrToInt(strings.Split(canalMsg.Database, "_")[1])
-//
-//	//TODO::日志记录
-//	utils.FilePutContents("handleCanalGuideOrderForNumericalStatementTable_"+utils.IntToStr(masterId)+"_"+now.Format("2006-01-02"), string(msg))
-//
-//	//2、判断操作(insert | update)
-//	if canalMsg.Type == md.CanalMsgInsertSqlType || canalMsg.Type == md.CanalMsgUpdateSqlType {
-//		var isUpdate bool
-//		//3、查找是否有数据
-//		var ordDate string
-//		for _, item := range canalMsg.Data {
-//			ordDate = time.Unix(utils.StrToInt64(item.CreateAt), 0).Format("2006-01-02")
-//		}
-//		statistics, err := db2.GetMasterGuideOrderStatistics(db.Db, masterId, ordDate)
-//		if err != nil {
-//			return err
-//		}
-//		if statistics == nil && canalMsg.Type == md.CanalMsgInsertSqlType {
-//			statistics = &model.MasterGuideOrderStatistics{
-//				MasterId:              masterId,
-//				PaymentTotal:          "",
-//				OrderCount:            0,
-//				EstimatedCommission:   "",
-//				EstimatedProfit:       "",
-//				LoseOrderCount:        0,
-//				PlaceOrderNumOfPeople: 0,
-//				EffectiveOrderCount:   0,
-//				EffectiveCommission:   "",
-//				ReceiveCommission:     "",
-//				LoseCommission:        "",
-//				AvgCommission:         "",
-//				CustomerUnitPrice:     "",
-//				Date:                  ordDate,
-//				CreateAt:              now.Format("2006-01-02 15:04:05"),
-//				UpdateAt:              now.Format("2006-01-02 15:04:05"),
-//			}
-//			_, err = db2.MasterGuideOrderStatisticsInsert(db.Db, statistics)
-//			if err != nil {
-//				return err
-//			}
-//		}
-//		if statistics == nil {
-//			return errors.New("过期订单数据不予处理")
-//		}
-//
-//		paymentTotal := statistics.PaymentTotal
-//		orderCount := statistics.OrderCount
-//		estimatedCommission := statistics.EstimatedCommission
-//		estimatedProfit := statistics.EstimatedProfit
-//		loseOrderCount := statistics.LoseOrderCount
-//		placeOrderNumOfPeople := statistics.PlaceOrderNumOfPeople
-//		effectiveOrderCount := statistics.EffectiveOrderCount
-//		effectiveCommission := statistics.EffectiveCommission
-//		loseCommission := statistics.LoseCommission
-//		avgCommission := statistics.AvgCommission
-//		customerUnitPrice := statistics.CustomerUnitPrice
-//		effectivePaymentTotal := statistics.EffectivePaymentTotal
-//		if canalMsg.Type == md.CanalMsgInsertSqlType {
-//			for _, item := range canalMsg.Data {
-//				orderCount++
-//				if item.State == "4" {
-//					loseOrderCount++
-//				} else {
-//					effectiveOrderCount++
-//				}
-//
-//				cacheKey := fmt.Sprintf(ZhiOsGuidePlaceOrderNumOfPeopleHashMapCacheKey, utils.IntToStr(masterId), ordDate)
-//				get, _ := cache.HGetString(cacheKey, item.Uid)
-//				paymentTotal = utils.Float64ToStr(utils.StrToFloat64(paymentTotal) + utils.StrToFloat64(item.PaidPrice))                   //付款金额
-//				estimatedCommission = utils.Float64ToStr(utils.StrToFloat64(estimatedCommission) + utils.StrToFloat64(item.BenefitAll))    //预估佣金(元)
-//				estimatedProfit = utils.Float64ToStr(utils.StrToFloat64(estimatedProfit) + utils.StrToFloat64(item.SysCommission))         //预估利润(元)
-//				effectiveCommission = utils.Float64ToStr(utils.StrToFloat64(effectiveCommission) + utils.StrToFloat64(item.BenefitAll))    //有效佣金(元)
-//				effectivePaymentTotal = utils.Float64ToStr(utils.StrToFloat64(effectivePaymentTotal) + utils.StrToFloat64(item.PaidPrice)) //有效付款金额(元)
-//
-//				estimatedCommissionValue, _ := decimal.NewFromString(estimatedCommission)
-//				orderCountValue := decimal.NewFromInt(int64(orderCount))
-//				avgCommission = estimatedCommissionValue.Div(orderCountValue).String() //平均佣金
-//
-//				if get == "" {
-//					placeOrderNumOfPeople++ //下单人数
-//					cache.HSet(cacheKey, item.Uid, "1")
-//				} else {
-//					cache.HSet(cacheKey, item.Uid, utils.IntToStr(utils.StrToInt(get)+1))
-//				}
-//				cache.Expire(cacheKey, md.ZhiOsUserVisitIpAddressHashMapCacheTime)
-//				paymentTotalValue, _ := decimal.NewFromString(paymentTotal)
-//				if placeOrderNumOfPeople == 0 {
-//					return errors.New("divider cannot be 0 in division operation")
-//				}
-//				placeOrderNumOfPeopleValue := decimal.NewFromInt(int64(placeOrderNumOfPeople))
-//				customerUnitPrice = paymentTotalValue.Div(placeOrderNumOfPeopleValue).String() //客单价
-//				isUpdate = true
-//			}
-//		}
-//
-//		if canalMsg.Type == md.CanalMsgUpdateSqlType {
-//			judgeSate := JudgeSate(*canalMsg)
-//			if judgeSate > 0 {
-//				if judgeSate == 2 {
-//					//TODO::未收货失效
-//					loseOrderCount++
-//					effectiveOrderCount--
-//					for _, item := range canalMsg.Data {
-//						loseCommission = utils.Float64ToStr(utils.StrToFloat64(loseCommission) + utils.StrToFloat64(item.BenefitAll))
-//						effectiveCommission = utils.Float64ToStr(utils.StrToFloat64(effectiveCommission) - utils.StrToFloat64(item.BenefitAll))
-//						effectivePaymentTotal = utils.Float64ToStr(utils.StrToFloat64(effectivePaymentTotal) - utils.StrToFloat64(item.PaidPrice))
-//						isUpdate = true
-//					}
-//				}
-//				if judgeSate == 3 {
-//					//TODO::已收货失效
-//					loseOrderCount++
-//					effectiveOrderCount--
-//					for _, item := range canalMsg.Data {
-//						loseCommission = utils.Float64ToStr(utils.StrToFloat64(loseCommission) + utils.StrToFloat64(item.BenefitAll))
-//						effectiveCommission = utils.Float64ToStr(utils.StrToFloat64(effectiveCommission) - utils.StrToFloat64(item.BenefitAll))
-//						effectivePaymentTotal = utils.Float64ToStr(utils.StrToFloat64(effectivePaymentTotal) - utils.StrToFloat64(item.PaidPrice))
-//						isUpdate = true
-//					}
-//				}
-//
-//				if judgeSate == 1 || judgeSate == 3 {
-//					//TODO::收货额外处理
-//
-//					//查找是否有数据
-//					var ordConfirmAt string
-//					for _, item := range canalMsg.Data {
-//						ordConfirmAt = time.Unix(utils.StrToInt64(item.ConfirmAt), 0).Format("2006-01-02")
-//					}
-//					ordConfirmStatistics, err := db2.GetMasterGuideOrderStatistics(db.Db, masterId, ordConfirmAt)
-//					if err != nil {
-//						return err
-//					}
-//					if ordConfirmStatistics == nil {
-//						return errors.New("过期收货订单数据不予处理")
-//					}
-//
-//					if judgeSate == 1 {
-//						//TODO::收货
-//						for _, item := range canalMsg.Data {
-//							ordConfirmStatistics.ReceiveCommission = utils.Float64ToStr(utils.StrToFloat64(ordConfirmStatistics.ReceiveCommission) + utils.StrToFloat64(item.BenefitAll))
-//						}
-//					}
-//					if judgeSate == 3 {
-//						//TODO::已收货失效
-//						for _, item := range canalMsg.Data {
-//							ordConfirmStatistics.ReceiveCommission = utils.Float64ToStr(utils.StrToFloat64(ordConfirmStatistics.ReceiveCommission) - utils.StrToFloat64(item.BenefitAll))
-//						}
-//					}
-//					ordConfirmStatistics.UpdateAt = now.Format("2006-01-02 15:04:05")
-//					_, err = db2.MasterGuideOrderStatisticsUpdate(db.Db, ordConfirmStatistics.Id, ordConfirmStatistics,
-//						"receive_commission", "update_at")
-//				}
-//			}
-//		}
-//
-//		if isUpdate {
-//			statistics.PaymentTotal = paymentTotal
-//			statistics.OrderCount = orderCount
-//			statistics.EstimatedCommission = estimatedCommission
-//			statistics.EstimatedProfit = estimatedProfit
-//			statistics.LoseOrderCount = loseOrderCount
-//			statistics.PlaceOrderNumOfPeople = placeOrderNumOfPeople
-//			statistics.EffectiveOrderCount = effectiveOrderCount
-//			statistics.EffectiveCommission = effectiveCommission
-//			statistics.LoseCommission = loseCommission
-//			statistics.AvgCommission = avgCommission
-//			statistics.CustomerUnitPrice = customerUnitPrice
-//			statistics.EffectivePaymentTotal = effectivePaymentTotal
-//			statistics.UpdateAt = now.Format("2006-01-02 15:04:05")
-//			_, err = db2.MasterGuideOrderStatisticsUpdate(db.Db, statistics.Id, statistics,
-//				"payment_total", "order_count", "estimated_commission", "estimated_profit", "lose_order_count",
-//				"place_order_num_of_people", "effective_order_count", "effective_commission", "lose_commission",
-//				"avg_commission", "customer_unit_price", "effective_payment_total", "update_at")
-//			if err != nil {
-//				return err
-//			}
-//		}
-//
-//	}
-//	return nil
-//}
-//
-//
-//// JudgeSate 处理订单状态(judgeSate[0:不需要处理 1:收货 2:未收货失效 3:已收货失效])
-//func JudgeSate(message md.CanalGuideOrderMessage[md.CanalGuideOrder]) (judgeSate int) {
-//	oldData := message.Old
-//
-//	//1、获取 旧的订单状态
-//	var oldOrdState string
-//	for _, item := range oldData {
-//		if item.State != "" {
-//			oldOrdState = item.State
-//		}
-//	}
-//	if oldOrdState == "" {
-//		return
-//	}
-//
-//	//2、获取 新的订单状态
-//	var nowOrdState string
-//	for _, item := range message.Data {
-//		if item.State != "" {
-//			nowOrdState = item.State
-//		}
-//	}
-//	if nowOrdState == "" {
-//		return
-//	}
-//
-//	if oldOrdState == nowOrdState {
-//		return
-//	}
-//
-//	//3、进行状态比较判断
-//	if oldOrdState == "0" {
-//		if nowOrdState == "4" { //未收货失效
-//			return 2
-//		}
-//		if nowOrdState == "1" || nowOrdState == "2" || nowOrdState == "3" || nowOrdState == "5" { //收货
-//			return 1
-//		}
-//	} else {
-//		if nowOrdState == "4" { //已收货失效
-//			return 3
-//		}
-//	}
-//	return
-//}
+func CanalMallOrderForNumericalStatementConsume(queue md.MqQueue) {
+	ch, err := rabbit.Cfg.Pool.GetChannel()
+	if err != nil {
+		logx.Error(err)
+		return
+	}
+	defer ch.Release()
+	//1、将自己绑定到交换机上
+	ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey)
+	//2、取出数据进行消费
+	ch.Qos(1)
+	delivery := ch.Consume(queue.Name, false)
+
+	var res amqp.Delivery
+	var ok bool
+	for {
+		res, ok = <-delivery
+		if ok == true {
+			//fmt.Println(string(res.Body))
+			fmt.Println(">>>>>>>>>>>>>>>>CanalMallOrderForNumericalStatementConsume<<<<<<<<<<<<<<<<<<<<<<<<<")
+			err = handleCanalMallOrderForNumericalStatementTable(res.Body)
+			if err != nil {
+				fmt.Println("err ::: ", err)
+				utils.FilePutContents("CanalMallOrderForNumericalStatementConsume_ERR", "[err]:"+err.Error())
+				_ = res.Reject(false)
+				////TODO::重新推回队列末尾，避免造成队列堵塞
+				//var msg *md.OneCirclesStructForSignIn
+				//json.Unmarshal(res.Body, &msg)
+				//ch.Publish(queue.ExchangeName, msg, queue.RoutKey)
+			} else {
+				_ = res.Ack(true)
+			}
+		} else {
+			panic(errors.New("error getting message"))
+		}
+	}
+	fmt.Println("get msg done")
+}
+
+func handleCanalMallOrderForNumericalStatementTable(msg []byte) error {
+	//1、解析canal采集至mq中queue的数据结构体
+	var canalMsg *md.CanalMallOrder
+	err := json.Unmarshal(msg, &canalMsg)
+	if err != nil {
+		return err
+	}
+	now := time.Now()
+
+	//2、获取masterId
+	masterId := utils.StrToInt(strings.Split(canalMsg.Database, "_")[1])
+
+	//TODO::日志记录
+	utils.FilePutContents("handleCanalMallOrderForNumericalStatementTable_"+utils.IntToStr(masterId)+"_"+now.Format("2006-01-02"), string(msg))
+
+	//2、判断操作(insert | update)
+	if canalMsg.Type == md.CanalMsgInsertSqlType || canalMsg.Type == md.CanalMsgUpdateSqlType {
+		var isUpdate bool
+		//3、查找是否有数据
+		var ordDate string
+		for _, item := range canalMsg.Data {
+			ordDate = utils.TimeParseStd(item.CreateTime).Format("2006-01-02")
+		}
+
+		statistics, err := db2.GetMasterMallOrderStatistics(db.Db, masterId, ordDate)
+		if err != nil {
+			return err
+		}
+		if statistics == nil && canalMsg.Type == md.CanalMsgInsertSqlType {
+			statistics = &model.MasterMallOrderStatistics{
+				MasterId:              masterId,
+				PaymentTotal:          "",
+				OrderCount:            0,
+				EstimatedCommission:   "",
+				LoseOrderCount:        0,
+				PlaceOrderNumOfPeople: 0,
+				EffectiveOrderCount:   0,
+				EffectiveCommission:   "",
+				ReceiveCommission:     "",
+				LoseCommission:        "",
+				AvgCommission:         "",
+				CustomerUnitPrice:     "",
+				Date:                  ordDate,
+				CreateAt:              now.Format("2006-01-02 15:04:05"),
+				UpdateAt:              now.Format("2006-01-02 15:04:05"),
+			}
+			_, err = db2.MasterMallOrderStatisticsInsert(db.Db, statistics)
+			if err != nil {
+				return err
+			}
+		}
+		if statistics == nil {
+			return errors.New("过期订单数据不予处理")
+		}
+
+		paymentTotal := statistics.PaymentTotal
+		orderCount := statistics.OrderCount
+		estimatedCommission := statistics.EstimatedCommission
+		loseOrderCount := statistics.LoseOrderCount
+		placeOrderNumOfPeople := statistics.PlaceOrderNumOfPeople
+		effectiveOrderCount := statistics.EffectiveOrderCount
+		effectiveCommission := statistics.EffectiveCommission
+		loseCommission := statistics.LoseCommission
+		avgCommission := statistics.AvgCommission
+		customerUnitPrice := statistics.CustomerUnitPrice
+		effectivePaymentTotal := statistics.EffectivePaymentTotal
+
+		if canalMsg.Type == md.CanalMsgInsertSqlType {
+			for _, item := range canalMsg.Data {
+				orderCount++
+				if item.State == "0" {
+					//未支付不统计
+					return errors.New("未支付不统计")
+				} else if item.State == "6" {
+					//失效
+					loseOrderCount++
+				} else {
+					effectiveOrderCount++
+				}
+
+				paymentTotal = utils.Float64ToStr(utils.StrToFloat64(paymentTotal) + utils.StrToFloat64(item.CostPrice))                        //付款金额
+				estimatedCommission = utils.Float64ToStr(utils.StrToFloat64(estimatedCommission) + utils.StrToFloat64(item.EstimateCommission)) //预估佣金(元)
+				effectiveCommission = utils.Float64ToStr(utils.StrToFloat64(effectiveCommission) + utils.StrToFloat64(item.EstimateCommission)) //有效佣金(元)
+				effectivePaymentTotal = utils.Float64ToStr(utils.StrToFloat64(effectivePaymentTotal) + utils.StrToFloat64(item.CostPrice))      //有效付款金额(元)
+
+				estimatedCommissionValue, _ := decimal.NewFromString(estimatedCommission)
+				orderCountValue := decimal.NewFromInt(int64(orderCount))
+				avgCommission = estimatedCommissionValue.Div(orderCountValue).String() //平均佣金
+
+				if item.State != "6" {
+					cacheKey := fmt.Sprintf(ZhiOsMallPlaceOrderNumOfPeopleHashMapCacheKey, utils.IntToStr(masterId), ordDate)
+					get, _ := cache.HGetString(cacheKey, item.Uid)
+					if get == "" {
+						placeOrderNumOfPeople++ //下单人数
+						cache.HSet(cacheKey, item.Uid, "1")
+					} else {
+						cache.HSet(cacheKey, item.Uid, utils.IntToStr(utils.StrToInt(get)+1))
+					}
+					cache.Expire(cacheKey, md.ZhiOsUserVisitIpAddressHashMapCacheTime)
+					paymentTotalValue, _ := decimal.NewFromString(paymentTotal)
+					if placeOrderNumOfPeople == 0 {
+						return errors.New("divider cannot be 0 in division operation")
+					}
+					placeOrderNumOfPeopleValue := decimal.NewFromInt(int64(placeOrderNumOfPeople))
+					customerUnitPrice = paymentTotalValue.Div(placeOrderNumOfPeopleValue).String()
+				}
+				//客单价
+				isUpdate = true
+			}
+		}
+
+		if canalMsg.Type == md.CanalMsgUpdateSqlType {
+			judgeSate := JudgeMallOrdSate(*canalMsg)
+			if judgeSate > 0 {
+				if judgeSate == 2 {
+					//TODO::未收货失效
+					loseOrderCount++
+					effectiveOrderCount--
+					for _, item := range canalMsg.Data {
+						loseCommission = utils.Float64ToStr(utils.StrToFloat64(loseCommission) + utils.StrToFloat64(item.EstimateCommission))
+						effectiveCommission = utils.Float64ToStr(utils.StrToFloat64(effectiveCommission) - utils.StrToFloat64(item.EstimateCommission))
+						effectivePaymentTotal = utils.Float64ToStr(utils.StrToFloat64(effectivePaymentTotal) - utils.StrToFloat64(item.CostPrice))
+						isUpdate = true
+					}
+				}
+				if judgeSate == 3 {
+					//TODO::已收货失效
+					loseOrderCount++
+					effectiveOrderCount--
+					for _, item := range canalMsg.Data {
+						loseCommission = utils.Float64ToStr(utils.StrToFloat64(loseCommission) + utils.StrToFloat64(item.EstimateCommission))
+						effectiveCommission = utils.Float64ToStr(utils.StrToFloat64(effectiveCommission) - utils.StrToFloat64(item.EstimateCommission))
+						effectivePaymentTotal = utils.Float64ToStr(utils.StrToFloat64(effectivePaymentTotal) - utils.StrToFloat64(item.CostPrice))
+						isUpdate = true
+					}
+				}
+
+				if judgeSate == 1 || judgeSate == 3 {
+					//TODO::收货额外处理
+
+					//查找是否有数据
+					var ordConfirmAt string
+					for _, item := range canalMsg.Data {
+						ordConfirmAt = utils.TimeParseStd(item.ConfirmTime).Format("2006-01-02")
+					}
+					ordConfirmStatistics, err := db2.GetMasterMallOrderStatistics(db.Db, masterId, ordConfirmAt)
+					if err != nil {
+						return err
+					}
+					if ordConfirmStatistics == nil {
+						return errors.New("过期收货订单数据不予处理")
+					}
+
+					if judgeSate == 1 {
+						//TODO::收货
+						for _, item := range canalMsg.Data {
+							ordConfirmStatistics.ReceiveCommission = utils.Float64ToStr(utils.StrToFloat64(ordConfirmStatistics.ReceiveCommission) + utils.StrToFloat64(item.EstimateCommission))
+						}
+					}
+					if judgeSate == 3 {
+						//TODO::已收货失效
+						for _, item := range canalMsg.Data {
+							ordConfirmStatistics.ReceiveCommission = utils.Float64ToStr(utils.StrToFloat64(ordConfirmStatistics.ReceiveCommission) - utils.StrToFloat64(item.EstimateCommission))
+						}
+					}
+					ordConfirmStatistics.UpdateAt = now.Format("2006-01-02 15:04:05")
+					_, err = db2.MasterMallOrderStatisticsUpdate(db.Db, ordConfirmStatistics.Id, ordConfirmStatistics,
+						"receive_commission", "update_at")
+				}
+			}
+		}
+
+		if isUpdate {
+			statistics.PaymentTotal = paymentTotal
+			statistics.OrderCount = orderCount
+			statistics.EstimatedCommission = estimatedCommission
+			statistics.LoseOrderCount = loseOrderCount
+			statistics.PlaceOrderNumOfPeople = placeOrderNumOfPeople
+			statistics.EffectiveOrderCount = effectiveOrderCount
+			statistics.EffectiveCommission = effectiveCommission
+			statistics.LoseCommission = loseCommission
+			statistics.AvgCommission = avgCommission
+			statistics.CustomerUnitPrice = customerUnitPrice
+			statistics.EffectivePaymentTotal = effectivePaymentTotal
+			statistics.UpdateAt = now.Format("2006-01-02 15:04:05")
+			_, err = db2.MasterMallOrderStatisticsUpdate(db.Db, statistics.Id, statistics,
+				"payment_total", "order_count", "estimated_commission", "lose_order_count",
+				"place_order_num_of_people", "effective_order_count", "effective_commission", "lose_commission",
+				"avg_commission", "customer_unit_price", "effective_payment_total", "update_at")
+			if err != nil {
+				return err
+			}
+		}
+
+	}
+	return nil
+}
+
+// JudgeMallOrdSate 处理订单状态(judgeSate[0:不需要处理 1:收货 2:未收货失效 3:已收货失效])
+func JudgeMallOrdSate(message md.CanalMallOrder) (judgeSate int) {
+	oldData := message.Old
+
+	//1、获取 旧的订单状态
+	var oldOrdState string
+	for _, item := range oldData {
+		if item.State != "" {
+			oldOrdState = item.State
+		}
+	}
+	if oldOrdState == "" {
+		return
+	}
+
+	//2、获取 新的订单状态
+	var nowOrdState string
+	for _, item := range message.Data {
+		if item.State != "" {
+			nowOrdState = item.State
+		}
+	}
+	if nowOrdState == "" {
+		return
+	}
+
+	if oldOrdState == nowOrdState {
+		return
+	}
+
+	//3、进行状态比较判断
+	if oldOrdState == "3" || oldOrdState == "4" || oldOrdState == "5" {
+		if nowOrdState == "6" {
+			//已收货失效
+			return 3
+		}
+	} else {
+		if nowOrdState == "6" {
+			//未收货失效
+			return 2
+		}
+		if nowOrdState == "3" || nowOrdState == "4" || nowOrdState == "5" {
+			//收货
+			return 1
+		}
+	}
+	return
+}

consume/canal_o2o_order_for_numerical_statement_consume.go

@@ -0,0 +1,307 @@
+package consume
+
+import (
+	"applet/app/db"
+	"applet/app/utils"
+	"applet/app/utils/cache"
+	"applet/app/utils/logx"
+	"applet/consume/md"
+	"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
+	db2 "code.fnuoos.com/go_rely_warehouse/zyos_model.git/src/models/official"
+	"code.fnuoos.com/go_rely_warehouse/zyos_model.git/src/models/official/model"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"github.com/shopspring/decimal"
+	"github.com/streadway/amqp"
+	"strings"
+	"time"
+)
+
+const ZhiOsO2oPlaceOrderNumOfPeopleHashMapCacheKey = "%s:zhiOs_o2o_place_order_num_of_people_hash_map_cache:%s" //下单人数缓存hashMap键
+
+func CanalO2oOrderForNumericalStatementConsume(queue md.MqQueue) {
+	ch, err := rabbit.Cfg.Pool.GetChannel()
+	if err != nil {
+		logx.Error(err)
+		return
+	}
+	defer ch.Release()
+	//1、将自己绑定到交换机上
+	ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey)
+	//2、取出数据进行消费
+	ch.Qos(100)
+	delivery := ch.Consume(queue.Name, false)
+
+	var res amqp.Delivery
+	var ok bool
+	for {
+		res, ok = <-delivery
+		if ok == true {
+			//fmt.Println(string(res.Body))
+			fmt.Println(">>>>>>>>>>>>>>>>CanalO2oOrderForNumericalStatementConsume<<<<<<<<<<<<<<<<<<<<<<<<<")
+			err = handleCanalO2oOrderForNumericalStatementTable(res.Body)
+			if err != nil {
+				fmt.Println("err ::: ", err)
+				utils.FilePutContents("CanalO2oOrderForNumericalStatementConsume_ERR", "[err]:"+err.Error())
+				_ = res.Reject(false)
+				////TODO::重新推回队列末尾，避免造成队列堵塞
+				//var msg *md.OneCirclesStructForSignIn
+				//json.Unmarshal(res.Body, &msg)
+				//ch.Publish(queue.ExchangeName, msg, queue.RoutKey)
+			} else {
+				_ = res.Ack(true)
+			}
+		} else {
+			panic(errors.New("error getting message"))
+		}
+	}
+	fmt.Println("get msg done")
+}
+
+func handleCanalO2oOrderForNumericalStatementTable(msg []byte) error {
+	//1、解析canal采集至mq中queue的数据结构体
+	var canalMsg *md.CanalO2oOrderMessage[md.CanalO2oOrder]
+	err := json.Unmarshal(msg, &canalMsg)
+	if err != nil {
+		return err
+	}
+	now := time.Now()
+
+	//2、获取masterId
+	masterId := utils.StrToInt(strings.Split(canalMsg.Database, "_")[1])
+
+	//TODO::日志记录
+	utils.FilePutContents("handleCanalO2oOrderForNumericalStatementTable_"+utils.IntToStr(masterId)+"_"+now.Format("2006-01-02"), string(msg))
+
+	//2、判断操作(insert | update)
+	if canalMsg.Type == md.CanalMsgInsertSqlType || canalMsg.Type == md.CanalMsgUpdateSqlType {
+		var isUpdate bool
+		//3、查找是否有数据
+		var ordDate string
+		for _, item := range canalMsg.Data {
+			ordDate = utils.TimeParseStd(item.CreateTime).Format("2006-01-02")
+		}
+
+		statistics, err := db2.GetMasterO2oOrderStatistics(db.Db, masterId, ordDate)
+		if err != nil {
+			return err
+		}
+		if statistics == nil && canalMsg.Type == md.CanalMsgInsertSqlType {
+			statistics = &model.MasterO2oOrderStatistics{
+				MasterId:              masterId,
+				PaymentTotal:          "",
+				OrderCount:            0,
+				EstimatedCommission:   "",
+				LoseOrderCount:        0,
+				PlaceOrderNumOfPeople: 0,
+				EffectiveOrderCount:   0,
+				EffectiveCommission:   "",
+				ReceiveCommission:     "",
+				LoseCommission:        "",
+				AvgCommission:         "",
+				CustomerUnitPrice:     "",
+				Date:                  ordDate,
+				CreateAt:              now.Format("2006-01-02 15:04:05"),
+				UpdateAt:              now.Format("2006-01-02 15:04:05"),
+			}
+			_, err = db2.MasterO2oOrderStatisticsInsert(db.Db, statistics)
+			if err != nil {
+				return err
+			}
+		}
+		if statistics == nil {
+			return errors.New("过期订单数据不予处理")
+		}
+
+		paymentTotal := statistics.PaymentTotal
+		orderCount := statistics.OrderCount
+		estimatedCommission := statistics.EstimatedCommission
+		loseOrderCount := statistics.LoseOrderCount
+		placeOrderNumOfPeople := statistics.PlaceOrderNumOfPeople
+		effectiveOrderCount := statistics.EffectiveOrderCount
+		effectiveCommission := statistics.EffectiveCommission
+		loseCommission := statistics.LoseCommission
+		avgCommission := statistics.AvgCommission
+		customerUnitPrice := statistics.CustomerUnitPrice
+		effectivePaymentTotal := statistics.EffectivePaymentTotal
+
+		if canalMsg.Type == md.CanalMsgInsertSqlType {
+			for _, item := range canalMsg.Data {
+				orderCount++
+				if item.State == "0" {
+					//未支付不统计
+					return errors.New("未支付不统计")
+				} else if item.State == "6" {
+					//失效
+					loseOrderCount++
+				} else {
+					effectiveOrderCount++
+				}
+
+				paymentTotal = utils.Float64ToStr(utils.StrToFloat64(paymentTotal) + utils.StrToFloat64(item.CostPrice))                        //付款金额
+				estimatedCommission = utils.Float64ToStr(utils.StrToFloat64(estimatedCommission) + utils.StrToFloat64(item.EstimateCommission)) //预估佣金(元)
+				effectiveCommission = utils.Float64ToStr(utils.StrToFloat64(effectiveCommission) + utils.StrToFloat64(item.EstimateCommission)) //有效佣金(元)
+				effectivePaymentTotal = utils.Float64ToStr(utils.StrToFloat64(effectivePaymentTotal) + utils.StrToFloat64(item.CostPrice))      //有效付款金额(元)
+
+				estimatedCommissionValue, _ := decimal.NewFromString(estimatedCommission)
+				orderCountValue := decimal.NewFromInt(int64(orderCount))
+				avgCommission = estimatedCommissionValue.Div(orderCountValue).String() //平均佣金
+
+				if item.State != "6" && item.State != "5" {
+					cacheKey := fmt.Sprintf(ZhiOsO2oPlaceOrderNumOfPeopleHashMapCacheKey, utils.IntToStr(masterId), ordDate)
+					get, _ := cache.HGetString(cacheKey, item.Uid)
+					if get == "" {
+						placeOrderNumOfPeople++ //下单人数
+						cache.HSet(cacheKey, item.Uid, "1")
+					} else {
+						cache.HSet(cacheKey, item.Uid, utils.IntToStr(utils.StrToInt(get)+1))
+					}
+					cache.Expire(cacheKey, md.ZhiOsUserVisitIpAddressHashMapCacheTime)
+					paymentTotalValue, _ := decimal.NewFromString(paymentTotal)
+					if placeOrderNumOfPeople == 0 {
+						return errors.New("divider cannot be 0 in division operation")
+					}
+					placeOrderNumOfPeopleValue := decimal.NewFromInt(int64(placeOrderNumOfPeople))
+					customerUnitPrice = paymentTotalValue.Div(placeOrderNumOfPeopleValue).String()
+				}
+				//客单价
+				isUpdate = true
+			}
+		}
+
+		if canalMsg.Type == md.CanalMsgUpdateSqlType {
+			judgeSate := JudgeO2oOrdSate(*canalMsg)
+			if judgeSate > 0 {
+				if judgeSate == 2 {
+					//TODO::未收货失效
+					loseOrderCount++
+					effectiveOrderCount--
+					for _, item := range canalMsg.Data {
+						loseCommission = utils.Float64ToStr(utils.StrToFloat64(loseCommission) + utils.StrToFloat64(item.EstimateCommission))
+						effectiveCommission = utils.Float64ToStr(utils.StrToFloat64(effectiveCommission) - utils.StrToFloat64(item.EstimateCommission))
+						effectivePaymentTotal = utils.Float64ToStr(utils.StrToFloat64(effectivePaymentTotal) - utils.StrToFloat64(item.CostPrice))
+						isUpdate = true
+					}
+				}
+				if judgeSate == 3 {
+					//TODO::已收货失效
+					loseOrderCount++
+					effectiveOrderCount--
+					for _, item := range canalMsg.Data {
+						loseCommission = utils.Float64ToStr(utils.StrToFloat64(loseCommission) + utils.StrToFloat64(item.EstimateCommission))
+						effectiveCommission = utils.Float64ToStr(utils.StrToFloat64(effectiveCommission) - utils.StrToFloat64(item.EstimateCommission))
+						effectivePaymentTotal = utils.Float64ToStr(utils.StrToFloat64(effectivePaymentTotal) - utils.StrToFloat64(item.CostPrice))
+						isUpdate = true
+					}
+				}
+
+				if judgeSate == 1 || judgeSate == 3 {
+					//TODO::收货额外处理
+
+					//查找是否有数据
+					var ordConfirmAt string
+					for _, item := range canalMsg.Data {
+						ordConfirmAt = utils.TimeParseStd(item.FinishTime).Format("2006-01-02")
+					}
+					ordConfirmStatistics, err := db2.GetMasterO2oOrderStatistics(db.Db, masterId, ordConfirmAt)
+					if err != nil {
+						return err
+					}
+					if ordConfirmStatistics == nil {
+						return errors.New("过期收货订单数据不予处理")
+					}
+
+					if judgeSate == 1 {
+						//TODO::收货
+						for _, item := range canalMsg.Data {
+							ordConfirmStatistics.ReceiveCommission = utils.Float64ToStr(utils.StrToFloat64(ordConfirmStatistics.ReceiveCommission) + utils.StrToFloat64(item.EstimateCommission))
+						}
+					}
+					if judgeSate == 3 {
+						//TODO::已收货失效
+						for _, item := range canalMsg.Data {
+							ordConfirmStatistics.ReceiveCommission = utils.Float64ToStr(utils.StrToFloat64(ordConfirmStatistics.ReceiveCommission) - utils.StrToFloat64(item.EstimateCommission))
+						}
+					}
+					ordConfirmStatistics.UpdateAt = now.Format("2006-01-02 15:04:05")
+					_, err = db2.MasterO2oOrderStatisticsUpdate(db.Db, ordConfirmStatistics.Id, ordConfirmStatistics,
+						"receive_commission", "update_at")
+				}
+			}
+		}
+
+		if isUpdate {
+			statistics.PaymentTotal = paymentTotal
+			statistics.OrderCount = orderCount
+			statistics.EstimatedCommission = estimatedCommission
+			statistics.LoseOrderCount = loseOrderCount
+			statistics.PlaceOrderNumOfPeople = placeOrderNumOfPeople
+			statistics.EffectiveOrderCount = effectiveOrderCount
+			statistics.EffectiveCommission = effectiveCommission
+			statistics.LoseCommission = loseCommission
+			statistics.AvgCommission = avgCommission
+			statistics.CustomerUnitPrice = customerUnitPrice
+			statistics.EffectivePaymentTotal = effectivePaymentTotal
+			statistics.UpdateAt = now.Format("2006-01-02 15:04:05")
+			_, err = db2.MasterO2oOrderStatisticsUpdate(db.Db, statistics.Id, statistics,
+				"payment_total", "order_count", "estimated_commission", "lose_order_count",
+				"place_order_num_of_people", "effective_order_count", "effective_commission", "lose_commission",
+				"avg_commission", "customer_unit_price", "effective_payment_total", "update_at")
+			if err != nil {
+				return err
+			}
+		}
+
+	}
+	return nil
+}
+
+// JudgeO2oOrdSate 处理订单状态(judgeSate[0:不需要处理 1:收货 2:未收货失效 3:已收货失效])
+func JudgeO2oOrdSate(message md.CanalO2oOrderMessage[md.CanalO2oOrder]) (judgeSate int) {
+	oldData := message.Old
+
+	//1、获取 旧的订单状态
+	var oldOrdState string
+	for _, item := range oldData {
+		if item.State != "" {
+			oldOrdState = item.State
+		}
+	}
+	if oldOrdState == "" {
+		return
+	}
+
+	//2、获取 新的订单状态
+	var nowOrdState string
+	for _, item := range message.Data {
+		if item.State != "" {
+			nowOrdState = item.State
+		}
+	}
+	if nowOrdState == "" {
+		return
+	}
+
+	if oldOrdState == nowOrdState {
+		return
+	}
+
+	//3、进行状态比较判断
+	if oldOrdState == "3" || oldOrdState == "4" {
+		if nowOrdState == "5" || nowOrdState == "6" {
+			//已收货失效
+			return 3
+		}
+	} else {
+		if nowOrdState == "5" || nowOrdState == "6" {
+			//未收货失效
+			return 2
+		}
+		if nowOrdState == "3" || nowOrdState == "4" {
+			//收货
+			return 1
+		}
+	}
+	return
+}

consume/canal_o2o_pay_order_for_numerical_statement_consume.go

@@ -0,0 +1,173 @@
+package consume
+
+import (
+	"applet/app/db"
+	"applet/app/utils"
+	"applet/app/utils/cache"
+	"applet/app/utils/logx"
+	"applet/consume/md"
+	"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
+	db2 "code.fnuoos.com/go_rely_warehouse/zyos_model.git/src/models/official"
+	"code.fnuoos.com/go_rely_warehouse/zyos_model.git/src/models/official/model"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"github.com/shopspring/decimal"
+	"github.com/streadway/amqp"
+	"strings"
+	"time"
+)
+
+const ZhiOsO2oPayPlaceOrderNumOfPeopleHashMapCacheKey = "%s:zhiOs_o2o_pay_place_order_num_of_people_hash_map_cache:%s" //下单人数缓存hashMap键
+
+func CanalO2oPayOrderForNumericalStatementConsume(queue md.MqQueue) {
+	ch, err := rabbit.Cfg.Pool.GetChannel()
+	if err != nil {
+		logx.Error(err)
+		return
+	}
+	defer ch.Release()
+	//1、将自己绑定到交换机上
+	ch.Bind(queue.Name, queue.ExchangeName, queue.RoutKey)
+	//2、取出数据进行消费
+	ch.Qos(100)
+	delivery := ch.Consume(queue.Name, false)
+
+	var res amqp.Delivery
+	var ok bool
+	for {
+		res, ok = <-delivery
+		if ok == true {
+			//fmt.Println(string(res.Body))
+			fmt.Println(">>>>>>>>>>>>>>>>CanalO2oPayOrderForNumericalStatementConsume<<<<<<<<<<<<<<<<<<<<<<<<<")
+			err = handleCanalO2oPayOrderForNumericalStatementTable(res.Body)
+			if err != nil {
+				fmt.Println("err ::: ", err)
+				utils.FilePutContents("CanalO2oPayOrderForNumericalStatementConsume_ERR", "[err]:"+err.Error())
+				_ = res.Reject(false)
+				////TODO::重新推回队列末尾，避免造成队列堵塞
+				//var msg *md.OneCirclesStructForSignIn
+				//json.Unmarshal(res.Body, &msg)
+				//ch.Publish(queue.ExchangeName, msg, queue.RoutKey)
+			} else {
+				_ = res.Ack(true)
+			}
+		} else {
+			panic(errors.New("error getting message"))
+		}
+	}
+	fmt.Println("get msg done")
+}
+
+func handleCanalO2oPayOrderForNumericalStatementTable(msg []byte) error {
+	//1、解析canal采集至mq中queue的数据结构体
+	var canalMsg *md.CanalMallOrder
+	err := json.Unmarshal(msg, &canalMsg)
+	if err != nil {
+		return err
+	}
+	now := time.Now()
+
+	//2、获取masterId
+	masterId := utils.StrToInt(strings.Split(canalMsg.Database, "_")[1])
+
+	//TODO::日志记录
+	utils.FilePutContents("handleCanalO2oPayOrderForNumericalStatementTable_"+utils.IntToStr(masterId)+"_"+now.Format("2006-01-02"), string(msg))
+
+	//2、判断操作(insert | update)
+	if canalMsg.Type == md.CanalMsgInsertSqlType || canalMsg.Type == md.CanalMsgUpdateSqlType {
+		var isUpdate bool
+		//3、查找是否有数据
+		var ordDate string
+		for _, item := range canalMsg.Data {
+			ordDate = utils.TimeParseStd(item.CreateTime).Format("2006-01-02")
+		}
+
+		statistics, err := db2.GetMasterO2oPayOrderStatistics(db.Db, masterId, ordDate)
+		if err != nil {
+			return err
+		}
+		if statistics == nil && canalMsg.Type == md.CanalMsgInsertSqlType {
+			statistics = &model.MasterO2oPayOrderStatistics{
+				MasterId:              masterId,
+				PaymentTotal:          "",
+				OrderCount:            0,
+				EstimatedCommission:   "",
+				PlaceOrderNumOfPeople: 0,
+				AvgCommission:         "",
+				CustomerUnitPrice:     "",
+				Date:                  ordDate,
+				CreateAt:              now.Format("2006-01-02 15:04:05"),
+				UpdateAt:              now.Format("2006-01-02 15:04:05"),
+			}
+			_, err = db2.MasterO2oPayOrderStatisticsInsert(db.Db, statistics)
+			if err != nil {
+				return err
+			}
+		}
+		if statistics == nil {
+			return errors.New("过期订单数据不予处理")
+		}
+
+		paymentTotal := statistics.PaymentTotal
+		orderCount := statistics.OrderCount
+		estimatedCommission := statistics.EstimatedCommission
+		placeOrderNumOfPeople := statistics.PlaceOrderNumOfPeople
+		avgCommission := statistics.AvgCommission
+		customerUnitPrice := statistics.CustomerUnitPrice
+
+		if canalMsg.Type == md.CanalMsgInsertSqlType {
+			for _, item := range canalMsg.Data {
+				orderCount++
+				if item.State == "0" {
+					//未支付不统计
+					return errors.New("未支付不统计")
+				}
+				if item.State == "1" || item.State == "4" {
+					paymentTotal = utils.Float64ToStr(utils.StrToFloat64(paymentTotal) + utils.StrToFloat64(item.CostPrice))                        //付款金额
+					estimatedCommission = utils.Float64ToStr(utils.StrToFloat64(estimatedCommission) + utils.StrToFloat64(item.EstimateCommission)) //预估佣金(元)
+
+					estimatedCommissionValue, _ := decimal.NewFromString(estimatedCommission)
+					orderCountValue := decimal.NewFromInt(int64(orderCount))
+					avgCommission = estimatedCommissionValue.Div(orderCountValue).String() //平均佣金
+
+					cacheKey := fmt.Sprintf(ZhiOsO2oPayPlaceOrderNumOfPeopleHashMapCacheKey, utils.IntToStr(masterId), ordDate)
+					get, _ := cache.HGetString(cacheKey, item.Uid)
+					if get == "" {
+						placeOrderNumOfPeople++ //下单人数
+						cache.HSet(cacheKey, item.Uid, "1")
+					} else {
+						cache.HSet(cacheKey, item.Uid, utils.IntToStr(utils.StrToInt(get)+1))
+					}
+					cache.Expire(cacheKey, md.ZhiOsUserVisitIpAddressHashMapCacheTime)
+					paymentTotalValue, _ := decimal.NewFromString(paymentTotal)
+					if placeOrderNumOfPeople == 0 {
+						return errors.New("divider cannot be 0 in division operation")
+					}
+					placeOrderNumOfPeopleValue := decimal.NewFromInt(int64(placeOrderNumOfPeople))
+					customerUnitPrice = paymentTotalValue.Div(placeOrderNumOfPeopleValue).String() //客单价
+					isUpdate = true
+				}
+			}
+		}
+
+		if isUpdate {
+			statistics.PaymentTotal = paymentTotal
+			statistics.OrderCount = orderCount
+			statistics.EstimatedCommission = estimatedCommission
+			statistics.PlaceOrderNumOfPeople = placeOrderNumOfPeople
+			statistics.AvgCommission = avgCommission
+			statistics.CustomerUnitPrice = customerUnitPrice
+			statistics.UpdateAt = now.Format("2006-01-02 15:04:05")
+			_, err = db2.MasterO2oPayOrderStatisticsUpdate(db.Db, statistics.Id, statistics,
+				"payment_total", "order_count", "estimated_commission",
+				"place_order_num_of_people",
+				"avg_commission", "customer_unit_price", "update_at")
+			if err != nil {
+				return err
+			}
+		}
+
+	}
+	return nil
+}

consume/init.go

@@ -80,15 +80,19 @@ func initConsumes() {
 	//jobs[consumeMd.ZhiOsUserVisitIpAddressConsumeFunName] = ZhiOsUserVisitIpAddressConsume
 	//jobs[consumeMd.CanalUserVirtualCcoinFlowFunName] = CanalUserVirtualCoinFlowConsume
 	//jobs[consumeMd.CanalGuideOrderForNumericalStatementConsumeFunName] = CanalGuideOrderForNumericalStatementConsume
+	jobs[consumeMd.CanalMallOrderForNumericalStatementConsumeFunName] = CanalMallOrderForNumericalStatementConsume
+	//jobs[consumeMd.CanalO2oOrderForNumericalStatementConsumeFunName] = CanalO2oOrderForNumericalStatementConsume
+	//jobs[consumeMd.CanalO2oPayOrderForNumericalStatementConsumeFunName] = CanalO2oPayOrderForNumericalStatementConsume
+	//jobs[consumeMd.CanalB2cOrderForNumericalStatementConsumeFunName] = CanalB2cOrderForNumericalStatementConsume
 
 	//////////////////////////////////////// oneCircles /////////////////////////////////////////////////////
 
-	jobs[consumeMd.OneCirclesSignInGreenEnergyFunName] = OneCirclesSignInGreenEnergyConsume
-	jobs[consumeMd.OneCirclesStartLevelDividendFunName] = OneCirclesStartLevelDividendConsume
-	jobs[consumeMd.OneCirclesActivityCoinAutoExchangeGreenEnergyFunName] = OneCirclesActivityCoinAutoExchangeGreenEnergyConsume
-	jobs[consumeMd.OneCirclesActivityCoinAutoExchangeGreenEnergyForTeamFunName] = OneCirclesActivityCoinAutoExchangeGreenEnergyForTeamConsume
-	jobs[consumeMd.OneCirclesSettlementPublicGiveActivityCoinFunName] = OneCirclesSettlementPublicGiveActivityCoinConsume
-	jobs[consumeMd.OneCirclesAddPublicPlatoonUserRelationCommissionFunName] = OneCirclesAddPublicPlatoonUserRelationCommissionConsume
+	//jobs[consumeMd.OneCirclesSignInGreenEnergyFunName] = OneCirclesSignInGreenEnergyConsume
+	//jobs[consumeMd.OneCirclesStartLevelDividendFunName] = OneCirclesStartLevelDividendConsume
+	//jobs[consumeMd.OneCirclesActivityCoinAutoExchangeGreenEnergyFunName] = OneCirclesActivityCoinAutoExchangeGreenEnergyConsume
+	//jobs[consumeMd.OneCirclesActivityCoinAutoExchangeGreenEnergyForTeamFunName] = OneCirclesActivityCoinAutoExchangeGreenEnergyForTeamConsume
+	//jobs[consumeMd.OneCirclesSettlementPublicGiveActivityCoinFunName] = OneCirclesSettlementPublicGiveActivityCoinConsume
+	//jobs[consumeMd.OneCirclesAddPublicPlatoonUserRelationCommissionFunName] = OneCirclesAddPublicPlatoonUserRelationCommissionConsume
 
 	//jobs[consumeMd.OneCirclesSignInCopyGreenEnergyFunName] = OneCirclesSignInCopyGreenEnergyConsume
 

consume/md/consume_key.go

@@ -83,6 +83,42 @@ var RabbitMqQueueKeyList = []*MqQueue{
 		BindKey:        "",
 		ConsumeFunName: "CanalGuideOrderForNumericalStatementConsume",
 	},
+	{
+		ExchangeName:   "canal.topic",
+		Name:           "canal_mall_order_for_numerical_statement",
+		Type:           TopicQueueType,
+		IsPersistent:   false,
+		RoutKey:        "canal_mall_order",
+		BindKey:        "",
+		ConsumeFunName: "CanalMallOrderForNumericalStatementConsume",
+	},
+	{
+		ExchangeName:   "canal.topic",
+		Name:           "canal_b2c_order_for_numerical_statement",
+		Type:           TopicQueueType,
+		IsPersistent:   false,
+		RoutKey:        "canal_b2c_order",
+		BindKey:        "",
+		ConsumeFunName: "CanalB2cOrderForNumericalStatementConsume",
+	},
+	{
+		ExchangeName:   "canal.topic",
+		Name:           "canal_o2o_order_for_numerical_statement",
+		Type:           TopicQueueType,
+		IsPersistent:   false,
+		RoutKey:        "canal_o2o_order",
+		BindKey:        "",
+		ConsumeFunName: "CanalO2oOrderForNumericalStatementConsume",
+	},
+	{
+		ExchangeName:   "canal.topic",
+		Name:           "canal_o2o_pay_order_for_numerical_statement",
+		Type:           TopicQueueType,
+		IsPersistent:   false,
+		RoutKey:        "canal_o2o_pay_to_merchant",
+		BindKey:        "",
+		ConsumeFunName: "CanalO2oPayOrderForNumericalStatementConsume",
+	},
 	{
 		ExchangeName:   "canal.topic",
 		Name:           "canal_gim_message",
@@ -587,6 +623,10 @@ const (
 	ZhiosSupplierAfterOrderFunName                              = "ZhiosSupplierAfterOrder"
 	CanalOrderConsumeFunName                                    = "CanalOrderConsume"
 	CanalGuideOrderForNumericalStatementConsumeFunName          = "CanalGuideOrderForNumericalStatementConsume"
+	CanalMallOrderForNumericalStatementConsumeFunName           = "CanalMallOrderForNumericalStatementConsume"
+	CanalO2oOrderForNumericalStatementConsumeFunName            = "CanalO2oOrderForNumericalStatementConsume"
+	CanalO2oPayOrderForNumericalStatementConsumeFunName         = "CanalO2oPayOrderForNumericalStatementConsume"
+	CanalB2cOrderForNumericalStatementConsumeFunName            = "CanalB2cOrderForNumericalStatementConsume"
 	CanalGuideOrderConsumeFunName                               = "CanalGuideOrderConsume"
 	ZhiOsUserVisitIpAddressConsumeFunName                       = "ZhiOsUserVisitIpAddressConsume"
 	DouShenUserRegisterConsumeForOfficialFunName                = "DouShenUserRegisterConsumeForOfficial"

consume/md/md_canal_b2c_order_consume.go

@@ -0,0 +1,74 @@
+package md
+
+type CanalB2cOrder struct {
+	OrdId                 string  `json:"ord_id" xorm:"not null pk BIGINT(20)"`
+	MainOrdId             string  `json:"main_ord_id" xorm:"not null comment('主订单号') index BIGINT(20)"`
+	Uid                   string  `json:"uid" xorm:"comment('用户id') index INT(11)"`
+	BelongStoreId         string  `json:"belong_store_id" xorm:"comment('归属店铺id') INT(11)"`
+	BuyerName             string  `json:"buyer_name" xorm:"comment('购买人') VARCHAR(255)"`
+	BuyerPhone            string  `json:"buyer_phone" xorm:"comment('购买人手机号') VARCHAR(255)"`
+	CostPrice             string  `json:"cost_price" xorm:"comment('价格') DECIMAL(12,2)"`
+	CostVirtualCoin       string  `json:"cost_virtual_coin" xorm:"comment('消耗的虚拟币') DECIMAL(12,2)"`
+	VirtualCoinId         string  `json:"virtual_coin_id" xorm:"comment('使用的虚拟币id') INT(11)"`
+	State                 string  `json:"state" xorm:"comment('订单状态：0未支付 1已支付 ...(其余状态根据订单类型不同)') TINYINT(1)"`
+	PayTime               string  `json:"pay_time" xorm:"comment('支付时间') DATETIME"`
+	PickUp                string  `json:"pick_up" xorm:"not null comment('取货方式：1堂食 2打包带走') TINYINT(1)"`
+	PayChannel            string  `json:"pay_channel" xorm:"not null comment('支付方式：1balance 2alipay 3wx_pay 4zhios_pay_alipay') TINYINT(1)"`
+	ShippingTime          string  `json:"shipping_time" xorm:"comment('发货时间') DATETIME"`
+	DeliveryWay           string  `json:"delivery_way" xorm:"default 1 comment('发货方式（1:自己联系）') TINYINT(1)"`
+	LogisticCompany       string  `json:"logistic_company" xorm:"not null default '' comment('物流公司') VARCHAR(255)"`
+	LogisticNum           string  `json:"logistic_num" xorm:"not null default '' comment('物流单号') VARCHAR(255)"`
+	ReceiverPhone         string  `json:"receiver_phone" xorm:"not null default '' comment('收货人手机号') VARCHAR(20)"`
+	ReceiverName          string  `json:"receiver_name" xorm:"not null default '' comment('收货人名字') VARCHAR(255)"`
+	ReceiverAddressDetail string  `json:"receiver_address_detail" xorm:"not null default '' comment('收货人地址') VARCHAR(255)"`
+	ShippingType          string  `json:"shipping_type" xorm:"not null default 1 comment('运送方式：1快递送货') TINYINT(1)"`
+	CouponDiscount        string  `json:"coupon_discount" xorm:"not null default 0.00 comment('优惠券折扣额') DECIMAL(12,2)"`
+	DiscountPrice         string  `json:"discount_price" xorm:"not null default 0.00 comment('立减') DECIMAL(12,2)"`
+	UserCouponId          string  `json:"user_coupon_id" xorm:"comment('使用的优惠券id') BIGINT(20)"`
+	ReturnInsuranceFee    string  `json:"return_insurance_fee" xorm:"not null default 0.00 comment('退货无忧费用') DECIMAL(12,2)"`
+	IsReceipt             string  `json:"is_receipt" xorm:"not null default 0 comment('是否开具发票 0否 1是') TINYINT(255)"`
+	ShippingFee           string  `json:"shipping_fee" xorm:"not null default 0.00 comment('运费') DECIMAL(12,2)"`
+	Comment               string  `json:"comment" xorm:"not null comment('备注') VARCHAR(2048)"`
+	ProvinceName          string  `json:"province_name" xorm:"not null default '' comment('收货省份') VARCHAR(255)"`
+	CityName              string  `json:"city_name" xorm:"not null default '' comment('收货城市') VARCHAR(255)"`
+	CountyName            string  `json:"county_name" xorm:"not null default '' comment('收货区域') VARCHAR(255)"`
+	PayNum                string  `json:"pay_num" xorm:"not null default '' comment('交易流水') VARCHAR(255)"`
+	ConfirmTime           string  `json:"confirm_time" xorm:"comment('确认时间') DATETIME"`
+	EstimateCommission    string  `json:"estimate_commission" xorm:"not null default 0.0000 comment('预计佣金（三方分账完之后站长的佣金）') DECIMAL(12,4)"`
+	CreateTime            string  `json:"create_time" xorm:"not null default 'CURRENT_TIMESTAMP' comment('创建时间') DATETIME"`
+	UpdateTime            string  `json:"update_time" xorm:"not null default 'CURRENT_TIMESTAMP' comment('更新时间') DATETIME"`
+	DeletedTime           string  `json:"deleted_time" xorm:"comment('删除时间') DATETIME"`
+	FinishTime            string  `json:"finish_time" xorm:"comment('完成时间') DATETIME"`
+	OrderType             string  `json:"order_type" xorm:"not null default 1 comment('订单类型：1(小店订单)') TINYINT(3)"`
+	Data                  string  `json:"data" xorm:"not null comment('订单相关的数据') TEXT"`
+	SettleTime            string  `json:"settle_time" xorm:"comment('结算时间') DATETIME"`
+	CommissionTime        string  `json:"commission_time" xorm:"comment('分佣时间') DATETIME"`
+	ShareUid              string  `json:"share_uid" xorm:"comment('分享人') INT(11)"`
+	TotalPrice            float32 `json:"total_price" xorm:"comment('订单总金额') FLOAT(8,2)"`
+	PickUpNum             string  `json:"pick_up_num" xorm:"comment('取餐号') VARCHAR(255)"`
+	TradeNo               string  `json:"trade_no" xorm:"comment('支付平台的订单号') VARCHAR(50)"`
+	PayTradeNo            string  `json:"pay_trade_no" xorm:"comment('支付联盟支付的订单号') VARCHAR(50)"`
+	ConsumptionType       string  `json:"consumption_type" xorm:"default 1 comment('消费类型：1到店消费2立即制作') TINYINT(1)"`
+	TableNum              string  `json:"table_num" xorm:"default '' comment('桌号') VARCHAR(12)"`
+	IsThreePartySplit     string  `json:"is_three_party_split" xorm:"comment('是否进行了三方分账,0：否，1:是') TINYINT(1)"`
+	MainCommission        string  `json:"main_commission" xorm:"not null default 0.0000 comment('总佣金') DECIMAL(12,4)"`
+	SkuPriceInfo          string  `json:"sku_price_info" xorm:"comment('新版本支付规则（规则计算信息）') VARCHAR(500)"`
+	PlatformCommission    string  `json:"platform_commission" xorm:"comment('平台所得的佣金(平台不是站长)') DECIMAL(12,4)"`
+	MealFee               string  `json:"meal_fee" xorm:"comment('餐位费') VARCHAR(50)"`
+	SupplierMerchantId    string  `json:"supplier_merchant_id" xorm:"comment('供应商id') INT(11)"`
+	SupplierOrdId         string  `json:"supplier_ord_id" xorm:"comment('供应商订单id') VARCHAR(255)"`
+	IsSetOutAch           string  `xorm:"not null default 0 INT(1)" json:"is_set_out_ach"`
+}
+
+type CanalB2cOrderMessage[T any] struct {
+	Data     []T      `json:"data"`
+	Database string   `json:"database"`
+	ES       int64    `json:"es"`
+	ID       int64    `json:"id"`
+	IsDdl    bool     `json:"isDdl"`
+	Old      []T      `json:"old"`
+	PkNames  []string `json:"pkNames"`
+	Table    string   `json:"table"`
+	TS       int64    `json:"ts"`
+	Type     string   `json:"type"`
+}

consume/md/md_canal_o2o_order_consume.go

@@ -0,0 +1,73 @@
+package md
+
+type CanalO2oOrder struct {
+	OrdId                 string `json:"ord_id" xorm:"not null pk BIGstring(20)"`
+	MainOrdId             string `json:"main_ord_id" xorm:"not null comment('主订单号') index BIGstring(20)"`
+	Uid                   string `json:"uid" xorm:"comment('用户id') index string(11)"`
+	BelongStoreId         string `json:"belong_store_id" xorm:"comment('归属店铺id') string(11)"`
+	BuyerName             string `json:"buyer_name" xorm:"comment('购买人') VARCHAR(255)"`
+	BuyerPhone            string `json:"buyer_phone" xorm:"comment('购买人手机号') VARCHAR(255)"`
+	CostPrice             string `json:"cost_price" xorm:"comment('价格') DECIMAL(12,2)"`
+	CostVirtualCoin       string `json:"cost_virtual_coin" xorm:"comment('消耗的虚拟币') DECIMAL(12,2)"`
+	VirtualCoinId         string `json:"virtual_coin_id" xorm:"comment('使用的虚拟币id') string(11)"`
+	State                 string `json:"state" xorm:"comment('订单状态：0未支付 1已支付 ...(其余状态根据订单类型不同)') TINYstring(1)"`
+	PayTime               string `json:"pay_time" xorm:"comment('支付时间') DATETIME"`
+	PickUp                string `json:"pick_up" xorm:"not null comment('取货方式：1堂食 2打包带走') TINYstring(1)"`
+	PayChannel            string `json:"pay_channel" xorm:"not null comment('支付方式：1balance 2alipay 3wx_pay 4zhios_pay_alipay') TINYstring(1)"`
+	ShippingTime          string `json:"shipping_time" xorm:"comment('发货时间') DATETIME"`
+	LogisticCompany       string `json:"logistic_company" xorm:"not null default '' comment('物流公司') VARCHAR(255)"`
+	LogisticNum           string `json:"logistic_num" xorm:"not null default '' comment('物流单号') VARCHAR(255)"`
+	ReceiverPhone         string `json:"receiver_phone" xorm:"not null default '' comment('收货人手机号') VARCHAR(20)"`
+	ReceiverName          string `json:"receiver_name" xorm:"not null default '' comment('收货人名字') VARCHAR(255)"`
+	ReceiverAddressDetail string `json:"receiver_address_detail" xorm:"not null default '' comment('收货人地址') VARCHAR(255)"`
+	ShippingType          string `json:"shipping_type" xorm:"not null default 1 comment('运送方式：1快递送货') TINYstring(1)"`
+	CouponDiscount        string `json:"coupon_discount" xorm:"not null default 0.00 comment('优惠券折扣额') DECIMAL(12,2)"`
+	DiscountPrice         string `json:"discount_price" xorm:"not null default 0.00 comment('立减') DECIMAL(12,2)"`
+	UserCouponId          string `json:"user_coupon_id" xorm:"comment('使用的优惠券id') BIGstring(20)"`
+	ReturnInsuranceFee    string `json:"return_insurance_fee" xorm:"not null default 0.00 comment('退货无忧费用') DECIMAL(12,2)"`
+	IsReceipt             string `json:"is_receipt" xorm:"not null default 0 comment('是否开具发票 0否 1是') TINYstring(255)"`
+	ShippingFee           string `json:"shipping_fee" xorm:"not null default 0.00 comment('运费') DECIMAL(12,2)"`
+	Comment               string `json:"comment" xorm:"not null comment('备注') VARCHAR(2048)"`
+	ProvinceName          string `json:"province_name" xorm:"not null default '' comment('收货省份') VARCHAR(255)"`
+	CityName              string `json:"city_name" xorm:"not null default '' comment('收货城市') VARCHAR(255)"`
+	CountyName            string `json:"county_name" xorm:"not null default '' comment('收货区域') VARCHAR(255)"`
+	PayNum                string `json:"pay_num" xorm:"not null default '' comment('交易流水') VARCHAR(255)"`
+	ConfirmTime           string `json:"confirm_time" xorm:"comment('确认时间') DATETIME"`
+	EstimateCommission    string `json:"estimate_commission" xorm:"not null default 0.0000 comment('预计佣金（三方分账完之后站长的佣金）') DECIMAL(12,4)"`
+	CreateTime            string `json:"create_time" xorm:"not null default 'CURRENT_TIMESTAMP' comment('创建时间') DATETIME"`
+	UpdateTime            string `json:"update_time" xorm:"not null default 'CURRENT_TIMESTAMP' comment('更新时间') DATETIME"`
+	DeletedTime           string `json:"deleted_time" xorm:"comment('删除时间') DATETIME"`
+	FinishTime            string `json:"finish_time" xorm:"comment('完成时间') DATETIME"`
+	OrderType             string `json:"order_type" xorm:"not null default 1 comment('订单类型：1(小店订单)') TINYstring(3)"`
+	Data                  string `json:"data" xorm:"not null comment('订单相关的数据') TEXT"`
+	SettleTime            string `json:"settle_time" xorm:"comment('结算时间') DATETIME"`
+	CommissionTime        string `json:"commission_time" xorm:"comment('分佣时间') DATETIME"`
+	ShareUid              string `json:"share_uid" xorm:"comment('分享人') string(11)"`
+	TotalPrice            string `json:"total_price" xorm:"comment('订单总金额') FLOAT(8,2)"`
+	PickUpNum             string `json:"pick_up_num" xorm:"comment('取餐号') VARCHAR(255)"`
+	TradeNo               string `json:"trade_no" xorm:"comment('支付平台的订单号') VARCHAR(50)"`
+	PayTradeNo            string `json:"pay_trade_no" xorm:"comment('支付联盟支付的订单号') VARCHAR(50)"`
+	ConsumptionType       string `json:"consumption_type" xorm:"not null default 1 comment('消费类型：1到店消费2立即制作') TINYstring(1)"`
+	TableNum              string `json:"table_num" xorm:"default '' comment('桌号') VARCHAR(12)"`
+	IsThreePartySplit     string `json:"is_three_party_split" xorm:"comment('是否进行了三方分账,0：否，1:是') TINYstring(1)"`
+	MainCommission        string `json:"main_commission" xorm:"not null default 0.0000 comment('总佣金') DECIMAL(12,4)"`
+	SkuPriceInfo          string `json:"sku_price_info" xorm:"comment('新版本支付规则（规则计算信息）') VARCHAR(500)"`
+	PlatformCommission    string `json:"platform_commission" xorm:"comment('平台所得的佣金(平台不是站长)') DECIMAL(12,4)"`
+	MealFee               string `json:"meal_fee" xorm:"comment('餐位费') VARCHAR(50)"`
+	ReturnMoneySettleAt   string `json:"return_money_settle_at" xorm:"default 0 comment('小口袋定制返现时间') string(11)"`
+	IsSetReduce           string `json:"is_set_reduce" xorm:"default 0 comment('小口袋定制设置退回 0否 1是') string(1)"`
+	IsSetOutAch           string `xorm:"not null default 0 string(1)" json:"is_set_out_ach"`
+}
+
+type CanalO2oOrderMessage[T any] struct {
+	Data     []T      `json:"data"`
+	Database string   `json:"database"`
+	ES       string   `json:"es"`
+	ID       string   `json:"id"`
+	IsDdl    bool     `json:"isDdl"`
+	Old      []T      `json:"old"`
+	PkNames  []string `json:"pkNames"`
+	Table    string   `json:"table"`
+	TS       string   `json:"ts"`
+	Type     string   `json:"type"`
+}

go.mod

@@ -12,7 +12,7 @@ require (
 	code.fnuoos.com/go_rely_warehouse/zyos_go_order_relate_rule.git v1.9.10-0.20240719082936-c249de79edce
 	code.fnuoos.com/go_rely_warehouse/zyos_go_pay.git v1.6.2-0.20231116085701-9ba6e19f877b
 	code.fnuoos.com/go_rely_warehouse/zyos_go_third_party_api.git v1.1.21-0.20240611024753-7cd929a03014
-	code.fnuoos.com/go_rely_warehouse/zyos_model.git v0.0.4-0.20240717064604-5e4000e89365
+	code.fnuoos.com/go_rely_warehouse/zyos_model.git v0.0.4-0.20240722120838-b9d594f8a46d
 	github.com/afex/hystrix-go v0.0.0-20180502004556-fa1af6a1f4f5
 	github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751
 	github.com/boombuler/barcode v1.0.1