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

cms

Browse Source
This commit is contained in:
苏尹岚
2020-10-26 17:58:51 +08:00
parent bbcb8ec37c
commit 64c0bc05ed
4 changed files with 151 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files

147
business/jxutils/datares/datares.go
View File

@@ -14,7 +14,6 @@ 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"
"github.com/qiniu/api.v7/storage"
@@ -152,82 +151,82 @@ func GetDataResource(ctx *jxcontext.Context, hashCode string) (resourceURL strin
// 这个函数，可能部分平台成功，部分失败
func UploadImage2Vendors(ctx *jxcontext.Context, parentTask tasksch.ITask, dataRes *model.DataResource, imgData []byte, isAsync bool) (hint string, err error) {
var vendorIDs []int
if dataRes.EbaiURL == "" {
vendorIDs = append(vendorIDs, model.VendorIDEBAI)
}
if dataRes.MtwmURL == "" {
vendorIDs = append(vendorIDs, model.VendorIDMTWM)
}
// var vendorIDs []int
// if dataRes.EbaiURL == "" {
// vendorIDs = append(vendorIDs, model.VendorIDEBAI)
// }
// if dataRes.MtwmURL == "" {
// vendorIDs = append(vendorIDs, model.VendorIDMTWM)
// }
// if dataRes.JdsURL == "" {
// vendorIDs = append(vendorIDs, model.VendorIDJDShop)
// }
if len(vendorIDs) > 0 {
imgName := jxutils.GetShortNameFromURL(dataRes.MainURL)
task := tasksch.NewSeqTask(fmt.Sprintf("上传图片至平台1:%s,%s", dataRes.Name, dataRes.MainURL), ctx,
func(task *tasksch.SeqTask, step int, params ...interface{}) (result interface{}, err error) {
switch step {
case 0:
if imgData == nil {
if imgData, _, err = jxutils.DownloadFileByURL(dataRes.MainURL); err != nil {
return nil, err
}
}
case 1:
uploadTask := tasksch.NewParallelTask(fmt.Sprintf("上传图片至平台2:%s,%s", dataRes.Name, imgName),
tasksch.NewParallelConfig().SetIsContinueWhenError(true), ctx,
func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
vendorID := batchItemList[0].(int)
if handler := partner.GetPurchasePlatformFromVendorID(vendorID); handler != nil {
// TODO vendorOrgCode
imgHint, err2 := handler.UploadImg(ctx, "", dataRes.MainURL, imgData, imgName, int(dataRes.UseType))
if err = err2; err == nil {
retVal = [][]interface{}{
[]interface{}{
vendorID,
imgHint,
},
}
}
}
return retVal, err
}, vendorIDs)
tasksch.HandleTask(uploadTask, task, false).Run()
resultList, err2 := uploadTask.GetResult(0)
err = err2
if len(resultList) > 0 {
db := dao.GetDB()
for _, v := range resultList {
result := v.([]interface{})
vendorID := result[0].(int)
imgHint := result[1].(string)
updateField := ""
if vendorID == model.VendorIDEBAI {
dataRes.EbaiURL = imgHint
updateField = "EbaiURL"
} else if vendorID == model.VendorIDMTWM {
dataRes.MtwmURL = imgHint
updateField = "MtwmURL"
}
// else if vendorID == model.VendorIDJDShop {
// dataRes.JdsURL = imgHint
// updateField = "JdsURL"
// }
dao.UpdateEntity(db, dataRes, updateField)
}
}
}
return result, err
}, 2)
tasksch.HandleTask(task, parentTask, false).Run()
if !isAsync {
if _, err = task.GetResult(0); err == nil {
hint = "1"
}
} else {
hint = task.GetID()
}
}
// if len(vendorIDs) > 0 {
// imgName := jxutils.GetShortNameFromURL(dataRes.MainURL)
// task := tasksch.NewSeqTask(fmt.Sprintf("上传图片至平台1:%s,%s", dataRes.Name, dataRes.MainURL), ctx,
// func(task *tasksch.SeqTask, step int, params ...interface{}) (result interface{}, err error) {
// switch step {
// case 0:
// if imgData == nil {
// if imgData, _, err = jxutils.DownloadFileByURL(dataRes.MainURL); err != nil {
// return nil, err
// }
// }
// case 1:
// uploadTask := tasksch.NewParallelTask(fmt.Sprintf("上传图片至平台2:%s,%s", dataRes.Name, imgName),
// tasksch.NewParallelConfig().SetIsContinueWhenError(true), ctx,
// func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
// vendorID := batchItemList[0].(int)
// if handler := partner.GetPurchasePlatformFromVendorID(vendorID); handler != nil {
// // TODO vendorOrgCode
// imgHint, err2 := handler.UploadImg(ctx, "", dataRes.MainURL, imgData, imgName, int(dataRes.UseType))
// if err = err2; err == nil {
// retVal = [][]interface{}{
// []interface{}{
// vendorID,
// imgHint,
// },
// }
// }
// }
// return retVal, err
// }, vendorIDs)
// tasksch.HandleTask(uploadTask, task, false).Run()
// resultList, err2 := uploadTask.GetResult(0)
// err = err2
// if len(resultList) > 0 {
// db := dao.GetDB()
// for _, v := range resultList {
// result := v.([]interface{})
// vendorID := result[0].(int)
// imgHint := result[1].(string)
// updateField := ""
// if vendorID == model.VendorIDEBAI {
// dataRes.EbaiURL = imgHint
// updateField = "EbaiURL"
// } else if vendorID == model.VendorIDMTWM {
// dataRes.MtwmURL = imgHint
// updateField = "MtwmURL"
// }
// // else if vendorID == model.VendorIDJDShop {
// // dataRes.JdsURL = imgHint
// // updateField = "JdsURL"
// // }
// dao.UpdateEntity(db, dataRes, updateField)
// }
// }
// }
// return result, err
// }, 2)
// tasksch.HandleTask(task, parentTask, false).Run()
// if !isAsync {
// if _, err = task.GetResult(0); err == nil {
// hint = "1"
// }
// } else {
// hint = task.GetID()
// }
// }
return hint, err
}

225
business/msghub/msghub.go
View File

@@ -1,225 +0,0 @@
package msghub
import (
"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/model"
"git.rosy.net.cn/jx-callback/business/model/dao"
"git.rosy.net.cn/jx-callback/globals"
)
const (
ServerMsgRegister = "register"
ServerMsgUnregister = "unregister"
ServerMsgPing = "ping"
ServerMsgNewOrder = "newOrder"
ServerMsgFinishedPickup = "finishedPickup" // 此消息当前当成刷新订单列表及状态来使用，并不是拣货
ServerMsgKeyOrderStatusChanged = "keyOrderStatusChanged" // 重要订单状态变化
ServerMsgNewWait4ApproveAfsOrder = "newWait4ApproveAfsOrder"
ServerMsgKeyAfsOrderStatusChanged = "keyAfsOrderStatusChanged" // 重要售后单状态变化
)
const (
maxGetOrderTimeDuration = 24 * time.Hour
minPollingDuration = 1 * time.Minute
defPollingDuration = 5 * time.Minute
maxPollingDuration = 10 * time.Minute
)
var (
channelMap map[int]map[chan<- *ServerMsg]int
msgChan chan *ServerMsg
)
type MsgOp struct {
StoreID int
Chan2Listen chan<- *ServerMsg
Chan2Close chan int
}
type ServerMsg struct {
Type string
StoreID int
MsgData interface{}
}
func init() {
utils.CallFuncAsync(routinueFunc)
}
func routinueFunc() {
msgChan = make(chan *ServerMsg, 100)
channelMap = make(map[int]map[chan<- *ServerMsg]int)
for {
msg, ok := <-msgChan
if ok {
switch msg.Type {
case ServerMsgRegister:
registerMsg := msg.MsgData.(*MsgOp)
if channelMap[registerMsg.StoreID] == nil {
channelMap[registerMsg.StoreID] = make(map[chan<- *ServerMsg]int)
}
channelMap[registerMsg.StoreID][registerMsg.Chan2Listen] = 1
case ServerMsgUnregister:
registerMsg := msg.MsgData.(*MsgOp)
delete(channelMap[registerMsg.StoreID], registerMsg.Chan2Listen)
close(registerMsg.Chan2Close)
case ServerMsgNewOrder, ServerMsgFinishedPickup, ServerMsgKeyOrderStatusChanged, ServerMsgNewWait4ApproveAfsOrder, ServerMsgKeyAfsOrderStatusChanged:
globals.SugarLogger.Debugf("msghub routinueFunc, msg:%s", utils.Format4Output(msg, true))
utils.CallFuncAsync(func() {
for chan2Send := range channelMap[msg.StoreID] {
chan2Send <- msg
delete(channelMap[msg.StoreID], chan2Send)
}
})
}
}
}
}
func registerChan(storeID int, chan2Listen chan<- *ServerMsg) {
msgChan <- &ServerMsg{
Type: ServerMsgRegister,
MsgData: &MsgOp{
StoreID: storeID,
Chan2Listen: chan2Listen,
},
}
}
func unregisterChan(storeID int, chan2Listen chan<- *ServerMsg) {
chan2Close := make(chan int)
msgChan <- &ServerMsg{
Type: ServerMsgUnregister,
MsgData: &MsgOp{
StoreID: storeID,
Chan2Listen: chan2Listen,
Chan2Close: chan2Close,
},
}
<-chan2Close
}
func getPendingOrderList(storeID int, lastOrderTime time.Time, lastOrderSeqID int64) (vendorOrderIDs []string, err error) {
if utils.IsTimeZero(lastOrderTime) || time.Now().Sub(lastOrderTime) > maxGetOrderTimeDuration {
lastOrderTime = time.Now().Add(-maxGetOrderTimeDuration)
}
orderList, err := dao.GetStoreOrderAfterTime(dao.GetDB(), storeID, lastOrderTime, lastOrderSeqID)
for _, v := range orderList {
vendorOrderIDs = append(vendorOrderIDs, v.VendorOrderID)
}
return vendorOrderIDs, err
}
func GetMsg(ctx *jxcontext.Context, storeID int, lastOrderTime time.Time, lastOrderSeqID int64, msgTypeList []string, waitingSecond int) (msg *ServerMsg, err error) {
vendorOrderIDs, err := getPendingOrderList(storeID, lastOrderTime, lastOrderSeqID)
if err == nil {
msg = &ServerMsg{
Type: ServerMsgNewOrder,
StoreID: storeID,
MsgData: 0,
}
if len(vendorOrderIDs) == 0 {
chan2Listen := make(chan *ServerMsg, 1)
registerChan(storeID, chan2Listen)
pollingDuration := defPollingDuration
if waitingSecond != 0 {
pollingDuration = time.Duration(waitingSecond) * time.Second
if globals.IsProductEnv() {
if pollingDuration > maxPollingDuration {
pollingDuration = maxPollingDuration
} else if pollingDuration < minPollingDuration {
pollingDuration = minPollingDuration
}
}
}
timer := time.NewTimer(pollingDuration)
select {
case msg2, ok := <-chan2Listen:
timer.Stop()
if ok {
msg = msg2
}
case <-timer.C:
unregisterChan(storeID, chan2Listen)
}
close(chan2Listen)
} else {
globals.SugarLogger.Debugf("GetMsg vendorOrderIDs:%s", utils.Format4Output(vendorOrderIDs, true))
msg.MsgData = len(vendorOrderIDs)
}
}
return msg, err
}
func OnNewOrder(order *model.GoodsOrder) {
globals.SugarLogger.Debugf("msghub OnNewOrder, order:%s", utils.Format4Output(order, true))
utils.CallFuncAsync(func() {
msgChan <- &ServerMsg{
Type: ServerMsgNewOrder,
StoreID: jxutils.GetSaleStoreIDFromOrder(order),
MsgData: 1,
// MsgData: []*model.GoodsOrderExt{
// &model.GoodsOrderExt{
// GoodsOrder: *order,
// },
// },
}
})
}
func OnFinishedPickup(order *model.GoodsOrder) {
globals.SugarLogger.Debugf("msghub OnFinishedPickup, order:%s", utils.Format4Output(order, true))
utils.CallFuncAsync(func() {
msgChan <- &ServerMsg{
Type: ServerMsgFinishedPickup,
StoreID: jxutils.GetSaleStoreIDFromOrder(order),
MsgData: 1,
// MsgData: []*model.GoodsOrderExt{
// &model.GoodsOrderExt{
// GoodsOrder: *order,
// },
// },
}
})
}
func OnKeyOrderStatusChanged(order *model.GoodsOrder) {
globals.SugarLogger.Debugf("msghub OnKeyOrderStatusChanged, order:%s", utils.Format4Output(order, true))
utils.CallFuncAsync(func() {
msgChan <- &ServerMsg{
Type: ServerMsgKeyOrderStatusChanged,
StoreID: jxutils.GetSaleStoreIDFromOrder(order),
MsgData: order,
}
})
}
func OnNewWait4ApproveAfsOrder(order *model.AfsOrder) {
globals.SugarLogger.Debugf("msghub OnNewWait4ApproveAfsOrder, order:%s", utils.Format4Output(order, true))
utils.CallFuncAsync(func() {
msgChan <- &ServerMsg{
Type: ServerMsgNewWait4ApproveAfsOrder,
StoreID: jxutils.GetSaleStoreIDFromAfsOrder(order),
MsgData: order,
}
})
}
func OnKeyAfsOrderStatusChanged(order *model.AfsOrder) {
globals.SugarLogger.Debugf("msghub OnKeyAfsOrderStatusChanged, order:%s", utils.Format4Output(order, true))
utils.CallFuncAsync(func() {
msgChan <- &ServerMsg{
Type: ServerMsgKeyAfsOrderStatusChanged,
StoreID: jxutils.GetSaleStoreIDFromAfsOrder(order),
MsgData: order,
}
})
}

73
controllers/cms.go Normal file
View File

@@ -0,0 +1,73 @@
package controllers
import (
"git.rosy.net.cn/jx-callback/business/jxstore/cms"
"github.com/astaxie/beego"
)
type CmsController struct {
beego.Controller
}
// @Title 新增配置
// @Description 新增配置
// @Param token header string true "认证token"
// @Param type formData string true "配置类型（当前只支持PricePack）"
// @Param key formData string true "配置名"
// @Param value formData string true "配置值"
// @Success 200 {object} controllers.CallResult
// @Failure 200 {object} controllers.CallResult
// @router /NewConfig [post]
func (c *CmsController) NewConfig() {
c.callNewConfig(func(params *tCmsNewConfigParams) (retVal interface{}, errCode string, err error) {
err = cms.AddConfig(params.Ctx, params.Key, params.Type, params.Value)
return retVal, "", err
})
}
// @Title 删除配置
// @Description 删除配置
// @Param token header string true "认证token"
// @Param type query string true "配置类型（当前只支持PricePack）"
// @Param key query string true "配置名"
// @Success 200 {object} controllers.CallResult
// @Failure 200 {object} controllers.CallResult
// @router /DeleteConfig [delete]
func (c *CmsController) DeleteConfig() {
c.callDeleteConfig(func(params *tCmsDeleteConfigParams) (retVal interface{}, errCode string, err error) {
err = cms.DeleteConfig(params.Ctx, params.Key, params.Type)
return retVal, "", err
})
}
// @Title 修改配置
// @Description 修改配置
// @Param token header string true "认证token"
// @Param type formData string true "配置类型（当前只支持PricePack）"
// @Param key formData string true "配置名"
// @Param value formData string true "配置值"
// @Success 200 {object} controllers.CallResult
// @Failure 200 {object} controllers.CallResult
// @router /UpdateConfig [put]
func (c *CmsController) UpdateConfig() {
c.callUpdateConfig(func(params *tCmsUpdateConfigParams) (retVal interface{}, errCode string, err error) {
retVal, err = cms.UpdateConfig(params.Ctx, params.Key, params.Type, params.Value)
return retVal, "", err
})
}
// @Title 查询配置
// @Description 查询配置
// @Param token header string fasle "认证token"
// @Param type query string false "配置类型（当前只支持PricePack）"
// @Param key query string false "配置名"
// @Param keyword query string false "关键字"
// @Success 200 {object} controllers.CallResult
// @Failure 200 {object} controllers.CallResult
// @router /QueryConfigs [get]
func (c *CmsController) QueryConfigs() {
c.callQueryConfigs(func(params *tCmsQueryConfigsParams) (retVal interface{}, errCode string, err error) {
retVal, err = cms.QueryConfigs(params.Key, params.Type, params.Keyword)
return retVal, "", err
})
}

5
routers/router.go
View File

@@ -51,6 +51,11 @@ func init() {
&controllers.BillController{},
),
),
beego.NSNamespace("/cms",
beego.NSInclude(
&controllers.CmsController{},
),
),
)
beego.AddNamespace(ns)
beego.AutoRouter(&controllers.WXPayController{})