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" ) type tStoreSkuBindExt struct { model.StoreSkuBind PricePercentage int VendorStoreID string `orm:"column(vendor_store_id)"` JdID int64 `orm:"column(jd_id)"` } // 京东到家,以有库存表示关注(认领) 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 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 = ? ` + 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 { return "", err } task := tasksch.NewParallelTask("SyncStoresSkus京东", 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 for _, v := range batchItemList { storeSku := v.(*tStoreSkuBindExt) alreadyAddStock := false if storeSku.JdSyncStatus&model.SyncFlagChangedMask != 0 { batchSkuIDs = append(batchSkuIDs, storeSku.SkuID) if storeSku.JdSyncStatus&(model.SyncFlagDeletedMask|model.SyncFlagNewMask) != 0 { // 关注或取消关注 stock := &jdapi.SkuStock{ OutSkuId: utils.Int2Str(storeSku.SkuID), StockQty: model.MaxStoreSkuStockQty, } if storeSku.DeletedAt != utils.DefaultTimeValue { stock.StockQty = 0 } else { alreadyAddStock = true } if stock.StockQty != 0 || !storeskulock.IsJdStoreSkuLocked(stationNo, storeSku.JdID) { skuStockList = append(skuStockList, stock) } } if storeSku.JdSyncStatus&(model.SyncFlagPriceMask|model.SyncFlagNewMask) != 0 { skuPriceInfoList = append(skuPriceInfoList, &jdapi.SkuPriceInfo{ OutSkuId: utils.Int2Str(storeSku.SkuID), Price: constrainPrice(jxutils.CaculateSkuVendorPrice(storeSku.Price, storeSku.PricePercentage)), }) } if storeSku.JdSyncStatus&(model.SyncFlagSaleMask|model.SyncFlagNewMask) != 0 { vendibility := &jdapi.StockVendibility{ OutSkuId: utils.Int2Str(storeSku.SkuID), DoSale: true, } if storeSku.Status != model.StoreSkuBindStatusNormal { vendibility.DoSale = false } else if !alreadyAddStock { // 如果是设置可售则自动将库存加满 stock := &jdapi.SkuStock{ OutSkuId: utils.Int2Str(storeSku.SkuID), StockQty: model.MaxStoreSkuStockQty, } 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 { errList = append(errList, err) } } if (err == nil || isContinueWhenError) && len(skuStockList) > 0 { if _, err = api.JdAPI.BatchUpdateCurrentQtys("", stationNo, skuStockList, ctx.GetUserName()); err == nil { syncMask |= model.SyncFlagNewMask | model.SyncFlagDeletedMask } else { errList = append(errList, err) } } if (err == nil || isContinueWhenError) && len(skuPriceInfoList) > 0 { if _, err = api.JdAPI.UpdateVendorStationPrice("", stationNo, skuPriceInfoList); err == nil { syncMask |= model.SyncFlagPriceMask } else { errList = append(errList, err) } } } if len(errList) == 0 { syncMask = -1 } if syncMask != 0 && len(batchSkuIDs) > 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 { errList = append(errList, err) } } if len(errList) == 1 { err = errList[0] } else if len(errList) > 1 { err = fmt.Errorf("%v", errList) } return nil, err }, storeSkus) tasksch.HandleTask(task, parentTask, false).Run() if !isAsync { _, err = task.GetResult(0) } return task.ID, err } 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 } skus, 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))), }) } 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 } func (p *PurchaseHandler) DeleteRemoteStoreSkus(ctx *jxcontext.Context, parentTask tasksch.ITask, storeID int, isAsync, isContinueWhenError bool) (hint string, err error) { return hint, err } func constrainPrice(price int) int { if price <= 0 { price = 1 } return price }