feat: sync xcx order shipping to wechat

go/gopath/src/fohow.com/apps/controllers/cron_controller/init.go

@@ -35,12 +35,7 @@ func (self *CronController) Cron() {
 	// 该接口要指定的ip列表访问
 	if inAllowedList, _ := tool.Contain(ip, strings.Split(beego.AppConfig.String("inetIp"), ",")); inAllowedList {
 		//beego.BeeLogger.Error("inAllowedList: %v", inAllowedList)
-		if name == "orderCompleteDispatchTime" {
-			isUpdateTime, _ := self.GetBool("is_update_time", false)
-			OrderCompleteDispatchTime(isUpdateTime)
-		} else {
-			exec(name)
-		}
+		self.run(name)
 		beego.BeeLogger.Info("Request ip[%s] in AllowedList ip, task execute", ip)
 	} else {
 		beego.BeeLogger.Warn("Request ip[%s] is not intranet ip, check current user", ip)
@@ -53,7 +48,7 @@ func (self *CronController) Cron() {
 				beego.BeeLogger.Warn("id=%d,currentUserId=%d", id, currentUser.Id)
 				if currentUser.Id == id {
 					beego.BeeLogger.Info("current user id=[%d], exec task", id)
-					exec(name)
+					self.run(name)
 					break
 				}
 			}
@@ -62,6 +57,19 @@ func (self *CronController) Cron() {
 	self.ServeJSON()
 }
 
+func (self *CronController) run(name string) {
+	if name == "orderCompleteDispatchTime" {
+		isUpdateTime, _ := self.GetBool("is_update_time", false)
+		OrderCompleteDispatchTime(isUpdateTime)
+		return
+	}
+	if name == "syncWechatOrderShipping" {
+		self.Data["json"] = SyncWechatOrderShipping()
+		return
+	}
+	exec(name)
+}
+
 func exec(name string) {
 	switch name {
 	case "take_cash": //提现
@@ -88,6 +96,8 @@ func exec(name string) {
 		//go patchCent()
 	case "clear_update_orders": //清理过期物流数据
 		clearUpdateOrders()
+	case "syncWechatOrderShipping": //同步小程序发货信息到微信发货管理
+		SyncWechatOrderShipping()
 	default:
 		beego.BeeLogger.Error("Didn't get task wity name:[%s]", name)
 	}

go/gopath/src/fohow.com/apps/controllers/cron_controller/wechat_order_shipping.go

@@ -0,0 +1,288 @@
+package cron_controller
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+	"unicode/utf8"
+
+	"github.com/astaxie/beego"
+
+	"fohow.com/apps/models/balance_model"
+	"fohow.com/apps/models/order_model"
+	"fohow.com/apps/models/user_model"
+	"fohow.com/libs/wx_mp"
+)
+
+var wechatOrderShippingLock sync.Mutex
+
+type WechatOrderShippingResult struct {
+	StartedAt           string `json:"started_at"`
+	FinishedAt          string `json:"finished_at"`
+	GoodsScanned        int    `json:"goods_scanned"`
+	GoodsUploaded       int    `json:"goods_uploaded"`
+	GoodsSkipped        int    `json:"goods_skipped"`
+	RechargeScanned     int    `json:"recharge_scanned"`
+	RechargeUploaded    int    `json:"recharge_uploaded"`
+	RechargeSkipped     int    `json:"recharge_skipped"`
+	WechatAlreadySynced int    `json:"wechat_already_synced"`
+	Failed              int    `json:"failed"`
+}
+
+func StartWechatOrderShippingScheduler() {
+	if !beego.AppConfig.DefaultBool("WechatOrderShippingCronEnabled", true) {
+		beego.BeeLogger.Warn("wechat order shipping scheduler disabled")
+		return
+	}
+	go func() {
+		for {
+			next := nextWechatOrderShippingRun(time.Now())
+			sleep := next.Sub(time.Now())
+			if sleep < 0 {
+				sleep = time.Minute
+			}
+			beego.BeeLogger.Warn("wechat order shipping scheduler next run at %s", next.Format("2006-01-02 15:04:05"))
+			time.Sleep(sleep)
+			SyncWechatOrderShipping()
+			time.Sleep(time.Minute)
+		}
+	}()
+}
+
+func nextWechatOrderShippingRun(now time.Time) time.Time {
+	next := time.Date(now.Year(), now.Month(), now.Day(), 1, 3, 0, 0, now.Location())
+	if !next.After(now) {
+		next = next.AddDate(0, 0, 1)
+	}
+	return next
+}
+
+func SyncWechatOrderShipping() *WechatOrderShippingResult {
+	if !wechatOrderShippingLock.TryLock() {
+		beego.BeeLogger.Warn("wechat order shipping task already running")
+		return &WechatOrderShippingResult{StartedAt: time.Now().Format("2006-01-02 15:04:05")}
+	}
+	defer wechatOrderShippingLock.Unlock()
+
+	now := time.Now()
+	ret := &WechatOrderShippingResult{StartedAt: now.Format("2006-01-02 15:04:05")}
+	startAt := now.AddDate(0, 0, -7)
+	accessToken := wx_mp.GetAccessToken(beego.AppConfig.String("WxFohowXcxAppId"), beego.AppConfig.String("WxFohowXcxAppSecret"))
+	if accessToken == "" {
+		beego.BeeLogger.Error("wechat order shipping access token is empty")
+		ret.Failed++
+		ret.FinishedAt = time.Now().Format("2006-01-02 15:04:05")
+		return ret
+	}
+
+	syncDispatchedGoodsOrders(accessToken, startAt, now, ret)
+	syncRechargeCashOrders(accessToken, startAt, now, ret)
+
+	ret.FinishedAt = time.Now().Format("2006-01-02 15:04:05")
+	beego.BeeLogger.Warn("wechat order shipping task finished: %+v", ret)
+	return ret
+}
+
+func syncDispatchedGoodsOrders(accessToken string, startAt, endAt time.Time, ret *WechatOrderShippingResult) {
+	orders := order_model.GetRecentXcxWechatDispatchedOrders(startAt, endAt)
+	ret.GoodsScanned = len(orders)
+	detailsMap := order_model.GetAllDetailsByOrderIds(orderIdsOf(orders), true)
+	for _, order := range orders {
+		wxUser := user_model.GetWxUserById(order.WxUserId, true)
+		if wxUser == nil || wxUser.Openid == "" {
+			ret.GoodsSkipped++
+			beego.BeeLogger.Warn("wechat order shipping goods order=%s skipped: empty openid", order.OrderId)
+			continue
+		}
+		itemDesc := buildGoodsItemDesc(detailsMap[order.OrderId])
+		if itemDesc == "" {
+			itemDesc = "商品订单"
+		}
+		logisticsType, shippingItem, ok := buildGoodsShippingItem(order, itemDesc)
+		if !ok {
+			ret.GoodsSkipped++
+			continue
+		}
+		if !uploadWechatOrderShippingIfNeeded(accessToken, order.OrderId, order.TradeNo, wxUser.Openid, logisticsType, shippingItem, ret) {
+			ret.GoodsSkipped++
+			continue
+		}
+		ret.GoodsUploaded++
+	}
+}
+
+func syncRechargeCashOrders(accessToken string, startAt, endAt time.Time, ret *WechatOrderShippingResult) {
+	orders := balance_model.GetRecentPaidRechargeCashOrders(startAt, endAt)
+	ret.RechargeScanned = len(orders)
+	for _, order := range orders {
+		wxUser := user_model.GetWxUserById(order.WxUserId, true)
+		if wxUser == nil || wxUser.Openid == "" {
+			ret.RechargeSkipped++
+			beego.BeeLogger.Warn("wechat order shipping recharge order=%s skipped: empty openid", order.OrderId)
+			continue
+		}
+		shippingItem := wx_mp.OrderShippingItem{ItemDesc: "余额充值"}
+		if !uploadWechatOrderShippingIfNeeded(accessToken, order.OrderId, order.TradeNo, wxUser.Openid, wx_mp.OrderShippingLogisticsVirtual, shippingItem, ret) {
+			ret.RechargeSkipped++
+			continue
+		}
+		ret.RechargeUploaded++
+	}
+}
+
+func uploadWechatOrderShippingIfNeeded(accessToken, orderId, transactionId, openid string, logisticsType int64, shippingItem wx_mp.OrderShippingItem, ret *WechatOrderShippingResult) bool {
+	if transactionId == "" {
+		beego.BeeLogger.Warn("wechat order shipping order=%s skipped: empty transaction_id", orderId)
+		return false
+	}
+	queryRet, err := wx_mp.GetOrderShippingInfo(accessToken, transactionId)
+	if err != nil {
+		ret.Failed++
+		beego.BeeLogger.Error("wechat order shipping query order=%s transaction=%s err=%s", orderId, transactionId, err)
+		return false
+	}
+	if queryRet.ErrCode != 0 {
+		ret.Failed++
+		beego.BeeLogger.Error("wechat order shipping query order=%s transaction=%s errcode=%d errmsg=%s", orderId, transactionId, queryRet.ErrCode, queryRet.ErrMsg)
+		return false
+	}
+	if queryRet.Order.OrderState != wx_mp.OrderShippingStateWaitShipping {
+		ret.WechatAlreadySynced++
+		beego.BeeLogger.Warn("wechat order shipping order=%s skipped: wechat order_state=%d", orderId, queryRet.Order.OrderState)
+		return false
+	}
+
+	req := wx_mp.NewUploadOrderShippingInfoRequest(transactionId, openid, logisticsType, shippingItem)
+	uploadRet, err := wx_mp.UploadOrderShippingInfo(accessToken, req)
+	if err != nil {
+		ret.Failed++
+		beego.BeeLogger.Error("wechat order shipping upload order=%s transaction=%s err=%s", orderId, transactionId, err)
+		return false
+	}
+	if uploadRet.ErrCode != 0 {
+		ret.Failed++
+		beego.BeeLogger.Error("wechat order shipping upload order=%s transaction=%s errcode=%d errmsg=%s", orderId, transactionId, uploadRet.ErrCode, uploadRet.ErrMsg)
+		return false
+	}
+	beego.BeeLogger.Warn("wechat order shipping upload order=%s transaction=%s success", orderId, transactionId)
+	return true
+}
+
+func buildGoodsShippingItem(order *order_model.Order, itemDesc string) (int64, wx_mp.OrderShippingItem, bool) {
+	if order.PickWay == order_model.PICK_SHOP {
+		return wx_mp.OrderShippingLogisticsPickup, wx_mp.OrderShippingItem{ItemDesc: itemDesc}, true
+	}
+	expressCompany := normalizeWechatExpressCompany(order.ExpressCode, order.ExpressCompany)
+	if strings.TrimSpace(order.ExpressOrderNo) == "" || expressCompany == "" {
+		beego.BeeLogger.Warn("wechat order shipping goods order=%s skipped: express_no=%s express_code=%s express_company=%s", order.OrderId, order.ExpressOrderNo, order.ExpressCode, order.ExpressCompany)
+		return 0, wx_mp.OrderShippingItem{}, false
+	}
+	item := wx_mp.OrderShippingItem{
+		TrackingNo:     strings.TrimSpace(order.ExpressOrderNo),
+		ExpressCompany: expressCompany,
+		ItemDesc:       itemDesc,
+	}
+	if contact := maskPhone(order.Tel); contact != "" {
+		item.Contact = &wx_mp.OrderShippingContact{ReceiverContact: contact}
+	}
+	return wx_mp.OrderShippingLogisticsExpress, item, true
+}
+
+func buildGoodsItemDesc(details []*order_model.OrderDetail) string {
+	parts := make([]string, 0, len(details))
+	for _, detail := range details {
+		name := strings.TrimSpace(detail.ProductName)
+		if name == "" {
+			continue
+		}
+		if detail.Count > 0 {
+			name = fmt.Sprintf("%s*%d", name, detail.Count)
+		}
+		parts = append(parts, name)
+	}
+	return truncateRunes(strings.Join(parts, ";"), 120)
+}
+
+func orderIdsOf(orders []*order_model.Order) []string {
+	ids := make([]string, 0, len(orders))
+	for _, order := range orders {
+		if order.OrderId != "" {
+			ids = append(ids, order.OrderId)
+		}
+	}
+	return ids
+}
+
+func normalizeWechatExpressCompany(expressCode, expressCompany string) string {
+	code := strings.ToUpper(strings.TrimSpace(expressCode))
+	code = strings.Replace(code, "-", "_", -1)
+	codeMap := map[string]string{
+		"SF":     "SF",
+		"STO":    "STO",
+		"ZTO":    "ZTO",
+		"YTO":    "YTO",
+		"YT":     "YTO",
+		"YD":     "YD",
+		"YUNDA":  "YD",
+		"EMS":    "EMS",
+		"EMSG":   "EMS",
+		"JD":     "JD",
+		"DB":     "DBL",
+		"DBL":    "DBL",
+		"ZJS":    "ZJS",
+		"YZGN":   "YZPY",
+		"YZPY":   "YZPY",
+		"HT":     "HTKY",
+		"HTKY":   "HTKY",
+		"FAST":   "FAST",
+		"QF":     "QFKD",
+		"QFKD":   "QFKD",
+		"FEDEX":  "FEDEX",
+		"UPS":    "UPS",
+		"ZTKY":   "ZTKY",
+		"ZTOKY":  "ZTKY",
+		"DEPPON": "DBL",
+	}
+	if v, ok := codeMap[code]; ok {
+		return v
+	}
+	company := strings.ToLower(strings.TrimSpace(expressCompany))
+	nameMap := map[string]string{
+		"顺丰":  "SF",
+		"申通":  "STO",
+		"中通":  "ZTO",
+		"圆通":  "YTO",
+		"韵达":  "YD",
+		"邮政":  "EMS",
+		"ems": "EMS",
+		"京东":  "JD",
+		"德邦":  "DBL",
+	}
+	for name, id := range nameMap {
+		if strings.Contains(company, strings.ToLower(name)) {
+			return id
+		}
+	}
+	return code
+}
+
+func maskPhone(phone string) string {
+	phone = strings.TrimSpace(phone)
+	if len(phone) < 7 {
+		return ""
+	}
+	runes := []rune(phone)
+	if len(runes) < 7 {
+		return ""
+	}
+	return fmt.Sprintf("%s****%s", string(runes[:3]), string(runes[len(runes)-4:]))
+}
+
+func truncateRunes(s string, max int) string {
+	if max <= 0 || utf8.RuneCountInString(s) <= max {
+		return s
+	}
+	runes := []rune(s)
+	return string(runes[:max])
+}

go/gopath/src/fohow.com/apps/models/balance_model/cash_balance.go

@@ -384,6 +384,30 @@ func GetRechargeCashOrderByOId(oId string, useCache bool) *RechargeCashOrder {
 	return item
 }
 
+// 查找最近一段时间内已支付的小程序余额充值订单
+func GetRecentPaidRechargeCashOrders(startAt, endAt time.Time) (orders []*RechargeCashOrder) {
+	sql := `
+		SELECT
+			*
+		FROM
+			recharge_cash_orders
+		WHERE
+			state = 1
+		AND pay_way = ?
+		AND trade_no IS NOT NULL
+		AND trade_no != ''
+		AND paied_at >= ?
+		AND paied_at < ?
+		ORDER BY paied_at ASC;
+	`
+
+	if _, err := orm.NewOrm().Raw(sql, PAY_WAY_TYPE_RECHARGE_WXPAY, startAt.Unix(), endAt.Unix()).QueryRows(&orders); err != nil {
+		beego.BeeLogger.Debug("balance_model.GetRecentPaidRechargeCashOrders() err=[%s]", err)
+		return nil
+	}
+	return orders
+}
+
 func GetCashBalanceBySourceAndRId(source, rId string) *CashBalance {
 	item := &CashBalance{}
 	if err := orm.NewOrm().QueryTable(item).

go/gopath/src/fohow.com/apps/models/order_model/order.go

@@ -601,6 +601,32 @@ func GetSevenDaysFullOrdersByDispatchTime() (orders []*Order) {
 	return orders
 }
 
+// 查找最近一段时间内小程序微信支付的已发货订单
+func GetRecentXcxWechatDispatchedOrders(startAt, endAt time.Time) (orders []*Order) {
+	sql := `
+		SELECT
+			*
+		FROM
+			orders
+		WHERE
+			status = ?
+		AND source = ?
+		AND pay_way = ?
+		AND trade_no IS NOT NULL
+		AND trade_no != ''
+		AND dispatch_time IS NOT NULL
+		AND dispatch_time >= ?
+		AND dispatch_time < ?
+		ORDER BY dispatch_time ASC;
+	`
+
+	if _, err := orm.NewOrm().Raw(sql, STATUS_DISPATCH, SOURCE_XCX, PAY_WAY_WEIXIN, startAt, endAt).QueryRows(&orders); err != nil {
+		beego.BeeLogger.Debug("order_model.GetRecentXcxWechatDispatchedOrders() err=[%s]", err)
+		return nil
+	}
+	return orders
+}
+
 // 查找未支付订单
 func GetUnpayOrders() (orders []*Order) {
 

go/gopath/src/fohow.com/libs/wx_mp/order_shipping.go

@@ -0,0 +1,150 @@
+package wx_mp
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/astaxie/beego"
+)
+
+const (
+	OrderShippingLogisticsExpress = 1
+	OrderShippingLogisticsVirtual = 3
+	OrderShippingLogisticsPickup  = 4
+
+	OrderShippingDeliveryUnified = 1
+
+	OrderShippingStateWaitShipping = 1
+)
+
+type OrderShippingOrderKey struct {
+	OrderNumberType int64  `json:"order_number_type"`
+	TransactionId   string `json:"transaction_id,omitempty"`
+	MchId           string `json:"mchid,omitempty"`
+	OutTradeNo      string `json:"out_trade_no,omitempty"`
+}
+
+type OrderShippingContact struct {
+	ConsignorContact string `json:"consignor_contact,omitempty"`
+	ReceiverContact  string `json:"receiver_contact,omitempty"`
+}
+
+type OrderShippingItem struct {
+	TrackingNo     string                `json:"tracking_no,omitempty"`
+	ExpressCompany string                `json:"express_company,omitempty"`
+	ItemDesc       string                `json:"item_desc"`
+	Contact        *OrderShippingContact `json:"contact,omitempty"`
+}
+
+type UploadOrderShippingInfoRequest struct {
+	OrderKey       OrderShippingOrderKey `json:"order_key"`
+	LogisticsType  int64                 `json:"logistics_type"`
+	DeliveryMode   int64                 `json:"delivery_mode"`
+	IsAllDelivered bool                  `json:"is_all_delivered,omitempty"`
+	ShippingList   []OrderShippingItem   `json:"shipping_list"`
+	UploadTime     string                `json:"upload_time"`
+	Payer          struct {
+		Openid string `json:"openid"`
+	} `json:"payer"`
+}
+
+type UploadOrderShippingInfoResponse struct {
+	ErrCode int64  `json:"errcode"`
+	ErrMsg  string `json:"errmsg"`
+}
+
+type GetOrderShippingInfoResponse struct {
+	ErrCode int64  `json:"errcode"`
+	ErrMsg  string `json:"errmsg"`
+	Order   struct {
+		OrderState int64 `json:"order_state"`
+		Shipping   struct {
+			FinishShipping bool `json:"finish_shipping"`
+		} `json:"shipping"`
+	} `json:"order"`
+}
+
+func GetOrderShippingInfo(accessToken, transactionId string) (*GetOrderShippingInfoResponse, error) {
+	req := map[string]string{
+		"transaction_id": transactionId,
+	}
+	body, err := callOrderShippingAPI(
+		"https://api.weixin.qq.com/wxa/sec/order/get_order",
+		accessToken,
+		req,
+	)
+	if err != nil {
+		return nil, err
+	}
+	ret := &GetOrderShippingInfoResponse{}
+	if err := json.Unmarshal(body, ret); err != nil {
+		return nil, err
+	}
+	return ret, nil
+}
+
+func UploadOrderShippingInfo(accessToken string, req *UploadOrderShippingInfoRequest) (*UploadOrderShippingInfoResponse, error) {
+	body, err := callOrderShippingAPI(
+		"https://api.weixin.qq.com/wxa/sec/order/upload_shipping_info",
+		accessToken,
+		req,
+	)
+	if err != nil {
+		return nil, err
+	}
+	ret := &UploadOrderShippingInfoResponse{}
+	if err := json.Unmarshal(body, ret); err != nil {
+		return nil, err
+	}
+	return ret, nil
+}
+
+func NewUploadOrderShippingInfoRequest(transactionId, openid string, logisticsType int64, shipping OrderShippingItem) *UploadOrderShippingInfoRequest {
+	req := &UploadOrderShippingInfoRequest{
+		OrderKey: OrderShippingOrderKey{
+			OrderNumberType: 2,
+			TransactionId:   transactionId,
+		},
+		LogisticsType: logisticsType,
+		DeliveryMode:  OrderShippingDeliveryUnified,
+		ShippingList:  []OrderShippingItem{shipping},
+		UploadTime:    time.Now().Format(time.RFC3339),
+	}
+	req.Payer.Openid = openid
+	return req
+}
+
+func callOrderShippingAPI(apiURL, accessToken string, req interface{}) ([]byte, error) {
+	postBody, err := json.Marshal(req)
+	if err != nil {
+		return nil, err
+	}
+	u := fmt.Sprintf("%s?access_token=%s", apiURL, url.QueryEscape(accessToken))
+	httpReq, err := http.NewRequest("POST", u, bytes.NewReader(postBody))
+	if err != nil {
+		return nil, err
+	}
+	httpReq.Header.Set("Content-Type", "application/json")
+
+	client := &http.Client{Timeout: 20 * time.Second}
+	resp, err := client.Do(httpReq)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return nil, fmt.Errorf("wechat order shipping api http status=%d body=%s", resp.StatusCode, string(body))
+	}
+	beego.BeeLogger.Warn("wechat order shipping api url=%s req=%s resp=%s", apiURL, string(postBody), string(body))
+	return body, nil
+}

go/gopath/src/fohow.com/main.go

@@ -10,6 +10,7 @@ import (
 	"github.com/astaxie/beego/orm"
 	"github.com/astaxie/beego/plugins/cors"
 
+	"fohow.com/apps/controllers/cron_controller"
 	_ "fohow.com/routers"
 )
 
@@ -68,5 +69,7 @@ func main() {
 		AllowCredentials: true,
 	}))
 
+	cron_controller.StartWechatOrderShippingScheduler()
+
 	beego.Run()
 }