- 重构京东门店商品同步操作

business/model/dao/store_sku.go

@@ -134,48 +134,60 @@ func GetStoreCategories(db *DaoDB, vendorID, storeID int, level int) (cats []*St
 	return cats, err
 }
 
-// 单门店模式厂商适用
+// 单多门店模式厂商通用
 func GetStoreSkus(db *DaoDB, vendorID, storeID int, skuIDs []int) (skus []*StoreSkuSyncInfo, err error) {
+	isSingleStorePF := model.MultiStoresVendorMap[vendorID] != 1
 	tableName := "t1"
-	if model.MultiStoresVendorMap[vendorID] == 1 { // 多店模式平台
+	if !isSingleStorePF {
 		tableName = "t2"
 	}
 	vendorSkuNameField := "0"
 	if vendorID == model.VendorIDWSC {
 		vendorSkuNameField = "t1.wsc_id2"
 	}
+	fieldPrefix := ConvertDBFieldPrefix(model.VendorNames[vendorID])
 	sql := `
 	SELECT t1.id bind_id, t1.price, t1.unit_price, t1.status store_sku_status, %s.%s_id vendor_sku_id, t1.%s_sync_status sku_sync_status, %s vendor_name_id,
 	t2.*,
 	t3.id name_id, t3.prefix, t3.name, t3.unit, IF(t3.%s <> '', t3.%s, t3.img) img, t3.upc, t3.%s desc_img,
-	t4.%s_category_id vendor_vendor_cat_id, t4.%s_price_percentage cat_price_percentage,
-	t5.%s_sync_status cat_sync_status, t5.%s_id vendor_cat_id,
-	t5sku.%s_sync_status sku_cat_sync_status, t5sku.%s_id sku_vendor_cat_id
+	t4.%s_category_id vendor_vendor_cat_id, t4.%s_price_percentage cat_price_percentage
+	`
+	fmtParams := []interface{}{
+		tableName, fieldPrefix, fieldPrefix, vendorSkuNameField, GetImgFieldName(vendorID), GetImgFieldName(vendorID), GetDescImgFieldName(vendorID),
+		fieldPrefix, fieldPrefix,
+	}
+	if isSingleStorePF {
+		sql += `,
+		t5.%s_sync_status cat_sync_status, t5.%s_id vendor_cat_id,
+		t5sku.%s_sync_status sku_cat_sync_status, t5sku.%s_id sku_vendor_cat_id`
+		fmtParams = append(fmtParams, fieldPrefix, fieldPrefix, fieldPrefix, fieldPrefix)
+	}
+	sql += `
 	FROM store_sku_bind t1
 	LEFT JOIN sku t2 ON t1.sku_id = t2.id AND t2.deleted_at = ? AND t2.status = ?
 	LEFT JOIN sku_name t3 ON t2.name_id = t3.id AND t3.deleted_at = ? AND t3.status = ?
-	LEFT JOIN sku_category t4 ON t3.category_id = t4.id AND t4.deleted_at = ?
-	LEFT JOIN store_sku_category_map t5 ON t4.id = t5.category_id AND t5.store_id = t1.store_id AND t5.deleted_at = ?
-	LEFT JOIN store_sku_category_map t5sku ON t2.category_id = t5sku.category_id AND t5sku.store_id = t1.store_id AND t5sku.deleted_at = ?
-	WHERE t1.store_id = ? AND (t1.%s_sync_status <> 0 OR (%s.%s_id <> 0 AND t3.id IS NULL))
-	`
+	LEFT JOIN sku_category t4 ON t3.category_id = t4.id AND t4.deleted_at = ?`
 	sqlParams := []interface{}{
 		utils.DefaultTimeValue,
 		model.SkuStatusNormal,
 		utils.DefaultTimeValue,
 		model.SkuStatusNormal,
 		utils.DefaultTimeValue,
-		utils.DefaultTimeValue,
-		utils.DefaultTimeValue,
-		storeID,
 	}
+	if isSingleStorePF {
+		sql += `
+		LEFT JOIN store_sku_category_map t5 ON t4.id = t5.category_id AND t5.store_id = t1.store_id AND t5.deleted_at = ?
+		LEFT JOIN store_sku_category_map t5sku ON t2.category_id = t5sku.category_id AND t5sku.store_id = t1.store_id AND t5sku.deleted_at = ?`
+		sqlParams = append(sqlParams, utils.DefaultTimeValue, utils.DefaultTimeValue)
+	}
+	sql += " WHERE t1.store_id = ? AND (t1.%s_sync_status <> 0 OR (%s.%s_id <> 0 AND t3.id IS NULL))"
+	fmtParams = append(fmtParams, fieldPrefix, tableName, fieldPrefix)
+	sqlParams = append(sqlParams, storeID)
 	if len(skuIDs) > 0 {
 		sql += " AND t1.sku_id IN (" + GenQuestionMarks(len(skuIDs)) + ")"
 		sqlParams = append(sqlParams, skuIDs)
 	}
-	fieldPrefix := ConvertDBFieldPrefix(model.VendorNames[vendorID])
-	sql = fmt.Sprintf(sql, tableName, fieldPrefix, fieldPrefix, vendorSkuNameField, GetImgFieldName(vendorID), GetImgFieldName(vendorID), GetDescImgFieldName(vendorID),
-		fieldPrefix, fieldPrefix, fieldPrefix, fieldPrefix, fieldPrefix, fieldPrefix, fieldPrefix, tableName, fieldPrefix)
+	sql = fmt.Sprintf(sql, fmtParams...)
 	sql += " ORDER BY t1.price"
 	// globals.SugarLogger.Debug(sql)
 	if err = GetRows(db, &skus, sql, sqlParams...); err != nil {
@@ -217,7 +229,8 @@ func GetFullStoreSkus(db *DaoDB, vendorID, storeID int) (skus []*StoreSkuSyncInf
 		model.SkuStatusNormal,
 	}
 	fieldPrefix := ConvertDBFieldPrefix(model.VendorNames[vendorID])
-	sql = fmt.Sprintf(sql, fieldPrefix, fieldPrefix, GetImgFieldName(vendorID), GetImgFieldName(vendorID), fieldPrefix, fieldPrefix, fieldPrefix, fieldPrefix, fieldPrefix)
+	sql = fmt.Sprintf(sql, fieldPrefix, fieldPrefix, GetImgFieldName(vendorID), GetImgFieldName(vendorID),
+		fieldPrefix, fieldPrefix, fieldPrefix, fieldPrefix, fieldPrefix)
 	// globals.SugarLogger.Debug(sql)
 	// globals.SugarLogger.Debug(utils.Format4Output(sqlParams, false))
 	if err = GetRows(db, &skus, sql, sqlParams...); err != nil {
@@ -229,6 +242,7 @@ func GetFullStoreSkus(db *DaoDB, vendorID, storeID int) (skus []*StoreSkuSyncInf
 func SetStoreSkuSyncStatus(db *DaoDB, vendorID int, storeIDs []int, skuIDs []int, syncStatus int) (num int64, err error) {
 	globals.SugarLogger.Debugf("SetStoreSkuSyncStatus, storeIDs:%v, vendorID:%d", storeIDs, vendorID)
 
+	isSingleStorePF := model.MultiStoresVendorMap[vendorID] != 1
 	fieldPrefix := ConvertDBFieldPrefix(model.VendorNames[vendorID])
 	sql := `
 	UPDATE store_sku_bind t1
@@ -242,7 +256,7 @@ func SetStoreSkuSyncStatus(db *DaoDB, vendorID int, storeIDs []int, skuIDs []int
 		utils.DefaultTimeValue,
 		syncStatus,
 	}
-	if (syncStatus & model.SyncFlagNewMask) != 0 {
+	if isSingleStorePF && (syncStatus&model.SyncFlagNewMask) != 0 {
 		sql += `,
 		t1.%s_id = 0
 		`
@@ -266,6 +280,7 @@ func SetStoreSkuSyncStatus(db *DaoDB, vendorID int, storeIDs []int, skuIDs []int
 func SetStoreCategorySyncStatus(db *DaoDB, vendorID int, storeIDs []int, catIDs []int, syncStatus int) (num int64, err error) {
 	globals.SugarLogger.Debugf("SetStoreCategorySyncStatus, storeIDs:%v, vendorID:%d", storeIDs, vendorID)
 
+	isSingleStorePF := model.MultiStoresVendorMap[vendorID] != 1
 	fieldPrefix := ConvertDBFieldPrefix(model.VendorNames[vendorID])
 	sql := `
 	UPDATE store_sku_category_map t1
@@ -279,7 +294,7 @@ func SetStoreCategorySyncStatus(db *DaoDB, vendorID int, storeIDs []int, catIDs
 		utils.DefaultTimeValue,
 		syncStatus,
 	}
-	if (syncStatus & model.SyncFlagNewMask) != 0 {
+	if isSingleStorePF && (syncStatus&model.SyncFlagNewMask) != 0 {
 		sql += `,
 		t1.%s_id = 0
 		`

business/partner/purchase/jd/store_sku.go

@@ -24,57 +24,32 @@ type tStoreSkuBindExt struct {
 }
 
 // 京东到家，以有库存表示关注（认领）
-func (p *PurchaseHandler) SyncStoreSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, storeID int, skuIDs []int, isAsync, isContinueWhenError bool) (hint string, err error) {
-	globals.SugarLogger.Debugf("jd SyncStoresSkus, storeID:%d, skuIDs:%v", storeID, skuIDs)
-	sqlWhere0 := `
-	WHERE (t1.jd_sync_status <> 0) AND t1.store_id = ?
-	`
-	sqlWhere := sqlWhere0
-	sqlWhereParams := []interface{}{
-		storeID,
-	}
-
-	if len(skuIDs) > 0 {
-		sqlWhere += " AND t1.sku_id IN (" + dao.GenQuestionMarks(len(skuIDs)) + ")"
-		sqlWhereParams = append(sqlWhereParams, skuIDs)
-	}
-
-	sql := `
-	SELECT t3.jd_id, t1.*, t2.price_percentage, t2.vendor_store_id, t5.jd_price_percentage cat_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 = ?
-	JOIN sku t3 ON t1.sku_id = t3.id AND t3.deleted_at = ?
-	JOIN sku_name t4 ON t4.id = t3.name_id
-	JOIN sku_category t5 ON t5.id = t4.category_id
-	` + sqlWhere + " ORDER BY t1.updated_at"
-	var storeSkus []*tStoreSkuBindExt
-	sqlParams := []interface{}{
-		model.VendorIDJD,
-		utils.DefaultTimeValue,
-		utils.DefaultTimeValue,
-	}
-	db := dao.GetDB()
-	if err = dao.GetRows(db, &storeSkus, sql, append(sqlParams, sqlWhereParams...)...); err != nil {
+func (p *PurchaseHandler) syncStoreSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, db *dao.DaoDB, storeID int, storeSkus []*dao.StoreSkuSyncInfo, isAsync, isContinueWhenError bool) (hint string, err error) {
+	globals.SugarLogger.Debugf("jd syncStoreSkus, storeID:%d", storeID)
+	storeDetail, err := dao.GetStoreDetail(db, storeID, model.VendorIDJD)
+	if err != nil {
 		return "", err
 	}
-	task := tasksch.NewParallelTask("SyncStoresSkus京东", tasksch.NewParallelConfig().SetBatchSize(jdapi.MaxStoreSkuBatchSize).SetIsContinueWhenError(isContinueWhenError), ctx,
+	task := tasksch.NewParallelTask("syncStoreSkus京东", tasksch.NewParallelConfig().SetBatchSize(jdapi.MaxStoreSkuBatchSize).SetIsContinueWhenError(isContinueWhenError), ctx,
 		func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
 			var skuPriceInfoList []*jdapi.SkuPriceInfo
 			var skuVendibilityList []*jdapi.StockVendibility
 			var skuStockList []*jdapi.SkuStock
-			stationNo := batchItemList[0].(*tStoreSkuBindExt).VendorStoreID
-			var batchSkuIDs []int
+			stationNo := storeDetail.VendorStoreID
+			var batchBindIDs []int
 			for _, v := range batchItemList {
-				storeSku := v.(*tStoreSkuBindExt)
+				storeSku := v.(*dao.StoreSkuSyncInfo)
 				alreadyAddStock := false
-				if storeSku.JdSyncStatus&model.SyncFlagChangedMask != 0 {
-					batchSkuIDs = append(batchSkuIDs, storeSku.SkuID)
-					if storeSku.JdSyncStatus&(model.SyncFlagDeletedMask|model.SyncFlagNewMask) != 0 { // 关注或取消关注
+				if storeSku.SkuSyncStatus&model.SyncFlagChangedMask != 0 || storeSku.BindID == 0 || storeSku.NameID == 0 {
+					if storeSku.BindID > 0 {
+						batchBindIDs = append(batchBindIDs, storeSku.BindID)
+					}
+					if storeSku.SkuSyncStatus&(model.SyncFlagDeletedMask|model.SyncFlagNewMask) != 0 || storeSku.BindID == 0 || storeSku.NameID == 0 { // 关注或取消关注
 						stock := &jdapi.SkuStock{
-							OutSkuId: utils.Int2Str(storeSku.SkuID),
+							OutSkuId: utils.Int2Str(storeSku.ID),
 							StockQty: model.MaxStoreSkuStockQty,
 						}
-						if storeSku.DeletedAt != utils.DefaultTimeValue {
+						if storeSku.SkuSyncStatus&model.SyncFlagDeletedMask != 0 || storeSku.DeletedAt != utils.DefaultTimeValue || storeSku.BindID == 0 || storeSku.NameID == 0 {
 							stock.StockQty = 0
 						} else {
 							alreadyAddStock = true
@@ -83,22 +58,22 @@ func (p *PurchaseHandler) SyncStoreSkus(ctx *jxcontext.Context, parentTask tasks
 							skuStockList = append(skuStockList, stock)
 						}
 					}
-					if storeSku.JdSyncStatus&(model.SyncFlagPriceMask|model.SyncFlagNewMask) != 0 {
+					if storeSku.SkuSyncStatus&(model.SyncFlagPriceMask|model.SyncFlagNewMask) != 0 {
 						skuPriceInfoList = append(skuPriceInfoList, &jdapi.SkuPriceInfo{
-							OutSkuId: utils.Int2Str(storeSku.SkuID),
-							Price:    constrainPrice(jxutils.CaculateSkuVendorPrice(storeSku.Price, storeSku.PricePercentage, storeSku.CatPricePercentage)),
+							OutSkuId: utils.Int2Str(storeSku.ID),
+							Price:    constrainPrice(jxutils.CaculateSkuVendorPrice(int(storeSku.Price), int(storeDetail.PricePercentage), storeSku.CatPricePercentage)),
 						})
 					}
-					if storeSku.JdSyncStatus&(model.SyncFlagSaleMask|model.SyncFlagNewMask) != 0 {
+					if storeSku.SkuSyncStatus&(model.SyncFlagSaleMask|model.SyncFlagNewMask) != 0 {
 						vendibility := &jdapi.StockVendibility{
-							OutSkuId: utils.Int2Str(storeSku.SkuID),
+							OutSkuId: utils.Int2Str(storeSku.ID),
 							DoSale:   true,
 						}
-						if storeSku.Status != model.StoreSkuBindStatusNormal {
+						if storeSku.StoreSkuStatus != model.StoreSkuBindStatusNormal {
 							vendibility.DoSale = false
 						} else if !alreadyAddStock { // 如果是设置可售则自动将库存加满
 							stock := &jdapi.SkuStock{
-								OutSkuId: utils.Int2Str(storeSku.SkuID),
+								OutSkuId: utils.Int2Str(storeSku.ID),
 								StockQty: model.MaxStoreSkuStockQty,
 							}
 							skuStockList = append(skuStockList, stock)
@@ -139,13 +114,13 @@ func (p *PurchaseHandler) SyncStoreSkus(ctx *jxcontext.Context, parentTask tasks
 			if len(errList) == 0 {
 				syncMask = -1
 			}
-			if syncMask != 0 && len(batchSkuIDs) > 0 {
+			if syncMask != 0 && len(batchBindIDs) > 0 {
 				db := dao.GetDB() // 多线程问题
 				sql := `
-			UPDATE store_sku_bind t1
-			SET t1.jd_sync_status = t1.jd_sync_status & ?
-			` + sqlWhere0 + " AND t1.sku_id IN (" + dao.GenQuestionMarks(len(batchSkuIDs)) + ")"
-				if _, err = dao.ExecuteSQL(db, sql, ^syncMask, storeID, batchSkuIDs); err != nil {
+				UPDATE store_sku_bind t1
+				SET t1.jd_sync_status = t1.jd_sync_status & ?
+				WHERE t1.id IN (` + dao.GenQuestionMarks(len(batchBindIDs)) + ")"
+				if _, err = dao.ExecuteSQL(db, sql, ^syncMask, batchBindIDs); err != nil {
 					errList = append(errList, err)
 				}
 			}
@@ -163,6 +138,16 @@ func (p *PurchaseHandler) SyncStoreSkus(ctx *jxcontext.Context, parentTask tasks
 	return task.ID, err
 }
 
+func (p *PurchaseHandler) SyncStoreSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, storeID int, skuIDs []int, isAsync, isContinueWhenError bool) (hint string, err error) {
+	globals.SugarLogger.Debugf("jd SyncStoresSkus, storeID:%d, skuIDs:%v", storeID, skuIDs)
+	db := dao.GetDB()
+	storeSkus, err := dao.GetStoreSkus(db, model.VendorIDJD, storeID, skuIDs)
+	if err != nil {
+		return "", err
+	}
+	return p.syncStoreSkus(ctx, parentTask, db, storeID, storeSkus, isAsync, isContinueWhenError)
+}
+
 func (p *PurchaseHandler) FullSyncStoreSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, storeID int, isAsync, isContinueWhenError bool) (hint string, err error) {
 	globals.SugarLogger.Debugf("jd FullSyncStoreSkus, storeID:%d", storeID)
 	db := dao.GetDB()
@@ -170,105 +155,11 @@ func (p *PurchaseHandler) FullSyncStoreSkus(ctx *jxcontext.Context, parentTask t
 	if err != nil {
 		return "", err
 	}
-	skus, err := dao.GetFullStoreSkus(db, model.VendorIDJD, storeID)
+	storeSkus, err := dao.GetFullStoreSkus(db, model.VendorIDJD, storeID)
 	if err != nil {
 		return "", err
 	}
-	return p.syncStoreSkus(ctx, parentTask, db, storeID, skus, isAsync, isContinueWhenError)
-}
-
-// todo 之后应该与SyncStoreSkus合并
-func (p *PurchaseHandler) syncStoreSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, db *dao.DaoDB, storeID int, skus []*dao.StoreSkuSyncInfo, isAsync, isContinueWhenError bool) (hint string, err error) {
-	globals.SugarLogger.Debugf("jd syncStoreSkus, storeID:%d, len(skus):%d", storeID, len(skus))
-	if len(skus) == 0 {
-		return "", nil
-	}
-	storeDetail, err := dao.GetStoreDetail(db, storeID, model.VendorIDJD)
-	if err != nil {
-		return "", err
-	}
-	stationNo := storeDetail.VendorStoreID
-	task := tasksch.NewParallelTask("SyncStoresSkus京东", tasksch.NewParallelConfig().SetBatchSize(jdapi.MaxStoreSkuBatchSize).SetIsContinueWhenError(isContinueWhenError), ctx,
-		func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (interface{}, error) {
-			var skuPriceInfoList []*jdapi.SkuPriceInfo
-			var skuVendibilityList []*jdapi.StockVendibility
-			var skuStockList []*jdapi.SkuStock
-			var batchSkuIDs []int
-			for _, v := range batchItemList {
-				storeSku := v.(*dao.StoreSkuSyncInfo)
-				alreadyAddStock := false
-				if storeSku.SkuSyncStatus&model.SyncFlagChangedMask != 0 || storeSku.BindID == 0 {
-					if storeSku.BindID != 0 {
-						batchSkuIDs = append(batchSkuIDs, storeSku.BindID)
-					}
-					if storeSku.SkuSyncStatus&(model.SyncFlagDeletedMask|model.SyncFlagNewMask) != 0 || storeSku.BindID == 0 { // 关注或取消关注
-						stock := &jdapi.SkuStock{
-							OutSkuId: utils.Int2Str(storeSku.ID),
-							StockQty: model.MaxStoreSkuStockQty,
-						}
-						if storeSku.DeletedAt != utils.DefaultTimeValue || storeSku.BindID == 0 {
-							stock.StockQty = 0
-						} else {
-							alreadyAddStock = true
-						}
-						if stock.StockQty != 0 || !storeskulock.IsJdStoreSkuLocked(stationNo, storeSku.JdID) {
-							skuStockList = append(skuStockList, stock)
-						}
-					}
-					if storeSku.SkuSyncStatus&(model.SyncFlagPriceMask|model.SyncFlagNewMask) != 0 {
-						skuPriceInfoList = append(skuPriceInfoList, &jdapi.SkuPriceInfo{
-							OutSkuId: utils.Int2Str(storeSku.ID),
-							Price:    constrainPrice(jxutils.CaculateSkuVendorPrice(int(storeSku.Price), int(storeDetail.PricePercentage), storeSku.CatPricePercentage)),
-						})
-					}
-					if storeSku.SkuSyncStatus&(model.SyncFlagSaleMask|model.SyncFlagNewMask) != 0 {
-						vendibility := &jdapi.StockVendibility{
-							OutSkuId: utils.Int2Str(storeSku.ID),
-							DoSale:   true,
-						}
-						if storeSku.StoreSkuStatus != model.StoreSkuBindStatusNormal {
-							vendibility.DoSale = false
-						} else if !alreadyAddStock { // 如果是设置可售则自动将库存加满
-							stock := &jdapi.SkuStock{
-								OutSkuId: utils.Int2Str(storeSku.ID),
-								StockQty: model.MaxStoreSkuStockQty,
-							}
-							skuStockList = append(skuStockList, stock)
-						}
-						if vendibility.DoSale || !storeskulock.IsJdStoreSkuLocked(stationNo, storeSku.JdID) {
-							skuVendibilityList = append(skuVendibilityList, vendibility)
-						}
-					}
-				}
-			}
-			globals.SugarLogger.Debugf("jd syncStoreSkus sync detail, storeID:%d, skuVendibilityList:%s, skuPriceInfoList:%s, skuStockList:%s", storeID, utils.Format4Output(skuVendibilityList, true), utils.Format4Output(skuPriceInfoList, true), utils.Format4Output(skuStockList, true))
-			if globals.EnableJdStoreWrite {
-				// todo 以下可以优化为并行操作
-				if len(skuVendibilityList) > 0 {
-					_, err = api.JdAPI.BatchUpdateVendibility("", stationNo, skuVendibilityList, ctx.GetUserName())
-				}
-				if err == nil && len(skuStockList) > 0 {
-					_, err = api.JdAPI.BatchUpdateCurrentQtys("", stationNo, skuStockList, ctx.GetUserName())
-				}
-				if err == nil && len(skuPriceInfoList) > 0 {
-					_, err = api.JdAPI.UpdateVendorStationPrice("", stationNo, skuPriceInfoList)
-				}
-			}
-			if err == nil && len(batchSkuIDs) > 0 {
-				db := dao.GetDB() // 多线程问题
-				sql := `
-			UPDATE store_sku_bind t1
-			SET t1.jd_sync_status = 0
-			WHERE t1.id IN (` + dao.GenQuestionMarks(len(batchSkuIDs)) + ")"
-				_, err = dao.ExecuteSQL(db, sql, batchSkuIDs)
-			}
-			return nil, err
-		}, skus)
-	tasksch.HandleTask(task, parentTask, false).Run()
-	if !isAsync {
-		_, err = task.GetResult(0)
-	}
-	return task.ID, err
+	return p.syncStoreSkus(ctx, parentTask, db, storeID, storeSkus, isAsync, isContinueWhenError)
 }
 
 func (p *PurchaseHandler) DeleteRemoteStoreSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, storeID int, isAsync, isContinueWhenError bool) (hint string, err error) {