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

- write legacy jxoder almost ok.

Browse Source
This commit is contained in:
gazebo
2018-07-17 18:01:49 +08:00
parent 77d1ea1d77
commit f9096a7b17
14 changed files with 342 additions and 272 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
business/controller/controller.go
View File

@@ -15,6 +15,7 @@ import (
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/baseapi/utils/routinepool"
"git.rosy.net.cn/jx-callback/business/scheduler"
_ "git.rosy.net.cn/jx-callback/business/scheduler/defsch" // 导入缺省订单调度器
"git.rosy.net.cn/jx-callback/globals"
"git.rosy.net.cn/jx-callback/globals/api"
@@ -39,6 +40,7 @@ func init() {
routinePool = routinepool.New(1000, 1000)
OrderManager = NewOrderManager()
WaybillManager = NewWaybillManager()
scheduler.CurrentScheduler.RegisterOrderManager(OrderManager)
}
func (m *SyncMapWithTimeout) StoreWithTimeout(key, value interface{}, timeout time.Duration) {
@@ -133,10 +135,10 @@ func addOrderOrWaybillStatus(status *model.OrderStatus, db orm.Ormer) (isDuplica
}
func CallMsgHandler(handler func(), primaryID string) {
handler()
// controller.routinePool.CallFun(func() {
// handler()
// }, primaryID)
// handler()
routinePool.CallFun(func() {
handler()
}, primaryID)
}
func GetDataCityCodeFromOrder(order *model.GoodsOrder) (retVal string, err error) {

3
business/controller/elm/elm.go
View File

@@ -4,6 +4,7 @@ import (
"git.rosy.net.cn/baseapi/platformapi/elmapi"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/controller"
"git.rosy.net.cn/jx-callback/business/model"
)
type Controller struct {
@@ -46,7 +47,7 @@ func (c *Controller) OnCallbackMsg(msg *elmapi.CallbackMsg) (retVal *elmapi.Call
innerMsg.MsgType = msg.Type
controller.CallMsgHandler(func() {
retVal = new(OrderController).onOrderUserUrgeOrder(&innerMsg)
}, innerMsg.OrderID)
}, controller.ComposeUniversalOrderID(innerMsg.OrderID, model.VendorIDELM))
}
} else if msg.Type >= elmapi.MsgTypeWaybillWait4DeliveryVendor && msg.Type <= elmapi.MsgTypeRejectedSystemError {
var innerMsg elmapi.CallbackWaybillStatusMsg

48
business/controller/elm/order.go
View File

@@ -25,42 +25,46 @@ func init() {
func (c *OrderController) OnOrderStatusMsg(msg *elmapi.CallbackOrderStatusMsg) (retVal *elmapi.CallbackResponse) {
controller.CallMsgHandler(func() {
retVal = c.onOrderStatusMsg(msg)
}, msg.OrderID)
}, controller.ComposeUniversalOrderID(msg.OrderID, model.VendorIDELM))
return retVal
}
func (c *OrderController) OnOrderNewMsg(msg map[string]interface{}) (retVal *elmapi.CallbackResponse) {
controller.CallMsgHandler(func() {
retVal = c.onOrderNew(msg)
}, msg["orderId"].(string))
}, controller.ComposeUniversalOrderID(msg["orderId"].(string), model.VendorIDELM))
return retVal
}
func (c *OrderController) OnOrderCancelRefundMsg(msg *elmapi.CallbackOrderCancelRefundMsg) (retVal *elmapi.CallbackResponse) {
controller.CallMsgHandler(func() {
retVal = c.onOrderCancelRefundMsg(msg)
}, msg.OrderID)
}, controller.ComposeUniversalOrderID(msg.OrderID, model.VendorIDELM))
return retVal
}
func (c *OrderController) orderStatusMsg2Status(msg *elmapi.CallbackOrderStatusMsg) *model.OrderStatus {
orderStatus := &model.OrderStatus{
VendorOrderID: msg.OrderID,
VendorID: model.VendorIDELM,
OrderType: model.OrderTypeOrder,
VendorStatus: c.stateAndType2Str(msg.State, msg.MsgType),
StatusTime: utils.Timestamp2Time(msg.UpdateTime),
VendorOrderID: msg.OrderID,
VendorID: model.VendorIDELM,
OrderType: model.OrderTypeOrder,
RefVendorOrderID: msg.OrderID,
RefVendorID: model.VendorIDELM,
VendorStatus: c.stateAndType2Str(msg.State, msg.MsgType),
StatusTime: utils.Timestamp2Time(msg.UpdateTime),
}
return orderStatus
}
func (c *OrderController) cancelRefundMsg2Status(msg *elmapi.CallbackOrderCancelRefundMsg) *model.OrderStatus {
orderStatus := &model.OrderStatus{
VendorOrderID: msg.OrderID,
VendorID: model.VendorIDELM,
OrderType: model.OrderTypeOrder,
VendorStatus: c.stateAndType2Str(msg.RefundStatus, msg.MsgType),
StatusTime: utils.Timestamp2Time(msg.UpdateTime),
VendorOrderID: msg.OrderID,
VendorID: model.VendorIDELM,
OrderType: model.OrderTypeOrder,
RefVendorOrderID: msg.OrderID,
RefVendorID: model.VendorIDELM,
VendorStatus: c.stateAndType2Str(msg.RefundStatus, msg.MsgType),
StatusTime: utils.Timestamp2Time(msg.UpdateTime),
}
return orderStatus
}
@@ -151,8 +155,8 @@ func (c *OrderController) getOrderInfo(orderID string) (order *model.GoodsOrder,
order.Skus = append(order.Skus, sku)
order.SkuCount++
order.GoodsCount += sku.Count
order.SalePrice += sku.SalePrice
order.Weight += sku.Weight
order.SalePrice += sku.SalePrice * int64(sku.Count)
order.Weight += sku.Weight * sku.Count
}
}
}
@@ -176,12 +180,14 @@ func (c *OrderController) onOrderNew(msg map[string]interface{}) (response *elma
func (c *OrderController) onOrderUserUrgeOrder(msg *elmapi.CallbackOrderUrgeMsg) *elmapi.CallbackResponse {
status := &model.OrderStatus{
VendorOrderID: msg.OrderID,
VendorID: model.VendorIDELM,
OrderType: model.OrderTypeOrder,
Status: model.OrderStatusApplyUrgeOrder,
VendorStatus: utils.Int2Str(msg.MsgType),
StatusTime: utils.Timestamp2Time(msg.UpdateTime),
VendorOrderID: msg.OrderID,
VendorID: model.VendorIDELM,
OrderType: model.OrderTypeOrder,
RefVendorOrderID: msg.OrderID,
RefVendorID: model.VendorIDELM,
Status: model.OrderStatusApplyUrgeOrder,
VendorStatus: utils.Int2Str(msg.MsgType),
StatusTime: utils.Timestamp2Time(msg.UpdateTime),
}
return elmapi.Err2CallbackResponse(controller.OrderManager.OnOrderStatusChanged(status), status.VendorStatus)
}

2
business/controller/elm/waybill.go
View File

@@ -15,7 +15,7 @@ type WaybillController struct {
func (c *WaybillController) OnWaybillStatusMsg(msg *elmapi.CallbackWaybillStatusMsg) (retVal *elmapi.CallbackResponse) {
controller.CallMsgHandler(func() {
retVal = c.onWaybillStatusMsg(msg)
}, msg.OrderID)
}, controller.ComposeUniversalOrderID(msg.OrderID, model.VendorIDELM))
return retVal
}

18
business/controller/jd/order.go
View File

@@ -22,7 +22,7 @@ func init() {
func (c *OrderController) OnOrderMsg(msg *jdapi.CallbackOrderMsg) (retVal *jdapi.CallbackResponse) {
controller.CallMsgHandler(func() {
retVal = c.onOrderMsg(msg)
}, msg.BillID)
}, controller.ComposeUniversalOrderID(msg.BillID, model.VendorIDJD))
return retVal
}
@@ -116,8 +116,8 @@ func (c *OrderController) getOrderInfo(msg *jdapi.CallbackOrderMsg) (order *mode
order.Skus = append(order.Skus, sku)
order.SkuCount++
order.GoodsCount += sku.Count
order.SalePrice += sku.SalePrice
order.Weight += sku.Weight
order.SalePrice += sku.SalePrice * int64(sku.Count)
order.Weight += sku.Weight * sku.Count
}
}
return order, err
@@ -150,11 +150,13 @@ func (c *OrderController) onOrderAdjust(msg *jdapi.CallbackOrderMsg) *jdapi.Call
func (c *OrderController) callbackMsg2Status(msg *jdapi.CallbackOrderMsg) *model.OrderStatus {
orderStatus := &model.OrderStatus{
VendorOrderID: msg.BillID,
VendorID: model.VendorIDJD,
OrderType: model.OrderTypeOrder,
VendorStatus: msg.StatusID,
StatusTime: utils.Str2Time(msg.Timestamp),
VendorOrderID: msg.BillID,
VendorID: model.VendorIDJD,
OrderType: model.OrderTypeOrder,
RefVendorOrderID: msg.BillID,
RefVendorID: model.VendorIDJD,
VendorStatus: msg.StatusID,
StatusTime: utils.Str2Time(msg.Timestamp),
}
return orderStatus
}

2
business/controller/jd/waybill.go
View File

@@ -13,7 +13,7 @@ type WaybillController struct {
func (c *WaybillController) OnWaybillMsg(msg *jdapi.CallbackDeliveryStatusMsg) (retVal *jdapi.CallbackResponse) {
controller.CallMsgHandler(func() {
retVal = c.onWaybillMsg(msg)
}, msg.OrderID)
}, controller.ComposeUniversalOrderID(msg.OrderID, model.VendorIDJD))
return retVal
}

4
business/controller/mtps/waybill.go
View File

@@ -22,7 +22,7 @@ func init() {
func (c *WaybillController) OnWaybillMsg(msg *mtpsapi.CallbackOrderMsg) (retVal *mtpsapi.CallbackResponse) {
controller.CallMsgHandler(func() {
retVal = c.onWaybillMsg(msg)
}, msg.MtPeisongID)
}, msg.OrderID)
return retVal
}
@@ -40,7 +40,7 @@ func (c *WaybillController) OnWaybillExcept(msg *mtpsapi.CallbackOrderExceptionM
}
order.VendorOrderID, order.OrderVendorID = controller.SplitUniversalOrderID(msg.OrderID)
retVal = mtpsapi.Err2CallbackResponse(controller.WaybillManager.OnWaybillStatusChanged(order), "mtps OnWaybillExcept")
}, msg.MtPeisongID)
}, msg.OrderID)
return retVal
}

206
business/controller/order.go
View File

@@ -3,17 +3,17 @@ package controller
import (
"fmt"
"git.rosy.net.cn/jx-callback/business/scheduler"
"git.rosy.net.cn/baseapi"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/model"
"git.rosy.net.cn/jx-callback/globals"
"git.rosy.net.cn/jx-callback/legacy/models"
"github.com/astaxie/beego/orm"
)
// 所有公共接口调用前，要求在order里或status中设置合适的Status
type OrderController struct {
orderMap SyncMapWithTimeout
}
func NewOrderManager() *OrderController {
@@ -22,12 +22,19 @@ func NewOrderManager() *OrderController {
func (c *OrderController) OnOrderNew(order *model.GoodsOrder) (err error) {
db := orm.NewOrm()
return c.saveOrder(order, false, db)
isDuplicated, err := addOrderOrWaybillStatus(c.order2Status(order), db)
if err == nil && !isDuplicated {
if err = c.saveOrder(order, false, db); err == nil {
err = scheduler.CurrentScheduler.OnOrderNew(order)
}
}
return err
}
func (c *OrderController) OnOrderAdjust(order *model.GoodsOrder) (err error) {
db := orm.NewOrm()
isDuplicated, err := addOrderOrWaybillStatus(c.order2Status(order), db)
status := c.order2Status(order)
isDuplicated, err := addOrderOrWaybillStatus(status, db)
if err == nil && !isDuplicated {
err = utils.CallFuncLogError(func() error {
_, err = db.Raw("DELETE FROM order_sku WHERE vendor_order_id = ? AND vendor_id = ?", order.VendorOrderID, order.VendorID).Exec()
@@ -36,7 +43,6 @@ func (c *OrderController) OnOrderAdjust(order *model.GoodsOrder) (err error) {
if err != nil {
return err
}
err = utils.CallFuncLogError(func() error {
_, err = db.Raw("DELETE FROM goods_order WHERE vendor_order_id = ? AND vendor_id = ?", order.VendorOrderID, order.VendorID).Exec()
return err
@@ -44,13 +50,17 @@ func (c *OrderController) OnOrderAdjust(order *model.GoodsOrder) (err error) {
if err != nil {
return err
}
if err = c.saveOrder(order, true, db); err == nil {
err = scheduler.CurrentScheduler.OnOrderStatusChanged(status)
}
}
return c.saveOrder(order, true, db)
return err
}
func (c *OrderController) OnOrderStatusChanged(orderStatus *model.OrderStatus) (err error) {
isDuplicated, err := c.addOrderStatus(orderStatus, nil)
if err == nil && !isDuplicated {
err = scheduler.CurrentScheduler.OnOrderStatusChanged(orderStatus)
if globals.HandleLegacyJxOrder {
c.legacyJxOrderStatusChanged(orderStatus, nil)
}
@@ -60,53 +70,41 @@ func (c *OrderController) OnOrderStatusChanged(orderStatus *model.OrderStatus) (
// private
func (c *OrderController) saveOrder(order *model.GoodsOrder, isAdjust bool, db orm.Ormer) (err error) {
isDuplicated, err := addOrderOrWaybillStatus(c.order2Status(order), db)
if !isDuplicated {
c.handleAutoAcceptOrder(order.VendorOrderID, order.VendorID, order.ConsigneeMobile, order.StoreID, db, func(isAccept bool) {
// c.purchasePlatformHandlers[order.VendorID].AcceptOrRefuseOrder(order, isAccept)
if isAccept {
order.Status = model.OrderStatusAccepted
} else {
order.Status = model.OrderStatusFailed
// 忽略查找JX信息错误
c.updateOrderOtherInfo(order, db)
db.Begin()
// globals.SugarLogger.Debugf("new order:%v", order)
order.WaybillVendorID = model.VendorIDUnknown
order.OrderFinishedAt = utils.DefaultTimeValue
order.ID = 0
created, _, err2 := db.ReadOrCreate(order, "VendorOrderID", "VendorID")
if err = err2; err == nil {
if created {
sql := "INSERT INTO order_sku(vendor_order_id, vendor_id, count, vendor_sku_id, sku_id, jx_sku_id, sku_name, shop_price, sale_price, weight, order_created_at) VALUES"
params := []interface{}{}
for _, sku := range order.Skus {
sql += "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?),"
params = append(params, sku.VendorOrderID, sku.VendorID, sku.Count, sku.VendorSkuID, sku.SkuID, sku.JxSkuID, sku.SkuName, sku.ShopPrice, sku.SalePrice, sku.Weight, order.OrderCreatedAt)
}
})
// 忽略查找JX信息错误
c.updateOrderOtherInfo(order, db)
db.Begin()
// globals.SugarLogger.Debugf("new order:%v", order)
order.OrderFinishedAt = utils.DefaultTimeValue
order.ID = 0
created, _, err2 := db.ReadOrCreate(order, "VendorOrderID", "VendorID")
if err = err2; err == nil {
c.orderMap.Store(ComposeUniversalOrderID(order.VendorOrderID, order.VendorID), order.ID)
if created {
sql := "INSERT INTO order_sku(vendor_order_id, vendor_id, count, vendor_sku_id, sku_id, jx_sku_id, sku_name, shop_price, sale_price, weight, order_created_at) VALUES"
params := []interface{}{}
for _, sku := range order.Skus {
sql += "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?),"
params = append(params, sku.VendorOrderID, sku.VendorID, sku.Count, sku.VendorSkuID, sku.SkuID, sku.JxSkuID, sku.SkuName, sku.ShopPrice, sku.SalePrice, order.Weight, order.OrderCreatedAt)
}
sql = sql[:len(sql)-1] + ";"
if _, err = db.Raw(sql, params...).Exec(); err != nil {
db.Rollback()
baseapi.SugarLogger.Infof("insert order_sku error:%v", err)
} else {
db.Commit()
if globals.HandleLegacyJxOrder {
c.legacyWriteJxOrder(order, db, isAdjust)
}
}
sql = sql[:len(sql)-1] + ";"
if _, err = db.Raw(sql, params...).Exec(); err != nil {
db.Rollback()
baseapi.SugarLogger.Infof("insert order_sku error:%v", err)
} else {
order.DuplicatedCount++
db.Update(order, "DuplicatedCount")
db.Commit()
baseapi.SugarLogger.Infof("duplicated order:%s vendorID:%d, msg received", order.VendorOrderID, order.VendorID)
if globals.HandleLegacyJxOrder {
c.legacyWriteJxOrder(order, db, isAdjust)
}
}
} else {
db.Rollback()
globals.SugarLogger.Warnf("create order:%v, error:%v", order, err)
order.DuplicatedCount++
db.Update(order, "DuplicatedCount")
db.Commit()
baseapi.SugarLogger.Infof("duplicated order:%s vendorID:%d, msg received", order.VendorOrderID, order.VendorID)
}
} else {
db.Rollback()
globals.SugarLogger.Warnf("create order:%v, error:%v", order, err)
}
return err
}
@@ -182,112 +180,50 @@ func (c *OrderController) updateOrderOtherInfo(order *model.GoodsOrder, db orm.O
return err
}
func (c *OrderController) handleAutoAcceptOrder(orderID string, vendorID int, userMobile string, jxStoreID int, db orm.Ormer, handler func(accepted bool)) int {
handleType := 0
if userMobile != "" {
if db == nil {
db = orm.NewOrm()
}
user := &models.BlackClient{
Mobile: userMobile,
}
if err := db.Read(user, "Mobile"); err != nil {
if err != orm.ErrNoRows {
globals.SugarLogger.Errorf("read data error:%v, data:%v, vendorID:%d", err, user, vendorID)
}
// 在访问数据库出错的情况下，也需要自动接单
handleType = 1
} else {
// 强制拒单
globals.SugarLogger.Infof("force reject order:%s, vendorID:%d", orderID, vendorID)
handleType = -1
}
} else {
globals.SugarLogger.Infof("order:%s, vendorID:%d, mobile is empty, should accept order", orderID, vendorID)
handleType = 1
}
if handleType == 1 {
handler(true)
} else if handleType == -1 {
handler(false)
}
return handleType
}
func (c *OrderController) addOrderStatus(orderStatus *model.OrderStatus, db orm.Ormer) (isDuplicated bool, err error) {
if db == nil {
db = orm.NewOrm()
}
isDuplicated, err = addOrderOrWaybillStatus(orderStatus, db)
if !isDuplicated && orderStatus.Status > model.OrderStatusUnknown {
order := &model.GoodsOrder{
VendorOrderID: orderStatus.VendorOrderID,
VendorID: orderStatus.VendorID,
if !isDuplicated && orderStatus.Status > model.OrderStatusNew {
params := orm.Params{
"status": orderStatus.Status,
"vendor_status": orderStatus.VendorStatus,
}
if err = c.updateOrderPKID(order, db); err == nil {
order.Status = orderStatus.Status
order.VendorStatus = orderStatus.VendorStatus
utils.CallFuncLogError(func() error {
columns := []string{"Status", "VendorStatus"}
if orderStatus.Status >= model.OrderStatusEndBegin {
order.OrderFinishedAt = orderStatus.StatusTime
columns = append(columns, "OrderFinishedAt")
}
_, err := db.Update(order, columns...)
return err
}, "update order")
if orderStatus.Status >= model.OrderStatusEndBegin {
params["order_finished_at"] = orderStatus.StatusTime
}
utils.CallFuncLogError(func() error {
_, err = db.QueryTable("goods_order").Filter("vendor_order_id", orderStatus.VendorOrderID).Filter("vendor_id", orderStatus.VendorID).Update(params)
return err
}, "update order")
}
return isDuplicated, err
}
func (c *OrderController) updateOrderPKID(order *model.GoodsOrder, db orm.Ormer) (err error) {
value, ok := c.orderMap.Load(ComposeUniversalOrderID(order.VendorOrderID, order.VendorID))
if !ok {
err = db.Read(order, "VendorOrderID", "VendorID")
// todo 这里应该要报警，但测试阶段先去掉
// utils.CallFuncLogError(func() error {
// err = db.Read(order, "VendorOrderID", "VendorID")
// return err
// }, "can not get order info from db")
} else {
order.ID = value.(int64)
}
return err
}
func (c *OrderController) order2Status(order *model.GoodsOrder) (retVal *model.OrderStatus) {
retVal = &model.OrderStatus{
VendorOrderID: order.VendorOrderID,
VendorID: order.VendorID,
OrderType: model.OrderTypeOrder,
Status: order.Status,
VendorStatus: order.VendorStatus,
StatusTime: order.OrderCreatedAt,
VendorOrderID: order.VendorOrderID,
VendorID: order.VendorID,
OrderType: model.OrderTypeOrder,
RefVendorOrderID: order.VendorOrderID,
RefVendorID: order.VendorID,
Status: order.Status,
VendorStatus: order.VendorStatus,
StatusTime: order.OrderCreatedAt,
}
return retVal
}
//Waybill
func (c *OrderController) OnWaybillStatusChanged(bill *model.Waybill, db orm.Ormer) (err error) {
return c.updateOrderByWaybill(bill, db)
}
func (c *OrderController) updateOrderByWaybill(bill *model.Waybill, db orm.Ormer) (err error) {
if db == nil {
db = orm.NewOrm()
}
order := &model.GoodsOrder{
VendorOrderID: bill.VendorOrderID,
VendorID: bill.OrderVendorID,
}
if err = db.Read(order, "VendorOrderID", "VendorID"); err == nil {
if order.Status < model.OrderStatusEndBegin {
order.WaybillStatus = bill.Status
order.WaybillVendorStatus = bill.VendorStatus
db.Update(order, "WaybillStatus", "WaybillVendorStatus")
}
func (c *OrderController) UpdateWaybillVendorID(bill *model.Waybill) (err error) {
db := orm.NewOrm()
params := orm.Params{
"waybill_vendor_id": bill.WaybillVendorID,
}
utils.CallFuncLogError(func() error {
_, err = db.QueryTable("goods_order").Filter("vendor_order_id", bill.VendorOrderID).Filter("vendor_id", bill.OrderVendorID).Update(params)
return err
}, "update order")
return err
}

128
business/controller/order_legacy.go
View File

@@ -1,7 +1,6 @@
package controller
import (
"fmt"
"strings"
"git.rosy.net.cn/baseapi"
@@ -45,6 +44,13 @@ const (
)
var (
VENDOR_ID2CARRIER_NO = map[int]string{
model.VendorIDJD: DD_CARRIER_NO,
model.VendorIDELM: EME_SELF_CARRIER_NO,
model.VendorIDDada: DDDELIVERY_CARRIER_NO,
model.VendorIDMTPS: MTPS_CARRIER_NO,
}
CARRIERS_NAMES = map[string]string{
DD_CARRIER_NO: "达达专送",
SELLER_CARRIER_NO: "门店自送",
@@ -62,7 +68,14 @@ var (
}
)
func (c *OrderController) legacyMapOrderStatus(orderStatus int) (retVal int8) {
func legacyMapCoordinateType(coordType int) (jxorderCoordType int) {
if coordType == model.CoordinateTypeMars {
return 2
}
return 0
}
func legacyMapOrderStatus(orderStatus int) (retVal int8) {
switch orderStatus {
case model.OrderStatusNew:
retVal = JX_ORDER_STATUS_WAIT_TO_ACCEPT
@@ -76,6 +89,28 @@ func (c *OrderController) legacyMapOrderStatus(orderStatus int) (retVal int8) {
retVal = JX_ORDER_STATUS_ADJUST
case model.OrderStatusApplyCancel:
retVal = JX_ORDER_STATUS_EXCEPTION_APPLY
case model.OrderStatusCanceled:
retVal = JX_ORDER_STATUS_CANCEL
}
return retVal
}
func legacyMapWaybillStatus(status int) (retVal int8) {
switch status {
case model.WaybillStatusNew:
retVal = JX_DELIVERY_STATUS_WAIT_TO_GRAB
case model.WaybillStatusAccepted:
retVal = JX_DELIVERY_STATUS_GRAB_DONE
case model.WaybillStatusCourierArrived:
retVal = JX_DELIVERY_STATUS_GET_GOODS_DONE
case model.WaybillStatusDelivering:
retVal = JX_DELIVERY_STATUS_GET_GOODS_DONE
case model.WaybillStatusDelivered:
retVal = JX_DELIVERY_STATUS_DELIVERY_DONE
case model.WaybillStatusCanceled:
retVal = JX_DELIVERY_STATUS_DELIVERY_CANCEL
case model.WaybillStatusFailed:
retVal = JX_DELIVERY_STATUS_DELIVERY_FAIL
}
return retVal
}
@@ -106,7 +141,7 @@ func (c *OrderController) legacyWriteJxOrder(order *model.GoodsOrder, db orm.Orm
JxStoreId: utils.Int2Str(GetJxStoreIDFromOrder(order)),
JxStoreName: order.StoreName,
OrderNum: order.OrderSeq,
OrderStatus: c.legacyMapOrderStatus(order.Status),
OrderStatus: legacyMapOrderStatus(order.Status),
OrderStatusTime: utils.Time2Str(order.OrderCreatedAt),
BusinessTag: businessTags,
SkuCount: order.SkuCount,
@@ -114,23 +149,30 @@ func (c *OrderController) legacyWriteJxOrder(order *model.GoodsOrder, db orm.Orm
BuyerFullName: order.ConsigneeName,
BuyerFullAddress: order.ConsigneeAddress,
BuyerMobile: order.ConsigneeMobile,
BuyerCoordType: order.CoordinateType,
BuyerCoordType: legacyMapCoordinateType(order.CoordinateType),
BuyerLng: IntCoordinate2Standard(order.ConsigneeLng),
BuyerLat: IntCoordinate2Standard(order.ConsigneeLat),
CityName: "all",
OrderStartTime: utils.Time2Str(order.OrderCreatedAt),
JdStoreId: order.VendorStoreID,
// DeliveryPackageWeight: float64(order.Weight) / 1000,
}
if order.ExpectedDeliveredTime.Sub(utils.DefaultTimeValue) > 0 {
jxorder.OrderPreEndDelivTime = utils.Time2Str(order.ExpectedDeliveredTime)
}
_, err = db.Insert(jxorder)
if err != nil {
db.Rollback()
globals.SugarLogger.Infof("insert jxorder error:%v", err)
return err
}
sql := "INSERT INTO jxordersku2(vender_id, order_id, jx_sku_id, sku_name, jx_store_id, sku_price, sku_count, is_gift, promotion_type, sku_plat_discount, sku_vender_discount) VALUES"
sql := "INSERT INTO jxordersku2(vender_id, order_id, jx_sku_id, sku_name, jx_store_id, sku_price, sku_count, is_gift, promotion_type, sku_plat_discount, sku_vender_discount, sku_img) VALUES"
params := []interface{}{}
for _, sku := range order.Skus {
sql += "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?),"
params = append(params, sku.VendorID, sku.VendorOrderID, sku.JxSkuID, sku.SkuName, GetJxStoreIDFromOrder(order), sku.SalePrice, sku.Count, sku.SkuType, sku.PromotionType, 0, 0)
sql += "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?),"
params = append(params, sku.VendorID, sku.VendorOrderID, sku.JxSkuID, sku.SkuName, GetJxStoreIDFromOrder(order), sku.SalePrice, sku.Count, sku.SkuType, sku.PromotionType, 0, 0, "")
}
sql = sql[:len(sql)-1] + ";"
if _, err = db.Raw(sql, params...).Exec(); err != nil {
@@ -160,27 +202,10 @@ func (c *OrderController) legacyWriteJxOrder(order *model.GoodsOrder, db orm.Orm
`, vendorOrderID /*, vendorOrderID*/).Exec()
return err
}, "update jxordersku")
}
return err
}
func (c *OrderController) legacyGetOrderDeliveryChannel(status *model.OrderStatus) (retVal string) {
switch status.VendorID {
case model.VendorIDJD:
retVal = DD_CARRIER_NO
case model.VendorIDELM:
retVal = EME_SELF_CARRIER_NO
case model.VendorIDDada:
retVal = DDDELIVERY_CARRIER_NO
case model.VendorIDMTPS:
retVal = MTPS_CARRIER_NO
default:
panic(fmt.Sprintf("unkown vendorID:%v", status.VendorID))
}
return retVal
}
func (c *OrderController) legacyJxOrderStatusChanged(status *model.OrderStatus, db orm.Ormer) (err error) {
if db == nil {
db = orm.NewOrm()
@@ -190,9 +215,60 @@ func (c *OrderController) legacyJxOrderStatusChanged(status *model.OrderStatus,
}
if err = db.Read(jxorder, "OrderId"); err == nil {
utils.CallFuncLogError(func() error {
jxorder.OrderStatus = c.legacyMapOrderStatus(status.Status)
jxorder.OrderStatus = legacyMapOrderStatus(status.Status)
jxorder.OrderStatusTime = utils.Time2Str(status.StatusTime)
_, err = db.Update(jxorder, "OrderStatus", "OrderStatusTime")
updateFields := []string{
"OrderStatus",
"OrderStatusTime",
}
if status.Status >= model.OrderStatusEndBegin {
jxorder.DeliveryFinishTime = utils.Time2Str(status.StatusTime)
updateFields = append(updateFields, "DeliveryFinishTime")
}
_, err = db.Update(jxorder, updateFields...)
return err
}, "legacyJxOrderStatusChanged")
} else {
globals.SugarLogger.Infof("read legacyJxOrder error:%v, orderID:%d", err, jxorder.OrderId)
}
return err
}
func (c *WaybillController) legacyWaybillStatusChanged(bill *model.Waybill, db orm.Ormer) (err error) {
if db == nil {
db = orm.NewOrm()
}
jxorder := &legacyorder.Jxorder2{
OrderId: utils.Str2Int64(bill.VendorOrderID),
}
if err = db.Read(jxorder, "OrderId"); err == nil {
utils.CallFuncLogError(func() error {
updateFields := []string{
"DeliveryCarrierNo",
"DeliveryCarrierName",
"DeliveryManNo",
"DeliveryManName",
"DeliveryManPhone",
"DeliveryBillNo",
"DeliveryStatus",
}
// jxorder.DeliveryPackageWeight
jxorder.DeliveryCarrierNo = VENDOR_ID2CARRIER_NO[bill.WaybillVendorID]
jxorder.DeliveryCarrierName = CARRIERS_NAMES[jxorder.DeliveryCarrierNo]
jxorder.DeliveryManNo = bill.CourierMobile
jxorder.DeliveryManName = bill.CourierName
jxorder.DeliveryManPhone = bill.CourierMobile
jxorder.DeliveryBillNo = bill.VendorWaybillID
jxorder.DeliveryStatus = legacyMapWaybillStatus(bill.Status)
// jxorder.DeliveryConfirmTime
if bill.Status == model.WaybillStatusNew {
updateFields = append(updateFields, "DeliveryStartTime")
jxorder.DeliveryStartTime = utils.Time2Str(bill.WaybillCreatedAt)
} else if bill.Status >= model.WaybillStatusEndBegin {
jxorder.DeliveryFinishTime = utils.Time2Str(bill.WaybillCreatedAt)
updateFields = append(updateFields, "DeliveryFinishTime")
}
_, err = db.Update(jxorder, updateFields...)
return err
}, "legacyJxOrderStatusChanged")
} else {

84
business/controller/waybill.go
View File

@@ -3,13 +3,13 @@ package controller
import (
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/model"
"git.rosy.net.cn/jx-callback/business/scheduler"
"git.rosy.net.cn/jx-callback/globals"
"github.com/astaxie/beego/orm"
)
// 所有公共接口调用前，要求在order里或status中设置合适的Status
type WaybillController struct {
WaybillMap SyncMapWithTimeout
}
func NewWaybillManager() *WaybillController {
@@ -24,13 +24,15 @@ func (w *WaybillController) onWaybillNew(bill *model.Waybill) (err error) {
bill.ID = 0
created, _, err2 := db.ReadOrCreate(bill, "VendorWaybillID", "WaybillVendorID")
if err = err2; err == nil {
w.WaybillMap.Store(ComposeUniversalOrderID(bill.VendorWaybillID, bill.WaybillVendorID), bill.ID)
if !created {
bill.DuplicatedCount++
db.Update(bill, "DuplicatedCount")
globals.SugarLogger.Infof("duplicated bill:%v vendorID:%d, msg received", bill.VendorWaybillID, bill.WaybillVendorID)
}
err = OrderManager.OnWaybillStatusChanged(bill, db)
err = scheduler.CurrentScheduler.OnWaybillStatusChanged(bill)
if globals.HandleLegacyJxOrder {
w.legacyWaybillStatusChanged(bill, db)
}
} else {
globals.SugarLogger.Warnf("create bill:%v, error:%v", bill, err)
}
@@ -47,7 +49,10 @@ func (w *WaybillController) OnWaybillStatusChanged(bill *model.Waybill) (err err
db := orm.NewOrm()
isDuplicated, err2 := w.addWaybillStatus(bill, db)
if err = err2; err == nil && !isDuplicated {
err = OrderManager.OnWaybillStatusChanged(bill, db)
err = scheduler.CurrentScheduler.OnWaybillStatusChanged(bill)
if globals.HandleLegacyJxOrder {
w.legacyWaybillStatusChanged(bill, db)
}
}
}
return err
@@ -57,14 +62,17 @@ func (w *WaybillController) onWaybillAccepted(bill *model.Waybill) (err error) {
db := orm.NewOrm()
isDuplicated, err := w.addWaybillStatus(bill, db)
if !isDuplicated {
if err = w.updateWaybillPKID(bill, db); err == nil {
utils.CallFuncLogError(func() error {
_, err = db.Update(bill, "CourierName", "CourierMobile")
return err
}, "update waybill info")
if err == nil {
err = OrderManager.OnWaybillStatusChanged(bill, db)
}
params := orm.Params{
"courier_name": bill.CourierName,
"courier_mobile": bill.CourierMobile,
}
utils.CallFuncLogError(func() error {
_, err = db.QueryTable("waybill").Filter("vendor_waybill_id", bill.VendorWaybillID).Filter("waybill_vendor_id", bill.WaybillVendorID).Update(params)
return err
}, "update waybill courier info")
err = scheduler.CurrentScheduler.OnWaybillStatusChanged(bill)
if globals.HandleLegacyJxOrder {
w.legacyWaybillStatusChanged(bill, db)
}
}
return err
@@ -76,46 +84,32 @@ func (w *WaybillController) addWaybillStatus(bill *model.Waybill, db orm.Ormer)
}
waybillStatus := w.waybill2Status(bill)
isDuplicated, err = addOrderOrWaybillStatus(waybillStatus, db)
if !isDuplicated && waybillStatus.Status > model.WaybillStatusUnknown {
if err = w.updateWaybillPKID(bill, db); err == nil {
bill.Status = waybillStatus.Status
bill.VendorStatus = waybillStatus.VendorStatus
utils.CallFuncLogError(func() error {
columns := []string{"Status", "VendorStatus"}
if waybillStatus.Status >= model.OrderStatusEndBegin {
bill.WaybillFinishedAt = waybillStatus.StatusTime
columns = append(columns, "WaybillFinishedAt")
}
_, err = db.Update(bill, columns...)
return err
}, "update waybill")
if !isDuplicated && waybillStatus.Status > model.WaybillStatusNew {
params := orm.Params{
"status": bill.Status,
"vendor_status": bill.VendorStatus,
}
if bill.Status >= model.WaybillStatusEndBegin {
params["waybill_finished_at"] = bill.WaybillCreatedAt
}
utils.CallFuncLogError(func() error {
_, err = db.QueryTable("waybill").Filter("vendor_waybill_id", bill.VendorWaybillID).Filter("waybill_vendor_id", bill.WaybillVendorID).Update(params)
return err
}, "update waybill status")
}
return isDuplicated, err
}
func (w *WaybillController) waybill2Status(bill *model.Waybill) (retVal *model.OrderStatus) {
retVal = &model.OrderStatus{
VendorOrderID: bill.VendorWaybillID,
VendorID: bill.WaybillVendorID,
OrderType: model.OrderTypeWaybill,
Status: bill.Status,
VendorStatus: bill.VendorStatus,
StatusTime: bill.WaybillCreatedAt,
VendorOrderID: bill.VendorWaybillID,
VendorID: bill.WaybillVendorID,
OrderType: model.OrderTypeWaybill,
RefVendorOrderID: bill.VendorOrderID,
RefVendorID: bill.OrderVendorID,
Status: bill.Status,
VendorStatus: bill.VendorStatus,
StatusTime: bill.WaybillCreatedAt,
}
return retVal
}
func (w *WaybillController) updateWaybillPKID(bill *model.Waybill, db orm.Ormer) (err error) {
if value, ok := w.WaybillMap.Load(ComposeUniversalOrderID(bill.VendorWaybillID, bill.WaybillVendorID)); !ok {
err = db.Read(bill, "VendorWaybillID", "WaybillVendorID")
// todo 这里应该要报警，但测试阶段先去掉
// utils.CallFuncLogError(func() error {
// err = db.Read(bill, "VendorWaybillID", "WaybillVendorID")
// return err
// }, "can not get waybill info from db")
} else {
bill.ID = value.(int64)
}
return err
}

4
business/model/const.go
View File

@@ -1,5 +1,6 @@
package model
// VendorIDJD, VendorIDMTWM与VendorIDELM的定义和老系统是兼容的
const (
VendorIDUnknown = -1
VendorIDPurchaseBegin = 0
@@ -57,7 +58,7 @@ const (
OrderStatusFinishedPickup = 15 // 拣货完成
OrderStatusDelivering = 20 // 开始配送，配送员已取货，从这里开始就是运单消息了
OrderStatusEndBegin = 100 // 以上的状态就是结束状态
OrderStatusEndBegin = 100 // 以下的状态就是结束状态
OrderStatusDelivered = 105 // 妥投
OrderStatusFinished = 110 // 订单已完成
OrderStatusCanceled = 115 // 订单已取消
@@ -78,6 +79,7 @@ const (
WaybillStatusCourierArrived = 15 // 此状态是可选的，明确写出来是因为还是较重要的状态，但业务逻辑不应依赖此状态
WaybillStatusDelivering = 20
WaybillStatusEndBegin = 100 // 以下的状态就是结束状态
WaybillStatusDelivered = 105
WaybillStatusCanceled = 115
WaybillStatusFailed = 120

22
business/model/order.go
View File

@@ -23,9 +23,9 @@ type GoodsOrder struct {
ConsigneeName string `orm:"size(32)"`
ConsigneeMobile string `orm:"size(32)"`
ConsigneeAddress string `orm:"size(255)"`
CoordinateType int
ConsigneeLng int // 坐标 * （10的六次方）
ConsigneeLat int // 坐标 * （10的六次方）
CoordinateType int
SkuCount int // 商品类别数量，即有多少种商品（注意在某些情况下，相同SKU的商品由于售价不同，也会当成不同商品在这个值里）
GoodsCount int // 商品个数
Status int // 参见OrderStatus*相关的常量定义
@@ -36,8 +36,6 @@ type GoodsOrder struct {
ExpectedDeliveredTime time.Time `orm:"type(datetime)"` // 预期送达时间
CancelApplyReason string `orm:"size(255)"` // ""表示没有申请，不为null表示用户正在取消申请
WaybillVendorID int `orm:"column(waybill_vendor_id)"`
WaybillStatus int
WaybillVendorStatus string `orm:"size(16)"`
DuplicatedCount int // 重复新订单消息数，这个一般不是由于消息重发造成的（消息重发由OrderStatus过滤），一般是业务逻辑造成的
OrderCreatedAt time.Time `orm:"type(datetime);index"`
OrderFinishedAt time.Time `orm:"type(datetime)"`
@@ -111,14 +109,16 @@ func (w *Waybill) TableIndex() [][]string {
// 包含订单与运单的状态及事件vendor status
type OrderStatus struct {
ID int64 `orm:"column(id)"`
VendorOrderID string `orm:"column(vendor_order_id);size(48)"`
VendorID int `orm:"column(vendor_id)"`
OrderType int // 0:订单，1：运单
Status int // 如果Status为OrderStatusEvent，表示VendorStatus只是一个通知事件，不是状态变化
VendorStatus string `orm:"size(16)"`
StatusTime time.Time `orm:"type(datetime);index"`
DuplicatedCount int // 收到的重复状态转换（或消息）数，一般是由于重发造成的
ID int64 `orm:"column(id)"`
VendorOrderID string `orm:"column(vendor_order_id);size(48)"`
VendorID int `orm:"column(vendor_id)"`
OrderType int // 0:订单，1：运单
RefVendorOrderID string `orm:"column(ref_vendor_order_id);size(48)"`
RefVendorID int `orm:"column(ref_vendor_id)"`
Status int // 如果Status为OrderStatusEvent，表示VendorStatus只是一个通知事件，不是状态变化
VendorStatus string `orm:"size(16)"`
StatusTime time.Time `orm:"type(datetime);index"`
DuplicatedCount int // 收到的重复状态转换（或消息）数，一般是由于重发造成的
ModelTimeInfo
}

59
business/scheduler/defsch/defsch.go
View File

@@ -3,6 +3,9 @@ package defsch
import (
"git.rosy.net.cn/jx-callback/business/model"
"git.rosy.net.cn/jx-callback/business/scheduler"
"git.rosy.net.cn/jx-callback/globals"
"git.rosy.net.cn/jx-callback/legacy/models"
"github.com/astaxie/beego/orm"
)
type DefScheduler struct {
@@ -20,15 +23,63 @@ func (s *DefScheduler) OnOrderNew(order *model.GoodsOrder) (err error) {
return nil
}
func (s *DefScheduler) OnOrderAccepted(order *model.GoodsOrder) (err error) {
return nil
}
func (s *DefScheduler) OnOrderStatusChanged(status *model.OrderStatus) (err error) {
return nil
}
// 以下是运单
func (s *DefScheduler) OnWaybillStatusChanged(bill *model.Waybill) (err error) {
if bill.Status == model.WaybillStatusAccepted {
s.CurOrderManager.UpdateWaybillVendorID(bill)
} else if bill.Status == model.WaybillStatusAcceptCanceled {
bill.WaybillVendorID = model.VendorIDUnknown
s.CurOrderManager.UpdateWaybillVendorID(bill)
}
return nil
}
func (s *DefScheduler) getPurchasePlatformFromVendorID(vendorID int) scheduler.PurchasePlatformHandler {
return s.PurchasePlatformHandlers[vendorID]
}
// c.handleAutoAcceptOrder(order.VendorOrderID, order.VendorID, order.ConsigneeMobile, order.StoreID, db, func(isAccept bool) {
// // c.purchasePlatformHandlers[order.VendorID].AcceptOrRefuseOrder(order, isAccept)
// if isAccept {
// order.Status = model.OrderStatusAccepted
// } else {
// order.Status = model.OrderStatusFailed
// }
// })
func (s *DefScheduler) handleAutoAcceptOrder(orderID string, vendorID int, userMobile string, jxStoreID int, db orm.Ormer, handler func(accepted bool)) int {
handleType := 0
if userMobile != "" {
if db == nil {
db = orm.NewOrm()
}
user := &models.BlackClient{
Mobile: userMobile,
}
if err := db.Read(user, "Mobile"); err != nil {
if err != orm.ErrNoRows {
globals.SugarLogger.Errorf("read data error:%v, data:%v, vendorID:%d", err, user, vendorID)
}
// 在访问数据库出错的情况下，也需要自动接单
handleType = 1
} else {
// 强制拒单
globals.SugarLogger.Infof("force reject order:%s, vendorID:%d", orderID, vendorID)
handleType = -1
}
} else {
globals.SugarLogger.Infof("order:%s, vendorID:%d, mobile is empty, should accept order", orderID, vendorID)
handleType = 1
}
if handleType == 1 {
handler(true)
} else if handleType == -1 {
handler(false)
}
return handleType
}

24
business/scheduler/scheduler.go
View File

@@ -41,7 +41,8 @@ type DeliveryPlatformHandler interface {
}
type OrderManager interface {
MasterWaybillStatusChanged(*model.Waybill) (err error)
OnOrderStatusChanged(status *model.OrderStatus) (err error) // 此消息是否使用还不确定
UpdateWaybillVendorID(bill *model.Waybill) (err error)
}
type Scheduler interface {
@@ -51,7 +52,6 @@ type Scheduler interface {
// 以下是订单
OnOrderNew(order *model.GoodsOrder) (err error)
OnOrderAccepted(order *model.GoodsOrder) (err error)
OnOrderStatusChanged(status *model.OrderStatus) (err error)
// 以下是运单
@@ -59,36 +59,36 @@ type Scheduler interface {
}
type BaseScheduler struct {
orderManager OrderManager
purchasePlatformHandlers map[int]PurchasePlatformHandler
deliveryPlatformHandlers map[int]DeliveryPlatformHandler
CurOrderManager OrderManager
PurchasePlatformHandlers map[int]PurchasePlatformHandler
DeliveryPlatformHandlers map[int]DeliveryPlatformHandler
}
func (c *BaseScheduler) Init() {
c.purchasePlatformHandlers = make(map[int]PurchasePlatformHandler)
c.deliveryPlatformHandlers = make(map[int]DeliveryPlatformHandler)
c.PurchasePlatformHandlers = make(map[int]PurchasePlatformHandler)
c.DeliveryPlatformHandlers = make(map[int]DeliveryPlatformHandler)
}
func (c *BaseScheduler) RegisterOrderManager(handler OrderManager) {
c.orderManager = handler
c.CurOrderManager = handler
}
func (c *BaseScheduler) RegisterPurchasePlatform(vendorID int, handler PurchasePlatformHandler) {
if !(vendorID >= model.VendorIDPurchaseBegin && vendorID <= model.VendorIDPurchaseEnd) {
panic(fmt.Sprintf("purchase vendor:%d is illegal", vendorID))
}
if _, ok := c.purchasePlatformHandlers[vendorID]; ok {
if _, ok := c.PurchasePlatformHandlers[vendorID]; ok {
panic(fmt.Sprintf("purchase vendor:%d, already exists", vendorID))
}
c.purchasePlatformHandlers[vendorID] = handler
c.PurchasePlatformHandlers[vendorID] = handler
}
func (c *BaseScheduler) RegisterDeliveryPlatform(vendorID int, handler DeliveryPlatformHandler) {
if !(vendorID >= model.VendorIDDeliveryBegin && vendorID <= model.VendorIDDeliveryEnd) {
panic(fmt.Sprintf("delivery vendor:%d is illegal", vendorID))
}
if _, ok := c.deliveryPlatformHandlers[vendorID]; ok {
if _, ok := c.DeliveryPlatformHandlers[vendorID]; ok {
panic(fmt.Sprintf("delivery vendor:%d, already exists", vendorID))
}
c.deliveryPlatformHandlers[vendorID] = handler
c.DeliveryPlatformHandlers[vendorID] = handler
}