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

- refactor SyncStoreSkus, LoopStoresMap

Browse Source
This commit is contained in:
gazebo
2018-10-26 11:23:10 +08:00
parent c07fb48d02
commit 5f35965fee
5 changed files with 230 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files

194
business/jxstore/cms/sync.go
View File

@@ -16,6 +16,11 @@ import (
"git.rosy.net.cn/jx-callback/globals/api" "git.rosy.net.cn/jx-callback/globals/api"
) )
type LoopStoreMapInfo struct {
VendorID int
StoreMapList []*model.StoreMap
}
type VendorSync struct { type VendorSync struct {
MultiStoreVendorIDs []int MultiStoreVendorIDs []int
SingleStoreVendorIDs []int SingleStoreVendorIDs []int
@@ -191,19 +196,37 @@ func (v *VendorSync) SyncReorderCategories(ctx *jxcontext.Context, db *dao.DaoDB
func (v *VendorSync) SyncStore(ctx *jxcontext.Context, db *dao.DaoDB, vendorID, storeID int, isAsync bool, userName string) (hint string, err error) { func (v *VendorSync) SyncStore(ctx *jxcontext.Context, db *dao.DaoDB, vendorID, storeID int, isAsync bool, userName string) (hint string, err error) {
globals.SugarLogger.Debugf("SyncStore, storeID:%d", storeID) globals.SugarLogger.Debugf("SyncStore, storeID:%d", storeID)
hint, err = v.LoopStoreMap(ctx, db, "SyncStore", isAsync, userName, storeID, func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (interface{}, error) {
storeMap := batchItemList[0].(*model.StoreMap) var vendorIDs []int
if (vendorID == -1 || vendorID == storeMap.VendorID) && (storeMap.SyncStatus != 0) { if vendorID != -1 {
if handler := v.GetStoreHandler(storeMap.VendorID); handler != nil { vendorIDs = []int{
if err = handler.UpdateStore(db, storeID, userName); err == nil { vendorID,
}
}
hint, err = v.LoopStoresMap(ctx, db, "SyncStore", isAsync, vendorIDs, []int{storeID}, func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (interface{}, error) {
loopMapInfo := batchItemList[0].(*LoopStoreMapInfo)
handler := v.GetStoreHandler(loopMapInfo.VendorID)
if len(loopMapInfo.StoreMapList) > 1 {
loopStoreTask := tasksch.NewParallelTask("SyncStore loop stores", nil, ctx.GetUserName(), func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
storeMap := batchItemList[0].(*model.StoreMap)
if err = handler.UpdateStore(db, storeMap.StoreID, userName); err == nil {
storeMap.SyncStatus = 0 storeMap.SyncStatus = 0
_, err = dao.UpdateEntity(db, storeMap, model.FieldSyncStatus) _, err = dao.UpdateEntity(db, storeMap, model.FieldSyncStatus)
} }
} return nil, err
}, loopMapInfo.StoreMapList)
t.AddChild(loopStoreTask).Run()
_, err = loopStoreTask.GetResult(0)
return nil, err
}
storeMap := loopMapInfo.StoreMapList[0]
if err = handler.UpdateStore(db, storeMap.StoreID, userName); err == nil {
storeMap.SyncStatus = 0
_, err = dao.UpdateEntity(db, storeMap, model.FieldSyncStatus)
} }
return nil, err return nil, err
}) })
return "", err return hint, err
} }
func (v *VendorSync) SyncSku(ctx *jxcontext.Context, db *dao.DaoDB, nameID, skuID int, isAsync bool, userName string) (hint string, err error) { func (v *VendorSync) SyncSku(ctx *jxcontext.Context, db *dao.DaoDB, nameID, skuID int, isAsync bool, userName string) (hint string, err error) {
@@ -253,32 +276,75 @@ func (v *VendorSync) SyncSku(ctx *jxcontext.Context, db *dao.DaoDB, nameID, skuI
// //
func (v *VendorSync) SyncStoresSkus(ctx *jxcontext.Context, db *dao.DaoDB, vendorIDs []int, storeIDs []int, skuIDs []int, isAsync bool, userName string) (hint string, err error) { func (v *VendorSync) SyncStoresSkus(ctx *jxcontext.Context, db *dao.DaoDB, vendorIDs []int, storeIDs []int, skuIDs []int, isAsync bool, userName string) (hint string, err error) {
globals.SugarLogger.Debug("SyncStoresSkus") globals.SugarLogger.Debug("SyncStoresSkus")
hint, err = v.LoopStoreVendors(ctx, db, vendorIDs, "SyncStoresSkus", isAsync, userName, func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (interface{}, error) { hint, err = v.LoopStoresMap(ctx, db, "SyncStoresSkus", isAsync, vendorIDs, storeIDs, func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (interface{}, error) {
handler := v.GetStoreHandler(batchItemList[0].(int)) loopMapInfo := batchItemList[0].(*LoopStoreMapInfo)
_, err = handler.SyncStoresSkus(ctx, t, db, storeIDs, skuIDs, false) handler := v.GetStoreHandler(loopMapInfo.VendorID)
if len(loopMapInfo.StoreMapList) > 1 {
loopStoreTask := tasksch.NewSeqTask("SyncStoresSkus loop stores", ctx.GetUserName(), func(task *tasksch.SeqTask, step int, params ...interface{}) (result interface{}, err error) {
storeID := loopMapInfo.StoreMapList[step].StoreID
_, err = handler.SyncStoreSkus(ctx, task, storeID, skuIDs, false)
return nil, err
}, len(loopMapInfo.StoreMapList))
t.AddChild(loopStoreTask).Run()
_, err = loopStoreTask.GetResult(0)
return nil, err
}
_, err = handler.SyncStoreSkus(ctx, t, loopMapInfo.StoreMapList[0].StoreID, skuIDs, false)
return nil, err return nil, err
}) })
return hint, err return hint, err
} }
func (v *VendorSync) LoopStoreMap(ctx *jxcontext.Context, db *dao.DaoDB, taskName string, isAsync bool, userName string, storeID int, handler tasksch.WorkFunc) (hint string, err error) { func (v *VendorSync) LoopStoresMap(ctx *jxcontext.Context, db *dao.DaoDB, taskName string, isAsync bool, vendorIDs []int, storeIDs []int, handler tasksch.WorkFunc) (hint string, err error) {
storeMaps, err := GetStoreVendorMaps(ctx, db, storeID, -1) sql := `
if err == nil { SELECT t1.*
task := tasksch.NewParallelTask(taskName, nil, userName, handler, storeMaps) FROM store_map t1
ctx.SetTaskOrAddChild(task, nil) WHERE t1.deleted_at = ?
tasksch.ManageTask(task).Run() `
hint = task.ID sqlParams := []interface{}{
if !isAsync { utils.DefaultTimeValue,
_, err = task.GetResult(0) }
if len(vendorIDs) > 0 {
sql += " AND t1.vendor_id IN (" + dao.GenQuestionMarks(len(vendorIDs)) + ")"
sqlParams = append(sqlParams, vendorIDs)
}
if len(storeIDs) > 0 {
sql += " AND t1.store_id IN (" + dao.GenQuestionMarks(len(storeIDs)) + ")"
sqlParams = append(sqlParams, storeIDs)
}
var storeMapList []*model.StoreMap
if err = dao.GetRows(db, &storeMapList, sql, sqlParams...); err != nil {
return "", err
}
if len(storeMapList) == 0 {
return "", nil
}
vendorStoreMap := make(map[int][]*model.StoreMap)
for _, v := range storeMapList {
vendorStoreMap[v.VendorID] = append(vendorStoreMap[v.VendorID], v)
}
loopInfoList := make([]*LoopStoreMapInfo, len(vendorStoreMap))
index := 0
for k, v := range vendorStoreMap {
loopInfoList[index] = &LoopStoreMapInfo{
VendorID: k,
StoreMapList: v,
} }
index++
}
task := tasksch.NewParallelTask(taskName, nil, ctx.GetUserName(), handler, loopInfoList)
ctx.SetTaskOrAddChild(task, nil)
tasksch.ManageTask(task).Run()
hint = task.ID
if !isAsync {
_, err = task.GetResult(0)
} }
return hint, makeSyncError(err) return hint, makeSyncError(err)
} }
func (v *VendorSync) LoopMultiStoresVendors(ctx *jxcontext.Context, db *dao.DaoDB, taskName string, isAsync bool, userName string, handler tasksch.WorkFunc) (hint string, err error) { func (v *VendorSync) LoopMultiStoresVendors(ctx *jxcontext.Context, db *dao.DaoDB, taskName string, isAsync bool, userName string, handler tasksch.WorkFunc) (hint string, err error) {
if taskName == "" {
taskName = "LoopMultiStoresVendors"
}
task := tasksch.NewParallelTask(taskName, nil, userName, handler, v.MultiStoreVendorIDs) task := tasksch.NewParallelTask(taskName, nil, userName, handler, v.MultiStoreVendorIDs)
ctx.SetTaskOrAddChild(task, nil) ctx.SetTaskOrAddChild(task, nil)
tasksch.ManageTask(task).Run() tasksch.ManageTask(task).Run()
@@ -288,58 +354,38 @@ func (v *VendorSync) LoopMultiStoresVendors(ctx *jxcontext.Context, db *dao.DaoD
return task.ID, makeSyncError(err) return task.ID, makeSyncError(err)
} }
func (v *VendorSync) LoopStoreVendors(ctx *jxcontext.Context, db *dao.DaoDB, vendorIDs []int, taskName string, isAsync bool, userName string, handler tasksch.WorkFunc) (hint string, err error) { // func (v *VendorSync) LoopStoreVendors(ctx *jxcontext.Context, db *dao.DaoDB, vendorIDs []int, taskName string, isAsync bool, userName string, handler tasksch.WorkFunc) (hint string, err error) {
if taskName == "" { // if taskName == "" {
taskName = "LoopStoreVendors" // taskName = "LoopStoreVendors"
} // }
var vendorIDMap map[int]int // var vendorIDMap map[int]int
if len(vendorIDs) != 0 { // if len(vendorIDs) != 0 {
vendorIDMap = make(map[int]int) // vendorIDMap = make(map[int]int)
for _, v := range vendorIDs { // for _, v := range vendorIDs {
vendorIDMap[v] = 1 // vendorIDMap[v] = 1
} // }
} // }
allHandlers := make([]int, len(v.MultiStoreVendorIDs)+len(v.SingleStoreVendorIDs)) // allHandlers := make([]int, len(v.MultiStoreVendorIDs)+len(v.SingleStoreVendorIDs))
copy(allHandlers, v.MultiStoreVendorIDs) // copy(allHandlers, v.MultiStoreVendorIDs)
copy(allHandlers[len(v.MultiStoreVendorIDs):], v.SingleStoreVendorIDs) // copy(allHandlers[len(v.MultiStoreVendorIDs):], v.SingleStoreVendorIDs)
if vendorIDMap != nil { // if vendorIDMap != nil {
count := 0 // count := 0
for _, v := range allHandlers { // for _, v := range allHandlers {
if vendorIDMap[v] == 1 { // if vendorIDMap[v] == 1 {
allHandlers[count] = v // allHandlers[count] = v
count++ // count++
} // }
} // }
allHandlers = allHandlers[:count] // allHandlers = allHandlers[:count]
} // }
task := tasksch.NewParallelTask(taskName, nil, userName, handler, allHandlers) // task := tasksch.NewParallelTask(taskName, nil, userName, handler, allHandlers)
ctx.SetTaskOrAddChild(task, nil) // ctx.SetTaskOrAddChild(task, nil)
tasksch.ManageTask(task).Run() // tasksch.ManageTask(task).Run()
if !isAsync { // if !isAsync {
_, err = task.GetResult(0) // _, err = task.GetResult(0)
} // }
return task.ID, err // return task.ID, err
} // }
func (v *VendorSync) LoopSingleStoreVendors(ctx *jxcontext.Context, db *dao.DaoDB, taskName string, isAsync bool, userName string, handler tasksch.WorkFunc) (hint string, err error) {
if taskName == "" {
taskName = "LoopSingleStoreVendors"
}
var storeMaps []*model.StoreMap
if err = dao.GetRows(db, &storeMaps, `
SELECT *
FROM store_map
WHERE deleted_at = ? AND vendor_id IN (`+dao.GenQuestionMarks(len(v.SingleStoreVendorIDs))+")", utils.DefaultTimeValue, v.SingleStoreVendorIDs); err == nil {
task := tasksch.NewParallelTask(taskName, nil, userName, handler, storeMaps)
ctx.SetTaskOrAddChild(task, nil)
tasksch.ManageTask(task).Run()
hint = task.ID
if !isAsync {
_, err = task.GetResult(0)
}
}
return hint, makeSyncError(err)
}
func (v *VendorSync) RefreshAllSkusID(ctx *jxcontext.Context, isAsync bool, vendorIDs []int, storeIDs []int) (hint string, err error) { func (v *VendorSync) RefreshAllSkusID(ctx *jxcontext.Context, isAsync bool, vendorIDs []int, storeIDs []int) (hint string, err error) {
task := tasksch.NewParallelTask("RefreshAllSkusID", nil, ctx.GetUserName(), func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) { task := tasksch.NewParallelTask("RefreshAllSkusID", nil, ctx.GetUserName(), func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {

2
business/partner/partner.go
View File

@@ -85,7 +85,7 @@ type IPurchasePlatformHandler interface {
// OpenStore(vendorStoreID string, userName string) error // OpenStore(vendorStoreID string, userName string) error
// CloseStore(vendorStoreID, closeNotice, userName string) error // CloseStore(vendorStoreID, closeNotice, userName string) error
SyncStoresSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, db *dao.DaoDB, storeIDs []int, skuIDs []int, isAsync bool) (hint string, err error) SyncStoreSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, storeID int, skuIDs []int, isAsync bool) (hint string, err error)
RefreshAllStoresID(ctx *jxcontext.Context, parentTask tasksch.ITask, isAsync bool) (hint string, err error) RefreshAllStoresID(ctx *jxcontext.Context, parentTask tasksch.ITask, isAsync bool) (hint string, err error)
GetVendorID() int GetVendorID() int

60
business/partner/purchase/ebai/store_sku.go
View File

@@ -14,7 +14,6 @@ import (
"git.rosy.net.cn/jx-callback/business/model/dao" "git.rosy.net.cn/jx-callback/business/model/dao"
"git.rosy.net.cn/jx-callback/globals" "git.rosy.net.cn/jx-callback/globals"
"git.rosy.net.cn/jx-callback/globals/api" "git.rosy.net.cn/jx-callback/globals/api"
"github.com/astaxie/beego/orm"
) )
const ( const (
@@ -73,33 +72,13 @@ var (
} }
) )
func (p *PurchaseHandler) SyncStoresSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, db *dao.DaoDB, storeIDs []int, skuIDs []int, isAsync bool) (hint string, err error) { func (p *PurchaseHandler) SyncStoreSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, storeID int, skuIDs []int, isAsync bool) (hint string, err error) {
task := tasksch.NewSeqTask("ebai.SyncStoresSkus", ctx.GetUserName(), func(t *tasksch.SeqTask, step int, params ...interface{}) (result interface{}, err error) { userName := ctx.GetUserName()
storeID := storeIDs[step] globals.SugarLogger.Debugf("SyncStoreSkus storeID:%d, skuIDs:%v, userName:%s", storeID, skuIDs, userName)
err = p.syncOneStoreSkus(t, db, storeID, skuIDs, false, ctx.GetUserName()) db := dao.GetDB()
return nil, err
}, len(storeIDs))
ctx.SetTaskOrAddChild(task, parentTask)
task.Run()
if !isAsync {
_, err = task.GetResult(0)
}
return task.ID, err
}
func (p *PurchaseHandler) syncOneStoreSkus(parentTask tasksch.ITask, db *dao.DaoDB, storeID int, skuIDs []int, isAsync bool, userName string) (err error) {
globals.SugarLogger.Debugf("syncOneStoreSkus storeID:%d, skuIDs:%v, userName:%s", storeID, skuIDs, userName)
_, err2 := dao.GetStoreMapByStoreID(db, storeID, model.VendorIDEBAI)
if err = err2; err != nil {
if err == orm.ErrNoRows {
err = nil
}
return err
}
if err = p.syncOneStoreCategoriesFromRemote2Local(db, storeID, userName); err != nil { if err = p.syncOneStoreCategoriesFromRemote2Local(db, storeID, userName); err != nil {
return err return "", err
} }
sql := ` sql := `
SELECT t1.*, t2.spec_quality, t2.spec_unit, t2.weight, t2.status sku_status, SELECT t1.*, t2.spec_quality, t2.spec_unit, t2.weight, t2.status sku_status,
@@ -109,24 +88,28 @@ func (p *PurchaseHandler) syncOneStoreSkus(parentTask tasksch.ITask, db *dao.Dao
t4p.id parent_cat_id, t5p.ebai_id parent_cat_ebai_id, t5p.ebai_sync_status parent_cat_ebai_sync_status, t4p.id parent_cat_id, t5p.ebai_id parent_cat_ebai_id, t5p.ebai_sync_status parent_cat_ebai_sync_status,
cat1.vendor_category_id ebai_cat3_id, cat2.vendor_category_id ebai_cat2_id, cat2.parent_id ebai_cat1_id cat1.vendor_category_id ebai_cat3_id, cat2.vendor_category_id ebai_cat2_id, cat2.parent_id ebai_cat1_id
FROM store_sku_bind t1 FROM store_sku_bind t1
JOIN sku t2 ON t1.sku_id = t2.id JOIN sku t2 ON t1.sku_id = t2.id AND t2.deleted_at = ?
JOIN sku_name t3 ON t2.name_id = t3.id JOIN sku_name t3 ON t2.name_id = t3.id AND t3.deleted_at = ?
JOIN sku_category t4 ON t3.category_id = t4.id JOIN sku_category t4 ON t3.category_id = t4.id AND t4.deleted_at = ?
JOIN sku_category t4p ON t4p.id = t4.parent_id JOIN sku_category t4p ON t4p.id = t4.parent_id AND t4p.deleted_at = ?
LEFT JOIN store_sku_category_map t5 ON t5.store_id = t1.store_id AND t5.category_id = t4.id LEFT JOIN store_sku_category_map t5 ON t5.store_id = t1.store_id AND t5.category_id = t4.id AND t5.deleted_at = ?
LEFT JOIN store_sku_category_map t5p ON t5p.store_id = t1.store_id AND t5p.category_id = t4p.id LEFT JOIN store_sku_category_map t5p ON t5p.store_id = t1.store_id AND t5p.category_id = t4p.id AND t5p.deleted_at = ?
LEFT JOIN sku_vendor_category cat1 ON t4.ebai_category_id = cat1.vendor_category_id AND cat1.vendor_id = ? LEFT JOIN sku_vendor_category cat1 ON t4.ebai_category_id = cat1.vendor_category_id AND cat1.vendor_id = ?
LEFT JOIN sku_vendor_category cat2 ON cat1.parent_id = cat2.vendor_category_id AND cat1.vendor_id = ? LEFT JOIN sku_vendor_category cat2 ON cat1.parent_id = cat2.vendor_category_id AND cat1.vendor_id = ?
WHERE t1.store_id = ? AND (t1.ebai_sync_status <> 0) WHERE t1.store_id = ? AND (t1.ebai_sync_status <> 0)
` `
sqlParams := []interface{}{ sqlParams := []interface{}{
utils.DefaultTimeValue,
utils.DefaultTimeValue,
utils.DefaultTimeValue,
utils.DefaultTimeValue,
utils.DefaultTimeValue,
utils.DefaultTimeValue,
model.VendorIDEBAI, model.VendorIDEBAI,
model.VendorIDEBAI, model.VendorIDEBAI,
storeID, storeID,
} }
if skuIDs != nil && len(skuIDs) > 0 { if len(skuIDs) > 0 {
sql += " AND t1.sku_id IN (" + dao.GenQuestionMarks(len(skuIDs)) + ")" sql += " AND t1.sku_id IN (" + dao.GenQuestionMarks(len(skuIDs)) + ")"
sqlParams = append(sqlParams, skuIDs) sqlParams = append(sqlParams, skuIDs)
} }
@@ -147,11 +130,11 @@ func (p *PurchaseHandler) syncOneStoreSkus(parentTask tasksch.ITask, db *dao.Dao
} }
for k := range catList2Add { for k := range catList2Add {
if err = dao.AddStoreCategoryMap(db, storeID, k, model.VendorIDEBAI, "", model.SyncFlagNewMask, userName); err != nil { if err = dao.AddStoreCategoryMap(db, storeID, k, model.VendorIDEBAI, "", model.SyncFlagNewMask, userName); err != nil {
return err return "", err
} }
} }
if err = p.SyncOneStoreCategories(db, storeID, userName); err != nil { if err = p.SyncOneStoreCategories(db, storeID, userName); err != nil {
return err return "", err
} }
if err = dao.GetRows(db, &storeSkuInfoList, sql, sqlParams...); err == nil { if err = dao.GetRows(db, &storeSkuInfoList, sql, sqlParams...); err == nil {
task := tasksch.NewParallelTask("syncOneStoreSkus skus", nil, userName, func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) { task := tasksch.NewParallelTask("syncOneStoreSkus skus", nil, userName, func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
@@ -185,7 +168,7 @@ func (p *PurchaseHandler) syncOneStoreSkus(parentTask tasksch.ITask, db *dao.Dao
_, err = task.GetResult(0) _, err = task.GetResult(0)
} }
} }
return err return "", err
} }
func (p *PurchaseHandler) SyncStoresCategories(db *dao.DaoDB, storeIDs []int, userName string) (err error) { func (p *PurchaseHandler) SyncStoresCategories(db *dao.DaoDB, storeIDs []int, userName string) (err error) {
@@ -381,6 +364,7 @@ func (p *PurchaseHandler) syncOneStoreCategoriesFromRemote2Local(db *dao.DaoDB,
// 从本地同步分类信息到饿百 // 从本地同步分类信息到饿百
// 测试过程中出现过，父分类创建成功后马上创建子分类会报没有父分类错 // 测试过程中出现过，父分类创建成功后马上创建子分类会报没有父分类错
// todo 对于deleted_at的处理有问题
func (p *PurchaseHandler) SyncOneStoreCategories(db *dao.DaoDB, storeID int, userName string) (err error) { func (p *PurchaseHandler) SyncOneStoreCategories(db *dao.DaoDB, storeID int, userName string) (err error) {
globals.SugarLogger.Debugf("SyncOneStoreCategories storeID:%d, userName:%s", storeID, userName) globals.SugarLogger.Debugf("SyncOneStoreCategories storeID:%d, userName:%s", storeID, userName)

2
business/partner/purchase/elm/store_sku.go
View File

@@ -18,7 +18,7 @@ func (p *PurchaseHandler) ReadStoreSku(storeID, skuID int) (skuNameExt *model.Sk
return nil, nil return nil, nil
} }
func (p *PurchaseHandler) SyncStoresSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, db *dao.DaoDB, storeIDs []int, skuIDs []int, isAsync bool) (hint string, err error) { func (p *PurchaseHandler) SyncStoreSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, storeID int, skuIDs []int, isAsync bool) (hint string, err error) {
return hint, err return hint, err
} }

186
business/partner/purchase/jd/store_sku.go
View File

@@ -23,116 +23,102 @@ type tStoreSkuBindExt struct {
} }
// 京东到家，以有库存表示关注（认领） // 京东到家，以有库存表示关注（认领）
func (p *PurchaseHandler) SyncStoresSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, db *dao.DaoDB, storeIDs []int, skuIDs []int, isAsync bool) (hint string, err error) { func (p *PurchaseHandler) SyncStoreSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, storeID int, skuIDs []int, isAsync bool) (hint string, err error) {
globals.SugarLogger.Debugf("jd SyncStoresSkus, storeIDs:%v, skuIDs:%v", storeIDs, skuIDs) globals.SugarLogger.Debugf("jd SyncStoresSkus, storeID:%d, skuIDs:%v", storeID, skuIDs)
sqlWhere := `
parallelCount := 1 WHERE (t1.jd_sync_status <> 0) AND t1.store_id = ?
if len(skuIDs) < MaxSkuBatchSize { `
parallelCount = 10 sqlWhereParams := []interface{}{
storeID,
} }
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 { if len(skuIDs) > 0 {
sqlWhere += " AND t1.sku_id IN (" + dao.GenQuestionMarks(len(skuIDs)) + ")" sqlWhere += " AND t1.sku_id IN (" + dao.GenQuestionMarks(len(skuIDs)) + ")"
sqlWhereParams = append(sqlWhereParams, skuIDs) sqlWhereParams = append(sqlWhereParams, skuIDs)
} }
sql := ` sql := `
SELECT t1.*, t2.price_percentage SELECT t1.*, t2.price_percentage
FROM store_sku_bind t1 FROM store_sku_bind t1
JOIN store_map t2 ON t1.store_id = t2.store_id AND t2.vendor_id = ? AND t2.deleted_at = ? JOIN store_map t2 ON t1.store_id = t2.store_id AND t2.vendor_id = ? AND t2.deleted_at = ?
` + sqlWhere ` + sqlWhere
var storeSkus []*tStoreSkuBindExt var storeSkus []*tStoreSkuBindExt
sqlParams := []interface{}{ sqlParams := []interface{}{
model.VendorIDJD, model.VendorIDJD,
utils.DefaultTimeValue, utils.DefaultTimeValue,
} }
if err = dao.GetRows(db, &storeSkus, sql, append(sqlParams, sqlWhereParams...)...); err == nil { db := dao.GetDB()
outStationNo := utils.Int2Str(storeID) if err = dao.GetRows(db, &storeSkus, sql, append(sqlParams, sqlWhereParams...)...); err != nil {
task := tasksch.NewParallelTask("SyncStoresSkus inner", tasksch.NewParallelConfig().SetBatchSize(MaxSkuBatchSize), ctx.GetUserName(), func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (interface{}, error) { return "", err
var skuPriceInfoList []*jdapi.SkuPriceInfo }
var skuVendibilityList []*jdapi.StockVendibility outStationNo := utils.Int2Str(storeID)
var skuStockList []*jdapi.SkuStock 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
for _, v := range batchItemList { var skuVendibilityList []*jdapi.StockVendibility
storeSku := v.(*tStoreSkuBindExt) var skuStockList []*jdapi.SkuStock
alreadyAddStock := false for _, v := range batchItemList {
if storeSku.JdSyncStatus&model.SyncFlagChangedMask != 0 { storeSku := v.(*tStoreSkuBindExt)
if storeSku.JdSyncStatus&(model.SyncFlagDeletedMask|model.SyncFlagNewMask) != 0 { // 关注或取消关注 alreadyAddStock := false
stock := &jdapi.SkuStock{ if storeSku.JdSyncStatus&model.SyncFlagChangedMask != 0 {
OutSkuId: utils.Int2Str(storeSku.SkuID), if storeSku.JdSyncStatus&(model.SyncFlagDeletedMask|model.SyncFlagNewMask) != 0 { // 关注或取消关注
StockQty: MaxStockQty, stock := &jdapi.SkuStock{
} OutSkuId: utils.Int2Str(storeSku.SkuID),
if storeSku.DeletedAt != utils.DefaultTimeValue { StockQty: MaxStockQty,
stock.StockQty = 0
} else {
alreadyAddStock = true
}
skuStockList = append(skuStockList, stock)
}
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
} else if !alreadyAddStock { // 如果是设置可售则自动将库存加满
stock := &jdapi.SkuStock{
OutSkuId: utils.Int2Str(storeSku.SkuID),
StockQty: MaxStockQty,
}
skuStockList = append(skuStockList, stock)
}
skuVendibilityList = append(skuVendibilityList, vendibility)
}
} }
if storeSku.DeletedAt != utils.DefaultTimeValue {
stock.StockQty = 0
} else {
alreadyAddStock = true
}
skuStockList = append(skuStockList, stock)
} }
if globals.EnableStoreWrite { if storeSku.JdSyncStatus&(model.SyncFlagPriceMask|model.SyncFlagNewMask) != 0 {
// todo 以下可以优化为并行操作 skuPriceInfoList = append(skuPriceInfoList, &jdapi.SkuPriceInfo{
globals.SugarLogger.Debug(utils.Format4Output(skuVendibilityList, false), utils.Format4Output(skuPriceInfoList, false), utils.Format4Output(skuStockList, false)) OutSkuId: utils.Int2Str(storeSku.SkuID),
if len(skuVendibilityList) > 0 { Price: jxutils.CaculateSkuVendorPrice(storeSku.Price, storeSku.PricePercentage),
_, 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 if storeSku.JdSyncStatus&(model.SyncFlagSaleMask|model.SyncFlagNewMask) != 0 {
}, storeSkus) vendibility := &jdapi.StockVendibility{
t.AddChild(task).Run() OutSkuId: utils.Int2Str(storeSku.SkuID),
if _, err = task.GetResult(0); err == nil { DoSale: true,
sql := ` }
UPDATE store_sku_bind t1 if storeSku.Status != model.StoreSkuBindStatusNormal {
SET t1.jd_sync_status = 0 vendibility.DoSale = false
` + sqlWhere } else if !alreadyAddStock { // 如果是设置可售则自动将库存加满
_, err = dao.ExecuteSQL(db, sql, sqlWhereParams...) stock := &jdapi.SkuStock{
OutSkuId: utils.Int2Str(storeSku.SkuID),
StockQty: MaxStockQty,
}
skuStockList = append(skuStockList, stock)
}
skuVendibilityList = append(skuVendibilityList, vendibility)
}
}
}
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 return nil, err
}, storeIDs) }, storeSkus)
ctx.SetTaskOrAddChild(task, parentTask) ctx.SetTaskOrAddChild(task, parentTask)
task.Run() task.Run()
if isAsync { if _, err = task.GetResult(0); err == nil {
return task.ID, nil sql := `
UPDATE store_sku_bind t1
SET t1.jd_sync_status = 0
` + sqlWhere
_, err = dao.ExecuteSQL(db, sql, sqlWhereParams...)
} }
_, err = task.GetResult(0) return task.ID, err
return "", err
} }