File size: 4,501 Bytes
d61821a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 | """Read immutable source snapshots directly from a local Git object database."""
from __future__ import annotations
from dataclasses import dataclass
from hashlib import sha256
from pathlib import Path
import subprocess
from typing import Iterator
class GitSnapshotError(RuntimeError):
"""Raised when a repository or pinned commit cannot be read safely."""
@dataclass(frozen=True, slots=True)
class SourceFile:
path: str
text: str
@dataclass(frozen=True, slots=True)
class SourceChunk:
chunk_id: str
path: str
line_start: int
line_end: int
text: str
class GitSnapshot:
def __init__(self, repository: Path):
self.repository = repository.resolve()
if not (self.repository / ".git").exists():
raise GitSnapshotError(f"Not a Git checkout: {self.repository}")
def _git(self, arguments: list[str], timeout_seconds: int = 60) -> bytes:
result = subprocess.run(
["git", *arguments],
cwd=self.repository,
check=False,
capture_output=True,
timeout=timeout_seconds,
)
if result.returncode != 0:
detail = result.stderr.decode("utf-8", errors="replace").strip()
raise GitSnapshotError(f"git {' '.join(arguments)} failed: {detail}")
return result.stdout
def verify_commit(self, commit: str) -> None:
self._git(["cat-file", "-e", f"{commit}^{{commit}}"])
def tracked_paths(self, commit: str, suffixes: tuple[str, ...] = (".go",)) -> tuple[str, ...]:
self.verify_commit(commit)
raw = self._git(["ls-tree", "-r", "--name-only", "-z", commit])
paths = tuple(
path
for path in raw.decode("utf-8", errors="surrogateescape").split("\0")
if path and path.endswith(suffixes)
)
return tuple(sorted(paths))
def read_file(self, commit: str, path: str) -> SourceFile:
if Path(path).is_absolute() or ".." in Path(path).parts:
raise GitSnapshotError(f"Unsafe repository path: {path}")
raw = self._git(["show", f"{commit}:{path}"])
return SourceFile(path=path, text=raw.decode("utf-8", errors="replace"))
def iter_files(
self,
commit: str,
suffixes: tuple[str, ...] = (".go",),
) -> Iterator[SourceFile]:
for path in self.tracked_paths(commit, suffixes):
yield self.read_file(commit, path)
def chunk_file(
source: SourceFile,
chunk_lines: int,
overlap_lines: int,
char_limit: int,
) -> tuple[SourceChunk, ...]:
if chunk_lines <= 0 or char_limit <= 0 or not 0 <= overlap_lines < chunk_lines:
raise ValueError("invalid line chunking policy")
lines = source.text.splitlines(keepends=True)
if not lines:
return ()
stride = chunk_lines - overlap_lines
chunks: list[SourceChunk] = []
for start in range(0, len(lines), stride):
selected = lines[start : start + chunk_lines]
if not selected:
break
block = "".join(selected)
block_line_start = start + 1
char_stride = max(char_limit - 512, 1)
for char_start in range(0, len(block), char_stride):
text = block[char_start : char_start + char_limit]
if not text:
break
line_start = block_line_start + block[:char_start].count("\n")
line_end = line_start + text.count("\n")
if text and not text.endswith("\n"):
line_end += 1
identity = sha256(
f"{source.path}\0{line_start}\0{line_end}\0{text}".encode("utf-8")
).hexdigest()
chunks.append(
SourceChunk(
chunk_id=identity,
path=source.path,
line_start=line_start,
line_end=line_end,
text=text,
)
)
if char_start + char_limit >= len(block):
break
if line_end == len(lines):
break
return tuple(chunks)
def chunk_snapshot(
snapshot: GitSnapshot,
commit: str,
chunk_lines: int,
overlap_lines: int,
char_limit: int,
suffixes: tuple[str, ...] = (".go",),
) -> tuple[SourceChunk, ...]:
chunks: list[SourceChunk] = []
for source in snapshot.iter_files(commit, suffixes):
chunks.extend(chunk_file(source, chunk_lines, overlap_lines, char_limit))
return tuple(chunks)
|