from concurrent import futures import time, json import grpc from utils.log_manager import bom_log from protobuf import classify_pb2 from protobuf import classify_pb2_grpc from predic_fac import PredictorFac from utils.status import code2msg class Classify(classify_pb2_grpc.classifyServicer): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.predictorfac = PredictorFac() def Classify(self, request, context): bom_log.info(f'grpc收到数据: {request.keyword}') try: dic_data = json.loads(request.keyword) code, res = self.predictorfac.predict(dic_data) except json.decoder.JSONDecodeError: code, res = self.predictorfac.predict(request.keyword) if code != 1: res = code2msg(code) bom_log.error(f'grpc识别报错: ' + res['message']) else: res['status'] = code return classify_pb2.ClassifyReply(message=json.dumps(res)) def fac_test_predic(self, data): res = self.predictorfac.predict(data) return res def grpc_server(port): # 启动 rpc 服务 server = grpc.server(futures.ThreadPoolExecutor(max_workers=10)) classify_pb2_grpc.add_classifyServicer_to_server(Classify(), server) server.add_insecure_port(f'[::]:{port}') server.start() try: while True: time.sleep(60 * 60 * 24) # one day in seconds except KeyboardInterrupt: server.stop(0) if __name__ == '__main__': port = 50051 grpc_server(port) # data = read_from_excel('DZ0901_V1.4_BOM.xlsx', 'DZ0901_V1.3BOM清单') # print(data) # print(Classify().fac_test_predic(data)) # for i in range(65, 91): # print(chr(i))