|
|
@@ -15,22 +15,25 @@ import (
|
|
|
)
|
|
|
|
|
|
var (
|
|
|
- Mgo *mongodb.MongodbSim
|
|
|
- MgoP *mongodb.MongodbSim
|
|
|
- MgoSpider *mongodb.MongodbSim
|
|
|
- AnalysisDB *gorm.DB
|
|
|
- JianyuDB *gorm.DB
|
|
|
- JianyuSubjectDB *gorm.DB
|
|
|
- endTime string //取数据，时间范围的截止时间；如果为空，就取当前时间
|
|
|
- accountOrderPool = make(chan DwdFAccountOrder, 5000)
|
|
|
+ Mgo *mongodb.MongodbSim
|
|
|
+ MgoP *mongodb.MongodbSim
|
|
|
+ MgoSpider *mongodb.MongodbSim
|
|
|
+ AnalysisDB *gorm.DB
|
|
|
+ JianyuDB *gorm.DB
|
|
|
+ JianyuSubjectDB *gorm.DB
|
|
|
+ ClickhouseDB *gorm.DB
|
|
|
+ endTime string //取数据，时间范围的截止时间；如果为空，就取当前时间
|
|
|
+ accountOrderPool = make(chan DwdFAccountOrder, 5000)
|
|
|
+ accountOrderChangePool = make(chan DwdFAccountOrderChange, 5000)
|
|
|
+ accountReturnPool = make(chan DwdFAccountReturn, 5000)
|
|
|
+ accountReturnChangePool = make(chan DwdFAccountReturnChange, 5000)
|
|
|
)
|
|
|
|
|
|
-// var endTime = "2024-11-02"
|
|
|
-
|
|
|
func main() {
|
|
|
local, _ := time.LoadLocation("Asia/Shanghai")
|
|
|
c := cron.New(cron.WithLocation(local), cron.WithSeconds())
|
|
|
- eid, err := c.AddFunc(GF.Cron.Spec, dealData)
|
|
|
+ //eid, err := c.AddFunc(GF.Cron.Spec, dealData) //tidb 数据库
|
|
|
+ eid, err := c.AddFunc(GF.Cron.Spec, dealDataForClickhouse) //Clickhouse数据库
|
|
|
if err != nil {
|
|
|
log.Info("main", zap.Any("AddFunc err", err))
|
|
|
}
|
|
|
@@ -45,11 +48,36 @@ func main() {
|
|
|
// dealData 处理订单相关数据
|
|
|
func dealData() {
|
|
|
//1.清空数据表
|
|
|
- truncateTables() //清空数据表
|
|
|
endTime = time.Now().Format("2006-01-02 15:04:05")
|
|
|
+ if GF.Cron.Url != "" {
|
|
|
+ // 请求头
|
|
|
+ headers := map[string]string{
|
|
|
+ "Content-Type": "application/json",
|
|
|
+ }
|
|
|
+
|
|
|
+ // 请求体
|
|
|
+ body := []byte(`{
|
|
|
+ "msgtype": "text",
|
|
|
+ "text": {
|
|
|
+ "content": "开始清零订单回款数据",
|
|
|
+ "mentioned_list":["@all"]
|
|
|
+ }
|
|
|
+ }`)
|
|
|
+ _, err := HTTPRequest("POST", GF.Cron.Url, headers, body)
|
|
|
+ if err != nil {
|
|
|
+ log.Info("HTTPRequest ", zap.Error(err), zap.String("body", string(body)))
|
|
|
+ }
|
|
|
+ }
|
|
|
+ truncateTables() //清空数据表
|
|
|
log.Info("配置取值时间范围为空", zap.String("默认取值截止时间为：", endTime))
|
|
|
+ //重新打开通道
|
|
|
+ accountOrderPool = make(chan DwdFAccountOrder, 5000)
|
|
|
+ accountOrderChangePool = make(chan DwdFAccountOrderChange, 5000)
|
|
|
+ accountReturnPool = make(chan DwdFAccountReturn, 5000)
|
|
|
+ accountReturnChangePool = make(chan DwdFAccountReturnChange, 5000)
|
|
|
|
|
|
- go saveAccountOrder() //保存订单数据
|
|
|
+ go saveData() //保存订单数据
|
|
|
+ //go saveAccountOrder() //保存订单数据
|
|
|
go dealAllDataAccountOrder2()
|
|
|
//dealAllDataAccountOrder() //1.处理归集后-存量剑鱼订单表-dwd_f_account_order
|
|
|
go dealAllDataAccountOrderChangeRecord() //2.处理归集后-存量业绩表更表-dwd_f_account_order_change
|
|
|
@@ -57,265 +85,46 @@ func dealData() {
|
|
|
go dealAllDataAccountReturnChange() //4.处理归集后-剑鱼回款变更表-dwd_f_account_return_change
|
|
|
}
|
|
|
|
|
|
-// dealAllDataAccountOrder 处理归集后-存量剑鱼订单表-dwd_f_account_order
|
|
|
-func dealAllDataAccountOrder() {
|
|
|
- defer func() {
|
|
|
- if r := recover(); r != nil {
|
|
|
- log.Info("Panic recovered", zap.Any("reason", r))
|
|
|
+func dealDataForClickhouse() {
|
|
|
+ //1.清空数据表
|
|
|
+ endTime = time.Now().Format("2006-01-02 15:04:05")
|
|
|
+ if GF.Cron.Url != "" {
|
|
|
+ // 请求头
|
|
|
+ headers := map[string]string{
|
|
|
+ "Content-Type": "application/json",
|
|
|
}
|
|
|
- log.Info("aaaaaaaaaaaa")
|
|
|
- }()
|
|
|
- now := time.Now()
|
|
|
- var total int64
|
|
|
- AnalysisDB.Debug().Model(&DataexportOrder{}).Where("autoUpdate < ?", endTime).Count(&total)
|
|
|
- log.Info("dealAllDataAccountOrder", zap.Any("总数是", total))
|
|
|
-
|
|
|
- rowsPerPage := 100 // 每页的数量
|
|
|
- totalPages := (int(total) / rowsPerPage) + 1 //总页数
|
|
|
|
|
|
- for page := 0; page < totalPages; page++ {
|
|
|
- offset := page * rowsPerPage
|
|
|
- rows, err := AnalysisDB.Debug().Model(&DataexportOrder{}).Where("autoUpdate < ?", endTime).Order("id desc").Offset(offset).Limit(rowsPerPage).Rows()
|
|
|
- defer func() {
|
|
|
- err = rows.Close()
|
|
|
- if err != nil {
|
|
|
- log.Info("Err rows.Close", zap.Error(err))
|
|
|
- }
|
|
|
- }()
|
|
|
- log.Info("dealAllDataAccountOrder", zap.Any("current page ", page))
|
|
|
- for rows.Next() {
|
|
|
- var dataExOrder DataexportOrder //原来的订单表
|
|
|
- var accountOrder DwdFAccountOrder //归集后的订单表
|
|
|
- var user1 DwdFUserbaseBaseinfo //个人用户表
|
|
|
- var returnRecord ReturnMoneyRecord //原来的回款表
|
|
|
- var returnRecords []ReturnMoneyRecord //原来的回款表
|
|
|
- var contract Contract //合同表
|
|
|
- // ScanRows 方法用于将一行记录扫描至结构体
|
|
|
- err = AnalysisDB.ScanRows(rows, &dataExOrder)
|
|
|
- if err != nil {
|
|
|
- log.Info("dealAllDataAccountOrder,ScanRows err ", zap.Error(err))
|
|
|
- }
|
|
|
-
|
|
|
- filter := dataExOrder.Filter
|
|
|
- filterMap := make(map[string]interface{})
|
|
|
- err := json.Unmarshal([]byte(filter), &filterMap)
|
|
|
- if err != nil {
|
|
|
- log.Info("dealAllDataAccountOrder, filter.json.Unmarshal", zap.Error(err))
|
|
|
- }
|
|
|
- //订单编号不为空
|
|
|
- if dataExOrder.OrderCode != "" {
|
|
|
- // 业务逻辑...
|
|
|
- accountOrder.OrderCode = dataExOrder.OrderCode
|
|
|
- var osr OrderSaleRecord //原来的业绩变更表
|
|
|
- err = JianyuDB.Order("id asc").Where("ordercode = ? ", dataExOrder.OrderCode).First(&osr).Error
|
|
|
- if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
|
|
- log.Info("dealAllDataAccountOrder", zap.Error(err))
|
|
|
- }
|
|
|
- //有业绩表更表记录
|
|
|
- if osr.ID > 0 {
|
|
|
- if osr.SalerName == "-" {
|
|
|
- accountOrder.SalerName = "运营"
|
|
|
- } else {
|
|
|
- accountOrder.SalerName = osr.SalerName
|
|
|
- }
|
|
|
- accountOrder.SalerDept = osr.SalerDept
|
|
|
- }
|
|
|
- accountOrder.CompanyName = dataExOrder.CompanyName
|
|
|
- if len(dataExOrder.UserID) > 20 { //个人身份
|
|
|
- err = JianyuSubjectDB.Where("userid = ? ", dataExOrder.UserID).First(&user1).Error
|
|
|
- if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
|
|
- log.Info("dealAllDataAccountOrder", zap.Error(err))
|
|
|
- }
|
|
|
- accountOrder.UserRegtime = user1.LRegistedate
|
|
|
- } else {
|
|
|
- //企业身份
|
|
|
- rs := new(struct {
|
|
|
- PositionId int `gorm:"column:position_id"`
|
|
|
- LRegistedate *time.Time `gorm:"column:l_registedate"`
|
|
|
- })
|
|
|
- if dataExOrder.UserID != "" {
|
|
|
- sql := fmt.Sprintf(`SELECT im.position_id,ub.l_registedate FROM
|
|
|
- (SELECT userid,position_id FROM Jianyu_subjectdb.dwd_f_userbase_id_mapping WHERE position_id = %s) im
|
|
|
- LEFT JOIN Jianyu_subjectdb.dwd_f_userbase_baseinfo ub
|
|
|
- ON im.userid=ub.userid`, dataExOrder.UserID)
|
|
|
- err = JianyuSubjectDB.Raw(sql).Scan(&rs).Error
|
|
|
- if err != nil {
|
|
|
- log.Info("处理用户订单表数据", zap.String("查询企业用户注册时间失败", dataExOrder.UserID))
|
|
|
- }
|
|
|
- accountOrder.UserRegtime = rs.LRegistedate
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- accountOrder.CreateTime = dataExOrder.CreateTime
|
|
|
- if dataExOrder.IsBackstageOrder == 0 {
|
|
|
- accountOrder.ReturnTime = dataExOrder.PayTime
|
|
|
- accountOrder.TotalReceived = util.IntAll(filterMap["originalAmount"]) //累计已收
|
|
|
- } else if dataExOrder.IsBackstageOrder == 1 {
|
|
|
- err = AnalysisDB.Where("order_code = ? ", dataExOrder.OrderCode).Where("state = 1").Order("return_time asc").First(&returnRecord).Error
|
|
|
- if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
|
|
- log.Info("dealAllDataAccountOrder", zap.Error(err))
|
|
|
- }
|
|
|
- accountOrder.ReturnTime = returnRecord.ReturnTime
|
|
|
- err = AnalysisDB.Where("order_code = ? ", dataExOrder.OrderCode).Find(&returnRecords).Error
|
|
|
- if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
|
|
- log.Info("dealAllDataAccountOrder", zap.Error(err))
|
|
|
- }
|
|
|
- totalMoney := 0
|
|
|
- if len(returnRecords) > 0 {
|
|
|
- for _, v := range returnRecords {
|
|
|
- totalMoney += v.ReturnMoney
|
|
|
- }
|
|
|
- }
|
|
|
- accountOrder.TotalReceived = totalMoney
|
|
|
- }
|
|
|
- accountOrder.SaleTime = dataExOrder.SaleTime
|
|
|
- accountOrder.OriginalPrice = dataExOrder.OriginalPrice
|
|
|
- accountOrder.ContractMoney = util.IntAll(filterMap["originalAmount"])
|
|
|
- accountOrder.Commission = dataExOrder.Commission
|
|
|
- accountOrder.ProceduresMoney = dataExOrder.ProceduresMoney
|
|
|
- accountOrder.ReceivableAmount = accountOrder.ContractMoney - accountOrder.Commission - accountOrder.ProceduresMoney
|
|
|
- accountOrder.ProductType = dataExOrder.ProductType
|
|
|
- if dataExOrder.ProductType == "大会员" || dataExOrder.ProductType == "大会员-子账号" || dataExOrder.ProductType == "大会员-补充包" {
|
|
|
- switch util.IntAll(filterMap["level"]) {
|
|
|
- case 1:
|
|
|
- accountOrder.DataSpec = "专家版"
|
|
|
- case 2:
|
|
|
- accountOrder.DataSpec = "智慧版"
|
|
|
- case 3:
|
|
|
- accountOrder.DataSpec = "商机版"
|
|
|
- case 4:
|
|
|
- accountOrder.DataSpec = "试用版"
|
|
|
- case 5:
|
|
|
- accountOrder.DataSpec = "定制版"
|
|
|
- case 6:
|
|
|
- accountOrder.DataSpec = "商机版2.0"
|
|
|
- case 7:
|
|
|
- accountOrder.DataSpec = "专家版2.0"
|
|
|
- }
|
|
|
- //付费类型 VipType
|
|
|
- switch util.IntAll(filterMap["recordPayType"]) {
|
|
|
- case 1:
|
|
|
- accountOrder.VipType = "购买"
|
|
|
- case 2:
|
|
|
- accountOrder.VipType = "续费"
|
|
|
- case 3:
|
|
|
- accountOrder.VipType = "升级"
|
|
|
- case 4:
|
|
|
- accountOrder.VipType = "试用"
|
|
|
- }
|
|
|
- } else {
|
|
|
- accountOrder.DataSpec = dataExOrder.DataSpec
|
|
|
- ////付费类型 VipType
|
|
|
- switch dataExOrder.VipType {
|
|
|
- case 0:
|
|
|
- accountOrder.VipType = "购买"
|
|
|
- case 1:
|
|
|
- accountOrder.VipType = "续费"
|
|
|
- case 2:
|
|
|
- accountOrder.VipType = "升级"
|
|
|
- case 3:
|
|
|
- accountOrder.VipType = "试用"
|
|
|
- }
|
|
|
- }
|
|
|
- //OrderStatus 订单状态
|
|
|
- if dataExOrder.RefundStatus == 1 || dataExOrder.RefundStatus == 2 {
|
|
|
- accountOrder.OrderStatus = "已完成"
|
|
|
- } else {
|
|
|
- switch dataExOrder.OrderStatus {
|
|
|
- case 1:
|
|
|
- accountOrder.OrderStatus = "已完成"
|
|
|
- case -1:
|
|
|
- accountOrder.OrderStatus = "逻辑删除"
|
|
|
- case -2:
|
|
|
- accountOrder.OrderStatus = "已取消"
|
|
|
- case -3:
|
|
|
- accountOrder.OrderStatus = "已取消"
|
|
|
- case 0:
|
|
|
- accountOrder.OrderStatus = "未完成"
|
|
|
- }
|
|
|
- }
|
|
|
- //回款状态
|
|
|
- if dataExOrder.IsBackstageOrder == 0 {
|
|
|
- if accountOrder.OrderStatus == "已支付" {
|
|
|
- accountOrder.ReturnStatus = "全额回款"
|
|
|
- } else {
|
|
|
- accountOrder.ReturnStatus = "未支付"
|
|
|
- }
|
|
|
- } else if dataExOrder.IsBackstageOrder == 1 {
|
|
|
- switch dataExOrder.ReturnStatus {
|
|
|
- case 1:
|
|
|
- accountOrder.ReturnStatus = "全额回款"
|
|
|
- case 2:
|
|
|
- accountOrder.ReturnStatus = "部分回款"
|
|
|
- case 0:
|
|
|
- accountOrder.ReturnStatus = "未回款"
|
|
|
- }
|
|
|
- }
|
|
|
- //refund_status 回款状态
|
|
|
- switch dataExOrder.RefundStatus {
|
|
|
- case 0:
|
|
|
- accountOrder.RefundStatus = "未退款"
|
|
|
- case 1:
|
|
|
- accountOrder.RefundStatus = "全额退款"
|
|
|
- case 2:
|
|
|
- accountOrder.RefundStatus = "部分退款"
|
|
|
- }
|
|
|
- //
|
|
|
- accountOrder.UserPhone = dataExOrder.UserPhone
|
|
|
- accountOrder.UserID = dataExOrder.UserID
|
|
|
- if dataExOrder.VipStarttime == nil || dataExOrder.VipStarttime.Year() > 1000 {
|
|
|
- accountOrder.VipStarttime = dataExOrder.VipStarttime
|
|
|
- accountOrder.VipEndtime = dataExOrder.VipEndtime
|
|
|
- }
|
|
|
- // 合同状态
|
|
|
- err = AnalysisDB.Where("order_code = ? ", dataExOrder.OrderCode).Find(&contract).Error
|
|
|
- if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
|
|
- log.Info("dealAllDataAccountOrder", zap.Error(err))
|
|
|
- }
|
|
|
- if contract.ContractStatus == 1 {
|
|
|
- accountOrder.ContractStatus = "已签协议"
|
|
|
- } else if contract.ContractStatus == 0 {
|
|
|
- accountOrder.ContractStatus = "未签协议"
|
|
|
- }
|
|
|
- accountOrder.ContractCode = contract.ContractCode
|
|
|
- accountOrder.ContractTime = contract.ContractTime
|
|
|
- if dataExOrder.SigningSubject == "h01" {
|
|
|
- accountOrder.SigningSubject = "北京剑鱼信息技术有限公司"
|
|
|
- } else if dataExOrder.SigningSubject == "h02" {
|
|
|
- accountOrder.SigningSubject = "北京拓普丰联信息科技股份有限公司"
|
|
|
- }
|
|
|
- accountOrder.OrderChannel = dataExOrder.OrderChannel
|
|
|
- accountOrder.DistributionChannel = dataExOrder.DistributionChannel
|
|
|
- if dataExOrder.IsBackstageOrder == 1 {
|
|
|
- accountOrder.IsBackstageOrder = "是"
|
|
|
- } else {
|
|
|
- accountOrder.IsBackstageOrder = "否"
|
|
|
- }
|
|
|
- switch dataExOrder.PayWay {
|
|
|
- case "ali", "ali_app", "ali_pc": //ali、ali_app、ali_pc 处理为“支付宝”
|
|
|
- accountOrder.PayWay = "支付宝"
|
|
|
- case "wx", "wx_app", "wx_js", "wx_pc": // wx、wx_app、wx_js、wx_pc 处理为“微信”
|
|
|
- accountOrder.PayWay = "微信"
|
|
|
- case "transferAccounts":
|
|
|
- accountOrder.PayWay = "对公转账"
|
|
|
- default:
|
|
|
- accountOrder.PayWay = dataExOrder.PayWay
|
|
|
- }
|
|
|
- //生成订单数据
|
|
|
- err = AnalysisDB.Create(&accountOrder).Error
|
|
|
- if err != nil {
|
|
|
- log.Info("dealAllDataAccountOrder Create ", zap.Error(err))
|
|
|
- }
|
|
|
+ // 请求体
|
|
|
+ body := []byte(`{
|
|
|
+ "msgtype": "text",
|
|
|
+ "text": {
|
|
|
+ "content": "开始清空 Clickhouse 订单回款相关数据",
|
|
|
+ "mentioned_list":["@all"]
|
|
|
}
|
|
|
+ }`)
|
|
|
+ _, err := HTTPRequest("POST", GF.Cron.Url, headers, body)
|
|
|
+ if err != nil {
|
|
|
+ log.Info("HTTPRequest ", zap.Error(err), zap.String("body", string(body)))
|
|
|
}
|
|
|
-
|
|
|
}
|
|
|
+ //clickhouse 数据清空
|
|
|
+ truncateClickhouse()
|
|
|
+ log.Info("配置取值时间范围为空", zap.String("默认取值截止时间为：", endTime))
|
|
|
+ //重新打开通道
|
|
|
+ accountOrderPool = make(chan DwdFAccountOrder, 5000)
|
|
|
+ accountOrderChangePool = make(chan DwdFAccountOrderChange, 5000)
|
|
|
+ accountReturnPool = make(chan DwdFAccountReturn, 5000)
|
|
|
+ accountReturnChangePool = make(chan DwdFAccountReturnChange, 5000)
|
|
|
|
|
|
- log.Info("dealAllDataAccountOrder 迭代结束", zap.Int64("数据总量", total))
|
|
|
- log.Info("dealAllDataAccountOrder", zap.Any("处理时长", time.Since(now).Minutes()))
|
|
|
+ go saveDataClickhouse() //保存订单数据
|
|
|
+ go dealAllDataAccountOrder2()
|
|
|
+ go dealAllDataAccountOrderChangeRecord() //2.处理归集后-存量业绩表更表-dwd_f_account_order_change
|
|
|
+ go dealAllDataAccountReturn() //3.处理归集后-存量剑鱼回款表-dwd_f_account_return
|
|
|
+ go dealAllDataAccountReturnChange() //4.处理归集后-剑鱼回款变更表-dwd_f_account_return_change
|
|
|
|
|
|
}
|
|
|
|
|
|
+// dealAllDataAccountOrder2 处理归集后-存量剑鱼订单表-dwd_f_account_order
|
|
|
func dealAllDataAccountOrder2() {
|
|
|
defer func() {
|
|
|
if r := recover(); r != nil {
|
|
|
@@ -426,7 +235,7 @@ func dealAllDataAccountOrder2() {
|
|
|
}
|
|
|
|
|
|
accountOrder.CreateTime = dataExOrder.CreateTime
|
|
|
- if dataExOrder.IsBackstageOrder == 0 {
|
|
|
+ if dataExOrder.IsBackstageOrder == 0 { //线上订单
|
|
|
accountOrder.ReturnTime = dataExOrder.PayTime
|
|
|
accountOrder.TotalReceived = util.IntAll(filterMap["originalAmount"]) //累计已收
|
|
|
} else if dataExOrder.IsBackstageOrder == 1 {
|
|
|
@@ -449,7 +258,17 @@ func dealAllDataAccountOrder2() {
|
|
|
}
|
|
|
accountOrder.SaleTime = dataExOrder.SaleTime
|
|
|
accountOrder.OriginalPrice = dataExOrder.OriginalPrice
|
|
|
- accountOrder.ContractMoney = util.IntAll(filterMap["originalAmount"])
|
|
|
+ //合同金额
|
|
|
+ if dataExOrder.IsBackstageOrder == 0 { //线上订单
|
|
|
+ if dataExOrder.OrderStatus == 1 {
|
|
|
+ accountOrder.ContractMoney = util.IntAll(filterMap["originalAmount"])
|
|
|
+ } else {
|
|
|
+ accountOrder.ContractMoney = dataExOrder.OrderMoney
|
|
|
+ }
|
|
|
+ } else { //线下订单
|
|
|
+ accountOrder.ContractMoney = dataExOrder.PayMoney
|
|
|
+ }
|
|
|
+
|
|
|
accountOrder.Commission = dataExOrder.Commission
|
|
|
accountOrder.ProceduresMoney = dataExOrder.ProceduresMoney
|
|
|
accountOrder.ReceivableAmount = accountOrder.ContractMoney - accountOrder.Commission - accountOrder.ProceduresMoney
|
|
|
@@ -595,31 +414,260 @@ func dealAllDataAccountOrder2() {
|
|
|
wg.Wait()
|
|
|
|
|
|
log.Info("dealAllDataAccountOrder 迭代结束", zap.Int64("数据总量", total))
|
|
|
+ close(accountOrderPool)
|
|
|
log.Info("dealAllDataAccountOrder", zap.Any("处理时长", time.Since(now).Minutes()))
|
|
|
}
|
|
|
|
|
|
-// saveAccountOrder 保存归集订单表
|
|
|
-func saveAccountOrder() {
|
|
|
- defer func() {
|
|
|
- if r := recover(); r != nil {
|
|
|
- log.Info("saveAccountOrder Panic recovered", zap.Any("reason", r))
|
|
|
+// saveData 保存数据
|
|
|
+func saveData() {
|
|
|
+ var wg sync.WaitGroup
|
|
|
+ // 启动保存每个通道数据的goroutine
|
|
|
+ wg.Add(4) // 因为有4个通道
|
|
|
+
|
|
|
+ //1.DwdFAccountOrder
|
|
|
+ go func() {
|
|
|
+ defer wg.Done() // 完成时减少计数器
|
|
|
+ batchSize := 100 // 设置批量插入大小
|
|
|
+ var batch []DwdFAccountOrder
|
|
|
+ for accountOrder := range accountOrderPool {
|
|
|
+ batch = append(batch, accountOrder)
|
|
|
+ if len(batch) >= batchSize {
|
|
|
+ err := AnalysisDB.Model(DwdFAccountOrder{}).CreateInBatches(batch, batchSize).Error
|
|
|
+ if err != nil {
|
|
|
+ log.Info("saveData DwdFAccountOrder", zap.Error(err))
|
|
|
+ }
|
|
|
+ batch = nil // 清空当前批次
|
|
|
+ }
|
|
|
+ }
|
|
|
+ // 插入剩余的批次
|
|
|
+ if len(batch) > 0 {
|
|
|
+ err := AnalysisDB.Model(DwdFAccountOrder{}).CreateInBatches(batch, len(batch)).Error
|
|
|
+ if err != nil {
|
|
|
+ log.Info("saveData DwdFAccountOrder", zap.Error(err))
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }()
|
|
|
+ //2.DwdFAccountOrderChange
|
|
|
+ go func() {
|
|
|
+ defer wg.Done()
|
|
|
+ batchSize := 100 // 设置批量插入大小
|
|
|
+ var batch []DwdFAccountOrderChange
|
|
|
+ for accountOrderChange := range accountOrderChangePool {
|
|
|
+ batch = append(batch, accountOrderChange)
|
|
|
+ if len(batch) >= batchSize {
|
|
|
+ err := AnalysisDB.Model(DwdFAccountOrderChange{}).CreateInBatches(batch, batchSize).Error
|
|
|
+ if err != nil {
|
|
|
+ log.Info("saveData DwdFAccountOrderChange", zap.Error(err))
|
|
|
+ }
|
|
|
+ batch = nil // 清空当前批次
|
|
|
+ }
|
|
|
+ }
|
|
|
+ // 插入剩余的批次
|
|
|
+ if len(batch) > 0 {
|
|
|
+ err := AnalysisDB.Model(DwdFAccountOrderChange{}).CreateInBatches(batch, len(batch)).Error
|
|
|
+ if err != nil {
|
|
|
+ log.Info("saveData DwdFAccountOrderChange", zap.Error(err))
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }()
|
|
|
+ //3.DwdFAccountReturn
|
|
|
+ go func() {
|
|
|
+ defer wg.Done()
|
|
|
+ batchSize := 100 // 设置批量插入大小
|
|
|
+ var batch []DwdFAccountReturn
|
|
|
+ for accountReturn := range accountReturnPool {
|
|
|
+ batch = append(batch, accountReturn)
|
|
|
+ if len(batch) >= batchSize {
|
|
|
+ err := AnalysisDB.Model(DwdFAccountReturn{}).CreateInBatches(batch, batchSize).Error
|
|
|
+ if err != nil {
|
|
|
+ log.Info("saveData DwdFAccountReturn", zap.Error(err))
|
|
|
+ }
|
|
|
+ batch = nil // 清空当前批次
|
|
|
+ }
|
|
|
+ }
|
|
|
+ // 插入剩余的批次
|
|
|
+ if len(batch) > 0 {
|
|
|
+ err := AnalysisDB.Model(DwdFAccountReturn{}).CreateInBatches(batch, len(batch)).Error
|
|
|
+ if err != nil {
|
|
|
+ log.Info("saveData DwdFAccountReturn", zap.Error(err))
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }()
|
|
|
+ //4.DwdFAccountReturnChange
|
|
|
+ go func() {
|
|
|
+ defer wg.Done()
|
|
|
+ batchSize := 100 // 设置批量插入大小
|
|
|
+ var batch []DwdFAccountReturnChange
|
|
|
+ for accountReturnChange := range accountReturnChangePool {
|
|
|
+ batch = append(batch, accountReturnChange)
|
|
|
+ if len(batch) >= batchSize {
|
|
|
+ err := AnalysisDB.Model(DwdFAccountReturnChange{}).CreateInBatches(batch, batchSize).Error
|
|
|
+ if err != nil {
|
|
|
+ log.Info("saveData DwdFAccountReturnChange", zap.Error(err))
|
|
|
+ }
|
|
|
+ batch = nil // 清空当前批次
|
|
|
+ }
|
|
|
+ }
|
|
|
+ // 插入剩余的批次
|
|
|
+ if len(batch) > 0 {
|
|
|
+ err := AnalysisDB.Model(DwdFAccountReturnChange{}).CreateInBatches(batch, len(batch)).Error
|
|
|
+ if err != nil {
|
|
|
+ log.Info("saveData DwdFAccountReturnChange", zap.Error(err))
|
|
|
+ }
|
|
|
}
|
|
|
- log.Info("saveAccountOrder over")
|
|
|
}()
|
|
|
|
|
|
- for {
|
|
|
- select {
|
|
|
- case accountOrder, ok := <-accountOrderPool:
|
|
|
- if !ok {
|
|
|
- // 通道已关闭，退出循环
|
|
|
- return
|
|
|
+ // 等待所有goroutine完成
|
|
|
+ wg.Wait()
|
|
|
+
|
|
|
+ // 所有通道的数据都处理完毕后打印日志
|
|
|
+ log.Info("All account data has been processed and saved successfully.")
|
|
|
+ if GF.Cron.Url != "" {
|
|
|
+ // 请求头
|
|
|
+ headers := map[string]string{
|
|
|
+ "Content-Type": "application/json",
|
|
|
+ }
|
|
|
+ // 请求体
|
|
|
+ body := []byte(`{
|
|
|
+ "msgtype": "text",
|
|
|
+ "text": {
|
|
|
+ "content": "订单回款数据处理完毕",
|
|
|
+ "mentioned_list":["@all"]
|
|
|
+ }
|
|
|
+ }`)
|
|
|
+ _, err := HTTPRequest("POST", GF.Cron.Url, headers, body)
|
|
|
+ if err != nil {
|
|
|
+ log.Info("HTTPRequest ", zap.Error(err), zap.String("body", string(body)))
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+}
|
|
|
+
|
|
|
+// saveDataClickhouse 保存数据，Clickhouse
|
|
|
+func saveDataClickhouse() {
|
|
|
+ var wg sync.WaitGroup
|
|
|
+ //// 启动保存每个通道数据的goroutine
|
|
|
+ wg.Add(4) // 因为有4个通道
|
|
|
+ //1.DwdFAccountOrder
|
|
|
+ go func() {
|
|
|
+ defer wg.Done()
|
|
|
+ batchSize := 100 // 设置批量插入大小
|
|
|
+ var batch []DwdFAccountOrder
|
|
|
+ for accountOrder := range accountOrderPool {
|
|
|
+ batch = append(batch, accountOrder)
|
|
|
+ if len(batch) >= batchSize {
|
|
|
+ err := ClickhouseDB.Model(DwdFOrder{}).CreateInBatches(batch, batchSize).Error
|
|
|
+ if err != nil {
|
|
|
+ log.Info("saveDataClickhouse DwdFOrder", zap.Error(err))
|
|
|
+ }
|
|
|
+ batch = nil // 清空当前批次
|
|
|
+ }
|
|
|
+ }
|
|
|
+ // 插入剩余的批次
|
|
|
+ if len(batch) > 0 {
|
|
|
+ err := ClickhouseDB.Model(DwdFOrder{}).CreateInBatches(batch, len(batch)).Error
|
|
|
+ if err != nil {
|
|
|
+ log.Info("saveDataClickhouse DwdFOrder", zap.Error(err))
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }()
|
|
|
+ //
|
|
|
+
|
|
|
+ //2.DwdFAccountOrderChange
|
|
|
+ go func() {
|
|
|
+ defer wg.Done()
|
|
|
+ batchSize := 100 // 设置批量插入大小
|
|
|
+ var batch []DwdFAccountOrderChange
|
|
|
+ for accountOrderChange := range accountOrderChangePool {
|
|
|
+ batch = append(batch, accountOrderChange)
|
|
|
+ if len(batch) >= batchSize {
|
|
|
+ err := ClickhouseDB.Model(DwdFOrderChange{}).CreateInBatches(batch, batchSize).Error
|
|
|
+ if err != nil {
|
|
|
+ log.Info("saveDataClickhouse DwdFOrderChange", zap.Error(err))
|
|
|
+ }
|
|
|
+ batch = nil // 清空当前批次
|
|
|
+ }
|
|
|
+ }
|
|
|
+ // 插入剩余的批次
|
|
|
+ if len(batch) > 0 {
|
|
|
+ err := ClickhouseDB.Model(DwdFOrderChange{}).CreateInBatches(batch, len(batch)).Error
|
|
|
+ if err != nil {
|
|
|
+ log.Info("saveDataClickhouse DwdFOrderChange", zap.Error(err))
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }()
|
|
|
+ //3.DwdFAccountReturn
|
|
|
+ go func() {
|
|
|
+ defer wg.Done()
|
|
|
+ batchSize := 100 // 设置批量插入大小
|
|
|
+ var batch []DwdFAccountReturn
|
|
|
+ for accountReturn := range accountReturnPool {
|
|
|
+ batch = append(batch, accountReturn)
|
|
|
+ if len(batch) >= batchSize {
|
|
|
+ err := ClickhouseDB.Model(DwdFReturn{}).CreateInBatches(batch, batchSize).Error
|
|
|
+ if err != nil {
|
|
|
+ log.Info("saveDataClickhouse DwdFReturn", zap.Error(err))
|
|
|
+ }
|
|
|
+ batch = nil // 清空当前批次
|
|
|
+ }
|
|
|
+ }
|
|
|
+ // 插入剩余的批次
|
|
|
+ if len(batch) > 0 {
|
|
|
+ err := ClickhouseDB.Model(DwdFReturn{}).CreateInBatches(batch, len(batch)).Error
|
|
|
+ if err != nil {
|
|
|
+ log.Info("saveDataClickhouse DwdFReturn", zap.Error(err))
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ }()
|
|
|
+ //4.DwdFAccountReturnChange
|
|
|
+ go func() {
|
|
|
+ defer wg.Done()
|
|
|
+ batchSize := 100 // 设置批量插入大小
|
|
|
+ var batch []DwdFAccountReturnChange
|
|
|
+ for accountReturnChange := range accountReturnChangePool {
|
|
|
+ batch = append(batch, accountReturnChange)
|
|
|
+ if len(batch) >= batchSize {
|
|
|
+ err := ClickhouseDB.Model(DwdFReturnChange{}).CreateInBatches(batch, batchSize).Error
|
|
|
+ if err != nil {
|
|
|
+ log.Info("saveDataClickhouse DwdFReturnChange", zap.Error(err))
|
|
|
+ }
|
|
|
+ batch = nil // 清空当前批次
|
|
|
}
|
|
|
- err := AnalysisDB.Create(&accountOrder).Error
|
|
|
+ }
|
|
|
+ // 插入剩余的批次
|
|
|
+ if len(batch) > 0 {
|
|
|
+ err := ClickhouseDB.Model(DwdFReturnChange{}).CreateInBatches(batch, len(batch)).Error
|
|
|
if err != nil {
|
|
|
- log.Info("dealAllDataAccountOrder Create ", zap.Error(err), zap.String("order_code", accountOrder.OrderCode))
|
|
|
+ log.Info("saveDataClickhouse DwdFReturnChange", zap.Error(err))
|
|
|
}
|
|
|
}
|
|
|
+ }()
|
|
|
+
|
|
|
+ // 等待所有goroutine完成
|
|
|
+ wg.Wait()
|
|
|
+
|
|
|
+ // 所有通道的数据都处理完毕后打印日志
|
|
|
+ log.Info("All clickhouse data has been processed and saved successfully.")
|
|
|
+ if GF.Cron.Url != "" {
|
|
|
+ // 请求头
|
|
|
+ headers := map[string]string{
|
|
|
+ "Content-Type": "application/json",
|
|
|
+ }
|
|
|
+ // 请求体
|
|
|
+ body := []byte(`{
|
|
|
+ "msgtype": "text",
|
|
|
+ "text": {
|
|
|
+ "content": "Clickhouse 订单回款数据处理完毕",
|
|
|
+ "mentioned_list":["@all"]
|
|
|
+ }
|
|
|
+ }`)
|
|
|
+ _, err := HTTPRequest("POST", GF.Cron.Url, headers, body)
|
|
|
+ if err != nil {
|
|
|
+ log.Info("HTTPRequest ", zap.Error(err), zap.String("body", string(body)))
|
|
|
+ }
|
|
|
}
|
|
|
+
|
|
|
}
|
|
|
|
|
|
// truncateTables 清空订单相关数据表
|
