- refactor file structure.

This commit is contained in:
gazebo
2018-08-17 13:59:15 +08:00
parent d872d4d0ec
commit d40186e930
25 changed files with 153 additions and 150 deletions

View File

@@ -0,0 +1,359 @@
package orderman
import (
"fmt"
"time"
"git.rosy.net.cn/baseapi"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/jxcallback/scheduler"
"git.rosy.net.cn/jx-callback/business/jxutils"
"git.rosy.net.cn/jx-callback/business/model"
"git.rosy.net.cn/jx-callback/globals"
"github.com/astaxie/beego/orm"
)
func (c *OrderManager) LoadPendingOrders() []*model.GoodsOrder {
db := orm.NewOrm()
var orders []*model.GoodsOrder
tillTime := time.Now().Add(-pendingOrderGapMax)
_, err := db.Raw(`
SELECT *
FROM goods_order
WHERE order_created_at >= ?
AND status < ?
`, tillTime, model.OrderStatusEndBegin).QueryRows(&orders)
if err != nil {
globals.SugarLogger.Warnf("LoadPendingOrders load pending orders error:%v", err)
return nil
}
for _, order := range orders {
utils.CallFuncLogError(func() error {
_, err = db.QueryTable("order_sku").Filter("vendor_order_id", order.VendorOrderID).Filter("vendor_id", order.VendorID).All(&order.Skus)
return err
}, "LoadPendingOrders order:%v", order)
}
return orders
}
// msgVendorStatus的意思是事件本身的类型类似有时收到NewOrder事件去取订单状态不一定就是New的
// OnOrderAdjust也类似而OrderStatus要记录的是消息所以添加这个
func (c *OrderManager) OnOrderNew(order *model.GoodsOrder, msgVendorStatus string) (err error) {
// todo transaction
db := orm.NewOrm()
if order.Status == model.OrderStatusUnknown {
order.Status = model.OrderStatusNew
}
status := model.Order2Status(order)
status.Status = model.OrderStatusNew
status.VendorStatus = msgVendorStatus
isDuplicated, err := addOrderOrWaybillStatus(status, db)
if err == nil && !isDuplicated {
if isDuplicated, err = c.saveOrder(order, false, db); err == nil && !isDuplicated {
err = scheduler.CurrentScheduler.OnOrderNew(order, false)
}
}
return err
}
// todo 调整单的处理可能还需要再细化一点,当前只是简单的删除重建
func (c *OrderManager) OnOrderAdjust(order *model.GoodsOrder, msgVendorStatus string) (err error) {
// todo transaction
db := orm.NewOrm()
if order.Status == model.OrderStatusUnknown {
order.Status = model.OrderStatusNew
}
status := model.Order2Status(order)
status.Status = model.OrderStatusAdjust
status.VendorStatus = msgVendorStatus
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()
return err
}, "OnAdjustOrder delete order, orderID:%s", order.VendorOrderID)
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
}, "OnAdjustOrder delete order_sku, orderID:%s", order.VendorOrderID)
if err != nil {
return err
}
if isDuplicated, err = c.saveOrder(order, true, db); err == nil && !isDuplicated {
// 因为订单调度器需要的是真实状态所以用order的状态
err = scheduler.CurrentScheduler.OnOrderNew(order, false)
err = scheduler.CurrentScheduler.OnOrderStatusChanged(model.Order2Status(order), false)
}
}
return err
}
func (c *OrderManager) OnOrderStatusChanged(orderStatus *model.OrderStatus) (err error) {
isDuplicated, err := c.addOrderStatus(orderStatus, nil)
if err == nil && !isDuplicated {
err = scheduler.CurrentScheduler.OnOrderStatusChanged(orderStatus, false)
if globals.GenerateLegacyJxOrder {
c.legacyJxOrderStatusChanged(orderStatus, nil)
}
}
return err
}
// private
func (c *OrderManager) saveOrder(order *model.GoodsOrder, isAdjust bool, db orm.Ormer) (isDuplicated bool, err error) {
// 忽略查找JX信息错误
c.updateOrderOtherInfo(order, db)
order.ID = 0
order.WaybillVendorID = model.VendorIDUnknown
order.OrderFinishedAt = utils.DefaultTimeValue
order.OrderCreatedAt = order.StatusTime
// hardcode 兼容京东消息错序问题
orderStatus := &model.OrderStatus{}
if db.Raw(`
SELECT *
FROM order_status
WHERE order_type = ? AND vendor_order_id = ? AND vendor_id = ?
ORDER BY status_time DESC
LIMIT 1
`, model.OrderTypeOrder, order.VendorOrderID, order.VendorID).QueryRow(orderStatus) == nil {
order.Status = orderStatus.Status
order.VendorStatus = orderStatus.VendorStatus
order.StatusTime = orderStatus.StatusTime
}
globals.SugarLogger.Debugf("saveOrder isAdjust:%t, order:%v", isAdjust, order)
db.Begin()
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, sku_type, promotion_type, 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, sku.SkuType, sku.PromotionType, order.StatusTime)
}
sql = sql[:len(sql)-1] + ";"
if _, err = db.Raw(sql, params...).Exec(); err != nil {
db.Rollback()
baseapi.SugarLogger.Infof("saveOrder insert order:%v, order_sku error:%v", order, err)
} else {
db.Commit()
if globals.GenerateLegacyJxOrder {
c.legacyWriteJxOrder(order, db, isAdjust)
}
}
} else {
isDuplicated = true
order.DuplicatedCount++
db.Update(order, "DuplicatedCount")
db.Commit()
baseapi.SugarLogger.Infof("saveOrder duplicated orderid:%s msg received", order.VendorOrderID)
}
} else {
db.Rollback()
globals.SugarLogger.Warnf("saveOrder create order:%v, error:%v", order, err)
}
return isDuplicated, err
}
func (c *OrderManager) updateOrderSkuOtherInfo(order *model.GoodsOrder, db orm.Ormer) (err error) {
jxStoreID := jxutils.GetJxStoreIDFromOrder(order)
if jxStoreID == 0 {
globals.SugarLogger.Infof("updateOrderSkuOtherInfo [运营]订单在京西与平台都找不到京西门店信息orderID:%s, VendorStoreID:%s", order.VendorOrderID, order.VendorStoreID)
return nil
}
orderSkus := order.Skus
var sql string
if orderSkus[0].VendorID == model.VendorIDJD {
sql = `
SELECT t1.jdskuid, t1.skuid, t2.price
FROM skumapper t1
LEFT JOIN jx_sku_store_bind t2 ON t1.skuid = t2.jxskuid AND t2.jxstoreid = ?
WHERE t1.jdskuid IN (-1,`
} else if orderSkus[0].VendorID == model.VendorIDELM {
sql = `
SELECT t2.jxskuid, t2.jxskuid, t2.price
FROM jx_sku_store_bind t2
WHERE t2.jxstoreid = ? AND t2.jxskuid IN (-1,`
} else {
panic(fmt.Sprintf("wrong vendorid:%d", orderSkus[0].VendorID))
}
sqlParams := []interface{}{jxStoreID}
for _, v := range orderSkus {
if orderSkus[0].VendorID == model.VendorIDJD {
sql += "?,"
sqlParams = append(sqlParams, int(utils.Str2Int64(v.VendorSkuID)))
} else if v.SkuID != 0 {
sql += "?,"
sqlParams = append(sqlParams, v.SkuID)
}
}
sql = sql[:len(sql)-1] + ")"
var lists []orm.ParamsList
if num, err := db.Raw(sql, sqlParams...).ValuesList(&lists); err == nil {
skumapper := make(map[string]orm.ParamsList)
for _, v := range lists {
skumapper[v[0].(string)] = v
}
// globals.SugarLogger.Debug(skumapper)
for _, v := range orderSkus {
if orderSkus[0].VendorID == model.VendorIDJD {
if values, ok := skumapper[v.VendorSkuID]; ok {
v.JxSkuID = int(utils.Str2Int64(utils.Interface2String(values[1])))
v.ShopPrice = utils.Str2Int64WithDefault(utils.Interface2String(values[2]), 0)
order.ShopPrice += v.ShopPrice
if v.ShopPrice == 0 {
globals.SugarLogger.Infof("updateOrderSkuOtherInfo [运营2]京东订单sku门店价格为零一般原因为没有门店价格信息orderID:%s, StoreID:%d, SkuID:%d, sku:%v", order.VendorOrderID, jxStoreID, v.JxSkuID, v)
}
} else {
globals.SugarLogger.Infof("updateOrderSkuOtherInfo [运营2]京东订单sku找不到门店价格或商品映射orderID:%s, StoreID:%d, VendorSkuID:%s, sku:%v", order.VendorOrderID, jxStoreID, v.VendorSkuID, v)
}
} else {
if v.SkuID != 0 {
if values, ok := skumapper[utils.Int2Str(v.SkuID)]; ok {
v.ShopPrice = utils.Str2Int64WithDefault(utils.Interface2String(values[2]), 0)
order.ShopPrice += v.ShopPrice
if v.ShopPrice == 0 {
globals.SugarLogger.Infof("updateOrderSkuOtherInfo [运营2]饿了么订单sku门店价格为零orderID:%s, StoreID:%d, SkuID:%d, sku:%v", order.VendorOrderID, jxStoreID, v.SkuID, v)
}
} else {
globals.SugarLogger.Infof("updateOrderSkuOtherInfo [运营2]饿了么订单sku找不到门店价格orderID:%s, StoreID:%d, SkuID:%d, sku:%v", order.VendorOrderID, jxStoreID, v.SkuID, v)
}
} else {
globals.SugarLogger.Infof("updateOrderSkuOtherInfo [运营2]饿了么订单sku没有京西ID信息orderID:%s, VendorSkuID:%s, sku:%v", order.VendorOrderID, v.VendorSkuID, v)
}
}
}
} else {
globals.SugarLogger.Errorf("updateOrderSkuOtherInfo can not get sku info for orderID:%s, num:%d, error:%v", order.VendorOrderID, num, err)
}
return err
}
func (c *OrderManager) updateOrderOtherInfo(order *model.GoodsOrder, db orm.Ormer) (err error) {
var sql string
if order.VendorID == model.VendorIDJD {
sql = `
SELECT t1.jxstoreid
FROM jxstoremap t1
/* JOIN jxstore t2 ON t1.jxstoreid = t2.storeid */
WHERE t1.jdstoreid = ?
`
} else if order.VendorID == model.VendorIDELM {
sql = `
SELECT t1.jx_store_id
FROM jx_to_elm_store_map t1
/* JOIN jxstore t2 ON t1.jx_store_id = t2.storeid */
WHERE t1.elm_store_id = ?
`
} else {
panic(fmt.Sprintf("wrong vendorid:%d", order.VendorID))
}
var lists []orm.ParamsList
if num, err := db.Raw(sql, utils.Str2Int64(order.VendorStoreID)).ValuesList(&lists); err == nil && num == 1 {
order.JxStoreID = int(utils.Str2Int64(lists[0][0].(string)))
} else {
globals.SugarLogger.Infof("updateOrderOtherInfo [运营]订单orderID:%s在京西找不到相应的京西门店信息请处理, VendorStoreID:%s, num:%d, error:%v", order.VendorOrderID, order.VendorStoreID, num, err)
}
err = c.updateOrderSkuOtherInfo(order, db)
return err
}
func (c *OrderManager) addOrderStatus(orderStatus *model.OrderStatus, db orm.Ormer) (isDuplicated bool, err error) {
if db == nil {
db = orm.NewOrm()
}
isDuplicated, err = addOrderOrWaybillStatus(orderStatus, db)
if err == nil && !isDuplicated &&
(orderStatus.Status > model.OrderStatusUnknown ||
(orderStatus.Status == model.OrderStatusUnlocked || orderStatus.Status == model.OrderStatusLocked || orderStatus.Status == model.OrderStatusApplyCancel)) {
order := &model.GoodsOrder{
VendorOrderID: orderStatus.VendorOrderID,
VendorID: orderStatus.VendorID,
}
if err = db.ReadForUpdate(order, "VendorOrderID", "VendorID"); err == nil {
if (orderStatus.Status == model.OrderStatusUnlocked || orderStatus.Status == model.OrderStatusLocked || orderStatus.Status == model.OrderStatusApplyCancel) ||
(orderStatus.Status > model.OrderStatusUnknown && orderStatus.Status >= order.Status) { // todo 要求status不能回绕
order.VendorStatus = orderStatus.VendorStatus
order.StatusTime = orderStatus.StatusTime
updateFields := []string{
"VendorStatus", "StatusTime",
}
if orderStatus.Status > model.OrderStatusUnknown {
order.LockStatus = model.OrderStatusUnknown
order.Status = orderStatus.Status
updateFields = append(updateFields, "Status", "LockStatus")
} else {
if orderStatus.Status == model.OrderStatusUnlocked {
order.LockStatus = model.OrderStatusUnknown
updateFields = append(updateFields, "LockStatus")
} else if orderStatus.Status == model.OrderStatusLocked || orderStatus.Status == model.OrderStatusApplyCancel {
order.LockStatus = orderStatus.Status
updateFields = append(updateFields, "LockStatus")
}
}
if orderStatus.Status >= model.OrderStatusEndBegin {
order.OrderFinishedAt = orderStatus.StatusTime
updateFields = append(updateFields, "OrderFinishedAt")
}
utils.CallFuncLogError(func() error {
_, err = db.Update(order, updateFields...)
return err
}, "addOrderStatus update orderID:%s, status:%v", order.VendorOrderID, orderStatus)
} else {
isDuplicated = true
}
} else {
if err == orm.ErrNoRows { // todo 消息错序
err = nil
} else {
globals.SugarLogger.Warnf("addOrderStatus orderID:%s read failed with error:%v", order.VendorOrderID, err)
}
}
}
return isDuplicated, err
}
func (c *OrderManager) LoadOrder(vendorOrderID string, vendorID int) (order *model.GoodsOrder, err error) {
db := orm.NewOrm()
order = &model.GoodsOrder{
VendorOrderID: vendorOrderID,
VendorID: vendorID,
}
if err = db.Read(order, "VendorOrderID", "VendorID"); err == nil {
err = utils.CallFuncLogError(func() error {
_, err = db.QueryTable("order_sku").Filter("vendor_order_id", vendorOrderID).Filter("vendor_id", vendorID).All(&order.Skus)
return err
}, "LoadOrder orderID:%s", vendorOrderID)
}
if err != nil {
globals.SugarLogger.Infof("LoadOrder orderID:%s failed with error:%v", vendorOrderID, err)
}
return order, err
}
//Waybill
func (c *OrderManager) UpdateWaybillVendorID(bill *model.Waybill, revertStatus bool) (err error) {
globals.SugarLogger.Debugf("UpdateWaybillVendorID bill:%v", bill)
db := orm.NewOrm()
params := orm.Params{
"vendor_waybill_id": bill.VendorWaybillID,
"waybill_vendor_id": bill.WaybillVendorID,
}
// 如果运单被取消,则要保持在已拣货状态
if revertStatus && bill.WaybillVendorID == model.VendorIDUnknown {
params["status"] = model.OrderStatusFinishedPickup
}
utils.CallFuncLogError(func() error {
_, err = db.QueryTable("goods_order").Filter("vendor_order_id", bill.VendorOrderID).Filter("vendor_id", bill.OrderVendorID).Update(params)
return err
}, "UpdateWaybillVendorID update order, bill:%v", bill)
return err
}

View File

@@ -0,0 +1,315 @@
package orderman
import (
"strings"
"git.rosy.net.cn/baseapi"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/jxutils"
"git.rosy.net.cn/jx-callback/business/legacymodel"
"git.rosy.net.cn/jx-callback/business/model"
"git.rosy.net.cn/jx-callback/globals"
"github.com/astaxie/beego/orm"
)
// legacy
const (
JX_ORDER_STATUS_WAIT_TO_ACCEPT = -1 //未接单(41000)
JX_ORDER_STATUS_PICKING = 0 //拣货(32000->或定时召唤达达)
JX_ORDER_STATUS_WAIT_FOR_DELIVERY = 1 //待配送(2->或转商家自送)
JX_ORDER_STATUS_DELIVERING = 2 //配送中(33040)
JX_ORDER_STATUS_DELIVERY_DONE = 3 //已完成(33060->90000)
JX_ORDER_STATUS_EXCEPTION_APPLY = 4 //异常申请
JX_ORDER_STATUS_NOT_DELIVER = 5 //不配送
JX_ORDER_STATUS_AFTER_SALE = 6 //售后单
JX_ORDER_STATUS_CANCEL = 7 //已取消
JX_ORDER_STATUS_ADJUST = 8 //调整单
JX_DELIVERY_STATUS_NOT_DELIVERY = -1 //尚未申请配送
JX_DELIVERY_STATUS_WAIT_TO_GRAB = 0 //等待抢单
JX_DELIVERY_STATUS_GRAB_DONE = 1 //已抢单
JX_DELIVERY_STATUS_FAIL_TO_GET_GOODS = 2 //取货失败
JX_DELIVERY_STATUS_FAIL_TO_GET_GOODS_WAIT_TO_CONFIRM = 3 //取货失败待审核
JX_DELIVERY_STATUS_GET_GOODS_DONE = 4 //取货完成
JX_DELIVERY_STATUS_DELIVERY_FAIL = 5 //投递失败
JX_DELIVERY_STATUS_DELIVERY_DONE = 6 //已完成
JX_DELIVERY_STATUS_DELIVERY_CANCEL = 7 //已取消
JX_STATUS_UNKNOWN = -100
)
const (
DD_CARRIER_NO = "9966"
SELLER_CARRIER_NO = "2938"
MTPS_CARRIER_NO = "1123"
DDDELIVERY_CARRIER_NO = "3465"
EME_SELF_CARRIER_NO = "9999"
)
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: "门店自送",
MTPS_CARRIER_NO: "美团配送",
DDDELIVERY_CARRIER_NO: "达达众包",
EME_SELF_CARRIER_NO: "饿了么蜂鸟",
}
BUSINESS_TAGS_MAP = map[string]string{
"one_dingshida": "0",
"dj_aging_nextday": "1",
"dj_aging_immediately": "2",
"lengcang": "3",
"lengdong": "3",
}
)
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
case model.OrderStatusAccepted:
retVal = JX_ORDER_STATUS_PICKING
case model.OrderStatusDelivering:
retVal = JX_ORDER_STATUS_DELIVERING
case model.OrderStatusDelivered, model.OrderStatusFinished:
retVal = JX_ORDER_STATUS_DELIVERY_DONE
case model.OrderStatusAdjust:
retVal = JX_ORDER_STATUS_ADJUST
case model.OrderStatusApplyCancel:
retVal = JX_ORDER_STATUS_EXCEPTION_APPLY
case model.OrderStatusCanceled:
retVal = JX_ORDER_STATUS_CANCEL
case model.OrderStatusFinishedPickup:
retVal = JX_ORDER_STATUS_WAIT_FOR_DELIVERY
default:
retVal = JX_STATUS_UNKNOWN
}
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
default:
retVal = JX_STATUS_UNKNOWN
}
return retVal
}
func (c *OrderManager) legacyWriteJxOrder(order *model.GoodsOrder, db orm.Ormer, isDelFirst bool) (err error) {
var result map[string]interface{}
businessTags := ""
if order.VendorID == model.VendorIDJD && utils.UnmarshalUseNumber([]byte(order.OriginalData), &result) == nil {
tagList := strings.Split(result["businessTag"].(string), ";")
list := []string{}
for _, v := range tagList {
if tag, ok := BUSINESS_TAGS_MAP[v]; ok {
list = append(list, tag)
}
}
businessTags = strings.Join(list, "|") + "|"
}
db.Begin()
if isDelFirst {
db.Raw("DELETE FROM "+globals.JxorderTableName+" WHERE order_id = ?", utils.Str2Int64(order.VendorOrderID)).Exec()
db.Raw("DELETE FROM "+globals.JxorderskuTableName+" WHERE order_id = ?", utils.Str2Int64(order.VendorOrderID)).Exec()
}
orderStatus := legacyMapOrderStatus(order.Status)
if orderStatus < JX_ORDER_STATUS_PICKING {
orderStatus = JX_ORDER_STATUS_PICKING
}
jxorder := &legacymodel.Jxorder2{
VenderId: int8(order.VendorID),
OrderId: utils.Str2Int64(order.VendorOrderID),
JxStoreId: utils.Int2Str(jxutils.GetJxStoreIDFromOrder(order)),
JxStoreName: order.StoreName,
OrderNum: order.OrderSeq,
OrderStatus: orderStatus,
OrderStatusTime: utils.Time2Str(order.StatusTime),
BusinessTag: businessTags,
SkuCount: order.SkuCount,
OrderBuyerRemark: order.BuyerComment,
BuyerFullName: order.ConsigneeName,
BuyerFullAddress: order.ConsigneeAddress,
BuyerMobile: order.ConsigneeMobile,
BuyerCoordType: legacyMapCoordinateType(order.CoordinateType),
BuyerLng: jxutils.IntCoordinate2Standard(order.ConsigneeLng),
BuyerLat: jxutils.IntCoordinate2Standard(order.ConsigneeLat),
CityName: "all",
OrderStartTime: utils.Time2Str(order.StatusTime),
JdStoreId: order.VendorStoreID,
OrderTotalMoney: int(order.SalePrice),
OrderDiscountMoney: int(order.SalePrice - order.ActualPayPrice),
OrderBuyerPayableMoney: int(order.ActualPayPrice),
// DeliveryPackageWeight: float64(order.Weight) / 1000,
}
if order.VendorID == model.VendorIDELM {
jxorder.OrderStatus = JX_ORDER_STATUS_WAIT_FOR_DELIVERY
}
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:%v error:%v", jxorder, err)
return err
}
sql := "INSERT INTO " + globals.JxorderskuTableName + "(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, jxutils.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 {
db.Rollback()
baseapi.SugarLogger.Infof("insert jxordersku jxorder:%v error:%v", jxorder, err)
} else {
db.Commit()
vendorOrderID := utils.Str2Int64(order.VendorOrderID)
utils.CallFuncLogError(func() error {
_, err = db.Raw(`
UPDATE `+globals.JxorderTableName+` t1
JOIN jxstore t2 ON t2.storeid = t1.jx_store_id
SET t1.store_lng = t2.lng,
t1.store_lat = t2.lat
WHERE t1.order_id = ?;
`, vendorOrderID /*, vendorOrderID*/).Exec()
return err
}, "update jxorder")
utils.CallFuncLogError(func() error {
_, err = db.Raw(`
UPDATE `+globals.JxorderskuTableName+` t1
JOIN jx_sku t2 ON t2.id = t1.jx_sku_id
JOIN jx_sku_name t3 ON t3.id = t2.nameid
SET t1.sku_img = t3.img
WHERE t1.order_id = ?;
`, vendorOrderID /*, vendorOrderID*/).Exec()
return err
}, "update jxordersku")
}
return err
}
func (c *OrderManager) legacyJxOrderStatusChanged(status *model.OrderStatus, db orm.Ormer) (err error) {
orderStatus := legacyMapOrderStatus(status.Status)
if orderStatus != JX_STATUS_UNKNOWN {
if status.VendorID == model.VendorIDELM {
if orderStatus < JX_ORDER_STATUS_WAIT_FOR_DELIVERY {
orderStatus = JX_ORDER_STATUS_WAIT_FOR_DELIVERY
}
}
if db == nil {
db = orm.NewOrm()
}
jxorder := &legacymodel.Jxorder2{
OrderId: utils.Str2Int64(status.VendorOrderID),
}
if err = db.Read(jxorder, "OrderId"); err == nil {
utils.CallFuncLogError(func() error {
jxorder.OrderStatus = orderStatus
jxorder.OrderStatusTime = utils.Time2Str(status.StatusTime)
updateFields := []string{
"OrderStatus",
"OrderStatusTime",
}
if status.Status >= model.OrderStatusEndBegin {
jxorder.DeliveryFinishTime = utils.Time2Str(status.StatusTime)
updateFields = append(updateFields, "DeliveryFinishTime")
}
_, err = db.Update(jxorder, updateFields...)
if orderStatus == JX_ORDER_STATUS_DELIVERY_DONE || orderStatus == JX_ORDER_STATUS_DELIVERING {
db.Raw(`
UPDATE `+globals.JxorderTableName+` t1
JOIN waybill t2 ON t2.vendor_order_id = t1.order_id AND t2.status = 105
SET t1.delivery_price = IF(t2.waybill_vendor_id = 102, t2.desired_fee/100, t1.delivery_price),
t1.delivery_price1 = IF(t2.waybill_vendor_id = 101, t2.desired_fee/100, t1.delivery_price1)
WHERE t1.order_id = ?
`, jxorder.OrderId).Exec()
}
return err
}, "legacyJxOrderStatusChanged")
} else {
globals.SugarLogger.Infof("read legacyJxOrder orderID:%d error:%v, ", jxorder.OrderId, err)
}
}
return err
}
func (c *OrderManager) legacyWaybillStatusChanged(bill *model.Waybill, db orm.Ormer) (err error) {
deliveryStatus := legacyMapWaybillStatus(bill.Status)
if deliveryStatus != JX_STATUS_UNKNOWN {
if db == nil {
db = orm.NewOrm()
}
jxorder := &legacymodel.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 = deliveryStatus
// jxorder.DeliveryConfirmTime
if bill.Status == model.WaybillStatusNew {
updateFields = append(updateFields, "DeliveryStartTime")
jxorder.DeliveryStartTime = utils.Time2Str(bill.StatusTime)
} else if bill.Status >= model.WaybillStatusEndBegin {
jxorder.DeliveryFinishTime = utils.Time2Str(bill.StatusTime)
updateFields = append(updateFields, "DeliveryFinishTime")
}
_, err = db.Update(jxorder, updateFields...)
return err
}, "legacyJxOrderStatusChanged")
} else {
globals.SugarLogger.Infof("read legacyJxOrder, orderID:%d error:%v", jxorder.OrderId, err)
}
}
return err
}

View File

@@ -0,0 +1,144 @@
package orderman
import (
"sort"
"time"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/jxcallback/scheduler"
_ "git.rosy.net.cn/jx-callback/business/jxcallback/scheduler/defsch" // 导入缺省订单调度器
"git.rosy.net.cn/jx-callback/business/jxutils"
"git.rosy.net.cn/jx-callback/business/model"
"git.rosy.net.cn/jx-callback/business/partner"
"git.rosy.net.cn/jx-callback/globals"
"github.com/astaxie/beego/orm"
)
const (
pendingOrderGapMax = 2 * 24 * time.Hour // 每次重启机子时,要检查几天内的订单状态
maxTimeHandlePendingOrder = 2 * time.Second //处理pending order的最长时间
maxSleepGapHandlePendingOrder = 5 * time.Millisecond // 每个pending order的最长时间间隙
)
var (
CurOrderManager *OrderManager
)
// 所有公共接口调用前要求在order里或status中设置合适的Status
type OrderManager struct {
}
func NewOrderManager() *OrderManager {
return &OrderManager{}
}
type StatusTimer interface {
GetStatusTime() time.Time
}
type StatusTimerSlice []StatusTimer
func (s StatusTimerSlice) Len() int {
return len(s)
}
func (s StatusTimerSlice) Less(i, j int) bool {
return s[i].GetStatusTime().Sub(s[j].GetStatusTime()) < 0
}
func (s StatusTimerSlice) Swap(i, j int) {
tmp := s[i]
s[i] = s[j]
s[j] = tmp
}
func init() {
CurOrderManager = NewOrderManager()
partner.Init(CurOrderManager)
}
func addOrderOrWaybillStatus(status *model.OrderStatus, db orm.Ormer) (isDuplicated bool, err error) {
if status.OrderType == model.OrderTypeOrder {
globals.SugarLogger.Debugf("addOrderStatus order:%v", status)
} else {
globals.SugarLogger.Debugf("addOrderStatus waybill:%v", status)
}
status.ID = 0
created, _, err := db.ReadOrCreate(status, "VendorOrderID", "VendorID", "OrderType", "VendorStatus", "StatusTime")
if err == nil {
if !created {
globals.SugarLogger.Debugf("duplicated event:%v", status)
isDuplicated = true
status.DuplicatedCount++
utils.CallFuncLogError(func() error {
_, err = db.Update(status, "DuplicatedCount")
return err
}, "addOrderOrWaybillStatus update DuplicatedCount, status:%v", status)
}
}
if err != nil {
// todo 这里居然会有主键重复错误,逻辑上是不应该的
globals.SugarLogger.Warnf("addOrderOrWaybillStatus status:%v, access db error:%v", status, err)
}
return isDuplicated, err
}
// todo 最好还是改成全事件回放算了
func LoadPendingOrders() {
orders := CurOrderManager.LoadPendingOrders()
globals.SugarLogger.Infof("LoadPendingOrders orders count:%d", len(orders))
ordersCount := len(orders)
if ordersCount > 0 {
bills := CurOrderManager.LoadPendingWaybills()
globals.SugarLogger.Infof("LoadPendingOrders waybills count:%d", len(bills))
var sortOrders StatusTimerSlice
for _, order := range orders {
if order.Status > model.OrderStatusNew {
status := model.Order2Status(order)
sortOrders = append(sortOrders, status)
}
// order.Status = model.OrderStatusNew // 就是要以实际order状态来调用scheduler.OnOrderNew
order.StatusTime = order.OrderCreatedAt
sortOrders = append(sortOrders, order)
}
for _, bill := range bills {
if bill.Status > model.WaybillStatusNew {
bill2 := *bill
sortOrders = append(sortOrders, &bill2)
}
bill.Status = model.WaybillStatusNew
bill.StatusTime = bill.WaybillCreatedAt
sortOrders = append(sortOrders, bill)
}
sort.Sort(sortOrders)
sleepGap := maxTimeHandlePendingOrder / time.Duration(ordersCount)
if sleepGap > maxSleepGapHandlePendingOrder {
sleepGap = maxSleepGapHandlePendingOrder
}
lastTime := time.Now()
for _, item := range sortOrders {
if order, ok := item.(*model.GoodsOrder); ok {
jxutils.CallMsgHandlerAsync(func() {
scheduler.CurrentScheduler.OnOrderNew(order, true)
}, order.VendorOrderID)
} else if status, ok := item.(*model.OrderStatus); ok {
jxutils.CallMsgHandlerAsync(func() {
scheduler.CurrentScheduler.OnOrderStatusChanged(status, true)
}, status.VendorOrderID)
} else {
bill := item.(*model.Waybill)
jxutils.CallMsgHandlerAsync(func() {
scheduler.CurrentScheduler.OnWaybillStatusChanged(bill, true)
}, bill.VendorOrderID)
}
curTime := time.Now()
timeout := sleepGap - curTime.Sub(lastTime)
if timeout > 0 {
time.Sleep(timeout)
}
lastTime = curTime
}
}
}

View File

@@ -0,0 +1,109 @@
package orderman
import (
"time"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/jxcallback/scheduler"
"git.rosy.net.cn/jx-callback/business/model"
"git.rosy.net.cn/jx-callback/globals"
"github.com/astaxie/beego/orm"
)
func (w *OrderManager) LoadPendingWaybills() []*model.Waybill {
db := orm.NewOrm()
var bills []*model.Waybill
tillTime := time.Now().Add(-pendingOrderGapMax)
_, err := db.Raw(`
SELECT t1.*
FROM waybill t1
JOIN goods_order t2 ON t2.vendor_order_id = t1.vendor_order_id
AND t2.vendor_id = t1.order_vendor_id
AND t2.order_created_at >= ?
AND t2.status < ?
WHERE t1.waybill_created_at >= ?
AND t1.status < ?
`, tillTime, model.OrderStatusEndBegin, tillTime, model.WaybillStatusEndBegin).QueryRows(&bills)
if err != nil {
globals.SugarLogger.Warnf("LoadPendingWaybills load pending waybills error:%v", err)
return nil
}
return bills
}
func (w *OrderManager) onWaybillNew(bill2 *model.Waybill, db orm.Ormer) (isDuplicated bool, err error) {
globals.SugarLogger.Debugf("onWaybillNew bill:%v", bill2)
isDuplicated, err = addOrderOrWaybillStatus(model.Waybill2Status(bill2), db)
if err == nil && !isDuplicated {
bill2.ID = 0
bill2.WaybillCreatedAt = bill2.StatusTime
bill2.WaybillFinishedAt = utils.DefaultTimeValue
billCopied := *bill2
bill := &billCopied
created, _, err2 := db.ReadOrCreate(bill, "VendorWaybillID", "WaybillVendorID")
if err = err2; err == nil {
if !created {
bill.DuplicatedCount++
if bill2.VendorOrderID == bill2.VendorWaybillID { // 购物平台(比如京东)重新建的运单,单号始终是与订单相同的
bill2.ID = bill.ID
bill2.DuplicatedCount = bill.DuplicatedCount
db.Update(bill2) //更新所有字段
} else {
db.Update(bill, "DuplicatedCount")
isDuplicated = true
globals.SugarLogger.Infof("onWaybillNew duplicated bill:%v msg received", bill2)
}
} else {
*bill2 = *bill
}
} else {
globals.SugarLogger.Warnf("onWaybillNew create bill:%v, error:%v", bill2, err)
}
}
return isDuplicated, err
}
func (w *OrderManager) OnWaybillStatusChanged(bill *model.Waybill) (err error) {
var isDuplicated bool
db := orm.NewOrm()
if bill.Status == model.WaybillStatusNew {
isDuplicated, err = w.onWaybillNew(bill, db)
} else {
var addParams orm.Params
if bill.Status == model.WaybillStatusAccepted {
addParams = orm.Params{
"courier_name": bill.CourierName,
"courier_mobile": bill.CourierMobile,
"desired_fee": bill.DesiredFee,
}
}
isDuplicated, err = w.addWaybillStatus(bill, db, addParams)
}
if err == nil && !isDuplicated {
scheduler.CurrentScheduler.OnWaybillStatusChanged(bill, false)
if globals.GenerateLegacyJxOrder {
w.legacyWaybillStatusChanged(bill, db)
}
}
return err
}
func (w *OrderManager) addWaybillStatus(bill *model.Waybill, db orm.Ormer, addParams orm.Params) (isDuplicated bool, err error) {
waybillStatus := model.Waybill2Status(bill)
isDuplicated, err = addOrderOrWaybillStatus(waybillStatus, db)
if err == nil && !isDuplicated && waybillStatus.Status > model.WaybillStatusUnknown { // todo 这里应该和addOrderStatus一样的改法状态不能回绕
params := utils.MergeMaps(orm.Params{
"status": bill.Status,
"vendor_status": bill.VendorStatus,
"status_time": bill.StatusTime,
}, addParams)
if bill.Status >= model.WaybillStatusEndBegin {
params["waybill_finished_at"] = bill.StatusTime
}
utils.CallFuncLogError(func() error {
_, err = db.QueryTable("waybill").Filter("vendor_waybill_id", bill.VendorWaybillID).Filter("waybill_vendor_id", bill.WaybillVendorID).Update(params)
return err
}, "addWaybillStatus update waybill status, bill:%v", bill)
}
return isDuplicated, err
}