+ CreateConsumerFromOrders

2019-09-06 13:45:10 +08:00
parent 898955729e
commit 9e0ca4f185
4 changed files with 131 additions and 1 deletions
--- a/business/jxstore/tempop/tempop.go
+++ b/business/jxstore/tempop/tempop.go
@@ -999,3 +999,99 @@ func RefreshMtpsWaybillFee(ctx *jxcontext.Context, isAsync, isContinueWhenError
 	}
 	return hint, err
 }
+
+func CreateConsumerFromOrders(ctx *jxcontext.Context, vendorIDs []int, fromDate, toDate time.Time, isForce, isAsync, isContinueWhenError bool) (hint string, err error) {
+	if utils.IsTimeZero(fromDate) {
+		return "", fmt.Errorf("fromDate必须指定")
+	}
+	if utils.IsTimeZero(toDate) {
+		toDate = time.Now()
+	}
+	fromDate = utils.Time2Date(fromDate)
+	toDate = utils.Time2Date(toDate)
+	if len(vendorIDs) == 0 {
+		vendorIDs = partner.GetPurchasePlatformVendorIDs()
+	}
+	var dateList []time.Time
+	curDate := fromDate
+	for {
+		if toDate.Sub(curDate) < 0 {
+			break
+		}
+		dateList = append(dateList, curDate)
+		curDate = curDate.Add(24 * time.Hour)
+	}
+
+	type GoodsOrderWithOriginal struct {
+		model.GoodsOrder
+		OriginalData string `orm:"type(text)" json:"-"`
+	}
+
+	db := dao.GetDB()
+	rootTask := tasksch.NewParallelTask("RefreshMtpsWaybillFee", tasksch.NewParallelConfig().SetParallelCount(1).SetIsContinueWhenError(isContinueWhenError), ctx,
+		func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
+			curDate := batchItemList[0].(time.Time)
+			sql := `
+			SELECT *
+			FROM goods_order t1
+			LEFT JOIN goods_order_original t2 ON t2.vendor_order_id = t1.vendor_order_id AND t2.vendor_id = t1.vendor_id
+			WHERE t1.order_created_at >= ? AND t1.order_created_at < ?`
+			sqlParams := []interface{}{
+				curDate,
+				curDate.Add(24 * time.Hour),
+			}
+			if len(vendorIDs) > 0 {
+				sql += " AND t1.vendor_id IN (" + dao.GenQuestionMarks(len(vendorIDs)) + ")"
+				sqlParams = append(sqlParams, vendorIDs)
+			}
+			if !isForce {
+				sql += " AND t1.vendor_user_id = ''"
+			}
+			var orderList []*GoodsOrderWithOriginal
+			if err = dao.GetRows(db, &orderList, sql, sqlParams...); err == nil {
+				if len(orderList) > 0 {
+					subTask := tasksch.NewParallelTask("RefreshMtpsWaybillFee", tasksch.NewParallelConfig().SetIsContinueWhenError(isContinueWhenError), ctx,
+						func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
+							order := batchItemList[0].(*GoodsOrderWithOriginal)
+							if isForce || order.VendorUserID == "" {
+								if handler := partner.GetPurchasePlatformFromVendorID(order.VendorID); handler != nil {
+									var order2 *model.GoodsOrder
+									// globals.SugarLogger.Debugf("fuck:%s", order.OriginalData)
+									if order.OriginalData != "" {
+										var orderData map[string]interface{}
+										if utils.UnmarshalUseNumber([]byte(order.OriginalData), &orderData) == nil {
+											order2 = handler.Map2Order(orderData)
+											if order2.VendorUserID == "" {
+												order2 = nil
+											}
+										}
+									}
+									if order2 == nil {
+										order2, err = handler.GetOrder(order.VendorOrderID)
+									}
+									if order2 != nil {
+										isNeedUpdateOrder := order.VendorUserID != order2.VendorUserID
+										order.VendorUserID = order2.VendorUserID
+										if isNeedUpdateOrder {
+											dao.UpdateEntity(db, &order.GoodsOrder, "VendorUserID")
+										}
+										err = cms.HandleOrder4Consignee(&order.GoodsOrder)
+									}
+								}
+							}
+							return nil, err
+						}, orderList)
+					tasksch.HandleTask(subTask, task, true).Run()
+					_, err = subTask.GetResult(0)
+				}
+			}
+			return nil, err
+		}, dateList)
+	tasksch.HandleTask(rootTask, nil, true).Run()
+	if !isAsync {
+		_, err = rootTask.GetResult(0)
+	} else {
+		hint = rootTask.ID
+	}
+	return hint, err
+}