@@ -0,0 +1,23 @@ | |||
package db | |||
import ( | |||
"applet/app/db/model" | |||
"xorm.io/xorm" | |||
) | |||
func GetIntegralProxyRecord(sess *xorm.Session, oid string) *model.IntegralProxyRecord { | |||
var data model.IntegralProxyRecord | |||
get, err := sess.Where("oid=?", oid).Get(&data) | |||
if get == false || err != nil { | |||
return nil | |||
} | |||
return &data | |||
} | |||
func GetIntegralProxyRecordById(eg *xorm.Engine, id string) *model.IntegralProxyRecord { | |||
var data model.IntegralProxyRecord | |||
get, err := eg.Where("id=?", id).Get(&data) | |||
if get == false || err != nil { | |||
return nil | |||
} | |||
return &data | |||
} |
@@ -0,0 +1,32 @@ | |||
package model | |||
import ( | |||
"time" | |||
) | |||
type IntegralProxyRecord struct { | |||
Id int `json:"id" xorm:"not null pk autoincr INT(11)"` | |||
Oid int64 `json:"oid" xorm:"default 0 BIGINT(20)"` | |||
Uid int `json:"uid" xorm:"default 0 INT(11)"` | |||
StoreUid int `json:"store_uid" xorm:"default 0 INT(11)"` | |||
Level int `json:"level" xorm:"default 0 INT(11)"` | |||
Amount string `json:"amount" xorm:"default 0.0000 DECIMAL(20,4)"` | |||
AfterAmount string `json:"after_amount" xorm:"default 0.0000 DECIMAL(20,4)"` | |||
CoinId int `json:"coin_id" xorm:"default 0 INT(11)"` | |||
Money string `json:"money" xorm:"default 0.00 DECIMAL(20,2)"` | |||
PayWay int `json:"pay_way" xorm:"default 0 INT(11)"` | |||
State int `json:"state" xorm:"default 0 INT(1)"` | |||
Time time.Time `json:"time" xorm:"DATETIME"` | |||
PayTime time.Time `json:"pay_time" xorm:"DATETIME"` | |||
OnAccount int `json:"on_account" xorm:"default 0 INT(11)"` | |||
RealOnAccount int `json:"real_on_account" xorm:"default 0 INT(11)"` | |||
Fee string `json:"fee" xorm:"default 0.0000 DECIMAL(20,4)"` | |||
SendTime int `json:"send_time" xorm:"default 0 INT(11)"` | |||
EveryAmount string `json:"every_amount" xorm:"default 0.0000 DECIMAL(20,4)"` | |||
AlipayNo string `json:"alipay_no" xorm:"default '' VARCHAR(100)"` | |||
Date string `json:"date" xorm:"comment('2024-01') VARCHAR(255)"` | |||
Type string `json:"type" xorm:"comment('2024-01') VARCHAR(255)"` | |||
SettleAt int `json:"settle_at" xorm:"default 0 INT(11)"` | |||
CommissionAt int `json:"commission_at" xorm:"default 0 INT(11)"` | |||
RunTime int `json:"run_time" xorm:"default 0 INT(11)"` | |||
} |
@@ -17,56 +17,50 @@ func Init() { | |||
// 增加消费任务队列 | |||
func initConsumes() { | |||
//jobs[consumeMd.ZhiosUserUpLvFunName] = ZhiosUserUpLv | |||
//jobs[consumeMd.CanalGuideOrderByUserUpLvConsume] = CanalGuideOrderByUserUpLvConsume | |||
//jobs[consumeMd.ZhiosOrderFreeFunName] = ZhiosOrderFree | |||
//jobs[consumeMd.ZhiosOrderTotalFunName] = ZhiosOrderTotal | |||
//jobs[consumeMd.ZhiosOrderTotalSecondFunName] = ZhiosOrderTotalSecond | |||
//// | |||
//jobs[consumeMd.ZhiosOrderSettleTotalFunName] = ZhiosSettleTotal | |||
//jobs[consumeMd.ZhiosOrderHjyFunName] = ZhiosOrderHjy | |||
//jobs[consumeMd.ZhiosOrderBuckleFunName] = ZhiosOrderBuckle | |||
//// | |||
//jobs[consumeMd.ZhiosSupplierAfterOrderFunName] = ZhiosSupplierAfterOrder | |||
//jobs[consumeMd.ZhiosGuideStoreOrderFunName] = ZhiosGuideStoreOrder | |||
jobs[consumeMd.ZhiosIntegralProxyRechargeFunName] = ZhiosIntegralProxyRecharge | |||
jobs[consumeMd.ZhiosUserUpLvFunName] = ZhiosUserUpLv | |||
jobs[consumeMd.CanalGuideOrderByUserUpLvConsume] = CanalGuideOrderByUserUpLvConsume | |||
jobs[consumeMd.ZhiosOrderFreeFunName] = ZhiosOrderFree | |||
jobs[consumeMd.ZhiosOrderTotalFunName] = ZhiosOrderTotal | |||
jobs[consumeMd.ZhiosOrderTotalSecondFunName] = ZhiosOrderTotalSecond | |||
// | |||
////jobs[consumeMd.ZhiosAppreciationDevFunName] = ZhiosAppreciation | |||
jobs[consumeMd.ZhiosOrderSettleTotalFunName] = ZhiosSettleTotal | |||
jobs[consumeMd.ZhiosOrderHjyFunName] = ZhiosOrderHjy | |||
jobs[consumeMd.ZhiosOrderBuckleFunName] = ZhiosOrderBuckle | |||
// | |||
//jobs[consumeMd.ZhiosAppreciationFunName] = ZhiosAppreciation | |||
//jobs[consumeMd.ZhiosValidUserFunName] = ZhiosValidUser | |||
// | |||
////jobs[consumeMd.ZhiosAcquisitionConditionDevFunName] = ZhiosAcquisitionCondition | |||
// | |||
//jobs[consumeMd.ZhiosAcquisitionConditionFunName] = ZhiosAcquisitionCondition | |||
//jobs[consumeMd.CanalOrderConsumeFunName] = CanalOrderConsume | |||
//jobs[consumeMd.CanalGuideOrderConsumeFunName] = CanalGuideOrderConsume | |||
//jobs[consumeMd.ZhiOsUserVisitIpAddressConsumeFunName] = ZhiOsUserVisitIpAddressConsume | |||
// | |||
//jobs[consumeMd.DouShenUserRegisterConsumeForOfficialFunName] = DouShenUserRegisterConsumeForOfficial | |||
//jobs[consumeMd.DouShenUserRegisterConsumeForOperationCenterFunName] = DouShenUserRegisterConsumeForOperationCenter | |||
//jobs[consumeMd.DouShenUserRegisterConsumeForMyRecommenderFunName] = DouShenUserRegisterConsumeForMyRecommender | |||
//jobs[consumeMd.DouShenUserRegisterConsumeForMyFansFunName] = DouShenUserRegisterConsumeForMyFans | |||
//jobs[consumeMd.DouShenUserRegisterConsumeForUserRegisterUpLvFunName] = DouShenUserRegisterConsumeForUserRegisterUpLv | |||
// | |||
//jobs[consumeMd.ZhiosFastReturnOrderPayFunName] = ZhiosFastReturnOrderPay | |||
//jobs[consumeMd.ZhiosFastReturnOrderSuccessFunName] = ZhiosFastReturnOrderSuccess | |||
//jobs[consumeMd.ZhiosFastReturnOrderRefundFunName] = ZhiosFastReturnOrderRefund | |||
//jobs[consumeMd.ZhiosFastReturnOrderRefundSecondFunName] = ZhiosFastReturnOrderRefundSecond | |||
// | |||
////jobs[consumeMd.CanalMallOrdForYouMiShangFunName] = CanalMallOrdForYouMiShang | |||
//jobs[consumeMd.YoumishangExchangeStoreFunName] = YoumishangExchangeStore | |||
// | |||
//jobs[consumeMd.ZhiosRechargeOrderFailFunName] = ZhiosRechargeOrderFail | |||
// | |||
//jobs[consumeMd.CloudIssuanceAsyncMLoginFunName] = CloudIssuanceAsyncMLoginConsume | |||
//jobs[consumeMd.ZhiosTikTokUpdateFunName] = ZhiosTikTokUpdate | |||
//jobs[consumeMd.ZhiosTikTokAllUpdateFunName] = ZhiosTikTokAllUpdate | |||
// | |||
//jobs[consumeMd.ZhiosCapitalPoolOrderTotalFunName] = ZhiosCapitalPoolOrderTotal | |||
//jobs[consumeMd.ZhiosExpressOrderFail] = ZhiosExpressOrderFail | |||
//jobs[consumeMd.ZhiosWithdrawReward] = ZhiosWithdrawReward | |||
// | |||
//jobs[consumeMd.ZhiosRechargeOrderFailDevFunName] = ZhiosRechargeOrderFailDev | |||
jobs[consumeMd.ZhiosSupplierAfterOrderFunName] = ZhiosSupplierAfterOrder | |||
jobs[consumeMd.ZhiosGuideStoreOrderFunName] = ZhiosGuideStoreOrder | |||
jobs[consumeMd.ZhiosAppreciationFunName] = ZhiosAppreciation | |||
jobs[consumeMd.ZhiosValidUserFunName] = ZhiosValidUser | |||
jobs[consumeMd.ZhiosAcquisitionConditionFunName] = ZhiosAcquisitionCondition | |||
jobs[consumeMd.CanalOrderConsumeFunName] = CanalOrderConsume | |||
jobs[consumeMd.CanalGuideOrderConsumeFunName] = CanalGuideOrderConsume | |||
jobs[consumeMd.ZhiOsUserVisitIpAddressConsumeFunName] = ZhiOsUserVisitIpAddressConsume | |||
jobs[consumeMd.DouShenUserRegisterConsumeForOfficialFunName] = DouShenUserRegisterConsumeForOfficial | |||
jobs[consumeMd.DouShenUserRegisterConsumeForOperationCenterFunName] = DouShenUserRegisterConsumeForOperationCenter | |||
jobs[consumeMd.DouShenUserRegisterConsumeForMyRecommenderFunName] = DouShenUserRegisterConsumeForMyRecommender | |||
jobs[consumeMd.DouShenUserRegisterConsumeForMyFansFunName] = DouShenUserRegisterConsumeForMyFans | |||
jobs[consumeMd.DouShenUserRegisterConsumeForUserRegisterUpLvFunName] = DouShenUserRegisterConsumeForUserRegisterUpLv | |||
jobs[consumeMd.ZhiosFastReturnOrderPayFunName] = ZhiosFastReturnOrderPay | |||
jobs[consumeMd.ZhiosFastReturnOrderSuccessFunName] = ZhiosFastReturnOrderSuccess | |||
jobs[consumeMd.ZhiosFastReturnOrderRefundFunName] = ZhiosFastReturnOrderRefund | |||
jobs[consumeMd.ZhiosFastReturnOrderRefundSecondFunName] = ZhiosFastReturnOrderRefundSecond | |||
jobs[consumeMd.YoumishangExchangeStoreFunName] = YoumishangExchangeStore | |||
jobs[consumeMd.ZhiosRechargeOrderFailFunName] = ZhiosRechargeOrderFail | |||
jobs[consumeMd.CloudIssuanceAsyncMLoginFunName] = CloudIssuanceAsyncMLoginConsume | |||
jobs[consumeMd.ZhiosTikTokUpdateFunName] = ZhiosTikTokUpdate | |||
jobs[consumeMd.ZhiosTikTokAllUpdateFunName] = ZhiosTikTokAllUpdate | |||
jobs[consumeMd.ZhiosCapitalPoolOrderTotalFunName] = ZhiosCapitalPoolOrderTotal | |||
jobs[consumeMd.ZhiosExpressOrderFail] = ZhiosExpressOrderFail | |||
jobs[consumeMd.ZhiosWithdrawReward] = ZhiosWithdrawReward | |||
////////////////////////////////////// V1 ///////////////////////////////////////////////////// | |||
//jobs[consumeMd.CloudIssuanceMsgCallBackFunName] = CloudIssuanceMsgCallBackConsume | |||
@@ -75,7 +69,7 @@ func initConsumes() { | |||
//jobs[consumeMd.SupplyCloudChainFenxiaoNewChangeFunName] = SupplyCloudChainFenxiaoNewChangeConsume | |||
//////////////////////////////////////// V3 ///////////////////////////////////////////////////// | |||
jobs[consumeMd.MallAddSupplyGoodsFunName] = MallAddSupplyGoodsConsume | |||
//jobs[consumeMd.MallAddSupplyGoodsFunName] = MallAddSupplyGoodsConsume | |||
} | |||
@@ -371,6 +371,15 @@ var RabbitMqQueueKeyList = []*MqQueue{ | |||
BindKey: "", | |||
ConsumeFunName: "ZhiosUserUpLv", | |||
}, | |||
{ | |||
ExchangeName: "zhios.integral_proxy_recharge.exchange", | |||
Name: "zhios_integral_proxy_recharge", | |||
Type: DirectQueueType, | |||
IsPersistent: false, | |||
RoutKey: "integral_proxy_recharge", | |||
BindKey: "", | |||
ConsumeFunName: "ZhiosIntegralProxyRecharge", | |||
}, | |||
//{ | |||
// ExchangeName: "zhios.order_buckle.exchange", | |||
// Name: "zhios_order_buckle_dev", | |||
@@ -383,6 +392,7 @@ var RabbitMqQueueKeyList = []*MqQueue{ | |||
} | |||
const ( | |||
ZhiosIntegralProxyRechargeFunName = "ZhiosIntegralProxyRecharge" | |||
ZhiosUserUpLvFunName = "ZhiosUserUpLv" | |||
CanalGuideOrderByUserUpLvConsume = "CanalGuideOrderByUserUpLvConsume" | |||
ZhiosOrderFreeFunName = "ZhiosOrderFree" | |||
@@ -0,0 +1,131 @@ | |||
package consume | |||
import ( | |||
"applet/app/db" | |||
"applet/app/utils" | |||
"applet/app/utils/logx" | |||
"applet/consume/md" | |||
"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit" | |||
"encoding/json" | |||
"errors" | |||
"fmt" | |||
"github.com/streadway/amqp" | |||
"time" | |||
"xorm.io/xorm" | |||
) | |||
func ZhiosIntegralProxyRecharge(queue md.MqQueue) { | |||
fmt.Println(">>>>>>>>>>>>>>>>>>>>>>>>") | |||
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(">>>>>>>>>>>>>>>>CanalOrderConsume<<<<<<<<<<<<<<<<<<<<<<<<<") | |||
err = handleZhiosIntegralProxyRecharge(res.Body) | |||
//_ = res.Reject(false) | |||
if err == nil { | |||
_ = res.Ack(true) | |||
} | |||
} else { | |||
panic(errors.New("error getting message")) | |||
} | |||
} | |||
fmt.Println("get msg done") | |||
} | |||
func handleZhiosIntegralProxyRecharge(msg []byte) error { | |||
//1、解析canal采集至mq中queue的数据结构体 | |||
var canalMsg *md.ZhiosAppreciation | |||
fmt.Println(string(msg)) | |||
var tmpString string | |||
err := json.Unmarshal(msg, &tmpString) | |||
if err != nil { | |||
fmt.Println("===with", err.Error()) | |||
return err | |||
} | |||
fmt.Println(tmpString) | |||
err = json.Unmarshal([]byte(tmpString), &canalMsg) | |||
if err != nil { | |||
fmt.Println("===with", err.Error()) | |||
return err | |||
} | |||
mid := canalMsg.Mid | |||
eg := db.DBs[mid] | |||
if eg == nil { | |||
return nil | |||
} | |||
sess := eg.NewSession() | |||
defer sess.Close() | |||
sess.Begin() | |||
order := db.GetIntegralProxyRecord(sess, canalMsg.Oid) | |||
now := int(time.Now().Unix()) | |||
if order.State != 1 || order.SendTime > now { | |||
sess.Rollback() | |||
return nil | |||
} | |||
if order.SendTime == 0 { | |||
order.SendTime = int(utils.TimeStdParseUnix(order.PayTime.Format("2006-01-02") + " 00:00:00")) | |||
} else if order.SendTime < int(time.Now().Unix()) { | |||
order.SendTime = int(utils.TimeStdParseUnix(time.Now().Format("2006-01-02") + " 00:00:00")) | |||
} | |||
order.SendTime += 30 * 86400 | |||
amount := "0" | |||
title := "" | |||
str := "代充值" | |||
if order.Type == "gift_goods" { | |||
str = "购买礼包奖励" | |||
} | |||
if canalMsg.Type == "" || canalMsg.Type == "first" { | |||
amount = order.Amount | |||
title = str + "到账" | |||
if order.RealOnAccount > 0 { | |||
order.RealOnAccount-- | |||
title = str + "第" + utils.IntToStr(order.OnAccount-order.RealOnAccount) + "期到账" | |||
amount = order.EveryAmount | |||
} | |||
} | |||
if canalMsg.Type == "second" { | |||
if order.RealOnAccount <= 0 { //已经没了 | |||
sess.Rollback() | |||
return nil | |||
} | |||
order.RealOnAccount-- | |||
title = str + "第" + utils.IntToStr(order.OnAccount-order.RealOnAccount) + "期到账" | |||
amount = order.EveryAmount | |||
} | |||
err = addIntegral(sess, order.CoinId, order.Oid, amount, title, order.Uid) | |||
if err != nil { | |||
sess.Rollback() | |||
return err | |||
} | |||
order.RunTime = utils.StrToInt(time.Now().Format("200601")) | |||
update, err := sess.Where("id=?", order.Id).Cols("run_time,real_on_account,send_time").Update(order) | |||
if update == 0 || err != nil { | |||
sess.Rollback() | |||
return err | |||
} | |||
sess.Commit() | |||
return nil | |||
} | |||
func addIntegral(sess *xorm.Session, coinId int, oid int64, amount, title string, uid int) error { | |||
err := UpdateUserVirtualCoinFinValidAndInterFlow(sess, | |||
amount, title, "0", 1, uid, coinId, 125, oid) | |||
if err != nil { | |||
return err | |||
} | |||
return nil | |||
} |