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() |