- sync.

business/jxstore/cms/cms.go

@@ -43,6 +43,7 @@ func InitServiceInfo(version, buildDate, gitCommit string) {
 			"bankName":               model.BankName,
 		},
 	}
+	Init()
 }
 
 func GetPurchaseHandler(vendorID int) partner.IPurchasePlatformHandler {

business/jxstore/cms/sku.go

@@ -10,6 +10,7 @@ import (
 	"git.rosy.net.cn/baseapi/utils"
 	"git.rosy.net.cn/jx-callback/business/model"
 	"git.rosy.net.cn/jx-callback/business/model/dao"
+	"git.rosy.net.cn/jx-callback/business/partner"
 )
 
 type SkuNamesInfo struct {
@@ -89,7 +90,9 @@ func ReorderCategories(parentID int, categoryIDs []int, userName string) (err er
 			}
 			// todo 这里应该也需要先置标记
 			if err == nil {
-				err = GetPurchaseHandler(model.VendorIDJD).ReorderCategories(parentCat, userName)
+				if handler := GetPurchaseHandler(model.VendorIDJD); handler != nil {
+					err = handler.(partner.IMultipleStoresHandler).ReorderCategories(parentCat, userName)
+				}
 			}
 		}
 	}
@@ -481,9 +484,8 @@ func DeleteSkuNamePlace(nameID, placeCode int, userName string) (num int64, err
 }
 
 func GetVendorSku(vendorID int, vendorSkuID string) (skuNameInfo *model.SkuNameExt, err error) {
-	handler := GetPurchaseHandler(vendorID)
-	if handler != nil {
-		return handler.ReadSku(vendorSkuID)
+	if handler := GetPurchaseHandler(vendorID); handler != nil {
+		return handler.(partner.IMultipleStoresHandler).ReadSku(vendorSkuID)
 	}
-	return nil, ErrCanNotVendor
+	return nil, ErrCanNotFindVendor
 }

business/jxstore/cms/store.go

@@ -25,8 +25,8 @@ type StoresInfo struct {
 }
 
 var (
-	ErrMissingInput = errors.New("没有有效的输入参数")
-	ErrCanNotVendor = errors.New("vendorID参数不合法")
+	ErrMissingInput     = errors.New("没有有效的输入参数")
+	ErrCanNotFindVendor = errors.New("vendorID参数不合法")
 )
 
 func GetPlaces(parentCode int, vendorID int, includeDisabled bool) ([]*model.Place, error) {
@@ -97,7 +97,7 @@ func GetStores(keyword string, params map[string]interface{}, offset, pageSize i
 
 	if params["storeID"] != nil {
 		sqlWhere += " AND t1.id = ?"
-		sqlParams = append(sqlParams, params["id"].(int))
+		sqlParams = append(sqlParams, params["storeID"].(int))
 	}
 	if params["name"] != nil {
 		sqlWhere += " AND t1.name LIKE ?"
@@ -195,8 +195,7 @@ func GetStores(keyword string, params map[string]interface{}, offset, pageSize i
 }
 
 func GetVendorStore(vendorStoreID string, vendorID int) (retVal *StoreExt, err error) {
-	handler := GetPurchaseHandler(vendorID)
-	if handler != nil {
+	if handler := GetPurchaseHandler(vendorID); handler != nil {
 		result, err2 := handler.ReadStore(vendorStoreID)
 		if err = err2; err == nil {
 			retVal = &StoreExt{
@@ -213,7 +212,7 @@ func GetVendorStore(vendorStoreID string, vendorID int) (retVal *StoreExt, err e
 		}
 		return nil, err
 	}
-	return nil, ErrCanNotVendor
+	return nil, ErrCanNotFindVendor
 }
 
 func UpdateStore(storeID int, payload map[string]interface{}, userName string) (num int64, err error) {
@@ -254,18 +253,22 @@ func GetStoreVendorMaps(db *dao.DaoDB, storeID int, vendorID int) (storeMaps []*
 }
 
 func AddStoreVendorMap(db *dao.DaoDB, storeID, vendorID int, storeMap *model.StoreMap, userName string) (outStoreMap *model.StoreMap, err error) {
-	store, err := GetPurchaseHandler(storeMap.VendorID).ReadStore(storeMap.VendorStoreID)
-	if err == nil {
-		dao.WrapAddIDCULEntity(storeMap, userName)
-		storeMap.StoreID = storeID
-		storeMap.VendorID = vendorID
-		storeMap.DeliveryType = store.DeliveryType
-		storeMap.Status = store.Status
-		storeMap.SyncStatus = model.SyncFlagModifiedMask // 新增绑定门店是修改的概念
-		if err = dao.CreateEntity(db, storeMap); err == nil {
-			outStoreMap = storeMap
-			err = CurVendorSync.SyncStore(db, storeMap.VendorID, store, true, userName)
+	if handler := GetPurchaseHandler(storeMap.VendorID); handler != nil {
+		store, err := handler.ReadStore(storeMap.VendorStoreID)
+		if err == nil {
+			dao.WrapAddIDCULEntity(storeMap, userName)
+			storeMap.StoreID = storeID
+			storeMap.VendorID = vendorID
+			storeMap.DeliveryType = store.DeliveryType
+			storeMap.Status = store.Status
+			storeMap.SyncStatus = model.SyncFlagModifiedMask // 新增绑定门店是修改的概念
+			if err = dao.CreateEntity(db, storeMap); err == nil {
+				outStoreMap = storeMap
+				err = CurVendorSync.SyncStore(db, storeMap.VendorID, store, true, userName)
+			}
 		}
+	} else {
+		err = ErrCanNotFindVendor
 	}
 	return outStoreMap, err
 }
@@ -283,9 +286,13 @@ func UpdateStoreVendorMap(db *dao.DaoDB, storeID, vendorID int, payload map[stri
 		db = dao.GetDB()
 	}
 	if vendorStoreID := utils.Interface2String(payload["vendorStoreID"]); vendorStoreID != "" {
-		jdStore, err2 := GetPurchaseHandler(vendorID).ReadStore(vendorStoreID)
-		if err = err2; err == nil {
-			payload["deliveryType"] = jdStore.DeliveryType
+		if handler := GetPurchaseHandler(vendorID); handler != nil {
+			jdStore, err2 := handler.ReadStore(vendorStoreID)
+			if err = err2; err == nil {
+				payload["deliveryType"] = jdStore.DeliveryType
+			}
+		} else {
+			err = ErrCanNotFindVendor
 		}
 	}
 	if err == nil {

business/jxstore/cms/sync.go

@@ -3,46 +3,75 @@ package cms
 import (
 	"errors"
 
-	"git.rosy.net.cn/baseapi/utils"
+	"git.rosy.net.cn/jx-callback/business/jxcallback/scheduler/basesch"
+	"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/partner"
+	"git.rosy.net.cn/jx-callback/globals"
 )
 
 type VendorSync struct {
 }
 
+var (
+	MultiStoresVendorHandlers []partner.IMultipleStoresHandler
+	SingleStoreVendorIDs      []int
+)
+
 var (
 	ErrHaveNotImplementedYet = errors.New("还没有实现")
 	ErrEntityNotExist        = errors.New("找不到相应实体")
 )
 
+func Init() {
+	for k, v := range basesch.FixedBaseScheduler.PurchasePlatformHandlers {
+		if multiHandler, ok := v.(partner.IMultipleStoresHandler); ok {
+			MultiStoresVendorHandlers = append(MultiStoresVendorHandlers, multiHandler)
+		} else if _, ok := v.(partner.ISingleStoreHandler); ok {
+			SingleStoreVendorIDs = append(SingleStoreVendorIDs, k)
+		} else {
+			panic("delivery platform type is wrong!")
+		}
+	}
+}
+
 func (v *VendorSync) SyncCategory(db *dao.DaoDB, categoryID int, isForce bool, userName string) (err error) {
 	if db == nil {
 		db = dao.GetDB()
 	}
-	var cats []*model.SkuCategory
-	cond := make(map[string]interface{})
-	if categoryID > 0 {
-		cond[model.FieldID] = categoryID
-	}
-	if err = dao.GetEntities(db, &cats, cond, true); err == nil {
-		for _, cat := range cats {
-			if (cat.JdSyncStatus & model.SyncFlagDeletedMask) != 0 { //删除
-				err = GetPurchaseHandler(model.VendorIDJD).DeleteCategory(cat, userName)
-			} else if (cat.JdSyncStatus&model.SyncFlagNewMask) != 0 || isForce { // 新增
-				err = GetPurchaseHandler(model.VendorIDJD).CreateCategory(cat, userName)
-			} else if (cat.JdSyncStatus & model.SyncFlagModifiedMask) != 0 { // 修改
-				err = GetPurchaseHandler(model.VendorIDJD).UpdateCategory(cat, userName)
-			}
-			if err == nil {
-				cat.JdSyncStatus = 0
-				_, err = dao.UpdateEntity(db, cat, model.FieldJdSyncStatus)
-			}
-			if err != nil {
-				break
-			}
+	// v.LoopSingleStoreVendors(db, "SyncCategory", userName, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
+	// 	storeMap := batchItemList[0].(*model.StoreMap)
+	// 	handler := GetPurchaseHandler(storeMap.VendorID).(partner.ISingleStoreHandler)
+	// 	return nil, nil
+	// })
+
+	err = v.LoopMultiStoresVendors(db, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
+		multiStoresHandler := batchItemList[0].(partner.IMultipleStoresHandler)
+		var cats []*model.SkuCategory
+		cond := make(map[string]interface{})
+		if categoryID > 0 {
+			cond[model.FieldID] = categoryID
 		}
-	}
+		if err = dao.GetEntities(db, &cats, cond, true); err == nil {
+			tasksch.RunTask("", func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
+				cat := batchItemList[0].(*model.SkuCategory)
+				if (cat.JdSyncStatus & model.SyncFlagDeletedMask) != 0 { //删除
+					err = multiStoresHandler.DeleteCategory(cat, userName)
+				} else if (cat.JdSyncStatus&model.SyncFlagNewMask) != 0 || isForce { // 新增
+					err = multiStoresHandler.CreateCategory(cat, userName)
+				} else if (cat.JdSyncStatus & model.SyncFlagModifiedMask) != 0 { // 修改
+					err = multiStoresHandler.UpdateCategory(cat, userName)
+				}
+				if err == nil {
+					cat.JdSyncStatus = 0
+					_, err = dao.UpdateEntity(db, cat, model.FieldJdSyncStatus)
+				}
+				return nil, err
+			}, nil, len(cats), 1, "", cats)
+		}
+		return nil, err
+	})
 	return err
 }
 
@@ -50,17 +79,18 @@ func (v *VendorSync) SyncStore(db *dao.DaoDB, vendorID int, store *model.Store,
 	if db == nil {
 		db = dao.GetDB()
 	}
-	storeMapList, err := v.GetStoreMapInfo(db, store.ID)
-	if err == nil {
-		for _, storeMap := range storeMapList {
-			if (vendorID == -1 || vendorID == storeMap.VendorID) && (isForce || storeMap.SyncStatus != 0) { // 对于门店，当前只有修改标记
-				if err = GetPurchaseHandler(storeMap.VendorID).UpdateStore(store.ID); err == nil {
+	err = v.LoopStoreMap(db, store.ID, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
+		storeMap := batchItemList[0].(*model.StoreMap)
+		if (vendorID == -1 || vendorID == storeMap.VendorID) && (isForce || storeMap.SyncStatus != 0) {
+			if handler := GetPurchaseHandler(storeMap.VendorID); handler != nil {
+				if err = handler.UpdateStore(store.ID); err == nil {
 					storeMap.SyncStatus = 0
-					dao.UpdateEntity(db, storeMap, model.FieldSyncStatus)
+					_, err = dao.UpdateEntity(db, storeMap, model.FieldSyncStatus)
 				}
 			}
 		}
-	}
+		return nil, err
+	})
 	return err
 }
 
@@ -68,60 +98,81 @@ func (v *VendorSync) SyncSku(db *dao.DaoDB, nameID, skuID int, isForce bool, use
 	if db == nil {
 		db = dao.GetDB()
 	}
-	var skuList []*model.Sku
-	cond := make(map[string]interface{})
-	if nameID != -1 {
-		cond[model.FieldNameID] = nameID
-	}
-	if skuID != -1 {
-		cond[model.FieldID] = skuID
-	}
-	if err = dao.GetEntities(db, &skuList, cond, true); err == nil {
-		for _, sku := range skuList {
-			if (skuID == -1 || skuID == sku.ID) && (isForce || sku.JdSyncStatus != 0) {
-				updateFields := []string{model.FieldJdSyncStatus}
-				if sku.JdSyncStatus&model.SyncFlagDeletedMask != 0 { // 删除
-					err = GetPurchaseHandler(model.VendorIDJD).CreateSku(sku, userName)
-					updateFields = append(updateFields, model.FieldJdID)
-				} else if sku.JdSyncStatus&model.SyncFlagNewMask != 0 { // 新增
-					err = GetPurchaseHandler(model.VendorIDJD).DeleteSku(sku, userName)
-				} else if sku.JdSyncStatus&model.SyncFlagModifiedMask != 0 { // 修改
-					err = GetPurchaseHandler(model.VendorIDJD).UpdateSku(sku, userName)
-				}
-				if err == nil {
-					sku.JdSyncStatus = 0
-					dao.UpdateEntity(db, sku, updateFields...)
-				}
-			}
+	// v.LoopSingleStoreVendors(db, "SyncSku", userName, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
+	// 	storeMap := batchItemList[0].(*model.StoreMap)
+	// 	handler := GetPurchaseHandler(storeMap.VendorID).(partner.ISingleStoreHandler)
+	// 	return nil, nil
+	// })
+	globals.SugarLogger.Debug("SyncSku1")
+	err = v.LoopMultiStoresVendors(db, func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
+		globals.SugarLogger.Debug("SyncSku2")
+		multiStoresHandler := batchItemList[0].(partner.IMultipleStoresHandler)
+		var skuList []*model.Sku
+		cond := make(map[string]interface{})
+		if nameID != -1 {
+			cond[model.FieldNameID] = nameID
 		}
-	}
+		if skuID != -1 {
+			cond[model.FieldID] = skuID
+		}
+		if err = dao.GetEntities(db, &skuList, cond, true); err == nil {
+			tasksch.RunTask("", func(batchItemList []interface{}, params ...interface{}) (interface{}, error) {
+				sku := batchItemList[0].(*model.Sku)
+				globals.SugarLogger.Debug("fuck:", sku)
+				if (skuID == -1 || skuID == sku.ID) && (isForce || sku.JdSyncStatus != 0) {
+					updateFields := []string{model.FieldJdSyncStatus}
+					if sku.JdSyncStatus&model.SyncFlagDeletedMask != 0 { // 删除
+						err = multiStoresHandler.CreateSku(sku, userName)
+						updateFields = append(updateFields, model.FieldJdID)
+					} else if sku.JdSyncStatus&model.SyncFlagNewMask != 0 { // 新增
+						err = multiStoresHandler.DeleteSku(sku, userName)
+					} else if sku.JdSyncStatus&model.SyncFlagModifiedMask != 0 { // 修改
+						err = multiStoresHandler.UpdateSku(sku, userName)
+					}
+					if err == nil {
+						sku.JdSyncStatus = 0
+						dao.UpdateEntity(db, sku, updateFields...)
+					}
+				}
+				return nil, err
+			}, nil, len(skuList), 1, "", skuList)
+		}
+		return nil, err
+	})
 	return err
 }
 
 func (v *VendorSync) SyncStoreSku(db *dao.DaoDB, storeID int, skuIDs []int, isForce bool, userName string) (err error) {
-	storeMapList, err := v.GetStoreMapInfo(db, storeID)
-	var skuNamesInfo *StoreSkuNamesInfo
-	if skuNamesInfo, err = GetStoreSkus(storeID, "", utils.Params2Map("skuID", skuIDs), 0, 10000); err == nil {
-		if skuNamesInfo.TotalCount != len(skuNamesInfo.SkuNames) {
-			panic("too many skus!")
-		}
-		for _, storeMap := range storeMapList {
-			switch storeMap.VendorID {
-			case model.VendorIDJD:
-				err = v.SyncStoreSku2JD(db, skuNamesInfo.SkuNames, isForce, userName)
-			}
-		}
+
+	return err
+}
+
+func (v *VendorSync) LoopStoreMap(db *dao.DaoDB, storeID int, handler tasksch.WorkFunc) (err error) {
+	storeMaps, err := GetStoreVendorMaps(db, storeID, -1)
+	if err == nil {
+		task := tasksch.RunTask("", handler, nil, len(storeMaps), 1, "", storeMaps)
+		_, err = task.GetResult(0)
 	}
 	return err
 }
 
-func (v *VendorSync) SyncStoreSku2JD(db *dao.DaoDB, skuInfoList []*StoreSkuNameExt, isForce bool, userName string) (err error) {
-	// for _, skuInfo := range skuInfoList {
-
-	// }
+func (v *VendorSync) LoopMultiStoresVendors(db *dao.DaoDB, handler tasksch.WorkFunc) (err error) {
+	task := tasksch.RunTask("", handler, nil, len(MultiStoresVendorHandlers), 1, "", MultiStoresVendorHandlers)
+	_, err = task.GetResult(0)
 	return err
 }
 
-func (v *VendorSync) GetStoreMapInfo(db *dao.DaoDB, storeID int) (storeMapList []*model.StoreMap, err error) {
-	return storeMapList, dao.GetEntities(db, &storeMapList, utils.Params2Map(model.FieldStoreID, storeID), true)
+func (v *VendorSync) LoopSingleStoreVendors(db *dao.DaoDB, taskName, userName string, handler tasksch.WorkFunc) (err error) {
+	var storeMaps []*model.StoreMap
+	if err = dao.GetRows(db, &storeMaps, `
+	SELECT *
+	FROM store_map
+	WHERE vendor_id IN (`+dao.GenQuestionMarks(len(SingleStoreVendorIDs))+")", SingleStoreVendorIDs); err == nil {
+		parellelCount := len(storeMaps)
+		if parellelCount > 20 {
+			parellelCount = 20
+		}
+		tasksch.RunManagedTask(taskName, handler, nil, parellelCount, 1, userName, storeMaps)
+	}
+	return err
 }

business/partner/partner.go

@@ -68,24 +68,31 @@ type IPurchasePlatformHandler interface {
 	// OpenStore(vendorStoreID string, userName string) error
 	// CloseStore(vendorStoreID, closeNotice, userName string) error
 
-	// Sku
-	CreateCategory(cat *model.SkuCategory, userName string) (err error)
-	ReadCategory(vendorCatID string) (cat *model.SkuCategory, err error)
+	SyncStoreSku(storeID int, skuIDs []int) (err error)
+}
+
+type IMultipleStoresHandler interface {
 	ReadCategories() (cats []*model.SkuCategory, err error)
+	// ReadCategory(vendorCatID string) (cat *model.SkuCategory, err error)
+
+	CreateCategory(cat *model.SkuCategory, userName string) (err error)
 	UpdateCategory(cat *model.SkuCategory, userName string) error
 	DeleteCategory(cat *model.SkuCategory, userName string) error
 	ReorderCategories(parentCat *model.SkuCategory, userName string) (err error)
 
+	// sku
 	CreateSku(sku *model.Sku, userName string) (err error)
 	ReadSku(vendorSkuID string) (skuNameExt *model.SkuNameExt, err error)
 	UpdateSku(sku *model.Sku, userName string) (err error)
 	DeleteSku(sku *model.Sku, userName string) (err error)
+}
 
-	//store sku
-	// CreateStoreSku(storeID int, sku *model.Sku, userName string) (err error)
-	// ReadStoreSku(vendorSkuID string) (skuName *model.SkuName, sellPlaces []*model.Place, sku *model.Sku, err error)
-	// UpdateStoreSku(storeID int, sku *model.Sku, userName string) (err error)
-	// DeleteStoreSku(storeID int, sku *model.Sku, userName string) (err error)
+type ISingleStoreHandler interface {
+	SyncCategories(catIDs []int) (err error)
+	ReadStoreCategories(storeID int) (cats []*model.SkuCategory, err error)
+
+	SyncSkus(skuIDs []int) (err error)
+	ReadStoreSku(storeID, skuID int) (skuNameExt *model.SkuNameExt, err error)
 }
 
 type IDeliveryPlatformHandler interface {

business/partner/purchase/elm/sku.go

@@ -1,43 +1 @@
 package elm
-
-import "git.rosy.net.cn/jx-callback/business/model"
-
-func (p *PurchaseHandler) CreateCategory(cat *model.SkuCategory, userName string) (err error) {
-	return nil
-}
-
-func (p *PurchaseHandler) ReadCategory(vendorCatID string) (cat *model.SkuCategory, err error) {
-	return nil, nil
-}
-
-func (p *PurchaseHandler) ReadCategories() (cats []*model.SkuCategory, err error) {
-	return nil, nil
-}
-
-func (p *PurchaseHandler) UpdateCategory(cat *model.SkuCategory, userName string) error {
-	return nil
-}
-
-func (p *PurchaseHandler) DeleteCategory(cat *model.SkuCategory, userName string) error {
-	return nil
-}
-
-func (p *PurchaseHandler) ReorderCategories(parentCat *model.SkuCategory, userName string) (err error) {
-	return nil
-}
-
-func (p *PurchaseHandler) CreateSku(sku *model.Sku, userName string) (err error) {
-	return nil
-}
-
-func (p *PurchaseHandler) ReadSku(vendorSkuID string) (skuNameExt *model.SkuNameExt, err error) {
-	return nil, nil
-}
-
-func (p *PurchaseHandler) UpdateSku(sku *model.Sku, userName string) (err error) {
-	return nil
-}
-
-func (p *PurchaseHandler) DeleteSku(sku *model.Sku, userName string) (err error) {
-	return nil
-}

business/partner/purchase/elm/store_sku.go

@@ -0,0 +1,20 @@
+package elm
+
+import "git.rosy.net.cn/jx-callback/business/model"
+
+func (p *PurchaseHandler) SyncStoreSku(storeID int, skuIDs []int) (err error) {
+	return nil
+}
+func (p *PurchaseHandler) SyncCategories(catIDs []int) (err error) {
+	return nil
+}
+func (p *PurchaseHandler) ReadStoreCategories(storeID int) (cats []*model.SkuCategory, err error) {
+	return nil, nil
+}
+
+func (p *PurchaseHandler) SyncSkus(skuIDs []int) (err error) {
+	return nil
+}
+func (p *PurchaseHandler) ReadStoreSku(storeID, skuID int) (skuNameExt *model.SkuNameExt, err error) {
+	return nil, nil
+}

business/partner/purchase/jd/store_sku.go

@@ -1 +1,5 @@
 package jd
+
+func (p *PurchaseHandler) SyncStoreSku(storeID int, skuIDs []int) (err error) {
+	return nil
+}