返回提交历史
Added
g4f/mcp/apply_patch.py
+339
-0
Modified
g4f/mcp/server.py
+2
-0
Modified
g4f/mcp/tools.py
+67
-3
XFEstudio/gpt4free
feat: add ApplyPatchTool for applying unified diff patches with fallback support
883c7174
代码差异
3 个文件
+408
-3
@@ -0,0 +1,339 @@
1
import re
2
import sys
3
import os
4
import shutil
5
import subprocess
6
import tempfile
7
from dataclasses import dataclass, field
8
from pathlib import Path
9
from typing import Optional
10
11
12
@dataclass
13
class PatchResult:
14
"""Result of a patch application operation."""
15
success: bool
16
output: str = ""
17
error: Optional[str] = None
18
files_changed: list[str] = field(default_factory=list)
19
20
21
def _find_patch_command() -> Optional[str]:
22
"""
23
Locate the patch command on the system.
24
Returns the command path or None if not found.
25
"""
26
# Check if patch is available in PATH
27
if shutil.which('patch'):
28
return 'patch'
29
30
# Windows-specific locations
31
if sys.platform == 'win32':
32
# Common Git for Windows locations
33
git_locations = [
34
r'C:\Program Files\Git\usr\bin\patch.exe',
35
r'C:\Program Files (x86)\Git\usr\bin\patch.exe',
36
r'C:\msys64\usr\bin\patch.exe',
37
]
38
for loc in git_locations:
39
if Path(loc).exists():
40
return loc
41
42
return None
43
44
45
def patch_is_available() -> bool:
46
"""Check if the patch command is available on the system"""
47
return _find_patch_command() is not None
48
49
50
def apply_patch(
51
patch_content: str,
52
target_file: str,
53
backup: bool = True,
54
dry_run: bool = False,
55
strip: int = 0
56
) -> PatchResult:
57
"""
58
Apply a unified diff patch. Falls back to Python if system patch unavailable.
59
"""
60
# Check patch availability first
61
patch_cmd = _find_patch_command()
62
63
if patch_cmd is None:
64
# Fallback to pure Python implementation
65
return _apply_patch_python(patch_content, target_file, backup, dry_run)
66
67
# Use system patch command
68
return _apply_patch_system(
69
patch_cmd, patch_content, target_file, backup, dry_run, strip
70
)
71
72
73
def _apply_patch_system(
74
patch_cmd: str,
75
patch_content: str,
76
target_file: str,
77
backup: bool,
78
dry_run: bool,
79
strip: int
80
) -> PatchResult:
81
"""Apply patch using system patch command"""
82
target_path = Path(target_file)
83
84
if not target_path.exists() and not dry_run:
85
return PatchResult(
86
success=False,
87
output="",
88
error=f"Target file not found: {target_file}"
89
)
90
91
patch_file = None
92
try:
93
# Normalize line endings to Unix-style (LF) for compatibility
94
normalized = patch_content.replace('\r\n', '\n').replace('\r', '\n')
95
# Ensure patch ends with a newline to avoid "unexpectedly ends in middle of line"
96
if normalized and not normalized.endswith('\n'):
97
normalized += '\n'
98
99
with tempfile.NamedTemporaryFile(
100
mode='w', suffix='.patch', delete=False, encoding='utf-8', newline='\n'
101
) as f:
102
f.write(normalized)
103
patch_file = f.name
104
105
cmd = [patch_cmd, '--binary']
106
if dry_run:
107
cmd.append('--dry-run')
108
if backup:
109
cmd.append('--backup')
110
111
cmd.extend(['-p', str(strip)])
112
cmd.extend(['-i', patch_file])
113
114
if target_path.is_dir():
115
cmd.extend(['-d', str(target_path)])
116
else:
117
cmd.append(str(target_path))
118
119
process = subprocess.run(
120
cmd,
121
capture_output=True,
122
text=True,
123
timeout=30,
124
cwd=str(target_path.parent) if target_path.is_file() else None
125
)
126
127
output = process.stdout + process.stderr
128
129
if process.returncode == 0:
130
return PatchResult(
131
success=True,
132
output=output,
133
files_changed=_parse_patched_files(output)
134
)
135
else:
136
return PatchResult(
137
success=False,
138
output=output,
139
error=_extract_error(output)
140
)
141
142
except subprocess.TimeoutExpired:
143
return PatchResult(
144
success=False, output="",
145
error="Patch application timed out"
146
)
147
except Exception as e:
148
return PatchResult(
149
success=False, output="",
150
error=f"System patching error: {str(e)}"
151
)
152
finally:
153
if patch_file and os.path.exists(patch_file):
154
try:
155
os.unlink(patch_file)
156
except OSError:
157
pass
158
159
160
def _apply_patch_python(
161
patch_content: str,
162
target_file: str,
163
backup: bool = False,
164
dry_run: bool = False
165
) -> PatchResult:
166
"""
167
Pure Python patch implementation as fallback.
168
Handles basic unified diffs without external dependencies.
169
"""
170
import difflib
171
import re
172
173
target_path = Path(target_file)
174
175
if not target_path.exists():
176
return PatchResult(
177
success=False, output="",
178
error=f"Target file not found: {target_file}"
179
)
180
181
if not target_path.is_file():
182
return PatchResult(
183
success=False, output="",
184
error=f"Python fallback only supports single files, got: {target_file}"
185
)
186
187
try:
188
# Read original content
189
original = target_path.read_text(encoding='utf-8')
190
original_lines = original.splitlines(True)
191
192
# Parse the unified diff
193
patched_lines = _parse_and_apply_unified_diff(original_lines, patch_content)
194
195
if patched_lines is None:
196
return PatchResult(
197
success=False, output="",
198
error="Failed to parse or apply patch"
199
)
200
201
# Check if any changes
202
changed = original_lines != patched_lines
203
204
if not changed:
205
return PatchResult(
206
success=True,
207
output="Patch already applied",
208
files_changed=[]
209
)
210
211
if dry_run:
212
return PatchResult(
213
success=True,
214
output=f"Dry run successful: {len(patched_lines)} lines",
215
files_changed=[str(target_path)]
216
)
217
218
# Apply changes
219
if backup:
220
backup_path = target_path.with_suffix(target_path.suffix + '.orig')
221
backup_path.write_text(original, encoding='utf-8')
222
223
target_path.write_text(''.join(patched_lines), encoding='utf-8')
224
225
return PatchResult(
226
success=True,
227
output=f"Successfully patched {target_file}",
228
files_changed=[str(target_path)]
229
)
230
231
except Exception as e:
232
return PatchResult(
233
success=False, output="",
234
error=f"Python patching error: {str(e)}"
235
)
236
237
238
def _parse_and_apply_unified_diff(
239
original_lines: list[str],
240
diff_text: str
241
) -> Optional[list[str]]:
242
"""
243
Minimal unified diff parser and applier.
244
Returns patched lines or None on failure.
245
"""
246
# Parse hunks from unified diff
247
hunks = re.findall(
248
r'@@ -(\d+),?(\d*) \+(\d+),?(\d*) @@(.*?)(?=@@|\Z)',
249
diff_text,
250
re.DOTALL
251
)
252
253
if not hunks:
254
return None
255
256
result_lines = original_lines.copy()
257
258
for hunk in reversed(hunks): # Apply in reverse to maintain positions
259
old_start, old_count, new_start, new_count, body = hunk
260
old_start = int(old_start) - 1 # Convert to 0-based
261
old_count = int(old_count) if old_count else 1
262
263
# Parse hunk body
264
body_lines = body.split('\n')[1:] # Skip first empty line
265
hunk_result = []
266
267
old_pos = old_start
268
for line in body_lines:
269
if not line:
270
continue
271
272
if line.startswith('+'):
273
hunk_result.append(line[1:] + '\n')
274
elif line.startswith('-'):
275
old_pos += 1
276
elif line.startswith(' '):
277
if old_pos < len(result_lines):
278
hunk_result.append(result_lines[old_pos])
279
old_pos += 1
280
else:
281
# Context line
282
pass
283
284
# Replace the hunk region
285
end_index = min(old_start + old_count, len(result_lines))
286
result_lines[old_start:end_index] = hunk_result
287
288
return result_lines
289
290
291
def _parse_patched_files(output: str) -> list[str]:
292
"""Extract patched file names from patch command output."""
293
files = []
294
for line in output.splitlines():
295
if line.startswith("patching file "):
296
files.append(line[len("patching file "):].strip())
297
return files
298
299
300
def _extract_error(output: str) -> str:
301
"""Extract error message from patch command output."""
302
for line in output.splitlines():
303
if "FAILED" in line or "failed" in line or "error" in line.lower():
304
return line.strip()
305
return output.strip() or "Unknown patch error"
306
307
308
def apply_patch_with_fallback(
309
patch_content: str,
310
target_file: str,
311
backup: bool = True,
312
dry_run: bool = False
313
) -> dict:
314
"""
315
Apply patch with automatic fallback detection.
316
Tries system patch first, falls back to Python implementation on failure.
317
Returns structured result.
318
"""
319
# Check system capabilities
320
using_python = not patch_is_available()
321
322
if using_python:
323
# System patch not available, use Python directly
324
result = apply_patch(patch_content, target_file, backup, dry_run)
325
else:
326
# Try system patch first
327
result = apply_patch(patch_content, target_file, backup, dry_run)
328
# If system patch failed, try Python fallback
329
if not result.success:
330
result = _apply_patch_python(patch_content, target_file, backup, dry_run)
331
using_python = True
332
333
return {
334
"success": result.success,
335
"output": result.output,
336
"error": result.error,
337
"files_changed": result.files_changed,
338
"implementation": "python_fallback" if using_python else "system_patch"
339
}
@@ -29,6 +29,7 @@ from ..image.copy_images import get_media_dir, copy_media, get_source_url
29
29
from .tools import (
30
30
MarkItDownTool, TextToAudioTool, WebSearchTool, WebScrapeTool, ImageGenerationTool,
31
31
PythonExecuteTool, FileReadTool, FileWriteTool, FileListTool, FileDeleteTool,
32
ApplyPatchTool
32
33
)
33
34
34
35
@@ -74,6 +75,7 @@ class MCPServer:
74
75
'text_to_audio': TextToAudioTool(),
75
76
'mark_it_down': MarkItDownTool(),
76
77
'python_execute': PythonExecuteTool(safe_mode=safe_mode),
78
'apply_patch': ApplyPatchTool(),
77
79
'file_read': FileReadTool(),
78
80
'file_write': FileWriteTool(),
79
81
'file_list': FileListTool(safe_mode=safe_mode),
@@ -13,9 +13,7 @@ This module provides MCP tool implementations that wrap gpt4free capabilities:
13
13
14
14
from __future__ import annotations
15
15
16
import os
17
from pathlib import Path
18
from typing import Any, Dict, List
16
from typing import Any, Dict
19
17
from abc import ABC, abstractmethod
20
18
import urllib.parse
21
19
@@ -820,3 +818,69 @@ class FileDeleteTool(MCPTool):
820
818
return {"path": rel_path, "deleted": True}
821
819
except Exception as exc:
822
820
return {"error": f"Delete failed: {exc}"}
821
822
class ApplyPatchTool(MCPTool):
823
"""Apply a unified diff patch to a file or directory using the system 'patch' command."""
824
825
@property
826
def description(self) -> str:
827
return (
828
"Apply a unified diff patch to a target file or directory using the system 'patch' command. "
829
"Provide the target path, patch content, and optional parameters for strip level, backup, and dry run. "
830
"Returns success status, output from the patch command, and any error messages."
831
)
832
833
@property
834
def input_schema(self) -> Dict[str, Any]:
835
return {
836
"type": "object",
837
"properties": {
838
"target_path": {
839
"type": "string",
840
"description": "Path to the target file or directory to patch"
841
},
842
"patch_content": {
843
"type": "string",
844
"description": "The unified diff patch content to apply"
845
},
846
"strip": {
847
"type": "integer",
848
"description": "Number of leading path components to strip from file paths in the patch (default: 1)",
849
"default": 1
850
},
851
"backup": {
852
"type": "boolean",
853
"description": "Whether to create backup files before patching (default: false)",
854
"default": False
855
},
856
"dry_run": {
857
"type": "boolean",
858
"description": (
859
"If true, perform a dry run without making changes (default: false). "
860
"The output will indicate whether the patch would apply cleanly."
861
),
862
"default": False
863
},
864
},
865
"required": ["target_path", "patch_content"],
866
}
867
868
async def execute(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
869
from .pa_provider import get_workspace_dir
870
from .apply_patch import apply_patch_with_fallback
871
872
target_path = arguments.get("target_path", "")
873
patch_content = arguments.get("patch_content", "")
874
backup = bool(arguments.get("backup", False))
875
dry_run = bool(arguments.get("dry_run", False))
876
workspace = get_workspace_dir()
877
try:
878
target = (workspace / target_path).resolve()
879
if not str(target).startswith(str(workspace.resolve())):
880
return {"error": "Access outside the workspace is not allowed"}
881
if not target.exists():
882
return {"error": f"File not found: {target_path}"}
883
except Exception as exc:
884
return {"error": f"Invalid target path: {exc}"}
885
886
return apply_patch_with_fallback(patch_content, str(target), backup, dry_run)