File size: 3,450 Bytes
793e953 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """
转写进度追踪器（内存级 + DB 状态）
为每个问诊维护：
- stage (preparing | uploading | transcribing | extracting | generating | done | error)
- percent (0-100)
- message (人类可读描述)
- engine (实际使用的引擎名称)
- updated_at
提供线程安全的更新与查询接口，供前端轮询。
"""
import threading
import time
from dataclasses import dataclass, asdict
from typing import Optional
@dataclass
class ProgressState:
"""单个问诊的进度状态"""
consultation_id: str
stage: str = "preparing" # preparing | uploading | transcribing | extracting | generating | done | error
percent: int = 0 # 0-100
message: str = "准备中" # 人类可读
engine: str = "xunfei_llm" # 当前实际使用的引擎
error: Optional[str] = None # 错误信息（如有）
updated_at: float = 0.0 # 上次更新时间戳
class ProgressTracker:
"""内存级进度追踪器（线程安全）"""
def __init__(self) -> None:
self._states: dict[str, ProgressState] = {}
self._lock = threading.Lock()
def init(self, consultation_id: str, engine: str = "xunfei_llm") -> ProgressState:
"""初始化/重置一个问诊的进度"""
with self._lock:
state = ProgressState(
consultation_id=consultation_id,
stage="preparing",
percent=0,
message="准备中…",
engine=engine,
updated_at=time.time(),
)
self._states[consultation_id] = state
return state
def update(
self,
consultation_id: str,
*,
stage: Optional[str] = None,
percent: Optional[int] = None,
message: Optional[str] = None,
error: Optional[str] = None,
) -> ProgressState:
"""更新进度状态"""
with self._lock:
state = self._states.get(consultation_id)
if state is None:
state = ProgressState(consultation_id=consultation_id)
self._states[consultation_id] = state
if stage is not None:
state.stage = stage
if percent is not None:
state.percent = max(state.percent, min(percent, 100))
if message is not None:
state.message = message
if error is not None:
state.error = error
state.stage = "error"
state.updated_at = time.time()
return state
def get(self, consultation_id: str) -> Optional[dict]:
"""获取进度状态（用于 API 返回）"""
with self._lock:
state = self._states.get(consultation_id)
if state is None:
return None
return asdict(state)
def clear(self, consultation_id: str) -> None:
"""清理进度记录"""
with self._lock:
self._states.pop(consultation_id, None)
# ================================================================
# 全局单例
# ================================================================
_tracker: Optional[ProgressTracker] = None
def get_progress_tracker() -> ProgressTracker:
"""获取全局进度追踪器（单例）"""
global _tracker
if _tracker is None:
_tracker = ProgressTracker()
return _tracker
|