fix: go rpc python error

app/gateway/rpc/init.go

@@ -40,8 +40,8 @@ func Init() {
 	ctx, CancelFunc = context.WithTimeout(context.Background(), 3*time.Second)
 
 	defer Register.Close()
-	initClient(config.Conf.Domain[consts.UserServiceName].Name, &UserClient)
-	initClient(config.Conf.Domain[consts.FavoriteServiceName].Name, &FavoriteClient)
+	// initClient(config.Conf.Domain[consts.UserServiceName].Name, &UserClient)
+	// initClient(config.Conf.Domain[consts.FavoriteServiceName].Name, &FavoriteClient)
 	initClient(config.Conf.Domain[consts.SearchServiceName].Name, &SearchEngineClient)
 	initClient(config.Conf.Domain[consts.IndexPlatformName].Name, &IndexPlatformClient)
 	initClient(config.Conf.Domain[consts.SearchVectorName].Name, &SearchVectorClient)

app/search_engine/repository/storage/init.go

@@ -2,9 +2,12 @@ package storage
 
 import (
 	"context"
+	"fmt"
 )
 
 func InitStorageDB(ctx context.Context) {
 	InitInvertedDB(ctx)
+	fmt.Println("InitInvertedDB finish")
 	InitGlobalTrieDB(ctx)
+	fmt.Println("InitGlobalTrieDB finish")
 }

app/search_engine/rpc/init.go

@@ -12,6 +12,7 @@ import (
 	"google.golang.org/grpc/resolver"
 
 	"github.com/CocaineCong/tangseng/config"
+	"github.com/CocaineCong/tangseng/consts"
 	"github.com/CocaineCong/tangseng/idl/pb/search_vector"
 	"github.com/CocaineCong/tangseng/pkg/discovery"
 )
@@ -31,7 +32,7 @@ func Init() {
 	ctx, CancelFunc = context.WithTimeout(context.Background(), 3*time.Second)
 
 	defer Register.Close()
-	initClient(config.Conf.Domain["search_vector"].Name, &SearchVectorClient)
+	initClient(config.Conf.Domain[consts.SearchVectorName].Name, &SearchVectorClient)
 }
 
 // initClient 初始化所有的rpc客户端

app/search_engine/service/recall/recall.go

@@ -7,11 +7,11 @@ import (
 	"github.com/samber/lo"
 	"github.com/spf13/cast"
 
-	"github.com/CocaineCong/tangseng/app/gateway/rpc"
 	"github.com/CocaineCong/tangseng/app/search_engine/analyzer"
 	"github.com/CocaineCong/tangseng/app/search_engine/ranking"
 	"github.com/CocaineCong/tangseng/app/search_engine/repository/db/dao"
 	"github.com/CocaineCong/tangseng/app/search_engine/repository/storage"
+	"github.com/CocaineCong/tangseng/app/search_engine/rpc"
 	pb "github.com/CocaineCong/tangseng/idl/pb/search_vector"
 	log "github.com/CocaineCong/tangseng/pkg/logger"
 	"github.com/CocaineCong/tangseng/repository/redis"

app/search_vector/consts/consts.py

@@ -1,4 +1,4 @@
 """
 default some constants
 """
-KAFKA_CONSUMER_VECTOR_INDEX_TOPIC = "kafka_consumer_vector_index_topic"
+KAFKA_CONSUMER_VECTOR_INDEX_TOPIC = "search-engine-csv-loader-topic"

app/search_vector/kafka_operate/kafka_operate.py

@@ -1,4 +1,5 @@
 """kafka operate"""
+import json
 from kafka import KafkaProducer, KafkaConsumer
 from kafka.errors import KafkaError
 from ..config.config import KAFKA_CLUSTER
@@ -75,7 +76,9 @@ def consume_messages_store_milvus(self, milvus_table):
             return
         print("Consuming messages...")
         for msg in self.consumer:
-            do_upload(milvus_table, 1, "mirror", msg, self.milvus_client)
+            data = json.loads(msg.value.decode('utf-8'))
+            do_upload(milvus_table, int(data["doc_id"]), data["title"],
+                      data["body"], self.milvus_client)
 
     def on_send_success(self, record_metadata):
         """

app/search_vector/milvus/operators.py

@@ -4,7 +4,8 @@
 from ..utils.logs import LOGGER
 
 
-def do_upload(table_name, doc_id, title, body, milvus_client):
+def do_upload(table_name: str, doc_id: int, title: str, body: str,
+              milvus_client):
     """
     upload info in milvus
 

main.py

@@ -185,5 +185,5 @@ def init_model():
 if __name__ == "__main__":
     # app.run(host=WEBSITE_HOST, port=WEBSITE_PORT, debug=True)
     # print("start server {}:{}".format(WEBSITE_HOST, WEBSITE_PORT))
-    consume_inverted_index()
+    # consume_inverted_index()
     asyncio.run(serve())