automas_admin/AUTO-MAS-test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 接入logger

Browse Source
This commit is contained in:
DLmaster361
2025-08-04 10:46:23 +08:00
parent 92ea6026f8
commit 76438459f7
10 changed files with 45 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
app/__init__.py
View File

@@ -24,6 +24,7 @@ __author__ = "DLmaster361 <DLmaster_361@163.com>"
__license__ = "GPL-3.0 license"
from .api import app
from .core import Config, logger
from .core import Config
from .utils import get_logger
__all__ = ["app", "Config", "logger"]
__all__ = ["app", "Config", "get_logger"]

5
app/api/api.py
View File

@@ -5,7 +5,10 @@ from datetime import datetime
from fastapi.middleware.cors import CORSMiddleware
from app.core import Config, logger
from app.core import Config
from app.utils import get_logger
logger = get_logger("API 模块")
app = FastAPI(
title="AUTO_MAA",

0
app/core/MAA.py
View File

3
app/core/__init__.py
View File

@@ -23,6 +23,5 @@ __author__ = "DLmaster361 <DLmaster_361@163.com>"
__license__ = "GPL-3.0 license"
from .config import Config
from .logger import logger
__all__ = ["Config", "logger"]
__all__ = ["Config"]

43
app/core/config.py
View File

@@ -34,7 +34,7 @@ from pathlib import Path
from typing import Union, Dict, List, Literal
from .logger import logger
from app.utils import get_logger
from app.models.ConfigBase import *
@@ -593,7 +593,13 @@ class AppConfig(GlobalConfig):
# self.if_ignore_silence = False
# self.if_database_opened = False
self.init_logger()
self.logger = get_logger("配置管理")
self.logger.info("")
self.logger.info("===================================")
self.logger.info("AUTO_MAA 后端应用程序")
self.logger.info(f"版本号： v{self.VERSION}")
self.logger.info(f"根目录： {self.root_path}")
self.logger.info("===================================")
# 检查目录
self.log_path.parent.mkdir(parents=True, exist_ok=True)
@@ -616,38 +622,7 @@ class AppConfig(GlobalConfig):
await self.QueueConfig.connect(self.config_path / "QueueConfig.json")
# self.check_data()
logger.info("程序初始化完成", module="配置管理")
def init_logger(self) -> None:
"""初始化日志记录器"""
logger.add(
sink=self.log_path,
level="DEBUG",
format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <cyan>{extra[module]}</cyan> | <level>{message}</level>",
enqueue=True,
backtrace=True,
diagnose=True,
rotation="1 week",
retention="1 month",
compression="zip",
)
logger.add(
sink=sys.stderr,
level="DEBUG",
format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <cyan>{extra[module]}</cyan> | <level>{message}</level>",
enqueue=True,
backtrace=True,
diagnose=True,
colorize=True,
)
logger.info("", module="配置管理")
logger.info("===================================", module="配置管理")
logger.info("AUTO_MAA 后端", module="配置管理")
logger.info(f"版本号： v{self.VERSION}", module="配置管理")
logger.info(f"根目录： {self.root_path}", module="配置管理")
logger.info("===================================", module="配置管理")
self.logger.info("程序初始化完成")
async def add_script(
self, script: Literal["MAA", "General"]

59
app/models/test.py
View File

@@ -1,59 +0,0 @@
import ConfigBase
from typing import Dict, Union, List
import uuid
from pathlib import Path
class TConfig(ConfigBase.ConfigBase):
def __init__(self):
super().__init__()
self.Main_Name = ConfigBase.ConfigItem("Main", "Name", "Default Name")
class TestConfig(ConfigBase.ConfigBase):
Main_Name = ConfigBase.ConfigItem("Main", "Name", "Default Name")
Bool_Enabled = ConfigBase.ConfigItem(
"Bool", "Enabled", False, ConfigBase.BoolValidator()
)
Mt = ConfigBase.MultipleConfig([TConfig])
root = Path.cwd()
test_config = TestConfig()
test_config.connect(root / "config.json")
test_config.set("Main", "Name", "New Name")
test_config.set("Bool", "Enabled", "qqq")
mt_test = ConfigBase.MultipleConfig([TestConfig])
mt_test.connect(root / "mt_config.json")
tc = mt_test.add(TestConfig)
for uid, config in mt_test.items():
print(uid, config.toDict())
config.set("Main", "Name", "Updated Name")
print(mt_test.toDict())
mt_test.add(TestConfig)
mt_test.setOrder(list(mt_test.keys())[::-1])
print(mt_test.toDict())
print("---------------------------------------------")
k: TestConfig = mt_test.add(TestConfig)
print(k.Mt.toDict())

9
app/core/logger.py → app/utils/__init__.py
View File

@@ -18,8 +18,11 @@
# Contact: DLmaster_361@163.com
__version__ = "5.0.0"
__author__ = "DLmaster361 <DLmaster_361@163.com>"
__license__ = "GPL-3.0 license"
from loguru import logger as _logger
logger = _logger.patch(lambda record: record["extra"].setdefault("module", "未知模块"))
logger.remove(0)
from .logger import get_logger
__all__ = ["get_logger"]

77
app/utils/config.py
View File

@@ -1,77 +0,0 @@
'''
配置文件管理, 使用yaml格式\n
注意: 此文件仅处理程序配置项而非用户配置
'''
from pathlib import Path
from typing import Any, ClassVar
import yaml
from pydantic import BaseModel
class BaseYAMLConfig(BaseModel):
# 必要项
CONFIG_PATH: ClassVar[Path]
@classmethod
def load(cls, config_path: Path | None = None) -> "BaseYAMLConfig":
"""
从 YAML 文件加载配置。如果文件不存在或为空，则使用类中定义的默认值。
"""
# 使用传入路径或类变量路径
if not cls.CONFIG_PATH:
raise ValueError("CONFIG_PATH必须被设置")
path = (config_path or cls.CONFIG_PATH).resolve()
if not path.exists():
# print(f"[Warning] Config file '{path}' not found. Using default values.")
data: dict[str, Any] = {}
else:
try:
with open(path, "r", encoding="utf-8") as f:
data = yaml.safe_load(f)
if data is None:
data = {}
if not isinstance(data, dict):
raise ValueError(f"Expected dict in {path}, got {type(data)}")
except Exception as e:
raise RuntimeError(f"Failed to load config from {path}: {e}")
try:
return cls.model_validate(data)
except Exception as e:
raise ValueError(f"Invalid config for {cls.__name__}: {e}")
class LogConfig (BaseModel):
"""
日志配置
"""
CONFIG_PATH:ClassVar[Path] = Path("config.yaml")
level: str = "DEBUG"
writelevel: str = "INFO"
log_dir: str = "logs"
max_size: int = 10

60
app/utils/logger.py
View File

@@ -1,45 +1,36 @@
from loguru import logger as _logger
import sys
import os
from app.utils.config import LogConfig
from pathlib import Path
config = LogConfig()
LOG_DIR = config.log_dir
os.makedirs(LOG_DIR, exist_ok=True)
(Path.cwd() / "debug").mkdir(parents=True, exist_ok=True)
_logger.remove()
console_format = (
"<green>{time:YYYY-MM-DD HH:mm:ss}</green>| "
"<level>{level: <8}</level> | "
"<yellow>{extra[module]}</yellow>- {message}"
_logger.add(
sink=Path.cwd() / "debug/app.log",
level="INFO",
format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <cyan>{extra[module]}</cyan> | <level>{message}</level>",
enqueue=True,
backtrace=True,
diagnose=True,
rotation="1 week",
retention="1 month",
compression="zip",
)
_logger.add(
sink=sys.stderr,
format=console_format,
level=config.level,
level="DEBUG",
format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <cyan>{extra[module]}</cyan> | <level>{message}</level>",
enqueue=True,
backtrace=True,
diagnose=True,
colorize=True,
enqueue=True,
)
file_format = "{time:YYYY-MM-DD HH:mm:ss}| {level: <8} | {extra[module]}- {message}"
_logger.add(
sink=os.path.join(LOG_DIR, "app_{time:YYYY-MM-DD}.log"),
format=file_format,
level=config.writelevel,
rotation="00:00",
retention="7 days",
colorize=False,
encoding="utf-8",
enqueue=True,
)
_logger = _logger.patch(lambda record: record["extra"].setdefault("module", "未知模块"))
@@ -53,18 +44,3 @@ def get_logger(module_name: str):
__all__ = ["get_logger"]
if __name__ == "__main__":
logger = get_logger("test1")
logger.debug("调试信息（只在控制台显示）")
logger.info("这是普通信息（控制台 + 文件）")
logger.warning("这是警告")
logger.error("发生错误")
logger2 = get_logger("test2")
logger2.error("发生错误")
try:
_ = 1 / 0
except Exception:
logger.exception("捕获异常")