72 lines
2.9 KiB
Python
72 lines
2.9 KiB
Python
import subprocess
|
|
from pathlib import Path
|
|
|
|
from app.repository.security import SKIPPED_PARTS, safe_repository_member
|
|
|
|
|
|
class RepositoryTools:
|
|
"""Git-backed read tools. A partial clone fetches a blob only when read_file needs it."""
|
|
|
|
def __init__(self, root: Path, max_file_bytes: int):
|
|
self.root, self.max_file_bytes = root.resolve(), max_file_bytes
|
|
|
|
def _run(self, args: list[str], timeout: int = 15) -> str:
|
|
result = subprocess.run(
|
|
args, cwd=self.root, text=True, capture_output=True, timeout=timeout, check=False
|
|
)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(result.stderr.strip() or "Repository command failed")
|
|
return result.stdout
|
|
|
|
def tree(self, limit: int = 300) -> str:
|
|
output = self._run(["git", "ls-tree", "-r", "--name-only", "HEAD"])
|
|
paths: list[str] = []
|
|
for path in output.splitlines():
|
|
if any(part in SKIPPED_PARTS for part in Path(path).parts):
|
|
continue
|
|
try:
|
|
safe_repository_member(path)
|
|
except ValueError:
|
|
continue
|
|
paths.append(path)
|
|
return "\n".join(paths[:limit])
|
|
|
|
def find_files(self, query: str) -> str:
|
|
if not query or len(query) > 100:
|
|
raise ValueError("Invalid search query")
|
|
return "\n".join(
|
|
line for line in self.tree(2_000).splitlines() if query.lower() in line.lower()
|
|
)[:12_000]
|
|
|
|
def search_text(self, query: str) -> str:
|
|
if not query or len(query) > 300:
|
|
raise ValueError("Invalid search query")
|
|
return "Text search would fetch too many blobs; use find_files then read_file."
|
|
|
|
def read_file(self, path: str, start_line: int = 1, end_line: int = 200) -> str:
|
|
member = safe_repository_member(path)
|
|
if start_line < 1 or end_line < start_line or end_line - start_line > 500:
|
|
raise ValueError("Invalid line range")
|
|
size = int(self._run(["git", "cat-file", "-s", f"HEAD:{member.as_posix()}"]).strip())
|
|
if size > self.max_file_bytes:
|
|
raise ValueError("File is unavailable or too large")
|
|
content = self._run(["git", "show", f"HEAD:{member.as_posix()}"], timeout=30)
|
|
lines = content.splitlines()
|
|
return "\n".join(
|
|
f"{i}: {line}" for i, line in enumerate(lines[start_line - 1 : end_line], start_line)
|
|
)
|
|
|
|
def current_commit(self) -> str:
|
|
return self._run(["git", "rev-parse", "HEAD"]).strip()
|
|
|
|
def recent_commits(self) -> str:
|
|
return self._run(["git", "log", "--oneline", "-10"])
|
|
|
|
def recent_diff(self) -> str:
|
|
return self._run(["git", "diff", "HEAD~1", "HEAD", "--stat"])[:12_000]
|
|
|
|
def file_info(self, path: str) -> str:
|
|
member = safe_repository_member(path)
|
|
size = self._run(["git", "cat-file", "-s", f"HEAD:{member.as_posix()}"]).strip()
|
|
return f"{path}: {size} bytes, extension={member.suffix or 'none'}"
|