utils.py

"""
Utility functions for the AI Web Automation Agent System.
Provides helper functions for logging, file operations, and data processing.
"""

import json
import logging
from pathlib import Path
from typing import Dict, Any, List, Optional
from datetime import datetime
import re


def setup_logger(name: str, log_file: Optional[str] = None, level: str = "INFO") -> logging.Logger:
    """
    Set up a logger with console and optional file output.

    Args:
        name: Logger name
        log_file: Optional log file path
        level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)

    Returns:
        Configured logger instance
    """
    logger = logging.getLogger(name)
    logger.setLevel(getattr(logging, level.upper()))

    # Console handler
    console_handler = logging.StreamHandler()
    console_handler.setLevel(getattr(logging, level.upper()))
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )
    console_handler.setFormatter(formatter)
    logger.addHandler(console_handler)

    # File handler (optional)
    if log_file:
        file_handler = logging.FileHandler(log_file)
        file_handler.setLevel(getattr(logging, level.upper()))
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)

    return logger


def sanitize_filename(filename: str) -> str:
    """
    Sanitize a filename by removing/replacing invalid characters.

    Args:
        filename: Original filename

    Returns:
        Sanitized filename safe for filesystem use
    """
    # Replace spaces with underscores
    filename = filename.replace(" ", "_")
    # Remove invalid characters
    filename = re.sub(r'[<>:"/\\|?*]', '', filename)
    # Convert to lowercase
    filename = filename.lower()
    # Limit length
    if len(filename) > 200:
        filename = filename[:200]
    return filename


def save_json(data: Dict[str, Any], filepath: Path) -> None:
    """
    Save dictionary data to a JSON file.

    Args:
        data: Dictionary to save
        filepath: Path to save JSON file
    """
    filepath.parent.mkdir(parents=True, exist_ok=True)
    with open(filepath, 'w', encoding='utf-8') as f:
        json.dump(data, f, indent=2, ensure_ascii=False)


def load_json(filepath: Path) -> Dict[str, Any]:
    """
    Load JSON data from a file.

    Args:
        filepath: Path to JSON file

    Returns:
        Dictionary with loaded data
    """
    if not filepath.exists():
        raise FileNotFoundError(f"JSON file not found: {filepath}")

    with open(filepath, 'r', encoding='utf-8') as f:
        return json.load(f)


def get_timestamp() -> str:
    """
    Get current timestamp in ISO format.

    Returns:
        ISO format timestamp string
    """
    return datetime.now().isoformat()


def format_step_name(step_number: int, action: str, description: str) -> str:
    """
    Format a step name for screenshot naming.

    Args:
        step_number: Step sequence number
        action: Action type (click, fill, navigate, etc.)
        description: Brief description of the step

    Returns:
        Formatted step name
    """
    sanitized_desc = sanitize_filename(description)
    return f"step_{step_number:02d}_{action}_{sanitized_desc}"


def extract_domain(url: str) -> str:
    """
    Extract domain from a URL.

    Args:
        url: Full URL

    Returns:
        Domain name
    """
    from urllib.parse import urlparse
    parsed = urlparse(url)
    return parsed.netloc


def parse_selector(selector_text: str) -> Dict[str, str]:
    """
    Parse a selector string into type and value.

    Args:
        selector_text: Selector string (e.g., "text:Create Project", "role:button")

    Returns:
        Dictionary with 'type' and 'value' keys
    """
    if ":" in selector_text:
        selector_type, value = selector_text.split(":", 1)
        return {"type": selector_type.strip(), "value": value.strip()}
    return {"type": "text", "value": selector_text.strip()}


def truncate_text(text: str, max_length: int = 100) -> str:
    """
    Truncate text to a maximum length with ellipsis.

    Args:
        text: Text to truncate
        max_length: Maximum length

    Returns:
        Truncated text
    """
    if len(text) <= max_length:
        return text
    return text[:max_length - 3] + "..."


def merge_dicts(*dicts: Dict[str, Any]) -> Dict[str, Any]:
    """
    Merge multiple dictionaries into one.

    Args:
        *dicts: Variable number of dictionaries to merge

    Returns:
        Merged dictionary
    """
    result = {}
    for d in dicts:
        result.update(d)
    return result


def ensure_list(item: Any) -> List[Any]:
    """
    Ensure item is a list. If not, wrap it in a list.

    Args:
        item: Any item

    Returns:
        List containing the item(s)
    """
    if isinstance(item, list):
        return item
    return [item]


def calculate_similarity(str1: str, str2: str) -> float:
    """
    Calculate similarity ratio between two strings using Levenshtein distance.

    Args:
        str1: First string
        str2: Second string

    Returns:
        Similarity ratio (0.0 to 1.0)
    """
    from difflib import SequenceMatcher
    return SequenceMatcher(None, str1.lower(), str2.lower()).ratio()


def find_best_match(target: str, candidates: List[str], threshold: float = 0.6) -> Optional[str]:
    """
    Find the best matching string from a list of candidates.

    Args:
        target: Target string to match
        candidates: List of candidate strings
        threshold: Minimum similarity threshold (0.0 to 1.0)

    Returns:
        Best matching candidate or None if no match above threshold
    """
    if not candidates:
        return None

    best_match = None
    best_score = 0.0

    for candidate in candidates:
        score = calculate_similarity(target, candidate)
        if score > best_score:
            best_score = score
            best_match = candidate

    return best_match if best_score >= threshold else None


def create_task_directory(base_dir: Path, app_name: str, task_name: str) -> Path:
    """
    Create a directory structure for a task.

    Args:
        base_dir: Base directory for all datasets
        app_name: Application name
        task_name: Task name

    Returns:
        Path to the created task directory
    """
    sanitized_app = sanitize_filename(app_name)
    sanitized_task = sanitize_filename(task_name)
    task_dir = base_dir / sanitized_app / sanitized_task
    task_dir.mkdir(parents=True, exist_ok=True)
    return task_dir