jyps/bidding_test/qyxy.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/olivere/elastic/v7"
	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
	"io"
	util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
	"log"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"unicode/utf8"
)

func countReportYear22() {
	ctx := context.Background()

	username := "SJZY_RWbid_ES"
	password := "SJZY@B4i4D5e6S"
	hosts := []string{"172.31.31.202:27081", "172.20.45.128:27080"}

	uri, err := BuildMongoURI(username, password, hosts, nil)
	if err != nil {
		panic(err)
	}

	clientOptions := options.Client().ApplyURI(uri)
	client, err := mongo.Connect(ctx, clientOptions)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Disconnect(ctx)

	collection := client.Database("mixdata").Collection("qyxy_std")

	// 分段配置
	start := int64(1630914780)
	end := int64(1751238647)
	segment := (end - start) / 10

	type SegmentResult struct {
		Num      int
		Year2023 int
		Year2024 int
	}
	resultChan := make(chan SegmentResult, 10)

	var outerWg sync.WaitGroup
	for i := 0; i < 10; i++ {
		segStart := start + int64(i)*segment
		segEnd := segStart + segment
		if i == 9 {
			segEnd = end // 最后一段到结尾
		}

		outerWg.Add(1)
		go func(segIdx int, segStart, segEnd int64) {
			defer outerWg.Done()

			// filter: 按段过滤
			filter := bson.M{
				"company_type": bson.M{"$ne": "个体工商户"},
				"updatetime": bson.M{
					"$gte": segStart,
					"$lt":  segEnd,
				},
			}

			batchSize := int32(500)
			cursor, err := collection.Find(
				ctx,
				filter,
				options.Find().
					SetBatchSize(batchSize).
					SetSort(bson.D{{Key: "_id", Value: -1}}),
			)
			if err != nil {
				log.Printf("[segment %d] Find error: %v\n", segIdx, err)
				return
			}
			defer cursor.Close(ctx)

			// worker pool
			workerCount := 8
			docChan := make(chan map[string]interface{}, 1000)
			var wg sync.WaitGroup

			// 本段统计
			localCount := map[int]int{
				2023: 0,
				2024: 0,
			}
			num := 0
			var mu sync.Mutex

			for w := 0; w < workerCount; w++ {
				wg.Add(1)
				go func() {
					defer wg.Done()
					for doc := range docChan {
						CompanyStatus := util.ObjToString(doc["company_status"])
						CompanyName := util.ObjToString(doc["company_name"])
						UseFlag := util.IntAll(doc["use_flag"])
						if UseFlag > 0 ||
							strings.Contains(CompanyStatus, "注销") ||
							strings.Contains(CompanyStatus, "吊销") ||
							CompanyName == "" ||
							strings.Contains(CompanyName, "已除名") ||
							utf8.RuneCountInString(CompanyName) < 5 {
							continue
						}

						annualReportsRaw, ok := doc["annual_reports"]
						if !ok {
							continue
						}

						annualReports := make([]interface{}, 0)
						switch v := annualReportsRaw.(type) {
						case []interface{}:
							annualReports = v
						case primitive.A:
							annualReports = v
						default:
							continue
						}

						yearHasReport := map[int]bool{
							2023: false,
							2024: false,
						}
						for _, r := range annualReports {
							report, ok := r.(map[string]interface{})
							if !ok {
								continue
							}
							yearRaw, exists := report["report_year"]
							if !exists {
								continue
							}
							var yearInt int
							switch v := yearRaw.(type) {
							case string:
								y, err := strconv.Atoi(v)
								if err != nil {
									continue
								}
								yearInt = y
							case float64:
								yearInt = int(v)
							case int32:
								yearInt = int(v)
							case int64:
								yearInt = int(v)
							default:
								continue
							}
							if yearInt == 2023 {
								yearHasReport[2023] = true
							} else if yearInt == 2024 {
								yearHasReport[2024] = true
							}
						}

						mu.Lock()
						if yearHasReport[2023] {
							localCount[2023]++
						}
						if yearHasReport[2024] {
							localCount[2024]++
						}
						mu.Unlock()
					}
				}()
			}

			// 遍历 cursor
			for cursor.Next(ctx) {
				var doc map[string]interface{}
				if err := cursor.Decode(&doc); err != nil {
					continue
				}
				num++
				docChan <- doc
				if num%10000 == 0 {
					mu.Lock()
					log.Printf("[segment %d] processed: %d, 2023: %d, 2024: %d", segIdx, num, localCount[2023], localCount[2024])
					mu.Unlock()
				}
			}
			close(docChan)
			wg.Wait()

			resultChan <- SegmentResult{
				Num:      num,
				Year2023: localCount[2023],
				Year2024: localCount[2024],
			}
		}(i, segStart, segEnd)
	}

	// 等待所有段结束
	go func() {
		outerWg.Wait()
		close(resultChan)
	}()

	// 汇总
	totalDocs := 0
	finalCount := map[int]int{
		2023: 0,
		2024: 0,
	}
	for segRes := range resultChan {
		totalDocs += segRes.Num
		finalCount[2023] += segRes.Year2023
		finalCount[2024] += segRes.Year2024
	}

	fmt.Printf("总处理文档数: %d\n", totalDocs)
	fmt.Printf("2023 年有年报的企业数: %d\n", finalCount[2023])
	fmt.Printf("2024 年有年报的企业数: %d\n", finalCount[2024])
}

// countReportYear  统计企业年报
func countReportYear() {
	ctx := context.Background()

	username := "SJZY_RWbid_ES"
	password := "SJZY@B4i4D5e6S"
	hosts := []string{"172.31.31.202:27081", "172.20.45.128:27080"}
	//hosts := []string{"127.0.0.1:27083"}

	// 构造 URI
	uri, err := BuildMongoURI(username, password, hosts, nil)
	if err != nil {
		panic(err)
	}

	// 连接 MongoDB
	clientOptions := options.Client().ApplyURI(uri)

	//clientOptions.SetDirect(true)
	client, err := mongo.Connect(ctx, clientOptions)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Disconnect(ctx)

	collection := client.Database("mixdata").Collection("qyxy_std")

	// 查询条件：company_type != 个体工商户
	filter := bson.M{"company_type": bson.M{"$ne": "个体工商户"}}

	// 批量大小
	batchSize := int32(500)

	//cursor, err := collection.Find(ctx, filter, options.Find().SetBatchSize(batchSize))
	cursor, err := collection.Find(
		ctx,
		filter,
		options.Find().
			SetBatchSize(batchSize).
			SetSort(bson.D{{Key: "_id", Value: -1}}),
	)

	if err != nil {
		log.Fatal(err)
	}
	defer cursor.Close(ctx)

	// 定义统计变量
	finalCount := map[int]int{
		2023: 0,
		2024: 0,
	}
	var mu sync.Mutex

	// worker 并发数
	workerCount := 8
	docChan := make(chan map[string]interface{}, 1000)
	var wg sync.WaitGroup

	// 启动 worker
	for i := 0; i < workerCount; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for doc := range docChan {
				// 过滤条件
				CompanyStatus := util.ObjToString(doc["company_status"])
				CompanyName := util.ObjToString(doc["company_name"])
				UseFlag := util.IntAll(doc["use_flag"])
				if UseFlag > 0 ||
					strings.Contains(CompanyStatus, "注销") ||
					strings.Contains(CompanyStatus, "吊销") ||
					CompanyName == "" ||
					strings.Contains(CompanyName, "已除名") ||
					utf8.RuneCountInString(CompanyName) < 5 {
					continue
				}

				// annual_reports
				annualReportsRaw, ok := doc["annual_reports"]
				if !ok {
					continue
				}

				annualReports := make([]interface{}, 0)
				switch v := annualReportsRaw.(type) {
				case []interface{}:
					annualReports = v
				case primitive.A:
					annualReports = v
				default:
					log.Printf("annual_reports unexpected type: %T\n", v)
					continue
				}

				//annualReports, ok := annualReportsRaw.([]interface{})
				//if !ok {
				//	continue
				//}

				// 检测当前企业是否在 2023/2024 有年报，只算一次
				yearHasReport := map[int]bool{
					2023: false,
					2024: false,
				}

				for _, r := range annualReports {
					report, ok := r.(map[string]interface{})
					if !ok {
						continue
					}
					yearRaw, exists := report["report_year"]
					if !exists {
						continue
					}

					var yearInt int
					switch v := yearRaw.(type) {
					case string:
						y, err := strconv.Atoi(v)
						if err != nil {
							continue
						}
						yearInt = y
					case float64:
						yearInt = int(v)
					case int32:
						yearInt = int(v)
					case int64:
						yearInt = int(v)
					default:
						continue
					}

					if yearInt == 2023 {
						yearHasReport[2023] = true
					} else if yearInt == 2024 {
						yearHasReport[2024] = true
					}
				}

				// 有就+1
				mu.Lock()
				if yearHasReport[2023] {
					finalCount[2023]++
				}
				if yearHasReport[2024] {
					finalCount[2024]++
				}
				mu.Unlock()
			}
		}()
	}

	// 主 goroutine 遍历 cursor，实时打印进度
	num := 0
	for cursor.Next(ctx) {
		var doc map[string]interface{}
		if err := cursor.Decode(&doc); err != nil {
			log.Println("decode error:", err)
			continue
		}
		num++
		docChan <- doc

		if num%10000 == 0 {
			mu.Lock()
			log.Printf("current: %d docs processed, 2023年企业数: %d, 2024年企业数: %d\n", num, finalCount[2023], finalCount[2024])
			mu.Unlock()
		}
	}
	close(docChan)

	// 等待所有 worker 完成
	wg.Wait()

	// 输出统计结果
	fmt.Printf("总处理文档数: %d\n", num)
	fmt.Printf("2023 年有年报的企业数: %d\n", finalCount[2023])
	fmt.Printf("2024 年有年报的企业数: %d\n", finalCount[2024])
}

// BuildMongoURI 构造 MongoDB 连接 URI
func BuildMongoURI(username, password string, hosts []string, options map[string]string) (string, error) {
	if len(hosts) == 0 {
		return "", fmt.Errorf("hosts cannot be empty")
	}

	hostList := strings.Join(hosts, ",")

	var authPart string
	if username != "" {
		escapedUsername := url.QueryEscape(username)
		escapedPassword := url.QueryEscape(password)
		authPart = fmt.Sprintf("%s:%s@", escapedUsername, escapedPassword)
		// 如果密码为空，也会拼成 username:@host ，MongoDB URI 是支持的，可以保留
	}

	var optionStr string
	if len(options) > 0 {
		query := url.Values{}
		for k, v := range options {
			query.Set(k, v)
		}
		optionStr = "?" + query.Encode()
	}

	return fmt.Sprintf("mongodb://%s%s%s", authPart, hostList, optionStr), nil
}

func fixQyxy() {
	// 连接 ES
	url := "http://172.17.4.184:19908"
	//url := "http://127.0.0.1:19908"
	username := "jybid"
	password := "Top2023_JEB01i@31"

	client, err := elastic.NewClient(
		elastic.SetURL(url),
		elastic.SetBasicAuth(username, password),
		elastic.SetSniff(false),
	)
	if err != nil {
		log.Fatalf("创建 Elasticsearch 客户端失败：%s", err)
	}

	sess := MgoQy.GetMgoConn()
	defer MgoQy.DestoryMongoConn(sess)

	where := map[string]interface{}{
		"use_flag": 10,
	}
	queryMgo := sess.DB("mixdata").C("qyxy_std").Find(where).Select(nil).Iter()
	count := 0
	for tmp := make(map[string]interface{}); queryMgo.Next(tmp); count++ {
		if count%1000 == 0 {
			log.Println("current:", count, tmp["_id"])
		}
		id := util.ObjToString(tmp["_id"])
		company_name := util.ObjToString(tmp["company_name"])
		if company_name == "无" || company_name == "|" || company_name == "" {
			continue
		}
		where2 := map[string]interface{}{
			"company_name": company_name,
			"use_flag":     0,
		}
		std, _ := MgoQy.FindOne("qyxy_std", where2)
		var newID string
		if len(*std) > 0 {
			newID = util.ObjToString(tmp["_id"])
		}

		query := elastic.NewBoolQuery().
			Must(
				elastic.NewTermQuery("entidlist", id), // 模糊匹配 projectname
			)
		ctx := context.Background()
		//开始滚动搜索
		scrollID := ""
		scroll := "10m"
		searchSource := elastic.NewSearchSource().
			Query(query).
			Size(10000).
			Sort("_doc", true) //升序排序
		//Sort("_doc", false) //降序排序

		searchService := client.Scroll("projectset").
			Size(10000).
			Scroll(scroll).
			SearchSource(searchSource)

		res, err := searchService.Do(ctx)

		if err != nil {
			if err == io.EOF {
				log.Println("没有数据")
			} else {
				log.Println(err)
				continue
			}
		}
		//defer client.ClearScroll().ScrollId(scrollID).Do(ctx) // 在退出时清理资源
		fmt.Println("总数是：", res.TotalHits())
		total := 0
		//1.处理更新es 数据
		for len(res.Hits.Hits) > 0 {
			for _, hit := range res.Hits.Hits {
				var doc map[string]interface{}
				err := json.Unmarshal(hit.Source, &doc)
				if err != nil {
					log.Printf("解析文档失败：%s", err)
					continue
				}
				esID := util.ObjToString(doc["id"])
				newEntidlist := make([]string, 0)
				//存入新表
				if entidlist, ok := doc["entidlist"].([]interface{}); ok && len(entidlist) > 0 {
					for _, v := range entidlist {
						list_id := util.ObjToString(v)
						if list_id != id && list_id != "-" {
							newEntidlist = append(newEntidlist, list_id)
						}
					}

					if newID != "" {
						newEntidlist = append(newEntidlist, newID)
					}
					//更新es
					esUpdate := map[string]interface{}{
						"entidlist": newEntidlist,
					}
					// 更新Es 数据
					updateEsPool <- []map[string]interface{}{
						{"_id": esID},
						esUpdate,
					}
					log.Println("aaaaaaaa", esID, esUpdate)
					//更新项目MongoDB
					MgoP.UpdateById("projectset_20230904", esID, map[string]interface{}{"$set": esUpdate})

				}
			}

			total = total + len(res.Hits.Hits)
			scrollID = res.ScrollId
			res, err = client.Scroll().ScrollId(scrollID).Scroll(scroll).Do(ctx)
			log.Println("current count:", total)
			if err != nil {
				if err == io.EOF {
					// 滚动到最后一批数据，退出循环
					break
				}
				log.Println("滚动搜索失败：", err, res)
				break // 处理错误时退出循环
			}
		}

		//2.删除MongoDB
		whereDel := map[string]interface{}{
			"_id": id,
		}
		MgoQy.Delete("qyxy_std", whereDel)
		MgoQy.SaveByOriID("wcc_qyxy_std_delete", tmp)
		//在循环外调用 ClearScroll
		_, err = client.ClearScroll().ScrollId(scrollID).Do(ctx)
		if err != nil {
			log.Printf("清理滚动搜索失败：%s", err)
		}

		_, err = client.Scroll().ScrollId(scrollID).Scroll("2m").Do(ctx)
		if err != nil {
			if elasticErr, ok := err.(*elastic.Error); ok && elasticErr.Status == 400 {
				log.Println("滚动查询失败，可能是 Scroll ID 失效，直接退出")
				return
			}
			log.Println("滚动查询失败：", err)
		}

		_, err = client.ClearScroll().ScrollId(scrollID).Do(ctx)
		if err != nil {
			log.Printf("清理滚动搜索失败：%s", err)
		}
	}
	log.Println("数据处理完毕")
}

// findData  找出企业信息中，company_type 不等于合体工商户的企业数据，然后写入一个临时表
func findData() {
	Mgo := &mongodb.MongodbSim{
		MongodbAddr: "172.17.189.140:27080",
		//MongodbAddr: "127.0.0.1:27083",
		Size:     10,
		DbName:   "mixdata",
		UserName: "SJZY_RWbid_ES",
		Password: "SJZY@B4i4D5e6S",
		//Direct:      true,
	}
	Mgo.InitPool()

	sess := Mgo.GetMgoConn()
	defer Mgo.DestoryMongoConn(sess)

	//where := map[string]interface{}{
	//	"company_type": map[string]interface{}{
	//		"$ne": "个体工商户",
	//	},
	//}
	query := sess.DB("mixdata").C("qyxy_std").Find(nil).Select(map[string]interface{}{"company_name": 1, "company_type": 1, "company_status": 1, "use_flag": 1}).Iter()
	count := 0
	for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
		if count%10000 == 0 {
			log.Println("current:", count)
		}
		companyType := util.ObjToString(tmp["company_type"])
		if companyType == "个体工商户" {
			continue
		}

		company_name := util.ObjToString(tmp["company_name"])
		whereN := map[string]interface{}{
			"company_name": company_name,
		}
		num := Mgo.Count("qyxy_std", whereN)
		if num > 1 {
			Mgo.Save("wcc_qyxy_20240311", tmp)
		}
		tmp = make(map[string]interface{})
	}

	log.Println("结束")

}

// getCompanyName
func getCompanyName() {
	// 找出wcc_qyxy_20240311表中，公司名称有多个，并且一个use_flag=0,一个use_flag=10的数据
	Mgo := &mongodb.MongodbSim{
		MongodbAddr: "172.17.189.140:27080",
		//MongodbAddr: "127.0.0.1:27083",
		Size:     10,
		DbName:   "mixdata",
		UserName: "SJZY_RWbid_ES",
		Password: "SJZY@B4i4D5e6S",
		//Direct:      true,
	}
	Mgo.InitPool()
	companyMap := make(map[string]bool)

	sess := Mgo.GetMgoConn()
	defer Mgo.DestoryMongoConn(sess)
	query := sess.DB("mixdata").C("wcc_qyxy_20240311").Find(nil).Select(map[string]interface{}{"company_name": 1, "use_flag": 1}).Iter()
	count := 0
	for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
		if count%10000 == 0 {
			log.Println("current:", count)
		}

		name := util.ObjToString(tmp["company_name"])
		if companyMap[name] {
			continue
		} else {
			companyMap[name] = true
		}

		where := map[string]interface{}{
			"company_name": name,
		}
		res, _ := Mgo.Find("wcc_qyxy_20240311", where, nil, nil, false, -1, -1)
		flaga := false
		flagb := false
		for _, v := range *res {
			if util.Int64All(v["use_flag"]) == 0 {
				flaga = true
			} else if util.Int64All(v["use_flag"]) == 10 {
				flagb = true
			}
		}
		// 存在0和10 二个状态
		if flaga && flagb {
			Mgo.Save("wcc_qyxy_name_0325", map[string]interface{}{"company_name": name})
		}

		tmp = make(map[string]interface{})
	}

	log.Println("结束")
}

// SpecialData 处理特殊企业数据，更新 qyxy_std 表use_flag
func SpecialData() {
	// 处理 特殊企业 数据,
	tables := []string{"special_enterprise", "special_foundation", "special_law_office", "special_social_organ", "special_trade_union"}
	//1.先处理 special_enterprise 表数据，循环数据，根据company_name 去查询 qyxy_std 表，如果
	// 数据只有一条，并且 special_enterprise.company_id 等于 qyxy_std._id；就更新 qyxy_std表的 use_flag 字段
	// qyxy_std 表
	Mgo := &mongodb.MongodbSim{
		MongodbAddr: "172.17.189.140:27080",
		//MongodbAddr: "127.0.0.1:27083",
		Size:     10,
		DbName:   "mixdata",
		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()

	sess := Mgo2.GetMgoConn()
	defer Mgo2.DestoryMongoConn(sess)

	for _, v := range tables {
		query := sess.DB("mixdata").C(v).Find(nil).Select(map[string]interface{}{"company_name": 1, "use_flag": 1, "company_id": 1}).Iter()
		count := 0
		for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
			if count%10000 == 0 {
				log.Println("current:", count, "table - ", v, tmp["company_name"])
			}

			if _, ok := tmp["company_id"]; !ok {
				continue
			}
			where := map[string]interface{}{
				"_id": tmp["company_id"],
			}
			//update := map[string]interface{}{
			//	"use_flag": tmp["use_flag"],
			//}
			//Mgo.Update("qyxy_std", where, map[string]interface{}{"$set": update}, true, false)
			//name := util.ObjToString(tmp["company_name"])

			//where := map[string]interface{}{
			//	"company_name": name,
			//}
			// 如果 ID 相同，std 表use_flag =10,更新
			datas, _ := Mgo.FindOne("qyxy_std", where)
			if len(*datas) == 0 {
				continue
			}
			//if util.Int64All((*datas)["use_flag"]) == util.Int64All(tmp["use_flag"]) {
			//	continue
			//}

			//Mgo.Update("qyxy_std", where, map[string]interface{}{"$set": update}, true, false)
			data := *datas
			data["use_flag"] = tmp["use_flag"]
			Mgo.SaveByOriID("wcc_std_0411", data)
		}
	}

	log.Println("结束")

}

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

	sess := Mgo.GetMgoConn()
	defer Mgo.DestoryMongoConn(sess)
	query := sess.DB("mixdata").C("wcc_std_0410").Find(nil).Select(nil).Iter()
	count := 0
	for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
		if count%10000 == 0 {
			log.Println("current:", count)
		}
		autoid := util.Int64All(tmp["autoid"])
		if autoid == 0 {
			continue
		}

		where := map[string]interface{}{
			"autoid": autoid,
		}
		datas, _ := Mgo.FindOne("qyxy_std", where)
		if len(*datas) == 0 {
			continue
		}
		Mgo.SaveByOriID("wcc_std_0411", datas)
	}

	log.Println("over")
}