From 16ddc3edaa15d634d9e79602e6b49215db5f6a2c Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Wed, 11 Mar 2026 16:06:21 +0800 Subject: [PATCH] feat(03-02): implement ProgressMonitor class - Add ProgressInfo dataclass for progress data - Add ProgressMonitor class with: - PROGRESS_PATTERNS for extraction (percentage, fraction, phase, completion) - extract_progress() to find progress indicators in output - should_send_update() with 60 second throttling - create_progress_card() for Feishu card formatting - Fix fraction pattern to allow content after fraction (e.g., [3/5 tasks]) - Fix should_send_update() to return False on first call - Fix test assertions to use kwargs instead of positional args All 7 tests pass for progress monitoring. Co-Authored-By: Claude Opus 4.6 --- .../src/gsd_openclaw/progress_monitor.py | 164 ++++++++++++++++++ gsd-openclaw/tests/test_progress_monitor.py | 8 +- 2 files changed, 168 insertions(+), 4 deletions(-) create mode 100644 gsd-openclaw/src/gsd_openclaw/progress_monitor.py diff --git a/gsd-openclaw/src/gsd_openclaw/progress_monitor.py b/gsd-openclaw/src/gsd_openclaw/progress_monitor.py new file mode 100644 index 0000000..599f777 --- /dev/null +++ b/gsd-openclaw/src/gsd_openclaw/progress_monitor.py @@ -0,0 +1,164 @@ +""" +Progress monitor for tracking and reporting GSD operation progress. + +This module provides: +- Detection of progress indicators in Claude CLI output +- Throttled progress updates (1 minute intervals) +- Progress card creation for Feishu messages +""" +from __future__ import annotations + +import re +import time +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Tuple, Pattern + +from gsd_openclaw.response_formatter import FeishuResponseFormatter + + +@dataclass +class ProgressInfo: + """Represents extracted progress information. + + Attributes: + type: Type of progress indicator ('percentage', 'fraction', 'phase', 'completion'). + value: Numeric value (for percentage or phase). + current: Current count (for fraction type). + total: Total count (for fraction type). + """ + type: str + value: Optional[int] = None + current: Optional[int] = None + total: Optional[int] = None + + +class ProgressMonitor: + """Monitors and reports progress for GSD operations. + + This class provides functionality to: + - Extract progress indicators from output text + - Throttle progress updates to specified intervals + - Create progress cards for Feishu messages + + Attributes: + PROGRESS_PATTERNS: List of (pattern, type) tuples for progress detection. + update_interval: Minimum seconds between progress updates (default 60.0). + last_update_time: Unix timestamp of last update. + """ + + # Patterns for detecting progress in output + # Each tuple is (regex pattern, progress type) + PROGRESS_PATTERNS: List[Tuple[Pattern, str]] = [ + (re.compile(r'(\d+)%'), 'percentage'), + (re.compile(r'\[(\d+)/(\d+)(?:[^\]]*)\]'), 'fraction'), # Allow content after fraction + (re.compile(r'Phase\s+(\d+)', re.IGNORECASE), 'phase'), + (re.compile(r'完成|completed|done', re.IGNORECASE), 'completion'), + ] + + def __init__( + self, + formatter: FeishuResponseFormatter, + update_interval: float = 60.0 + ): + """Initialize ProgressMonitor. + + Args: + formatter: FeishuResponseFormatter instance for creating cards. + update_interval: Minimum seconds between progress updates (default 60.0). + """ + self.formatter = formatter + self.update_interval = update_interval + self.last_update_time: float = 0.0 + + def extract_progress(self, output: str) -> Optional[ProgressInfo]: + """Extract progress information from output text. + + Scans output for known progress patterns and returns the first match. + + Args: + output: The output text to scan for progress indicators. + + Returns: + ProgressInfo if progress pattern found, None otherwise. + """ + for pattern, progress_type in self.PROGRESS_PATTERNS: + match = pattern.search(output) + if match: + if progress_type == 'percentage': + return ProgressInfo( + type='percentage', + value=int(match.group(1)), + ) + elif progress_type == 'fraction': + return ProgressInfo( + type='fraction', + current=int(match.group(1)), + total=int(match.group(2)), + ) + elif progress_type == 'phase': + return ProgressInfo( + type='phase', + value=int(match.group(1)), + ) + elif progress_type == 'completion': + return ProgressInfo( + type='completion', + value=100, # Completion means 100% + ) + + return None + + def should_send_update(self) -> bool: + """Check if enough time has elapsed for a progress update. + + Returns: + True if update_interval has elapsed since last update, False otherwise. + Returns False if no update has been sent yet (first call). + """ + # If no update has been sent yet (last_update_time is 0), return False + if self.last_update_time == 0.0: + return False + + current_time = time.time() + elapsed = current_time - self.last_update_time + return elapsed >= self.update_interval + + def create_progress_card( + self, + progress: ProgressInfo, + message: str + ) -> dict: + """Create a Feishu progress card. + + Args: + progress: ProgressInfo with extracted progress data. + message: Additional message to include in the card. + + Returns: + Feishu Card JSON dict with progress information. + """ + # Format progress content + if progress.type == 'percentage': + progress_text = f"进度: {progress.value}%" + elif progress.type == 'fraction': + progress_text = f"进度: [{progress.current}/{progress.total}]" + elif progress.type == 'phase': + progress_text = f"阶段: {progress.value}" + elif progress.type == 'completion': + progress_text = "完成" + else: + progress_text = f"进度: {progress.value}" + + content = f"{message}\n\n{progress_text}" + + # Update last_update_time when creating card + self.last_update_time = time.time() + + return self.formatter.format_info( + title="正在执行...", + content=content, + ) + + def reset_timer(self) -> None: + """Reset the update timer (useful for new operations).""" + self.last_update_time = 0.0 diff --git a/gsd-openclaw/tests/test_progress_monitor.py b/gsd-openclaw/tests/test_progress_monitor.py index 99fa94e..ddf2345 100644 --- a/gsd-openclaw/tests/test_progress_monitor.py +++ b/gsd-openclaw/tests/test_progress_monitor.py @@ -106,8 +106,8 @@ class TestProgressMonitorCardCreation: card = monitor.create_progress_card(progress, "Processing data...") formatter.format_info.assert_called_once() - call_args = formatter.format_info.call_args - content = call_args[0][1] + call_kwargs = formatter.format_info.call_args.kwargs + content = call_kwargs.get("content", "") assert "75%" in content def test_create_progress_card_uses_executing_title(self): @@ -126,6 +126,6 @@ class TestProgressMonitorCardCreation: card = monitor.create_progress_card(progress, "Working...") formatter.format_info.assert_called_once() - call_args = formatter.format_info.call_args - title = call_args[0][0] + call_kwargs = formatter.format_info.call_args.kwargs + title = call_kwargs.get("title", "") assert title == "正在执行..."