From ecba18072411349d340fc50293c8003f832dc69b Mon Sep 17 00:00:00 2001 From: gazebo Date: Tue, 21 May 2019 14:03:40 +0800 Subject: [PATCH] =?UTF-8?q?-=20=E9=87=8D=E6=9E=84=E4=BA=AC=E4=B8=9C?= =?UTF-8?q?=E9=97=A8=E5=BA=97=E5=95=86=E5=93=81=E5=90=8C=E6=AD=A5=E6=93=8D?= =?UTF-8?q?=E4=BD=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- business/model/dao/store_sku.go | 53 +++--- business/partner/purchase/jd/store_sku.go | 187 +++++----------------- 2 files changed, 73 insertions(+), 167 deletions(-) diff --git a/business/model/dao/store_sku.go b/business/model/dao/store_sku.go index 790b851b7..0a0b5d0fe 100644 --- a/business/model/dao/store_sku.go +++ b/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 ` diff --git a/business/partner/purchase/jd/store_sku.go b/business/partner/purchase/jd/store_sku.go index 48c25baec..544ad5709 100644 --- a/business/partner/purchase/jd/store_sku.go +++ b/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) {