openclaw/skills/gsd-openclaw/src/gsd_openclaw/progress_monitor.py

"""
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