跨节点数据查询工程文件

import asyncio import logging import datetime import builtins import threading import x_com_python_lib class Context: request_id: int channel_id: int sender_service_key: x_com_python_lib.PyServiceKey # 解析参数 class MillisecondFormatter(logging.Formatter): def format(self, record): s = super().format(record) old_time = self.formatTime(record, self.datefmt) new_time = datetime.datetime.fromtimestamp(record.created).strftime('%Y-%m-%d %H:%M:%S.{:03d}'.format(int(record.msecs))) return s.replace(old_time, new_time) class XportLogHandler(logging.Handler): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) format_str = '%(asctime)s %(levelname)s {} %(filename)s %(lineno)d: %(message)s'.format(x_com_python_lib.service_name()) self.formatter = MillisecondFormatter(format_str) def emit(self, record): log_entry = self.format(record) x_com_python_lib.log.output(log_entry) def init(): TRACE_LEVEL_NUM = 5 logging.addLevelName(TRACE_LEVEL_NUM, "TRACE") def trace(self, message, *args, **kws): if self.isEnabledFor(TRACE_LEVEL_NUM): self._log(TRACE_LEVEL_NUM, message, args, **kws) # logging.Logger.trace = trace logger = logging.getLogger('xport') log_level = x_com_python_lib.log.level().upper() # if log_level == "TRACE": logger.setLevel(TRACE_LEVEL_NUM) elif log_level == "DEBUG": logger.setLevel(logging.DEBUG) elif log_level == "INFO": logger.setLevel(logging.INFO) elif log_level == "WARN": logger.setLevel(logging.WARNING) elif log_level == "ERROR": logger.setLevel(logging.ERROR) # xport_log_handler = XportLogHandler() logger.addHandler(xport_log_handler) builtins.logger = logger def getLogger() -> logging.Logger: return builtins.logger def check_stop_event(): if builtins.stop_event.is_set(): builtins.loop.stop() else: # 每秒检查一次 builtins.loop.call_later(1, check_stop_event) def run_loop_in_thread(): loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) builtins.loop = loop loop.call_soon(check_stop_event) loop.run_forever() def disptach_message(request_id, channel_id, sender_service_key, msg, param): builtins.loop.call_soon_threadsafe(lambda: builtins.loop.create_task(_inner_disptach_message(request_id, channel_id, sender_service_key, msg, param))) async def lanuch(): builtins.stop_event = threading.Event() thread = threading.Thread(target=run_loop_in_thread) thread.start() init() try: await x_com_python_lib.init() while True: await asyncio.sleep(10) except: builtins.stop_event.set() async def _inner_disptach_message(request_id, channel_id, sender_service_key, msg, param): ctx = Context() ctx.request_id = request_id ctx.channel_id = channel_id ctx.sender_service_key = sender_service_key try: if msg == "on_init": await builtins.SERVICE.on_init() x_com_python_lib.xport.resopnse_message(request_id, sender_service_key, msg, 0, "", None) elif msg == "TDayK": ret = await builtins.SERVICE.t_day_k(ctx, param) x_com_python_lib.xport.resopnse_message(request_id, sender_service_key, msg, 0, "", ret) elif msg == "TyMinK": ret = await builtins.SERVICE.ty_min_k(ctx, param) x_com_python_lib.xport.resopnse_message(request_id, sender_service_key, msg, 0, "", ret) elif msg == "on_finalize": await builtins.SERVICE.on_finalize() x_com_python_lib.xport.resopnse_message(request_id, sender_service_key, msg, 0, "", None) else: builtins.logger.warn("无该 {} 消息的处理函数".format(msg)) except Exception as e: x_com_python_lib.xport.resopnse_message(request_id, sender_service_key, msg, 1, str(e), None)