python和websocket構(gòu)建實時日志跟蹤器的步驟

with open(file_path) as f:
      # 首次讀取指定行數(shù)（NUM_LINES）的日志文件，發(fā)送給客戶端
            content = ''.join(deque(f, NUM_LINES))
            content = conv.convert(content, full=False)
            await websocket.send(content)
      
      # 如果發(fā)現(xiàn)客戶端有 tail 請求，則進行 tail 日志追蹤
            if tail:
        # 首先創(chuàng)建發(fā)起這次請求的心跳時間
                last_heartbeat = time.time()
                while True:
          # 每次 tail 服務(wù)端最新的日志記錄，返回給客戶端
                    content = f.read()
                    if content:
                        content = conv.convert(content, full=False)
                        await websocket.send(content)
                    else:
                        await asyncio.sleep(1)

                    # 檢測這次請求距上一次發(fā)起請求，是不是已經(jīng)超過了最長心跳檢測時長，如果是，發(fā)起心跳檢測
                    if time.time() - last_heartbeat > HEARTBEAT_INTERVAL:
                        try:
                            await websocket.send('ping')
                            pong = await asyncio.wait_for(websocket.recv(), 5)
                            logger.info(f"pong:{pong}")
                            if pong != 'pong':
                                raise Exception()
                        except Exception:
                            raise Exception('Ping error')
                        else:
                            last_heartbeat = time.time()
            else:
                await websocket.close()

async def consumer_handler(websocket: WebSocketClientProtocol) -> None:
    async for message in websocket:
        log_message(message)
        if message == "ping":
            await websocket.send("pong")

async def cousume(hostname: str, port: int, log_file: str, tail:bool=True) -> None:
    websocket_resource_url = f"ws://{hostname}:{port}{log_file}"
    if tail:
        websocket_resource_url = f"{websocket_resource_url}?tail=1"
    async with websockets.connect(websocket_resource_url) as websocket:
        await consumer_handler(websocket)

def log_message(message: str) -> None:
    logger.info(f"Message: {message}")

import os

from loguru import logger

class LoggerExtend(object):
    # 存放目錄名稱
    folder = '../logs'

    def __init__(self, filename, folder=None):

        self.folder = folder or self.folder

        if not os.path.exists(self.folder):
            os.mkdir(self.folder)

        self.file = self.folder + '/' + filename

        logger.add(self.file, rotation="100 MB")

    @property
    def get_logger(self):
        return logger

if __name__ == '__main__':
    logger = LoggerExtend(os.path.basename(__file__).replace(".py", ".log")).get_logger
    import time
    while True:

        logger.info("你好aaa")