|
| 1 | +import base64 |
| 2 | +import os |
| 3 | +import re |
| 4 | +from typing import Annotated, Any |
| 5 | +from urllib.parse import unquote, urlparse |
| 6 | + |
| 7 | +from fastmcp import Context |
| 8 | + |
| 9 | +from services.registry import mcp_for_unity_tool |
| 10 | +from services.tools import get_unity_instance_from_context |
| 11 | +from transport.unity_transport import send_with_unity_instance |
| 12 | +from transport.legacy.unity_connection import async_send_command_with_retry |
| 13 | + |
| 14 | + |
| 15 | +def _split_uri(uri: str) -> tuple[str, str]: |
| 16 | + """Split an incoming URI or path into (name, directory) suitable for Unity. |
| 17 | +
|
| 18 | + Rules: |
| 19 | + - unity://path/Assets/... → keep as Assets-relative (after decode/normalize) |
| 20 | + - file://... → percent-decode, normalize, strip host and leading slashes, |
| 21 | + then, if any 'Assets' segment exists, return path relative to that 'Assets' root. |
| 22 | + Otherwise, fall back to original name/dir behavior. |
| 23 | + - plain paths → decode/normalize separators; if they contain an 'Assets' segment, |
| 24 | + return relative to 'Assets'. |
| 25 | + """ |
| 26 | + raw_path: str |
| 27 | + if uri.startswith("unity://path/"): |
| 28 | + raw_path = uri[len("unity://path/"):] |
| 29 | + elif uri.startswith("file://"): |
| 30 | + parsed = urlparse(uri) |
| 31 | + host = (parsed.netloc or "").strip() |
| 32 | + p = parsed.path or "" |
| 33 | + # UNC: file://server/share/... -> //server/share/... |
| 34 | + if host and host.lower() != "localhost": |
| 35 | + p = f"//{host}{p}" |
| 36 | + # Use percent-decoded path, preserving leading slashes |
| 37 | + raw_path = unquote(p) |
| 38 | + else: |
| 39 | + raw_path = uri |
| 40 | + |
| 41 | + # Percent-decode any residual encodings and normalize separators |
| 42 | + raw_path = unquote(raw_path).replace("\\", "/") |
| 43 | + # Strip leading slash only for Windows drive-letter forms like "/C:/..." |
| 44 | + if os.name == "nt" and len(raw_path) >= 3 and raw_path[0] == "/" and raw_path[2] == ":": |
| 45 | + raw_path = raw_path[1:] |
| 46 | + |
| 47 | + # Normalize path (collapse ../, ./) |
| 48 | + norm = os.path.normpath(raw_path).replace("\\", "/") |
| 49 | + |
| 50 | + # If an 'Assets' segment exists, compute path relative to it (case-insensitive) |
| 51 | + parts = [p for p in norm.split("/") if p not in ("", ".")] |
| 52 | + idx = next((i for i, seg in enumerate(parts) |
| 53 | + if seg.lower() == "assets"), None) |
| 54 | + assets_rel = "/".join(parts[idx:]) if idx is not None else None |
| 55 | + |
| 56 | + effective_path = assets_rel if assets_rel else norm |
| 57 | + # For POSIX absolute paths outside Assets, drop the leading '/' |
| 58 | + # to return a clean relative-like directory (e.g., '/tmp' -> 'tmp'). |
| 59 | + if effective_path.startswith("/"): |
| 60 | + effective_path = effective_path[1:] |
| 61 | + |
| 62 | + name = os.path.splitext(os.path.basename(effective_path))[0] |
| 63 | + directory = os.path.dirname(effective_path) |
| 64 | + return name, directory |
| 65 | + |
| 66 | + |
| 67 | +@mcp_for_unity_tool(description="Searches a file with a regex pattern and returns line numbers and excerpts.") |
| 68 | +async def find_in_file( |
| 69 | + ctx: Context, |
| 70 | + uri: Annotated[str, "The resource URI to search under Assets/ or file path form supported by read_resource"], |
| 71 | + pattern: Annotated[str, "The regex pattern to search for"], |
| 72 | + project_root: Annotated[str | None, "Optional project root path"] = None, |
| 73 | + max_results: Annotated[int, "Cap results to avoid huge payloads"] = 200, |
| 74 | + ignore_case: Annotated[bool | str | None, "Case insensitive search"] = True, |
| 75 | +) -> dict[str, Any]: |
| 76 | + unity_instance = get_unity_instance_from_context(ctx) |
| 77 | + await ctx.info( |
| 78 | + f"Processing find_in_file: {uri} (unity_instance={unity_instance or 'default'})") |
| 79 | + |
| 80 | + name, directory = _split_uri(uri) |
| 81 | + |
| 82 | + # 1. Read file content via Unity |
| 83 | + read_resp = await send_with_unity_instance( |
| 84 | + async_send_command_with_retry, |
| 85 | + unity_instance, |
| 86 | + "manage_script", |
| 87 | + { |
| 88 | + "action": "read", |
| 89 | + "name": name, |
| 90 | + "path": directory, |
| 91 | + }, |
| 92 | + ) |
| 93 | + |
| 94 | + if not isinstance(read_resp, dict) or not read_resp.get("success"): |
| 95 | + return read_resp if isinstance(read_resp, dict) else {"success": False, "message": str(read_resp)} |
| 96 | + |
| 97 | + data = read_resp.get("data", {}) |
| 98 | + contents = data.get("contents") |
| 99 | + if not contents and data.get("contentsEncoded"): |
| 100 | + try: |
| 101 | + contents = base64.b64decode(data.get("encodedContents", "").encode( |
| 102 | + "utf-8")).decode("utf-8", "replace") |
| 103 | + except Exception: |
| 104 | + contents = contents or "" |
| 105 | + |
| 106 | + if contents is None: |
| 107 | + return {"success": False, "message": "Could not read file content."} |
| 108 | + |
| 109 | + # 2. Perform regex search |
| 110 | + flags = re.MULTILINE |
| 111 | + # Handle ignore_case which can be boolean or string from some clients |
| 112 | + ic = ignore_case |
| 113 | + if isinstance(ic, str): |
| 114 | + ic = ic.lower() in ("true", "1", "yes") |
| 115 | + if ic: |
| 116 | + flags |= re.IGNORECASE |
| 117 | + |
| 118 | + try: |
| 119 | + regex = re.compile(pattern, flags) |
| 120 | + except re.error as e: |
| 121 | + return {"success": False, "message": f"Invalid regex pattern: {e}"} |
| 122 | + |
| 123 | + matches = [] |
| 124 | + lines = contents.splitlines() |
| 125 | + |
| 126 | + # Helper to map index to line number |
| 127 | + def get_line_number(index, content_lines): |
| 128 | + # This is a bit slow for large files if we do it for every match, |
| 129 | + # but robust. |
| 130 | + # Better: iterate matches and count newlines? |
| 131 | + # Or just search line by line? |
| 132 | + # Searching line by line is safer for line-based results, but regex might span lines. |
| 133 | + pass |
| 134 | + |
| 135 | + # If the regex is not multiline specific (doesn't contain \n literal match logic), |
| 136 | + # we could iterate lines. But users might use multiline regexes. |
| 137 | + # Let's search the whole content and map back to lines. |
| 138 | + |
| 139 | + found = list(regex.finditer(contents)) |
| 140 | + |
| 141 | + results = [] |
| 142 | + count = 0 |
| 143 | + |
| 144 | + for m in found: |
| 145 | + if count >= max_results: |
| 146 | + break |
| 147 | + |
| 148 | + start_idx = m.start() |
| 149 | + end_idx = m.end() |
| 150 | + |
| 151 | + # Calculate line number |
| 152 | + # Count newlines up to start_idx |
| 153 | + line_num = contents.count('\n', 0, start_idx) + 1 |
| 154 | + |
| 155 | + # Get line content for excerpt |
| 156 | + # Find start of line |
| 157 | + line_start = contents.rfind('\n', 0, start_idx) + 1 |
| 158 | + # Find end of line |
| 159 | + line_end = contents.find('\n', start_idx) |
| 160 | + if line_end == -1: |
| 161 | + line_end = len(contents) |
| 162 | + |
| 163 | + line_content = contents[line_start:line_end] |
| 164 | + |
| 165 | + # Create excerpt |
| 166 | + # We can just return the line content as excerpt |
| 167 | + |
| 168 | + results.append({ |
| 169 | + "line": line_num, |
| 170 | + "content": line_content.strip(), # detailed match info? |
| 171 | + "match": m.group(0), |
| 172 | + "start": start_idx, |
| 173 | + "end": end_idx |
| 174 | + }) |
| 175 | + count += 1 |
| 176 | + |
| 177 | + return { |
| 178 | + "success": True, |
| 179 | + "data": { |
| 180 | + "matches": results, |
| 181 | + "count": len(results), |
| 182 | + "total_matches": len(found) |
| 183 | + } |
| 184 | + } |
| 185 | + |
0 commit comments