330 lines
12 KiB
Go
330 lines
12 KiB
Go
package cms
|
||
|
||
import (
|
||
"fmt"
|
||
"io"
|
||
"time"
|
||
|
||
"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/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/business/partner/putils"
|
||
"git.rosy.net.cn/jx-callback/globals/api/apimanager"
|
||
"github.com/360EntSecGroup-Skylar/excelize"
|
||
)
|
||
|
||
func storeOrSkuMap2List(intStrMap map[int]string) (ids []int) {
|
||
for k, v := range intStrMap {
|
||
if v != "" {
|
||
ids = append(ids, k)
|
||
}
|
||
}
|
||
return ids
|
||
}
|
||
|
||
func getStoreSkus(db *dao.DaoDB, storeID int, skuIDs []int) (skus []*dao.StoreSkuSyncInfo, err error) {
|
||
sql := `
|
||
SELECT
|
||
t1.id bind_id, t1.sku_id, t1.price, t1.unit_price, t1.status store_sku_status,
|
||
t1.jd_sync_status sku_sync_status, t1.jd_price vendor_price, t1.jd_lock_time lock_time,
|
||
t1.store_id, t1.deleted_at bind_deleted_at,t1.status_sale_begin,t1.status_sale_end,
|
||
t2.*,
|
||
t3.id name_id, t3.prefix, t3.name, t3.unit, t3.upc, t3.status name_status, t3.ex_prefix, t3.ex_prefix_begin, t3.ex_prefix_end
|
||
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 = ?*/
|
||
WHERE 1 = 1
|
||
AND t1.store_id = ?
|
||
AND t1.deleted_at = ?
|
||
AND t1.sku_id IN ( ` + dao.GenQuestionMarks(len(skuIDs)) + `)
|
||
ORDER BY t1.price`
|
||
sqlParams := []interface{}{
|
||
utils.DefaultTimeValue,
|
||
utils.DefaultTimeValue,
|
||
storeID,
|
||
utils.DefaultTimeValue,
|
||
skuIDs,
|
||
}
|
||
if err = dao.GetRows(db, &skus, sql, sqlParams...); err != nil {
|
||
return nil, err
|
||
}
|
||
return skus, err
|
||
}
|
||
|
||
func SyncStoreSku4FakeJD(ctx *jxcontext.Context, parentTask tasksch.ITask, storeID int, vendorStoreID string, inSkuMap map[int]string, isContinueWhenError bool) (err error) {
|
||
vendorID := model.VendorIDJD
|
||
db := dao.GetDB()
|
||
storeDetail, err := dao.GetStoreDetail(db, storeID, vendorID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
storeDetail.VendorOrgCode = apimanager.FakeJdOrgCode
|
||
skuIDs := storeOrSkuMap2List(inSkuMap)
|
||
skus, err := getStoreSkus(db, storeID, skuIDs)
|
||
if err != nil || len(skus) == 0 {
|
||
return err
|
||
}
|
||
formalizeStoreSkuList(skus)
|
||
storeSkuHandler := partner.GetPurchasePlatformFromVendorID(vendorID).(partner.IPurchasePlatformStoreSkuHandler)
|
||
|
||
var (
|
||
stockList, onlineList, offlineList, priceList []*partner.StoreSkuInfo
|
||
)
|
||
skuMap := make(map[*partner.StoreSkuInfo]*dao.StoreSkuSyncInfo)
|
||
now := jxutils.OperationTime2HourMinuteFormat(time.Now())
|
||
var failedList []*partner.StoreSkuInfoWithErr
|
||
for _, sku := range skus {
|
||
sku.SkuSyncStatus = model.SyncFlagSaleMask | model.SyncFlagPriceMask | model.SyncFlagStockMask
|
||
sku.VendorSkuID = inSkuMap[sku.SkuID]
|
||
sku.SkuID = int(utils.Str2Int64(sku.VendorSkuID)) // skuID与vendorID一样
|
||
sku.VendorOrgCode = apimanager.FakeJdOrgCode
|
||
|
||
sku.VendorPrice = 0
|
||
sku.MergedStatus = MergeSkuSaleStatusWithStoreOpTime(sku, storeDetail, now)
|
||
var bareSku *partner.StoreSkuInfo
|
||
if isStoreSkuSyncNeedDelete(sku) {
|
||
if !dao.IsVendorThingIDEmpty(sku.VendorSkuID) {
|
||
bareSku = storeSkuSyncInfo2Bare(sku)
|
||
stockList = append(stockList, bareSku)
|
||
} else {
|
||
// updateItems = append(updateItems, sku2Update(vendorID, sku, model.SyncFlagDeletedMask))
|
||
}
|
||
} else if model.IsSyncStatusNew(sku.SkuSyncStatus) {
|
||
calVendorPrice4StoreSku(sku, storeDetail.PricePercentagePackObj, int(storeDetail.PricePercentage))
|
||
if dao.IsVendorThingIDEmpty(sku.VendorSkuID) {
|
||
// todo 多平台商品库没有正常创建，直接跳过
|
||
} else {
|
||
sku.SkuSyncStatus |= model.SyncFlagSaleMask | model.SyncFlagPriceMask
|
||
bareSku = storeSkuSyncInfo2Bare(sku)
|
||
stockList = append(stockList, bareSku)
|
||
priceList = append(priceList, bareSku)
|
||
if sku.MergedStatus == model.SkuStatusNormal {
|
||
onlineList = append(onlineList, bareSku)
|
||
} else {
|
||
offlineList = append(offlineList, bareSku)
|
||
}
|
||
}
|
||
} else {
|
||
if dao.IsVendorThingIDEmpty(sku.VendorSkuID) {
|
||
// err = fmt.Errorf("门店:%d，修改没有创建的商品:%d", storeID, sku.SkuID)
|
||
err = utils.NewErrorCode(fmt.Sprintf("门店:%d，修改没有创建的商品:%d", storeID, sku.SkuID), "-1", 0)
|
||
failedList = putils.GetErrMsg2FailedSingleList(nil, err, storeID, vendorID, "异常同步错误")
|
||
if parentTask == nil {
|
||
return err
|
||
}
|
||
parentTask.AddBatchErr(err)
|
||
parentTask.AddFailedList(failedList)
|
||
} else {
|
||
isAdded2Update := false
|
||
if model.IsSyncStatusPrice(sku.SkuSyncStatus) {
|
||
bareSku = storeSkuSyncInfo2Bare(calVendorPrice4StoreSku(sku, storeDetail.PricePercentagePackObj, int(storeDetail.PricePercentage)))
|
||
priceList = append(priceList, bareSku)
|
||
}
|
||
if !isAdded2Update {
|
||
if model.IsSyncStatusSale(sku.SkuSyncStatus) {
|
||
if bareSku == nil {
|
||
bareSku = storeSkuSyncInfo2Bare(sku)
|
||
}
|
||
if sku.MergedStatus == model.SkuStatusNormal {
|
||
onlineList = append(onlineList, bareSku)
|
||
stockList = append(stockList, bareSku)
|
||
} else {
|
||
offlineList = append(offlineList, bareSku)
|
||
// 因为京东平台以是否有库存表示是否关注，所以不论是否可售，都要设置库存
|
||
stockList = append(stockList, bareSku)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
if bareSku != nil {
|
||
skuMap[bareSku] = sku
|
||
}
|
||
}
|
||
bareSku2Sync := func(bareSkuList []*partner.StoreSkuInfo) (skuList []*dao.StoreSkuSyncInfo) {
|
||
if len(bareSkuList) > 0 {
|
||
skuList = make([]*dao.StoreSkuSyncInfo, len(bareSkuList))
|
||
for k, v := range bareSkuList {
|
||
skuList[k] = skuMap[v]
|
||
}
|
||
}
|
||
return skuList
|
||
}
|
||
isContinueWhenError2 := true
|
||
realStoreID := int(utils.Str2Int64(vendorStoreID))
|
||
task := tasksch.NewParallelTask("SyncStoreSku4FakeJD", tasksch.NewParallelConfig().SetParallelCount(1).SetIsContinueWhenError(isContinueWhenError2), ctx,
|
||
func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
|
||
step := batchItemList[0].(int)
|
||
// globals.SugarLogger.Debugf("step:%d", step)
|
||
switch step {
|
||
case 0:
|
||
for k, list := range [][]*partner.StoreSkuInfo{stockList /*, onlineList*/} {
|
||
if len(list) > 0 {
|
||
_, err = putils.FreeBatchStoreSkuInfo("更新门店商品库存", func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) {
|
||
var failedList []*partner.StoreSkuInfoWithErr
|
||
failedList, err = storeSkuHandler.UpdateStoreSkusStock(ctx, storeDetail.VendorOrgCode, realStoreID, vendorStoreID, batchedStoreSkuList)
|
||
if len(failedList) > 0 {
|
||
task.AddFailedList(failedList)
|
||
}
|
||
successList := putils.UnselectStoreSkuListByVendorSkuIDs(batchedStoreSkuList, GetVendorSkuIDList(failedList))
|
||
if k == 0 && len(successList) > 0 {
|
||
updateStoreSku(dao.GetDB(), vendorID, bareSku2Sync(successList), model.SyncFlagStockMask)
|
||
}
|
||
return nil, len(successList), err
|
||
}, ctx, task, list, storeSkuHandler.GetStoreSkusBatchSize(partner.FuncUpdateStoreSkusStock), isContinueWhenError2)
|
||
}
|
||
}
|
||
case 1, 2:
|
||
statusList := onlineList
|
||
status := model.SkuStatusNormal
|
||
name := "可售门店商品"
|
||
if step == 2 {
|
||
statusList = offlineList
|
||
status = model.SkuStatusDontSale
|
||
name = "不可售门店商品"
|
||
}
|
||
if len(statusList) > 0 {
|
||
_, err = putils.FreeBatchStoreSkuInfo(name, func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) {
|
||
var failedList []*partner.StoreSkuInfoWithErr
|
||
failedList, err = storeSkuHandler.UpdateStoreSkusStatus(ctx, storeDetail.VendorOrgCode, realStoreID, vendorStoreID, batchedStoreSkuList, status)
|
||
if len(failedList) > 0 {
|
||
task.AddFailedList(failedList)
|
||
}
|
||
successList := putils.UnselectStoreSkuListByVendorSkuIDs(batchedStoreSkuList, GetVendorSkuIDList(failedList))
|
||
if len(successList) > 0 {
|
||
updateStoreSku(dao.GetDB(), vendorID, bareSku2Sync(successList), model.SyncFlagSaleMask)
|
||
}
|
||
return nil, len(successList), err
|
||
}, ctx, task, statusList, storeSkuHandler.GetStoreSkusBatchSize(partner.FuncUpdateStoreSkusStatus), isContinueWhenError2)
|
||
}
|
||
case 3:
|
||
if len(priceList) > 0 {
|
||
_, err = putils.FreeBatchStoreSkuInfo("更新门店商品价格", func(task tasksch.ITask, batchedStoreSkuList []*partner.StoreSkuInfo) (result interface{}, successCount int, err error) {
|
||
var failedList []*partner.StoreSkuInfoWithErr
|
||
failedList, err = storeSkuHandler.UpdateStoreSkusPrice(ctx, storeDetail.VendorOrgCode, realStoreID, vendorStoreID, batchedStoreSkuList)
|
||
if len(failedList) > 0 {
|
||
task.AddFailedList(failedList)
|
||
}
|
||
successList := putils.UnselectStoreSkuListByVendorSkuIDs(batchedStoreSkuList, GetVendorSkuIDList(failedList))
|
||
if len(successList) > 0 {
|
||
updateStoreSku(dao.GetDB(), vendorID, bareSku2Sync(successList), model.SyncFlagPriceMask)
|
||
}
|
||
return nil, len(successList), err
|
||
}, ctx, task, priceList, storeSkuHandler.GetStoreSkusBatchSize(partner.FuncUpdateStoreSkusPrice), isContinueWhenError2)
|
||
}
|
||
}
|
||
return retVal, err
|
||
}, []int{0, 1, 2, 3})
|
||
tasksch.HandleTask(task, parentTask, true).Run()
|
||
_, err = task.GetResult(0)
|
||
return err
|
||
}
|
||
|
||
func SyncFakeJdStoreSku(ctx *jxcontext.Context, parentTask tasksch.ITask, storeMap, skuMap map[int]string, isAsync, isContinueWhenError bool) (hint string, err error) {
|
||
storeIDs := storeOrSkuMap2List(storeMap)
|
||
task := tasksch.NewParallelTask("同步假京东", nil, ctx,
|
||
func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
|
||
storeID := batchItemList[0].(int)
|
||
err = SyncStoreSku4FakeJD(ctx, task, storeID, storeMap[storeID], skuMap, isContinueWhenError)
|
||
return retVal, err
|
||
}, storeIDs)
|
||
tasksch.HandleTask(task, parentTask, true).Run()
|
||
if isAsync {
|
||
hint = task.GetID()
|
||
} else {
|
||
_, err = task.GetResult(0)
|
||
hint = utils.Int2Str(len(storeIDs))
|
||
}
|
||
return hint, err
|
||
}
|
||
|
||
func excel2FakeJdThingMap(ctx *jxcontext.Context, reader io.Reader) (thingMapList []*model.FakeJdThingMap, err error) {
|
||
xlsx, err := excelize.OpenReader(reader)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
for sheetIndex := 0; sheetIndex < xlsx.SheetCount; sheetIndex++ {
|
||
rows, err2 := xlsx.GetRows(xlsx.GetSheetName(sheetIndex + 1))
|
||
if err2 != nil {
|
||
return nil, err2
|
||
}
|
||
for rowNum, row := range rows {
|
||
thingMap := &model.FakeJdThingMap{
|
||
JxID: int(utils.Str2Int64WithDefault(row[0], 0)),
|
||
JdID: utils.Str2Int64WithDefault(row[0], 0),
|
||
}
|
||
if thingMap.JxID == 0 || thingMap.JdID == 0 {
|
||
if rowNum == 0 {
|
||
continue
|
||
} else {
|
||
break
|
||
}
|
||
}
|
||
if sheetIndex == 0 {
|
||
thingMap.ThingType = model.ThingTypeStore
|
||
} else {
|
||
thingMap.ThingType = model.ThingTypeSku
|
||
}
|
||
dao.WrapAddIDCULEntity(thingMap, ctx.GetUserName())
|
||
thingMapList = append(thingMapList, thingMap)
|
||
}
|
||
}
|
||
return thingMapList, err
|
||
}
|
||
|
||
func getFakeThingMap(ctx *jxcontext.Context, db *dao.DaoDB) (storeMap, skuMap map[int]string, err error) {
|
||
var thingMapList []*model.FakeJdThingMap
|
||
err = dao.GetRows(db, &thingMapList, "SELECT t1.* FROM fake_jd_thing_map t1")
|
||
if err == nil {
|
||
storeMap, skuMap = make(map[int]string), make(map[int]string)
|
||
for _, v := range thingMapList {
|
||
if v.ThingType == model.ThingTypeStore {
|
||
storeMap[v.JxID] = utils.Int64ToStr(v.JdID)
|
||
} else {
|
||
skuMap[v.JxID] = utils.Int64ToStr(v.JdID)
|
||
}
|
||
}
|
||
}
|
||
return storeMap, skuMap, err
|
||
}
|
||
|
||
func UploadFakeJdThingMap(ctx *jxcontext.Context, reader io.Reader, isSyncNow, isAsync, isContinueWhenError bool) (hint string, err error) {
|
||
thingMapList, err := excel2FakeJdThingMap(ctx, reader)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
|
||
db := dao.GetDB()
|
||
dao.Begin(db)
|
||
defer func() {
|
||
if r := recover(); r != nil {
|
||
dao.Rollback(db)
|
||
panic(r)
|
||
}
|
||
}()
|
||
_, err = dao.ExecuteSQL(db, "DELETE t1 FROM fake_jd_thing_map t1")
|
||
if err != nil {
|
||
dao.Rollback(db)
|
||
return "", err
|
||
}
|
||
err = dao.CreateMultiEntities(db, thingMapList)
|
||
if err != nil {
|
||
dao.Rollback(db)
|
||
return "", err
|
||
}
|
||
dao.Commit(db)
|
||
|
||
if isSyncNow {
|
||
if storeMap, skuMap, err2 := getFakeThingMap(ctx, db); err2 == nil {
|
||
hint, err = SyncFakeJdStoreSku(ctx, nil, storeMap, skuMap, isAsync, isContinueWhenError)
|
||
}
|
||
} else {
|
||
hint = utils.Int2Str(len(thingMapList))
|
||
}
|
||
return hint, err
|
||
}
|