networkManage/api/internal/service/initNetwork.go

package service

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

	util "app.yhyue.com/moapp/jybase/common"
	. "app.yhyue.com/moapp/jybase/date"
	. "bp.jydev.jianyu360.cn/CRM/networkManage/api/common"
	"github.com/zeromicro/go-zero/core/logx"
)

type ContactInfo struct {
	Name  string
	Phone string
}
type Connection struct {
	Company_id   string
	Company_name string
	Itype        int
	QyxyId       string
	*ContactInfo
}
type InitNetwork struct {
	PositionId   int64
	EntId        int64
	DeptId       int64
	UserId       int64
	EntName      string
	BusinessType string
}

func (i *InitNetwork) Init() {
	logx.Info("进行初始化设置", fmt.Sprintf("%+v", i))
	array := []*Connection{}
	ids := []string{}
	idMap := map[string]bool{}
	yearLimit := NetworkCom.ProjectYearLimit()
	//①业主人脉：当前企业曾经合作过物业项目的采购单位的联系人作为业主人脉
	index1 := 0
	func() {
		rows, err := ClickhouseConn.Query(context.Background(), `select DISTINCT buyer,buyer_id from information.transaction_info_all prewhere has(winner,?) and zbtime>? order by zbtime desc limit ?`, i.EntName, yearLimit, C.InitNetworkMaxSize)
		if err != nil {
			logx.Error(err)
			return
		}
		for rows.Next() {
			var (
				buyer    string
				buyer_id string
			)
			if err := rows.Scan(&buyer, &buyer_id); err != nil {
				logx.Error(err)
				continue
			}
			if buyer == "" || buyer_id == "" {
				continue
			}
			if !idMap[buyer_id] {
				ids = append(ids, buyer_id)
				idMap[buyer_id] = true
			}
			array = append(array, &Connection{
				Company_id:   buyer_id,
				Company_name: buyer,
				Itype:        1,
			})
			index1++
			if index1 == C.InitNetworkMaxSize {
				break
			}
		}
	}()
	logx.Info("初始化设置业主人脉over。。。", index1)
	//②甲异业渠道（人脉）：“当前企业曾经合作过物业项目的采购单位的中标物业企业中”与当前企业不一样业态的企业的联系人列表
	index2 := 0
	func() {
		if i.BusinessType == "" {
			return
		}
		args := []interface{}{i.EntName, i.EntName}
		wh, newArgs := util.WhArgs(strings.Split(i.BusinessType, ","))
		args = append(args, newArgs...)
		args = append(args, yearLimit)
		args = append(args, C.InitNetworkMaxSize)
		rows, err := ClickhouseConn.Query(context.Background(), `select DISTINCT wr,wr_id from information.transaction_info_all ARRAY JOIN winner_id as wr_id,winner as wr where buyer_id IN (SELECT buyer_id from information.transaction_info_all prewhere has(winner,?) and buyer_id<>'') and wr<>? and LENGTH(winner)=LENGTH(winner_id) and hasAny(topscopeclass,[`+wh+`])=0 and wr_id<>'' AND zbtime>? order by zbtime desc limit ?`, args...)
		if err != nil {
			logx.Error(err)
			return
		}
		for rows.Next() {
			var (
				winner    string
				winner_id string
			)
			if err := rows.Scan(&winner, &winner_id); err != nil {
				logx.Error(err)
				continue
			}
			if winner == "" || winner_id == "" {
				continue
			}
			if !idMap[winner_id] {
				ids = append(ids, winner_id)
				idMap[winner_id] = true
			}
			array = append(array, &Connection{
				Company_id:   winner_id,
				Company_name: winner,
				Itype:        3,
			})
			index2++
			if index2 == C.InitNetworkMaxSize {
				break
			}
		}
	}()
	logx.Info("初始化设置甲异业渠道人脉over。。。", index2)
	//③招标代理机构（人脉）：当前企业曾经合作过的招标代理机构联系人信息
	index3 := 0
	func() {
		rows, err := ClickhouseConn.Query(context.Background(), `select DISTINCT agency,agency_id from information.transaction_info_all prewhere (buyer=? or has(winner,?)) and agency<>'' and agency_id<>'' and zbtime>? order by zbtime desc limit ?`, i.EntName, i.EntName, yearLimit, C.InitNetworkMaxSize)
		if err != nil {
			logx.Error(err)
			return
		}
		for rows.Next() {
			var (
				agency    string
				agency_id string
			)
			if err := rows.Scan(&agency, &agency_id); err != nil {
				logx.Error(err)
				continue
			}
			if agency == "" || agency_id == "" {
				continue
			}
			if !idMap[agency_id] {
				ids = append(ids, agency_id)
				idMap[agency_id] = true
			}
			array = append(array, &Connection{
				Company_id:   agency_id,
				Company_name: agency,
				Itype:        5,
			})
			index3++
			if index3 == C.InitNetworkMaxSize {
				break
			}
		}
	}()
	logx.Info("初始化设置招标代理机构人脉over。。。", index3)
	newIds := []string{}
	cis := map[string]*ContactInfo{}
	cids := map[string]string{}
	logx.Info("初始化设置总共企业数量", len(ids))
	for _, v := range ids {
		newIds = append(newIds, v)
		if len(newIds) == C.SelectBatchSize {
			for kk, vv := range i.GetContactInfo(newIds) {
				cis[kk] = vv
			}
			for kk, vv := range i.GetCompanyId(newIds) {
				cids[kk] = vv
			}
			newIds = []string{}
		}
	}
	if len(newIds) > 0 {
		for kk, vv := range i.GetContactInfo(newIds) {
			cis[kk] = vv
		}
		for kk, vv := range i.GetCompanyId(newIds) {
			cids[kk] = vv
		}
		newIds = []string{}
	}
	for _, v := range array {
		if cis[v.Company_id] != nil {
			v.ContactInfo = cis[v.Company_id]
		}
		v.QyxyId = cids[v.Company_id]
	}
	//
	nowFormat := NowFormat(Date_Full_Layout)
	newArray := []*Connection{}
	fields := []string{"position_id", "ent_id", "ent_dept_id", "ent_user_id", "itype", "company_name", "company_id", "qyxy_id", "contact_person", "contact_phone", "status", "source", "create_time", "update_time"}
	if CrmMysql.ExecTx("初始化人脉", func(tx *sql.Tx) bool {
		if CrmMysql.UpdateOrDeleteBySqlByTx(tx, `delete from crm.connection where position_id=? and source=1`, i.PositionId) < 0 {
			return false
		}
		pool := make(chan bool, C.PoolSize)
		lock := &sync.Mutex{}
		wait := &sync.WaitGroup{}
		for _, vt := range array {
			if vt.Company_name == i.EntName || vt.ContactInfo == nil || vt.ContactInfo.Phone == "" || vt.ContactInfo.Name == "" {
				continue
			}
			pool <- true
			wait.Add(1)
			go func(v *Connection) {
				defer func() {
					<-pool
					wait.Done()
				}()
				if CrmMysql.CountBySqlByTx(tx, `select count(1) from crm.connection where position_id=? and company_id=? and itype=?`, i.PositionId, v.Company_id, v.Itype) != 0 {
					logx.Info("过滤掉已存在的人脉", fmt.Sprintf("%+v", v))
					return
				}
				lock.Lock()
				defer lock.Unlock()
				logx.Info("保存人脉", fmt.Sprintf("%+v", v))
				newArray = append(newArray, v)
			}(vt)
		}
		wait.Wait()
		values := []interface{}{}
		for _, v := range newArray {
			values = append(values, i.PositionId, i.EntId, i.DeptId, i.UserId, v.Itype, v.Company_name, v.Company_id, v.QyxyId, v.ContactInfo.Name, v.ContactInfo.Phone, 1, 1, nowFormat, nowFormat)
			if len(values) == 200 {
				if r1, r2 := CrmMysql.InsertBatchByTx(tx, "crm.connection", fields, values); r1 <= 0 || r2 <= 0 {
					return false
				}
				values = []interface{}{}
			}
		}
		if len(values) > 0 {
			if r1, r2 := CrmMysql.InsertBatchByTx(tx, "crm.connection", fields, values); r1 <= 0 || r2 <= 0 {
				return false
			}
			values = []interface{}{}
		}
		return true
	}) {
		Network.DeleteCache(i.PositionId)
	}
}

//
func (i *InitNetwork) GetCompanyId(ids []string) map[string]string {
	m := map[string]string{}
	if len(ids) == 0 {
		return m
	}
	wh, args := util.WhArgs(ids)
	rows, err := ClickhouseConn.Query(context.Background(), `select id,company_id from information.ent_info prewhere id in(`+wh+`)`, args...)
	if err != nil {
		logx.Error(err)
		return m
	}
	for rows.Next() {
		var (
			id         string
			company_id string
		)
		if err := rows.Scan(&id, &company_id); err != nil {
			logx.Error(err)
			continue
		}
		m[id] = company_id
	}
	return m
}

//
func (i *InitNetwork) GetContactInfo(ids []string) map[string]*ContactInfo {
	m := map[string]*ContactInfo{}
	if len(ids) == 0 {
		return m
	}
	wh, args := util.WhArgs(ids)
	rows, err := ClickhouseConn.Query(context.Background(), `select id,phone,name from information.ent_contact prewhere id in (`+wh+`)`, args...)
	if err != nil {
		logx.Error(err)
		return m
	}
	for rows.Next() {
		var (
			id    string
			phone string
			name  string
		)
		if err := rows.Scan(&id, &phone, &name); err != nil {
			logx.Error(err)
			continue
		}
		if phone == "" || name == "" {
			continue
		}
		m[id] = &ContactInfo{
			Name:  name,
			Phone: phone,
		}
	}
	return m
}