gd_crontab/impl/crontab/log_hour_task.go

package crontab

import (
	"context"
	"gd_crontab/apis"
	"fmt"
	"runtime"
	"strings"
	"time"

	"gd_crontab/common.in/utils"
	"github.com/astaxie/beego/orm"
	"go.uber.org/zap"
)

type TGdApiCodeHour struct {
	Id           int64  `json:"id"`
	MerchantId   int64  `json:"merchant_id"`
	ApiId        int64  `json:"api_id"`
	Code         int    `json:"code"`
	Msg          string `json:"msg"`
	Count        int64  `json:"count"`
	Hour         int64  `json:"hour"`
	State        bool   `json:"state"`
	MerchantName string `json:"merchant_name"`
	ApiName      string `json:"api_name"`
}

type TGdProviderCodeHour struct {
	Id              int64  `json:"id"`
	ProviderApiId   int64  `json:"api_id"`
	Code            string `json:"code"`
	Msg             string `json:"msg"`
	Count           int64  `json:"count"`
	Hour            int64  `json:"hour"`
	State           bool   `json:"state"`
	ProviderName    string `json:"provider_name"`
	ProviderApiName string `json:"provider_api_name"`
}

type TGdReportHour struct {
	Id                int64
	MerchantId        int64
	ApiId             int64
	MerchantName      string
	Date              string `json:"date"`
	MerchantDataApiId int64
	ApiName           string  `json:"api_name"`
	Total             int64   `json:"total"`
	Valid             int64   `json:"valid"`
	Success           int64   `json:"success"`
	Query             int64   `json:"query"`
	Reuse             int64   `json:"reuse"`
	SumElapsed        float64 `json:"sum_elapsed"`
	PlatformError     int64   `json:"platform_error"`
	ProviderError     int64   `json:"provider_error"`
	Charge            int64   `json:"charge"`
	Hour              int64   `json:"hour"`
	NonredundantQuery int     `json:"nonredundant_query"`
	Nonredundant      int     `json:"nonredundant"`
	ChargeReuse       int64   `json:"charge_reuse"`
}

type TGdProviderReportHour struct {
	Id              int64
	Date            string  `json:"date"`
	ProviderName    string  `json:"provider_name"`
	MerchantId      int64   `json:"merchant_id"`
	ApiId           int64   `json:"api_id"`
	ProviderApiName string  `json:"provider_api_name"`
	Total           int64   `json:"total"`
	Success         int64   `json:"success"`
	Failed          int64   `json:"failed"`
	Query           int64   `json:"query"`
	SumElapsed      float64 `json:"sum_elapsed"`
	Charge          int64   `json:"charge"`
	ProviderApiId   int64   `json:"provider_api_id"`
	Hour            int64   `json:"hour"`
}

func computeApiReportNormal(start, end int64) (ret []TGdReportHour) {
	mreq := apis.LogQueryUserAccessCountExportReq{}
	mreply := apis.LogQueryUserAccessCountExportReply{}
	mreq.StartTimestamp = start
	mreq.EndTimestamp = end
	mreq.GroupByMerchant = true
	mreq.TabName = "t_gd_access_log_day"
	err := LogQueryUserAccessCountExport(context.Background(), &mreq, &mreply)
	if err != nil {
		l.Error("func",
			zap.String("call", "LogQueryUserAccessCountExport"),
			zap.String("args", utils.MarshalJsonString(mreq)),
			zap.String("error", err.Error()))
		return nil
	}
	ret = make([]TGdReportHour, len(mreply.LogQueryUserAcessCount))
	for i, v := range mreply.LogQueryUserAcessCount {
		item := &ret[i]
		item.Hour = start
		item.Date = time.Unix(start, 0).Format("2006-01-02")
		item.ApiId = v.ApiId
		item.Charge = v.Charge
		item.ApiName = v.ApiName
		item.MerchantName = v.MerchantName
		item.Valid = v.Valid
		item.Reuse = v.Reuse
		item.Total = v.Total
		item.Success = v.Success
		item.SumElapsed = v.SumElapsed
		item.ProviderError = v.ProviderError
		item.PlatformError = v.PlatformError
		item.Query = v.Query
		item.MerchantId = v.MerchantId
		item.NonredundantQuery = v.NonredundantQuery
		item.Nonredundant = v.Nonredundant
		item.ChargeReuse = v.ChargeReuse
	}
	return ret
}

func computeApiReportCodeDistribution(apiReport []TGdReportHour, start, end int64) (ret []TGdApiCodeHour) {

	for _, v := range apiReport {
		mreq := apis.LogQueryInterfaceCountReq{}
		mreq.MerchantId = v.MerchantId
		mreq.ApiId = v.ApiId
		mreq.StartTimestamp = start
		mreq.EndTimestamp = end
		mreq.TabName = "t_gd_access_log_day"
		mreq.GroupByState = true
		mreply := apis.LogQueryInterfaceAnalyzeErrorReply{}
		err := LogQueryInterfaceAnalyzeError(context.Background(), &mreq, &mreply)
		if err != nil {
			l.Error("func",
				zap.String("call", "LogQueryInterfaceAnalyzeError"),
				zap.String("args", utils.MarshalJsonString(mreq)),
				zap.String("error", err.Error()))
			return nil
		}
		codeInfos := make([]TGdApiCodeHour, len(mreply.List))
		for i, c := range mreply.List {
			codeInfos[i].ApiId = v.ApiId
			codeInfos[i].MerchantId = v.MerchantId
			codeInfos[i].MerchantName = v.MerchantName
			codeInfos[i].ApiName = v.ApiName
			codeInfos[i].Code = c.Code
			codeInfos[i].Msg = c.Msg
			codeInfos[i].Count = int64(c.Count)
			codeInfos[i].Hour = v.Hour
			codeInfos[i].State = c.State
		}
		ret = append(ret, codeInfos...)
	}
	return ret
}

func computeProviderReportNormal(start, end int64) (ret []TGdProviderReportHour) {
	mreq := apis.LogQueryProviderCountExportReq{}
	mreply := apis.LogQueryProviderCountExportReply{}
	mreq.StartTimestamp = start
	mreq.EndTimestamp = end
	mreq.GroupByMerchantApi = true
	mreq.TabName = "t_gd_thirdpart_log_day"
	err := LogQueryProviderCountExport(context.Background(), &mreq, &mreply)
	if err != nil {
		l.Error("func",
			zap.String("call", "LogQueryProviderCountExport"),
			zap.String("args", utils.MarshalJsonString(mreq)),
			zap.String("error", err.Error()))
		return nil
	}
	ret = make([]TGdProviderReportHour, len(mreply.LogQueryProviderCount))
	for i, v := range mreply.LogQueryProviderCount {
		item := &ret[i]
		item.Hour = start
		item.Date = time.Unix(start, 0).Format("2006-01-02")
		item.ProviderApiId = v.ProviderApiId
		item.Charge = v.Charge
		item.ProviderApiName = v.ProviderApiName
		item.ProviderName = v.ProviderName
		item.Total = v.Total
		item.Query = v.Query
		item.Success = v.Success
		item.SumElapsed = v.SumElapsed
		item.Failed = v.Failed
		item.MerchantId = v.MerchantId
		item.ApiId = v.ApiId
	}
	return ret
}

func computeProviderReportCodeDistribution(providerReport []TGdProviderReportHour, start, end int64) (ret []TGdProviderCodeHour) {
	m := map[int64]string{}
	for _, v := range providerReport {
		if _, ok := m[v.ProviderApiId]; ok == false {
			m[v.ProviderApiId] = "-"
		} else {
			continue
		}
		mreq := apis.ThirdPartyInterfaceErrorAnalyzeReq{}
		mreq.TabName = "t_gd_thirdpart_log_day"
		mreq.ApiId = v.ProviderApiId
		mreq.StartTimestamp = start
		mreq.EndTimestamp = end
		mreq.GroupByState = true
		mreply := apis.ThirdPartyInterfaceErrorAnalyzeReply{}
		err := ThirdPartyInterfaceErrorAnalyze(context.Background(), &mreq, &mreply)
		if err != nil {
			l.Error("func",
				zap.String("call", "ThirdPartyInterfaceErrorAnalyze"),
				zap.String("args", utils.MarshalJsonString(mreq)),
				zap.String("error", err.Error()))
			return nil
		}
		codeInfos := make([]TGdProviderCodeHour, len(mreply.List))
		for i, c := range mreply.List {
			codeInfos[i].ProviderApiId = v.ProviderApiId
			codeInfos[i].ProviderApiName = v.ProviderApiName
			codeInfos[i].ProviderName = v.ProviderName
			codeInfos[i].Code = c.Code
			codeInfos[i].Msg = c.Msg
			codeInfos[i].Count = int64(c.Count)
			codeInfos[i].Hour = v.Hour
			codeInfos[i].State = c.State
		}
		ret = append(ret, codeInfos...)
	}
	return ret
}

func insertApiReport(apiReport []TGdReportHour, codeInfos []TGdApiCodeHour) {
	if len(apiReport) > 0 {
		hourDb.Raw("delete from t_gd_report_hour where hour=?", apiReport[0].Hour).Exec()
		hourDb.InsertMulti(len(apiReport), &apiReport)
	}
	if len(codeInfos) > 0 {
		hourDb.Raw("delete from t_gd_api_code_hour where hour=?", codeInfos[0].Hour).Exec()
		hourDb.InsertMulti(len(codeInfos), &codeInfos)
	}
}

func insertProviderReport(providerReport []TGdProviderReportHour, codeInfos []TGdProviderCodeHour) {
	if len(providerReport) > 0 {
		newReport := []TGdProviderReportHour{}
		for i, _ := range providerReport {
			newReport = append(newReport, providerReport[i])
		}
		if len(newReport) > 0 {
			hourDb.Raw("delete from t_gd_provider_report_hour where hour=?", newReport[0].Hour).Exec()
			hourDb.InsertMulti(len(newReport), &newReport)
		}
	}
	if len(codeInfos) > 0 {
		hourDb.Raw("delete from t_gd_provider_code_hour where hour=?", codeInfos[0].Hour).Exec()
		hourDb.InsertMulti(len(codeInfos), &codeInfos)
	}
}

func insertApiReportLocal(apiReport []TGdReportHour, codeInfos []TGdApiCodeHour, db orm.Ormer) {
	if len(apiReport) > 0 {
		db.InsertMulti(len(apiReport), &apiReport)
	}
	if len(codeInfos) > 0 {
		db.Raw("delete from t_gd_api_code_hour where hour=?", codeInfos[0].Hour).Exec()
		db.InsertMulti(len(codeInfos), &codeInfos)
	}
}

func insertProviderReportLocal(providerReport []TGdProviderReportHour, codeInfos []TGdProviderCodeHour, db orm.Ormer) {
	if len(providerReport) > 0 {
		newReport := []TGdProviderReportHour{}
		for i, v := range providerReport {
			if (strings.Contains(v.ProviderName, "觅实") || strings.Contains(v.ProviderName, "A02-MS")) && strings.Contains(v.ProviderApiName, "MSN0") == false && strings.Contains(v.ProviderApiName, "MSM00001") == false && strings.Contains(v.ProviderApiName, "觅实N0") == false {
				continue
			}
			newReport = append(newReport, providerReport[i])
		}
		if len(newReport) > 0 {
			db.InsertMulti(len(newReport), &newReport)
		}
	}
	if len(codeInfos) > 0 {
		db.Raw("delete from t_gd_provider_code_hour where hour=?", codeInfos[0].Hour).Exec()
		db.InsertMulti(len(codeInfos), &codeInfos)
	}
}

func computeApiReportHour(start, end int64) {
	apiReport := computeApiReportNormal(start, end)
	codeInfos := computeApiReportCodeDistribution(apiReport, start, end)
	insertApiReport(apiReport, codeInfos)
}

func computeProviderReportHour(start, end int64) {
	providerReport := computeProviderReportNormal(start, end)
	codeInfos := computeProviderReportCodeDistribution(providerReport, start, end)
	insertProviderReport(providerReport, codeInfos)
}

func apiReportHour(hour int64) {
	end := hour
	for {
		if end-hour > 12*3600 {
			break
		}
		if hourDb.QueryTable("t_gd_report_hour").Filter("hour", hour).Exist() == false || (end-hour <= 2*3600) {
			computeApiReportHour(hour, hour+3600)
		}
		hour = hour - 3600
	}
}

func GetLocalDb(user, pass, addr string) orm.Ormer {
	if addr != "" {
		dataSource := fmt.Sprintf("%s:%s@tcp(%s)/%s?charset=%s", user, pass, addr, "db_gd_access_log", "utf8")
		if err := orm.RegisterDataBase("local", "mysql", dataSource, 2, 2); err != nil {
			fmt.Printf("%v\n", err.Error())
			return nil
		}
	}
	db = orm.NewOrm()
	db.Using("local")
	return db
}

func GernerateHourReport(start, end int64, isApi, isProvider bool, db orm.Ormer) {
	for i := start; i < end; i += 3600 {
		hourDb = orm.NewOrm()
		if isApi {
			apiReport := computeApiReportNormal(start, end)
			codeInfos := computeApiReportCodeDistribution(apiReport, start, end)
			insertApiReportLocal(apiReport, codeInfos, db)
		}
		if isProvider {
			providerReport := computeProviderReportNormal(start, end)
			codeInfos := computeProviderReportCodeDistribution(providerReport, start, end)
			insertProviderReportLocal(providerReport, codeInfos, db)

		}

	}
}

func providerReportHour(hour int64) {
	end := hour
	for {
		if end-hour > 12*3600 {
			break
		}
		if hourDb.QueryTable("t_gd_provider_report_hour").Filter("hour", hour).Exist() == false || (end-hour <= 2*3600) {
			computeProviderReportHour(hour, hour+3600)
		}
		hour = hour - 3600
	}
}

var hourDb orm.Ormer

func LogHourTask() {
	go func() {
		hourDb = orm.NewOrm()
		now := time.Now()
		hour := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, now.Location()).Unix() - 3600
		hourDb.Begin()
		apiReportHour(hour)
		providerReportHour(hour)
		hourDb.Commit()
		for {
			now := time.Now()
			next := now.Add(time.Hour)
			next = time.Date(next.Year(), next.Month(), next.Day(), next.Hour(), 2, 0, 0, next.Location())
			t := time.NewTimer(next.Sub(now))
			<-t.C
			t.Stop()
			hour := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, now.Location()).Unix()

			hourDb.Begin()
			apiReportHour(hour)
			providerReportHour(hour)
			hourDb.Commit()
			runtime.GC()
		}
	}()
}