#!/usr/bin/python # -*- coding: GBK -*- """ ·â°üÈÕ־ģ¿é ÌṩͳһµÄÈÕÖ¾¼Ç¼¹¦ÄÜ,Ö§³ÖÊ®Áù½øÖƸñʽÊä³ö """ import os import datetime from config import ConfigReader class PacketLogger: """·â°üÈÕÖ¾¼Ç¼Æ÷µ¥Àý""" _instance = None def __new__(cls): if cls._instance is None: cls._instance = super(PacketLogger, cls).__new__(cls) cls._instance._initialized = False return cls._instance @classmethod def instance(cls): """»ñÈ¡µ¥ÀýʵÀý""" if cls._instance is None: cls._instance = cls() return cls._instance def __init__(self): if hasattr(self, '_initialized') and self._initialized: return self._initialized = True # ¶ÁÈ¡ÅäÖã¨Ê¹Óõ¥Àý£© config = ConfigReader.instance() self.enable = config.get_enable_packet_log() self.log_path = config.get_packet_log_path() # ÈÕÖ¾Ïà¹ØÊôÐÔ self.log_dir = None self.log_file = None # ³õʼ»¯ÈÕÖ¾ if self.enable: self._init() def _init(self): """³õʼ»¯ÈÕ־ϵͳ""" try: # ´´½¨»ù´¡Â·¾¶ if not os.path.exists(self.log_path): os.makedirs(self.log_path) # ´´½¨ÈÕÆÚ×ÓĿ¼: C:\ServerLog\2026-02\EventServer today = datetime.datetime.now() date_path = today.strftime('%Y-%m') self.log_dir = os.path.join(self.log_path, date_path, 'EventServer') if not os.path.exists(self.log_dir): os.makedirs(self.log_dir) # ÉèÖÃÈÕÖ¾ÎļþÃû log_filename = 'packet_%s.log' % today.strftime('%Y%m%d') self.log_file = os.path.join(self.log_dir, log_filename) except Exception as e: print('[PacketLogger] Init error: %s' % str(e)) self.enable = False def _check_date_change(self): """¼ì²éÈÕÆÚ±ä»¯,±ØÒªÊ±Çл»ÈÕÖ¾Îļþ""" if not self.enable: return today = datetime.datetime.now() log_filename = 'packet_%s.log' % today.strftime('%Y%m%d') current_log_file = os.path.join(self.log_dir, log_filename) # ÈÕÆÚ±ä»¯Ê±ÖØÐ³õʼ»¯ if current_log_file != self.log_file: self._init() def log(self, cid, packet_data, packet_type, direction='RECV'): """ ¼Ç¼·â°üÈÕÖ¾ Args: cid: ¿Í»§¶ËID packet_data: ·â°üÊý¾Ý(×Ö·û´®) packet_type: ·â°üÀàÐÍ(Èç LOGIN, EVENT, HEARTBEATµÈ) direction: ·½Ïò(RECV=½ÓÊÕ, SEND=·¢ËÍ) """ return if not self.enable: return try: # ¼ì²éÈÕÆÚ±ä»¯ self._check_date_change() # Éú³Éʱ¼ä´Á(¾«È·µ½ºÁÃë) today = datetime.datetime.now() timestamp = today.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3] # ת»»ÎªÊ®Áù½øÖÆ×Ö·û´® hex_data = ' '.join(['%02X' % ord(b) if isinstance(b, (str, unicode)) else '%02X' % b for b in packet_data]) # Éú³ÉÈÕÖ¾ÐÐ log_line = '[%s] [%s] CID=%d Type=%s Len=%d Data=%s\n' % ( timestamp, direction, cid, packet_type, len(packet_data), hex_data ) # дÈëÎļþ with open(self.log_file, 'a') as f: f.write(log_line) except Exception as e: print('[PacketLogger] Write error: %s' % str(e)) def log_text(self, text, prefix='INFO'): """ ¼Ç¼Îı¾ÈÕÖ¾ Args: text: ÈÕÖ¾Îı¾ prefix: ÈÕ־ǰ׺(INFO, WARN, ERRORµÈ) """ if not self.enable: return try: # ¼ì²éÈÕÆÚ±ä»¯ self._check_date_change() # Éú³Éʱ¼ä´Á today = datetime.datetime.now() timestamp = today.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3] # Éú³ÉÈÕÖ¾ÐÐ log_line = '[%s] [%s] %s\n' % (timestamp, prefix, text) # дÈëÎļþ with open(self.log_file, 'a') as f: f.write(log_line) except Exception as e: print('[PacketLogger] Write error: %s' % str(e)) # È«¾ÖÈÕ־ʵÀý packet_logger = PacketLogger()