update

consume/canal_mall_order_for_numerical_statement_consume.go

@@ -228,6 +228,51 @@ func handleCanalMallOrderForNumericalStatementTable(msg []byte) error {
 					_, err = db2.MasterMallOrderStatisticsUpdate(db.Db, ordConfirmStatistics.Id, ordConfirmStatistics,
 						"receive_commission", "update_at")
 				}
+
+				if judgeSate == 4 {
+					//TODO::已支付
+					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
+					}
+				}
 			}
 		}
 
@@ -257,7 +302,7 @@ func handleCanalMallOrderForNumericalStatementTable(msg []byte) error {
 	return nil
 }
 
-// JudgeMallOrdSate 处理订单状态(judgeSate[0:不需要处理 1:收货 2:未收货失效 3:已收货失效])
+// JudgeMallOrdSate 处理订单状态(judgeSate[0:不需要处理 1:收货 2:未收货失效 3:已收货失效 4:已支付])
 func JudgeMallOrdSate(message md.CanalMallOrder) (judgeSate int) {
 	oldData := message.Old
 
@@ -293,6 +338,9 @@ func JudgeMallOrdSate(message md.CanalMallOrder) (judgeSate int) {
 			//已收货失效
 			return 3
 		}
+	} else if oldOrdState == "0" {
+		//已支付
+		return 4
 	} else {
 		if nowOrdState == "6" {
 			//未收货失效

consume/canal_o2o_order_for_numerical_statement_consume.go

@@ -228,6 +228,51 @@ func handleCanalO2oOrderForNumericalStatementTable(msg []byte) error {
 					_, err = db2.MasterO2oOrderStatisticsUpdate(db.Db, ordConfirmStatistics.Id, ordConfirmStatistics,
 						"receive_commission", "update_at")
 				}
+
+				if judgeSate == 4 {
+					//TODO::已支付
+					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
+					}
+				}
 			}
 		}
 
@@ -257,7 +302,7 @@ func handleCanalO2oOrderForNumericalStatementTable(msg []byte) error {
 	return nil
 }
 
-// JudgeO2oOrdSate 处理订单状态(judgeSate[0:不需要处理 1:收货 2:未收货失效 3:已收货失效])
+// JudgeO2oOrdSate 处理订单状态(judgeSate[0:不需要处理 1:收货 2:未收货失效 3:已收货失效 4:已支付])
 func JudgeO2oOrdSate(message md.CanalO2oOrderMessage[md.CanalO2oOrder]) (judgeSate int) {
 	oldData := message.Old
 
@@ -293,6 +338,9 @@ func JudgeO2oOrdSate(message md.CanalO2oOrderMessage[md.CanalO2oOrder]) (judgeSa
 			//已收货失效
 			return 3
 		}
+	} else if oldOrdState == "0" {
+		//已支付
+		return 4
 	} else {
 		if nowOrdState == "5" || nowOrdState == "6" {
 			//未收货失效

consume/canal_o2o_pay_order_for_numerical_statement_consume.go

@@ -61,7 +61,7 @@ func CanalO2oPayOrderForNumericalStatementConsume(queue md.MqQueue) {
 
 func handleCanalO2oPayOrderForNumericalStatementTable(msg []byte) error {
 	//1、解析canal采集至mq中queue的数据结构体
-	var canalMsg *md.CanalMallOrder
+	var canalMsg *md.CanalO2oPayToMerchantMessage[md.CanalO2oPayToMerchant]
 	err := json.Unmarshal(msg, &canalMsg)
 	if err != nil {
 		return err
@@ -124,7 +124,7 @@ func handleCanalO2oPayOrderForNumericalStatementTable(msg []byte) error {
 					return errors.New("未支付不统计")
 				}
 				if item.State == "1" || item.State == "4" {
-					paymentTotal = utils.Float64ToStr(utils.StrToFloat64(paymentTotal) + utils.StrToFloat64(item.CostPrice))                        //付款金额
+					paymentTotal = utils.Float64ToStr(utils.StrToFloat64(paymentTotal) + utils.StrToFloat64(item.PayAmount))                        //付款金额
 					estimatedCommission = utils.Float64ToStr(utils.StrToFloat64(estimatedCommission) + utils.StrToFloat64(item.EstimateCommission)) //预估佣金(元)
 
 					estimatedCommissionValue, _ := decimal.NewFromString(estimatedCommission)
@@ -151,6 +151,47 @@ func handleCanalO2oPayOrderForNumericalStatementTable(msg []byte) error {
 			}
 		}
 
+		if canalMsg.Type == md.CanalMsgUpdateSqlType {
+			judgeSate := judgeSateFunc(*canalMsg)
+			if judgeSate > 0 {
+				if judgeSate == 4 {
+					//TODO::已支付
+					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.PayAmount))                        //付款金额
+							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
@@ -171,3 +212,41 @@ func handleCanalO2oPayOrderForNumericalStatementTable(msg []byte) error {
 	}
 	return nil
 }
+
+// judgeSateFunc 处理订单状态(judgeSate[0:不需要处理 4:已支付])
+func judgeSateFunc(message md.CanalO2oPayToMerchantMessage[md.CanalO2oPayToMerchant]) (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" && nowOrdState >= "1" {
+		//已支付
+		return 4
+	}
+	return
+}

consume/init.go

@@ -75,15 +75,15 @@ func initConsumes() {
 	//jobs[consumeMd.MallAddSupplyGoodsFunName] = MallAddSupplyGoodsConsume
 
 	//////////////////////////////////////// bigData /////////////////////////////////////////////////////
-	//jobs[consumeMd.CanalOrderConsumeFunName] = CanalOrderConsume
-	//jobs[consumeMd.CanalGuideOrderConsumeFunName] = CanalGuideOrderConsume
-	//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
+	jobs[consumeMd.CanalOrderConsumeFunName] = CanalOrderConsume
+	jobs[consumeMd.CanalGuideOrderConsumeFunName] = CanalGuideOrderConsume
+	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-v1 /////////////////////////////////////////////////////
 	//jobs[consumeMd.OneCirclesSignInGreenEnergyFunName] = OneCirclesSignInGreenEnergyConsume
@@ -108,8 +108,8 @@ func initConsumes() {
 	//jobs[consumeMd.InstallmentPaymentAutoRepaidConsumeFunName] = InstallmentPaymentAutoRepaidConsume //分期付 - 自动扣款
 
 	////////////////////////////////////// SuperCloudIssuance /////////////////////////////////////////////////////
-	jobs[consumeMd.SuperCloudIssuanceMsgCallBackFunName] = SuperCloudIssuanceMsgCallBackConsume
-	jobs[consumeMd.SuperCloudIssuanceAsyncMLoginFunName] = SuperCloudIssuanceAsyncMLoginConsume
+	//jobs[consumeMd.SuperCloudIssuanceMsgCallBackFunName] = SuperCloudIssuanceMsgCallBackConsume
+	//jobs[consumeMd.SuperCloudIssuanceAsyncMLoginFunName] = SuperCloudIssuanceAsyncMLoginConsume
 
 	////////////////////////////////////// DMS /////////////////////////////////////////////////////
 	//jobs[consumeMd.CanalGimMessageConsumeFunName] = CanalGimMessageConsume

consume/md/md_canal_o2o_pay_to_merchant_consume.go

@@ -0,0 +1,39 @@
+package md
+
+type CanalO2oPayToMerchant struct {
+	Id                 string `json:"id" xorm:"pk autoincr BIGINT(20)"`
+	PayId              string `json:"pay_id" xorm:"not null comment('付款id') BIGINT(20)"`
+	Uid                string `json:"uid" xorm:"not null comment('用户id') INT(11)"`
+	MerchantId         string `json:"merchant_id" xorm:"comment('商户id') INT(11)"`
+	BelongStoreId      string `json:"belong_store_id" xorm:"comment('归属店铺id') INT(11)"`
+	PayAmount          string `json:"pay_amount" xorm:"comment('付款金额') DECIMAL(12,4)"`
+	ActualPayAmount    string `json:"actual_pay_amount" xorm:"comment('实际付款金额') DECIMAL(12,4)"`
+	CoinId             string `json:"coin_id" xorm:"comment('虚拟币id') INT(11)"`
+	TradeNo            string `json:"trade_no" xorm:"comment('支付平台的订单号') VARCHAR(50)"`
+	PayTradeNo         string `json:"pay_trade_no" xorm:"comment('支付联盟支付的订单号') VARCHAR(50)"`
+	MainCommission     string `json:"main_commission" xorm:"comment('总佣金') DECIMAL(12,4)"`
+	EstimateCommission string `json:"estimate_commission" xorm:"comment('预计佣金（三方分账完之后站长的佣金）') DECIMAL(12,4)"`
+	PlatformCommission string `json:"platform_commission" xorm:"comment('平台所得的佣金') DECIMAL(12,4)"`
+	PayChannel         string `json:"pay_channel" xorm:"comment('支付方式：1balance 2alipay 3wx_pay 4zhios_pay_alipay') TINYINT(2)"`
+	State              string `json:"state" xorm:"not null default 0 comment('支付状态：0：未支付，1：已支付') TINYINT(2)"`
+	PayTime            string `json:"pay_time" xorm:"comment('支付时间') DATETIME"`
+	Data               string `json:"data" xorm:"comment('回调数据') VARCHAR(5000)"`
+	Remarks            string `json:"remarks" xorm:"comment('备注') VARCHAR(255)"`
+	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"`
+	SettleTime         string `json:"settle_time" xorm:"not null default 'CURRENT_TIMESTAMP' comment('结算时间') DATETIME"`
+	CommissionTime     string `json:"commission_time" xorm:"not null default 'CURRENT_TIMESTAMP' comment('时间') DATETIME"`
+}
+
+type CanalO2oPayToMerchantMessage[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"`
+}