jypkg/marketanalysis/marketanalysis.go

package marketanalysis

import (
	"app.yhyue.com/moapp/jybase/common"
	"app.yhyue.com/moapp/jybase/encrypt"
	elastic "app.yhyue.com/moapp/jybase/es"
	"app.yhyue.com/moapp/jybase/mongodb"
	"app.yhyue.com/moapp/jybase/mysql"
	"app.yhyue.com/moapp/jybase/redis"
	"app.yhyue.com/moapp/jypkg/common/src/qfw/util/jy"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"strings"
	"sync"
	"time"
)

var (
	MAPool            chan bool
	MATimeout         int
	MAProjectNumLimit int
	MAKeyWordsCount   int
	MinKeyWordsCount  int //关键词数量最低值 小于此数量，默认走线上
	MaxPCount         int //项目最大限制数量
	ProjectCount      int
	PtIndex           = "projectset"
	PtType            = "projectset"
	FieldsDetail      = `"purchasing","projectname.pname"`
	Fields            = `"projectname.pname"`
	BWCount           = Top1000
	BWDCount          = BWDistribution
)

func MAInit(limit, timeOut, projectNumLimit, keyWordsCount, pCount, minKCount, maxPCount, bwc, bwdc int, ptIndex, ptType string, fields []string) {
	if limit == 0 {
		limit = 5
	}
	MAPool = make(chan bool, limit)
	for i := 0; i < limit; i++ {
		MAPool <- true
	}
	if timeOut <= 0 {
		timeOut = 20
	}
	MATimeout = timeOut
	if projectNumLimit <= 0 {
		projectNumLimit = 600000
	}
	MAProjectNumLimit = projectNumLimit
	if keyWordsCount <= 0 {
		keyWordsCount = 300
	}
	MAKeyWordsCount = keyWordsCount

	if minKCount <= 0 {
		minKCount = 100
	}
	MinKeyWordsCount = minKCount
	if maxPCount <= 0 {
		maxPCount = 5000000
	}
	MaxPCount = maxPCount
	//项目数量
	ProjectCount = pCount
	//es 索引
	if ptIndex != "" {
		PtIndex = ptIndex
		PtType = ptType
	}
	//
	if len(fields) > 1 {
		Fields = fmt.Sprintf(`"%s"`, fields[0])
		FieldsDetail = fmt.Sprintf(`"%s"`, strings.Join(fields, `","`))
	}
	//
	if bwc > 0 {
		BWCount = bwc
	}
	if bwdc > 0 {
		BWDCount = bwdc
	}
}

// AnalysisRequestParam 接口原请求参数
type AnalysisRequestParam struct {
	KeysItemsStr   string //分析内容【字符串】结构和o_member_jy.a_items保持一致
	RangeTime      string //时间【字符串】 时间戳开始-结束时间戳
	RangeTimeExtra string //时间【字符串】前段回显使用
	Area           string //省份【对象字符串】
	Industry       string //行业【对象字符串】
	BuyerClass     string //采购单位类型【字符串】多个采购单位类型用逗号拼接
	Buyer          string //采购单位
	Winner         string //中标单位
	Sort           int    //排序：默认0：成交时间倒序；1：项目金额倒序
	PageSize       int    //默认每页10条
	PageNum        int    //默认当前第一页
	IsDetail       bool   //是否是项目明细请求
	MatchingMode   string //匹配方式 title：标题 content：项目名称/标的物
}

type viewKeyWord struct {
	Keyword  []string `json:"key"`       //关键词
	Appended []string `json:"appendkey"` //附加词
	Exclude  []string `json:"notkey"`    //排除词
	MatchWay int      `json:"matchway"`  //匹配模式
}

// keyWordGroup 订阅词结构体
type keyWordGroup struct {
	A_Key      []viewKeyWord `json:"a_key"`
	ItemName   string        `json:"s_item"`
	UpdateTime int64         `json:"updatetime"`
}

// AnalysisRequestFormat 格式化后参数
type AnalysisRequestFormat struct {
	KeysItems    []keyWordGroup
	Area, City   []string //省份城市
	STime, ETime int64    //开始结束时间
	Industry     []string //行业
	BuyerClass   []string //采购单位类型
	Buyer        string   //采购单位
	Winner       string   //中标单位
	Sort         int      //排序：默认0：成交时间倒序；1：项目金额倒序
	PageSize     int      //默认每页10条
	PageNum      int      //默认当前第一页
}

type AnalysisEntity struct {
	MgoRecordId   string
	BaseParam     AnalysisRequestParam
	FormatParam   AnalysisRequestFormat
	UId, Pid      string
	ProjectInfo   projectInfo
	Offline       int // 1-离线 2-实时
	State         int // 状态：默认0：生成中；1：已生成；2：已取消；-1：生成失败
	MgoUserId     string
	Phone         string // 手机号
	PositionId    int64
	OriginalTotal int64 // 数据总数
	KeysTotal     int64 //关键词数量
	PositionType  int
	EntId         int
	EntUserId     int
	Source        string
	Mgo           *mongodb.MongodbSim
	MySql         *mysql.Mysql
	Rs            int //分析来源
}

type projectInfo struct {
	Count int64
	List  []ProjectList
}

type ProjectList struct {
	Name       string   `json:"name"`       //项目名称
	Id         string   `json:"id"`         //项目id
	Area       string   `json:"area"`       //地区
	DealTime   int64    `json:"dealTime"`   //成交时间
	BidStatus  string   `json:"bidStatus"`  //项目类型
	BuyerClass string   `json:"buyerClass"` //采购单位类型
	Winner     []string `json:"winner"`     //中标单位
	WinnerId   []string `json:"winnerId"`   //中标单位id
	Buyer      string   `json:"buyer"`      //采购单位
	BidAmount  float64  `json:"bidAmount"`  //中标金额
	Budget     float64  `json:"budget"`     //预算
}

// ForMatData 获取格式化请求参数
func (a *AnalysisEntity) ForMatData() error {
	//格式化订阅词
	if err := json.Unmarshal([]byte(a.BaseParam.KeysItemsStr), &a.FormatParam.KeysItems); err != nil {
		return fmt.Errorf("关键词组格式异常")
	}
	if a.FormatParam.KeysItems == nil || len(a.FormatParam.KeysItems) == 0 {
		return fmt.Errorf("请选择关键词组")
	}
	var flag bool
	if a.Rs == 0 { //普通分析（非pdf）
		// 判断关键词是不是为空
		for i := 0; i < len(a.FormatParam.KeysItems); i++ {
			items := a.FormatParam.KeysItems[i]
			for j := 0; j < len(items.A_Key); j++ {
				AKey := items.A_Key[j]
				if len(AKey.Keyword) > 0 {
					var keys []string
					for _, kv := range AKey.Keyword {
						if strings.TrimSpace(kv) != "" {
							keys = append(keys, kv)
						}
					}
					AKey.Keyword = []string{}
					if len(keys) > 0 {
						AKey.Keyword = keys
						flag = true
						break
					}
				}
				if len(AKey.Keyword) == 0 {
					items.A_Key = append(items.A_Key[:j], items.A_Key[j+1:]...)
				}
			}
			if flag {
				break
			}
			a.FormatParam.KeysItems = append(a.FormatParam.KeysItems[:i], a.FormatParam.KeysItems[i+1:]...)
		}
		if !flag {
			return fmt.Errorf("请选择关键词组")
		}
	}

	//格式化时间段
	if timeArr := strings.Split(a.BaseParam.RangeTime, "-"); len(timeArr) == 2 {
		a.FormatParam.STime = common.Int64All(timeArr[0])
		a.FormatParam.ETime = common.Int64All(timeArr[1])
		if a.FormatParam.STime == 0 || a.FormatParam.ETime == 0 {
			return fmt.Errorf("开始时间和结束时间不能为空")
		}
	} else {
		return fmt.Errorf("时间戳格式异常")
	}
	//格式化省份、城市
	if areaStr := strings.TrimSpace(a.BaseParam.Area); areaStr != "" {
		imap := map[string][]string{}
		if err := json.Unmarshal([]byte(a.BaseParam.Area), &imap); err != nil {
			return fmt.Errorf("非法地区信息")
		}
		var city, area []string
		for name, v := range imap {
			if len(v) == 0 {
				area = append(area, name)
			} else {
				for _, vv := range v {
					city = append(city, vv)
				}
			}
		}
		a.FormatParam.Area = area
		a.FormatParam.City = city
	}
	//格式化行业
	if industryStr := strings.TrimSpace(a.BaseParam.Industry); industryStr != "" {
		imap := map[string][]string{}
		if err := json.Unmarshal([]byte(industryStr), &imap); err != nil {
			return fmt.Errorf("非法行业信息")
		}
		var farr []string
		for name, v := range imap {
			for _, vv := range v {
				farr = append(farr, fmt.Sprintf("%s_%s", name, vv))
			}
		}
		if len(farr) > 0 {
			//P510 行业：其它
			if qt := jy.IndustryHandle(strings.Join(farr, ",")); len(qt) > 0 {
				farr = append(farr, qt...)
			}
		}
		a.FormatParam.Industry = farr
	}
	//格式化类型
	if buyerClassStr := strings.TrimSpace(a.BaseParam.BuyerClass); buyerClassStr != "" {
		a.FormatParam.BuyerClass = strings.Split(buyerClassStr, ",")
	}
	//中标企业
	a.FormatParam.Winner = a.BaseParam.Winner
	//采购单位
	a.FormatParam.Buyer = a.BaseParam.Buyer
	//排序
	a.FormatParam.Sort = common.If(a.BaseParam.Sort != 0 && a.BaseParam.Sort != 1, 0, a.BaseParam.Sort).(int)
	if a.BaseParam.PageNum*a.BaseParam.PageSize > ProjectCount {
		a.BaseParam.PageNum = ProjectCount / a.BaseParam.PageSize
	}
	//当前页码
	a.FormatParam.PageNum = common.If(a.BaseParam.PageNum < 1 || a.BaseParam.PageNum > 1000, 1, a.BaseParam.PageNum).(int)
	//默认每页10条
	a.FormatParam.PageSize = common.If(a.BaseParam.PageSize < 1 || a.BaseParam.PageSize > 100, 50, a.BaseParam.PageSize).(int)
	return nil
}

// ForMatData 获取格式化请求参数
func (a *AnalysisEntity) ForMatDataPdf() (string, error) {
	//格式化订阅词
	if err := json.Unmarshal([]byte(a.BaseParam.KeysItemsStr), &a.FormatParam.KeysItems); err != nil {
		return "", fmt.Errorf("关键词组格式异常")
	}

	//格式化时间段
	if timeArr := strings.Split(a.BaseParam.RangeTime, "-"); len(timeArr) == 2 {
		a.FormatParam.STime = common.Int64All(timeArr[0])
		a.FormatParam.ETime = common.Int64All(timeArr[1])
		if a.FormatParam.STime == 0 || a.FormatParam.ETime == 0 {
			return "", fmt.Errorf("开始时间和结束时间不能为空")
		}
	} else {
		return "", fmt.Errorf("时间戳格式异常")
	}
	//格式化省份、城市
	if areaStr := strings.TrimSpace(a.BaseParam.Area); areaStr != "" {
		imap := map[string][]string{}
		if err := json.Unmarshal([]byte(a.BaseParam.Area), &imap); err != nil {
			return "", fmt.Errorf("非法地区信息")
		}
		var city, area []string
		for name, v := range imap {
			if len(v) == 0 {
				area = append(area, name)
			} else {
				for _, vv := range v {
					city = append(city, vv)
				}
			}
		}
		a.FormatParam.Area = area
		a.FormatParam.City = city
	}
	//格式化行业
	if industryStr := strings.TrimSpace(a.BaseParam.Industry); industryStr != "" {
		imap := map[string][]string{}
		if err := json.Unmarshal([]byte(industryStr), &imap); err != nil {
			return "", fmt.Errorf("非法行业信息")
		}
		var farr []string
		for name, v := range imap {
			for _, vv := range v {
				farr = append(farr, fmt.Sprintf("%s_%s", name, vv))
			}
		}
		a.FormatParam.Industry = farr
	}
	//格式化类型
	if buyerClassStr := strings.TrimSpace(a.BaseParam.BuyerClass); buyerClassStr != "" {
		a.FormatParam.BuyerClass = strings.Split(buyerClassStr, ",")
	}
	//中标企业
	a.FormatParam.Winner = a.BaseParam.Winner
	//采购单位
	a.FormatParam.Buyer = a.BaseParam.Buyer
	//排序
	a.FormatParam.Sort = common.If(a.BaseParam.Sort != 0 && a.BaseParam.Sort != 1, 0, a.BaseParam.Sort).(int)
	if a.BaseParam.PageNum*a.BaseParam.PageSize > ProjectCount {
		a.BaseParam.PageNum = ProjectCount / a.BaseParam.PageSize
	}
	//当前页码
	a.FormatParam.PageNum = common.If(a.BaseParam.PageNum < 1 || a.BaseParam.PageNum > 1000, 1, a.BaseParam.PageNum).(int)
	//默认每页10条
	a.FormatParam.PageSize = common.If(a.BaseParam.PageSize < 1 || a.BaseParam.PageSize > 100, 50, a.BaseParam.PageSize).(int)
	data := map[string]interface{}{
		"s_keysItems":      a.BaseParam.KeysItemsStr,
		"s_rangeTime":      a.BaseParam.RangeTime,
		"s_rangeTimeExtra": a.BaseParam.RangeTimeExtra,
		"s_area":           a.BaseParam.Area,
		"s_industry":       a.BaseParam.Industry,
		"s_buyerClass":     a.BaseParam.BuyerClass,
		"s_matchingMode":   a.BaseParam.MatchingMode,
		"s_userId":         a.UId,
		"s_parentId":       a.Pid,
		"s_mgoUserId":      a.MgoUserId,
		"i_positionId":     a.PositionId,
		"s_phone":          a.Phone,
		"audit":            1,
	}
	if a.OriginalTotal > 0 {
		data["l_originalTotal"] = a.OriginalTotal
	}
	if a.Source != "" {
		data["source"] = a.Source
	}
	rs, b := a.Mgo.FindOne(ReportHistoryTable, data)
	if b && rs != nil && len(*rs) > 0 {
		return common.InterfaceToStr((*rs)["_id"]), errors.New("当期分析已存在")
	}
	return "", nil
}

// GetProjectInfoList 项目明细
func (a *AnalysisEntity) GetProjectInfoList() error {
	var (
		queryDefault = `,"sort": [{%s}],"from": %d,"size": %d`
		start        = (a.FormatParam.PageNum - 1) * a.FormatParam.PageSize
		sort         = `"jgtime": "desc"`
	)
	if a.FormatParam.Sort > 0 {
		sort = `"bidamount": "desc","budget": "desc"`
	}
	countSql := fmt.Sprintf(a.GetCommonQuerySql(), "")
	queryDefault = fmt.Sprintf(queryDefault, sort, start, a.FormatParam.PageSize)
	finalSql := fmt.Sprintf(a.GetCommonQuerySql(), queryDefault)
	log.Println("finalSql:", finalSql)
	count, hits := elastic.GetWithCount(PtIndex, PtType, countSql, finalSql)
	//hits, count := elastic.GetOA(PtIndex, PtType, finalSql)
	if count > 0 {
		a.ProjectInfo.Count = count
		source := *hits
		for _, v := range source {
			var winnerIdArr []string
			if common.ObjToString(v["s_winner"]) != "" && v["entidlist"] != nil {
				idObjs, _ := v["entidlist"].([]interface{})
				for _, v := range common.ObjArrToStringArr(idObjs) {
					if v != "" && v != "-" {
						v = encrypt.EncodeArticleId2ByCheck(v)
					}
					winnerIdArr = append(winnerIdArr, v)
				}
			}
			a.ProjectInfo.List = append(a.ProjectInfo.List, ProjectList{
				Name:       common.ObjToString(v["projectname"]),
				Id:         encrypt.EncodeArticleId2ByCheck(common.ObjToString(v["id"])),
				Area:       common.ObjToString(v["area"]),
				DealTime:   common.Int64All(v["jgtime"]), //截止时间
				BidStatus:  common.ObjToString(v["bidstatus"]),
				BuyerClass: common.ObjToString(v["buyerclass"]),
				Winner:     strings.Split(common.ObjToString(v["s_winner"]), ","),
				WinnerId:   winnerIdArr,
				Buyer:      common.ObjToString(v["buyer"]),
				BidAmount:  common.Float64All(v["bidamount"]), //中标金额
				Budget:     common.Float64All(v["budget"]),    //预算
			})
		}
	}
	return nil
}

// SaveAnalysisRecord 保存分析记录
func (a *AnalysisEntity) SaveAnalysisRecord() error {
	if a.Offline == ValueRealTime {
		a.State = ReportStateGenerated
	} else {
		a.State = ReportStateGenerating
	}
	data := map[string]interface{}{
		"s_keysItems":      a.BaseParam.KeysItemsStr,
		"s_rangeTime":      a.BaseParam.RangeTime,
		"s_rangeTimeExtra": a.BaseParam.RangeTimeExtra,
		"s_area":           a.BaseParam.Area,
		"s_industry":       a.BaseParam.Industry,
		"s_buyerClass":     a.BaseParam.BuyerClass,
		"s_matchingMode":   a.BaseParam.MatchingMode,
		"s_userId":         a.UId,
		"s_parentId":       a.Pid,
		"i_state":          common.If(a.Source == "analysisPDF", 1, a.State), //状态：默认0：生成中；1：已生成；2：已取消；-1：生成失败
		"l_updateTime":     time.Now().Unix(),                                //生成时间 or 取消时间
		"l_createTime":     time.Now().Unix(),
		"i_offline":        common.If(a.Source == "analysisPDF", 2, a.Offline),
		"s_mgoUserId":      a.MgoUserId,
		"i_positionId":     a.PositionId,
		"s_phone":          a.Phone,
	}
	if a.OriginalTotal > 0 {
		data["l_originalTotal"] = a.OriginalTotal
	}
	if a.KeysTotal > 0 {
		data["l_keysTotal"] = a.KeysTotal
	}
	if a.Source != "" {
		data["source"] = a.Source
	}
	a.MgoRecordId = a.Mgo.Save(ReportHistoryTable, data)
	if a.MgoRecordId == "" {
		return fmt.Errorf("分析创建异常")
	}
	return nil
}

// GetAnalysisFromMgoDb 从数据库中获取分析记录
func (a *AnalysisEntity) GetAnalysisFromMgoDb() error {
	if a.MgoRecordId == "" {
		return fmt.Errorf("缺少参数")
	}
	queryMap := map[string]interface{}{
		"_id":   mongodb.StringTOBsonId(a.MgoRecordId),
		"i_del": map[string]interface{}{"$ne": 1},
	}
	if a.UId == a.Pid { //主账号
		queryMap["s_parentId"] = a.Pid
	} else {
		queryMap["s_userId"] = a.UId
	}
	res, _ := a.Mgo.FindOne(ReportHistoryTable, queryMap)
	if res == nil || len(*res) == 0 {
		return fmt.Errorf("未查询到相关数据")
	}
	a.Offline = common.IntAll((*res)["i_offline"])
	a.State = common.IntAll((*res)["i_state"])
	a.BaseParam.KeysItemsStr, _ = (*res)["s_keysItems"].(string)
	a.BaseParam.RangeTime, _ = (*res)["s_rangeTime"].(string)
	a.BaseParam.RangeTimeExtra, _ = (*res)["s_rangeTimeExtra"].(string)
	a.BaseParam.Area, _ = (*res)["s_area"].(string)
	a.BaseParam.Industry, _ = (*res)["s_industry"].(string)
	a.BaseParam.BuyerClass, _ = (*res)["s_buyerClass"].(string)
	a.BaseParam.MatchingMode, _ = (*res)["s_matchingMode"].(string)
	return nil
}

// removeEmptyRecord 删除空报告记录
func (a *AnalysisEntity) removeEmptyRecord() {
	queryMap := map[string]interface{}{
		"_id": mongodb.StringTOBsonId(a.MgoRecordId),
	}
	if a.UId == a.Pid { //主账号
		queryMap["s_parentId"] = a.Pid
	} else {
		queryMap["s_userId"] = a.UId
	}
	a.Mgo.Update(ReportHistoryTable, queryMap, map[string]interface{}{
		"$set": map[string]interface{}{"i_del": 1},
	}, false, false)
	//log.Println("删除空报告", queryMap)
}

// GetRecordList 获取分析记录
func (a *AnalysisEntity) GetRecordList(pageNum, PageSize int, positionType int, entId, entUserId int) (total int, list []map[string]interface{}) {
	queryMap := map[string]interface{}{
		"i_del": map[string]interface{}{"$ne": 1},
	}
	if a.UId == a.Pid { //主账号
		queryMap["s_parentId"] = a.Pid
	} else {
		queryMap["s_userId"] = a.UId
	}
	if pageNum == 1 {
		total = a.Mgo.Count(ReportHistoryTable, queryMap)
		if total == 0 {
			return
		}
	} else {
		total = -1
	}
	res, _ := a.Mgo.Find(ReportHistoryTable, queryMap, `{"l_createTime":-1}`, nil, false, (pageNum-1)*PageSize, PageSize)
	if res == nil || len(*res) == 0 {
		return
	}
	var ids []string
	for _, m := range *res {
		ids = append(ids, common.InterfaceToStr(m["_id"]))
	}
	idMap := make(map[string]map[string]interface{})
	tName, _ := GetMongoColl(MarketScaleMain)
	mids, _ := a.Mgo.Find(tName, map[string]interface{}{
		"s_m_id": map[string]interface{}{
			"$in": ids,
		},
	}, "", `{"market_profile":1,"s_m_id":1}`, false, -1, -1)
	if mids != nil && len(*mids) > 0 {
		for _, m := range *mids {
			marketProfile, _ := m["market_profile"].(map[string]interface{})
			if common.IntAll(marketProfile["project_count"]) > 0 {
				idMap[common.InterfaceToStr(m["s_m_id"])] = marketProfile
			}
		}
	}

	//用户消息开关
	open := GetMsgOpen(a.Mgo, a.MgoUserId, positionType, entId, entUserId)
	for _, row := range *res {
		var status int
		if row["i_state"] != nil {
			status = common.IntAll(row["i_state"])
			if status == ReportStateFailed {
				status = ReportStateGenerating // 生成失败对外还是展示为生成中
			}
		}
		var (
			isDownload bool
		)
		marketProfile := make(map[string]interface{})
		if marketProfile = idMap[mongodb.BsonIdToSId(row["_id"])]; marketProfile != nil && common.IntAll(marketProfile["project_count"]) > 0 && status == 1 {
			isDownload = true
		}
		//P510  回显 移除 其它
		industry := common.ObjToString(row["s_industry"])
		if industry != "" {
			industry = strings.ReplaceAll(industry, "\"其它\"", "")
		}
		data := map[string]interface{}{
			"id":               encodeId(mongodb.BsonIdToSId(row["_id"])),
			"keysItems":        common.ObjToString(row["s_keysItems"]),
			"area":             common.ObjToString(row["s_area"]),
			"industry":         industry,
			"buyerclass":       common.ObjToString(row["s_buyerClass"]),
			"rangeTime":        common.ObjToString(row["s_rangeTime"]),
			"s_rangeTimeExtra": common.ObjToString(row["s_rangeTimeExtra"]),
			"createTime":       common.Int64All(row["l_createTime"]),
			"matchingMode":     common.ObjToString(row["s_matchingMode"]), //项目匹配方式
			"state":            common.If(row["i_state"] == nil, nil, status),
			"updateTime":       common.Int64All(row["l_updateTime"]),
			"msgOpen":          open,
			"isDownload":       isDownload,
			"marketProfile":    marketProfile,
		}
		list = append(list, data)
	}
	return
}

func (a *AnalysisEntity) GetRecordPdfList(pageNum, PageSize int) (total int, list []map[string]interface{}) {
	queryMap := map[string]interface{}{
		"i_del":  map[string]interface{}{"$ne": 1},
		"source": "analysisPDF",
	}
	if a.UId == a.Pid { //主账号
		queryMap["s_parentId"] = a.Pid
	} else {
		queryMap["s_userId"] = a.UId
	}
	if pageNum == 1 {
		total = a.Mgo.Count(ReportHistoryTable, queryMap)
		if total == 0 {
			return
		}
	} else {
		total = -1
	}
	res, _ := a.Mgo.Find(ReportHistoryTable, queryMap, `{"l_createTime":-1}`, nil, false, (pageNum-1)*PageSize, PageSize)
	if res == nil || len(*res) == 0 {
		return
	}
	for _, row := range *res {
		data := map[string]interface{}{
			"id":               encodeId(mongodb.BsonIdToSId(row["_id"])),
			"keysItems":        common.ObjToString(row["s_keysItems"]),
			"area":             common.ObjToString(row["s_area"]),
			"industry":         common.ObjToString(row["s_industry"]),
			"buyerclass":       common.ObjToString(row["s_buyerClass"]),
			"rangeTime":        common.ObjToString(row["s_rangeTime"]),
			"s_rangeTimeExtra": common.ObjToString(row["s_rangeTimeExtra"]),
			"createTime":       common.Int64All(row["l_createTime"]),
			"matchingMode":     common.ObjToString(row["s_matchingMode"]), //项目匹配方式
			"updateTime":       common.Int64All(row["l_updateTime"]),
			"audit":            common.Int64All(row["audit"]),
		}
		list = append(list, data)
	}
	return
}

// GetQueryItem 获取查询条件，前端回显使用
func (a *AnalysisEntity) getQueryItem() (map[string]interface{}, error) {
	return map[string]interface{}{
		"keysItems":        a.BaseParam.KeysItemsStr,
		"area":             a.BaseParam.Area,
		"industry":         a.BaseParam.Industry,
		"buyerclass":       a.BaseParam.BuyerClass,
		"rangeTime":        a.BaseParam.RangeTime,
		"s_rangeTimeExtra": a.BaseParam.RangeTimeExtra,
		"matchingMode":     a.BaseParam.MatchingMode,
	}, nil
}

// GetPartResult 分块儿获取报告内容
func (a *AnalysisEntity) GetPartResult(flag int) (map[string]interface{}, error) {
	defer common.Catch()
	if flag == MarketQueryItem { //返回查询内容
		return a.getQueryItem()
	}
	thisCacheKey := fmt.Sprintf(ReportCacheKey, a.MgoRecordId, flag)
	if cacheData := redis.Get(ReportCacheDB, thisCacheKey); cacheData != nil {
		if cacheMap, ok := cacheData.(map[string]interface{}); ok && len(cacheMap) > 0 {
			return cacheMap, nil
		}
	}
	rData, err := func() (map[string]interface{}, error) {
		//控制并发&&超时返回超时异常
		select {
		case <-time.After(time.Duration(MATimeout) * time.Second * 20):
			return nil, fmt.Errorf("查询超时，请稍后重试")
		case <-MAPool:
		}
		start := time.Now()
		defer func() {
			MAPool <- true
			log.Printf("report %s[%d] speed %d ms\n", a.MgoRecordId, flag, time.Now().Sub(start).Milliseconds())
		}()
		//校验报告是否合法
		//if flag != MarketScaleMain {
		//	if exists, _ := redis.Exists(ReportCacheDB, fmt.Sprintf(ReportCacheKey, a.MgoRecordId, 1)); !exists {
		//		return nil, fmt.Errorf("报告异常请求，请刷新重试")
		//	}
		//}
		// 1. 查mongo
		var rData map[string]interface{}
		var err error
		rData, err = a.GetMongoData(flag)
		// 查到数据则直接返回
		if err == nil && rData != nil {
			return rData, nil
		}
		// 2.没有查询到数据 判断是不是离线的
		//  正常情况下正在离线生的不会走到这里
		// 离线的应该生成完报告之后才会调用获取结果接口 这里处理是防止直接调接口传正在离线生的报告
		if a.Offline == ValueOffline || a.Offline == 0 {
			return nil, err
		}
		// 3. 实时则查es
		rData, err = a.RealTimeQuery(flag)
		if err == nil && len(rData) > 0 {
			// 4.存库
			a.SaveMongoReport(rData, flag)
			return rData, err
		}
		return nil, err
	}()
	if err == nil && rData != nil && len(rData) > 0 {
		delete(rData, "s_m_id")
		delete(rData, "_id")
		redis.Put(ReportCacheDB, thisCacheKey, rData, ReportCacheTime)
	}
	return rData, err
}

// 离线  获取采购单位和中部单位数据
func (a *AnalysisEntity) BWData() (buyers []string, winners []string) {
	sql := `"buyer_terms": {"terms": {"field": "buyer","size": 50000}},"winner_terms": {"terms": {"field": "s_winner","size": 50000}}`
	finalSql := fmt.Sprintf(a.GetCommonQuerySqlWithAggs(), sql)
	res := GetAggs(PtIndex, PtType, finalSql)
	if res == nil || len(res) == 0 {
		return
	}
	var bw = BWBuckets{}
	for name, object := range res {
		bArr, err := object.MarshalJSON()
		if len(bArr) == 0 || err != nil {
			continue
		}
		if name == "buyer_terms" {
			if json.Unmarshal(bArr, &bw.Buyerterms) != nil {
				continue
			}
		} else if name == "winner_terms" {
			if json.Unmarshal(bArr, &bw.Winnerterms) != nil {
				continue
			}
		}
	}
	for _, bv := range bw.Buyerterms.Buckets {
		buyers = append(buyers, bv.Key)
	}
	for _, wv := range bw.Winnerterms.Buckets {
		winners = append(winners, wv.Key)
	}
	return
}

// 实时查询
func (a *AnalysisEntity) RealTimeQuery(flag int) (map[string]interface{}, error) {
	switch flag {
	case MarketScaleMain:
		rData, err := a.MarketTime()
		//非离线
		if a.Offline != ValueOffline && err != nil { //若无报告内容，删除报告记录
			go a.removeEmptyRecord()
		}
		if rData != nil && a.Offline == ValueOffline {
			//采购单位和中部单位 数据
			rData["buyers"], rData["winners"] = a.BWData()
		}
		return rData, err
	case MarketTopProject:
		return a.ProjectTop10()
	case MarketProjectAllData:
		return a.AllData()
	case MarketScaleRefine:
		return a.MarketScaleRefineQuery()
	case MarketBuyerAndWinner:
		return a.BuyerWinnerAnalysis(), nil
	}
	return nil, fmt.Errorf("未知请求")
}

// GetMongoData 从mongo库查询数据
func (a *AnalysisEntity) GetMongoData(flag int) (map[string]interface{}, error) {
	collName, err := GetMongoColl(flag)
	if err != nil || collName == "" {
		return nil, fmt.Errorf("未知请求")
	}
	// 查询
	query := map[string]interface{}{
		"s_m_id": a.MgoRecordId,
	}
	data, b := a.Mgo.FindOne(collName, query)
	if !b || data == nil || len(*data) == 0 {
		log.Println("没有查询到数据", b, query, collName)
		return nil, nil
	}
	delete(*data, "s_m_id")
	delete(*data, "_id")
	return *data, nil
}

// GetMongoColl 获取mgo库对应的Coll
func GetMongoColl(flag int) (string, error) {
	switch flag {
	case MarketScaleMain:
		return CollMarketScaleMain, nil
	case MarketTopProject:
		return CollMarketTopProject, nil
	case MarketProjectAllData:
		return CollMarketProjectAllData, nil
	case MarketScaleRefine:
		return CollMarketScaleRefine, nil
	case MarketBuyerAndWinner:
		return CollMarketBuyerAndWinner, nil
	}
	return "", fmt.Errorf("未知类型")
}

// SaveMongoReport 数据存mongo库
func (a *AnalysisEntity) SaveMongoReport(rData map[string]interface{}, flag int) {
	collName, _ := GetMongoColl(flag)
	rData["s_m_id"] = a.MgoRecordId
	b, err := json.Marshal(rData)
	if err != nil {
		log.Println("JSON marshal error:", err)
	} else {
		var saveMap map[string]interface{}
		err = json.Unmarshal(b, &saveMap)
		if err != nil {
			log.Println("JSON Unmarshal error:", err)
			a.Mgo.Save(collName, rData)
		} else {
			a.Mgo.Save(collName, saveMap)
		}
	}
}

// GetAnalyzingReport 是否有正在分析的离线报告
func (a *AnalysisEntity) GetAnalyzingReport() string {
	query := map[string]interface{}{
		"s_userId": a.UId,
		"i_state":  map[string]interface{}{"$in": []int{ReportStateGenerating, ReportStateFailed}}, //状态：默认0：生成中；1：已生成；2：已取消；-1：生成失败
		"i_del":    map[string]interface{}{"$ne": 1},
	}
	//log.Println("query:", query)
	rs, b := a.Mgo.FindOne(ReportHistoryTable, query)
	//log.Println("rs,b:", rs, b)
	if b && rs != nil {
		return mongodb.BsonIdToSId((*rs)["_id"])
	}
	return ""
}

// IsOffline 判断是否符合在线分析的条件
func (a *AnalysisEntity) IsOffline() {
	//离线生成：订阅词（关键词+排除词）超过300个（数量支持配置），或单次分析数据超过60万条，则离线生成；
	//  查询数据量
	countSql := fmt.Sprintf(a.GetCommonQuerySql(), "")
	log.Println(a.Phone, "IsOffline count SQL:", countSql)
	now := time.Now()
	dataCount := elastic.Count(PtIndex, PtType, countSql)
	log.Println(a.Phone, "--IsOffline 数据量：", dataCount, "---", MaxPCount)
	if dataCount > int64(MaxPCount) { //提示用户 数据量过大，修改生成报告条件
		a.Offline = ValueError
		return
	}
	a.OriginalTotal = dataCount
	keyCount := 0
	for i := 0; i < len(a.FormatParam.KeysItems); i++ {
		items := a.FormatParam.KeysItems[i]
		for j := 0; j < len(items.A_Key); j++ {
			AKey := items.A_Key[j]
			for k := 0; k < len(AKey.Keyword); k++ {
				keyCount++
			}
			for k := 0; k < len(AKey.Appended); k++ {
				keyCount++
			}
			for k := 0; k < len(AKey.Exclude); k++ {
				keyCount++
			}
		}
	}
	a.KeysTotal = int64(keyCount)
	//关键词数量 < 100 ，默认走线上
	if keyCount < MinKeyWordsCount {
		a.Offline = ValueRealTime
		return
	}
	if keyCount > MAKeyWordsCount {
		// 查询配置
		if mac := a.getMarUserAccount(); mac != nil {
			if keyCount >= mac.Threshold {
				a.Offline = ValueOffline
				return
			}
		} else {
			a.Offline = ValueOffline
			return
		}
	}
	log.Println("IsOffline 统计数据量耗时：", time.Since(now))
	if int(dataCount) > MAProjectNumLimit {
		a.Offline = ValueOffline
		return
	}
	a.Offline = ValueRealTime
	return
}

type MarUserAccount struct {
	Threshold int // 关键词离线标准量
}

// 获取离线市场报告分析关键词标准信息
func (a *AnalysisEntity) getMarUserAccount() *MarUserAccount {
	rs := a.MySql.SelectBySql(fmt.Sprintf("SELECT threshold FROM %s where  position_id=? and state = 0;", TablejianyuMarUserAccount), a.PositionId)
	if rs != nil && len(*rs) > 0 {
		return &MarUserAccount{Threshold: common.IntAll((*rs)[0]["threshold"])}
	}
	return nil
}
func (a *AnalysisEntity) GetReportState() (generated bool, needUpdate bool, err error) {
	// 查库
	err = a.GetAnalysisFromMgoDb()
	if err != nil {
		return
	}
	// 已经生成的可以直接返回
	if a.State == ReportStateGenerated {
		generated = true
		return
	}
	// 该字段没有值 说明是还没有被更新的历史数据 需要更新
	// 且判断之后后续需要更新i_state和i_offline字段
	if a.Offline == 0 {
		needUpdate = true
	}
	// 如果没有值的话 说明这是历史数据  需要接着走下面的流程进行判断
	// 格式化数据 用于后面校验个数
	if err = a.ForMatData(); err != nil {
		return
	}
	return

}

// Cancel 取消正在分析中的报告
func (a *AnalysisEntity) Cancel() (bool, error) {
	// 取消报告
	queryMap := map[string]interface{}{
		"_id":     a.MgoRecordId,
		"i_state": map[string]interface{}{"$in": []int{ReportStateGenerating, ReportStateFailed}}, //状态：默认0：生成中；1：已生成；2：已取消；-1：生成失败
		"i_del":   map[string]interface{}{"$ne": 1},
	}
	if a.UId == a.Pid { //主账号
		queryMap["s_parentId"] = a.Pid
	} else {
		queryMap["s_userId"] = a.UId
	}
	//验证
	rs, b := a.Mgo.FindOne(ReportHistoryTable, queryMap)
	//
	if !b || rs == nil {
		return false, fmt.Errorf("未查询到该记录")
	}
	update := a.Mgo.UpdateById(ReportHistoryTable, a.MgoRecordId, map[string]interface{}{"$set": map[string]interface{}{
		"l_updateTime": time.Now().Unix(),
		"i_state":      ReportStateCanceled,
	}})
	if !update {
		log.Println("分析报告取消失败：", a.MgoRecordId)
		return false, fmt.Errorf("取消失败")
	}
	// redis里面放取消标识
	redis.Put(ReportCacheDB, fmt.Sprintf(ReportCanceledKey, a.MgoRecordId), 1, ReportCanceledTime)
	return update, nil
}

// UpdateOffline 更新报告是否是离线报告
func (a *AnalysisEntity) UpdateOffline(offline bool) bool {
	set := map[string]interface{}{
		"i_state":      common.If(offline, ReportStateGenerating, ReportStateGenerated),
		"i_offline":    common.If(offline, ValueOffline, ValueRealTime),
		"l_updateTime": time.Now().Unix(),
		"s_mgoUserId":  a.MgoUserId, // 这里更新这些字段是因为这几个字段是p437 版本新加上的 历史数据没有这些字段
		"s_positionId": a.PositionId,
		"s_phone":      a.Phone,
	}
	if a.OriginalTotal > 0 {
		set["l_originalTotal"] = a.OriginalTotal
	}
	data := map[string]interface{}{
		"$set": set,
	}
	update := a.Mgo.UpdateById(ReportHistoryTable, a.MgoRecordId, data)
	if !update {
		log.Println("UpdateOffline 更新报告状态失败:", data, a.MgoRecordId)
	}
	return update
}

// UpdateState 更新报告生成状态
func (a *AnalysisEntity) UpdateState(state int) bool {
	setMap := map[string]interface{}{
		"i_state":      state,
		"l_updateTime": time.Now().Unix(),
	}
	if state == ReportStateGenerated {
		setMap["l_finishTime"] = time.Now().Unix()
	}
	data := map[string]interface{}{
		"$set": setMap,
	}
	update := a.Mgo.UpdateById(ReportHistoryTable, a.MgoRecordId, data)
	if !update {
		log.Println("UpdateState 更新报告生成状态失败:", data, a.MgoRecordId)
	}
	return update
}

// Delete 删除正在分析中的报告
func (a *AnalysisEntity) Delete() (bool, error) {
	// 删除报告
	queryMap := map[string]interface{}{
		"_id":   a.MgoRecordId,
		"i_del": map[string]interface{}{"$ne": 1},
	}
	if a.UId == a.Pid { //主账号
		queryMap["s_parentId"] = a.Pid
	} else {
		queryMap["s_userId"] = a.UId
	}
	//验证
	rs, b := a.Mgo.FindOne(ReportHistoryTable, queryMap)
	//
	if !b || rs == nil {
		return false, fmt.Errorf("未查询到该记录")
	}
	update := a.Mgo.UpdateById(ReportHistoryTable, a.MgoRecordId, map[string]interface{}{"$set": map[string]interface{}{
		"l_updateTime": time.Now().Unix(),
		"i_del":        1,
	}})
	if !update {
		log.Println("分析报告删除失败：", a.MgoRecordId)
		return false, fmt.Errorf("删除失败")
	}
	// redis里面放取消标识
	redis.Put(ReportCacheDB, fmt.Sprintf(ReportCanceledKey, a.MgoRecordId), 1, ReportCanceledTime)
	return update, nil
}

// GetCommonQuerySql 公共筛选
func (a *AnalysisEntity) GetCommonQuerySql() string {
	var musts, bools []string
	//时间
	musts = append(musts, fmt.Sprintf(`{"range":{"jgtime":{"gte":%d,"lte":%d}}}`, a.FormatParam.STime, a.FormatParam.ETime))
	//地区
	if len(a.FormatParam.Area) > 0 || len(a.FormatParam.City) > 0 {
		var areaCity []string
		if len(a.FormatParam.Area) > 0 {
			areaCity = append(areaCity, fmt.Sprintf(`{"terms":{"area":["%s"]}}`, strings.Join(a.FormatParam.Area, `","`)))
		}
		if len(a.FormatParam.City) > 0 {
			areaCity = append(areaCity, fmt.Sprintf(`{"terms":{"city":["%s"]}}`, strings.Join(a.FormatParam.City, `","`)))
		}
		musts = append(musts, fmt.Sprintf(`{"bool":{"should":[%s],"minimum_should_match": 1}}`, strings.Join(areaCity, ",")))
	}
	//行业
	if len(a.FormatParam.Industry) > 0 {
		musts = append(musts, fmt.Sprintf(`{"terms":{"subscopeclass":["%s"]}}`, strings.Join(a.FormatParam.Industry, `","`)))
	}
	//类型
	if len(a.FormatParam.BuyerClass) > 0 {
		musts = append(musts, fmt.Sprintf(`{"terms":{"buyerclass":["%s"]}}`, strings.Join(a.FormatParam.BuyerClass, `","`)))
	}
	//分析报告中标状态限制
	musts = append(musts, fmt.Sprintf(queryBoolMust, pSearchDecMust))
	//订阅词
	for _, v := range getAllKeywordArr(a.FormatParam.KeysItems) {
		if sql := getKeyWordSql(v, a.BaseParam.MatchingMode); sql != "" {
			bools = append(bools, sql)
		}
	}
	//中标企业
	if a.FormatParam.Winner != "" {
		var winnerId string
		if mongodb.IsObjectIdHex(a.FormatParam.Winner) {
			winnerId = a.FormatParam.Winner
		} else {
			rData := elastic.Get("qyxy", "qyxy", fmt.Sprintf(`{"query": {"bool": {"must": [{"term": {"company_name": "%s"}}]}},"_source":["_id"],"size": 1}`, a.FormatParam.Winner))
			if rData != nil && len(*rData) > 0 {
				winnerId = common.InterfaceToStr((*rData)[0]["_id"])
			}
		}
		if winnerId != "" {
			musts = append(musts, fmt.Sprintf(`{"term":{"entidlist":"%s"}}`, winnerId))
		} else {
			musts = append(musts, fmt.Sprintf(`{"term":{"s_winner":"%s"}}`, a.FormatParam.Winner))
		}
	}
	//采购单位
	if a.FormatParam.Buyer != "" {
		musts = append(musts, fmt.Sprintf(`{"term":{"buyer":"%s"}}`, a.FormatParam.Buyer))
	}
	return fmt.Sprintf(`{"query":{"bool":{"must":[%s],"should":[%s],"minimum_should_match": %d}}%s}`, strings.Join(musts, ","), strings.Join(bools, ","), common.If(len(bools) > 0, 1, 0).(int), "%s")
}

// GetCommonQuerySqlWithAggs 此方法用于聚合查询
func (a *AnalysisEntity) GetCommonQuerySqlWithAggs() string {
	return fmt.Sprintf(a.GetCommonQuerySql(), `,"aggs":{%s},"size":0`)
}

// 市场概况+时间分布
func (a *AnalysisEntity) MarketTime() (map[string]interface{}, error) {
	var (
		sql                   []string
		monthB, yearB         bool
		MonthRange, YearRange string
		isOffline             = a.Offline == 1
	)
	sql = append(sql, fmt.Sprintf(aggsMarketAnalysis, "thismarket", fmt.Sprintf(`{"key":"%s","from":%d,"to":%d}`, "market", a.FormatParam.STime, a.FormatParam.ETime)))
	n_stime := a.FormatParam.STime
	var n_mae AnalysisEntity
	n_mae.FormatParam = a.FormatParam
	n_mae.BaseParam = a.BaseParam
	n_mae.MgoRecordId = a.MgoRecordId
	n_mae.FormatParam.ETime = a.FormatParam.ETime
	if time.Unix(a.FormatParam.STime, 0).AddDate(1, 0, 0).Unix() >= time.Unix(a.FormatParam.ETime, 0).Unix() {
		n_stime = getPreviousMarket(time.Unix(a.FormatParam.STime, 0), time.Unix(a.FormatParam.ETime, 0))
		sql = append(sql, fmt.Sprintf(aggsMarketAnalysis, "oldmarket", fmt.Sprintf(`{"key":"%s","from":%d,"to":%d}`, "market", n_stime, a.FormatParam.STime)))
	}

	if isOffline || time.Unix(a.FormatParam.STime, 0).AddDate(0, 1, 0).Unix() < time.Unix(a.FormatParam.ETime, 0).Unix() {
		var mon_time, year_time int64
		stime, etime := time.Unix(a.FormatParam.STime, 0), time.Unix(a.FormatParam.ETime, 0)
		monthB, MonthRange = GetMonthData(isOffline, stime, etime)
		sql = append(sql, fmt.Sprintf(projectTimeDistribution, "monthtime", MonthRange))
		if monthB {
			mon_time = stime.AddDate(0, -1, 0).Unix()
		} else {
			mon_time = a.FormatParam.STime
		}
		//年度数据
		yearB, YearRange = GetYearData(isOffline, stime, etime)
		sql = append(sql, fmt.Sprintf(projectTimeDistribution, "yeartime", YearRange))
		if yearB {
			year_time = stime.AddDate(-1, 0, 0).Unix()
		} else {
			year_time = a.FormatParam.STime
		}

		if n_stime > mon_time {
			n_stime = mon_time
		}
		if n_stime > year_time {
			n_stime = year_time
		}
	}
	//非离线
	if !isOffline {
		n_mae.FormatParam.STime = n_stime
	}
	finalSql := fmt.Sprintf(n_mae.GetCommonQuerySqlWithAggs(), strings.Join(sql, ","))
	log.Printf("final MarketScaleRefineQuery sql: %s", finalSql)
	rMap := make(map[string]interface{})
	rMapData := make(map[string]interface{})
	thisRow := marketTime{}
	res := GetAggs(PtIndex, PtType, finalSql)
	if res == nil || len(res) == 0 {
		return nil, fmt.Errorf("未查询到项目")
	}
	Rest(res, &thisRow)
	if thisRow.Thismarket.Buckets != nil && len(thisRow.Thismarket.Buckets) != 0 {
		Projectmarket := thisRow.Thismarket.Buckets[0]
		if !isOffline && Projectmarket.ProjectCount.DocCount == 0 {
			return nil, fmt.Errorf("未查询到项目数据")
		}
		// 实时分析数据量 如果关键词少于100，项目数据量可能大于MAProjectNumLimit（600000）
		//if Projectmarket.ProjectCount.DocCount > MAProjectNumLimit {
		//	return nil, fmt.Errorf("项目数量超出上限")
		//}
		rMapData["project_count"] = Projectmarket.ProjectCount.DocCount
		rMapData["projctamout"] = Projectmarket.ProjectAmount.Value
		rMapData["projectavgmoney"] = Projectmarket.ProjectAvgMoney.Value
		rMapData["buyercount"] = Projectmarket.BuyerCount.Value
		rMapData["winnercount"] = Projectmarket.WinnerCount.Value
		if thisRow.Oldmarket.Buckets != nil && len(thisRow.Oldmarket.Buckets) != 0 {
			oldProjectmarket := thisRow.Oldmarket.Buckets[0]
			//环比数据
			rMapData["projctamount_ratio"] = sequential(Projectmarket.ProjectAmount.Value, oldProjectmarket.ProjectAmount.Value)
			rMapData["project_count_ratio"] = sequential(float64(Projectmarket.ProjectCount.DocCount), float64(oldProjectmarket.ProjectCount.DocCount))
			rMapData["projectavgmoney_ratio"] = sequential(Projectmarket.ProjectAvgMoney.Value, oldProjectmarket.ProjectAvgMoney.Value)
			rMapData["buyercount_ratio"] = sequential(float64(Projectmarket.BuyerCount.Value), float64(oldProjectmarket.BuyerCount.Value))
			rMapData["winnercount_ratio"] = sequential(float64(Projectmarket.WinnerCount.Value), float64(oldProjectmarket.WinnerCount.Value))
		}
	}
	rMap["market_profile"] = rMapData

	if thisRow.Monthtime.Buckets != nil && len(thisRow.Monthtime.Buckets) != 0 {
		rMap["month_distribution"] = n_mae.TimeData(monthB, thisRow.Monthtime.Buckets)
	}
	if thisRow.Yeartime.Buckets != nil && len(thisRow.Yeartime.Buckets) != 0 {
		rMap["year_distribution"] = n_mae.TimeData(yearB, thisRow.Yeartime.Buckets)
	}
	return rMap, nil
}

// 时间分布月，年通用数据处理
func (a *AnalysisEntity) TimeData(_b bool, thisRow []Buckets) map[string]interface{} {
	var count_ss, amout_ss []map[string]interface{}

	for k, v := range thisRow {
		//环比多取一期数据
		if _b && k == 0 {
			continue
		}
		count := make(map[string]interface{})
		amount := make(map[string]interface{})
		count["minth"] = v.Key
		count["value"] = v.ScaleTotal.DocCount
		amount["minth"] = v.Key
		amount["value"] = v.ScaleAmount.Value
		//整月，年统计环比
		if _b {
			doccount := thisRow[k-1].ScaleTotal.DocCount
			amountvalue := thisRow[k-1].ScaleAmount.Value

			count["ratio"] = sequential(float64(v.ScaleTotal.DocCount), float64(doccount))
			amount["ratio"] = sequential(v.ScaleAmount.Value, amountvalue)
		}
		count_ss = append(count_ss, count)
		amout_ss = append(amout_ss, amount)
	}
	rMapData := make(map[string]interface{})
	rMapData["project_count"] = count_ss
	rMapData["project_amount"] = amout_ss
	return rMapData
}

// top10
func (a *AnalysisEntity) ProjectTop10() (rMap map[string]interface{}, err error) {
	finalSql := fmt.Sprintf(a.GetCommonQuerySql(), queryTop10)
	log.Println("ProjectTop10:", finalSql)
	hits := elastic.Get(PtIndex, PtType, finalSql)
	if hits == nil || len(*hits) == 0 {
		return
	}
	rMap = map[string]interface{}{}
	bArr := []map[string]interface{}{}
	source := *hits
	for _, v := range source {
		bA := map[string]interface{}{}
		if v["ids"] != nil && len(InterToSliceString(v["ids"])) > 0 {
			idss := InterToSliceString(v["ids"])
			bA["_id"] = encodeId(idss[len(idss)-1])
		} else {
			bA["_id"] = nil
		}
		if v["area"] == "" {
			bA["city"] = nil
		} else {
			bA["area"] = v["area"]
		}
		if v["city"] == "" {
			bA["city"] = nil
		} else {
			bA["city"] = v["city"]
		}
		bA["jgtime"] = v["jgtime"]
		bA["projectname"] = v["s_projectname"]
		bA["sortprice"] = v["sortprice"]
		ids := v["entidlist"]
		var id_s []string
		var winner_s []string
		for _, i := range InterToSliceString(ids) {
			iid := encodeId(i)
			id_s = append(id_s, iid)
		}
		bA["eidlist"] = id_s
		bA["winner_s"] = nil
		if common.ObjToString(v["s_winner"]) != "" {
			w_s := strings.Split(common.ObjToString(v["s_winner"]), ",")
			for _, vv := range w_s {
				winner_s = append(winner_s, vv)
			}
			bA["winner_s"] = winner_s
		}
		bArr = append(bArr, bA)
	}
	rMap["ProjectTop10"] = bArr
	return
}

func (a *AnalysisEntity) AllData() (rMap map[string]interface{}, err error) {
	rMap = map[string]interface{}{}
	var aggs []string
	area := a.FormatParam.Area
	city := a.FormatParam.City
	buyclass := a.FormatParam.BuyerClass
	aggs = append(aggs, aggsAllCM, fmt.Sprintf(queryAggsSortprice, sortpriceStr), aggsArea)
	if (len(area) + len(city)) != 1 {
		aaat3 := fmt.Sprintf(aggsAreaAmounttop3, Top30)
		aact3 := fmt.Sprintf(aggsAreaCounttop3, Top30)
		if a.Offline == ValueOffline {
			aaat3 = fmt.Sprintf(aggsAreaAmounttop3, Top30)
			aact3 = fmt.Sprintf(aggsAreaCounttop3, Top30)
		}
		aggs = append(aggs, aaat3, aact3)
	}
	if len(buyclass) != 1 {
		abat3 := fmt.Sprintf(aggsBuyerclassAmounttop3, Top30)
		abct3 := fmt.Sprintf(aggsBuyerclassCounttop3, Top30)
		if a.Offline == ValueOffline {
			abat3 = fmt.Sprintf(aggsBuyerclassAmounttop3, Top30)
			abct3 = fmt.Sprintf(aggsBuyerclassCounttop3, Top30)
		}
		//aggs = append(aggs, aggs_buyerclass, aggs_buyerclass_other, aggs_buyerclass_amounttop3, aggs_buyerclass_counttop3)
		aggs = append(aggs, aggsBuyerclass, abat3, abct3)
	}
	finalSql := fmt.Sprintf(a.GetCommonQuerySqlWithAggs(), strings.Join(aggs, ","))
	log.Println("allData sql:", finalSql)
	res := GetAggs(PtIndex, PtType, finalSql)
	if res == nil || len(res) == 0 {
		return
	}
	thisRow := AreaCTop{}
	for name, object := range res {
		bArr, err := object.MarshalJSON()
		if len(bArr) == 0 || err != nil {
			continue
		}
		if name == "project_amount" {
			if json.Unmarshal(bArr, &thisRow.Amount) != nil {
				continue
			}
		} else if name == "project_count" {
			if json.Unmarshal(bArr, &thisRow.ProjectCount) != nil {
				continue
			}
			//} else if name == "buyerclass_scale_other" {
			//	if json.Unmarshal(bArr, &thisRow.BuyerclassScaleOther) != nil {
			//		continue
			//	}
		} else if name == "project_count_not0" {
			if json.Unmarshal(bArr, &thisRow.CountNot0) != nil {
				continue
			}
		} else if name == "sortprice_ranges" {
			if json.Unmarshal(bArr, &thisRow.SortpriceRanges) != nil {
				continue
			}
		} else if name == "area_distribution" {
			if json.Unmarshal(bArr, &thisRow.AreaDistribution) != nil {
				continue
			}
		} else if name == "buyerclass_scale" {
			if json.Unmarshal(bArr, &thisRow.BuyerclassScale) != nil {
				continue
			}
		} else if name == "area_amount_top3" {
			if json.Unmarshal(bArr, &thisRow.AreaAmountTop3) != nil {
				continue
			}
		} else if name == "area_count_top3" {
			if json.Unmarshal(bArr, &thisRow.AreaCountTop3) != nil {
				continue
			}
		} else if name == "buyerclass_amount_top3" {
			if json.Unmarshal(bArr, &thisRow.BuyclassAmountTop3) != nil {
				continue
			}
		} else if name == "buyerclass_count_top3" {
			if json.Unmarshal(bArr, &thisRow.BuyclassCountTop3) != nil {
				continue
			}
		}
	}
	thisRow.Total = int64(thisRow.ProjectCount.DocCount)
	data := ProjectScale(thisRow)
	area_data := AreaDistribute(thisRow)
	customerData, customerOther := CustomerDistribute(thisRow)
	//customerOther, customerData := CustomerDistributeDetails(thisRow, customerDetails)
	var ids []string
	for _, v := range thisRow.AreaAmountTop3.Buckets {
		for _, va := range v.WinnerTop.Buckets {
			ids = append(ids, va.Winner)
		}
	}
	for _, v := range thisRow.BuyclassAmountTop3.Buckets {
		for _, va := range v.WinnerTop.Buckets {
			ids = append(ids, va.Winner)
		}
	}
	for _, v := range thisRow.AreaCountTop3.Buckets {
		for _, va := range v.WinnerTop.Buckets {
			ids = append(ids, va.Winner)
		}
	}
	for _, v := range thisRow.BuyclassCountTop3.Buckets {
		for _, va := range v.BidcountTop.Buckets {
			ids = append(ids, va.Winner)
		}
	}
	eid := IDToName(ids)
	rMap["projectScale"] = data
	rMap["area_infos"] = area_data
	rMap["customer_scale"] = customerData
	rMap["customer_scale_other"] = customerOther //客户分布（其它）
	rMap["scaleAreaAmountTop"] = AmountCompute(thisRow, "area", eid, a.Offline == ValueOffline)
	rMap["scaleBuyclassAmountTop"] = AmountCompute(thisRow, "buyclass", eid, a.Offline == ValueOffline)
	rMap["scaleAreaCountTop"] = CountCompute(thisRow, "area", eid, a.Offline == ValueOffline)
	rMap["scaleBuyclassCountTop"] = CountCompute(thisRow, "buyclass", eid, a.Offline == ValueOffline)
	if a.Offline == ValueOffline {
		rMap["count_not_0"] = thisRow.CountNot0.Count
	}
	return
}

// MarketScaleRefineQuery 细化聚合
func (a *AnalysisEntity) MarketScaleRefineQuery() (rMap map[string]interface{}, err error) {
	//关键词分组聚合
	var aggsGroup []string
	//获取总金额及总数
	aggsGroup = append(aggsGroup, `"projectAmount":{"sum":{"field":"sortprice"}}`)
	aggsGroup = append(aggsGroup, `"projectTotal":{"filter":{"match_all":{}}}`)
	itemDataMap := map[string]int64{}
	for _, group := range a.FormatParam.KeysItems {
		var bools []string
		for _, v := range getGroupKeywordArr(group.A_Key) {
			if sql := getKeyWordSql(v, a.BaseParam.MatchingMode); sql != "" {
				bools = append(bools, sql)
			}
		}
		if len(bools) > 0 {
			//临时调整 进行数据验证 Top3
			rwt := Top30
			rwa := Top30
			if a.Offline == ValueOffline { //离线
				rwt = Top30
				rwa = Top30
			}
			aggsGroup = append(aggsGroup, fmt.Sprintf(`"%s":{"filter":{"bool":{"should":[%s],"minimum_should_match": 1}},"aggs":{"project_count":{"filter":{"match_all":{}}},"project_amount":{"sum":{"field":"sortprice"}},"winner_total_top":{"terms":{"field":"entidlist","exclude":["-"],"order":[{"refine_winner_total":"desc"}],"size":%d},"aggs":{"refine_winner_total":{"filter":{"match_all":{}}}}},"winner_amount_top":{"terms":{"field":"entidlist","exclude":["-"],"order":[{"refine_winner_amount":"desc"}],"size":%d},"aggs":{"refine_winner_amount":{"sum":{"field":"sortprice"}}}}}}`, group.ItemName, strings.Join(bools, ","), rwt, rwa))
		}
		itemDataMap[group.ItemName] = group.UpdateTime
	}
	finalSql := fmt.Sprintf(a.GetCommonQuerySqlWithAggs(), strings.Join(aggsGroup, ","))
	fmt.Println("finalSql-----4", finalSql)
	rMap = map[string]interface{}{}
	res := GetAggs(PtIndex, PtType, finalSql)
	if res == nil || len(res) == 0 {
		return
	}
	scale := scaleRefineData{IdSwitch: map[string]string{}, IsOffline: a.Offline == ValueOffline}
	for name, object := range res {
		bArr, err := object.MarshalJSON()
		if len(bArr) == 0 || err != nil {
			continue
		}
		if name == "projectTotal" {
			st := simpleTotal{}
			if err := json.Unmarshal(bArr, &st); err == nil {
				scale.Total = st
			}
			continue
		} else if name == "projectAmount" {
			ss := simpleSum{}
			if err := json.Unmarshal(bArr, &ss); err == nil {
				scale.Amount = ss
			}
			continue
		}
		thisRow := scaleRefineRow{}
		if err := json.Unmarshal(bArr, &thisRow); err != nil {
			continue
		}
		thisRow.ItemName = name
		thisRow.UpdateTime = itemDataMap[name]
		scale.Data = append(scale.Data, &thisRow)
	}
	scale.formatData() //获取所需数据
	scale.doIdSwitch() //补充企业名称
	return map[string]interface{}{
		"scaleRefineAll":       scale.ReturnData.Overall,
		"scaleRefineTotalTop":  scale.ReturnData.TotalTop,
		"scaleRefineAmountTop": scale.ReturnData.AmountTop,
	}, nil
}

// 采购单位分析
func (a *AnalysisEntity) BuyerWinnerAnalysis() map[string]interface{} {
	var datas []string
	//采购单位-采购规模分布
	//buyer_scalefmt := fmt.Sprintf(buyer_procurement_scale, sortprice_str)

	//采购单位-项目数量 采购单位-项目金额
	bc := fmt.Sprintf(buyerCount, Top50, Top30)
	bs := fmt.Sprintf(buyerSortprice, Top50, Top30)
	if a.Offline == ValueOffline {
		bc = fmt.Sprintf(buyerCount, BWCount, Top30)
		bs = fmt.Sprintf(buyerSortprice, BWCount, Top30)
	}
	bps := fmt.Sprintf(buyerProcurementScale, BWDCount)
	datas = append(datas, bps, bc, bs)
	//中标单位-规模分布
	//winner_scalefmt := fmt.Sprintf(winner_procurement_scale, sortprice_str)

	//中标单位-项目数量 中标单位-项目金额
	wc := fmt.Sprintf(winnerCount, Top50, Top30) //正常需求30-3；目前是50-30，多获取数据 分析后进行数据损失补充，尽量减少 实时分析和离线分析的误差
	ws := fmt.Sprintf(winnerSortprice, Top50, Top30)
	if a.Offline == ValueOffline {
		wc = fmt.Sprintf(winnerCount, BWCount, Top30)
		ws = fmt.Sprintf(winnerSortprice, BWCount, Top30)
	}
	wps := fmt.Sprintf(winnerProcurementScale, BWDCount)
	datas = append(datas, wps, wc, ws)
	finalSql := fmt.Sprintf(a.GetCommonQuerySqlWithAggs(), strings.Join(datas, ","))
	log.Printf("final PurchasingAnalysiseQuery sql: %s", finalSql)
	t := time.Now()
	res := GetAggs(PtIndex, PtType, finalSql)
	if res == nil || len(res) == 0 {
		return nil
	}
	//计算entidlist大于1的中标单位金额
	entArrMap := make(map[string]float64)
	winnerKeyMap := make(map[string]int)
	newSql := strings.ReplaceAll(fmt.Sprintf(a.GetCommonQuerySql(), `,"sort": [{"sortprice": {"order": "desc"}}], "_source": ["entidlist","sortprice"],"size":10000`), `],"should":[`, fmt.Sprintf(`%s],"should":[`, `,{"script": {"script": {"source": "doc['entidlist'].length > 1"}}}`))
	log.Println("winner new sql", newSql)
	multiBid := elastic.Get(PtIndex, PtType, newSql)
	if multiBid != nil && len(*multiBid) > 0 {
		for _, m := range *multiBid {
			entidlist, _ := m["entidlist"].([]interface{})
			entArr := common.ObjArrToStringArr(entidlist)
			sortprice := common.Float64All(m["sortprice"])
			entLen := len(entArr)
			switch entLen {
			case 1:
				entArrMap[entArr[0]] = sortprice
				//winnerKeyMap[entArr[0]]++
			case 0:
			default:
				assessedAmount := sortprice / common.Float64All(entLen)
				for _, m2 := range entArr {
					//winnerKeyMap[m2]++
					entArrMap[m2] = assessedAmount
				}
			}
		}
	}
	log.Println("采购单位-中标单位分析报告es查询耗时===", time.Since(t))
	var thisBuyerWinnerRow BuyerWinnerRow
	for name, object := range res {
		bArr, err := object.MarshalJSON()
		if len(bArr) == 0 || err != nil {
			continue
		}
		if name == "buyer_amount_distribution" {
			if json.Unmarshal(bArr, &thisBuyerWinnerRow.BuyerAmountDistribution) != nil {
				continue
			}
		} else if name == "buyer_count_top3" {
			if json.Unmarshal(bArr, &thisBuyerWinnerRow.BuyerCountTop3) != nil {
				continue
			}
		} else if name == "buyer_amount_top3" {
			if json.Unmarshal(bArr, &thisBuyerWinnerRow.BuyerAmountTop3) != nil {
				continue
			}
		} else if name == "winner_amount_distribution" {
			if json.Unmarshal(bArr, &thisBuyerWinnerRow.WinnerAmountDistribution) != nil {
				continue
			}
		} else if name == "winner_count_top3" {
			if json.Unmarshal(bArr, &thisBuyerWinnerRow.WinnerCountTop3) != nil {
				continue
			}
		} else if name == "winner_amount_top3" {
			if json.Unmarshal(bArr, &thisBuyerWinnerRow.WinnerAmountTop3) != nil {
				continue
			}
		}
	}
	var winnerKeys []string
	for _, v := range thisBuyerWinnerRow.BuyerCountTop3.Buckets {
		for _, v1 := range v.SWinnerTop.Buckets {
			winnerKeyMap[v1.Key]++
		}
	}
	for _, v := range thisBuyerWinnerRow.BuyerAmountTop3.Buckets {
		for _, v1 := range v.SWinnerTop.Buckets {
			winnerKeyMap[v1.Key]++
		}
	}
	for _, v := range thisBuyerWinnerRow.WinnerCountTop3.SWinnerCount {
		winnerKeyMap[v.Key]++
	}
	for _, v := range thisBuyerWinnerRow.WinnerAmountTop3.SWinnerAmount {
		winnerKeyMap[v.Key]++
	}
	for k := range winnerKeyMap {
		winnerKeys = append(winnerKeys, k)
	}
	winnerName := GetEntNameByIds(winnerKeys)
	//rMap := make(map[string]interface{})
	var rMap = sync.Map{}
	sy := sync.WaitGroup{}
	sy.Add(2)
	go BuyerAnalysis(thisBuyerWinnerRow, &rMap, winnerName, &sy, a.Offline == ValueOffline)
	go WinningAnalysis(thisBuyerWinnerRow, &rMap, entArrMap, winnerName, &sy, a.Offline == ValueOffline)
	sy.Wait()
	log.Println("采购单位-中标单位分析报告程序计算耗时===", time.Since(t))
	rMaps := make(map[string]interface{})
	rMap.Range(func(key, value interface{}) bool {
		rMaps[common.InterfaceToStr(key)] = value
		return true
	})
	return rMaps
}

func (a *AnalysisEntity) GetPdfPageApi() (finalDate map[string]interface{}, err error) {
	var (
		wg   = sync.WaitGroup{}
		lock = &sync.Mutex{}
	)
	finalDate = make(map[string]interface{})
	for i := 1; i <= 5; i++ {
		wg.Add(1)
		go func(flag int) {
			defer wg.Done()
			mgoData, _ := a.GetMongoData(flag)
			lock.Lock()
			for s, i2 := range mgoData {
				finalDate[s] = i2
			}
			lock.Unlock()
		}(i)
	}
	wg.Wait()
	return
}