- bug fix for many sync operation.
This commit is contained in:
@@ -3,6 +3,7 @@ package cms
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
|
||||
"git.rosy.net.cn/baseapi/utils"
|
||||
"git.rosy.net.cn/jx-callback/business/jxcallback/scheduler/basesch"
|
||||
@@ -12,6 +13,7 @@ import (
|
||||
"git.rosy.net.cn/jx-callback/business/model/dao"
|
||||
"git.rosy.net.cn/jx-callback/business/partner"
|
||||
"git.rosy.net.cn/jx-callback/globals"
|
||||
"git.rosy.net.cn/jx-callback/globals/api"
|
||||
)
|
||||
|
||||
type VendorSync struct {
|
||||
@@ -45,20 +47,27 @@ var (
|
||||
)
|
||||
|
||||
func Init() {
|
||||
apiMap := map[int]interface{}{
|
||||
model.VendorIDJD: api.JdAPI,
|
||||
model.VendorIDELM: api.ElmAPI,
|
||||
model.VendorIDEBAI: api.EbaiAPI,
|
||||
}
|
||||
CurVendorSync.PurchaseHandlers = make(map[int]partner.IPurchasePlatformHandler)
|
||||
for k, v := range basesch.FixedBaseScheduler.PurchasePlatformHandlers {
|
||||
if multiHandler, ok := v.(partner.IMultipleStoresHandler); ok {
|
||||
CurVendorSync.MultiStoreVendorIDs = append(CurVendorSync.MultiStoreVendorIDs, k)
|
||||
CurVendorSync.PurchaseHandlers[k] = &MultiStoreHandlerWrapper{
|
||||
IMultipleStoresHandler: multiHandler,
|
||||
if !reflect.ValueOf(apiMap[k]).IsNil() {
|
||||
if multiHandler, ok := v.(partner.IMultipleStoresHandler); ok {
|
||||
CurVendorSync.MultiStoreVendorIDs = append(CurVendorSync.MultiStoreVendorIDs, k)
|
||||
CurVendorSync.PurchaseHandlers[k] = &MultiStoreHandlerWrapper{
|
||||
IMultipleStoresHandler: multiHandler,
|
||||
}
|
||||
} else if singleHandler, ok := v.(partner.ISingleStoreHandler); ok {
|
||||
CurVendorSync.SingleStoreVendorIDs = append(CurVendorSync.SingleStoreVendorIDs, k)
|
||||
CurVendorSync.PurchaseHandlers[k] = &SingleStoreHandlerWrapper{
|
||||
ISingleStoreHandler: singleHandler,
|
||||
}
|
||||
} else {
|
||||
panic(fmt.Sprintf("platform:%d type is wrong!", k))
|
||||
}
|
||||
} else if singleHandler, ok := v.(partner.ISingleStoreHandler); ok {
|
||||
CurVendorSync.SingleStoreVendorIDs = append(CurVendorSync.SingleStoreVendorIDs, k)
|
||||
CurVendorSync.PurchaseHandlers[k] = &SingleStoreHandlerWrapper{
|
||||
ISingleStoreHandler: singleHandler,
|
||||
}
|
||||
} else {
|
||||
panic(fmt.Sprintf("platform:%d type is wrong!", k))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -111,48 +120,60 @@ func (v *VendorSync) GetSingleStoreHandler(vendorID int) partner.ISingleStoreHan
|
||||
return nil
|
||||
}
|
||||
|
||||
func (v *VendorSync) SyncCategory(db *dao.DaoDB, categoryID int, isForce bool, userName string) (err error) {
|
||||
err = v.LoopMultiStoresVendors(db, "SyncCategory", userName, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
|
||||
func (v *VendorSync) syncCategories(multiStoresHandler partner.IMultipleStoresHandler, db *dao.DaoDB, cats []*model.SkuCategory, userName string) (err error) {
|
||||
syncStatusFieldName := multiStoresHandler.GetFieldSyncStatusName()
|
||||
task := tasksch.RunTask("syncCategories", false, nil, len(cats), 1, userName, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
|
||||
cat := batchItemList[0].(*model.SkuCategory)
|
||||
updateFields := []string{syncStatusFieldName}
|
||||
syncStatus := jxutils.GetObjFieldByName(cat, syncStatusFieldName).(int8)
|
||||
if (syncStatus & model.SyncFlagDeletedMask) != 0 { //删除
|
||||
err = multiStoresHandler.DeleteCategory(db, cat, userName)
|
||||
} else if (syncStatus & model.SyncFlagNewMask) != 0 { // 新增
|
||||
err = multiStoresHandler.CreateCategory(db, cat, userName)
|
||||
updateFields = append(updateFields, multiStoresHandler.GetFieldIDName())
|
||||
} else if (syncStatus & model.SyncFlagModifiedMask) != 0 { // 修改
|
||||
err = multiStoresHandler.UpdateCategory(db, cat, userName)
|
||||
}
|
||||
if err == nil {
|
||||
jxutils.SetObjFieldByName(cat, syncStatusFieldName, int8(0))
|
||||
_, err = dao.UpdateEntity(db, cat, updateFields...)
|
||||
}
|
||||
return nil, err
|
||||
}, cats)
|
||||
_, err = task.GetResult(0)
|
||||
return err
|
||||
}
|
||||
|
||||
func (v *VendorSync) SyncCategory(db *dao.DaoDB, categoryID int, isAsync bool, userName string) (hint string, err error) {
|
||||
globals.SugarLogger.Debug(v.MultiStoreVendorIDs)
|
||||
hint, err = v.LoopMultiStoresVendors(db, "SyncCategory", isAsync, userName, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
|
||||
multiStoresHandler := v.GetMultiStoreHandler(batchItemList[0].(int))
|
||||
syncStatusFieldName := multiStoresHandler.GetFieldSyncStatusName()
|
||||
var cats []*model.SkuCategory
|
||||
cond := make(map[string]interface{})
|
||||
if categoryID > 0 {
|
||||
cond[model.FieldID] = categoryID
|
||||
} else {
|
||||
cond[model.FieldLevel] = 1
|
||||
}
|
||||
err := dao.GetEntitiesByKV(db, &cats, cond, true)
|
||||
if err == nil {
|
||||
task := tasksch.RunTask("", false, nil, len(cats), 1, userName, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
|
||||
for _, v := range batchItemList {
|
||||
cat := v.(*model.SkuCategory)
|
||||
updateFields := []string{syncStatusFieldName}
|
||||
syncStatus := jxutils.GetObjFieldByName(cat, syncStatusFieldName).(int8)
|
||||
if (syncStatus & model.SyncFlagDeletedMask) != 0 { //删除
|
||||
err = multiStoresHandler.DeleteCategory(db, cat, userName)
|
||||
} else if (syncStatus&model.SyncFlagNewMask) != 0 || isForce { // 新增
|
||||
err = multiStoresHandler.CreateCategory(db, cat, userName)
|
||||
updateFields = append(updateFields, multiStoresHandler.GetFieldIDName())
|
||||
} else if (syncStatus & model.SyncFlagModifiedMask) != 0 { // 修改
|
||||
err = multiStoresHandler.UpdateCategory(db, cat, userName)
|
||||
}
|
||||
if err == nil {
|
||||
jxutils.SetObjFieldByName(cat, syncStatusFieldName, int8(0))
|
||||
_, err = dao.UpdateEntity(db, cat, updateFields...)
|
||||
} else {
|
||||
break
|
||||
}
|
||||
}
|
||||
return nil, err
|
||||
}, cats)
|
||||
_, err = task.GetResult(0)
|
||||
err = v.syncCategories(multiStoresHandler, db, cats, userName)
|
||||
}
|
||||
if err != nil || categoryID > 0 {
|
||||
return nil, err
|
||||
}
|
||||
cond[model.FieldLevel] = 2
|
||||
err = dao.GetEntitiesByKV(db, &cats, cond, true)
|
||||
if err == nil {
|
||||
err = v.syncCategories(multiStoresHandler, db, cats, userName)
|
||||
}
|
||||
return nil, err
|
||||
})
|
||||
return err
|
||||
return "", err
|
||||
}
|
||||
|
||||
func (v *VendorSync) SyncReorderCategories(db *dao.DaoDB, categoryID int, isForce bool, userName string) (err error) {
|
||||
err = v.LoopMultiStoresVendors(db, "SyncReorderCategories", userName, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
|
||||
func (v *VendorSync) SyncReorderCategories(db *dao.DaoDB, categoryID int, isAsync bool, userName string) (hint string, err error) {
|
||||
hint, err = v.LoopMultiStoresVendors(db, "SyncReorderCategories", isAsync, userName, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
|
||||
multiStoresHandler := v.GetMultiStoreHandler(batchItemList[0].(int))
|
||||
err2 := multiStoresHandler.ReorderCategories(db, categoryID, userName)
|
||||
if err2 == nil {
|
||||
@@ -162,14 +183,14 @@ func (v *VendorSync) SyncReorderCategories(db *dao.DaoDB, categoryID int, isForc
|
||||
}
|
||||
return nil, err2
|
||||
})
|
||||
return err
|
||||
return "", err
|
||||
}
|
||||
|
||||
func (v *VendorSync) SyncStore(db *dao.DaoDB, vendorID, storeID int, isForce bool, userName string) (err error) {
|
||||
func (v *VendorSync) SyncStore(db *dao.DaoDB, vendorID, storeID int, isAsync bool, userName string) (hint string, err error) {
|
||||
globals.SugarLogger.Debugf("SyncStore, storeID:%d", storeID)
|
||||
err = v.LoopStoreMap(db, "SyncStore", userName, storeID, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
|
||||
hint, err = v.LoopStoreMap(db, "SyncStore", isAsync, userName, storeID, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
|
||||
storeMap := batchItemList[0].(*model.StoreMap)
|
||||
if (vendorID == -1 || vendorID == storeMap.VendorID) && (isForce || storeMap.SyncStatus != 0) {
|
||||
if (vendorID == -1 || vendorID == storeMap.VendorID) && (storeMap.SyncStatus != 0) {
|
||||
if handler := v.GetStoreHandler(storeMap.VendorID); handler != nil {
|
||||
if err = handler.UpdateStore(db, storeID, userName); err == nil {
|
||||
storeMap.SyncStatus = 0
|
||||
@@ -179,12 +200,12 @@ func (v *VendorSync) SyncStore(db *dao.DaoDB, vendorID, storeID int, isForce boo
|
||||
}
|
||||
return nil, err
|
||||
})
|
||||
return err
|
||||
return "", err
|
||||
}
|
||||
|
||||
func (v *VendorSync) SyncSku(db *dao.DaoDB, nameID, skuID int, userName string) (err error) {
|
||||
func (v *VendorSync) SyncSku(db *dao.DaoDB, nameID, skuID int, isAsync bool, userName string) (hint string, err error) {
|
||||
globals.SugarLogger.Debugf("SyncSku, nameID:%d, skuID:%d, userName:%s", nameID, skuID, userName)
|
||||
err = v.LoopMultiStoresVendors(db, "SyncSku", userName, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
|
||||
hint, err = v.LoopMultiStoresVendors(db, "SyncSku", isAsync, userName, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
|
||||
multiStoresHandler := v.GetMultiStoreHandler(batchItemList[0].(int))
|
||||
syncStatusFieldName := multiStoresHandler.GetFieldSyncStatusName()
|
||||
var skuList []*model.Sku
|
||||
@@ -199,25 +220,21 @@ func (v *VendorSync) SyncSku(db *dao.DaoDB, nameID, skuID int, userName string)
|
||||
if err == nil {
|
||||
// globals.SugarLogger.Debug(utils.Format4Output(skuList, false))
|
||||
task := tasksch.RunTask("SyncSku", false, nil, len(skuList), 1, userName, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
|
||||
for _, v := range batchItemList {
|
||||
sku := v.(*model.Sku)
|
||||
syncStatus := jxutils.GetObjFieldByName(sku, syncStatusFieldName).(int8)
|
||||
if (skuID == -1 || skuID == sku.ID) && (syncStatus != 0) {
|
||||
updateFields := []string{syncStatusFieldName}
|
||||
if syncStatus&model.SyncFlagDeletedMask != 0 { // 删除
|
||||
err = multiStoresHandler.DeleteSku(db, sku, userName)
|
||||
} else if syncStatus&model.SyncFlagNewMask != 0 { // 新增
|
||||
err = multiStoresHandler.CreateSku(db, sku, userName)
|
||||
updateFields = append(updateFields, multiStoresHandler.GetFieldIDName())
|
||||
} else if syncStatus&model.SyncFlagModifiedMask != 0 { // 修改
|
||||
err = multiStoresHandler.UpdateSku(db, sku, userName)
|
||||
}
|
||||
if err == nil {
|
||||
jxutils.SetObjFieldByName(sku, syncStatusFieldName, int8(0))
|
||||
dao.UpdateEntity(db, sku, updateFields...)
|
||||
} else {
|
||||
break
|
||||
}
|
||||
sku := batchItemList[0].(*model.Sku)
|
||||
syncStatus := jxutils.GetObjFieldByName(sku, syncStatusFieldName).(int8)
|
||||
if (skuID == -1 || skuID == sku.ID) && (syncStatus != 0) {
|
||||
updateFields := []string{syncStatusFieldName}
|
||||
if syncStatus&model.SyncFlagDeletedMask != 0 { // 删除
|
||||
err = multiStoresHandler.DeleteSku(db, sku, userName)
|
||||
} else if syncStatus&model.SyncFlagNewMask != 0 { // 新增
|
||||
err = multiStoresHandler.CreateSku(db, sku, userName)
|
||||
updateFields = append(updateFields, multiStoresHandler.GetFieldIDName())
|
||||
} else if syncStatus&model.SyncFlagModifiedMask != 0 { // 修改
|
||||
err = multiStoresHandler.UpdateSku(db, sku, userName)
|
||||
}
|
||||
if err == nil {
|
||||
jxutils.SetObjFieldByName(sku, syncStatusFieldName, int8(0))
|
||||
dao.UpdateEntity(db, sku, updateFields...)
|
||||
}
|
||||
}
|
||||
return nil, err
|
||||
@@ -226,13 +243,13 @@ func (v *VendorSync) SyncSku(db *dao.DaoDB, nameID, skuID int, userName string)
|
||||
}
|
||||
return nil, err
|
||||
})
|
||||
return err
|
||||
return "", err
|
||||
}
|
||||
|
||||
//
|
||||
func (v *VendorSync) SyncStoresSkus(db *dao.DaoDB, vendorIDs []int, storeIDs []int, skuIDs []int, isAsync bool, userName string) (hint string, err error) {
|
||||
globals.SugarLogger.Debug("SyncStoresSkus")
|
||||
hint, err = v.LoopStoreVendors(db, isAsync, vendorIDs, "SyncStoresSkus", userName, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
|
||||
hint, err = v.LoopStoreVendors(db, vendorIDs, "SyncStoresSkus", isAsync, userName, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
|
||||
handler := v.GetStoreHandler(batchItemList[0].(int))
|
||||
_, err = handler.SyncStoresSkus(db, storeIDs, skuIDs, false, userName)
|
||||
return nil, err
|
||||
@@ -240,25 +257,31 @@ func (v *VendorSync) SyncStoresSkus(db *dao.DaoDB, vendorIDs []int, storeIDs []i
|
||||
return hint, err
|
||||
}
|
||||
|
||||
func (v *VendorSync) LoopStoreMap(db *dao.DaoDB, taskName, userName string, storeID int, handler tasksch.WorkFunc) (err error) {
|
||||
func (v *VendorSync) LoopStoreMap(db *dao.DaoDB, taskName string, isAsync bool, userName string, storeID int, handler tasksch.WorkFunc) (hint string, err error) {
|
||||
storeMaps, err := GetStoreVendorMaps(db, storeID, -1)
|
||||
if err == nil {
|
||||
task := tasksch.RunManagedTask(taskName, false, nil, len(storeMaps), 1, userName, handler, storeMaps)
|
||||
_, err = task.GetResult(0)
|
||||
hint = task.ID
|
||||
if !isAsync {
|
||||
_, err = task.GetResult(0)
|
||||
}
|
||||
}
|
||||
return makeSyncError(err)
|
||||
return "", makeSyncError(err)
|
||||
}
|
||||
|
||||
func (v *VendorSync) LoopMultiStoresVendors(db *dao.DaoDB, taskName, userName string, handler tasksch.WorkFunc) (err error) {
|
||||
func (v *VendorSync) LoopMultiStoresVendors(db *dao.DaoDB, taskName string, isAsync bool, userName string, handler tasksch.WorkFunc) (hint string, err error) {
|
||||
if taskName == "" {
|
||||
taskName = "LoopMultiStoresVendors"
|
||||
}
|
||||
task := tasksch.RunManagedTask(taskName, false, nil, len(v.MultiStoreVendorIDs), 1, userName, handler, v.MultiStoreVendorIDs)
|
||||
_, err = task.GetResult(0)
|
||||
return makeSyncError(err)
|
||||
hint = task.ID
|
||||
if !isAsync {
|
||||
_, err = task.GetResult(0)
|
||||
}
|
||||
return "", makeSyncError(err)
|
||||
}
|
||||
|
||||
func (v *VendorSync) LoopStoreVendors(db *dao.DaoDB, isAsync bool, vendorIDs []int, taskName, userName string, handler tasksch.WorkFunc) (hint string, err error) {
|
||||
func (v *VendorSync) LoopStoreVendors(db *dao.DaoDB, vendorIDs []int, taskName string, isAsync bool, userName string, handler tasksch.WorkFunc) (hint string, err error) {
|
||||
if taskName == "" {
|
||||
taskName = "LoopStoreVendors"
|
||||
}
|
||||
@@ -290,7 +313,7 @@ func (v *VendorSync) LoopStoreVendors(db *dao.DaoDB, isAsync bool, vendorIDs []i
|
||||
return "", makeSyncError(err)
|
||||
}
|
||||
|
||||
func (v *VendorSync) LoopSingleStoreVendors(db *dao.DaoDB, taskName, userName string, handler tasksch.WorkFunc) (err error) {
|
||||
func (v *VendorSync) LoopSingleStoreVendors(db *dao.DaoDB, taskName string, isAsync bool, userName string, handler tasksch.WorkFunc) (hint string, err error) {
|
||||
if taskName == "" {
|
||||
taskName = "LoopSingleStoreVendors"
|
||||
}
|
||||
@@ -304,9 +327,12 @@ func (v *VendorSync) LoopSingleStoreVendors(db *dao.DaoDB, taskName, userName st
|
||||
parellelCount = 20
|
||||
}
|
||||
task := tasksch.RunManagedTask(taskName, false, nil, parellelCount, 1, userName, handler, storeMaps)
|
||||
_, err = task.GetResult(0)
|
||||
hint = task.ID
|
||||
if !isAsync {
|
||||
_, err = task.GetResult(0)
|
||||
}
|
||||
}
|
||||
return makeSyncError(err)
|
||||
return "", makeSyncError(err)
|
||||
}
|
||||
|
||||
func (v *VendorSync) RefreshSkuIDs(nameID, skuID int, userName string) (err error) {
|
||||
@@ -330,7 +356,7 @@ func (v *VendorSync) RefreshSkuIDs(nameID, skuID int, userName string) (err erro
|
||||
db := dao.GetDB()
|
||||
if err = dao.GetRows(db, &ids, sql, sqlParams); err == nil {
|
||||
// globals.SugarLogger.Debug(utils.Format4Output(ids, false))
|
||||
err = v.LoopMultiStoresVendors(db, "RefreshSkuIDs", userName, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
|
||||
_, err = v.LoopMultiStoresVendors(db, "RefreshSkuIDs", false, userName, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
|
||||
multiStoresHandler := v.GetMultiStoreHandler(batchItemList[0].(int))
|
||||
err := multiStoresHandler.SyncSkusIDMap(db, ids, userName)
|
||||
globals.SugarLogger.Debug(err)
|
||||
|
||||
Reference in New Issue
Block a user