jx-callback/business/partner/purchase/jd/store_sku.go

package jd

import (
	"git.rosy.net.cn/baseapi/platformapi/jdapi"
	"git.rosy.net.cn/baseapi/utils"
	"git.rosy.net.cn/jx-callback/business/jxutils"
	"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/globals"
	"git.rosy.net.cn/jx-callback/globals/api"
)

const (
	MaxStockQty     = 100000000
	MaxSkuBatchSize = 50
)

type tStoreSkuBindExt struct {
	model.StoreSkuBind
	PricePercentage int
}

// 京东到家，以有库存表示关注（认领）
func (p *PurchaseHandler) SyncStoresSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, db *dao.DaoDB, storeIDs []int, skuIDs []int, isAsync bool) (hint string, err error) {
	globals.SugarLogger.Debugf("jd SyncStoresSkus, storeIDs:%v, skuIDs:%v", storeIDs, skuIDs)

	parallelCount := 1
	if len(skuIDs) < MaxSkuBatchSize {
		parallelCount = 10
	}
	task := tasksch.NewParallelTask("SyncStoresSkus", tasksch.NewParallelConfig().SetParallelCount(parallelCount), ctx.GetUserName(), func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (interface{}, error) {
		storeID := batchItemList[0].(int)
		sqlWhere := `
		WHERE (t1.jd_sync_status <> 0) AND t1.store_id = ?
		`
		sqlWhereParams := []interface{}{
			storeID,
		}

		if len(skuIDs) > 0 {
			sqlWhere += " AND t1.sku_id IN (" + dao.GenQuestionMarks(len(skuIDs)) + ")"
			sqlWhereParams = append(sqlWhereParams, skuIDs)
		}

		sql := `
		SELECT t1.*, t2.price_percentage
		FROM store_sku_bind t1
		JOIN store_map t2 ON t1.store_id = t2.store_id AND t2.vendor_id = ? AND t2.deleted_at = ?
		` + sqlWhere
		var storeSkus []*tStoreSkuBindExt
		sqlParams := []interface{}{
			model.VendorIDJD,
			utils.DefaultTimeValue,
		}
		if err = dao.GetRows(db, &storeSkus, sql, append(sqlParams, sqlWhereParams...)...); err == nil {
			outStationNo := utils.Int2Str(storeID)
			task := tasksch.NewParallelTask("SyncStoresSkus inner", tasksch.NewParallelConfig().SetBatchSize(MaxSkuBatchSize), ctx.GetUserName(), func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (interface{}, error) {
				var skuPriceInfoList []*jdapi.SkuPriceInfo
				var skuVendibilityList []*jdapi.StockVendibility
				var skuStockList []*jdapi.SkuStock

				for _, v := range batchItemList {
					storeSku := v.(*tStoreSkuBindExt)
					if storeSku.JdSyncStatus&model.SyncFlagChangedMask != 0 {
						if storeSku.JdSyncStatus&(model.SyncFlagPriceMask|model.SyncFlagNewMask) != 0 {
							skuPriceInfoList = append(skuPriceInfoList, &jdapi.SkuPriceInfo{
								OutSkuId: utils.Int2Str(storeSku.SkuID),
								Price:    jxutils.CaculateSkuVendorPrice(storeSku.Price, storeSku.PricePercentage),
							})
						}
						if storeSku.JdSyncStatus&(model.SyncFlagSaleMask|model.SyncFlagNewMask) != 0 {
							vendibility := &jdapi.StockVendibility{
								OutSkuId: utils.Int2Str(storeSku.SkuID),
								DoSale:   true,
							}
							if storeSku.Status != model.StoreSkuBindStatusNormal {
								vendibility.DoSale = false
							}
							skuVendibilityList = append(skuVendibilityList, vendibility)
						}
						if storeSku.JdSyncStatus&(model.SyncFlagDeletedMask|model.SyncFlagNewMask) != 0 { // 关注或取消关注
							stock := &jdapi.SkuStock{
								OutSkuId: utils.Int2Str(storeSku.SkuID),
								StockQty: MaxStockQty,
							}
							if storeSku.DeletedAt != utils.DefaultTimeValue {
								stock.StockQty = 0
							}
							skuStockList = append(skuStockList, stock)
						}
					}
				}
				if globals.EnableStoreWrite {
					// todo 以下可以优化为并行操作
					globals.SugarLogger.Debug(utils.Format4Output(skuVendibilityList, false), utils.Format4Output(skuPriceInfoList, false), utils.Format4Output(skuStockList, false))
					if len(skuVendibilityList) > 0 {
						_, err = api.JdAPI.BatchUpdateVendibility(outStationNo, "", skuVendibilityList, ctx.GetUserName())
					}
					if err == nil && len(skuPriceInfoList) > 0 {
						_, err = api.JdAPI.UpdateVendorStationPrice(outStationNo, "", skuPriceInfoList)
					}
					if err == nil && len(skuStockList) > 0 {
						_, err = api.JdAPI.BatchUpdateCurrentQtys(outStationNo, "", skuStockList, ctx.GetUserName())
					}
				}
				return nil, err
			}, storeSkus)
			t.AddChild(task).Run()
			if _, err = task.GetResult(0); err == nil {
				sql := `
				UPDATE store_sku_bind t1
				SET t1.jd_sync_status = 0
				` + sqlWhere
				_, err = dao.ExecuteSQL(db, sql, sqlWhereParams...)
			}
		}
		return nil, err
	}, storeIDs)

	ctx.SetTaskOrAddChild(task, parentTask)
	task.Run()
	if isAsync {
		return task.ID, nil
	}
	_, err = task.GetResult(0)
	return "", err
}