mirror of
https://github.com/Nanaloveyuki/py-logiliteal.git
synced 2025-10-19 16:46:23 +00:00
Compare commits
10 Commits
Author | SHA1 | Date | |
---|---|---|---|
|
5a7756b793 | ||
|
b83984d58a | ||
|
652b6d6d2a | ||
|
9bc021d991 | ||
|
d3d9613acc | ||
|
388b1c2b75 | ||
|
563cecac97 | ||
|
827658beab | ||
|
9ebd08e5e2 | ||
|
3291f19a25 |
4
.gitignore
vendored
4
.gitignore
vendored
@@ -1,5 +1,7 @@
|
||||
# Custom
|
||||
*.backup.json
|
||||
*_backup.json
|
||||
auto_upload.bat
|
||||
auto_upload.sh
|
||||
|
||||
# Byte-compiled / optimized / DLL files
|
||||
__pycache__/
|
||||
|
10
README.md
10
README.md
@@ -8,7 +8,7 @@
|
||||
py-logiliteal 允许嵌入到其他项目中, 并根据需要自定义日志记录器
|
||||
同时也支持pip安装
|
||||
```bash
|
||||
pip install py-logiliteal
|
||||
pip install logiliteal
|
||||
```
|
||||
|
||||
**支持高可扩展的样式**
|
||||
@@ -28,7 +28,7 @@ pip install py-logiliteal
|
||||
## 安装
|
||||
暂无安装包, 请使用release发布版或直接clone代码到本地/使用pip安装
|
||||
```bash
|
||||
pip install py-logiliteal
|
||||
pip install logiliteal
|
||||
```
|
||||
|
||||
## 文档
|
||||
@@ -37,8 +37,8 @@ pip install py-logiliteal
|
||||
## 示例
|
||||
```python
|
||||
# 导入
|
||||
from py_logiliteal import Logger
|
||||
# 或 import py_logiliteal(不推荐)
|
||||
from logiliteal import Logger
|
||||
# 或 import logiliteal(不推荐)
|
||||
|
||||
# 实例化
|
||||
logger = Logger()
|
||||
@@ -51,7 +51,7 @@ logger.warn("这是一条带有前缀的警告日志", prefix="114514")
|
||||
logger.critical("这是一条带有前缀并且日志等级不同的严重错误日志", prefix="114514", level=55)
|
||||
|
||||
# 自定义配置
|
||||
from py_logiliteal import set_config, get_config
|
||||
from logiliteal import set_config, get_config
|
||||
# 读取配置
|
||||
print(get_config("console_format"))
|
||||
# 默认会输出时间、日志等级、日志前缀、日志消息
|
||||
|
33
logger_config.json
Normal file
33
logger_config.json
Normal file
@@ -0,0 +1,33 @@
|
||||
{
|
||||
"file_level": "DEBUG",
|
||||
"file_name": "latest.log",
|
||||
"file_path": "./logs",
|
||||
"file_format": "{asctime} {levelname} | {prefix}{message}",
|
||||
"file_encoding": "utf-8",
|
||||
"enable_console": true,
|
||||
"enable_file": true,
|
||||
"console_color": true,
|
||||
"console_level": "INFO",
|
||||
"console_format": "{time} {levelname} | {prefix}{message}",
|
||||
"console_prefix": "Auto",
|
||||
"console_encoding": "utf-8",
|
||||
"asctime_format": "%Y-%m-%d %H:%M:%S",
|
||||
"time_format": "%H:%M:%S",
|
||||
"date_format": "%Y-%m-%d",
|
||||
"weekday_format": "%A",
|
||||
"level_name": {
|
||||
"DEBUG": "DEBUG",
|
||||
"INFO": "INFO",
|
||||
"WARN": "WARN",
|
||||
"ERRO": "ERRO",
|
||||
"CRIT": "CRIT"
|
||||
},
|
||||
"level_color": {
|
||||
"DEBUG": "#c1d5ff",
|
||||
"INFO": "#c1ffff",
|
||||
"WARN": "#fff600",
|
||||
"ERRO": "#ffa000",
|
||||
"CRIT": "#ff8181"
|
||||
},
|
||||
"time_color": "#28ffb6"
|
||||
}
|
2
setup.py
2
setup.py
@@ -6,7 +6,7 @@ long_description = (here / "README.md").read_text(encoding="utf-8")
|
||||
|
||||
setup(
|
||||
name="logiliteal",
|
||||
version="0.1.0",
|
||||
version="0.1.1",
|
||||
description="简洁,高扩展性,可自定义的日志库 / Simple, high extensibility, and customizable logging library",
|
||||
long_description=long_description,
|
||||
long_description_content_type="text/markdown",
|
||||
|
@@ -5,7 +5,10 @@ py-logiliteal's config settings, used to set py-logiliteal's global config
|
||||
# encoding = utf-8
|
||||
# python 3.13.5
|
||||
|
||||
from .utils import get_config, set_config, reset_config, get_asctime, get_date, get_time, get_weekday, fmt_console, fmt_placeholder, fmt_message, fmt_level_name, set_style
|
||||
from .utils import get_config, set_config, reset_config, create_backup
|
||||
from .utils import get_asctime, get_date, get_time, get_weekday
|
||||
from .utils import fmt_console, fmt_placeholder, fmt_message, fmt_level_name
|
||||
from .utils import set_style, set_color, set_bg_color
|
||||
from .levels import Logger
|
||||
|
||||
__all__ = [
|
||||
@@ -23,5 +26,6 @@ __all__ = [
|
||||
"set_style",
|
||||
"set_color",
|
||||
"set_bg_color",
|
||||
"create_backup",
|
||||
"Logger" # 日志记录器非实例化
|
||||
]
|
@@ -35,94 +35,70 @@ class Logger:
|
||||
rename(_get_full_path(file_path, file_name), _get_full_path(file_path, f"{get_asctime().replace(':', '-')}.log"))
|
||||
self.debug("日志文件已存在,已自动重命名")
|
||||
|
||||
@staticmethod
|
||||
def info(message: Any, prefix: str | None = None, level: int = 20) -> Optional[str]:
|
||||
"""
|
||||
信息日志
|
||||
Info log
|
||||
:param message: 消息内容 Message content
|
||||
:param prefix: 前缀 Prefix
|
||||
:param level: 日志级别 Log level(11~20)
|
||||
"""
|
||||
if level < 11 or level > 20:
|
||||
return None
|
||||
def _log(self, msg, pf, lvn):
|
||||
if is_enable_file:
|
||||
with open(_get_full_path(file_path, file_name), "a", encoding=file_encoding) as f:
|
||||
f.write(fmt_file(level, fmt_message(message, no_placeholder=True), prefix))
|
||||
f.write(fmt_file(lvn, fmt_message(msg, no_placeholder=True), pf))
|
||||
if is_enable_console:
|
||||
print(fmt_console(level, fmt_message(message, no_placeholder=True), prefix))
|
||||
return fmt_console(level, fmt_message(message, no_placeholder=True), prefix)
|
||||
print(fmt_console(lvn, fmt_message(msg, no_placeholder=True), pf))
|
||||
return fmt_console(lvn, fmt_message(msg, no_placeholder=True), pf)
|
||||
|
||||
@staticmethod
|
||||
def debug(message: Any, prefix: str | None = None, level: int = 10) -> Optional[str]:
|
||||
def debug(self, message: Any, prefix: str | None = None, level: int = 0) -> Optional[str]:
|
||||
"""
|
||||
调试日志
|
||||
Debug log
|
||||
:param message: 消息内容 Message content
|
||||
:param prefix: 前缀 Prefix
|
||||
:param level: 日志级别 Log level(0~10)
|
||||
:param level: 日志级别 Log level(0~9)
|
||||
"""
|
||||
if level < 0 or level > 10:
|
||||
return None
|
||||
if is_enable_file:
|
||||
with open(_get_full_path(file_path, file_name), "a", encoding=file_encoding) as f:
|
||||
f.write(fmt_file(level, fmt_message(message, no_placeholder=True), prefix))
|
||||
if is_enable_console:
|
||||
print(fmt_console(level, fmt_message(message, no_placeholder=True), prefix))
|
||||
return fmt_console(level, fmt_message(message, no_placeholder=True), prefix)
|
||||
return self._log(message, prefix, level)
|
||||
|
||||
@staticmethod
|
||||
def warn(message: Any, prefix: str | None = None, level: int = 31) -> Optional[str]:
|
||||
def info(self, message: Any, prefix: str | None = None, level: int = 10) -> Optional[str]:
|
||||
"""
|
||||
信息日志
|
||||
Info log
|
||||
:param message: 消息内容 Message content
|
||||
:param prefix: 前缀 Prefix
|
||||
:param level: 日志级别 Log level(10~19)
|
||||
"""
|
||||
return self._log(message, prefix, level)
|
||||
|
||||
def warn(self, message: Any, prefix: str | None = None, level: int = 20) -> Optional[str]:
|
||||
"""
|
||||
警告日志
|
||||
Warn log
|
||||
:param message: 消息内容 Message content
|
||||
:param prefix: 前缀 Prefix
|
||||
:param level: 日志级别 Log level(31~40)
|
||||
:param level: 日志级别 Log level(20~29)
|
||||
"""
|
||||
if level < 30 or level > 40:
|
||||
return None
|
||||
if is_enable_file:
|
||||
with open(_get_full_path(file_path, file_name), "a", encoding=file_encoding) as f:
|
||||
f.write(fmt_file(level, fmt_message(message, no_placeholder=True), prefix))
|
||||
if is_enable_console:
|
||||
print(fmt_console(level, fmt_message(message, no_placeholder=True), prefix))
|
||||
return fmt_console(level, fmt_message(message, no_placeholder=True), prefix)
|
||||
return self._log(message, prefix, level)
|
||||
|
||||
@staticmethod
|
||||
def error(message: Any, prefix: str | None = None, level: int = 41) -> Optional[str]:
|
||||
def error(self, message: Any, prefix: str | None = None, level: int = 30) -> Optional[str]:
|
||||
"""
|
||||
错误日志
|
||||
Error log
|
||||
:param message: 消息内容 Message content
|
||||
:param prefix: 前缀 Prefix
|
||||
:param level: 日志级别 Log level(41~50)
|
||||
:param level: 日志级别 Log level(30~39)
|
||||
"""
|
||||
if level < 40 or level > 50:
|
||||
return None
|
||||
if is_enable_file:
|
||||
with open(_get_full_path(file_path, file_name), "a", encoding=file_encoding) as f:
|
||||
f.write(fmt_file(level, fmt_message(message, no_placeholder=True), prefix))
|
||||
if is_enable_console:
|
||||
print(fmt_console(level, fmt_message(message, no_placeholder=True), prefix))
|
||||
return fmt_console(level, fmt_message(message, no_placeholder=True), prefix)
|
||||
return self._log(message, prefix, level)
|
||||
|
||||
@staticmethod
|
||||
def critical(message: Any, prefix: str | None = None, level: int = 51) -> Optional[str]:
|
||||
def critical(self, message: Any, prefix: str | None = None, level: int = 40) -> Optional[str]:
|
||||
"""
|
||||
严重错误日志
|
||||
Critical error log
|
||||
:param message: 消息内容 Message content
|
||||
:param prefix: 前缀 Prefix
|
||||
:param level: 日志级别 Log level(51~60)
|
||||
:param level: 日志级别 Log level(40~49)
|
||||
"""
|
||||
if level < 50 or level > 60:
|
||||
return None
|
||||
if is_enable_file:
|
||||
with open(_get_full_path(file_path, file_name), "a", encoding=file_encoding) as f:
|
||||
f.write(fmt_file(level, fmt_message(message, no_placeholder=True), prefix))
|
||||
if is_enable_console:
|
||||
print(fmt_console(level, fmt_message(message, no_placeholder=True), prefix))
|
||||
return fmt_console(level, fmt_message(message, no_placeholder=True), prefix)
|
||||
|
||||
return self._log(message, prefix, level)
|
||||
|
||||
def log(self, message: Any, prefix: str | None = None, level: int = 50) -> Optional[str]:
|
||||
"""
|
||||
自定义日志
|
||||
Custom log
|
||||
:param message: 消息内容 Message content
|
||||
:param prefix: 前缀 Prefix
|
||||
:param level: 日志级别 Log level(50~59...)
|
||||
"""
|
||||
return self._log(message, prefix, level)
|
@@ -5,7 +5,7 @@ Utility functions
|
||||
# encoding = utf-8
|
||||
# python 3.13.5
|
||||
|
||||
from .configs import get_config, set_config, reset_config
|
||||
from .configs import get_config, set_config, reset_config, create_backup
|
||||
from .time import get_asctime, get_date, get_time, get_weekday
|
||||
from .fmt import fmt_console, fmt_placeholder, fmt_message, fmt_level_name
|
||||
from .styles import set_color, set_bg_color, set_style
|
||||
@@ -14,6 +14,7 @@ __all__ = [
|
||||
"get_config",
|
||||
"set_config",
|
||||
"reset_config",
|
||||
"create_backup",
|
||||
"get_asctime",
|
||||
"get_date",
|
||||
"get_time",
|
||||
@@ -24,5 +25,5 @@ __all__ = [
|
||||
"fmt_level_name",
|
||||
"set_color",
|
||||
"set_bg_color",
|
||||
"set_style"
|
||||
"set_style",
|
||||
]
|
||||
|
@@ -6,7 +6,14 @@ py-logiliteal's config settings, used to set py-logiliteal's global config
|
||||
# encoding = utf-8
|
||||
# python 3.13.5
|
||||
|
||||
DEFAULT_CONFIG_PATH = "config.json"
|
||||
import json
|
||||
from os import remove
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from typing import Union, Optional, Tuple
|
||||
from logging import error
|
||||
|
||||
DEFAULT_CONFIG_PATH = "logger_config.json"
|
||||
DEFAULT_CONFIG = {
|
||||
"file_level": "DEBUG",
|
||||
"file_name": "latest.log",
|
||||
@@ -17,17 +24,59 @@ DEFAULT_CONFIG = {
|
||||
"enable_file": True,
|
||||
"console_color": True,
|
||||
"console_level": "INFO",
|
||||
"console_format": "{asctime} {levelname} | {prefix}{message}",
|
||||
"console_format": "{time} {levelname} | {prefix}{message}",
|
||||
"console_prefix": "Auto",
|
||||
"console_encoding": "utf-8",
|
||||
"date_format": "%Y-%m-%d %H:%M:%S",
|
||||
"asctime_format": "%Y-%m-%d %H:%M:%S",
|
||||
"time_format": "%H:%M:%S",
|
||||
"date_format": "%Y-%m-%d",
|
||||
"weekday_format": "%A",
|
||||
"level_name": {"DEBUG": "DEBUG", "INFO": "INFO", "WARN": "WARN", "ERRO": "ERRO", "CRIT": "CRIT"},
|
||||
"level_color": {"DEBUG": "#c1d5ff", "INFO": "#c1ffff", "WARN": "#fff600", "ERRO": "#ffa000", "CRIT": "#ff8181"},
|
||||
"time_color": "#28ffb6",
|
||||
}
|
||||
|
||||
from typing import Union, Optional
|
||||
from logging import error, info
|
||||
import json
|
||||
g_config_cache = None
|
||||
g_config_mtime = 0
|
||||
|
||||
def create_backup(config_path: Path) -> Tuple[bool, str]:
|
||||
"""
|
||||
创建配置文件备份
|
||||
:param config_path: 配置文件路径
|
||||
:param backup_prefix: 备份文件前缀
|
||||
:return: (是否成功, 备份路径或错误信息)
|
||||
"""
|
||||
try:
|
||||
if not config_path.exists():
|
||||
return False, f"配置文件不存在: {config_path}"
|
||||
backup_path = config_path.parent / f"logger_config_backup.json"
|
||||
|
||||
if backup_path.exists():
|
||||
remove(backup_path)
|
||||
shutil.copy2(config_path, backup_path)
|
||||
return True, str(backup_path)
|
||||
except PermissionError:
|
||||
return False, f"权限不足,无法创建备份: {config_path}"
|
||||
except Exception as e:
|
||||
return False, f"备份失败: {str(e)}"
|
||||
|
||||
def handle_config_exceptions(func):
|
||||
"""
|
||||
配置操作异常处理装饰器
|
||||
"""
|
||||
def wrapper(*args, **kwargs):
|
||||
try:
|
||||
return func(*args, **kwargs)
|
||||
except json.JSONDecodeError as e:
|
||||
error(f"配置文件格式错误: {e}")
|
||||
return False, f"配置文件格式错误: {str(e)}"
|
||||
except PermissionError:
|
||||
error(f"没有权限操作配置文件: {DEFAULT_CONFIG_PATH}")
|
||||
return False, "没有权限操作配置文件"
|
||||
except Exception as e:
|
||||
error(f"配置操作失败: {e}")
|
||||
return False, f"配置操作失败: {str(e)}"
|
||||
return wrapper
|
||||
|
||||
def get_config(select: str = None) -> Union[dict, str, bool, int, None]:
|
||||
"""
|
||||
@@ -35,27 +84,24 @@ def get_config(select: str = None) -> Union[dict, str, bool, int, None]:
|
||||
:param select: 配置项名称 Config item name
|
||||
:return: 配置项值 Config item value
|
||||
"""
|
||||
try:
|
||||
with open(DEFAULT_CONFIG_PATH, "r", encoding="utf-8") as f:
|
||||
config = json.load(f)
|
||||
if select:
|
||||
return config[select]
|
||||
return config
|
||||
except (FileNotFoundError, json.JSONDecodeError):
|
||||
with open(DEFAULT_CONFIG_PATH, "w", encoding="utf-8") as f:
|
||||
json.dump(DEFAULT_CONFIG, f, indent=4)
|
||||
if select:
|
||||
return DEFAULT_CONFIG[select]
|
||||
return DEFAULT_CONFIG
|
||||
except KeyError as e:
|
||||
error(f"配置项 '{select}' 不存在")
|
||||
return None
|
||||
except Exception as e:
|
||||
error(f"读取配置文件失败: {e}")
|
||||
return None
|
||||
global g_config_cache, g_config_mtime
|
||||
config_path = Path(DEFAULT_CONFIG_PATH)
|
||||
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
if config_path.exists():
|
||||
current_mtime = config_path.stat().st_mtime
|
||||
if current_mtime != g_config_mtime or g_config_cache is None:
|
||||
with open(config_path, "r", encoding="utf-8") as f:
|
||||
g_config_cache = json.load(f)
|
||||
g_config_mtime = current_mtime
|
||||
else:
|
||||
with open(config_path, "w", encoding="utf-8") as f:
|
||||
json.dump(DEFAULT_CONFIG, f, indent=4)
|
||||
g_config_cache = DEFAULT_CONFIG
|
||||
g_config_mtime = config_path.stat().st_mtime
|
||||
|
||||
if select:
|
||||
return g_config_cache.get(select)
|
||||
return g_config_cache
|
||||
|
||||
def set_config(select: str, value: Union[dict, str, bool, int, None]) -> tuple[bool, Optional[str]]:
|
||||
"""
|
||||
@@ -64,75 +110,49 @@ def set_config(select: str, value: Union[dict, str, bool, int, None]) -> tuple[b
|
||||
:param value: 配置项值 Config item value
|
||||
:return: (设置是否成功, 消息) (Set success or not, message)
|
||||
"""
|
||||
try:
|
||||
config_path = Path(DEFAULT_CONFIG_PATH)
|
||||
backup_path = Path(__file__).parent / "configs.backup.json"
|
||||
config_path = Path(DEFAULT_CONFIG_PATH)
|
||||
|
||||
if not config_path.exists():
|
||||
with open(config_path, "w", encoding="utf-8") as f:
|
||||
json.dump(DEFAULT_CONFIG, f, indent=4)
|
||||
if not config_path.exists():
|
||||
with open(config_path, "w", encoding="utf-8") as f:
|
||||
json.dump(DEFAULT_CONFIG, f, indent=4)
|
||||
|
||||
shutil.copy2(config_path, backup_path)
|
||||
backup_success, backup_info = create_backup(config_path)
|
||||
if not backup_success:
|
||||
return False, f"备份失败: {backup_info}"
|
||||
|
||||
with open(DEFAULT_CONFIG_PATH, "r+", encoding="utf-8") as f:
|
||||
config = json.load(f)
|
||||
config[select] = value
|
||||
f.seek(0)
|
||||
json.dump(config, f, indent=4)
|
||||
f.truncate()
|
||||
with open(config_path, "r+", encoding="utf-8") as f:
|
||||
config = json.load(f)
|
||||
config[select] = value
|
||||
f.seek(0)
|
||||
json.dump(config, f, indent=4)
|
||||
f.truncate()
|
||||
|
||||
with open(DEFAULT_CONFIG_PATH, "r", encoding="utf-8") as f:
|
||||
verify_config = json.load(f)
|
||||
if verify_config.get(select) != value:
|
||||
error(f"配置项 '{select}' 设置失败,值为 {value}")
|
||||
shutil.move(backup_path, DEFAULT_CONFIG_PATH)
|
||||
return False, f"配置项 '{select}' 设置失败,已恢复备份"
|
||||
with open(config_path, "r", encoding="utf-8") as f:
|
||||
verify_config = json.load(f)
|
||||
if verify_config.get(select) != value:
|
||||
shutil.move(backup_info, config_path)
|
||||
return False, f"配置项 '{select}' 设置失败,已恢复备份"
|
||||
|
||||
Path(backup_path).unlink(missing_ok=True)
|
||||
return True, f"配置项 '{select}' 设置成功"
|
||||
global g_config_cache
|
||||
g_config_cache = None
|
||||
return True, f"配置项 '{select}' 设置成功"
|
||||
|
||||
except json.JSONDecodeError as e:
|
||||
error(f"配置文件格式错误: {e}")
|
||||
return False, f"配置文件格式错误: {str(e)}"
|
||||
except PermissionError:
|
||||
error(f"没有权限操作配置文件: {DEFAULT_CONFIG_PATH}")
|
||||
return False, "没有权限操作配置文件"
|
||||
except Exception as e:
|
||||
error(f"设置配置文件失败: {e}")
|
||||
|
||||
if Path(backup_path).exists():
|
||||
shutil.move(backup_path, DEFAULT_CONFIG_PATH)
|
||||
return False, f"设置失败,已恢复备份: {str(e)}"
|
||||
return False, f"设置配置文件失败: {str(e)}"
|
||||
|
||||
@staticmethod
|
||||
@handle_config_exceptions
|
||||
def reset_config() -> tuple[bool, Optional[str]]:
|
||||
"""
|
||||
重置配置信息 Reset config info
|
||||
:return: (重置是否成功, 消息) (Reset success or not, message)
|
||||
"""
|
||||
try:
|
||||
config_path = Path(DEFAULT_CONFIG_PATH)
|
||||
config_path = Path(DEFAULT_CONFIG_PATH)
|
||||
|
||||
if not config_path.exists():
|
||||
with open(config_path, "w", encoding="utf-8") as f:
|
||||
json.dump(DEFAULT_CONFIG, f, indent=4)
|
||||
return True, "配置文件不存在,已创建默认配置"
|
||||
if config_path.exists():
|
||||
backup_success, backup_info = create_backup(config_path)
|
||||
if not backup_success:
|
||||
return False, f"备份失败: {backup_info}"
|
||||
|
||||
from .time import get_asctime
|
||||
timestamp = get_asctime()
|
||||
backup_path = f"{DEFAULT_CONFIG_PATH}_{timestamp}.backup.json"
|
||||
backup_path = backup_path.replace(":", "-")
|
||||
shutil.copy2(DEFAULT_CONFIG_PATH, backup_path)
|
||||
with open(DEFAULT_CONFIG_PATH, "w", encoding="utf-8") as f:
|
||||
json.dump(DEFAULT_CONFIG, f, indent=4)
|
||||
return True
|
||||
except PermissionError:
|
||||
error(f"权限不足,无法操作配置文件: {DEFAULT_CONFIG_PATH}")
|
||||
return False, "权限不足,无法重置配置"
|
||||
except json.JSONDecodeError:
|
||||
error("配置文件格式错误,无法解析")
|
||||
return False, "配置文件格式错误,无法重置"
|
||||
except Exception as e:
|
||||
error(f"重置配置文件失败: {e}")
|
||||
return False, f"重置配置失败: {str(e)}"
|
||||
with open(config_path, "w", encoding="utf-8") as f:
|
||||
json.dump(DEFAULT_CONFIG, f, indent=4)
|
||||
|
||||
global g_config_cache
|
||||
g_config_cache = None
|
||||
return True, "配置已重置为默认值"
|
||||
|
@@ -7,11 +7,13 @@ py-logiliteal's formatter, used to format log output
|
||||
# python 3.13.5
|
||||
|
||||
from .configs import get_config
|
||||
from typing import Any, Optional, Union
|
||||
from typing import Any, Optional
|
||||
from .time import get_asctime, get_time, get_weekday, get_date
|
||||
from .styles import set_color, set_bg_color
|
||||
import re
|
||||
|
||||
time_color = get_config("time_color")
|
||||
|
||||
def fmt_level(level: str) -> int:
|
||||
"""
|
||||
格式化日志级别
|
||||
@@ -20,15 +22,14 @@ def fmt_level(level: str) -> int:
|
||||
:return: 格式化后的日志级别 Formatted log level
|
||||
"""
|
||||
level_map = {
|
||||
"DEBUG": 10,
|
||||
"INFO": 20,
|
||||
"WARN": 30,
|
||||
"ERRO": 40,
|
||||
"CRIT": 50,
|
||||
"DEBUG": 0,
|
||||
"INFO": 10,
|
||||
"WARN": 20,
|
||||
"ERRO": 30,
|
||||
"CRIT": 40,
|
||||
"UNKN": 50
|
||||
}
|
||||
if level == "UNKN":
|
||||
return -1
|
||||
return level_map.get(level.upper(), 0)
|
||||
return level_map.get(level.upper(), 50)
|
||||
|
||||
def fmt_level_number(level: int) -> str:
|
||||
"""
|
||||
@@ -37,15 +38,15 @@ def fmt_level_number(level: int) -> str:
|
||||
:param level: 日志级别数字 Log level number
|
||||
:return: 格式化后的日志级别 Formatted log level
|
||||
"""
|
||||
if level <= 10 and level >= 0:
|
||||
if level < 10:
|
||||
return "DEBUG"
|
||||
elif level <= 20 and level > 10:
|
||||
elif level < 20:
|
||||
return "INFO"
|
||||
elif level <= 30 and level > 20:
|
||||
elif level < 30:
|
||||
return "WARN"
|
||||
elif level <= 40 and level > 30:
|
||||
elif level < 40:
|
||||
return "ERRO"
|
||||
elif level <= 50 and level > 40:
|
||||
elif level < 50:
|
||||
return "CRIT"
|
||||
else:
|
||||
return "UNKN"
|
||||
@@ -65,10 +66,10 @@ def fmt_placeholder(message: Any, use_date_color: bool = True) -> str:
|
||||
message = str(message)
|
||||
if use_date_color:
|
||||
message = message.format_map(SafeDict(
|
||||
asctime = set_color(get_asctime(),"#28ffb6"),
|
||||
time = set_color(get_time(),"#28ffb6"),
|
||||
weekday = set_color(get_weekday(),"#28ffb6"),
|
||||
date = set_color(get_date(),"#28ffb6")
|
||||
asctime = set_color(get_asctime(),time_color),
|
||||
time = set_color(get_time(),time_color),
|
||||
weekday = set_color(get_weekday(),time_color),
|
||||
date = set_color(get_date(),time_color)
|
||||
))
|
||||
else:
|
||||
message = message.format_map(SafeDict(
|
||||
@@ -88,46 +89,34 @@ def fmt_message(message: Any, no_placeholder: bool = False, no_color: bool = Fal
|
||||
"""
|
||||
|
||||
def process_color_tags(msg: str) -> str:
|
||||
from io import StringIO
|
||||
output = StringIO()
|
||||
stack = []
|
||||
result = []
|
||||
current_content = []
|
||||
last_end = 0
|
||||
pattern = re.compile(r'(<#([0-9a-fA-F]{6})>|</>)')
|
||||
current_color = None
|
||||
|
||||
for match in pattern.finditer(msg):
|
||||
text_between = msg[last_end:match.start()]
|
||||
if stack:
|
||||
current_content.append(text_between)
|
||||
else:
|
||||
result.append(text_between)
|
||||
|
||||
last_end = match.end()
|
||||
output.write(msg[last_end:match.start()])
|
||||
tag = match.group(1)
|
||||
last_end = match.end()
|
||||
|
||||
if tag.startswith('<#'):
|
||||
color_code = match.group(2)
|
||||
stack.append(color_code)
|
||||
stack.append(current_color)
|
||||
current_color = match.group(2)
|
||||
else:
|
||||
if stack:
|
||||
color = stack.pop()
|
||||
colored_content = set_color(''.join(current_content), f'#{color}')
|
||||
result.append(colored_content)
|
||||
current_content = []
|
||||
current_color = stack.pop()
|
||||
else:
|
||||
result.append(tag)
|
||||
output.write(tag)
|
||||
|
||||
remaining_text = msg[last_end:]
|
||||
if stack:
|
||||
current_content.append(remaining_text)
|
||||
else:
|
||||
result.append(remaining_text)
|
||||
output.write(msg[last_end:])
|
||||
result = output.getvalue()
|
||||
output.close()
|
||||
|
||||
for color in reversed(stack):
|
||||
result.append(f'<#{color}>')
|
||||
result.append(''.join(current_content))
|
||||
current_content = []
|
||||
|
||||
return ''.join(result)
|
||||
if current_color:
|
||||
result += ''.join(f'<#{color}>' for color in reversed(stack)) if stack else f'<#{current_color}>'
|
||||
return result
|
||||
if no_color:
|
||||
processed_message = str(message)
|
||||
else:
|
||||
@@ -161,14 +150,12 @@ def fmt_console(level: int, message: Any, prefix: str | None = None) -> Optional
|
||||
:param message: 消息内容 Message content
|
||||
:return: 格式化后的消息 Formatted message
|
||||
"""
|
||||
cl = get_config("console_level")
|
||||
fmt = get_config("console_format")
|
||||
if fmt_level(cl) > level:
|
||||
console_level = get_config("console_level")
|
||||
if level != -1 and fmt_level(console_level) > level:
|
||||
return None
|
||||
if prefix is None:
|
||||
prefix = ""
|
||||
fmt = fmt_placeholder(fmt)
|
||||
return fmt.format(
|
||||
fmt = get_config("console_format")
|
||||
prefix = prefix or ""
|
||||
return fmt_placeholder(fmt).format(
|
||||
levelname = fmt_level_name(fmt_level_number(level)),
|
||||
prefix = fmt_message(prefix, no_placeholder=True),
|
||||
message = fmt_message(message, no_placeholder=True)
|
||||
|
@@ -8,19 +8,21 @@ py-logiliteal's style tools, used to format log output
|
||||
|
||||
from typing import Union, Optional
|
||||
|
||||
def _get_hex_to_ansi(hex_color: str) -> Union[Optional[str], None]:
|
||||
def _get_hex_to_ansi(hex_color: str|None) -> Union[Optional[str], None]:
|
||||
"""
|
||||
将16进制颜色值转换为ANSI转义序列
|
||||
Convert hex color value to ANSI escape sequence
|
||||
:param hex_color: 16进制颜色值 Hex color value
|
||||
:return: ANSI转义序列 ANSI escape sequence
|
||||
"""
|
||||
if not hex_color:
|
||||
return ""
|
||||
if not hex_color.startswith("#"):
|
||||
return None
|
||||
r, g, b = int(hex_color[1:3], 16), int(hex_color[3:5], 16), int(hex_color[5:7], 16)
|
||||
return f"\033[38;2;{r};{g};{b}m"
|
||||
|
||||
def set_color(text: str, color: str) -> str:
|
||||
def set_color(text: str|None, color: str|None) -> str:
|
||||
"""
|
||||
设置文本颜色
|
||||
Set text color
|
||||
@@ -28,12 +30,16 @@ def set_color(text: str, color: str) -> str:
|
||||
:param color: 颜色值 Color value
|
||||
:return: 格式化后的文本 Formatted text
|
||||
"""
|
||||
if not text:
|
||||
return ""
|
||||
if not color:
|
||||
return text
|
||||
ansi = _get_hex_to_ansi(color)
|
||||
if not ansi:
|
||||
return text
|
||||
return f"{ansi}{text}\033[0m"
|
||||
|
||||
def set_bg_color(text: str, color: str) -> str:
|
||||
def set_bg_color(text: str|None, color: str|None) -> str:
|
||||
"""
|
||||
设置文本背景颜色
|
||||
Set text background color
|
||||
@@ -41,14 +47,17 @@ def set_bg_color(text: str, color: str) -> str:
|
||||
:param color: 颜色值 Color value
|
||||
:return: 格式化后的文本 Formatted text
|
||||
"""
|
||||
if not text:
|
||||
return ""
|
||||
if not color:
|
||||
return text
|
||||
ansi = _get_hex_to_ansi(color)
|
||||
if not ansi:
|
||||
return text
|
||||
# 将前景色ANSI代码转换为背景色代码 (38→48)
|
||||
ansi = ansi.replace("38;", "48;")
|
||||
return f"{ansi}{text}\033[0m"
|
||||
|
||||
def set_style(text: str, bold: bool = False, underline: bool = False, reverse: bool = False) -> str:
|
||||
def set_style(text: str|None, bold: bool = False, underline: bool = False, reverse: bool = False) -> str:
|
||||
"""
|
||||
设置文本样式
|
||||
Set text style
|
||||
@@ -58,6 +67,8 @@ def set_style(text: str, bold: bool = False, underline: bool = False, reverse: b
|
||||
:param reverse: 是否反相 Is reverse
|
||||
:return: 格式化后的文本 Formatted text
|
||||
"""
|
||||
if not text:
|
||||
return ""
|
||||
ansi = ""
|
||||
if bold:
|
||||
ansi += "\033[1m"
|
||||
|
@@ -9,10 +9,11 @@ Time utility module, used for time formatting and output, caching time formattin
|
||||
from datetime import datetime
|
||||
from .configs import get_config
|
||||
|
||||
import time
|
||||
|
||||
cache_time: str = ""
|
||||
cache_asctime: str = ""
|
||||
cache_date: str = ""
|
||||
cache_weekday: str = ""
|
||||
cache_time_ts: float = 0.0
|
||||
cache_fmt: str | None = None
|
||||
|
||||
def get_asctime() -> str:
|
||||
"""
|
||||
@@ -20,11 +21,7 @@ def get_asctime() -> str:
|
||||
Get current time(YYYY-MM-DD HH:MM:SS) and cache formatted result
|
||||
:return: 格式化后的时间 Formatted time
|
||||
"""
|
||||
global cache_asctime
|
||||
if cache_asctime:
|
||||
return cache_asctime
|
||||
cache_asctime = datetime.now().strftime(get_config("date_format"))
|
||||
return cache_asctime
|
||||
return _get_time(get_config("asctime_format"))
|
||||
|
||||
def get_time() -> str:
|
||||
"""
|
||||
@@ -32,24 +29,15 @@ def get_time() -> str:
|
||||
Get current time(HH:MM:SS) and cache formatted result
|
||||
:return: 格式化后的时间 Formatted time
|
||||
"""
|
||||
global cache_time
|
||||
if cache_time:
|
||||
return cache_time
|
||||
cache_time = datetime.now().strftime("%H:%M:%S")
|
||||
return cache_time
|
||||
return _get_time(get_config("time_format"))
|
||||
|
||||
def get_date() -> str:
|
||||
"""
|
||||
获取当前日期(YYYY-MM-DD),并缓存格式化结果
|
||||
获取当前日期(星期几),并缓存格式化结果
|
||||
Get current date(YYYY-MM-DD) and cache formatted result
|
||||
:return: 格式化后的日期 Formatted date
|
||||
"""
|
||||
global cache_date
|
||||
if cache_date:
|
||||
return cache_date
|
||||
cache_date = datetime.now().strftime("%Y-%m-%d")
|
||||
return cache_date
|
||||
return _get_time(get_config("date_format"))
|
||||
|
||||
def get_weekday() -> str:
|
||||
"""
|
||||
@@ -57,8 +45,21 @@ def get_weekday() -> str:
|
||||
Get current date(weekday) and cache formatted result
|
||||
:return: 格式化后的星期几 Formatted weekday
|
||||
"""
|
||||
global cache_weekday
|
||||
if cache_weekday:
|
||||
return cache_weekday
|
||||
cache_weekday = datetime.now().strftime("%A")
|
||||
return cache_weekday
|
||||
return _get_time(get_config("weekday_format"))
|
||||
|
||||
def _get_time(fmt: str) -> str:
|
||||
"""
|
||||
获取当前时间(根据指定格式),并缓存格式化结果
|
||||
Get current time(based on specified format) and cache formatted result
|
||||
:param fmt: 时间格式 Time format
|
||||
:return: 格式化后的时间 Formatted time
|
||||
"""
|
||||
global cache_time, cache_time_ts, cache_fmt
|
||||
if cache_fmt is None:
|
||||
cache_fmt = fmt
|
||||
now = time.time()
|
||||
if cache_time and (now - cache_time_ts < 1) and (cache_fmt == fmt):
|
||||
return cache_time
|
||||
cache_time = datetime.now().strftime(fmt)
|
||||
cache_time_ts = now
|
||||
return cache_time
|
@@ -7,11 +7,14 @@
|
||||
"enable_console": true,
|
||||
"enable_file": true,
|
||||
"console_color": true,
|
||||
"console_level": "DEBUG",
|
||||
"console_format": "{asctime} {levelname} | {prefix}{message}",
|
||||
"console_level": "INFO",
|
||||
"console_format": "{time} {levelname} | {prefix}{message}",
|
||||
"console_prefix": "Auto",
|
||||
"console_encoding": "utf-8",
|
||||
"date_format": "%Y-%m-%d %H:%M:%S",
|
||||
"asctime_format": "%Y-%m-%d %H:%M:%S",
|
||||
"time_format": "%H:%M:%S",
|
||||
"date_format": "%Y-%m-%d",
|
||||
"weekday_format": "%A",
|
||||
"level_name": {
|
||||
"DEBUG": "DEBUG",
|
||||
"INFO": "INFO",
|
||||
@@ -21,7 +24,7 @@
|
||||
},
|
||||
"level_color": {
|
||||
"DEBUG": "#c1d5ff",
|
||||
"INFO": "#c1f8ff",
|
||||
"INFO": "#c1ffff",
|
||||
"WARN": "#fff600",
|
||||
"ERRO": "#ffa000",
|
||||
"CRIT": "#ff8181"
|
@@ -4,7 +4,7 @@ from pathlib import Path
|
||||
project_root = Path(__file__).parent.parent
|
||||
sys.path.append(str(project_root))
|
||||
|
||||
from src.utils.configs import get_config, set_config, reset_config
|
||||
from src.logiliteal.utils.configs import get_config, set_config, reset_config, create_backup
|
||||
|
||||
print(f"\n配置更换测试")
|
||||
set_config("file_level", "DEBUG")
|
||||
|
@@ -4,7 +4,7 @@ from pathlib import Path
|
||||
project_root = Path(__file__).parent.parent
|
||||
sys.path.append(str(project_root))
|
||||
|
||||
from src.utils.fmt import fmt_level, fmt_level_number, fmt_console, fmt_message, fmt_placeholder
|
||||
from src.logiliteal.utils.fmt import fmt_level, fmt_level_number, fmt_console, fmt_message, fmt_placeholder
|
||||
|
||||
print("日志级别测试")
|
||||
print(fmt_level("DEBUG"))
|
||||
|
@@ -4,9 +4,12 @@ from pathlib import Path
|
||||
project_root = Path(__file__).parent.parent
|
||||
sys.path.append(str(project_root))
|
||||
|
||||
from src import Logger
|
||||
from src.logiliteal import Logger
|
||||
|
||||
log = Logger()
|
||||
|
||||
log.info("测试信息日志")
|
||||
log.debug("测试调试日志")
|
||||
log.warn("测试警告日志")
|
||||
log.error("测试错误日志")
|
||||
log.critical("测试严重错误日志")
|
25
tests/t-time.py
Normal file
25
tests/t-time.py
Normal file
@@ -0,0 +1,25 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
import time
|
||||
|
||||
project_root = Path(__file__).parent.parent
|
||||
sys.path.append(str(project_root))
|
||||
|
||||
from src.logiliteal.utils.time import get_asctime, get_time, get_date, get_weekday
|
||||
from src.logiliteal.levels import Logger
|
||||
|
||||
log = Logger()
|
||||
|
||||
log.info(get_asctime())
|
||||
log.info(get_time())
|
||||
log.info(get_date())
|
||||
log.info(get_weekday())
|
||||
|
||||
while True:
|
||||
try:
|
||||
log.info(get_asctime())
|
||||
time.sleep(1)
|
||||
log.info("时间分割线")
|
||||
except KeyboardInterrupt:
|
||||
log.info("测试结束")
|
||||
break
|
Reference in New Issue
Block a user