更新法人库

faren_tidb/all.go

@@ -204,15 +204,24 @@ func dealAllFromCompanyBase2() {
 		"company_type": map[string]interface{}{
 			"$ne": "个体工商户",
 		},
+		"_id": map[string]interface{}{
+			//"$lte": 964729447,
+			"$gt":  GF.Env.Startid,
+			"$lte": GF.Env.Endid,
+		},
+	}
+
+	if GF.Env.Startid >= GF.Env.Endid || GF.Env.Endid <= 0 {
+		jlog.Error("dealAllFromCompanyBase2", zap.Any("where", where), zap.Any("查询条件错误", "开始结束ID错误"))
 	}
 	// channel 作为队列
 	jobCh := make(chan map[string]interface{}, 1000) // 缓冲队列
 	entCh := make(chan EntInfo, 1000)                // 结果队列
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-
+	jlog.Error("dealAllFromCompanyBase2", zap.Any("where", where))
 	// 启动 worker 处理数据
-	workerNum := 6 // 并发度可调
+	workerNum := 10 // 并发度可调
 	var wg sync.WaitGroup
 	for i := 0; i < workerNum; i++ {
 		wg.Add(1)
@@ -253,7 +262,7 @@ func dealAllFromCompanyBase2() {
 	count := 0
 	for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
 		if count%1000 == 0 {
-			jlog.Info("dealAllFromCompanyBase", zap.Any("current:", count), zap.Any("company_name", tmp["company_name"]))
+			jlog.Info("dealAllFromCompanyBase", zap.Any("current:", count), zap.Any("company_name", tmp["company_name"]), zap.Any("id", tmp["_id"]))
 		}
 		select {
 		case jobCh <- tmp:

faren_tidb/config.go

@@ -38,6 +38,8 @@ type EnvConf struct {
 	Isw           bool   //是否保存标签里的权重
 	Esindex       string //ent_info
 	Pre           string //数据表前缀
+	Startid       int    //查询MongoDB  company_base 开始ID
+	Endid         int    //查询MongoDB  company_base 结束ID
 }
 
 type EsConf struct {

faren_tidb/config.toml

@@ -1,11 +1,11 @@
 [mongo]  ## 标讯地址
-    host = "127.0.0.1:27083"
-    #        host = "172.17.189.140:27080"
+# host = "127.0.0.1:27083"
+    host = "172.31.31.202:27081"
     db = "mixdata"
     coll = "qyxy_std"
     username = "SJZY_RWbid_ES"
     password = "SJZY@B4i4D5e6S"
-    direct = true  ## 本地代理时需要打开，
+    # direct = true  ## 本地代理时需要打开，
 
 ## 测试环境
 #    host = "192.168.3.149:27102"
@@ -17,22 +17,22 @@
 
 
 [mongoqy]  ## 181 凭安数据
-#    host = "172.17.4.181:27001"
-    host = "127.0.0.1:27001"
+    host = "172.17.4.181:27001"
+    # host = "127.0.0.1:27001"
     db = "mixdata"
     username = ""
     password = ""
-    direct = true  ## 本地代理时需要打开，
+    # direct = true  ## 本地代理时需要打开，
 
 
 
 
 [clickhouse] ## clickhouse 数据库
-#        host = "cc-2ze9tv451wov14w9e.clickhouse.ads.aliyuncs.com:9000"
-        host = "localhost:18129"
-        username = "biservice"
-        password = "Bi_top95215#"
-        db = "information"
+    host = "cc-2ze9tv451wov14w9e.clickhouse.ads.aliyuncs.com:9000"
+    # host = "localhost:18129"
+    username = "biservice"
+    password = "Bi_top95215#"
+    db = "information"
     ### 测试环境
 #    host = "172.20.45.129:18123"
 #    username = "jytop"
@@ -40,39 +40,40 @@
 #    db = "information"
 
 [mysql]  ## mysql
-#    host = "172.20.45.129:4000"
-#    host = "127.0.0.1:4001"
-#    username = "root"
-#    password = "=PDT49#80Z!RVv52_z"
-#    db = "global_common_data"
-
-    host = "172.20.45.129:4000"
-    username = "root"
-    password = "=PDT49#80Z!RVv52_z"
+    host = "172.17.162.25:4000"
+    username = "datagroup"
+    password = "Dgrpdb#2024@36"
     db = "global_common_data"
 
+# host = "172.20.45.129:4000"
+# username = "root"
+# password = "=PDT49#80Z!RVv52_z"
+# db = "global_common_data"
+
 [env]
     start = -2 ## 开始取2天前的数据，
     end = -1 ## 截止取1天前的数据，
     spec = "0 50 23 * * *"  ## 定时任务，每天23点50执行;更新法人库Bitmap
     spec2 = "0 0 18 * * 6"  ## 定时任务，没周六18点执行，读取更新凭安数据，
-#    pre = "dws_f_" ## 数据表 前缀
+    pre = "dws_f_" ## 数据表 前缀
+    startid = 0  ## 查询MongoDB  company_base 开始ID
+    endid = 27000 ## 查询MongoDB  company_base 结束ID
 
 
 
 [log]
 # 日志路径，为空将输出控制台
-    logpath = ""
-#logpath = "logs/log.out"
-# log size (M)
-maxsize = 10
-# compress log
-compress = true
-# log save  time (day)
-maxage =  7
-# save total log file total
-maxbackups = 10
-# log level
-loglevel  = "debug"
-# text or json output
-format = "json"
+#    logpath = ""
+    logpath = "logs/log.out"
+    # log size (M)
+    maxsize = 10
+    # compress log
+    compress = true
+    # log save  time (day)
+    maxage =  7
+    # save total log file total
+    maxbackups = 10
+    # log level
+    loglevel  = "debug"
+    # text or json output
+    format = "json"

faren_tidb/main.go

@@ -26,6 +26,6 @@ var (
 )
 
 func main() {
-	dealAllFromCompanyBase() //企业法人存量数据
+	dealAllFromCompanyBase2() //企业法人存量数据
 
 }