110 lines
4.0 KiB
Python
110 lines
4.0 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""
|
|||
|
|
测试API日志输出 - 验证API服务器的日志配置
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import sys
|
|||
|
|
import os
|
|||
|
|
import time
|
|||
|
|
import logging
|
|||
|
|
|
|||
|
|
# 设置Windows控制台支持UTF-8编码
|
|||
|
|
if os.name == 'nt': # Windows系统
|
|||
|
|
# 设置控制台编码为UTF-8
|
|||
|
|
os.system('chcp 65001 >nul 2>&1')
|
|||
|
|
# 重新配置stdout和stderr以支持UTF-8，禁用缓冲
|
|||
|
|
import io
|
|||
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace', line_buffering=True)
|
|||
|
|
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace', line_buffering=True)
|
|||
|
|
# 确保输出立即刷新
|
|||
|
|
sys.stdout.reconfigure(encoding='utf-8', errors='replace', line_buffering=True)
|
|||
|
|
sys.stderr.reconfigure(encoding='utf-8', errors='replace', line_buffering=True)
|
|||
|
|
|
|||
|
|
# 配置日志记录 - 增强中文支持
|
|||
|
|
logging.basicConfig(
|
|||
|
|
level=logging.INFO,
|
|||
|
|
format='[%(asctime)s] %(levelname)s - %(message)s',
|
|||
|
|
datefmt='%H:%M:%S',
|
|||
|
|
handlers=[
|
|||
|
|
logging.StreamHandler(sys.stdout), # 输出到控制台，立即刷新
|
|||
|
|
logging.FileHandler('test_api.log', encoding='utf-8') # 输出到文件
|
|||
|
|
],
|
|||
|
|
force=True # 强制重新配置日志
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# 设置所有logger立即刷新
|
|||
|
|
for handler in logging.root.handlers:
|
|||
|
|
if isinstance(handler, logging.StreamHandler):
|
|||
|
|
handler.stream = sys.stdout
|
|||
|
|
handler.flush = lambda: handler.stream.flush()
|
|||
|
|
|
|||
|
|
# 创建logger
|
|||
|
|
logger = logging.getLogger(__name__)
|
|||
|
|
|
|||
|
|
def test_api_logging():
|
|||
|
|
"""测试API日志输出功能"""
|
|||
|
|
|
|||
|
|
print("\n" + "="*60, flush=True)
|
|||
|
|
print("🧪 API日志输出测试开始", flush=True)
|
|||
|
|
print("="*60, flush=True)
|
|||
|
|
|
|||
|
|
# 测试logger输出
|
|||
|
|
print("\n📝 测试Logger输出:", flush=True)
|
|||
|
|
logger.info("🚀 API服务器启动")
|
|||
|
|
logger.info("📋 配置加载完成")
|
|||
|
|
logger.info("🏷️ 版本信息: v2.1.0")
|
|||
|
|
logger.warning("⚠️ 这是一个警告消息")
|
|||
|
|
logger.error("❌ 这是一个错误消息")
|
|||
|
|
|
|||
|
|
# 测试混合输出
|
|||
|
|
print("\n🔄 测试混合print和logger输出:", flush=True)
|
|||
|
|
task_id = "test-67890"
|
|||
|
|
|
|||
|
|
print(f"🚀 训练任务开始: {task_id}", flush=True)
|
|||
|
|
logger.info(f"🚀 训练任务开始: {task_id}")
|
|||
|
|
|
|||
|
|
print(f"📋 任务详情: 训练 TRANSFORMER 模型 - 药品 P001", flush=True)
|
|||
|
|
logger.info(f"📋 任务详情: 训练 TRANSFORMER 模型 - 药品 P001")
|
|||
|
|
|
|||
|
|
print(f"🏷️ 版本信息: 版本号 v1.0, 模型标识: P001", flush=True)
|
|||
|
|
logger.info(f"🏷️ 版本信息: 版本号 v1.0, 模型标识: P001")
|
|||
|
|
|
|||
|
|
# 模拟训练进度
|
|||
|
|
print("\n🏃 模拟训练进度:", flush=True)
|
|||
|
|
for epoch in range(1, 4):
|
|||
|
|
loss = 0.1234 - epoch * 0.01
|
|||
|
|
print(f"📊 Epoch {epoch}/3, 训练损失: {loss:.4f}, 测试损失: {loss+0.01:.4f}", flush=True)
|
|||
|
|
logger.info(f"📊 Epoch {epoch}/3, 训练损失: {loss:.4f}, 测试损失: {loss+0.01:.4f}")
|
|||
|
|
time.sleep(0.3)
|
|||
|
|
|
|||
|
|
# 强制刷新所有输出
|
|||
|
|
sys.stdout.flush()
|
|||
|
|
sys.stderr.flush()
|
|||
|
|
|
|||
|
|
# 测试完成输出
|
|||
|
|
print("\n✅ 模拟训练完成:", flush=True)
|
|||
|
|
print(f"📈 训练完成! 结果类型: <class 'dict'>", flush=True)
|
|||
|
|
logger.info("📈 训练完成")
|
|||
|
|
|
|||
|
|
print(f"💾 模型保存路径: saved_models/transformer/P001_v1.0.pth", flush=True)
|
|||
|
|
logger.info("💾 模型已保存")
|
|||
|
|
|
|||
|
|
print(f"✔️ 任务状态更新: 已完成", flush=True)
|
|||
|
|
logger.info("✔️ 任务状态更新: 已完成")
|
|||
|
|
|
|||
|
|
print("\n" + "="*60, flush=True)
|
|||
|
|
print("🎉 API日志输出测试完成!", flush=True)
|
|||
|
|
print("📁 详细日志已保存到: test_api.log", flush=True)
|
|||
|
|
print("="*60, flush=True)
|
|||
|
|
|
|||
|
|
# 检查日志文件
|
|||
|
|
if os.path.exists('test_api.log'):
|
|||
|
|
print("\n📖 日志文件内容预览:", flush=True)
|
|||
|
|
with open('test_api.log', 'r', encoding='utf-8') as f:
|
|||
|
|
lines = f.readlines()
|
|||
|
|
for line in lines[-5:]: # 显示最后5行
|
|||
|
|
print(f" {line.strip()}", flush=True)
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
test_api_logging()
|