- handle pending orders and waybills when starting.

business/controller/controller.go

@@ -2,6 +2,7 @@ package controller
 
 import (
 	"fmt"
+	"time"
 
 	"git.rosy.net.cn/baseapi/utils"
 	"git.rosy.net.cn/baseapi/utils/routinepool"
@@ -12,6 +13,10 @@ import (
 	"github.com/astaxie/beego/orm"
 )
 
+const (
+	pendingOrderGapMax = 2 * 24 * time.Hour // 每次重启机子时，要检查几天内的订单状态
+)
+
 var (
 	OrderManager   *OrderController
 	WaybillManager *WaybillController

business/controller/elm/order.go

@@ -142,6 +142,7 @@ func (c *OrderController) getOrderInfo(orderID string) (order *model.GoodsOrder,
 			OrderSeq:              int(utils.MustInterface2Int64(result["daySn"])),
 			OrderCreatedAt:        utils.Str2Time(result["createdAt"].(string)),
 			OriginalData:          utils.FilterMb4(string(utils.MustMarshal(result))),
+			ActualPayPrice:        jxutils.StandardPrice2Int(utils.MustInterface2Float64(result["totalPrice"])),
 			Skus:                  []*model.OrderSku{},
 		}
 		deliveryGeo := strings.Split(utils.Interface2String(result["deliveryGeo"]), ",")

business/controller/jd/order.go

@@ -81,6 +81,7 @@ func (c *OrderController) getOrderInfo(msg *jdapi.CallbackOrderMsg) (order *mode
 			OrderSeq:              int(utils.MustInterface2Int64(result["orderNum"])),
 			OrderCreatedAt:        utils.Str2Time(result["orderStartTime"].(string)),
 			OriginalData:          utils.FilterMb4(string(utils.MustMarshal(result))),
+			ActualPayPrice:        utils.MustInterface2Int64(result["orderBuyerPayableMoney"]),
 			Skus:                  []*model.OrderSku{},
 		}
 		coordinateType := utils.Interface2Int64WithDefault(result["buyerCoordType"], 1)
@@ -166,7 +167,7 @@ func (c *OrderController) callbackMsg2Status(msg *jdapi.CallbackOrderMsg) *model
 
 // PurchasePlatformHandler
 func (c *OrderController) AcceptOrRefuseOrder(order *model.GoodsOrder, isAcceptIt bool) (err error) {
-	globals.SugarLogger.Infof("AcceptOrRefuseOrder order:%v", order)
+	globals.SugarLogger.Infof("AcceptOrRefuseOrder order:%v", order.VendorOrderID)
 	return nil
 
 	_, err = api.JdAPI.OrderAcceptOperate(order.VendorOrderID, isAcceptIt)

business/controller/order.go

@@ -2,6 +2,7 @@ package controller
 
 import (
 	"fmt"
+	"time"
 
 	"git.rosy.net.cn/jx-callback/business/scheduler"
 
@@ -21,6 +22,29 @@ func NewOrderManager() *OrderController {
 	return &OrderController{}
 }
 
+func (c *OrderController) LoadPendingOrders() {
+	db := orm.NewOrm()
+	var orders []*model.GoodsOrder
+	_, err := db.Raw(`
+	SELECT *
+	FROM goods_order
+	WHERE order_created_at >= ?
+	AND status < ?
+	ORDER by order_created_at
+	`, time.Now().Add(-pendingOrderGapMax), model.OrderStatusEndBegin).QueryRows(&orders)
+	if err != nil {
+		globals.SugarLogger.Warnf("init load pending orders error:%v", err)
+		return
+	}
+	globals.SugarLogger.Info(len(orders))
+	for _, v := range orders {
+		v2 := v
+		routinePool.CallFunAsync(func() {
+			scheduler.CurrentScheduler.OnOrderNew(v2)
+		}, v2.VendorOrderID)
+	}
+}
+
 func (c *OrderController) OnOrderNew(order *model.GoodsOrder) (err error) {
 	db := orm.NewOrm()
 	order.StatusTime = order.OrderCreatedAt

business/controller/waybill.go

@@ -1,6 +1,8 @@
 package controller
 
 import (
+	"time"
+
 	"git.rosy.net.cn/baseapi/utils"
 	"git.rosy.net.cn/jx-callback/business/jxutils/weixinmsg"
 	"git.rosy.net.cn/jx-callback/business/model"
@@ -17,6 +19,40 @@ func NewWaybillManager() *WaybillController {
 	return &WaybillController{}
 }
 
+func (w *WaybillController) LoadPendingWaybills() {
+	db := orm.NewOrm()
+	var bills []*model.Waybill
+	_, 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 waybill_created_at >= ?
+	AND t1.status < ?
+	ORDER by waybill_created_at
+	`, time.Now().Add(-pendingOrderGapMax), model.WaybillStatusEndBegin,
+		time.Now().Add(-pendingOrderGapMax), model.WaybillStatusEndBegin).QueryRows(&bills)
+	if err != nil {
+		globals.SugarLogger.Warnf("init load pending waybills error:%v", err)
+		return
+	}
+	globals.SugarLogger.Info(len(bills))
+	for _, v := range bills {
+		v2 := v
+		routinePool.CallFunAsync(func() {
+			if v2.Status != model.WaybillStatusNew {
+				savedStatus := v2.Status
+				v2.Status = model.WaybillStatusNew
+				scheduler.CurrentScheduler.OnWaybillStatusChanged(v2)
+				v2.Status = savedStatus
+			}
+			scheduler.CurrentScheduler.OnWaybillStatusChanged(v2)
+		}, v2.VendorOrderID)
+	}
+}
+
 func (w *WaybillController) onWaybillNew(bill *model.Waybill) (err error) {
 	db := orm.NewOrm()
 	isDuplicated, err := addOrderOrWaybillStatus(model.Waybill2Status(bill), db)

business/jxutils/weixinmsg/weixinmsg.go

@@ -103,7 +103,7 @@ func NotifyNewOrder(order *model.GoodsOrder) (err error) {
 	sb.WriteString("份（")
 	sb.WriteString(jxutils.IntPrice2StandardString(order.Skus[0].SalePrice))
 	sb.WriteString("元/份）等，共支付了")
-	sb.WriteString(jxutils.IntPrice2StandardString(order.SalePrice))
+	sb.WriteString(jxutils.IntPrice2StandardString(order.ActualPayPrice))
 	sb.WriteString("元")
 	data := map[string]interface{}{
 		"first": map[string]interface{}{
@@ -139,6 +139,7 @@ func NotifyWaybillStatus(bill *model.Waybill, order *model.GoodsOrder) error {
 	var title string
 	var templateID string
 	remark := ""
+	titleColor := ""
 	switch bill.Status {
 	case model.WaybillStatusAccepted:
 		if bill.WaybillVendorID == model.VendorIDMTPS {
@@ -146,28 +147,30 @@ func NotifyWaybillStatus(bill *model.Waybill, order *model.GoodsOrder) error {
 		} else if bill.WaybillVendorID == model.VendorIDDada {
 			templateID = WX_DADA_DELIVERY_GRABDONE_TEMPLATE_ID
 		}
+		titleColor = WX_HIGHLEVEL_TEMPLATE_COLOR
 		remark = FormatDeliveryTime(order)
-		title = fmt.Sprintf("%s %s#订单长时间无人配送，我们已安排%s配送员%s电话号码%s负责配送。^_^", model.VendorChineseNames[bill.OrderVendorID], bill.VendorOrderID, model.VendorChineseNames[bill.WaybillVendorID], bill.CourierName, bill.CourierMobile)
+		title = fmt.Sprintf("%s %d#订单长时间无人配送，我们已安排%s配送员%s电话号码%s负责配送。^_^", model.VendorChineseNames[bill.OrderVendorID], order.OrderSeq, model.VendorChineseNames[bill.WaybillVendorID], bill.CourierName, bill.CourierMobile)
 	case model.WaybillStatusDelivered:
 		if bill.WaybillVendorID == model.VendorIDMTPS {
 			templateID = WX_MTPS_DELIVERY_DONE_TEMPLATE_ID
 		} else if bill.WaybillVendorID == model.VendorIDDada {
 			templateID = WX_DADA_DELIVERY_DONE_TEMPLATE_ID
 		}
+		titleColor = venderColors[bill.WaybillVendorID]
 		title = fmt.Sprintf("%s 第%d号订单的配送完成", model.VendorChineseNames[bill.OrderVendorID], order.OrderSeq)
 	}
 	if templateID != "" {
 		data := map[string]interface{}{
 			"first": map[string]interface{}{
 				"value": title,
-				"color": WX_HIGHLEVEL_TEMPLATE_COLOR,
+				"color": titleColor,
 			},
 			"keyword1": map[string]interface{}{
 				"value": bill.VendorOrderID,
 				"color": WX_NEW_ORDER_TEMPLATE_COLOR,
 			},
 			"keyword2": map[string]interface{}{
-				"value": bill.CourierName,
+				"value": fmt.Sprintf("%s（%s）", bill.CourierName, model.VendorChineseNames[bill.WaybillVendorID]),
 				"color": WX_NEW_ORDER_TEMPLATE_COLOR,
 			},
 			"keyword3": map[string]interface{}{

business/model/order.go

@@ -19,6 +19,7 @@ type GoodsOrder struct {
 	SubStoreName          string `orm:"size(64)"`
 	ShopPrice             int64  // 单位为分
 	SalePrice             int64  // 单位为分
+	ActualPayPrice        int64  // 单位为分
 	Weight                int    // 单位为克
 	ConsigneeName         string `orm:"size(32)"`
 	ConsigneeMobile       string `orm:"size(32)"`
@@ -60,8 +61,8 @@ type OrderSku struct {
 	SkuID          int       `orm:"column(sku_id)"`    // 外部系统里记录的 jxskuid
 	JxSkuID        int       `orm:"column(jx_sku_id)"` // 根据VendorSkuID在本地系统里查询出来的 jxskuid
 	SkuName        string    `orm:"size(255)"`
-	ShopPrice      int64
+	ShopPrice      int64     // 门店标价
-	SalePrice      int64
+	SalePrice      int64     // 售卖价
 	Weight         int       // 单位为克
 	SkuType        int       // 当前如果为gift就为1，否则缺省为0
 	PromotionType  int       // todo 当前是用于记录京东的PromotionType(生成jxorder用)，没有做转换
@@ -72,17 +73,17 @@ type OrderSku struct {
 // 所以这里不能用唯一索引
 func (o *OrderSku) TableIndex() [][]string {
 	return [][]string{
-		[]string{"VendorOrderID", "SkuID", "VendorID"},
+		[]string{"VendorOrderID", "SkuID"},
 	}
 }
 
 type Waybill struct {
 	ID                int64     `orm:"column(id)"`
-	VendorOrderID     string    `orm:"column(vendor_order_id);size(48)"`
-	OrderVendorID     int       `orm:"column(order_vendor_id)"`
 	VendorWaybillID   string    `orm:"column(vendor_waybill_id);size(48)"`
 	VendorWaybillID2  string    `orm:"column(vendor_waybill_id2);size(48)"` // 某些平台有多个ID，比如美团配送，当前美团配送的 delivery_id存这里
 	WaybillVendorID   int       `orm:"column(waybill_vendor_id)"`
+	VendorOrderID     string    `orm:"column(vendor_order_id);size(48)"`
+	OrderVendorID     int       `orm:"column(order_vendor_id)"`
 	CourierName       string    `orm:"size(32)"`
 	CourierMobile     string    `orm:"size(32)"`
 	Status            int       // 参见WaybillStatus*相关的常量定义
@@ -125,6 +126,6 @@ type OrderStatus struct {
 
 func (v *OrderStatus) TableIndex() [][]string {
 	return [][]string{
-		[]string{"VendorOrderID", "VendorStatus", "Status"},
+		[]string{"VendorOrderID", "Status", "VendorStatus"},
 	}
 }

business/scheduler/defsch/defsch.go

@@ -15,11 +15,11 @@ import (
 )
 
 const (
-	defTime2Delivered                      = 1 * time.Hour     // 正常订单都是1小时达
+	time2Delivered                         = 1 * time.Hour     // 正常订单都是1小时达
-	defTime2Schedule3rdCarrier             = 330 * time.Second // 京东要求5分钟后才能转自送，保险起见，设置为5分半钟
+	time2Schedule3rdCarrier                = 330 * time.Second // 京东要求5分钟后才能转自送，保险起见，设置为5分半钟
 	time2Schedule3rdCarrierGap4OrderStatus = 3 * time.Minute   // 京东要求是运单状态为待抢单且超时5分钟，但为了防止没有运单事件，所以就拣货完成事件开始算，添加3分钟
-	defTime2AutoPickupMin                  = 10 * time.Minute
+	time2AutoPickupMin                     = 15 * time.Minute
-	time2AutoPickupGap                     = 2 * time.Minute
+	time2AutoPickupGap                     = 5 * time.Minute
 )
 
 type WatchOrderInfo struct {
@@ -52,13 +52,13 @@ func init() {
 			},
 		},
 		model.OrderStatusAccepted: &scheduler.StatusActionConfig{ // 自动拣货
-			Timeout: defTime2AutoPickupMin,
+			Timeout: time2AutoPickupMin,
 			TimeoutAction: func(order *model.GoodsOrder) (err error) {
 				return sch.GetPurchasePlatformFromVendorID(order.VendorID).PickedUpGoods(order)
 			},
 		},
 		model.OrderStatusFinishedPickup: &scheduler.StatusActionConfig{ // 尝试召唤更多物流
-			Timeout: defTime2Schedule3rdCarrier,
+			Timeout: time2Schedule3rdCarrier,
 			TimeoutAction: func(order *model.GoodsOrder) (err error) {
 				return sch.createWaybillOn3rdProviders(order, nil)
 			},
@@ -74,11 +74,14 @@ func (s *DefScheduler) OnOrderNew(order *model.GoodsOrder) (err error) {
 	}
 	s.orderMap.Store(jxutils.ComposeUniversalOrderID(order.VendorOrderID, order.VendorID), watchInfo)
 	s.resetTimer(watchInfo, model.OrderStatusNew, order.OrderCreatedAt, 0)
+	if order.Status > model.OrderStatusNew {
+		return s.OnOrderStatusChanged(model.Order2Status(order))
+	}
 	return err
 }
 
 func (s *DefScheduler) OnOrderStatusChanged(status *model.OrderStatus) (err error) {
-	if status.Status > model.OrderStatusUnknown {
+	if status.Status > model.OrderStatusUnknown { // 只处理状态转换，一般消息不处理
 		globals.SugarLogger.Debugf("OnOrderStatusChanged, status:%v", status)
 		if savedOrderInfo := s.loadSavedOrderFromMap(status); savedOrderInfo != nil {
 			s.updateOrderByStatus(savedOrderInfo.order, status)
@@ -120,7 +123,7 @@ func (s *DefScheduler) OnWaybillStatusChanged(bill *model.Waybill) (err error) {
 				if bill.OrderVendorID == bill.WaybillVendorID {
 					if savedOrderInfo.timerStatus == model.OrderStatusFinishedPickup { // 如果当前TIMER还是OrderStatusFinishedPickup（在OnOrderStatusChanged中设置的），则重置
 						s.resetTimer(savedOrderInfo, model.OrderStatusFinishedPickup, bill.WaybillCreatedAt, 0)
-					} else {
+					} else if savedOrderInfo.timerStatus != 0 {
 						globals.SugarLogger.Infof("OnWaybillStatusChanged met other timer, status:%d", savedOrderInfo.timerStatus)
 					}
 				}
@@ -215,7 +218,8 @@ func (s *DefScheduler) createWaybillOn3rdProviders(order *model.GoodsOrder, excl
 func (s *DefScheduler) cancelOtherWaybills(savedOrderInfo *WatchOrderInfo, bill *model.Waybill) (err error) {
 	globals.SugarLogger.Debugf("cancelOtherWaybills, order:%v, bill:%v", savedOrderInfo.order, bill)
 	for _, v := range savedOrderInfo.waybills {
-		if bill == nil || v.WaybillVendorID != bill.OrderVendorID && !(v.WaybillVendorID == bill.WaybillVendorID && v.VendorWaybillID == bill.VendorWaybillID) {
+		if bill == nil || (v.WaybillVendorID != bill.OrderVendorID && !(v.WaybillVendorID == bill.WaybillVendorID && v.VendorWaybillID == bill.VendorWaybillID)) {
+			globals.SugarLogger.Debugf("cancelOtherWaybills, cancel bill:%v", bill)
 			_ = s.GetDeliveryPlatformFromVendorID(v.WaybillVendorID).CancelWaybill(v)
 		}
 	}
@@ -271,7 +275,7 @@ func (s *DefScheduler) loadSavedOrderFromMap(status *model.OrderStatus) *WatchOr
 
 func (s *DefScheduler) getBeginTime4LatestPickup(order *model.GoodsOrder) (retVal time.Time) {
 	if order.ExpectedDeliveredTime != utils.DefaultTimeValue {
-		return order.ExpectedDeliveredTime.Add(-defTime2Delivered)
+		return order.ExpectedDeliveredTime.Add(-time2Delivered)
 	}
 	return order.StatusTime
 }

globals/api/api.go

@@ -8,6 +8,7 @@ import (
 	"git.rosy.net.cn/baseapi/platformapi/mtpsapi"
 	"git.rosy.net.cn/baseapi/platformapi/weixinapi"
 	"git.rosy.net.cn/baseapi/utils"
+	_ "git.rosy.net.cn/jx-callback/globals/db"
 	"github.com/astaxie/beego"
 
 	"github.com/astaxie/beego/orm"

globals/db/db.go

@@ -0,0 +1,30 @@
+package db
+
+import (
+	"git.rosy.net.cn/jx-callback/business/legacyorder"
+	"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"
+	"github.com/astaxie/beego/orm"
+	_ "github.com/go-sql-driver/mysql" // import your used driver
+)
+
+func init() {
+	// set default database
+	orm.RegisterDataBase("default", "mysql", beego.AppConfig.String("dbConnectStr"), 30)
+	models.RegisterModels()
+	if globals.CallNew {
+		orm.RegisterModel(new(model.GoodsOrder))
+		orm.RegisterModel(new(model.OrderSku))
+		orm.RegisterModel(new(model.Waybill))
+		orm.RegisterModel(new(model.OrderStatus))
+
+		// orm.RegisterModel(new(legacyorder.Elemeorder2))
+		// orm.RegisterModel(new(legacyorder.Jdorder2))
+		orm.RegisterModel(new(legacyorder.Jxorder2))
+		orm.RegisterModel(new(legacyorder.Jxordersku2))
+	}
+	// create table
+	orm.RunSyncdb("default", false, true)
+}

globals/globals.go

@@ -2,13 +2,8 @@ package globals
 
 import (
 	"git.rosy.net.cn/baseapi"
-	"git.rosy.net.cn/jx-callback/business/legacyorder"
-	"git.rosy.net.cn/jx-callback/business/model"
-	"git.rosy.net.cn/jx-callback/legacy/models"
 	"github.com/astaxie/beego"
 	"github.com/astaxie/beego/logs"
-	"github.com/astaxie/beego/orm"
-	_ "github.com/go-sql-driver/mysql" // import your used driver
 	"go.uber.org/zap"
 )
 
@@ -35,24 +30,4 @@ func init() {
 	logger, _ := zap.NewDevelopment()
 	SugarLogger = logger.Sugar()
 	baseapi.Init(SugarLogger)
-	initDB()
-}
-
-func initDB() {
-	// set default database
-	orm.RegisterDataBase("default", "mysql", beego.AppConfig.String("dbConnectStr"), 30)
-	models.RegisterModels()
-	if CallNew {
-		orm.RegisterModel(new(model.GoodsOrder))
-		orm.RegisterModel(new(model.OrderSku))
-		orm.RegisterModel(new(model.Waybill))
-		orm.RegisterModel(new(model.OrderStatus))
-
-		// orm.RegisterModel(new(legacyorder.Elemeorder2))
-		// orm.RegisterModel(new(legacyorder.Jdorder2))
-		orm.RegisterModel(new(legacyorder.Jxorder2))
-		orm.RegisterModel(new(legacyorder.Jxordersku2))
-	}
-	// create table
-	orm.RunSyncdb("default", false, true)
 }

main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	bzcon "git.rosy.net.cn/jx-callback/business/controller"
 	"git.rosy.net.cn/jx-callback/legacy/jd/controller"
 	"git.rosy.net.cn/jx-callback/legacy/tasks"
 	_ "git.rosy.net.cn/jx-callback/routers"
@@ -13,6 +14,8 @@ func main() {
 		tasks.RefreshWeixinToken()
 		tasks.RefreshElmToken()
 	}
+	bzcon.OrderManager.LoadPendingOrders()
+	bzcon.WaybillManager.LoadPendingWaybills()
 
 	if beego.BConfig.RunMode == "dev" {
 		beego.BConfig.WebConfig.DirectoryIndex = true