67 lines
3.0 KiB
Python
67 lines
3.0 KiB
Python
|
import re
|
||
|
|
import traceback
|
||
|
|
from loguru import logger
|
||
|
|
|
||
|
|
class DrGraphSession:
|
||
|
|
def __init__(self, stepIndex: int, msg: str, session_id: str):
|
||
|
|
logger.info(f"DrGraphSession.__init__: stepIndex={stepIndex}, msg={msg}, session_id={session_id}")
|
||
|
|
self.stepIndex = stepIndex
|
||
|
|
self.session_id = session_id
|
||
|
|
|
||
|
|
match = re.search(r";(-\d+)", msg);
|
||
|
|
level = -3
|
||
|
|
if match:
|
||
|
|
level = int(match.group(1))
|
||
|
|
value = value.replace(f";{level}", "")
|
||
|
|
level = -3 + level
|
||
|
|
|
||
|
|
if "警告" in value or value.startswith("WARNING"):
|
||
|
|
self.log_warning(f"第 {self.stepIndex} 步 - {value}", level = level)
|
||
|
|
elif "异常" in value or value.startswith("EXCEPTION"):
|
||
|
|
self.log_exception(f"第 {self.stepIndex} 步 - {value}", level = level)
|
||
|
|
elif "成功" in value or value.startswith("SUCCESS"):
|
||
|
|
self.log_success(f"第 {self.stepIndex} 步 - {value}", level = level)
|
||
|
|
elif "开始" in value or value.startswith("START"):
|
||
|
|
self.log_success(f"第 {self.stepIndex} 步 - {value}", level = level)
|
||
|
|
elif "失败" in value or value.startswith("ERROR"):
|
||
|
|
self.log_error(f"第 {self.stepIndex} 步 - {value}", level = level)
|
||
|
|
else:
|
||
|
|
self.log_info(f"第 {self.stepIndex} 步 - {value}", level = level)
|
||
|
|
|
||
|
|
def log_prefix(self) -> str:
|
||
|
|
"""Get log prefix with session ID and desc."""
|
||
|
|
return f"〖Session{self.session_id}〗"
|
||
|
|
|
||
|
|
def parse_source_pos(self, level: int):
|
||
|
|
pos = (traceback.format_stack())[level - 1].strip().split('\n')[0]
|
||
|
|
match = re.search(r"File \"(.+?)\", line (\d+), in (\w+)", pos);
|
||
|
|
if match:
|
||
|
|
file = match.group(1).replace("F:\\DrGraph_Python\\FastAPI\\", "")
|
||
|
|
pos = f"{file}:{match.group(2)} in {match.group(3)}"
|
||
|
|
return pos
|
||
|
|
|
||
|
|
def log_info(self, msg: str, level: int = -2):
|
||
|
|
"""Log info message with session ID."""
|
||
|
|
pos = self.parse_source_pos(level)
|
||
|
|
logger.info(f"{self.log_prefix()} {msg} >>> @ {pos}")
|
||
|
|
|
||
|
|
def log_success(self, msg: str, level: int = -2):
|
||
|
|
"""Log success message with session ID."""
|
||
|
|
pos = self.parse_source_pos(level)
|
||
|
|
logger.success(f"{self.log_prefix()} {msg} >>> @ {pos}")
|
||
|
|
|
||
|
|
def log_warning(self, msg: str, level: int = -2):
|
||
|
|
"""Log warning message with session ID."""
|
||
|
|
pos = self.parse_source_pos(level)
|
||
|
|
logger.warning(f"{self.log_prefix()} {msg} >>> @ {pos}")
|
||
|
|
|
||
|
|
def log_error(self, msg: str, level: int = -2):
|
||
|
|
"""Log error message with session ID."""
|
||
|
|
pos = self.parse_source_pos(level)
|
||
|
|
logger.error(f"{self.log_prefix()} {msg} >>> @ {pos}")
|
||
|
|
|
||
|
|
def log_exception(self, msg: str, level: int = -2):
|
||
|
|
"""Log exception message with session ID."""
|
||
|
|
pos = self.parse_source_pos(level)
|
||
|
|
logger.exception(f"{self.log_prefix()} {msg} >>> @ {pos}")
|