AskaEth
e6875f0b4b
- 13-service async plugin framework - Textual TUI with CLI fallback - Plugin hot-reload + permission system - Web management panel (aiohttp) - Bridge-based inter-module communication - 10 regression tests Fixes applied: - PBKDF2-SHA256 auth (was plain SHA256) - Auth bypass removed (was allow-all on fail) - Bare excepts replaced with logged errors - CatFramework/DreamSu -> SenSu naming unified - ServiceManager: health checks + startup_order - Env var credentials (SENSU_ADMIN_PASSWORD etc)
64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import sys
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
# 添加项目根目录到Python路径
|
|
project_root = Path(__file__).parent
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
async def debug_log_format():
|
|
"""调试日志格式"""
|
|
try:
|
|
from services.log_service import LogService
|
|
from services.init_service import InitService
|
|
|
|
print("🐱 调试日志格式...")
|
|
|
|
# 初始化配置
|
|
init_service = InitService()
|
|
configs = await init_service.initialize_framework()
|
|
base_config = configs['base']
|
|
|
|
# 创建日志服务
|
|
log_service = LogService(base_config)
|
|
|
|
# 等待日志服务初始化
|
|
await asyncio.sleep(1)
|
|
|
|
# 添加测试消费者
|
|
def test_consumer(log_record):
|
|
print("📝 消费者收到的日志记录:")
|
|
print(f" 所有键: {list(log_record.keys())}")
|
|
if 'simple_message' in log_record:
|
|
print(f" simple_message: {log_record['simple_message']}")
|
|
if 'formatted_message' in log_record:
|
|
print(f" formatted_message: {log_record['formatted_message']}")
|
|
if 'level' in log_record:
|
|
print(f" level: {log_record['level']}")
|
|
print("---")
|
|
|
|
log_service.add_log_consumer(test_consumer)
|
|
|
|
# 测试日志记录
|
|
import logging
|
|
logger = logging.getLogger("test_logger")
|
|
print("\n🔧 发送测试日志...")
|
|
logger.info("这是一条测试信息日志")
|
|
logger.debug("这是一条测试调试日志")
|
|
logger.warning("这是一条测试警告日志")
|
|
logger.error("这是一条测试错误日志")
|
|
|
|
# 等待日志处理完成
|
|
await asyncio.sleep(0.5)
|
|
|
|
except Exception as e:
|
|
print(f"❌ 调试失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(debug_log_format())
|