buryingPoint/online_datasync/entity/action_order.go

package entity

import (
	"encoding/json"
	"fmt"
	"log"
	. "online_datasync/config"
	. "online_datasync/db"
	"strings"
	"time"

	. "app.yhyue.com/moapp/jybase/common"
	. "app.yhyue.com/moapp/jybase/date"
)

var (
	Action_order *action_order
)

type action_order struct {
}

func (a *action_order) TableName() string {
	return "action_order"
}

//
func (a *action_order) SaveFields() []string {
	return []string{"userid", "product_code", "createtime", "pay_time", "pay_way", "order_status", "order_money", "pay_money", "order_snapshot_id", "bill_status", "order_type", "timestamp"}
}

//
func (a *action_order) Run(start_unix, end_unix int64, start_layout, end_layout string) {
	log.Println("开始同步", a.TableName(), "表。。。")
	a.add()
	if start_unix > 0 {
		a.update(start_unix, end_unix, start_layout, end_layout)
	}
	log.Println("同步", a.TableName(), "表结束。。。")
}

//
func (a *action_order) add() {
	log.Println("开始同步新增", a.TableName(), "表。。。")
	index := 0
	action_order_array, action_order_spec_array, action_product_record_array := []interface{}{}, []interface{}{}, []interface{}{}
	Mysql_From_Jianyu.SelectByBath(Config.SelectBathSize, func(l *[]map[string]interface{}) {
		for _, v := range *l {
			index++
			TimeTask.Dataexport_order_id = Int64All(v["id"])
			product_code := a.getProductCode(ObjToString(v["product_type"]))
			//
			action_order_array = append(action_order_array, v["user_id"], product_code, v["create_time"], v["pay_time"], a.getPayWay(ObjToString(v["pay_way"])), v["order_status"], v["order_money"], v["pay_money"], v["id"], v["applybill_status"], a.getOrderType(IntAllDef(v["vip_type"], -1)), NowFormat(Date_Full_Layout))
			//
			var buyerclass_num interface{}
			var area_num interface{}
			var city_num interface{}
			if v["filter"] != nil {
				filter := make(map[string]interface{})
				if err := json.Unmarshal([]byte(ObjToString(v["filter"])), &filter); err == nil {
					filter_buyerclass, _ := filter["buyerclass"].([]interface{})
					buyerclass_num = len(filter_buyerclass)
					filter_area, _ := filter["area"].([]interface{})
					area_num = len(filter_area)
					filter_city, _ := filter["city"].([]interface{})
					city_num = len(filter_city)
				}
			}
			action_order_spec_array = append(action_order_spec_array, v["user_id"], v["id"], product_code, v["original_price"], v["discount_price"], v["data_count"], v["order_money"], v["order_money"], v["create_time"], v["vip_starttime"], v["vip_endtime"], v["filter"], buyerclass_num, area_num, city_num)
			//
			var status interface{}
			if v["vip_starttime"] != nil && v["vip_endtime"] != nil {
				vip_starttime, start_err := time.ParseInLocation(Date_Full_Layout, ObjToString(v["vip_starttime"]), time.Local)
				vip_endtime, end_err := time.ParseInLocation(Date_Full_Layout, ObjToString(v["vip_endtime"]), time.Local)
				if start_err == nil && end_err == nil {
					now := time.Now()
					if now.After(vip_starttime) && now.Before(vip_endtime) {
						status = 1
					} else {
						status = 2
					}
				}
			}
			action_product_record_array = append(action_product_record_array, v["user_id"], product_code, v["vip_starttime"], v["vip_endtime"], v["create_time"], v["id"], status, NowFormat(Date_Full_Layout))
			//
			if index%Config.InsertBathSize == 0 {
				log.Println("同步新增", a.TableName(), "表", index)
				Mysql_Main.InsertIgnoreBatch(a.TableName(), a.SaveFields(), action_order_array)
				action_order_array = []interface{}{}
				//
				log.Println("同步新增", Action_order_spec.TableName(), "表", index)
				Mysql_Main.InsertIgnoreBatch(Action_order_spec.TableName(), Action_order_spec.SaveFields(), action_order_spec_array)
				action_order_spec_array = []interface{}{}
				//
				log.Println("同步新增", Action_product_record.TableName(), "表", index)
				Mysql_Main.InsertIgnoreBatch(Action_product_record.TableName(), Action_product_record.SaveFields(), action_product_record_array)
				action_product_record_array = []interface{}{}
			}
		}
	}, `select id,user_id,create_time,pay_time,pay_way,order_status,order_money,pay_money,applybill_status,vip_type,product_type,vip_starttime,vip_endtime,original_price,discount_price,data_count,filter from dataexport_order where id>? order by id`, TimeTask.Dataexport_order_id)
	if len(action_order_array) > 0 {
		Mysql_Main.InsertIgnoreBatch(a.TableName(), a.SaveFields(), action_order_array)
		action_order_array = []interface{}{}
		//
		Mysql_Main.InsertIgnoreBatch(Action_order_spec.TableName(), Action_order_spec.SaveFields(), action_order_spec_array)
		action_order_spec_array = []interface{}{}
		//
		Mysql_Main.InsertIgnoreBatch(Action_product_record.TableName(), Action_product_record.SaveFields(), action_product_record_array)
		action_product_record_array = []interface{}{}
	}
	log.Println("同步新增", a.TableName(), "表结束。。。", index)
	return
}

//
func (a *action_order) getOrderType(vip_type int) int {
	if vip_type == 0 {
		return 1 //试用
	} else if vip_type == 1 {
		return 3 //续费
	} else if vip_type == 2 {
		return 4 //升级
	}
	return 2
}

//
func (a *action_order) getProductCode(product_type string) string {
	if strings.ToUpper(product_type) == "VIP订阅" {
		return "101"
	} else if product_type == "历史数据" {
		return "102"
	} else if product_type == "招标文件解读" {
		return "103"
	} else if strings.HasPrefix(product_type, "大会员") {
		return "104"
	} else if product_type == "剑鱼币" {
		return "105"
	} else if product_type == "中标必听课" {
		return "106"
	} else if product_type == "企业商机管理" {
		return "107"
	} else if product_type == "剑鱼课程" {
		return "108"
	} else if product_type == "招投标课程" {
		return "109"
	} else if product_type == "数据报告" {
		return "110"
	} else if product_type == "线上课程" {
		return "111"
	}
	return ""
}

//
func (a *action_order) getPayWay(pay_way string) string {
	if pay_way == "ali_app" || pay_way == "wx_app" {
		return "app"
	} else if pay_way == "wx_js" {
		return "wechat"
	} else if pay_way == "ali_pc" || pay_way == "wx_pc" {
		return "pc"
	}
	return ""
}

//
func (a *action_order) update(start_unix, end_unix int64, start_layout, end_layout string) {
	log.Println("开始同步更新", a.TableName(), "表 。。。")
	index := 0
	array := [][]interface{}{}
	fields := []string{"order_snapshot_id", "pay_time", "pay_way", "order_status", "pay_money", "timestamp"}
	Mysql_From_Jianyu.SelectByBath(Config.SelectBathSize, func(l *[]map[string]interface{}) {
		for _, v := range *l {
			index++
			pay_way := a.getPayWay(ObjToString(v["pay_way"]))
			array = append(array, []interface{}{v["id"], v["pay_time"], pay_way, v["order_status"], v["pay_money"], NowFormat(Date_Full_Layout)})
			if index%Config.UpdateBathSize == 0 {
				log.Println("同步更新", a.TableName(), "表", index)
				Mysql_Main.UpdateBath(a.TableName(), fields, array)
				array = [][]interface{}{}
			}
		}
	}, `select distinct a.id,a.pay_time,a.pay_way,a.order_status,a.pay_money,a.applybill_status from dataexport_order a
		left join apply_invoice b on (a.id=b.order_id)
		left join invoice c on (a.order_code=c.order_code)
		where (a.pay_time>=? and a.pay_time<?) or (b.apply_date>=? and b.apply_date<?) or (c.create_time>=? and c.create_time<?) or (c.billing_time>=? and c.billing_time<?) or (c.operable_time>=? and c.operable_time<?)`, start_layout, end_layout, fmt.Sprint(start_unix), fmt.Sprint(end_unix), start_unix, end_unix, start_unix, end_unix, start_layout, end_layout)
	if len(array) > 0 {
		Mysql_Main.UpdateBath(a.TableName(), fields, array)
		array = [][]interface{}{}
	}
	log.Println("同步更新", a.TableName(), "表结束。。。", index)
}