修复订单回款 通道关闭错误

account_order/account_order_change.go

@@ -19,6 +19,7 @@ func dealAllDataAccountOrderChangeRecord() {
 	allDataOrderSaleRecordForAccountOrderChange() //业绩变更表数据处理
 	allDataRefundRecordForAccountOrderChange()    //退款表数据处理
 	log.Info("dealAllDataAccountOrderChangeRecord ", zap.String("处理归集后的业绩变更表", "处理完毕"))
+	close(accountOrderChangePool)
 }
 
 // allDataOrderSaleRecordForAccountOrderChange  业绩变更表数据处理-order_sale_record
@@ -28,7 +29,7 @@ func allDataOrderSaleRecordForAccountOrderChange() {
 			log.Info("Panic recovered", zap.Any("reason", r))
 		}
 
-		log.Info("aaaaaaaaaaaa")
+		log.Info("allDataOrderSaleRecordForAccountOrderChange")
 	}()
 	now := time.Now()
 	var total int64
@@ -130,7 +131,18 @@ func allDataOrderSaleRecordForAccountOrderChange() {
 				if err != nil {
 					log.Info("allDataOrderSaleRecordForAccountOrderChange, filter.json.Unmarshal", zap.Error(err))
 				}
-				account_order_change.ContractMoney = util.IntAll(filterMap["originalAmount"])
+				//合同金额
+				if data_exOrder.IsBackstageOrder == 0 { //线上订单
+					if data_exOrder.OrderStatus == 1 {
+						account_order_change.ContractMoney = util.IntAll(filterMap["originalAmount"])
+					} else {
+						account_order_change.ContractMoney = data_exOrder.OrderMoney
+					}
+				} else {
+					//线下订单
+					account_order_change.ContractMoney = data_exOrder.PayMoney
+				}
+
 				account_order_change.Commission = data_exOrder.Commission
 				account_order_change.ProceduresMoney = data_exOrder.ProceduresMoney
 				account_order_change.ReceivableAmount = account_order_change.ContractMoney - account_order_change.Commission - account_order_change.ProceduresMoney
@@ -288,10 +300,12 @@ func allDataOrderSaleRecordForAccountOrderChange() {
 				//
 				account_order_change.OrderSaleRecordID = order_sale_record.ID
 				//生成订单数据
-				err = AnalysisDB.Create(&account_order_change).Error
-				if err != nil {
-					log.Info("allDataOrderSaleRecordForAccountOrderChange Create ", zap.Error(err))
-				}
+				accountOrderChangePool <- account_order_change
+				//err = AnalysisDB.Create(&account_order_change).Error
+				//if err != nil {
+				//	log.Info("allDataOrderSaleRecordForAccountOrderChange Create ", zap.Error(err))
+				//}
+				//
 			}
 		}
 	}
@@ -307,7 +321,7 @@ func allDataRefundRecordForAccountOrderChange() {
 			log.Info("Panic recovered", zap.Any("reason", r))
 		}
 
-		log.Info("aaaaaaaaaaaa")
+		log.Info("allDataRefundRecordForAccountOrderChange")
 	}()
 	now := time.Now()
 	var total int64
@@ -414,8 +428,18 @@ func allDataRefundRecordForAccountOrderChange() {
 				if err != nil {
 					log.Info("allDataRefundRecordForAccountOrderChange, filter.json.Unmarshal", zap.Error(err))
 				}
-				account_order_change.ContractMoney = util.IntAll(filterMap["originalAmount"]) // 合同金额
-				account_order_change.Commission = data_exOrder.Commission                     //佣金
+
+				// 合同金额
+				if data_exOrder.IsBackstageOrder == 0 { //线上订单
+					if data_exOrder.OrderStatus == 1 {
+						account_order_change.ContractMoney = util.IntAll(filterMap["originalAmount"])
+					} else {
+						account_order_change.ContractMoney = data_exOrder.OrderMoney
+					}
+				} else { //线下订单
+					account_order_change.ContractMoney = data_exOrder.PayMoney
+				}
+				account_order_change.Commission = data_exOrder.Commission //佣金
 				account_order_change.ProceduresMoney = data_exOrder.ProceduresMoney
 				account_order_change.ReceivableAmount = account_order_change.ContractMoney - account_order_change.Commission - account_order_change.ProceduresMoney
 				account_order_change.ChangeValue = 0 - refund_record.RefundMoney //业绩变动额
@@ -571,10 +595,12 @@ func allDataRefundRecordForAccountOrderChange() {
 				//
 				account_order_change.RefundRecordID = refund_record.ID //退款表ID
 				//生成订单数据
-				err = AnalysisDB.Create(&account_order_change).Error
-				if err != nil {
-					log.Info("allDataRefundRecordForAccountOrderChange Create ", zap.Error(err))
-				}
+				accountOrderChangePool <- account_order_change
+				//err = AnalysisDB.Create(&account_order_change).Error
+				//if err != nil {
+				//	log.Info("allDataRefundRecordForAccountOrderChange Create ", zap.Error(err))
+				//}
+				//
 			}
 		}
 

account_order/account_return.go

@@ -14,6 +14,7 @@ import (
 func dealAllDataAccountReturn() {
 	allDataReturnMoneyRecordForAccountReturn() //处理迭代回款表-data_analysis.return_money_record
 	allDataDataexportOrderForAccountReturn()   //处理迭代订单表-data_analysis.dataexport_order
+	close(accountReturnPool)
 }
 
 // allDataReturnMoneyRecordForAccountReturn 回款表数据处理-return_money_record
@@ -22,7 +23,7 @@ func allDataReturnMoneyRecordForAccountReturn() {
 		if r := recover(); r != nil {
 			log.Info("Panic recovered", zap.Any("reason", r))
 		}
-		log.Info("aaaaaaaaaaaa")
+		log.Info("allDataReturnMoneyRecordForAccountReturn")
 	}()
 	now := time.Now()
 	var total int64
@@ -203,10 +204,12 @@ func allDataReturnMoneyRecordForAccountReturn() {
 				//数据状态
 				account_return.ReturnMoneyRecordState = int(return_money_record.State)
 				//生成数据
-				err = AnalysisDB.Create(&account_return).Error
-				if err != nil {
-					log.Info("allDataReturnMoneyRecordForAccountReturn Create ", zap.Error(err))
-				}
+				accountReturnPool <- account_return
+				//err = AnalysisDB.Create(&account_return).Error
+				//if err != nil {
+				//	log.Info("allDataReturnMoneyRecordForAccountReturn Create ", zap.Error(err))
+				//}
+				//
 			}
 		}
 	}
@@ -224,7 +227,7 @@ func allDataDataexportOrderForAccountReturn() {
 		if r := recover(); r != nil {
 			log.Info("Panic recovered", zap.Any("reason", r))
 		}
-		log.Info("aaaaaaaaaaaa")
+		log.Info("allDataDataexportOrderForAccountReturn")
 	}()
 	now := time.Now()
 	var total int64
@@ -380,10 +383,11 @@ func allDataDataexportOrderForAccountReturn() {
 				account_return.DataexportOrderID = data_exOrder.ID
 				account_return.ReturnMoneyRecordState = 1
 				//生成数据
-				err = AnalysisDB.Create(&account_return).Error
-				if err != nil {
-					log.Info("allDataDataexportOrderForAccountReturn Create ", zap.Error(err))
-				}
+				accountReturnPool <- account_return
+				//err = AnalysisDB.Create(&account_return).Error
+				//if err != nil {
+				//	log.Info("allDataDataexportOrderForAccountReturn Create ", zap.Error(err))
+				//}
 			}
 		}
 

account_order/account_return_change.go

@@ -14,7 +14,7 @@ import (
 func dealAllDataAccountReturnChange() {
 	allDataOrderSaleRecordForAccountReturnChange() //迭代业绩变更表-order_sale_record
 	allDataRefundRecordForAccountReturnChange()    //迭代业绩变更表-order_sale_record
-
+	close(accountReturnChangePool)
 }
 
 // allDataOrderSaleRecordForAccountReturnChange  业绩变更表数据处理-jianyu.order_sale_record
@@ -200,10 +200,11 @@ func allDataOrderSaleRecordForAccountReturnChange() {
 					account_return_change.OperatePerson = return_money_record.OperatePerson
 					account_return_change.OrderSaleRecordID = order_sale_record.ID
 					//生成数据
-					err = AnalysisDB.Create(&account_return_change).Error
-					if err != nil {
-						log.Info("allDataOrderSaleRecordForAccountReturnChange Create ", zap.Error(err))
-					}
+					accountReturnChangePool <- account_return_change
+					//err = AnalysisDB.Create(&account_return_change).Error
+					//if err != nil {
+					//	log.Info("allDataOrderSaleRecordForAccountReturnChange Create ", zap.Error(err))
+					//}
 				}
 			}
 		}
@@ -385,10 +386,11 @@ func allDataRefundRecordForAccountReturnChange() {
 				account_return_change.OperatePerson = "jianyu@admin"
 				account_return_change.RefundRecordID = refund_record.ID
 				//生成数据
-				err = AnalysisDB.Create(&account_return_change).Error
-				if err != nil {
-					log.Info("allDataRefundRecordForAccountReturnChange Create ", zap.Error(err))
-				}
+				accountReturnChangePool <- account_return_change
+				//err = AnalysisDB.Create(&account_return_change).Error
+				//if err != nil {
+				//	log.Info("allDataRefundRecordForAccountReturnChange Create ", zap.Error(err))
+				//}
 			}
 		}
 	}

account_order/config.go

@@ -26,6 +26,7 @@ type MgoConf struct {
 // CronConf 定时任务
 type CronConf struct {
 	Spec   string
+	Url    string
 	Start  int
 	End    int
 	Delete int

account_order/config.toml

@@ -20,10 +20,8 @@
 
 [cron] ## 定时任务
 #    spec = "0 */1 * * * *"  ## 5分钟执行一次
-    spec = "0 00 23 * * *"   ## 每天23点执行
-#    start = -1               ## 表示开始时间昨天凌晨
-#    end = 0                  ## 表示截止时间到今天凌晨
-#    delete = -30             ## 删除30天之前的数据
+    spec = "0 03 7,12,15,20 * * *"   ## 每天7,12,15,20点执行
+    url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=0fc090e3-ceda-448d-8aba-d00a2994be7e"## 微信群通知机器人
 
 
 [mysql]
@@ -53,13 +51,7 @@
 #    password= "=PDT49#80Z!RVv52_z"
 #    table = "ods_datamonitoring_bidding"    ## 存储结果数据表
 
-[env]
 
-    spec = "0 00 01 * * *"  ## 定时任务，每天凌晨1点执行
-
-[email]
-    api = "http://172.17.162.36:19281/_send/_mail"
-    to = "wangchengcheng@topnet.net.cn"
 
 ## 日志
 # 日志

account_order/main.go

@@ -15,14 +15,17 @@ 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
+	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"
@@ -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,7 @@ 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))
-		}
-		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))
-				}
-			}
-		}
-
-	}
-
-	log.Info("dealAllDataAccountOrder 迭代结束", zap.Int64("数据总量", total))
-	log.Info("dealAllDataAccountOrder", zap.Any("处理时长", time.Since(now).Minutes()))
-
-}
-
+// dealAllDataAccountOrder2 处理归集后-存量剑鱼订单表-dwd_f_account_order
 func dealAllDataAccountOrder2() {
 	defer func() {
 		if r := recover(); r != nil {
@@ -426,7 +196,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 +219,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,6 +375,7 @@ func dealAllDataAccountOrder2() {
 	wg.Wait()
 
 	log.Info("dealAllDataAccountOrder 迭代结束", zap.Int64("数据总量", total))
+	close(accountOrderPool)
 	log.Info("dealAllDataAccountOrder", zap.Any("处理时长", time.Since(now).Minutes()))
 }
 
@@ -622,6 +403,78 @@ func saveAccountOrder() {
 	}
 }
 
+// saveData 保存数据
+func saveData() {
+	var wg sync.WaitGroup
+	// 启动保存每个通道数据的goroutine
+	wg.Add(4) // 因为有4个通道
+
+	go func() {
+		defer wg.Done() // 完成时减少计数器
+		for accountOrder := range accountOrderPool {
+			err := AnalysisDB.Create(&accountOrder).Error
+			if err != nil {
+				log.Info("dealAllDataAccountOrder Create", zap.Error(err), zap.String("order_code", accountOrder.OrderCode))
+			}
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+		for accountOrderChange := range accountOrderChangePool {
+			err := AnalysisDB.Create(&accountOrderChange).Error
+			if err != nil {
+				log.Info("dealAllDataAccountOrderChange Create", zap.Error(err), zap.String("order_code", accountOrderChange.OrderCode))
+			}
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+		for accountReturn := range accountReturnPool {
+			err := AnalysisDB.Create(&accountReturn).Error
+			if err != nil {
+				log.Info("dealAllDataAccountReturn Create", zap.Error(err), zap.String("order_code", accountReturn.OrderCode))
+			}
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+		for accountReturnChange := range accountReturnChangePool {
+			err := AnalysisDB.Create(&accountReturnChange).Error
+			if err != nil {
+				log.Info("dealAllDataAccountReturnChange Create", zap.Error(err), zap.String("order_code", accountReturnChange.OrderCode))
+			}
+		}
+	}()
+
+	// 等待所有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)))
+		}
+	}
+
+}
+
 // truncateTables 清空订单相关数据表
 func truncateTables() {
 	var (

account_order/tools.go

@@ -0,0 +1,48 @@
+package main
+
+import (
+	"bytes"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"time"
+)
+
+// HTTPRequest 用于发送 HTTP 请求
+func HTTPRequest(method, url string, headers map[string]string, body []byte) (string, error) {
+	// 创建一个新的 HTTP 请求
+	client := &http.Client{
+		Timeout: 10 * time.Second, // 设置请求超时时间
+	}
+
+	// 创建 HTTP 请求
+	req, err := http.NewRequest(method, url, bytes.NewBuffer(body))
+	if err != nil {
+		return "", fmt.Errorf("failed to create request: %v", err)
+	}
+
+	// 设置请求头
+	for key, value := range headers {
+		req.Header.Set(key, value)
+	}
+
+	// 发送请求
+	resp, err := client.Do(req)
+	if err != nil {
+		return "", fmt.Errorf("failed to send request: %v", err)
+	}
+	defer resp.Body.Close()
+
+	// 读取响应体
+	respBody, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return "", fmt.Errorf("failed to read response body: %v", err)
+	}
+
+	// 如果响应状态码不是 200，则返回错误
+	if resp.StatusCode != http.StatusOK {
+		return "", fmt.Errorf("request failed with status: %d", resp.StatusCode)
+	}
+
+	return string(respBody), nil
+}

bidding_tags/config.toml

@@ -52,9 +52,10 @@ loglevel  = "debug"
 format = "json"
 
 [env]
-    start = -1 ## 开始取2天前的数据，
+    start = -1 ## 开始取2天前的数据，开始调整为1小时前
     end = 0 ## 截止取1天前的数据，
-    spec = "0 00 01 * * *"  ## 定时任务，每天凌晨1点执行
+#    spec = "0 00 01 * * *"  ## 定时任务，每天凌晨1点执行
+    spec = "@hourly"  ## 定时任务，每个小时
     esindex = "bidding"
     doai = true  ## 是否启用调用大模型标记
     ## 情报_环境采购

bidding_tags/main.go

@@ -150,10 +150,10 @@ func main() {
 	//where := map[string]interface{}{
 	//	"_id": map[string]interface{}{
 	//		//"$lt": mongodb.StringTOBsonId("6715b7e8b25c3e1debabb8eb"),
-	//		"$lt": mongodb.StringTOBsonId("672ce44bb25c3e1deb1109d7"), //2024-11-08;调用ai8
+	//		"$lt": mongodb.StringTOBsonId("667045f766cf0db42abc123b"), //2024-11-08;调用ai8
 	//	},
 	//}
-	//dealTopInformationAi7(where)//处理存量数据
+	//dealTopInformationAi7(where) //处理存量数据
 	//------------------------------------------------//
 	local, _ := time.LoadLocation("Asia/Shanghai")
 	c := cron.New(cron.WithLocation(local), cron.WithSeconds())
@@ -172,9 +172,19 @@ func main() {
 func dealInc() {
 	jlog.Info("dealInc", zap.String("开始处理增量数据", ""))
 	now := time.Now()
-	start := time.Date(now.Year(), now.Month(), now.Day()+GF.Env.Start, 0, 0, 0, 0, now.Location())
-	end := time.Date(now.Year(), now.Month(), now.Day()+GF.Env.End, 0, 0, 0, 0, now.Location())
+	//start := time.Date(now.Year(), now.Month(), now.Day()+GF.Env.Start, 0, 0, 0, 0, now.Location())
+	//end := time.Date(now.Year(), now.Month(), now.Day()+GF.Env.End, 0, 0, 0, 0, now.Location())
 
+	//where := map[string]interface{}{
+	//	"comeintime": map[string]interface{}{
+	//		"$gte": start.Unix(),
+	//		"$lt":  end.Unix(),
+	//	},
+	//}
+
+	//按照小时处理数据
+	start := time.Date(now.Year(), now.Month(), now.Day(), now.Hour()+GF.Env.Start, 0, 0, 0, now.Location())
+	end := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, now.Location())
 	where := map[string]interface{}{
 		"comeintime": map[string]interface{}{
 			"$gte": start.Unix(),
@@ -192,9 +202,8 @@ func dealInc() {
 
 	// 暂时只使用大模型
 	//dealTopInformationAi6(where)
-	dealTopInformationAi7(where)
+	go dealTopInformationAi7(where)
 
-	jlog.Info("dealInc", zap.Any("开始处理增量数据", "数据处理完毕"))
 }
 
 // dealAll 处理存量数据
@@ -757,7 +766,7 @@ func dealTopInformationAi7(where interface{}) {
 	it := sess.DB(GF.Mongo.DB).C(GF.Mongo.Coll).Find(where).Select(nil).Sort("-_id").Iter()
 	for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
 		if count%100 == 0 {
-			jlog.Info("dealTopInformationAi6", zap.Any("current", count), zap.Any("title", tmp["title"]), zap.Any("_id", tmp["_id"]))
+			jlog.Info("dealTopInformationAi7", zap.Any("current", count), zap.Any("title", tmp["title"]), zap.Any("_id", tmp["_id"]))
 		}
 
 		// 判断之前物业标签是否有无
@@ -833,13 +842,13 @@ func dealTopInformationAi7(where interface{}) {
 	wg.Wait()
 
 	if len(lines) > 0 {
-		resu, err := ZpAI7(key, model, lines)
+		resu, err := ZpAI8(key, model, lines)
 		if err != nil {
 			log.Println(err)
 		}
 		dealResu(resu)
 	}
-	jlog.Info("dealTopInformationAi6", zap.Any("数据处理完毕", count))
+	jlog.Info("dealTopInformationAi7", zap.Any("数据处理完毕", count))
 }
 
 func dealAi(datas []RequestData) {