liulei/jx-callback
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

- 新同步逻辑调整，京东还有问题

Browse Source
This commit is contained in:
gazebo
2019-07-18 09:11:31 +08:00
parent b5b3d82130
commit 2c2659b0ed
10 changed files with 98 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
business/jxstore/cms/store.go
View File

@@ -794,7 +794,7 @@ func UpdateStoreVendorMap(ctx *jxcontext.Context, db *dao.DaoDB, storeID, vendor
valid := dao.StrictMakeMapByStructObject(payload, storeMap, userName)
if valid["pricePercentagePack"] != nil {
if pricePercentagePack := utils.Interface2String(valid["pricePercentagePack"]); pricePercentagePack != "" {
_, err2 := dao.QueryConfigs(db, model.ConfigTypePricePack, pricePercentagePack, "")
_, err2 := dao.QueryConfigs(db, pricePercentagePack, model.ConfigTypePricePack, "")
if err = err2; err != nil {
return 0, err
}

81
business/jxstore/cms/sync_store_sku.go
View File

@@ -101,7 +101,7 @@ func SyncStorCategories(ctx *jxcontext.Context, parentTask tasksch.ITask, vendor
return hint, err
}
func SyncStoreSkuNew(ctx *jxcontext.Context, vendorID, storeID int, vendorStoreID string, nameIDs, skuIDs []int, isAsync, isContinueWhenError bool) (hint string, err error) {
func SyncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, vendorID, storeID int, vendorStoreID string, nameIDs, skuIDs []int, isAsync, isContinueWhenError bool) (hint string, err error) {
singleStoreHandler, _ := partner.GetPurchasePlatformFromVendorID(vendorID).(partner.ISingleStoreStoreSkuHandler)
if singleStoreHandler != nil {
if err = CreateStoreCategoryByStoreSku(ctx, vendorID, storeID, vendorStoreID, nameIDs, skuIDs); err != nil {
@@ -116,11 +116,11 @@ func SyncStoreSkuNew(ctx *jxcontext.Context, vendorID, storeID int, vendorStoreI
_, err = SyncStorCategories(ctx, task, vendorID, storeID, vendorStoreID, nameIDs, skuIDs, false, isContinueWhenError)
}
case 1:
err = syncStoreSkuNew(ctx, task, vendorID, storeID, nameIDs, skuIDs, isContinueWhenError)
err = syncStoreSkuNew(ctx, task, false, vendorID, storeID, nameIDs, skuIDs, isContinueWhenError)
}
return result, err
}, 2)
tasksch.HandleTask(task, nil, true).Run()
tasksch.HandleTask(task, parentTask, true).Run()
if !isAsync {
_, err = task.GetResult(0)
} else {
@@ -129,19 +129,28 @@ func SyncStoreSkuNew(ctx *jxcontext.Context, vendorID, storeID int, vendorStoreI
return hint, err
}
func FullSyncStoreSkuNew(ctx *jxcontext.Context, vendorID, storeID int, vendorStoreID string, isAsync, isContinueWhenError bool) (hint string, err error) {
func FullSyncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, vendorID, storeID int, vendorStoreID string, isAsync, isContinueWhenError bool) (hint string, err error) {
singleStoreHandler, _ := partner.GetPurchasePlatformFromVendorID(vendorID).(partner.ISingleStoreStoreSkuHandler)
task := tasksch.NewParallelTask("FullSyncStoreSkuNew", tasksch.NewParallelConfig().SetParallelCount(1).SetIsContinueWhenError(isContinueWhenError), ctx,
func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
step := batchItemList[0].(int)
switch step {
case 0:
_, err = ClearRemoteStoreStuffAndSetNew(ctx, task, vendorID, storeID, vendorStoreID, false, isContinueWhenError)
if singleStoreHandler != nil {
_, err = ClearRemoteStoreStuffAndSetNew(ctx, task, vendorID, storeID, vendorStoreID, false, isContinueWhenError)
} else {
_, err = dao.SetStoreSkuSyncStatus(dao.GetDB(), vendorID, []int{storeID}, nil, model.SyncFlagStoreSkuOnlyMask)
}
case 1:
_, err = SyncStoreSkuNew(ctx, vendorID, storeID, vendorStoreID, nil, nil, false, isContinueWhenError)
if singleStoreHandler != nil {
_, err = SyncStoreSkuNew(ctx, task, vendorID, storeID, vendorStoreID, nil, nil, false, isContinueWhenError)
} else {
err = syncStoreSkuNew(ctx, task, false, vendorID, storeID, nil, nil, isContinueWhenError)
}
}
return retVal, err
}, []int{0, 1})
tasksch.HandleTask(task, nil, true).Run()
tasksch.HandleTask(task, parentTask, true).Run()
if !isAsync {
_, err = task.GetResult(0)
} else {
@@ -150,6 +159,10 @@ func FullSyncStoreSkuNew(ctx *jxcontext.Context, vendorID, storeID int, vendorSt
return hint, err
}
func isStoreSkuSyncNeedDelete(storeSku *dao.StoreSkuSyncInfo) bool {
return model.IsSyncStatusDelete(storeSku.StoreSkuSyncStatus) || storeSku.DeletedAt != utils.DefaultTimeValue || storeSku.BindID == 0 || storeSku.NameID == 0
}
func storeSkuSyncInfo2Bare(inSku *dao.StoreSkuSyncInfo) (outSku *partner.StoreSkuInfo) {
outSku = &partner.StoreSkuInfo{
SkuID: inSku.SkuID,
@@ -157,10 +170,10 @@ func storeSkuSyncInfo2Bare(inSku *dao.StoreSkuSyncInfo) (outSku *partner.StoreSk
NameID: inSku.NameID,
VendorNameID: inSku.VendorNameID,
Status: jxutils.MergeSkuStatus(inSku.Status, inSku.StoreSkuStatus),
Price: inSku.VendorPrice,
Status: inSku.MergedStatus,
VendorPrice: inSku.VendorPrice,
}
if !model.IsSyncStatusDelete(inSku.StoreSkuSyncStatus) {
if !isStoreSkuSyncNeedDelete(inSku) {
outSku.Stock = model.MaxStoreSkuStockQty
}
return outSku
@@ -172,6 +185,14 @@ func calVendorPrice4StoreSku(inSku *dao.StoreSkuSyncInfo, pricePercentagePack mo
return inSku
}
func formalizeStoreSkuList(inSkuList []*dao.StoreSkuSyncInfo) []*dao.StoreSkuSyncInfo {
for _, skuItem := range inSkuList {
skuItem.MergedStatus = jxutils.MergeSkuStatus(skuItem.Status, skuItem.StoreSkuStatus)
skuItem.SkuName = jxutils.ComposeSkuName(skuItem.Prefix, skuItem.Name, skuItem.Comment, skuItem.Unit, skuItem.SpecQuality, skuItem.SpecUnit, 0)
}
return inSkuList
}
func sku2Update(vendorID int, sku *dao.StoreSkuSyncInfo, syncStatus int8) (item *dao.KVUpdateItem) {
if syncStatus&(model.SyncFlagDeletedMask|model.SyncFlagNewMask|model.SyncFlagModifiedMask) != 0 {
sku.StoreSkuSyncStatus = 0
@@ -204,7 +225,7 @@ func updateStoreSku(db *dao.DaoDB, vendorID int, storeSkuList []*dao.StoreSkuSyn
return num, err
}
func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, vendorID, storeID int, nameIDs, skuIDs []int, isContinueWhenError bool) (err error) {
func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, isFull bool, vendorID, storeID int, nameIDs, skuIDs []int, isContinueWhenError bool) (err error) {
db := dao.GetDB()
storeDetail, err := dao.GetStoreDetail(db, storeID, vendorID)
if err != nil {
@@ -212,10 +233,17 @@ func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, vendorID,
}
vendorStoreID := storeDetail.VendorStoreID
skus, err := dao.GetStoreSkus(db, vendorID, storeID, skuIDs)
var skus []*dao.StoreSkuSyncInfo
if isFull {
skus, err = dao.GetFullStoreSkus(db, vendorID, storeID)
} else {
skus, err = dao.GetStoreSkus(db, vendorID, storeID, skuIDs)
}
if err != nil {
return err
}
formalizeStoreSkuList(skus)
singleStoreHandler, _ := partner.GetPurchasePlatformFromVendorID(vendorID).(partner.ISingleStoreStoreSkuHandler)
storeSkuHandler := partner.GetPurchasePlatformFromVendorID(vendorID).(partner.IPurchasePlatformStoreSkuHandler)
@@ -228,7 +256,7 @@ func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, vendorID,
for _, sku := range skus {
var bareSku *partner.StoreSkuInfo
if model.IsSyncStatusDelete(sku.StoreSkuSyncStatus) {
if isStoreSkuSyncNeedDelete(sku) {
if !dao.IsVendorThingIDEmpty(sku.VendorSkuID) {
bareSku = storeSkuSyncInfo2Bare(sku)
if singleStoreHandler == nil {
@@ -241,11 +269,13 @@ func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, vendorID,
}
} else if model.IsSyncStatusNew(sku.StoreSkuSyncStatus) {
calVendorPrice4StoreSku(sku, storeDetail.PricePercentagePackObj, int(storeDetail.PricePercentage))
bareSku = storeSkuSyncInfo2Bare(sku)
if singleStoreHandler == nil {
bareSku = storeSkuSyncInfo2Bare(sku)
stockList = append(stockList, bareSku)
} else {
createList = append(createList, sku)
if bareSku.Status == model.SkuStatusNormal && !dao.IsVendorThingIDEmpty(sku.VendorCatID) {
createList = append(createList, sku)
}
}
} else {
if model.IsSyncStatusUpdate(sku.StoreSkuSyncStatus) {
@@ -281,16 +311,16 @@ func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, vendorID,
}
return skuList
}
task := tasksch.NewParallelTask("syncStoreSkuNew", tasksch.NewParallelConfig().SetParallelCount(1).SetIsContinueWhenError(isContinueWhenError), ctx,
func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
step := batchItemList[0].(int)
globals.SugarLogger.Debugf("step:%d", step)
switch step {
case 0:
if len(deleteList) > 0 {
_, err = putils.FreeBatchStoreSkuInfo(func(batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, err error) {
if err = singleStoreHandler.DeleteStoreSkus(ctx, storeID, vendorStoreID, batchedStoreSkuList); err == nil {
_, err = updateStoreSku(db, vendorID, bareSku2Sync(batchedStoreSkuList), model.SyncFlagDeletedMask)
_, err = updateStoreSku(dao.GetDB(), vendorID, bareSku2Sync(batchedStoreSkuList), model.SyncFlagDeletedMask)
}
return nil, err
}, ctx, task, deleteList, singleStoreHandler.GetStoreSkusBatchSize(partner.FuncDeleteStoreSkus), isContinueWhenError)
@@ -298,8 +328,9 @@ func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, vendorID,
case 1:
if len(createList) > 0 {
_, err = putils.FreeBatchStoreSkuSyncInfo(func(batchedStoreSkuList []*dao.StoreSkuSyncInfo) (result interface{}, err error) {
globals.SugarLogger.Debug(utils.Format4Output(batchedStoreSkuList, false))
if err = singleStoreHandler.CreateStoreSkus(ctx, storeID, vendorStoreID, batchedStoreSkuList); err == nil {
_, err = updateStoreSku(db, vendorID, batchedStoreSkuList, model.SyncFlagNewMask)
_, err = updateStoreSku(dao.GetDB(), vendorID, batchedStoreSkuList, model.SyncFlagNewMask)
}
return nil, err
}, ctx, task, createList, singleStoreHandler.GetStoreSkusBatchSize(partner.FuncCreateStoreSkus), isContinueWhenError)
@@ -307,8 +338,8 @@ func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, vendorID,
case 2:
if len(updateList) > 0 {
_, err = putils.FreeBatchStoreSkuSyncInfo(func(batchedStoreSkuList []*dao.StoreSkuSyncInfo) (result interface{}, err error) {
if err = singleStoreHandler.CreateStoreSkus(ctx, storeID, vendorStoreID, batchedStoreSkuList); err == nil {
_, err = updateStoreSku(db, vendorID, batchedStoreSkuList, model.SyncFlagModifiedMask)
if err = singleStoreHandler.UpdateStoreSkus(ctx, storeID, vendorStoreID, batchedStoreSkuList); err == nil {
_, err = updateStoreSku(dao.GetDB(), vendorID, batchedStoreSkuList, model.SyncFlagModifiedMask)
}
return nil, err
}, ctx, task, updateList, singleStoreHandler.GetStoreSkusBatchSize(partner.FuncUpdateStoreSkus), isContinueWhenError)
@@ -317,7 +348,7 @@ func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, vendorID,
if len(stockList) > 0 {
_, err = putils.FreeBatchStoreSkuInfo(func(batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, err error) {
if err = storeSkuHandler.UpdateStoreSkusStock(ctx, storeID, vendorStoreID, stockList); err == nil {
_, err = updateStoreSku(db, vendorID, bareSku2Sync(batchedStoreSkuList), model.SyncFlagModifiedMask) // ?
_, err = updateStoreSku(dao.GetDB(), vendorID, bareSku2Sync(batchedStoreSkuList), model.SyncFlagModifiedMask) // ?
}
return nil, err
}, ctx, task, stockList, storeSkuHandler.GetStoreSkusBatchSize(partner.FuncUpdateStoreSkusStock), isContinueWhenError)
@@ -325,8 +356,8 @@ func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, vendorID,
case 4:
if len(statusList) > 0 {
_, err = putils.FreeBatchStoreSkuInfo(func(batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, err error) {
if err = storeSkuHandler.UpdateStoreSkusStock(ctx, storeID, vendorStoreID, statusList); err == nil {
_, err = updateStoreSku(db, vendorID, bareSku2Sync(batchedStoreSkuList), model.SyncFlagSaleMask)
if err = storeSkuHandler.UpdateStoreSkusStatus(ctx, storeID, vendorStoreID, statusList); err == nil {
_, err = updateStoreSku(dao.GetDB(), vendorID, bareSku2Sync(batchedStoreSkuList), model.SyncFlagSaleMask)
}
return nil, err
}, ctx, task, statusList, storeSkuHandler.GetStoreSkusBatchSize(partner.FuncUpdateStoreSkusStatus), isContinueWhenError)
@@ -334,8 +365,8 @@ func syncStoreSkuNew(ctx *jxcontext.Context, parentTask tasksch.ITask, vendorID,
case 5:
if len(priceList) > 0 {
_, err = putils.FreeBatchStoreSkuInfo(func(batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, err error) {
if err = storeSkuHandler.UpdateStoreSkusStock(ctx, storeID, vendorStoreID, priceList); err == nil {
_, err = updateStoreSku(db, vendorID, bareSku2Sync(batchedStoreSkuList), model.SyncFlagPriceMask)
if err = storeSkuHandler.UpdateStoreSkusPrice(ctx, storeID, vendorStoreID, priceList); err == nil {
_, err = updateStoreSku(dao.GetDB(), vendorID, bareSku2Sync(batchedStoreSkuList), model.SyncFlagPriceMask)
}
return nil, err
}, ctx, task, priceList, storeSkuHandler.GetStoreSkusBatchSize(partner.FuncUpdateStoreSkusPrice), isContinueWhenError)