jyps/xlsx/clickhouse.go

package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"github.com/xuri/excelize/v2"
	"gorm.io/driver/clickhouse"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"
	"io"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
	"log"
	"net/http"
	"net/url"
	"sync"
	"time"
)

// EntMapCode 映射关系表
type EntMapCode struct {
	AId         string `json:"a_id"`
	BId         string `json:"b_id"`
	AName       string `json:"a_name"`
	BName       string `json:"b_name"`
	Code        string `json:"code"`         //关系属性代码
	InvestRatio string `json:"invest_ratio"` //投资比例
	InvestPrice string `json:"invest_price"` //投资金额
	CreateTime  int64  `json:"create_time"`
	UpdateTime  int64  `json:"update_time"`
}

func (EntMapCode) TableName() string {
	return "ent_map_code"
}

// clickhouseData 获取clickhouse 数据
func clickhouseData() {
	host := "cc-2ze9tv451wov14w9e.clickhouse.ads.aliyuncs.com:9000"
	username := "biservice"
	password := "Bi_top95215#"

	Mgo := &mongodb.MongodbSim{
		MongodbAddr: "172.17.189.140:27080",
		//MongodbAddr: "127.0.0.1:27083",
		Size:     10,
		DbName:   "qfw",
		UserName: "SJZY_RWbid_ES",
		Password: "SJZY@B4i4D5e6S",
		//Direct:      true,
	}
	Mgo.InitPool()

	//host := "192.168.3.207:19000"
	//username := "jytop"
	//password := "pwdTopJy123"

	encodedPassword := url.QueryEscape(password)
	dn := fmt.Sprintf("clickhouse://%s:%s@%s/information?dial_timeout=10s&read_timeout=20s", username, encodedPassword, host)
	db, err := gorm.Open(clickhouse.Open(dn), &gorm.Config{
		Logger: logger.Default.LogMode(logger.Silent),
	})
	if err != nil {
		log.Fatal("链接数据库失败")
	} else {
		log.Println("链接成功")
	}

	f, err := excelize.OpenFile("./河南单温.xlsx")
	if err != nil {
		fmt.Println(err)
		return
	}
	defer func() {
		if err := f.Close(); err != nil {
			fmt.Println(err)
		}
	}()
	Sheet := "单位"
	rows, err := f.GetRows(Sheet)
	if err != nil {
		fmt.Println(err)
		return
	}

	defer func() {
		if err := f.Close(); err != nil {
			log.Println("f.close,", err)
		}
	}()
	var wg sync.WaitGroup
	sem := make(chan bool, 15) // 控制同时最多只有10个协程
	for i := 1; i < len(rows); i++ {
		if len(rows[i]) > 1 && rows[i][1] != "" {
			insert := make(map[string]interface{})
			company := rows[i][0]
			log.Println(i, "----", company)
			insert["bname"] = rows[i][0]
			insert["aname"] = rows[i][1]
			if len(rows[i]) > 2 {
				insert["type"] = rows[i][2]
			}
			Mgo.Save("wcc_henan_0603", insert)
			continue
		}
		//
		//
		wg.Add(1)
		sem <- true
		go func(i int) {
			defer wg.Done()
			company := rows[i][0]
			insert := make(map[string]interface{})
			log.Println(i, "----", company)
			insert["bname"] = rows[i][0]
			exist := EntMapCode{}
			db.Where(&EntMapCode{BName: company}).First(&exist)
			if exist.AName != "" {
				insert["aname"] = exist.AName
			}

			if exist.Code == "0101" {
				insert["type"] = "管辖关系"
				//f.SetCellValue(Sheet, fmt.Sprintf("%s%d", "C", i+1), "管辖关系")
			} else if exist.Code == "0102" {
				insert["type"] = "直属关系"
				//f.SetCellValue(Sheet, fmt.Sprintf("%s%d", "C", i+1), "直属关系")
			} else if exist.Code == "0103" {
				insert["type"] = "组成关系"
				//f.SetCellValue(Sheet, fmt.Sprintf("%s%d", "C", i+1), "组成关系")
			}

			Mgo.Save("wcc_henan_0603", insert)
			<-sem // 释放一个位置，允许其他协程开始执行
		}(i)
	}

	wg.Wait() // 等待所有协程结束

	err = f.Save()
	if err != nil {
		log.Println("f save err", err)
	}
	log.Println("结束")
	//for i := 1; i < len(rows); i++ {
	//	company := rows[i][0]
	//	log.Println("aaa", company)
	//
	//	exist := EntMapCode{}
	//	db.Where(&EntMapCode{BName: company}).First(&exist)
	//	if exist.AName != "" {
	//		f.SetCellValue("单位", fmt.Sprintf("%s%d", "B", i+1), exist.AName)
	//	}
	//
	//	if exist.Code == "0101" {
	//		f.SetCellValue("单位", fmt.Sprintf("%s%d", "C", i+1), "管辖关系")
	//	} else if exist.Code == "0102" {
	//		f.SetCellValue("单位", fmt.Sprintf("%s%d", "C", i+1), "直属关系")
	//	} else if exist.Code == "0103" {
	//		f.SetCellValue("单位", fmt.Sprintf("%s%d", "C", i+1), "组成关系")
	//	}
	//}
	//
	//f.Save()
	//log.Println("结束")

}

// getName  更新表格数据
func getName() {
	Mgo := &mongodb.MongodbSim{
		MongodbAddr: "172.17.189.140:27080",
		//MongodbAddr: "127.0.0.1:27083",
		Size:     10,
		DbName:   "qfw",
		UserName: "SJZY_RWbid_ES",
		Password: "SJZY@B4i4D5e6S",
		//Direct:      true,
	}
	Mgo.InitPool()

	//181
	Mgo2 := &mongodb.MongodbSim{
		MongodbAddr: "172.17.4.181:27001",
		//MongodbAddr: "127.0.0.1:27001",
		DbName:   "mixdata",
		Size:     10,
		UserName: "",
		Password: "",
		//Direct:      true,
	}
	Mgo2.InitPool()

	f, err := excelize.OpenFile("./河南单位.xlsx")
	if err != nil {
		fmt.Println(err)
		return
	}
	defer func() {
		if err := f.Close(); err != nil {
			fmt.Println(err)
		}
	}()

	rows, err := f.GetRows("单位")
	if err != nil {
		fmt.Println(err)
		return
	}

	for i := 1; i < len(rows); i++ {
		name := rows[i][0]
		log.Println(i, "----", name)
		//res, _ := Mgo2.FindOne("special_enterprise", map[string]interface{}{"company_name": name})
		//if res != nil && len(*res) > 0 {
		//	f.SetCellValue("单位", fmt.Sprintf("%s%d", "E", i+1), "special_enterprise")
		//	f.SetCellValue("单位", fmt.Sprintf("%s%d", "F", i+1), (*res)["company_status"])
		//	f.SetCellValue("单位", fmt.Sprintf("%s%d", "G", i+1), (*res)["company_type"])
		//} else {
		//	ra, _ := Mgo2.FindOne("special_gov_unit", map[string]interface{}{"company_name": name})
		//	if ra != nil && len(*ra) > 0 {
		//		f.SetCellValue("单位", fmt.Sprintf("%s%d", "E", i+1), "special_gov_unit")
		//		f.SetCellValue("单位", fmt.Sprintf("%s%d", "G", i+1), (*ra)["company_type"])
		//		if util.ObjToString((*ra)["company_status"]) != "" {
		//			f.SetCellValue("单位", fmt.Sprintf("%s%d", "F", i+1), (*ra)["company_status"])
		//		} else {
		//			if util.IntAll((*ra)["use_flag"]) < 5 {
		//				f.SetCellValue("单位", fmt.Sprintf("%s%d", "F", i+1), "正常")
		//			} else {
		//				f.SetCellValue("单位", fmt.Sprintf("%s%d", "F", i+1), "废弃")
		//			}
		//		}
		//	}
		//}
		// 更新D 列
		//rb, _ := Mgo.FindOne("wcc_zhengfujigou", map[string]interface{}{"bname": name})
		//if rb != nil && len(*rb) > 0 {
		//	f.SetCellValue("单位", fmt.Sprintf("%s%d", "D", i+1), (*rb)["aname"])
		//}
		//
		data := map[string]interface{}{
			"detail": name,
		}

		res := getAreaInfo(data)
		if res != nil {
			city := res["city"]
			district := res["district"]
			f.SetCellValue("单位", fmt.Sprintf("%s%d", "B", i+1), city)
			f.SetCellValue("单位", fmt.Sprintf("%s%d", "C", i+1), district)
		}
	}

	f.Save()
	log.Println("结束")
}

// getAreaInfo 调用抽取接口，获取省市区
func getAreaInfo(data map[string]interface{}) map[string]interface{} {
	info := map[string]interface{}{}
	client := &http.Client{Timeout: 2 * time.Second}
	jsonStr, _ := json.Marshal(data)
	resp, err := client.Post("http://127.0.0.1:9996/service/region", "application/json", bytes.NewBuffer(jsonStr))
	if err != nil {
		return info
	}
	res, err := io.ReadAll(resp.Body)
	if err != nil {
		return info
	}
	err = json.Unmarshal(res, &info)
	if err != nil {
		return info
	}
	return info
}