158 lines
5.3 KiB
Python
158 lines
5.3 KiB
Python
"""WebSocket 日志流"""
|
|
import asyncio
|
|
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Dict, AsyncGenerator
|
|
|
|
from ..config import LOG_QUEUE_MAX_SIZE
|
|
|
|
|
|
class LogStreamer:
|
|
"""日志流管理器"""
|
|
|
|
def __init__(self):
|
|
self._queues: Dict[str, asyncio.Queue] = {}
|
|
self._subscribers: Dict[str, list] = {}
|
|
self._log_lines: Dict[str, list] = {}
|
|
self._completed: set[str] = set()
|
|
|
|
def create_queue(self, task_id: str) -> asyncio.Queue:
|
|
"""创建任务的日志队列（重复调用时保留已有日志）"""
|
|
if task_id in self._queues:
|
|
return self._queues[task_id]
|
|
|
|
queue = asyncio.Queue(maxsize=LOG_QUEUE_MAX_SIZE)
|
|
self._queues[task_id] = queue
|
|
self._log_lines.setdefault(task_id, [])
|
|
self._completed.discard(task_id)
|
|
return queue
|
|
|
|
async def emit(self, task_id: str, message: str, level: str = "info"):
|
|
"""发送日志消息"""
|
|
log_entry = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"level": level,
|
|
"message": message,
|
|
}
|
|
|
|
# 收集日志
|
|
if task_id in self._log_lines:
|
|
self._log_lines[task_id].append(log_entry)
|
|
|
|
if task_id not in self._queues:
|
|
return
|
|
|
|
try:
|
|
self._queues[task_id].put_nowait(log_entry)
|
|
except asyncio.QueueFull:
|
|
# 队列满了，丢弃最旧的消息
|
|
try:
|
|
self._queues[task_id].get_nowait()
|
|
self._queues[task_id].put_nowait(log_entry)
|
|
except asyncio.QueueEmpty:
|
|
pass
|
|
|
|
async def emit_step(self, task_id: str, step: str):
|
|
"""发送步骤标记"""
|
|
await self.emit(task_id, f"[{step}]", level="step")
|
|
|
|
async def emit_error(self, task_id: str, error: str):
|
|
"""发送错误消息"""
|
|
await self.emit(task_id, error, level="error")
|
|
|
|
async def emit_warning(self, task_id: str, warning: str):
|
|
"""发送警告消息"""
|
|
await self.emit(task_id, warning, level="warn")
|
|
|
|
async def subscribe(self, task_id: str) -> AsyncGenerator[dict, None]:
|
|
"""订阅任务日志"""
|
|
queue = self._queues.get(task_id)
|
|
if not queue:
|
|
return
|
|
|
|
# Messages are stored both in history and the live queue. Replay the
|
|
# history once, then discard its queue copies before waiting for new logs.
|
|
existing = list(self._log_lines.get(task_id, []))
|
|
completed = task_id in self._completed
|
|
while not queue.empty():
|
|
queue.get_nowait()
|
|
|
|
for entry in existing:
|
|
yield entry
|
|
|
|
if completed:
|
|
return
|
|
|
|
# 流式推送新消息
|
|
while True:
|
|
try:
|
|
msg = await asyncio.wait_for(queue.get(), timeout=30)
|
|
if msg is None: # 结束信号
|
|
break
|
|
yield msg
|
|
except asyncio.TimeoutError:
|
|
# 发送心跳
|
|
yield {"timestamp": datetime.now().isoformat(), "level": "heartbeat", "message": ""}
|
|
except Exception:
|
|
break
|
|
|
|
def complete(self, task_id: str):
|
|
"""标记任务日志结束"""
|
|
self._completed.add(task_id)
|
|
if task_id in self._queues:
|
|
try:
|
|
self._queues[task_id].put_nowait(None)
|
|
except asyncio.QueueFull:
|
|
pass
|
|
|
|
def cleanup(self, task_id: str):
|
|
"""清理任务日志队列"""
|
|
self._queues.pop(task_id, None)
|
|
self._completed.discard(task_id)
|
|
|
|
def save_log(self, task_id: str, build_dir: Path):
|
|
"""将收集的日志保存到独立日志目录（不会随构建目录删除）"""
|
|
lines = self._log_lines.get(task_id)
|
|
if not lines:
|
|
return
|
|
|
|
# 保存到 backend/logs/ 目录
|
|
logs_dir = Path(__file__).parent.parent / "logs"
|
|
logs_dir.mkdir(exist_ok=True)
|
|
log_path = logs_dir / f"{task_id}.log"
|
|
with open(log_path, "w", encoding="utf-8") as f:
|
|
for entry in lines:
|
|
ts = entry["timestamp"]
|
|
level = entry["level"].upper()
|
|
msg = entry["message"]
|
|
f.write(f"[{ts}] [{level}] {msg}\n")
|
|
|
|
# 同时保存到构建目录（如果存在）
|
|
if build_dir and build_dir.exists():
|
|
with open(build_dir / "build.log", "w", encoding="utf-8") as f:
|
|
for entry in lines:
|
|
ts = entry["timestamp"]
|
|
level = entry["level"].upper()
|
|
msg = entry["message"]
|
|
f.write(f"[{ts}] [{level}] {msg}\n")
|
|
|
|
def get_log(self, build_dir: Path, task_id: str = None) -> str:
|
|
"""读取已保存的打包日志（优先从独立日志目录读取）"""
|
|
# 优先从独立日志目录读取
|
|
if task_id:
|
|
logs_dir = Path(__file__).parent.parent / "logs"
|
|
log_path = logs_dir / f"{task_id}.log"
|
|
if log_path.exists():
|
|
return log_path.read_text(encoding="utf-8")
|
|
# 回退到构建目录
|
|
if build_dir:
|
|
log_path = build_dir / "build.log"
|
|
if log_path.exists():
|
|
return log_path.read_text(encoding="utf-8")
|
|
return ""
|
|
|
|
|
|
# 全局单例
|
|
log_streamer = LogStreamer()
|