feat: add progress callback mechanism for real-time updates in data extraction
- Add ProgressInfo and ProgressCalculator classes for structured progress tracking - Implement RealtimeOutput for immediate log display without buffering - Add progress callback support to DiscreteMaterialPlanExtractor - Use queue-based thread-safe communication for GUI progress updates - Fix log output issue where all content appeared at once after completion Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
97
gui/progress.py
Normal file
97
gui/progress.py
Normal file
@@ -0,0 +1,97 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
进度信息模块
|
||||
|
||||
定义用于进度回调的数据结构。
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Dict, Any, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class ProgressInfo:
|
||||
"""
|
||||
进度信息
|
||||
|
||||
用于在后台任务和 GUI 之间传递进度信息。
|
||||
"""
|
||||
stage: str # 阶段标识: 'login', 'query', 'download', 'logout', 'convert', 'complete'
|
||||
current: int # 当前进度值
|
||||
total: int # 总量
|
||||
message: str # 显示给用户的消息
|
||||
detail: Dict[str, Any] = field(default_factory=dict) # 额外详细信息
|
||||
|
||||
@property
|
||||
def percent(self) -> int:
|
||||
"""计算进度百分比(0-100)"""
|
||||
if self.total <= 0:
|
||||
return 0
|
||||
return int(self.current * 100 / self.total)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"ProgressInfo(stage={self.stage}, {self.current}/{self.total}, {self.message})"
|
||||
|
||||
|
||||
class ProgressCalculator:
|
||||
"""
|
||||
进度计算器
|
||||
|
||||
将各阶段的进度映射到总体进度百分比。
|
||||
"""
|
||||
|
||||
# 各阶段在总进度中的占比
|
||||
STAGE_WEIGHTS = {
|
||||
'login': 5, # 登录: 0-5%
|
||||
'query': 5, # 查询: 5-10%
|
||||
'download': 65, # 下载: 10-75%
|
||||
'logout': 5, # 注销: 75-80%
|
||||
'convert': 15, # 转换: 80-95%
|
||||
'complete': 5, # 完成: 95-100%
|
||||
}
|
||||
|
||||
def __init__(self):
|
||||
"""初始化进度计算器"""
|
||||
self._stage_offsets = self._calculate_stage_offsets()
|
||||
|
||||
def _calculate_stage_offsets(self) -> Dict[str, int]:
|
||||
"""计算各阶段的起始偏移量(百分比)"""
|
||||
offsets = {}
|
||||
offset = 0
|
||||
for stage, weight in self.STAGE_WEIGHTS.items():
|
||||
offsets[stage] = offset
|
||||
offset += weight
|
||||
return offsets
|
||||
|
||||
def calculate_overall_percent(self, progress: ProgressInfo) -> int:
|
||||
"""
|
||||
计算总体进度百分比
|
||||
|
||||
Args:
|
||||
progress: 进度信息
|
||||
|
||||
Returns:
|
||||
总体进度百分比 (0-100)
|
||||
"""
|
||||
stage = progress.stage
|
||||
|
||||
if stage == 'complete':
|
||||
return 100
|
||||
|
||||
if stage not in self._stage_offsets:
|
||||
return 0
|
||||
|
||||
# 计算阶段起始百分比
|
||||
stage_offset = self._stage_offsets[stage]
|
||||
|
||||
# 计算阶段内的进度百分比
|
||||
stage_percent = progress.percent
|
||||
|
||||
# 计算该阶段的权重
|
||||
stage_weight = self.STAGE_WEIGHTS[stage]
|
||||
|
||||
# 总进度 = 阶段偏移 + (阶段内进度 * 阶段权重 / 100)
|
||||
overall = stage_offset + int(stage_percent * stage_weight / 100)
|
||||
|
||||
return min(overall, 100)
|
||||
Reference in New Issue
Block a user