debug_env/server/tools/advanced_tools.py

"""
Advanced debugging tools for code analysis and inspection.

Provides functions for:
- Code search (grep-like pattern matching)
- File structure analysis (AST parsing)
- Type checking (mypy integration)
- Test coverage analysis (coverage.py integration)
- Directory listing and navigation
- Import dependency analysis
"""

import ast
import json
import logging
import os
import re
import subprocess
from pathlib import Path
from typing import Any, Dict, List

logger = logging.getLogger(__name__)


def search_code(
    workdir: str,
    pattern: str,
    file_types: str = "py",
    case_sensitive: bool = False,
    max_results: int = 50,
    context_lines: int = 2,
) -> Dict[str, Any]:
    """
    Search for code patterns across files in the task directory.

    Args:
        workdir: Absolute path to task working directory
        pattern: Text or regex pattern to search for
        file_types: File type filter ("py", "all", "txt", "json", "md")
        case_sensitive: Whether to match case
        max_results: Maximum matches to return (1-500)
        context_lines: Lines of context around each match (0-10)

    Returns:
        Dict with matches, match count, files searched, and timing info
    """
    try:
        workdir_path = Path(workdir)
        if not workdir_path.is_dir():
            return {"error": f"Invalid directory: {workdir}", "matches": [], "total_matches": 0}

        # Determine file extensions to search
        if file_types == "all":
            search_patterns = ["*"]
        elif file_types == "py":
            search_patterns = ["*.py"]
        else:
            search_patterns = [f"*.{file_types}"]

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

        matches = []
        files_searched = 0

        # Search files
        for search_pattern in search_patterns:
            for file_path in workdir_path.rglob(search_pattern):
                if not file_path.is_file():
                    continue

                files_searched += 1
                try:
                    content = file_path.read_text(encoding="utf-8")
                    lines = content.split("\n")

                    for line_num, line in enumerate(lines, 1):
                        for match in compiled_pattern.finditer(line):
                            if len(matches) >= max_results:
                                break

                            matches.append({
                                "filename": str(file_path.relative_to(workdir_path)),
                                "line_number": line_num,
                                "line_content": line,
                                "match_position": match.start(),
                                "before": lines[max(0, line_num - context_lines - 1):line_num - 1],
                                "after": lines[line_num:min(len(lines), line_num + context_lines)],
                            })
                except (UnicodeDecodeError, OSError):
                    continue

        logger.info(f"Search pattern '{pattern}': found {len(matches)} matches in {files_searched} files")

        return {
            "matches": matches,
            "total_matches": len(matches),
            "files_searched": files_searched,
            "search_time_ms": 0,  # Would be measured in production
        }

    except Exception as e:
        logger.error(f"Error searching code: {e}")
        return {"error": str(e), "matches": [], "total_matches": 0}


def get_file_structure(
    workdir: str,
    path: str,
    include_docstrings: bool = True,
    include_signatures: bool = True,
    max_depth: int = 0,
) -> Dict[str, Any]:
    """
    Parse and return the structure (AST) of a Python file.

    Args:
        workdir: Absolute path to task working directory
        path: File path relative to workdir
        include_docstrings: Include docstrings in output
        include_signatures: Include function/method signatures
        max_depth: Maximum nesting depth (0 for unlimited)

    Returns:
        Dict with classes, functions, imports, and constants
    """
    try:
        file_path = Path(workdir) / path
        if not file_path.is_file():
            return {"error": f"File not found: {path}"}

        content = file_path.read_text(encoding="utf-8")
        tree = ast.parse(content)

        structure = {
            "filename": path,
            "imports": [],
            "classes": [],
            "functions": [],
            "constants": [],
        }

        for node in ast.walk(tree):
            # Extract imports
            if isinstance(node, ast.Import):
                for alias in node.names:
                    structure["imports"].append({
                        "module": alias.name,
                        "items": [alias.asname or alias.name],
                        "line_number": node.lineno,
                    })

            elif isinstance(node, ast.ImportFrom):
                items = [alias.name for alias in node.names]
                structure["imports"].append({
                    "module": node.module or "",
                    "items": items,
                    "line_number": node.lineno,
                })

            # Extract classes
            elif isinstance(node, ast.ClassDef):
                class_info = {
                    "name": node.name,
                    "line_number": node.lineno,
                    "methods": [],
                    "attributes": [],
                }

                if include_docstrings:
                    class_info["docstring"] = ast.get_docstring(node)

                for item in node.body:
                    if isinstance(item, ast.FunctionDef):
                        class_info["methods"].append({
                            "name": item.name,
                            "line_number": item.lineno,
                        })

                structure["classes"].append(class_info)

        # Extract top-level functions
        for node in tree.body:
            if isinstance(node, ast.FunctionDef):
                func_info = {
                    "name": node.name,
                    "line_number": node.lineno,
                }

                if include_signatures:
                    func_info["parameters"] = [arg.arg for arg in node.args.args]

                if include_docstrings:
                    func_info["docstring"] = ast.get_docstring(node)

                structure["functions"].append(func_info)

        logger.info(f"Extracted structure from {path}")
        return structure

    except SyntaxError as e:
        logger.error(f"Syntax error in {path}: {e}")
        return {"error": f"Syntax error: {e}"}
    except Exception as e:
        logger.error(f"Error getting file structure: {e}")
        return {"error": str(e)}


def run_type_check(
    workdir: str,
    path: str,
    strict: bool = False,
    ignore_missing_imports: bool = True,
    show_error_codes: bool = True,
    max_errors: int = 100,
) -> Dict[str, Any]:
    """
    Run static type checking (mypy) on Python files.

    Args:
        workdir: Absolute path to task working directory
        path: File or directory path relative to workdir
        strict: Enable strict type checking mode
        ignore_missing_imports: Ignore untyped third-party imports
        show_error_codes: Show mypy error codes
        max_errors: Maximum errors to report

    Returns:
        Dict with type check results and error list
    """
    try:
        check_path = Path(workdir) / path
        if not check_path.exists():
            return {"error": f"Path not found: {path}", "status": "error"}

        cmd = ["mypy", str(check_path)]

        if strict:
            cmd.append("--strict")
        if ignore_missing_imports:
            cmd.append("--ignore-missing-imports")
        if show_error_codes:
            cmd.append("--show-error-codes")

        cmd.extend(["--no-error-summary", f"--max-errors={max_errors}"])

        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            cwd=workdir,
            timeout=30,
        )

        # Parse output
        errors = []
        for line in result.stdout.split("\n"):
            if not line.strip():
                continue

            # mypy output format: file:line:column: error: message [error-code]
            match = re.match(
                r"^([^:]+):(\d+):(\d+): error: (.+)(?:\s+\[([^\]]+)\])?$",
                line
            )
            if match:
                errors.append({
                    "file": match.group(1),
                    "line": int(match.group(2)),
                    "column": int(match.group(3)),
                    "message": match.group(4),
                    "error_code": match.group(5),
                })

        status = "success" if result.returncode == 0 else "errors_found"

        logger.info(f"Type check {path}: {len(errors)} errors, status={status}")

        return {
            "path": path,
            "status": status,
            "error_count": len(errors),
            "errors": errors,
            "check_time_ms": 0,
        }

    except subprocess.TimeoutExpired:
        logger.error(f"Type check timeout for {path}")
        return {"error": "Type check timeout", "status": "error"}
    except Exception as e:
        logger.error(f"Error running type check: {e}")
        return {"error": str(e), "status": "error"}


def get_test_coverage(
    workdir: str,
    path: str = ".",
    branch_coverage: bool = False,
    include_missing: bool = True,
    fail_under: int = 0,
) -> Dict[str, Any]:
    """
    Analyze test coverage for Python files.

    Args:
        workdir: Absolute path to task working directory
        path: File or directory path relative to workdir
        branch_coverage: Include branch coverage analysis
        include_missing: Show which lines are not covered
        fail_under: Minimum required coverage percentage

    Returns:
        Dict with coverage percentage, file breakdown, and statistics
    """
    try:
        cmd = ["coverage", "run", "-m", "pytest", workdir]

        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            cwd=workdir,
            timeout=60,
        )

        # Get coverage report
        cmd = ["coverage", "json", "--pretty-print"]
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            cwd=workdir,
            timeout=30,
        )

        coverage_data = json.loads(result.stdout)
        total_coverage = coverage_data.get("totals", {}).get("percent_covered", 0)

        logger.info(f"Coverage for {path}: {total_coverage}%")

        return {
            "total_coverage": round(total_coverage, 1),
            "files": [],
            "functions": [],
            "status": "covered" if total_coverage >= fail_under else "uncovered",
            "check_time_ms": 0,
        }

    except Exception as e:
        logger.error(f"Error analyzing coverage: {e}")
        return {"error": str(e), "total_coverage": 0, "status": "error"}


def list_directory(
    workdir: str,
    path: str = ".",
    recursive: bool = True,
    max_depth: int = 0,
    include_hidden: bool = False,
    file_types: str = "all",
) -> Dict[str, Any]:
    """
    List files and directories in the task working directory.

    Args:
        workdir: Absolute path to task working directory
        path: Subdirectory path relative to workdir
        recursive: Include subdirectories recursively
        max_depth: Maximum directory nesting depth (0 for unlimited)
        include_hidden: Include hidden files/folders starting with .
        file_types: Filter by file type ("py", "all", "test", "source")

    Returns:
        Dict with directory tree, file count, and metadata
    """
    try:
        root_path = Path(workdir) / path
        if not root_path.is_dir():
            return {"error": f"Directory not found: {path}"}

        def walk_tree(directory: Path, depth: int = 0) -> List[Dict]:
            if max_depth > 0 and depth >= max_depth:
                return []

            items = []
            try:
                for entry in sorted(directory.iterdir()):
                    # Skip hidden files if needed
                    if entry.name.startswith(".") and not include_hidden:
                        continue

                    # Filter by file type
                    if file_types == "py" and entry.is_file() and not entry.suffix == ".py":
                        continue
                    elif file_types == "test" and entry.is_file() and not entry.name.startswith("test_"):
                        continue

                    item = {
                        "name": entry.name,
                        "type": "directory" if entry.is_dir() else "file",
                        "path": str(entry.relative_to(Path(workdir))),
                    }

                    if entry.is_file():
                        item["size_bytes"] = entry.stat().st_size
                        item["is_test"] = entry.name.startswith("test_")

                    if entry.is_dir() and recursive:
                        item["contents"] = walk_tree(entry, depth + 1)

                    items.append(item)
            except PermissionError:
                pass

            return items

        items = walk_tree(root_path)

        # Count files
        total_files = sum(1 for entry in root_path.rglob("*") if entry.is_file())
        total_dirs = sum(1 for entry in root_path.rglob("*") if entry.is_dir())

        logger.info(f"Listed directory {path}: {total_files} files, {total_dirs} directories")

        return {
            "root_path": path,
            "total_files": total_files,
            "total_directories": total_dirs,
            "items": items,
            "tree_depth": max_depth,
        }

    except Exception as e:
        logger.error(f"Error listing directory: {e}")
        return {"error": str(e)}


def get_dependencies(
    workdir: str,
    path: str,
    include_stdlib: bool = False,
    include_external: bool = True,
    direction: str = "to",
    depth: int = 1,
) -> Dict[str, Any]:
    """
    Extract and analyze Python import dependencies.

    Args:
        workdir: Absolute path to task working directory
        path: File or directory path relative to workdir
        include_stdlib: Include Python standard library imports
        include_external: Include third-party library imports
        direction: Show imports "to" or "from"
        depth: Maximum depth to follow dependencies

    Returns:
        Dict with dependency graph, cycles, and unused imports
    """
    try:
        target_path = Path(workdir) / path
        if not target_path.exists():
            return {"error": f"Path not found: {path}"}

        dependencies = {
            "path": path,
            "direct_dependencies": [],
            "indirect_dependencies": [],
            "circular_dependencies": [],
            "unused_imports": [],
            "dependency_count": 0,
        }

        # Find Python files to analyze
        if target_path.is_file():
            py_files = [target_path]
        else:
            py_files = list(target_path.rglob("*.py"))

        # Parse imports from each file
        for py_file in py_files:
            try:
                content = py_file.read_text(encoding="utf-8")
                tree = ast.parse(content)

                for node in ast.walk(tree):
                    if isinstance(node, ast.Import):
                        for alias in node.names:
                            dependencies["direct_dependencies"].append({
                                "module": alias.name,
                                "type": "import",
                                "line_number": node.lineno,
                                "is_local": False,
                            })

                    elif isinstance(node, ast.ImportFrom):
                        dependencies["direct_dependencies"].append({
                            "module": node.module or "",
                            "type": "from...import",
                            "items": [alias.name for alias in node.names],
                            "line_number": node.lineno,
                            "is_local": not (node.module and ("." not in node.module)),
                        })

            except (SyntaxError, UnicodeDecodeError):
                continue

        dependencies["dependency_count"] = len(dependencies["direct_dependencies"])

        logger.info(f"Analyzed dependencies for {path}: {dependencies['dependency_count']} imports")

        return dependencies

    except Exception as e:
        logger.error(f"Error analyzing dependencies: {e}")
        return {"error": str(e)}