liulei/jx-callback
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

- 特殊处理同步价格至平台出错，较大改动

Browse Source
This commit is contained in:
gazebo
2019-06-06 15:01:48 +08:00
parent 89451ede5f
commit a66d69a265
10 changed files with 214 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
business/jxstore/cms/store_sku.go
View File

@@ -1586,7 +1586,7 @@ func GetVendorStoreSkusInfo(ctx *jxcontext.Context, storeID int, vendorIDs, skuI
}
}
return nil, err
})
}, true)
if err != nil {
skuVendorMap = nil
}

94
business/jxstore/cms/sync.go
View File

@@ -4,6 +4,7 @@ import (
"errors"
"fmt"
"reflect"
"strings"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/jxutils"
@@ -213,7 +214,7 @@ func (v *VendorSync) SyncStore(ctx *jxcontext.Context, db *dao.DaoDB, vendorID,
vendorID,
}
}
hint, err = v.LoopStoresMap(ctx, db, fmt.Sprintf("同步门店信息:%d", storeID), isAsync, false, vendorIDs, []int{storeID}, func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (interface{}, error) {
hint, err = v.LoopStoresMap(ctx, db, fmt.Sprintf("同步门店信息:%d", storeID), isAsync, false, vendorIDs, []int{storeID}, func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (resultList interface{}, err error) {
loopMapInfo := batchItemList[0].(*LoopStoreMapInfo)
handler := v.GetStoreHandler(loopMapInfo.VendorID)
if len(loopMapInfo.StoreMapList) > 1 {
@@ -229,16 +230,19 @@ func (v *VendorSync) SyncStore(ctx *jxcontext.Context, db *dao.DaoDB, vendorID,
return resultList, err
}, loopMapInfo.StoreMapList)
t.AddChild(loopStoreTask).Run()
return loopStoreTask.GetResult(0)
resultList, err = loopStoreTask.GetResult(0)
} else {
storeMap := loopMapInfo.StoreMapList[0]
if err = handler.UpdateStore(db, storeMap.StoreID, userName); err == nil {
storeMap.SyncStatus = 0
_, err = dao.UpdateEntity(db, storeMap, model.FieldSyncStatus)
}
if err == nil {
resultList = []interface{}{1}
}
}
storeMap := loopMapInfo.StoreMapList[0]
if err = handler.UpdateStore(db, storeMap.StoreID, userName); err == nil {
storeMap.SyncStatus = 0
_, err = dao.UpdateEntity(db, storeMap, model.FieldSyncStatus)
}
err = jxutils.AddVendorInfo2Err(err, loopMapInfo.VendorID)
return []interface{}{1}, err
})
return resultList, partner.AddVendorInfo2Err(err, loopMapInfo.VendorID)
}, true)
return hint, makeSyncError(err)
}
@@ -378,13 +382,12 @@ func (v *VendorSync) SyncStoresCategory(ctx *jxcontext.Context, db *dao.DaoDB, v
}, len(loopMapInfo.StoreMapList))
t.AddChild(loopStoreTask).Run()
_, err = loopStoreTask.GetResult(0)
return nil, err
} else {
_, err = handler.SyncStoreCategory(ctx, t, loopMapInfo.StoreMapList[0].StoreID, false)
}
_, err = handler.SyncStoreCategory(ctx, t, loopMapInfo.StoreMapList[0].StoreID, false)
err = jxutils.AddVendorInfo2Err(err, loopMapInfo.VendorID)
}
return nil, err
})
return nil, partner.AddVendorInfo2Err(err, loopMapInfo.VendorID)
}, isContinueWhenError)
return hint, makeSyncError(err)
}
@@ -392,7 +395,7 @@ func (v *VendorSync) SyncStoresCategory(ctx *jxcontext.Context, db *dao.DaoDB, v
func (v *VendorSync) SyncStoresSkus(ctx *jxcontext.Context, db *dao.DaoDB, vendorIDs []int, storeIDs []int, skuIDs []int, isForce, isAsync, isContinueWhenError bool) (hint string, err error) {
globals.SugarLogger.Debug("SyncStoresSkus")
isManageIt := len(storeIDs) != 1 || len(skuIDs) == 0 || len(skuIDs) > 8
hint, err = v.LoopStoresMap(ctx, db, fmt.Sprintf("同步门店商品信息:%v", storeIDs), isAsync, isManageIt, vendorIDs, storeIDs,
task, hint, err := v.LoopStoresMap2(ctx, db, fmt.Sprintf("同步门店商品信息:%v", storeIDs), isAsync, isManageIt, vendorIDs, storeIDs,
func(t *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (interface{}, error) {
loopMapInfo := batchItemList[0].(*LoopStoreMapInfo)
if handler := v.GetStoreHandler(loopMapInfo.VendorID); handler != nil {
@@ -413,14 +416,26 @@ func (v *VendorSync) SyncStoresSkus(ctx *jxcontext.Context, db *dao.DaoDB, vendo
}, len(loopMapInfo.StoreMapList))
t.AddChild(loopStoreTask).Run()
_, err = loopStoreTask.GetResult(0)
return nil, err
} else {
_, err = handler.SyncStoreSkus(ctx, t, loopMapInfo.StoreMapList[0].StoreID, skuIDs, false, isContinueWhenError)
}
_, err = handler.SyncStoreSkus(ctx, t, loopMapInfo.StoreMapList[0].StoreID, skuIDs, false, isContinueWhenError)
err = jxutils.AddVendorInfo2Err(err, loopMapInfo.VendorID)
}
return nil, err
})
return hint, makeSyncError(err)
return nil, partner.AddVendorInfo2Err(err, loopMapInfo.VendorID)
}, isContinueWhenError)
if vendorErr := partner.IsErrChangePriceFailed(task.GetOriginalErr()); vendorErr != nil {
platformList := make([]string, len(task.GetDetailErrList()))
for k, v := range task.GetDetailErrList() {
if vendorErr := partner.IsErrVendorError(v); vendorErr != nil {
platformList[k] = model.VendorChineseNames[vendorErr.VendorID()]
} else {
platformList[k] = "未知"
}
}
err = fmt.Errorf("修改价格失败\n失败平台：%s", strings.Join(platformList, ","))
} else {
err = makeSyncError(err)
}
return hint, err
}
func (v *VendorSync) FullSyncStoresSkus(ctx *jxcontext.Context, db *dao.DaoDB, vendorIDs []int, storeIDs []int, isAsync, isContinueWhenError bool) (hint string, err error) {
@@ -443,13 +458,12 @@ func (v *VendorSync) FullSyncStoresSkus(ctx *jxcontext.Context, db *dao.DaoDB, v
}, len(loopMapInfo.StoreMapList))
t.AddChild(loopStoreTask).Run()
_, err = loopStoreTask.GetResult(0)
return nil, err
} else {
_, err = handler.FullSyncStoreSkus(ctx, t, loopMapInfo.StoreMapList[0].StoreID, false, isContinueWhenError)
}
_, err = handler.FullSyncStoreSkus(ctx, t, loopMapInfo.StoreMapList[0].StoreID, false, isContinueWhenError)
err = jxutils.AddVendorInfo2Err(err, loopMapInfo.VendorID)
}
return nil, err
})
return nil, partner.AddVendorInfo2Err(err, loopMapInfo.VendorID)
}, isContinueWhenError)
return hint, makeSyncError(err)
}
@@ -468,17 +482,16 @@ func (v *VendorSync) DeleteRemoteStoreSkus(ctx *jxcontext.Context, db *dao.DaoDB
}, len(loopMapInfo.StoreMapList))
t.AddChild(loopStoreTask).Run()
_, err = loopStoreTask.GetResult(0)
return nil, err
} else {
_, err = handler.DeleteRemoteStoreSkus(ctx, t, loopMapInfo.StoreMapList[0].StoreID, false, isContinueWhenError)
}
_, err = handler.DeleteRemoteStoreSkus(ctx, t, loopMapInfo.StoreMapList[0].StoreID, false, isContinueWhenError)
err = jxutils.AddVendorInfo2Err(err, loopMapInfo.VendorID)
}
return nil, err
})
return nil, partner.AddVendorInfo2Err(err, loopMapInfo.VendorID)
}, isContinueWhenError)
return hint, makeSyncError(err)
}
func (v *VendorSync) LoopStoresMap(ctx *jxcontext.Context, db *dao.DaoDB, taskName string, isAsync, isManageIt bool, vendorIDs []int, storeIDs []int, handler tasksch.WorkFunc) (hint string, err error) {
func (v *VendorSync) LoopStoresMap2(ctx *jxcontext.Context, db *dao.DaoDB, taskName string, isAsync, isManageIt bool, vendorIDs []int, storeIDs []int, handler tasksch.WorkFunc, isContinueWhenError bool) (task tasksch.ITask, hint string, err error) {
sql := `
SELECT t1.*
FROM store_map t1
@@ -498,11 +511,11 @@ func (v *VendorSync) LoopStoresMap(ctx *jxcontext.Context, db *dao.DaoDB, taskNa
sql += " ORDER BY t1.store_id, t1.vendor_id"
var storeMapList []*model.StoreMap
if err = dao.GetRows(db, &storeMapList, sql, sqlParams...); err != nil {
return "", err
return nil, "", err
}
if len(storeMapList) == 0 {
return "", nil
return nil, "", nil
}
vendorStoreMap := make(map[int][]*model.StoreMap)
for _, v := range storeMapList {
@@ -520,16 +533,21 @@ func (v *VendorSync) LoopStoresMap(ctx *jxcontext.Context, db *dao.DaoDB, taskNa
if len(loopInfoList) == 1 {
taskName = fmt.Sprintf("%s,处理平台%s", taskName, model.VendorChineseNames[loopInfoList[0].VendorID])
}
task := tasksch.NewParallelTask(taskName, tasksch.NewParallelConfig().SetIsContinueWhenError(true), ctx, handler, loopInfoList)
task = tasksch.NewParallelTask(taskName, tasksch.NewParallelConfig().SetIsContinueWhenError(true), ctx, handler, loopInfoList)
tasksch.HandleTask(task, nil, isManageIt).Run()
if !isAsync {
if _, err = task.GetResult(0); err == nil {
hint = "1" // todo 暂时这样
}
} else {
hint = task.ID
hint = task.GetID()
}
return hint, makeSyncError(err)
return task, hint, err
}
func (v *VendorSync) LoopStoresMap(ctx *jxcontext.Context, db *dao.DaoDB, taskName string, isAsync, isManageIt bool, vendorIDs []int, storeIDs []int, handler tasksch.WorkFunc, isContinueWhenError bool) (hint string, err error) {
_, hint, err = v.LoopStoresMap2(ctx, db, taskName, isAsync, isManageIt, vendorIDs, storeIDs, handler, isContinueWhenError)
return hint, err
}
func (v *VendorSync) LoopMultiStoresVendors(ctx *jxcontext.Context, db *dao.DaoDB, taskName string, isAsync bool, userName string, handler tasksch.WorkFunc) (hint string, err error) {

7
business/jxutils/jxutils_cms.go
View File

@@ -374,10 +374,3 @@ func GuessVendorIDFromVendorStoreID(vendorStoreID int64) (vendorID int) {
func GetVendorName(vendorID int) (vendorName string) {
return model.VendorChineseNames[vendorID]
}
func AddVendorInfo2Err(inErr error, vendorID int) (outErr error) {
if inErr != nil {
outErr = fmt.Errorf("处理平台%s, %s", model.VendorChineseNames[vendorID], inErr.Error())
}
return outErr
}

10
business/jxutils/tasksch/parallel_task.go
View File

@@ -134,13 +134,13 @@ func (task *ParallelTask) Run() {
}
} else {
globals.SugarLogger.Infof("ParallelTask.Run %s, subtask(job:%s, params:%s) result:%v, failed with error:%v", task.Name, utils.Format4Output(job, true), utils.Format4Output(task.params, true), result, err)
task.locker.Lock()
task.detailErrList = append(task.detailErrList, err)
task.locker.Unlock()
if !task.IsContinueWhenError { // 出错
chanRetVal = err
goto end
}
task.locker.Lock()
task.detailErrMsgList = append(task.detailErrMsgList, err.Error())
task.locker.Unlock()
}
}
}
@@ -188,8 +188,12 @@ func (task *ParallelTask) Run() {
}
}
if taskErr != nil {
task.OriginalErr = taskErr
task.Err = NewTaskError(task.Name, taskErr)
} else {
if len(task.detailErrList) > 0 {
task.OriginalErr = task.detailErrList[0]
}
task.Err = task.buildTaskErrFromDetail()
}
task.Result = taskResult

10
business/jxutils/tasksch/sequence_task.go
View File

@@ -44,13 +44,13 @@ func (task *SeqTask) Run() {
})
task.finishedOneJob(1, err)
if taskErr = err; taskErr != nil {
task.locker.Lock()
task.detailErrList = append(task.detailErrList, err)
task.locker.Unlock()
globals.SugarLogger.Infof("SeqTask.Run %s step:%d failed with error:%v", task.Name, i, err)
if !task.IsContinueWhenError {
break
}
task.locker.Lock()
task.detailErrMsgList = append(task.detailErrMsgList, err.Error())
task.locker.Unlock()
} else if result != nil {
taskResult = append(taskResult, utils.Interface2Slice(result)...)
}
@@ -68,8 +68,12 @@ func (task *SeqTask) Run() {
}
}
if taskErr != nil {
task.OriginalErr = taskErr
task.Err = NewTaskError(task.Name, taskErr)
} else {
if len(task.detailErrList) > 0 {
task.OriginalErr = task.detailErrList[0]
}
task.Err = task.buildTaskErrFromDetail()
}
task.Result = taskResult

44
business/jxutils/tasksch/task.go
View File

@@ -57,6 +57,8 @@ type ITask interface {
AddChild(task ITask) ITask
GetChildren() TaskList
SetParent(parentTask ITask)
GetOriginalErr() error
GetDetailErrList() []error
json.Marshaler
}
@@ -105,15 +107,17 @@ type BaseTask struct {
NoticeMsg string `json:"noticeMsg"`
Result []interface{} `json:"-"`
Children TaskList `json:"children"`
Err error `json:"err"`
Result []interface{} `json:"-"`
Children TaskList `json:"children"`
Err error `json:"err"`
OriginalErr error `json:"-"`
detailErrMsgList []string
finishChan chan struct{}
C <-chan struct{} `json:"-"`
params []interface{}
quitChan chan int
detailErrList []error
finishChan chan struct{}
C <-chan struct{} `json:"-"`
params []interface{}
quitChan chan int
locker sync.RWMutex
parent ITask
@@ -163,7 +167,7 @@ func (t *BaseTask) GetID() string {
func (t *BaseTask) GetResult(duration time.Duration) (retVal []interface{}, err error) {
if t.GetStatus() >= TaskStatusEndBegin {
return t.Result, t.Err
return t.Result, t.OriginalErr
}
if duration == 0 {
duration = time.Hour * 10000 // duration为0表示无限等待
@@ -173,7 +177,7 @@ func (t *BaseTask) GetResult(duration time.Duration) (retVal []interface{}, err
case <-t.finishChan:
t.isGetResultCalled = true
timer.Stop()
return t.Result, t.Err
return t.Result, t.OriginalErr
case <-timer.C:
}
return nil, ErrTaskNotFinished
@@ -271,6 +275,18 @@ func (t *BaseTask) GetNoticeMsg() string {
return t.NoticeMsg
}
func (t *BaseTask) GetOriginalErr() error {
t.locker.RLock()
defer t.locker.RUnlock()
return t.OriginalErr
}
func (t *BaseTask) GetDetailErrList() []error {
t.locker.RLock()
defer t.locker.RUnlock()
return t.detailErrList
}
func AddChild(parentTask ITask, task ITask) ITask {
if parentTask != nil {
return parentTask.AddChild(task)
@@ -365,8 +381,12 @@ func (t *BaseTask) setStatus(status int) {
}
func (t *BaseTask) buildTaskErrFromDetail() (err error) {
if len(t.detailErrMsgList) > 0 {
return NewTaskError(t.Name, fmt.Errorf("总共:%d, 失败:%d, 详情:\n%s", t.TotalItemCount, t.FailedItemCount, strings.Join(t.detailErrMsgList, "\n")))
if len(t.detailErrList) > 0 {
strList := make([]string, len(t.detailErrList))
for k, v := range t.detailErrList {
strList[k] = v.Error()
}
return NewTaskError(t.Name, fmt.Errorf("总共:%d, 失败:%d, 详情:\n%s", t.TotalItemCount, t.FailedItemCount, strings.Join(strList, "\n")))
}
return nil
}

89
business/partner/partner_err.go Normal file
View File

@@ -0,0 +1,89 @@
package partner
import (
"fmt"
"git.rosy.net.cn/jx-callback/business/model"
)
const (
ErrCodeUnknown = 1
ErrCodeChangePriceFailed = 100
)
type ErrorWithCode struct {
errMsg string
intCode int
vendorID int
storeID int
skuID int
}
func NewErrorCode(errMsg string, code, vendorID int) *ErrorWithCode {
retVal := &ErrorWithCode{
errMsg: errMsg,
intCode: code,
vendorID: vendorID,
}
return retVal
}
func (e *ErrorWithCode) SetStoreID(storeID int) {
e.storeID = storeID
}
func (e *ErrorWithCode) SetSkuID(skuID int) {
e.skuID = skuID
}
func (e *ErrorWithCode) Error() string {
return fmt.Sprintf("平台:%s, code:%d, %s", model.VendorChineseNames[e.VendorID()], e.intCode, e.errMsg)
}
func (e *ErrorWithCode) String() string {
return e.Error()
}
func (e *ErrorWithCode) Code() int {
return e.intCode
}
func (e *ErrorWithCode) ErrMsg() string {
return e.errMsg
}
func (e *ErrorWithCode) VendorID() int {
return e.vendorID
}
func (e *ErrorWithCode) StoreID() int {
return e.storeID
}
func (e *ErrorWithCode) SkuID() int {
return e.skuID
}
func IsErrChangePriceFailed(err error) *ErrorWithCode {
if vendorErr, ok := err.(*ErrorWithCode); ok && vendorErr.Code() == ErrCodeChangePriceFailed {
return vendorErr
}
return nil
}
func IsErrVendorError(err error) *ErrorWithCode {
if vendorErr, ok := err.(*ErrorWithCode); ok {
return vendorErr
}
return nil
}
func AddVendorInfo2Err(inErr error, vendorID int) (outErr error) {
outErr = inErr
if inErr != nil {
if IsErrVendorError(inErr) == nil {
outErr = NewErrorCode(inErr.Error(), ErrCodeUnknown, vendorID)
}
}
return outErr
}

3
business/partner/purchase/ebai/store_sku.go
View File

@@ -13,6 +13,7 @@ import (
"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/globals"
"git.rosy.net.cn/jx-callback/globals/api"
)
@@ -306,6 +307,8 @@ func (p *PurchaseHandler) SyncStoreSkus(ctx *jxcontext.Context, parentTask tasks
storeSku.EbaiSyncStatus = 0
_, err = dao.UpdateEntity(nil, &storeSku.StoreSkuBind, updateFields...)
}
} else if isErrModifyPrice(err) {
err = partner.NewErrorCode(err.Error(), partner.ErrCodeChangePriceFailed, model.VendorIDEBAI)
}
return nil, err
}, storeSkuInfoList)

4
business/partner/purchase/jd/store_sku.go
View File

@@ -3,6 +3,8 @@ package jd
import (
"fmt"
"git.rosy.net.cn/jx-callback/business/partner"
"git.rosy.net.cn/baseapi/platformapi/jdapi"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/jxutils"
@@ -107,7 +109,7 @@ func (p *PurchaseHandler) syncStoreSkus(ctx *jxcontext.Context, parentTask tasks
if _, err = api.JdAPI.UpdateVendorStationPrice("", stationNo, skuPriceInfoList); err == nil {
syncMask |= model.SyncFlagPriceMask
} else {
errList = append(errList, err)
errList = append(errList, partner.NewErrorCode(err.Error(), partner.ErrCodeChangePriceFailed, model.VendorIDJD))
}
}
}

17
business/partner/purchase/mtwm/store_sku.go
View File

@@ -13,6 +13,7 @@ import (
"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/globals"
"git.rosy.net.cn/jx-callback/globals/api"
)
@@ -335,7 +336,7 @@ func (p *PurchaseHandler) SyncStoreSkus(ctx *jxcontext.Context, parentTask tasks
}
}
if err != nil {
if extErr, ok := err.(*utils.ErrorWithCode); ok && extErr.IntCode() == 1 && strings.Index(extErr.ErrMsg(), "折扣商品原价不允许修改") >= 0 {
if isErrModifyPrice(err) {
shouldCallSellStatus = true
}
}
@@ -355,6 +356,8 @@ func (p *PurchaseHandler) SyncStoreSkus(ctx *jxcontext.Context, parentTask tasks
if len(updateFields) > 0 {
_, err = dao.UpdateEntity(db, storeSkuBind, updateFields...)
}
} else if isErrModifyPrice(err) {
err = partner.NewErrorCode(err.Error(), partner.ErrCodeChangePriceFailed, model.VendorIDEBAI)
}
return nil, err
}, skus)
@@ -371,6 +374,18 @@ func (p *PurchaseHandler) SyncStoreSkus(ctx *jxcontext.Context, parentTask tasks
return hint, err
}
func isErrModifyPrice(err error) bool {
if errExt, ok := err.(*utils.ErrorWithCode); ok && errExt.IntCode() == 1 {
for _, v := range []string{
"折扣商品原价不允许修改",
} {
if strings.Index(errExt.ErrMsg(), v) >= 0 {
return true
}
}
}
return false
}
func (p *PurchaseHandler) RefreshStoresAllSkusID(ctx *jxcontext.Context, parentTask tasksch.ITask, isAsync bool, storeIDs []int) (hint string, err error) {
return hint, err
}