networkManage/api/internal/service/CoopHistoryService.go

package service

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"app.yhyue.com/moapp/jybase/common"
	"app.yhyue.com/moapp/jybase/encrypt"
	elastic "app.yhyue.com/moapp/jybase/es"
	"app.yhyue.com/moapp/jybase/redis"
	T "bp.jydev.jianyu360.cn/CRM/networkManage/api/common"
	"bp.jydev.jianyu360.cn/CRM/networkManage/api/internal/types"
	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
	"github.com/zeromicro/go-zero/core/logx"
)

var (
	INDEX_1   = "transaction_info_all"
	sql_2_0   = `SELECT buyer_id FROM information.transaction_info_all WHERE project_id = ?`
	sql_2_2   = `select id as buyer_id from ent_info where company_name=? order by id desc limit 1`
	sql_2_1   = `SELECT project_id, project_name, zbtime FROM information.transaction_info_all WHERE buyer_id = ? AND has(winner, ?) ORDER BY zbtime DESC`
	sql_2_1_1 = `SELECT project_id, project_name, zbtime FROM information.transaction_info_all WHERE buyer_id = ? AND agency = ? ORDER BY zbtime DESC`
	//sql_2_2    = `select DISTINCT  b.company_id,b.company_name,a.relate_id,a.relate_name,b.contact_person   from  connection_introduce a  INNER JOIN connection b on  a.position_id= ? and a.connection_id=b.id  and   a.relate_Id = ? and  a.itype =1  and  b.itype=4 and  b.status=1 `
	//sql_2_3    = `select a.a_id as a_id, a.b_id as b_id, a.a_name as a_name, a.b_name as b_name, a.code as code, c.legal_person as c_person, d.legal_person d_person from crm.ent_map_code a left join ent_info c on (a.a_id = ? or a.b_id = ? ) and a.code in('0101', '0201') and a.a_id = c.id left join ent_info d on (a.a_id = ? or a.b_id = ? ) and a.code in('0101', '0201') and a.b_id = d.id where (a.a_id = ? or a.b_id = ? ) and a.code in('0101', '0201')`
	es_query  = `{"query": {"bool": {"must": [{"terms": {"buyer_id": ["%s"]}},{"exists": {"field": "winner_id"}}],"must_not": {"terms": {"topscopeclass": ["%s"]}}}},"aggs": {"winner_count": {"terms": {"field": "winner","size": 1000,"order": {"_count": "desc"}},"aggs": {"amount_all": {"sum": {"field": "project_money"}},"ent_id": {"terms": {"field": "winner_id"}},"buyer_id": {"terms": {"field": "buyer_id"}}}}},"size": 1}`
	es_query1 = `{"query": {"bool": {"must": [{"terms": {"buyer_id": ["%s"]}}],"must_not": [{"term": {"agency_id": {"value": ""}}}]}},"aggs": {"agency_count": {"terms": {"field": "agency","size": 1000,"order": {"_count": "desc"}},"aggs": {"amount_all": {"sum": {"field": "project_money"}},"ent_id": {"terms": {"field": "agency_id"}},"buyer_id": {"terms": {"field": "buyer_id"}}}}},"size": 1}`
)

type ResultData struct {
	SourceType   string  `json:"SourceType"`
	EntName      string  `json:"EntName"`
	EntId        string  `json:"EntId"`
	EntPerson    string  `json:"EntPerson"`
	Middleman    string  `json:"Middleman"`
	ProjectNum   int     `json:"ProjectNum"`
	TotalAmount  float64 `json:"TotalAmount"`
	RecentTime   int64   `json:"RecentTime"`
	NearlyYears  bool    `json:"NearlyYears"`
	SupEnt       string  `json:"SupEnt"`       // 上级
	NextEnt      string  `json:"NextEnt"`      // 下级
	Relationship string  `json:"Relationship"` // 关系
	BuyerId      string  `json:"BuyerId"`
	IsIgnore     bool    `json:"isIgnore"`
}

type ResultDatas []*ResultData

func (n *ResultDatas) Len() int {
	return len(*n)
}

func (n *ResultDatas) Less(i, j int) bool {
	if (*n)[i].RecentTime == (*n)[j].RecentTime {
		return (*n)[i].EntName < (*n)[j].EntName
	}
	return (*n)[i].RecentTime > (*n)[j].RecentTime
}

func (n *ResultDatas) Swap(i, j int) {
	(*n)[i], (*n)[j] = (*n)[j], (*n)[i]
}

type ProjectTmp struct {
	BuyerId string `ch:"buyer_id"`
}

type Cooperate struct {
	Near   bool
	ZbTime int64
}

type AggStruct struct {
	Buckets []struct {
		Key        string `json:"key,omitempty"`
		Doc_count  int64  `json:"doc_count,omitempty"`
		Amount_all struct {
			Value float64 `json:"value,omitempty"`
		} `json:"amount_all"`
		EntId struct {
			Buckets []struct {
				Key       string `json:"key,omitempty"`
				Doc_count int64  `json:"doc_count,omitempty"`
			} `json:"buckets"`
		} `json:"ent_id"`
		BuyerId struct {
			Buckets []struct {
				Key       string `json:"key,omitempty"`
				Doc_count int64  `json:"doc_count,omitempty"`
			} `json:"buckets"`
		} `json:"buyer_id"`
	} `json:"buckets"`
}

func GetPrList(req *types.CoopHistoryReq) (result []*ResultData, size_1, size_2, size_3, size_4 int64) {
	pTmp := ProjectTmp{}
	var err error
	if req.Pid != "" {
		err = T.ClickhouseConn.QueryRow(context.TODO(), sql_2_0, req.Pid).ScanStruct(&pTmp)
	} else if req.Buyer != "" {
		err = T.ClickhouseConn.QueryRow(context.TODO(), sql_2_2, req.Buyer).ScanStruct(&pTmp)
	} else {
		return
	}
	if err != nil {
		return nil, 0, 0, 0, 0
	}
	//propertyForm := T.NetworkCom.GetMyProbusfor(req.EntAccountId)
	scopeClass := FindBusiness(req.EntId, req.EntUserId)
	// firstparty：甲方 supplier：供应商 adiffb：同甲异业 middleman：中间人 agency：招标代理机构 sup_sub: 上下级
	// 1、同甲异业数据/ 3、招标代理机构渠道
	if pTmp.BuyerId != "" {
		// 中间人可介绍业主
		var r3 []map[string]interface{}
		r3 = FindMiddleman([]string{fmt.Sprintf("'%s'", pTmp.BuyerId)}, req.PositionId, r3)
		if r3 != nil && len(r3) > 0 {
			size_4 = int64(len(r3))
			if req.ChannelType == "0" || req.ChannelType == "4" {
				for _, m := range r3 {
					tmp := ResultData{
						SourceType:   "middleman",
						EntName:      common.ObjToString(m["b_name"]),
						EntId:        common.ObjToString(m["b_id"]),
						EntPerson:    common.ObjToString(m["personName"]),
						Relationship: "业主的关系人",
						BuyerId:      pTmp.BuyerId,
					}
					result = append(result, &tmp)
				}
			}
		}
		// 关联单位
		var r4 []map[string]interface{}
		r4 = Findfirstparty([]string{fmt.Sprintf("'%s'", pTmp.BuyerId)}, r4)
		if r4 != nil && len(r4) > 0 {
			size_2 = int64(len(r4))
			if req.ChannelType == "0" || req.ChannelType == "2" {
				for _, m := range r4 {
					tmp := ResultData{
						SourceType:   "sup_sub",
						EntName:      common.ObjToString(m["b_name"]),
						EntId:        common.ObjToString(m["b_id"]),
						EntPerson:    common.ObjToString(m["personName"]),
						Relationship: common.ObjToString(m["relationship"]),
						BuyerId:      pTmp.BuyerId,
					}
					result = append(result, &tmp)
				}
			}
		}
		r1 := GetWinnerData(strings.Split(scopeClass, ","), pTmp.BuyerId)
		r2 := GetAgencyData(pTmp.BuyerId)
		winners, agency := []string{}, []string{}
		for _, m := range r1 {
			winners = append(winners, common.ObjToString(m["name"]))
		}
		for _, m := range r2 {
			agency = append(agency, common.ObjToString(m["name"]))
		}
		logx.Info("CandidateChannel LastTimeCoopBath start ", req.PositionId, " ", pTmp.BuyerId, " ", len(winners), " ", len(agency))
		ltcs1, ltcs2 := LastTimeCoopBath(req.PositionId, pTmp.BuyerId, winners, agency)
		logx.Info("CandidateChannel LastTimeCoopBath over ", req.PositionId)
		if r1 != nil && len(r1) > 0 {
			size_1 = int64(len(r1))
			if req.ChannelType == "0" || req.ChannelType == "1" {
				for _, m := range r1 {
					tmp := ResultData{
						SourceType:  "adiffb",
						EntName:     common.ObjToString(m["name"]),
						ProjectNum:  common.IntAll(m["coop_size"]),
						TotalAmount: common.Float64All(m["coop_amount"]),
						BuyerId:     pTmp.BuyerId,
					}
					if ltc := ltcs1[common.ObjToString(m["name"])]; ltc != nil {
						tmp.NearlyYears = ltc.Near
						tmp.RecentTime = ltc.ZbTime
					}
					result = append(result, &tmp)
				}
			}
		}
		if r2 != nil && len(r2) > 0 {
			size_3 = int64(len(r2))
			if req.ChannelType == "0" || req.ChannelType == "3" {
				for _, m := range r2 {
					tmp := ResultData{
						SourceType:  "agency",
						EntName:     common.ObjToString(m["name"]),
						ProjectNum:  common.IntAll(m["coop_size"]),
						TotalAmount: common.Float64All(m["coop_amount"]),
						BuyerId:     pTmp.BuyerId,
					}
					if ltc := ltcs2[common.ObjToString(m["name"])]; ltc != nil {
						tmp.NearlyYears = ltc.Near
						tmp.RecentTime = ltc.ZbTime
					}
					result = append(result, &tmp)
				}
			}
		}
	}
	return
}
func GetWinnerData(scopeClass []string, bid string) (result1 []map[string]interface{}) {
	result1 = []map[string]interface{}{}
	winnerKey := fmt.Sprintf("networkManage_coopHistory_winner_%s", common.GetMd5String(strings.Join(scopeClass, ",")+"_"+bid))
	wb, err := redis.GetBytes("newother", winnerKey)
	if err == nil {
		json.Unmarshal(*wb, &result1)
		return
	}
	m1 := make(map[string]interface{}) //采购单位-中标单位
	noIds := []string{}
	if len(scopeClass) > 0 {
		//同甲异态 中标企业
		aggs, _, _ := elastic.GetAggs(INDEX_1, INDEX_1, fmt.Sprintf(es_query, strings.ReplaceAll(bid, ",", `","`), strings.Join(scopeClass, "\",\"")))
		logx.Info("es聚合查询结果:", fmt.Sprintf(es_query, strings.ReplaceAll(bid, ",", `","`), strings.Join(scopeClass, "\",\"")))
		var m1Buckets = AggStruct{}
		if aggs != nil && aggs["winner_count"] != nil {
			bs, err := aggs["winner_count"].MarshalJSON()
			if err != nil {
				logx.Info(err)
			} else {
				if len(bs) == 0 {
					logx.Info(err)
				} else {
					err := json.Unmarshal(bs, &m1Buckets)
					logx.Info(err)
					if len(m1Buckets.Buckets) > 0 {
						for _, v := range m1Buckets.Buckets {
							if v.Key != "" && len(v.EntId.Buckets) > 0 && len(v.BuyerId.Buckets) > 0 {
								mv := map[string]interface{}{"name": v.Key, "amount": v.Amount_all.Value, "count": v.Doc_count, "buyerId": v.BuyerId.Buckets[0].Key}
								if len(v.EntId.Buckets) == 1 && v.EntId.Buckets[0].Key != "" {
									mv["nameId"] = v.EntId.Buckets[0].Key
								} else {
									noIds = append(noIds, v.Key)
								}
								m1[v.Key] = mv
							}
						}
					}
				}
			}
		}
	}
	idName := T.NetworkCom.GetEntIdByName(noIds)
	for k, v := range m1 {
		v1 := v.(map[string]interface{})
		tmp := make(map[string]interface{})
		tmp["name"] = k
		nameId, _ := v1["nameId"].(string)
		if nameId == "" {
			nameId = idName[k]
		}
		tmp["nameId"] = nameId
		tmp["coop_size"] = v1["count"]
		tmp["coop_amount"] = v1["amount"]
		result1 = append(result1, tmp)
	}
	redis.Put("newother", winnerKey, result1, T.C.CacheTimeOut)
	return
}

func GetAgencyData(bid string) (result2 []map[string]interface{}) {
	result2 = []map[string]interface{}{}
	agencyKey := fmt.Sprintf("networkManage_coopHistory_agency_%s", bid)
	wb, err := redis.GetBytes("newother", agencyKey)
	if err == nil {
		json.Unmarshal(*wb, &result2)
		return
	}
	m2 := make(map[string]interface{}) //采购单位-代理机构
	noIds := []string{}
	//代理机构
	aggs1, _, _ := elastic.GetAggs(INDEX_1, INDEX_1, fmt.Sprintf(es_query1, strings.ReplaceAll(bid, ",", `","`)))
	var m2Buckets = AggStruct{}
	if aggs1 != nil && aggs1["agency_count"] != nil {
		bs, err := aggs1["agency_count"].MarshalJSON()
		if err != nil {
			logx.Info(err)
		} else {
			if len(bs) == 0 {
				logx.Info(err)
			} else {
				err = json.Unmarshal(bs, &m2Buckets)
				logx.Info(err)
				if len(m2Buckets.Buckets) > 0 {
					for _, v := range m2Buckets.Buckets {
						if v.Key != "" && len(v.EntId.Buckets) > 0 && len(v.BuyerId.Buckets) > 0 {
							if v.EntId.Buckets[0].Key == "" {
								noIds = append(noIds, v.Key)
							}
							m2[v.Key] = map[string]interface{}{"name": v.Key, "amount": v.Amount_all.Value, "count": v.Doc_count, "nameId": v.EntId.Buckets[0].Key, "buyerId": v.BuyerId.Buckets[0].Key}
						}
					}
				}
			}
		}
	}
	idName := T.NetworkCom.GetEntIdByName(noIds)
	for k, v := range m2 {
		v1 := v.(map[string]interface{})
		tmp := make(map[string]interface{})
		tmp["name"] = k
		nameId, _ := v1["nameId"].(string)
		if nameId == "" {
			nameId = idName[k]
		}
		tmp["nameId"] = nameId
		tmp["coop_size"] = v1["count"]
		tmp["coop_amount"] = v1["amount"]
		result2 = append(result2, tmp)
	}
	redis.Put("newother", agencyKey, result2, T.C.CacheTimeOut)
	return
}

type P_History struct {
	ProjectId   string `ch:"project_id"`
	ProjectName string `ch:"project_name"`
	ZbTime      int64  `ch:"zbtime"`
	Href        string `ch:"href"`
}

// @Author jianghan
// @Description 合作历史
// @Date 2024/4/20
func GetData3(source string, buyerId, winnerId string) (result []*P_History) {
	var rows driver.Rows
	if source == "1" {
		rows, _ = T.ClickhouseConn.Query(context.TODO(), sql_2_1_1, buyerId, winnerId)
	} else if source == "2" {
		rows, _ = T.ClickhouseConn.Query(context.TODO(), sql_2_1, buyerId, winnerId)
	}
	for rows.Next() {
		pHis := P_History{}
		_ = rows.ScanStruct(&pHis)
		pHis.Href = fmt.Sprintf("/article/content/%s.html", encrypt.CommonEncodeArticle("content", pHis.ProjectId))
		result = append(result, &pHis)
	}
	return
}

// @Author jianghan
// @Description 上次合作时间
// @Date 2024/4/24
func LastTimeCoop(buyerId, ent, stype string) (bool, int64) {
	zbtime := int64(0)
	near := false
	if stype == "adiffb" {
		sql := `SELECT zbtime FROM information.transaction_info_all WHERE buyer_id = ? AND has(winner, ?) ORDER BY zbtime DESC LIMIT 1`
		_ = T.ClickhouseConn.QueryRow(context.TODO(), sql, buyerId, ent).Scan(&zbtime)
	} else if stype == "agency" {
		sql := `SELECT zbtime FROM information.transaction_info_all WHERE buyer_id = ? AND agency = ? ORDER BY zbtime DESC LIMIT 1`
		_ = T.ClickhouseConn.QueryRow(context.TODO(), sql, buyerId, ent).Scan(&zbtime)
	}
	timestamp := time.Now().AddDate(-3, 0, 0).Unix()
	if timestamp <= zbtime {
		near = true
	}
	return near, zbtime
}

func LastTimeCoopBath(positionId int64, buyerId string, winners, agencys []string) (map[string]*Cooperate, map[string]*Cooperate) {
	adiffb, agency := map[string]*Cooperate{}, map[string]*Cooperate{}
	if buyerId == "" {
		return adiffb, agency
	}
	var toSearch = func(tp int, query string, ws []string) {
		logx.Info("LastTimeCoopBath once start ", positionId)
		defer logx.Info("LastTimeCoopBath once over ", positionId)
		wh, newArgs := T.NetworkCom.WhArgs(ws)
		query = fmt.Sprintf(query, wh)
		args := []interface{}{buyerId}
		args = append(args, newArgs...)
		rows, err := T.ClickhouseConn.Query(context.Background(), query, args...)
		if err != nil {
			logx.Error(err)
			return
		}
		for rows.Next() {
			var (
				one    string
				zbtime int64
			)
			if err := rows.Scan(&one, &zbtime); err != nil {
				logx.Error(err)
				continue
			}
			c := &Cooperate{
				ZbTime: zbtime,
			}
			timestamp := time.Now().AddDate(-3, 0, 0).Unix()
			if timestamp <= c.ZbTime {
				c.Near = true
			}
			if tp == 1 {
				adiffb[one] = c
			} else {
				agency[one] = c
			}
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			logx.Error(err)
		}
	}
	if len(winners) > 0 {
		key := fmt.Sprintf("networkManage_lastTimeCoop_winner_%s", common.GetMd5String(buyerId+"_"+strings.Join(winners, ",")))
		wb, err := redis.GetBytes("newother", key)
		if err == nil {
			json.Unmarshal(*wb, &adiffb)
		} else {
			sql := `SELECT winner_one,max(zbtime) FROM information.transaction_info_all ARRAY JOIN winner AS winner_one WHERE buyer_id=? AND hasAny(winner,[%s])) group by winner_one`
			wns := []string{}
			for _, v := range winners {
				wns = append(wns, v)
				if len(wns) == 50 {
					toSearch(1, sql, wns)
					wns = []string{}
				}
			}
			if len(wns) > 0 {
				toSearch(1, sql, wns)
			}
			redis.Put("newother", key, adiffb, T.C.CacheTimeOut)
		}
	}
	if len(agencys) > 0 {
		key := fmt.Sprintf("networkManage_lastTimeCoop_agency_%s", common.GetMd5String(buyerId+"_"+strings.Join(agencys, ",")))
		wb, err := redis.GetBytes("newother", key)
		if err == nil {
			json.Unmarshal(*wb, &agency)
		} else {
			sql := `SELECT agency,max(zbtime) FROM information.transaction_info_all WHERE buyer_id=? AND agency in (%s) group by agency`
			acs := []string{}
			for _, v := range agencys {
				acs = append(acs, v)
				if len(acs) == 50 {
					toSearch(2, sql, acs)
					acs = []string{}
				}
			}
			if len(acs) > 0 {
				toSearch(2, sql, acs)
			}
			redis.Put("newother", key, agency, T.C.CacheTimeOut)
		}
	}
	return adiffb, agency
}