XFE Git
XFE Studio Git
Git 首页 全局搜索
XFE 主站 文档 NuGet
公开
关注 0 Fork 0 Star 1
UTF-8
import re
import sys
import os
import shutil
import subprocess
import tempfile
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional


@dataclass
class PatchResult:
    """Result of a patch application operation."""

    success: bool
    output: str = ""
    error: Optional[str] = None
    files_changed: list[str] = field(default_factory=list)


def _find_patch_command() -> Optional[str]:
    """
    Locate the patch command on the system.
    Returns the command path or None if not found.
    """
    # Check if patch is available in PATH
    if shutil.which("patch"):
        return "patch"

    # Windows-specific locations
    if sys.platform == "win32":
        # Common Git for Windows locations
        git_locations = [
            r"C:\Program Files\Git\usr\bin\patch.exe",
            r"C:\Program Files (x86)\Git\usr\bin\patch.exe",
            r"C:\msys64\usr\bin\patch.exe",
        ]
        for loc in git_locations:
            if Path(loc).exists():
                return loc

    return None


def patch_is_available() -> bool:
    """Check if the patch command is available on the system"""
    return _find_patch_command() is not None


def apply_patch(
    patch_content: str,
    target_file: str,
    backup: bool = True,
    dry_run: bool = False,
    strip: int = 0,
) -> PatchResult:
    """
    Apply a unified diff patch. Falls back to Python if system patch unavailable.
    """
    # Check patch availability first
    patch_cmd = _find_patch_command()

    if patch_cmd is None:
        # Fallback to pure Python implementation
        return _apply_patch_python(patch_content, target_file, backup, dry_run)

    # Use system patch command
    return _apply_patch_system(
        patch_cmd, patch_content, target_file, backup, dry_run, strip
    )


def _apply_patch_system(
    patch_cmd: str,
    patch_content: str,
    target_file: str,
    backup: bool,
    dry_run: bool,
    strip: int,
) -> PatchResult:
    """Apply patch using system patch command"""
    target_path = Path(target_file)

    if not target_path.exists() and not dry_run:
        return PatchResult(
            success=False, output="", error=f"Target file not found: {target_file}"
        )

    patch_file = None
    try:
        # Normalize line endings to Unix-style (LF) for compatibility
        normalized = patch_content.replace("\r\n", "\n").replace("\r", "\n")
        # Ensure patch ends with a newline to avoid "unexpectedly ends in middle of line"
        if normalized and not normalized.endswith("\n"):
            normalized += "\n"

        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".patch", delete=False, encoding="utf-8", newline="\n"
        ) as f:
            f.write(normalized)
            patch_file = f.name

        cmd = [patch_cmd, "--binary"]
        if dry_run:
            cmd.append("--dry-run")
        if backup:
            cmd.append("--backup")

        cmd.extend(["-p", str(strip)])
        cmd.extend(["-i", patch_file])

        if target_path.is_dir():
            cmd.extend(["-d", str(target_path)])
        else:
            cmd.append(str(target_path))

        process = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=30,
            cwd=str(target_path.parent) if target_path.is_file() else None,
        )

        output = process.stdout + process.stderr

        if process.returncode == 0:
            return PatchResult(
                success=True, output=output, files_changed=_parse_patched_files(output)
            )
        else:
            return PatchResult(
                success=False, output=output, error=_extract_error(output)
            )

    except subprocess.TimeoutExpired:
        return PatchResult(
            success=False, output="", error="Patch application timed out"
        )
    except Exception as e:
        return PatchResult(
            success=False, output="", error=f"System patching error: {str(e)}"
        )
    finally:
        if patch_file and os.path.exists(patch_file):
            try:
                os.unlink(patch_file)
            except OSError:
                pass


def _apply_patch_python(
    patch_content: str, target_file: str, backup: bool = False, dry_run: bool = False
) -> PatchResult:
    """
    Pure Python patch implementation as fallback.
    Handles basic unified diffs without external dependencies.
    """
    import difflib
    import re

    target_path = Path(target_file)

    if not target_path.exists():
        return PatchResult(
            success=False, output="", error=f"Target file not found: {target_file}"
        )

    if not target_path.is_file():
        return PatchResult(
            success=False,
            output="",
            error=f"Python fallback only supports single files, got: {target_file}",
        )

    try:
        # Read original content
        original = target_path.read_text(encoding="utf-8")
        original_lines = original.splitlines(True)

        # Parse the unified diff
        patched_lines = _parse_and_apply_unified_diff(original_lines, patch_content)

        if patched_lines is None:
            return PatchResult(
                success=False, output="", error="Failed to parse or apply patch"
            )

        # Check if any changes
        changed = original_lines != patched_lines

        if not changed:
            return PatchResult(
                success=True, output="Patch already applied", files_changed=[]
            )

        if dry_run:
            return PatchResult(
                success=True,
                output=f"Dry run successful: {len(patched_lines)} lines",
                files_changed=[str(target_path)],
            )

        # Apply changes
        if backup:
            backup_path = target_path.with_suffix(target_path.suffix + ".orig")
            backup_path.write_text(original, encoding="utf-8")

        target_path.write_text("".join(patched_lines), encoding="utf-8")

        return PatchResult(
            success=True,
            output=f"Successfully patched {target_file}",
            files_changed=[str(target_path)],
        )

    except Exception as e:
        return PatchResult(
            success=False, output="", error=f"Python patching error: {str(e)}"
        )


def _parse_and_apply_unified_diff(
    original_lines: list[str], diff_text: str
) -> Optional[list[str]]:
    """
    Minimal unified diff parser and applier.
    Returns patched lines or None on failure.
    """
    # Parse hunks from unified diff
    hunks = re.findall(
        r"@@ -(\d+),?(\d*) \+(\d+),?(\d*) @@(.*?)(?=@@|\Z)", diff_text, re.DOTALL
    )

    if not hunks:
        return None

    result_lines = original_lines.copy()

    for hunk in reversed(hunks):  # Apply in reverse to maintain positions
        old_start, old_count, new_start, new_count, body = hunk
        old_start = int(old_start) - 1  # Convert to 0-based
        old_count = int(old_count) if old_count else 1

        # Parse hunk body
        body_lines = body.split("\n")[1:]  # Skip first empty line
        hunk_result = []

        old_pos = old_start
        for line in body_lines:
            if not line:
                continue

            if line.startswith("+"):
                hunk_result.append(line[1:] + "\n")
            elif line.startswith("-"):
                old_pos += 1
            elif line.startswith(" "):
                if old_pos < len(result_lines):
                    hunk_result.append(result_lines[old_pos])
                old_pos += 1
            else:
                # Context line
                pass

        # Replace the hunk region
        end_index = min(old_start + old_count, len(result_lines))
        result_lines[old_start:end_index] = hunk_result

    return result_lines


def _parse_patched_files(output: str) -> list[str]:
    """Extract patched file names from patch command output."""
    files = []
    for line in output.splitlines():
        if line.startswith("patching file "):
            files.append(line[len("patching file ") :].strip())
    return files


def _extract_error(output: str) -> str:
    """Extract error message from patch command output."""
    for line in output.splitlines():
        if "FAILED" in line or "failed" in line or "error" in line.lower():
            return line.strip()
    return output.strip() or "Unknown patch error"


def apply_patch_with_fallback(
    patch_content: str, target_file: str, backup: bool = True, dry_run: bool = False
) -> dict:
    """
    Apply patch with automatic fallback detection.
    Tries system patch first, falls back to Python implementation on failure.
    Returns structured result.
    """
    # Check system capabilities
    using_python = not patch_is_available()

    if using_python:
        # System patch not available, use Python directly
        result = apply_patch(patch_content, target_file, backup, dry_run)
    else:
        # Try system patch first
        result = apply_patch(patch_content, target_file, backup, dry_run)
        # If system patch failed, try Python fallback
        if not result.success:
            result = _apply_patch_python(patch_content, target_file, backup, dry_run)
            using_python = True

    return {
        "success": result.success,
        "output": result.output,
        "error": result.error,
        "files_changed": result.files_changed,
        "implementation": "python_fallback" if using_python else "system_patch",
    }