package jd import ( "fmt" "git.rosy.net.cn/baseapi/platformapi/jdapi" "git.rosy.net.cn/baseapi/utils" "git.rosy.net.cn/jx-callback/business/jxutils" "git.rosy.net.cn/jx-callback/business/jxutils/jxcontext" "git.rosy.net.cn/jx-callback/business/jxutils/storeskulock" "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/globals" "git.rosy.net.cn/jx-callback/globals/api" ) // 京东到家,以有库存表示关注(认领) 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 } batchSize := jdapi.MaxStoreSkuBatchSize // storeSkusLen := len(storeSkus) // if storeSkusLen < jdapi.MaxStoreSkuBatchSize/2 { // batchSize = 1 // } else if storeSkusLen < jdapi.MaxStoreSkuBatchSize { // batchSize = (storeSkusLen + 1) / 2 // } else if storeSkusLen < jdapi.MaxStoreSkuBatchSize*2 { // batchSize = (storeSkusLen + 2) / 3 // } task := tasksch.NewParallelTask("syncStoreSkus京东", tasksch.NewParallelConfig().SetBatchSize(batchSize).SetIsContinueWhenError(isContinueWhenError), ctx, func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) { doWork := func(batchItemList []interface{}) (isPartialFailed bool, err error) { var skuPriceInfoList []*jdapi.SkuPriceInfo var skuVendibilityList []*jdapi.StockVendibility var skuStockList []*jdapi.SkuStock vendorSkuID4Price := "" vendorSkuID4Qty := "" stationNo := storeDetail.VendorStoreID var batchBindIDs []int for _, v := range batchItemList { storeSku := v.(*dao.StoreSkuSyncInfo) alreadyAddStock := false if storeSku.StoreSkuSyncStatus&model.SyncFlagChangedMask != 0 || storeSku.BindID == 0 || storeSku.NameID == 0 { if storeSku.BindID > 0 { batchBindIDs = append(batchBindIDs, storeSku.BindID) } if storeSku.StoreSkuSyncStatus&(model.SyncFlagDeletedMask|model.SyncFlagNewMask) != 0 || storeSku.BindID == 0 || storeSku.NameID == 0 { // 关注或取消关注 stock := &jdapi.SkuStock{ OutSkuId: utils.Int2Str(storeSku.SkuID), StockQty: model.MaxStoreSkuStockQty, } if storeSku.StoreSkuSyncStatus&model.SyncFlagDeletedMask != 0 || storeSku.DeletedAt != utils.DefaultTimeValue || storeSku.BindID == 0 || storeSku.NameID == 0 { stock.StockQty = 0 } else { alreadyAddStock = true } if stock.StockQty != 0 || !storeskulock.IsJdStoreSkuLocked(stationNo, storeSku.JdID) { vendorSkuID4Qty = storeSku.VendorSkuID skuStockList = append(skuStockList, stock) } } if storeSku.StoreSkuSyncStatus&(model.SyncFlagPriceMask|model.SyncFlagNewMask) != 0 { vendorSkuID4Price = storeSku.VendorSkuID pricePercentage := jxutils.GetPricePercentage(storeDetail.PricePercentagePackObj, int(storeSku.UnitPrice), int(storeDetail.PricePercentage)) skuPriceInfoList = append(skuPriceInfoList, &jdapi.SkuPriceInfo{ OutSkuId: utils.Int2Str(storeSku.SkuID), Price: constrainPrice(jxutils.CaculateSkuVendorPrice(int(storeSku.Price), pricePercentage, storeSku.CatPricePercentage)), }) } if storeSku.StoreSkuSyncStatus&(model.SyncFlagSaleMask|model.SyncFlagNewMask) != 0 { vendibility := &jdapi.StockVendibility{ OutSkuId: utils.Int2Str(storeSku.SkuID), DoSale: true, } if storeSku.StoreSkuStatus != model.StoreSkuBindStatusNormal { vendibility.DoSale = false } else if !alreadyAddStock { // 如果是设置可售则自动将库存加满 stock := &jdapi.SkuStock{ OutSkuId: utils.Int2Str(storeSku.SkuID), StockQty: model.MaxStoreSkuStockQty, } vendorSkuID4Qty = storeSku.VendorSkuID skuStockList = append(skuStockList, stock) } if vendibility.DoSale || !storeskulock.IsJdStoreSkuLocked(stationNo, storeSku.JdID) { skuVendibilityList = append(skuVendibilityList, vendibility) } } } } syncMask := 0 errList := []error{} if globals.EnableJdStoreWrite { // todo 以下可以优化为并行操作 globals.SugarLogger.Debug(utils.Format4Output(skuVendibilityList, false), utils.Format4Output(skuPriceInfoList, false), utils.Format4Output(skuStockList, false)) if len(skuVendibilityList) > 0 { if _, err = api.JdAPI.BatchUpdateVendibility("", stationNo, skuVendibilityList, ctx.GetUserName()); err == nil { syncMask |= model.SyncFlagSaleMask } else { if !isPartialFailed { isPartialFailed = isErrPartialFailed(err) } errList = append(errList, err) } } if (err == nil || isContinueWhenError) && len(skuStockList) > 0 { if len(skuStockList) == 1 { err = api.JdAPI.UpdateCurrentQty(stationNo, utils.Str2Int64WithDefault(vendorSkuID4Qty, 0), skuStockList[0].StockQty) } else { _, err = api.JdAPI.BatchUpdateCurrentQtys("", stationNo, skuStockList, ctx.GetUserName()) } if err == nil { syncMask |= model.SyncFlagNewMask | model.SyncFlagDeletedMask } else { if !isPartialFailed { isPartialFailed = isErrPartialFailed(err) } errList = append(errList, err) } } if (err == nil || isContinueWhenError) && len(skuPriceInfoList) > 0 { if len(skuPriceInfoList) == 1 { _, err = api.JdAPI.UpdateStationPrice(utils.Str2Int64WithDefault(vendorSkuID4Price, 0), stationNo, skuPriceInfoList[0].Price) } else { _, err = api.JdAPI.UpdateVendorStationPrice("", stationNo, skuPriceInfoList) } if err == nil { syncMask |= model.SyncFlagPriceMask } else { if !isPartialFailed { isPartialFailed = isErrPartialFailed(err) } // errList = append(errList, partner.NewErrorCode(err.Error(), partner.ErrCodeChangePriceFailed, model.VendorIDJD)) errList = append(errList, err) } } } if len(errList) == 0 { syncMask = -1 } if syncMask != 0 && len(batchBindIDs) > 0 { // db := dao.GetDB() // 多线程问题 sql := ` 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) } } if len(errList) == 1 { err = errList[0] } else if len(errList) > 1 { err = fmt.Errorf("%v", errList) } return isPartialFailed, err } isErrPartialFailed, err := doWork(batchItemList) if isErrPartialFailed && len(batchItemList) > 1 { for _, v := range batchItemList { doWork([]interface{}{v}) } } return nil, err }, storeSkus) tasksch.HandleTask(task, parentTask, false).Run() if !isAsync { _, err = task.GetResult(0) } return task.ID, err } func isErrPartialFailed(err error) bool { if errExt, ok := err.(*utils.ErrorWithCode); ok && errExt.Code() == jdapi.ResponseInnerCodePartialFailed { return true } return false } 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() _, err = dao.SetStoreSkuSyncStatus(db, model.VendorIDJD, []int{storeID}, nil, model.SyncFlagStoreSkuOnlyMask) if err != nil { return "", err } storeSkus, err := dao.GetFullStoreSkus(db, model.VendorIDJD, storeID) if err != nil { return "", err } return p.syncStoreSkus(ctx, parentTask, db, storeID, storeSkus, isAsync, isContinueWhenError) } func constrainPrice(price int) int { if price <= 0 { price = 1 } return price } func (p *PurchaseHandler) GetStoresSku(ctx *jxcontext.Context, parentTask tasksch.ITask, storeIDs []int) (storeSkuList []*model.StoreSkuBind, err error) { db := dao.GetDB() skuList, err := dao.GetSkus(db, nil, nil, []int{model.SkuStatusNormal}, nil) if err != nil { return nil, err } var skuInfoList []*jdapi.BaseStockCenterRequest skuMap := make(map[int64]int) for _, sku := range skuList { if !jxutils.IsEmptyID(sku.JdID) { skuInfoList = append(skuInfoList, &jdapi.BaseStockCenterRequest{ SkuId: sku.JdID, }) skuMap[sku.JdID] = sku.ID } } for _, storeID := range storeIDs { storeDetail, err := dao.GetStoreDetail(db, storeID, model.VendorIDJD) if err != nil { return nil, err } for _, sku := range skuInfoList { sku.StationNo = storeDetail.VendorStoreID } task := tasksch.NewParallelTask("jd 获取京东门店商品信息", tasksch.NewParallelConfig().SetBatchSize(jdapi.MaxStoreSkuBatchSize), ctx, func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) { batchSkuInfoList := make([]*jdapi.BaseStockCenterRequest, len(batchItemList)) batchSkuList := make([]int64, len(batchItemList)) for k, v := range batchItemList { batchSkuInfoList[k] = v.(*jdapi.BaseStockCenterRequest) batchSkuList[k] = batchSkuInfoList[k].SkuId } stockInfo, err := api.JdAPI.QueryOpenUseable(batchSkuInfoList) if err != nil { return nil, err } priceInfo, err := api.JdAPI.GetStationInfoList(storeDetail.VendorStoreID, batchSkuList) if err != nil { return nil, err } var batchStoreSkuList []*model.StoreSkuBind batchStoreSkuMap := make(map[int64]*model.StoreSkuBind) for _, v := range stockInfo { if v.UsableQty > 0 { batchSku := &model.StoreSkuBind{ StoreID: storeID, SkuID: skuMap[v.SkuID], } if v.Vendibility == 0 { batchSku.Status = model.SkuStatusNormal } else { batchSku.Status = model.SkuStatusDontSale } batchStoreSkuMap[v.SkuID] = batchSku batchStoreSkuList = append(batchStoreSkuList, batchSku) } } for _, v := range priceInfo { if storeSku := batchStoreSkuMap[v.SkuID]; storeSku != nil { storeSku.Price = int(v.Price) } } return batchStoreSkuList, err }, skuInfoList) tasksch.AddChild(parentTask, task).Run() result, err := task.GetResult(0) if err != nil { return nil, err } for _, v := range result { storeSkuList = append(storeSkuList, v.(*model.StoreSkuBind)) } } return storeSkuList, err } func (p *PurchaseHandler) SyncStoreProducts(ctx *jxcontext.Context, parentTask tasksch.ITask, storeID int, skuIDs []int, isAsync, isContinueWhenError bool) (hint string, err error) { globals.SugarLogger.Debugf("jd SyncStoreProducts, storeID:%d", storeID) db := dao.GetDB() storeDetail, err := dao.GetStoreDetail(db, storeID, model.VendorIDJD) if err != nil { return "", err } storeSkuList, err := dao.GetStoreSkus2(db, model.VendorIDJD, storeID, skuIDs, false) if err != nil { return "", err } task := tasksch.NewParallelTask("SyncStoreProducts京东", tasksch.NewParallelConfig().SetIsContinueWhenError(isContinueWhenError), ctx, func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) { storeSku := batchItemList[0].(*dao.StoreSkuSyncInfo) if storeSku.VendorSkuID != "" && storeSku.StoreSkuStatus == model.SkuStatusNormal { if globals.EnableJdStoreWrite { synchronized, err2 := api.JdAPI.SyncProduct(storeDetail.VendorStoreID, storeSku.VendorSkuID) if err = err2; err == nil && synchronized { retVal = []int{1} } } else { retVal = []int{1} } } return retVal, err }, storeSkuList) tasksch.HandleTask(task, parentTask, true).Run() if !isAsync { result, err2 := task.GetResult(0) if err = err2; err == nil { hint = utils.Int2Str(len(result)) } } else { hint = task.GetID() } return hint, err }