jx-callback/business/jxstore/initdata/temp_op.go

package initdata

import (
	"time"

	"git.rosy.net.cn/baseapi/utils"
	"git.rosy.net.cn/jx-callback/business/jxutils/jxcontext"
	"git.rosy.net.cn/jx-callback/business/jxutils/tasksch"
	"git.rosy.net.cn/jx-callback/business/model"
	"git.rosy.net.cn/jx-callback/business/model/dao"
	"git.rosy.net.cn/jx-callback/business/model/legacymodel2"
	"git.rosy.net.cn/jx-callback/globals"
	"git.rosy.net.cn/jx-callback/globals/api"
)

func TransferLegacyJdOrder(ctx *jxcontext.Context, isAsync, isContinueWhenError bool) (hint string, err error) {
	sql := `
	SELECT t1.*
	FROM jdorder t1
	LEFT JOIN goods_order t2 ON t2.vendor_order_id = t1.vendor_order_id
	WHERE t2.id IS NULL
	`
	db := dao.GetDB()
	var jdOrderList []*legacymodel2.Jdorder
	if err = dao.GetRows(db, &jdOrderList, sql); err != nil {
		return "", err
	}

	task := tasksch.NewParallelTask("TransferLegacyJdOrder", tasksch.NewParallelConfig().SetIsContinueWhenError(isContinueWhenError).SetBatchSize(40), ctx.GetUserName(), func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
		var orderDetailList []*model.GoodsOrderOriginal
		for _, v := range batchItemList {
			jdOrder := v.(*legacymodel2.Jdorder)
			var detail map[string]interface{}
			if err = utils.UnmarshalUseNumber([]byte(jdOrder.Data), &detail); err != nil {
				return nil, err
			}
			resultList := detail["result"].(map[string]interface{})["resultList"].([]interface{})
			if len(resultList) > 0 {
				originalData := resultList[0].(map[string]interface{})
				if originalData["orgCode"].(string) == "320406" {
					orderDetail := &model.GoodsOrderOriginal{
						VendorOrderID:  jdOrder.VendorOrderID,
						VendorID:       model.VendorIDJD,
						OrderCreatedAt: utils.Str2Time(originalData["orderPurchaseTime"].(string)),
						OriginalData:   string(utils.MustMarshal(originalData)),
					}
					orderDetailList = append(orderDetailList, orderDetail)
				}
			}
		}
		if len(orderDetailList) > 0 {
			err = dao.CreateMultiEntities(db, orderDetailList)
		}
		return nil, err
	}, jdOrderList)
	tasksch.ManageTask(task).Run()

	if !isAsync {
		_, err = task.GetResult(0)
	} else {
		hint = task.ID
	}
	return hint, err
}

func TransferLegacyElmOrder(ctx *jxcontext.Context, isAsync, isContinueWhenError bool) (hint string, err error) {
	sql := `
	SELECT t1.*
	FROM elemeorder t1
	LEFT JOIN goods_order t2 ON t2.vendor_order_id = t1.orderid
	WHERE t2.id IS NULL
	`
	db := dao.GetDB()
	var elmOrderList []*legacymodel2.Elemeorder
	if err = dao.GetRows(db, &elmOrderList, sql); err != nil {
		return "", err
	}

	task := tasksch.NewParallelTask("TransferLegacyElmOrder", tasksch.NewParallelConfig().SetIsContinueWhenError(isContinueWhenError).SetBatchSize(40), ctx.GetUserName(), func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
		var orderDetailList []*model.GoodsOrderOriginal
		for _, v := range batchItemList {
			elmOrder := v.(*legacymodel2.Elemeorder)
			var detail map[string]interface{}
			if err = utils.UnmarshalUseNumber([]byte(elmOrder.Data), &detail); err != nil {
				return nil, err
			}
			orderDetail := &model.GoodsOrderOriginal{
				VendorOrderID:  elmOrder.Orderid,
				VendorID:       model.VendorIDELM,
				OrderCreatedAt: utils.Str2Time(detail["activeAt"].(string)),
				OriginalData:   elmOrder.Data,
			}
			orderDetailList = append(orderDetailList, orderDetail)
		}
		if len(orderDetailList) > 0 {
			err = dao.CreateMultiEntities(db, orderDetailList)
		}
		return nil, err
	}, elmOrderList)
	tasksch.ManageTask(task).Run()

	if !isAsync {
		_, err = task.GetResult(0)
	} else {
		hint = task.ID
	}
	return hint, err
}

func saveJdOrderList(existJdIDMap map[string]int, jdOrderList []interface{}) (err error) {
	var orderDetailList []*model.GoodsOrderOriginal
	for _, v := range jdOrderList {
		orderMap := v.(map[string]interface{})
		orderID := utils.Int64ToStr(utils.MustInterface2Int64(orderMap["orderId"]))
		if existJdIDMap[orderID] == 0 {
			orderDetail := &model.GoodsOrderOriginal{
				VendorOrderID:  orderID,
				VendorID:       model.VendorIDJD,
				OrderCreatedAt: utils.Str2Time(orderMap["orderPurchaseTime"].(string)),
				OriginalData:   string(utils.MustMarshal(orderMap)),
			}
			orderDetailList = append(orderDetailList, orderDetail)
		}
	}
	if len(orderDetailList) > 0 {
		// for _, v := range orderDetailList {
		// 	globals.SugarLogger.Debug(v.VendorOrderID)
		// }
		// db := dao.GetDB()
		// err = dao.CreateMultiEntities(db, orderDetailList)
	}
	return err
}

func PullJdOrder(ctx *jxcontext.Context, fromTime, toTime time.Time, isAsync, isContinueWhenError bool) (hint string, err error) {
	var existJdIDs []string
	db := dao.GetDB()
	if err = dao.GetRows(db, &existJdIDs, `
	SELECT vendor_order_id
	FROM goods_order_original
	WHERE vendor_id = ?`, model.VendorIDJD); err != nil {
		return "", err
	}
	existJdIDMap := make(map[string]int)
	for _, v := range existJdIDs {
		existJdIDMap[v] = 1
	}

	pageSize := 100
	hours := int((toTime.Sub(fromTime)-time.Hour)/time.Hour + 1)
	globals.SugarLogger.Debug(hours)
	rootTask := tasksch.NewSeqTask("PullJdOrder", ctx.GetUserName(), func(task *tasksch.SeqTask, step int, params ...interface{}) (result interface{}, err error) {
		subFromTime := fromTime.Add(time.Duration(step) * time.Hour)
		subToTime := fromTime.Add(time.Duration(step+1) * time.Hour)
		if subToTime.Sub(toTime) > 0 {
			subToTime = toTime
		}
		if step < hours-1 {
			subToTime = subToTime.Add(-1 * time.Second) // 减一秒
		}
		commonParams := map[string]interface{}{
			"pageSize":                pageSize,
			"orderPurchaseTime_begin": utils.Time2Str(subFromTime),
			"orderPurchaseTime_end":   utils.Time2Str(subToTime),
		}
		orderList, totalCount, err := api.JdAPI.OrderQuery(utils.MergeMaps(commonParams, utils.Params2Map("pageNo", 1)))
		if err != nil {
			return "", err
		}
		if err = saveJdOrderList(existJdIDMap, orderList); err != nil {
			return "", err
		}
		leftPageCount := (totalCount - 1) / pageSize
		globals.SugarLogger.Debugf("subFromTime:%s, subToTime:%s, leftPageCount:%d, totalCount:%d", utils.Time2Str(subFromTime), utils.Time2Str(subToTime), leftPageCount, totalCount)
		if leftPageCount > 0 {
			pageNoList := make([]int, leftPageCount)
			for k := range pageNoList {
				pageNoList[k] = k + 2
			}
			subTask := tasksch.NewParallelTask("PullJdOrder sub", tasksch.NewParallelConfig().SetIsContinueWhenError(isContinueWhenError).SetParallelCount(10), ctx.GetUserName(), func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
				subOrderList, subTotalCount, err := api.JdAPI.OrderQuery(utils.MergeMaps(commonParams, utils.Params2Map("pageNo", batchItemList[0])))
				if err == nil {
					if totalCount != subTotalCount {
						globals.SugarLogger.Warnf("PullJdOrder totalCount:%d, subTotalCount:%d are not same", totalCount, subTotalCount)
					}
					err = saveJdOrderList(existJdIDMap, subOrderList)
				}
				return nil, err
			}, pageNoList)
			task.AddChild(subTask).Run()
			_, err = subTask.GetResult(0)
		}
		return nil, err
	}, hours)
	tasksch.ManageTask(rootTask).Run()

	if !isAsync {
		_, err = rootTask.GetResult(0)
	} else {
		hint = rootTask.ID
	}
	return hint, err
}