webber/webber-api/src/domains/tools/search/grep.py

"""
Content search tool using regex patterns.
"""
import re
from pathlib import Path
from typing import Literal

from src.domains.tools.base import BaseTool, ToolResult
from src.domains.tools.gitignore import filter_gitignored
from src.shared.logging import get_logger, logged

logger = get_logger(__name__)


class GrepContentTool(BaseTool):
    """
    Search file contents using regex patterns.

    Similar to grep/ripgrep but implemented in Python for portability.
    """

    name = "grep_content"
    description = """Search file contents using regex pattern.

Args:
    pattern: Regex pattern to search for (Python re syntax)
    path: Directory or file to search (default: working directory)
    file_glob: Filter files by glob pattern (e.g., "*.py", "*.ts")
    context_lines: Lines of context before/after matches (default: 0)
    case_sensitive: Whether search is case-sensitive (default: True)
    output_mode: "content" for matching lines, "files" for file paths only
    honor_gitignore: Filter out gitignored files (default: True)

Returns:
    Matching lines with file paths and line numbers, or list of files.
    Format: "filepath:line_num: content"
    By default, excludes files matching .gitignore patterns and common ignored
    directories like .venv/, node_modules/, __pycache__/, etc.

Examples:
    - pattern="def.*init" file_glob="*.py" - Find init methods in Python files
    - pattern="TODO" - Find all TODO comments
    - pattern="class\\s+\\w+" - Find class definitions

IMPORTANT:
- Use this tool to search for code patterns
- Escape special regex characters (\\, ., *, etc.)
- Never guess where code is - use grep to find it
"""

    def __init__(
        self,
        allowed_paths: list[str] | None = None,
        max_results: int = 100,
        max_file_size: int = 1_000_000,  # 1MB
        honor_gitignore: bool = True
    ):
        """
        Initialize GrepContentTool.

        Args:
            allowed_paths: List of allowed directory prefixes
            max_results: Maximum matches to return
            max_file_size: Skip files larger than this (bytes)
            honor_gitignore: Whether to filter out gitignored files by default
        """
        self.allowed_paths = allowed_paths or []
        self.max_results = max_results
        self.max_file_size = max_file_size
        self.honor_gitignore = honor_gitignore

    @logged()
    async def execute(  # type: ignore[override]  # see BaseTool.execute
        self,
        pattern: str,
        path: str | None = None,
        file_glob: str | None = None,
        context_lines: int = 0,
        case_sensitive: bool = True,
        output_mode: Literal["content", "files"] = "content",
        honor_gitignore: bool | None = None
    ) -> ToolResult:
        """
        Search for pattern in files.

        Args:
            pattern: Regex pattern to search
            path: Directory or file to search
            file_glob: Filter to files matching glob
            context_lines: Context lines around matches
            case_sensitive: Case-sensitive search
            output_mode: "content" or "files"
            honor_gitignore: Filter out gitignored files (default: instance setting)

        Returns:
            ToolResult with matching content or file list
        """
        should_filter_gitignore = honor_gitignore if honor_gitignore is not None else self.honor_gitignore
        search_path = Path(path) if path else Path.cwd()

        # Validate path
        if not self._validate_path(search_path, self.allowed_paths):
            return self._error(f"Path not in allowed paths: {search_path}")

        if not search_path.exists():
            return self._error(f"Path not found: {search_path}")

        # Compile regex
        try:
            flags = 0 if case_sensitive else re.IGNORECASE
            regex = re.compile(pattern, flags)
        except re.error as e:
            return self._error(f"Invalid regex pattern: {e}")

        # Collect files to search
        if search_path.is_file():
            files_to_search = [search_path]
        else:
            glob_pattern = file_glob or "**/*"
            files_to_search = [
                f for f in search_path.glob(glob_pattern)
                if f.is_file()
            ]

        # Filter by allowed paths
        if self.allowed_paths:
            files_to_search = [
                f for f in files_to_search
                if self._validate_path(f, self.allowed_paths)
            ]

        # Filter out gitignored files
        if should_filter_gitignore:
            files_to_search = filter_gitignored(files_to_search, search_path)

        # Search files
        matches = []
        files_with_matches = set()
        total_matches = 0

        for file_path in files_to_search:
            # Skip large files
            try:
                if file_path.stat().st_size > self.max_file_size:
                    continue
            except OSError:
                continue

            # Skip binary files (heuristic)
            if self._is_likely_binary(file_path):
                continue

            file_matches = await self._search_file(
                file_path, regex, context_lines
            )

            if file_matches:
                files_with_matches.add(str(file_path.resolve()))
                total_matches += len(file_matches)
                matches.extend(file_matches)

                # Check result limit
                if len(matches) >= self.max_results:
                    break

        # Format output
        truncated = total_matches > self.max_results

        if output_mode == "files":
            result = "\n".join(sorted(files_with_matches))
            if not result:
                result = f"No files found matching pattern '{pattern}'"
        else:
            result = "\n".join(matches[:self.max_results])
            if not result:
                result = f"No matches found for pattern '{pattern}'"

        return self._success(
            data=result,
            truncated=truncated,
            total_matches=total_matches,
            files_matched=len(files_with_matches)
        )

    async def _search_file(
        self,
        file_path: Path,
        regex: re.Pattern,
        context_lines: int
    ) -> list[str]:
        """Search a single file for matches."""
        try:
            content = file_path.read_text(encoding='utf-8', errors='replace')
            lines = content.splitlines()
        except (PermissionError, UnicodeDecodeError, OSError):
            return []

        matches = []
        matched_line_nums = set()

        # Find all matching lines
        for i, line in enumerate(lines):
            if regex.search(line):
                matched_line_nums.add(i)

        # Add context and format
        for match_num in sorted(matched_line_nums):
            start = max(0, match_num - context_lines)
            end = min(len(lines), match_num + context_lines + 1)

            for i in range(start, end):
                prefix = ">" if i == match_num else " "
                line_num = i + 1  # 1-based
                formatted = f"{file_path}:{line_num}:{prefix} {lines[i]}"
                matches.append(formatted)

            # Add separator between match groups
            if context_lines > 0:
                matches.append("--")

        # Remove trailing separator
        if matches and matches[-1] == "--":
            matches.pop()

        return matches

    def _is_likely_binary(self, file_path: Path) -> bool:
        """Check if file is likely binary based on extension or content."""
        binary_extensions = {
            '.pyc', '.pyo', '.so', '.dll', '.exe', '.bin',
            '.png', '.jpg', '.jpeg', '.gif', '.ico', '.svg',
            '.pdf', '.zip', '.tar', '.gz', '.bz2', '.xz',
            '.woff', '.woff2', '.ttf', '.eot',
            '.mp3', '.mp4', '.wav', '.avi', '.mov',
            '.db', '.sqlite', '.sqlite3',
        }

        if file_path.suffix.lower() in binary_extensions:
            return True

        # Check first bytes for null characters
        try:
            with open(file_path, 'rb') as f:
                chunk = f.read(1024)
                if b'\x00' in chunk:
                    return True
        except (PermissionError, OSError):
            return True

        return False