iOSBuildServer/backend/services/log_streamer.py
shen f5919a8229 feat: 构建超时控制、日志持久化、任务删除、pod install 及多项 UI 优化
- 新增 BUILD_TIMEOUT_HOURS 配置,超时自动标记失败
- 日志保存到独立 backend/logs/ 目录,不随构建目录删除
- 新增任务删除 API 及前端删除按钮
- 构建流程增加 pod install 步骤
- 通过文件传递配置避免命令行参数截断
- Provisioning Profile 支持按名称搜索
- dSYM 下载支持目录打包为 zip
- 前端端口可配置、历史页增加下载列和耗时显示
- 日志弹窗支持展开/收起详细日志
- 自动生成 upload_key 标识
2026-06-08 18:56:43 +08:00

141 lines
4.7 KiB
Python

"""WebSocket 日志流"""
import asyncio
import json
from datetime import datetime
from pathlib import Path
from typing import Dict, AsyncGenerator
class LogStreamer:
"""日志流管理器"""
def __init__(self):
self._queues: Dict[str, asyncio.Queue] = {}
self._subscribers: Dict[str, list] = {}
self._log_lines: Dict[str, list] = {}
def create_queue(self, task_id: str) -> asyncio.Queue:
"""创建任务的日志队列"""
queue = asyncio.Queue(maxsize=1000)
self._queues[task_id] = queue
self._log_lines[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
# 回放已缓存的历史日志
existing = self._log_lines.get(task_id, [])
for entry in existing:
yield entry
# 流式推送新消息
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):
"""标记任务日志结束"""
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)
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()