jyps/test/company.go

package main

import (
	util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
	"log"
	"strings"
	"sync"
)

// getCompany  获取公司信息
func getCompany() {
	// 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()

	//1.获取凭安特企，河南数据
	where := map[string]interface{}{
		"report_year": 2023,
	}

	defer util.Catch()
	sess := Mgo2.GetMgoConn()
	defer Mgo2.DestoryMongoConn(sess)
	count := 0
	it := sess.DB("mixdata").C("annual_report_asset").Find(where).Select(nil).Iter()
	for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
		if count%10000 == 0 {
			log.Println("corrent:", count, tmp["company_id"])
		}

		isa := false
		isb := false
		//总资产,个体户选择不公示
		if tmp["total_amount"] == nil || util.ObjToString(tmp["total_amount"]) == "企业选择不公示" || util.ObjToString(tmp["total_amount"]) == "个体户选择不公示" {
			isa = true
		}
		if tmp["business_income"] == nil || util.ObjToString(tmp["business_income"]) == "企业选择不公示" || util.ObjToString(tmp["business_income"]) == "个体户选择不公示" {
			isb = true
		}
		if isa && isb {
			continue
		}
		insert := map[string]interface{}{
			"company_id":      tmp["company_id"],
			"total_amount":    tmp["total_amount"],
			"business_income": tmp["business_income"],
			"use_flag":        tmp["use_flag"],
		}

		Mgo2.InsertOrUpdate("wcc", "wcc_annual_report_asset", insert)
	}
}

// updateReport 更新补充年报信息
func updateReport() {
	// 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()

	MgoB := &mongodb.MongodbSim{
		MongodbAddr: "172.31.31.202:27081,172.20.45.128:27080",
		//MongodbAddr: "127.0.0.1:27083",
		Size:     10,
		DbName:   "qfw",
		UserName: "SJZY_RWbid_ES",
		Password: "SJZY@B4i4D5e6S",
		//Direct:      true,
	}
	MgoB.InitPool()
	defer util.Catch()
	sess := MgoB.GetMgoConn()
	defer MgoB.DestoryMongoConn(sess)
	count := 0
	ch := make(chan bool, 15)
	wg := &sync.WaitGroup{}

	it := sess.DB("qfw").C("wcc_2025_guangdong_qyxy").Find(nil).Select(nil).Iter()
	for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
		if count%10000 == 0 {
			log.Println("current:", count, tmp["company_name"])
		}
		ch <- true
		wg.Add(1)

		go func(tmp map[string]interface{}) {
			defer func() {
				<-ch
				wg.Done()
			}()
			where := map[string]interface{}{
				"company_id":  tmp["id"],
				"report_year": 2023,
			}

			id := mongodb.BsonIdToSId(tmp["_id"])
			report, _ := Mgo2.FindOne("annual_report_asset", where)
			if (*report) != nil {
				update := make(map[string]interface{})
				total_amount := util.ObjToString((*report)["total_amount"])
				if total_amount != "" && !strings.Contains(total_amount, "不公示") {
					update["total_amount"] = (*report)["total_amount"]
					update["total_amount2"] = strings.TrimSpace(strings.ReplaceAll(total_amount, "万元", ""))
				}
				business_income := util.ObjToString((*report)["business_income"])
				if business_income != "" && !strings.Contains(business_income, "不公示") {
					update["business_income"] = (*report)["business_income"]
					update["business_income2"] = strings.TrimSpace(strings.ReplaceAll(business_income, "万元", ""))
				}

				if len(update) > 0 {
					MgoB.UpdateById("wcc_2025_guangdong_qyxy", id, map[string]interface{}{"$set": update})
				}
			}
		}(tmp)

		tmp = map[string]interface{}{}
	}

	wg.Wait()
	log.Println("数据处理完毕")
}

// updateCapital 获取更新注册资金
func updateCapital() {
	MgoB2 := &mongodb.MongodbSim{
		MongodbAddr: "172.31.31.202:27081,172.20.45.128:27080",
		//MongodbAddr: "127.0.0.1:27083",
		Size:     10,
		DbName:   "mixdata",
		UserName: "SJZY_RWbid_ES",
		Password: "SJZY@B4i4D5e6S",
		//Direct:      true,
	}
	MgoB2.InitPool()

	MgoB := &mongodb.MongodbSim{
		MongodbAddr: "172.31.31.202:27081,172.20.45.128:27080",
		//MongodbAddr: "127.0.0.1:27083",
		Size:     10,
		DbName:   "qfw",
		UserName: "SJZY_RWbid_ES",
		Password: "SJZY@B4i4D5e6S",
		//Direct:      true,
	}
	MgoB.InitPool()
	defer util.Catch()
	sess := MgoB.GetMgoConn()
	defer MgoB.DestoryMongoConn(sess)
	count := 0
	ch := make(chan bool, 15)
	wg := &sync.WaitGroup{}

	it := sess.DB("qfw").C("wcc_2025_guangdong_qyxy").Find(nil).Select(nil).Iter()
	for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
		if count%10000 == 0 {
			log.Println("current:", count, tmp["company_name"])
		}
		ch <- true
		wg.Add(1)

		go func(tmp map[string]interface{}) {
			defer func() {
				<-ch
				wg.Done()
			}()
			where := map[string]interface{}{
				"_id": tmp["id"],
			}
			
			id := tmp["_id"]
			report, _ := MgoB2.FindOne("qyxy_std", where)
			if (*report) != nil {
				update := make(map[string]interface{})
				if _, ok := (*report)["capital"]; ok {
					update["capital"] = (*report)["capital"]
				}
				if _, ok := (*report)["real_capital"]; ok {
					update["real_capital"] = (*report)["real_capital"]
				}
				if len(update) > 0 {
					MgoB.UpdateById("wcc_2025_guangdong_qyxy", id, map[string]interface{}{"$set": update})
				}
			}
		}(tmp)

		tmp = map[string]interface{}{}
	}

	wg.Wait()
	log.Println("数据处理完毕")
}