application/api/internal/service/network.go

package service

import (
	"context"
	"database/sql"
	"fmt"
	"math"
	"sort"
	"strings"
	"sync"

	. "app.yhyue.com/moapp/jybase/common"
	. "app.yhyue.com/moapp/jybase/date"
	. "app.yhyue.com/moapp/jybase/es"
	. "app.yhyue.com/moapp/jybase/sort"
	. "bp.jydev.jianyu360.cn/CRM/application/api/common"
	"bp.jydev.jianyu360.cn/CRM/application/api/internal/types"
	"github.com/shopspring/decimal"
	"github.com/zeromicro/go-zero/core/logx"
)

var Network = &network{}

type network struct {
}
type networkTree struct {
	Count    int64               `json:"count"`
	Name     string              `json:"name"`
	Children []*networkTreeChild `json:"children"`
}
type networkTreeChild struct {
	Count int64  `json:"count"`
	Name  string `json:"name"`
	Id    string `json:"id"`
	Type  int    `json:"type"`
}
type projectInfo struct {
	BuyerCount    int
	ProjectCount  int
	ProjectAmount float64
}
type firstpartyNetwork struct {
	CompanyId   string
	CompanyName string
	Name        string
}

//人脉库-添加/修改人脉
func (n *network) AddOrUpdate(in *types.AddOrUpdateReq) *types.Reply {
	reply := &types.Reply{}
	nowFormat := NowFormat(Date_Full_Layout)
	var saveIntroduce = func(tx *sql.Tx, cid int64, isUpdate bool) bool {
		if in.Type != "middleman" {
			return true
		}
		values := []interface{}{}
		if in.Introduce_owner_id != "" {
			for k, v := range strings.Split(in.Introduce_owner_id, ",") {
				values = append(values, in.PositionId, in.EntId, in.EntDeptId, in.EntUserId, cid, v, strings.Split(in.Introduce_owner_name, ",")[k], 1, nowFormat)
			}
		}
		if in.Introduce_project_id != "" {
			for k, v := range strings.Split(in.Introduce_project_id, ",") {
				values = append(values, in.PositionId, in.EntId, in.EntDeptId, in.EntUserId, cid, v, strings.Split(in.Introduce_project_name, ",")[k], 2, nowFormat)
			}
		}
		if len(values) == 0 {
			return false
		}
		var r2 int64
		if isUpdate {
			r2 = CrmMysql.UpdateOrDeleteBySqlByTx(tx, `delete from crm.connection_introduce where connection_id=? and position_id=?`, in.Id, in.PositionId)
		}
		r3, _ := CrmMysql.InsertBatchByTx(tx, "crm.connection_introduce", []string{"position_id", "ent_id", "ent_dept_id", "ent_user_id", "connection_id", "relate_id", "relate_name", "itype", "create_time"}, values)
		return r2 >= 0 && r3 > 0
	}
	if in.Id > 0 {
		if CrmMysql.ExecTx("更新人脉", func(tx *sql.Tx) bool {
			r1 := CrmMysql.UpdateOrDeleteBySqlByTx(tx, `update crm.connection set company_name=?,company_id=?,contact_person=?,contact_phone=?,update_time=? where id=? and position_id=?`, in.Company_name, in.Company_id, in.Contact_person, in.Contact_phone, nowFormat, in.Id, in.PositionId)
			return r1 >= 0 && saveIntroduce(tx, in.Id, true)
		}) {
			reply.Data = map[string]interface{}{
				"status": 1,
			}
		} else {
			reply.Data = map[string]interface{}{
				"status": 0,
			}
		}
	} else {
		itype := n.TypeConvert(in.Type)
		var r1 int64
		if CrmMysql.ExecTx("新增人脉", func(tx *sql.Tx) bool {
			_, r1 = CrmMysql.InsertBatchByTx(tx, "crm.connection", []string{"position_id", "ent_id", "ent_dept_id", "ent_user_id", "itype", "company_name", "company_id", "contact_person", "contact_phone", "status", "create_time", "update_time"}, []interface{}{in.PositionId, in.EntId, in.EntDeptId, in.EntUserId, itype, in.Company_name, in.Company_id, in.Contact_person, in.Contact_phone, 1, nowFormat, nowFormat})
			return r1 > 0 && saveIntroduce(tx, r1, false)
		}) {
			reply.Data = map[string]interface{}{
				"status": 1,
				"id":     r1,
			}
		} else {
			reply.Data = map[string]interface{}{
				"status": 0,
			}
		}
	}
	return reply
}

//人脉库-业主名称联想
func (n *network) Associate(in *types.AssociateReq) (reply *types.Reply) {
	//类型；firstparty：甲方 supplier：供应商 adiffb：同甲异业 middleman：中间人 middleman_owner：中间人-业主 middleman_project：中间人-项目 agency：招标代理机构
	res := []map[string]interface{}{}
	reply = &types.Reply{Data: res}
	in.Name = strings.TrimSpace(in.Name)
	if in.Name == "" {
		return
	}
	pageSize := 10
	if in.Type == "adiffb" {
		probusfors := NetworkCom.GetMyProbusfor(in.EntAccountId)
		if len(probusfors) > 0 {
			args := []interface{}{in.EntName}
			wh, newArgs := NetworkCom.WhArgs(probusfors)
			args = append(args, newArgs...)
			q := `select DISTINCT b.winner_id,b.winner from information.transaction_info a 
				inner join information.transaction_info b on (has(a.winner, ?) and a.buyer_id<>'' and a.buyer_id=b.buyer_id and hasAny(b.property_form,[` + wh + `])) ORDER BY b.project_id`
			rows, err := ClickhouseConn.Query(context.Background(), q, args...)
			if err != nil {
				logx.Error(err)
			} else {
				repeat := map[string]bool{}
				for rows.Next() {
					var (
						winner_id []string
						winner    []string
					)
					if err := rows.Scan(&winner_id, &winner); err != nil {
						logx.Error(err)
						continue
					}
					for k, v := range winner {
						if repeat[v] || !strings.Contains(v, in.Name) {
							continue
						}
						repeat[v] = true
						if k >= len(winner_id) {
							continue
						}
						res = append(res, map[string]interface{}{
							"company_name": v,
							"company_id":   winner_id[k],
						})
					}
				}
				rows.Close()
				if err := rows.Err(); err != nil {
					logx.Error(err)
				}
				if len(res) > pageSize {
					res = res[:pageSize]
				}
			}
		}
	} else {
		must := []string{fmt.Sprintf(`{"multi_match":{"query":"%s","type":"phrase","fields":["company_name"]}}`, in.Name)}
		switch in.Type {
		case "firstparty":
			must = append(must, fmt.Sprintf(`{"terms":{"company_label":["%s"]}}`, strings.Join(NetworkCom.GetEntTagSeat(2), `","`)))
			must = append(must, `{"terms":{"company_unit_type":[1,2]}}`)
		case "supplier":
			must = append(must, fmt.Sprintf(`{"terms":{"company_label":["%s"]}}`, strings.Join(NetworkCom.GetEntTagSeat(2), `","`)))
			must = append(must, `{"term":{"company_unit_type":3}}`)
		case "middleman_owner":
			must = append(must, `{"terms":{"company_unit_type":[1,2]}}`)
		case "agency":
			must = append(must, `{"term":{"company_unit_type":4}}`)
		}
		q := fmt.Sprintf(`{"query":{"bool":{"must":[%s]}},"size":%d,"_source":["id","company_name"]}`, strings.Join(must, ","), pageSize)
		logx.Info("人脉库-业主名称联想", q)
		datas := VarEs.Get("ent_info", "ent_info", q)
		if datas != nil {
			for _, v := range *datas {
				res = append(res, map[string]interface{}{
					"company_name": ObjToString(v["company_name"]),
					"company_id":   ObjToString(v["id"]),
				})
			}
		}
	}
	reply.Data = res
	return
}

//人脉库-全部人脉项目
func (n *network) AllProject(in *types.AllprojectReq) (reply *types.Reply) {
	pool := make(chan bool, 5)
	wait := &sync.WaitGroup{}
	lock := &sync.Mutex{}
	reply = &types.Reply{}
	wh, newArgs := NetworkCom.WhArgs(NetworkCom.GetMyProbusfor(in.EntAccountId))
	var count int64
	var list []*networkTree
	if in.Id != "" {
		if in.Type == 1 {
			result := n.FirstpartyNetwork(in.Name, []string{in.Id})
			if result[in.Id] != nil {
				nameIndex := map[string]int{}
				for _, v := range result[in.Id] {
					if _, ok := nameIndex[v.Name]; !ok {
						nameIndex[v.Name] = len(list)
						list = append(list, &networkTree{
							Name: v.Name,
						})
					}
					pool <- true
					wait.Add(1)
					go func(cIndex int, cId, cName string) {
						defer func() {
							<-pool
							wait.Done()
						}()
						ntc := &networkTreeChild{
							Name: cName,
							Id:   cId,
							Type: 1,
						}
						if wh != "" {
							thisArgs := []interface{}{ntc.Id}
							thisArgs = append(thisArgs, newArgs...)
							ntc.Count = NetworkCom.Count(`select count(1) from information.transaction_info where buyer_id=? and hasAny(property_form,[`+wh+`])`, thisArgs...)
						}
						lock.Lock()
						count++
						list[cIndex].Count += ntc.Count
						list[cIndex].Children = append(list[cIndex].Children, ntc)
						lock.Unlock()
					}(nameIndex[v.Name], v.CompanyId, v.CompanyName)
				}
				wait.Wait()
			}
		}
	} else {
		q := `SELECT a.company_id,a.company_name,a.itype,COUNT(b.id) AS ipc FROM crm.connection a 
			LEFT JOIN crm.connection_introduce b ON (b.position_id=? AND b.itype=2 AND a.id=b.connection_id) WHERE a.position_id=?`
		args := []interface{}{in.PositionId, in.PositionId}
		if in.Name != "" {
			q += ` and company_name like ?`
			args = append(args, "%"+in.Name+"%")
		}
		q += ` GROUP BY a.id ORDER BY a.create_time DESC`
		datas := CrmMysql.SelectBySql(q, args...)
		list = []*networkTree{
			&networkTree{
				Name:     "甲方",
				Children: []*networkTreeChild{},
			},
			&networkTree{
				Name:     "供应商",
				Children: []*networkTreeChild{},
			},
			&networkTree{
				Name:     "同甲异业渠道",
				Children: []*networkTreeChild{},
			},
			&networkTree{
				Name:     "中间人",
				Children: []*networkTreeChild{},
			},
			&networkTree{
				Name:     "招标代理",
				Children: []*networkTreeChild{},
			},
		}
		//
		for _, vt := range *datas {
			pool <- true
			wait.Add(1)
			go func(v map[string]interface{}) {
				defer func() {
					<-pool
					wait.Done()
				}()
				itype := IntAll(v["itype"])
				if itype <= 0 || itype > len(list) {
					return
				}
				ntc := &networkTreeChild{
					Name: ObjToString(v["company_name"]),
					Id:   ObjToString(v["company_id"]),
					Type: IntAll(v["itype"]),
				}
				if wh != "" {
					thisArgs := []interface{}{ntc.Id}
					thisArgs = append(thisArgs, newArgs...)
					if itype == 1 {
						ntc.Count = NetworkCom.Count(`select count(1) from information.transaction_info where buyer_id=? and hasAny(property_form,[`+wh+`])`, thisArgs...)
					} else if itype == 2 || itype == 3 {
						ntc.Count = NetworkCom.Count(`select count(1) from information.transaction_info where has(winner_id,?) and hasAny(property_form,[`+wh+`])`, thisArgs...)
					} else if itype == 4 {
						ntc.Count = Int64All(v["ipc"])
					} else if itype == 5 {
						ntc.Count = NetworkCom.Count(`select count(1) from information.transaction_info where agency_id=? and hasAny(property_form,[`+wh+`])`, thisArgs...)
					}
				}
				lock.Lock()
				count += ntc.Count
				list[itype-1].Count += ntc.Count
				list[itype-1].Children = append(list[itype-1].Children, ntc)
				lock.Unlock()
			}(vt)
		}
		wait.Wait()
	}
	reply = &types.Reply{
		Data: map[string]interface{}{
			"count": count,
			"list":  list,
		},
	}
	return reply
}

//人脉库-列表
func (n *network) List(in *types.NetWorkListReq) *types.Reply {
	q := `select a.company_id,a.company_name,a.itype,a.contact_person as person,a.contact_phone as phone,count(if(b.itype=1,1,null)) as buyer_count,count(if(b.itype=2,1,null)) as project_count,GROUP_CONCAT(b.relate_id) as relate_id,a.create_time from crm.connection a 
		left join crm.connection_introduce b on (a.id=b.connection_id) where a.position_id=?`
	args := []interface{}{in.PositionId}
	if in.Type != "" {
		q += ` and a.itype=?`
		args = append(args, n.TypeConvert(in.Type))
	}
	if in.Starttime != "" {
		q += ` and a.create_time>=?`
		args = append(args, in.Starttime)
	}
	if in.Endtime != "" {
		q += ` and a.create_time<=?`
		args = append(args, in.Endtime)
	}
	if in.Name != "" {
		q += ` and a.company_name like ?`
		args = append(args, "%"+in.Name+"%")
	}
	q += ` GROUP BY a.id order by a.create_time desc`
	listTemp := CrmMysql.SelectBySql(q, args...)
	firstparty_array, supplier_array, adiffb_array, agency_array, middleman_array := []string{}, []string{}, []string{}, []string{}, []string{}
	for _, v := range *listTemp {
		switch Int64All(v["itype"]) {
		case 1:
			firstparty_array = append(firstparty_array, ObjToString(v["company_id"]))
		case 2:
			supplier_array = append(supplier_array, ObjToString(v["company_id"]))
		case 3:
			adiffb_array = append(adiffb_array, ObjToString(v["company_id"]))
		case 4:
			if relate_id := ObjToString(v["relate_id"]); relate_id != "" {
				middleman_array = append(middleman_array, strings.Split(relate_id, ",")...)
			}
		case 5:
			agency_array = append(agency_array, ObjToString(v["company_id"]))
		}
	}
	//
	firstparty_count, supplier_count, adiffb_count, middleman_count, agency_count := 0, 0, 0, 0, 0
	list := []*map[string]interface{}{}
	firstparty_project := n.Introduce_Firstparty(n.FirstpartyNetwork("", firstparty_array))
	supplier_project := n.Introduce_Supplier(supplier_array)
	adiffb_project := n.Introduce_Supplier(adiffb_array)
	agency_project := n.Introduce_Agency(agency_array)
	middleman_project := n.Introduce_Middleman(middleman_array)
	for _, v := range *listTemp {
		itype := ""
		buyer_count, project_count, expect_amount := 0, 0, float64(0)
		company_id := ObjToString(v["company_id"])
		switch Int64All(v["itype"]) {
		case 1:
			itype = "甲方"
			firstparty_count++
			if firstparty_project[company_id] != nil {
				buyer_count = firstparty_project[company_id].BuyerCount
				project_count = firstparty_project[company_id].ProjectCount
				expect_amount = firstparty_project[company_id].ProjectAmount
			}
		case 2:
			itype = "供应商"
			supplier_count++
			if supplier_project[company_id] != nil {
				buyer_count = supplier_project[company_id].BuyerCount
				project_count = supplier_project[company_id].ProjectCount
				expect_amount = supplier_project[company_id].ProjectAmount
			}
		case 3:
			itype = "同甲异业渠道"
			adiffb_count++
			if adiffb_project[company_id] != nil {
				buyer_count = adiffb_project[company_id].BuyerCount
				project_count = adiffb_project[company_id].ProjectCount
				expect_amount = adiffb_project[company_id].ProjectAmount
			}
		case 4:
			itype = "中间人"
			middleman_count++
			buyer_count = IntAll(v["buyer_count"])
			project_count = IntAll(v["project_count"])
			if relate_id := ObjToString(v["relate_id"]); relate_id != "" {
				for _, v := range strings.Split(relate_id, ",") {
					if middleman_project[v] != nil {
						expect_amount += middleman_project[v].ProjectAmount
					}
				}
			}
			if middleman_project[company_id] != nil {
				expect_amount = middleman_project[company_id].ProjectAmount
			}
		case 5:
			itype = "招标代理机构"
			agency_count++
			if agency_project[company_id] != nil {
				buyer_count = agency_project[company_id].BuyerCount
				project_count = agency_project[company_id].ProjectCount
				expect_amount = agency_project[company_id].ProjectAmount
			}
		}
		list = append(list, &map[string]interface{}{
			"company_id":    company_id,
			"company_name":  v["company_name"],
			"type":          itype,
			"person":        v["person"],
			"phone":         v["phone"],
			"buyer_count":   buyer_count,
			"monitor_count": 0,
			"expect_amount": RetainDecimal(expect_amount/10000, 2),
			"project_count": project_count,
			"create_time":   v["create_time"],
		})
	}
	csList := &ComSortList{
		SortKeys: []*ComSortKey{
			&ComSortKey{
				Keys:  []string{"expect_amount"},
				Order: -1,
				Type:  "float",
			},
		},
		List: list,
	}
	if in.Order_amount == -1 {
		csList.SortKeys[0].Order = -1
	}
	if in.Order_amount != 0 {
		sort.Sort(csList)
	}
	length := int64(len(csList.List))
	var pageSize int64 = 100
	total_page := int64(math.Ceil(float64(length) / float64(pageSize)))
	finalList := []*map[string]interface{}{}
	if length > 0 {
		if in.Current_page <= 0 {
			in.Current_page = 1
		}
		if in.Current_page > total_page {
			in.Current_page = total_page
		}
		start := (in.Current_page - 1) * pageSize
		end := start + pageSize
		if end > length {
			end = length
		}
		finalList = csList.List[start:end]
	}
	return &types.Reply{
		Data: map[string]interface{}{
			"total_page":       total_page,
			"firstparty_count": firstparty_count,
			"supplier_count":   supplier_count,
			"adiffb_count":     adiffb_count,
			"middleman_count":  middleman_count,
			"agency_count":     agency_count,
			"list":             finalList,
		},
	}
}

//
func (n *network) FirstpartyNetwork(name string, values []string) map[string][]*firstpartyNetwork {
	result := map[string][]*firstpartyNetwork{}
	wh, args := NetworkCom.WhArgs(values)
	q := `select DISTINCT a.a_id,a.b_id as company_id,a.b_name as company_name,b.name from information.ent_map_code a 
				inner join information.ent_code b on (a.a_id in (` + wh + `) and b.pcode in ('0100','0200') and a.code=b.code)`
	if name != "" {
		q += ` where c.company_name like ?`
		args = append(args, "%"+name+"%")
	}
	q += ` order by a.a_id,b.name,a.b_name`
	rows, err := ClickhouseConn.Query(context.Background(), q, args...)
	if err != nil {
		logx.Error(err)
		return result
	}
	for rows.Next() {
		var (
			a_id         string
			company_id   string
			company_name string
			name         string
		)
		if err := rows.Scan(&a_id, &company_id, &company_name, &name); err != nil {
			logx.Error(err)
			continue
		}
		result[a_id] = append(result[a_id], &firstpartyNetwork{
			CompanyId:   company_id,
			CompanyName: company_name,
			Name:        name,
		})
	}
	return result
}

//
func (n *network) Introduce_Firstparty(fpn map[string][]*firstpartyNetwork) map[string]*projectInfo {
	values := []string{}
	vm := map[string]*projectInfo{}
	for _, v := range fpn {
		for _, vv := range v {
			vm[vv.CompanyId] = &projectInfo{}
			values = append(values, vv.CompanyId)
		}
	}
	wh, args := NetworkCom.WhArgs(values)
	rows, err := ClickhouseConn.Query(context.Background(), `select buyer_id,count(project_id) AS project_count,sum(project_money) AS project_amount from information.transaction_info where buyer_id in (`+wh+`) group by buyer_id`, args...)
	if err != nil {
		logx.Error(err)
		return nil
	}
	for rows.Next() {
		var (
			buyer_id       string
			project_count  uint64
			project_amount decimal.Decimal
		)
		if err := rows.Scan(&buyer_id, &project_count, &project_amount); err != nil {
			logx.Error(err)
			continue
		}
		if vm[buyer_id] == nil {
			continue
		}
		vm[buyer_id].ProjectCount += int(project_count)
		pf, _ := project_amount.Float64()
		vm[buyer_id].ProjectAmount += pf
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		logx.Error(err)
	}
	result := map[string]*projectInfo{}
	for k, v := range fpn {
		if result[k] == nil {
			result[k] = &projectInfo{}
		}
		result[k].BuyerCount = len(v)
		for _, vv := range v {
			if vm[vv.CompanyId] == nil {
				continue
			}
			result[k].ProjectCount += vm[vv.CompanyId].ProjectCount
			result[k].ProjectAmount += vm[vv.CompanyId].ProjectAmount
		}
	}
	return result
}

//
func (n *network) Introduce_Supplier(values []string) map[string]*projectInfo {
	vm := map[string]*projectInfo{}
	for _, v := range values {
		vm[v] = &projectInfo{}
	}
	wh, newArgs := NetworkCom.WhArgs(values)
	rows, err := ClickhouseConn.Query(context.Background(), `select a.winner_id,count(a.buyer_id) AS buyer_count,count(b.project_id) AS project_count,sum(b.project_money) AS project_amount from information.transaction_info a 
		inner join information.transaction_info b on (hasAny(a.winner_id,[`+wh+`]) and a.buyer_id<>'' and a.buyer_id=b.buyer_id)
		group by a.winner_id`, newArgs...)
	if err != nil {
		logx.Error(err)
		return nil
	}
	for rows.Next() {
		var (
			winner_id      []string
			buyer_count    uint64
			project_count  uint64
			project_amount decimal.Decimal
		)
		if err := rows.Scan(&winner_id, &buyer_count, &project_count, &project_amount); err != nil {
			logx.Error(err)
			continue
		}
		for _, v := range winner_id {
			if vm[v] == nil {
				continue
			}
			vm[v].BuyerCount += int(buyer_count)
			vm[v].ProjectCount += int(project_count)
			pf, _ := project_amount.Float64()
			vm[v].ProjectAmount += pf
		}
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		logx.Error(err)
	}
	return vm
}

//
func (n *network) Introduce_Agency(values []string) map[string]*projectInfo {
	vm := map[string]*projectInfo{}
	for _, v := range values {
		vm[v] = &projectInfo{}
	}
	wh, newArgs := NetworkCom.WhArgs(values)
	rows, err := ClickhouseConn.Query(context.Background(), `select a.agency_id,count(a.buyer_id) AS buyer_count,count(b.project_id) AS project_count,sum(b.project_money) AS project_amount from information.transaction_info a 
		inner join information.transaction_info b on (a.agency_id in (`+wh+`) and a.buyer_id<>'' and a.buyer_id=b.buyer_id)
		group by a.agency_id`, newArgs...)
	if err != nil {
		logx.Error(err)
		return nil
	}
	for rows.Next() {
		var (
			agency_id      string
			buyer_count    uint64
			project_count  uint64
			project_amount decimal.Decimal
		)
		if err := rows.Scan(&agency_id, &buyer_count, &project_count, &project_amount); err != nil {
			logx.Error(err)
			continue
		}
		if vm[agency_id] == nil {
			continue
		}
		vm[agency_id].BuyerCount += int(buyer_count)
		vm[agency_id].ProjectCount += int(project_count)
		pf, _ := project_amount.Float64()
		vm[agency_id].ProjectAmount += pf
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		logx.Error(err)
	}
	return vm
}

//
func (n *network) Introduce_Middleman(values []string) map[string]*projectInfo {
	vm := map[string]*projectInfo{}
	wh, newArgs := NetworkCom.WhArgs(values)
	rows, err := ClickhouseConn.Query(context.Background(), `select project_id,project_money from information.transaction_info where project_id in (`+wh+`)`, newArgs...)
	if err != nil {
		logx.Error(err)
		return nil
	}
	for rows.Next() {
		var (
			project_id    string
			project_money decimal.Decimal
		)
		if err := rows.Scan(&project_id, &project_money); err != nil {
			logx.Error(err)
			continue
		}
		pf, _ := project_money.Float64()
		vm[project_id] = &projectInfo{
			ProjectAmount: pf,
		}
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		logx.Error(err)
	}
	return vm
}

//
func (n *network) TypeConvert(itype string) int {
	//firstparty：甲方 supplier：供应商 adiffb：同甲异业 middleman：中间人 agency：招标代理机构
	switch itype {
	case "firstparty":
		return 1
	case "supplier":
		return 2
	case "adiffb":
		return 3
	case "middleman":
		return 4
	case "agency":
		return 5
	}
	return 0
}