Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095"""Bounded reads, exact edits, and ripgrep-backed search over workspace files.
Every external command runs through the supervised `run` job plugin, so thismodule never spawns a process of its own and every child stays in a job's group."""
from __future__ import annotations
import asyncioimport codecsimport difflibimport fnmatchimport jsonimport osimport reimport shleximport statimport tempfilefrom dataclasses import dataclassfrom io import BufferedReaderfrom pathlib import Pathfrom collections.abc import Awaitable, Callable, Generatorfrom typing import Generic, Iterable, Sequence, TypeVar
from albedo_api import PythonApi, ReadyList, Textimport albedo_tracefrom albedo_plugins import hookfrom albedo_plugins import run as jobs
READ_LIMIT = 16_000READ_CHUNK_BYTES = 64 * 1024LINE_SEPARATORS = "\n\r\v\f\x1c\x1d\x1e\x85\u2028\u2029"LINE_BOUNDARIES = re.compile(rf"\r\n|[{re.escape(LINE_SEPARATORS)}]")SEARCH_TIMEOUT = 30LINE_WIDTH = 160CANDIDATES = 4LISTED = 10CONTEXT = 2PREVIEW_LINES = 2WITHHELD = 3
@dataclass(frozen=True)class Match: """One matching line, or with `context=` a surrounding line that prints grep-style as `path-line- text`."""
path: str line: int text: str context: bool = False
def __str__(self) -> str: mark = "-" if self.context else ":" return f"{self.path}{mark}{self.line}{mark} {self.text}"
def to_dict(self) -> dict[str, str | int | bool]: fields: dict[str, str | int | bool] = { "path": self.path, "line": self.line, "text": self.text, } if self.context: fields["context"] = True return fields
class Rows(ReadyList): """A list that prints one row per line, so a REPL result reads like output. Like Text, it may be awaited or used directly."""
def __init__( self, items: Iterable[object] = (), truncated: bool = False, note: str = "" ) -> None: super().__init__(items) self.truncated = truncated self.note = note
def __repr__(self) -> str: if not self: return f"[]\n{self.note}" if self.note else "[]" body = "\n".join(str(item) for item in self) return body + ("\n[truncated]" if self.truncated else "")
__str__ = __repr__
Result = TypeVar("Result")
class Search(Generic[Result]): """Work that runs a supervised job, so its result exists only once awaited: `Search[Rows]` awaits to Rows. Using it without `await` explains that instead of printing a coroutine."""
def __init__( self, call: str, run: Callable[[], Awaitable[Result]], result: str = "rows" ) -> None: self._call = call self._run = run self._result = result
def __await__(self) -> Generator[object, None, Result]: return self._run().__await__()
def __getitem__(self: Search[Rows], index: int | slice) -> Search[object]: """`await files.find(...)[:10]` slices before it awaits; apply the slice to the rows instead of failing on precedence."""
async def run() -> object: rows = await self._run() picked = rows[index] if isinstance(rows, Rows) and isinstance(index, slice): return Rows(picked, truncated=rows.truncated, note=rows.note) return picked
return Search(self._call, run, self._result)
def _unawaited(self) -> TypeError: return TypeError( f"{self._call}(...) runs in the background; " f"use `await {self._call}(...)` to get its {self._result}" )
def __repr__(self) -> str: return f"<{self._call}(...) has not run: use `await {self._call}(...)` for its {self._result}>"
__str__ = __repr__
def __iter__(self): raise self._unawaited()
def __len__(self) -> int: raise self._unawaited()
async def _run(command: str) -> tuple[int | None, str, bool]: """One supervised shell job, awaited to completion, with its bounded output. The plugin's own work, so neither traced as the cell's command nor refused.""" job = jobs.start(["/bin/sh", "-c", command], SEARCH_TIMEOUT, traced=False) try: await job clipped = job.capture.seen > job.capture.retained limit = min(jobs.preview_limit, job.capture.retained) return job.exit_code, str(job.read(offset=0, limit=limit)), clipped except asyncio.CancelledError: await job.stop() raise finally: if job.exit_code is not None or job.timed_out or job.termination is not None: jobs.forget(job)
def _numbered(number: int, lines: Sequence[str]) -> str: text = lines[number - 1] if 0 < number <= len(lines) else "" return f"{number:>6} | {text[:LINE_WIDTH]}"
class _ReadWindow: """Track selected lines, the current line, and the first rejected row."""
def __init__(self, start_line: int, last_line: int | None, max_chars: int): self.start_line = start_line self.last_line = last_line self.max_chars = max_chars self.lines: list[str] = [] self.used_chars = 0 self.stopped_line: int | None = None self.stopped_row_chars = 0 self.number = 1 self.line_chars = 0 self.fragments: list[str] = []
def wants_line(self) -> bool: return ( self.stopped_line is None and self.number >= self.start_line and (self.last_line is None or self.number <= self.last_line) )
def extend(self, text: str, start: int, end: int) -> None: self.line_chars += end - start if self.wants_line(): row_chars = max(6, len(str(self.number))) + 3 + self.line_chars if self.used_chars + row_chars + 1 <= self.max_chars: if start < end: self.fragments.append(text[start:end]) else: # Keep counting a giant line, but release its retained prefix. self.fragments.clear()
def finish_line(self) -> None: if self.wants_line(): row_chars = max(6, len(str(self.number))) + 3 + self.line_chars if self.used_chars + row_chars + 1 > self.max_chars: self.stopped_line = self.number self.stopped_row_chars = row_chars else: self.lines.append("".join(self.fragments)) self.used_chars += row_chars + 1 self.number += 1 self.line_chars = 0 self.fragments.clear()
def _scan_read( source: BufferedReader, window: _ReadWindow, end_line: int | None) -> int: chunk = source.read(READ_CHUNK_BYTES) at_eof = len(chunk) < READ_CHUNK_BYTES or not source.peek(1) text = chunk.decode("utf-8", errors="replace") if at_eof else "" # Native splitting is cheap for a bounded ASCII file. Unicode uses spans # instead, so dense replacement characters never build a list of substrings. if at_eof and text.isascii(): lines = text.splitlines() total = len(lines) last = total if window.last_line is None else min(total, window.last_line) window.lines = lines[window.start_line - 1 : last] return total
decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") if not at_eof: text = decoder.decode(chunk) skip_lf = False while True: if skip_lf and text.startswith("\n"): text = text[1:] separator_count = 0 if any(separator in text for separator in LINE_SEPARATORS): separator_count = sum( text.count(separator) for separator in LINE_SEPARATORS ) separator_count -= text.count("\r\n") window_complete = window.stopped_line is not None or ( window.last_line is not None and window.number > window.last_line ) if not separator_count: window.extend(text, 0, len(text)) elif window_complete or window.number + separator_count < window.start_line: window.number += separator_count if end_line is not None and window.number > end_line: return end_line # Skipped lines need only an unfinished-line flag for EOF. # A requested line begins after a separator resets this state. window.line_chars = int(text[-1] not in LINE_SEPARATORS) else: position = 0 for boundary in LINE_BOUNDARIES.finditer(text): window.extend(text, position, boundary.start()) window.finish_line() if end_line is not None and window.number > end_line: return end_line position = boundary.end() window.extend(text, position, len(text)) skip_lf = text.endswith("\r") if at_eof: if window.line_chars: window.finish_line() return window.number - 1 chunk = source.read(READ_CHUNK_BYTES) at_eof = not chunk text = decoder.decode(chunk, final=at_eof)
class Files: """Workspace file access with the diagnostics an exact edit needs."""
def read( self, path: str, start_line: int = 1, end_line: int | None = None, *, limit: int | None = None, max_chars: int = READ_LIMIT, ) -> Text: """Numbered lines from start_line through end_line, at most `limit` lines.
`limit` counts lines. `max_chars` is a separate character budget (at most 200000) that keeps a huge window from flooding the context; when it stops a read early, the result says so and names the line to resume from. Lines are never shortened. The numbers are what `edit(line_hint=)` takes. `$ALBEDO_HOME/creds.json` is refused: `commands.accounts()` lists its accounts without secrets. """ if ( start_line < 1 or (end_line is not None and end_line < start_line) or (limit is not None and limit < 1) or not 0 < max_chars <= 200_000 ): raise ValueError( "start_line >= 1, end_line >= start_line, limit >= 1 line, " "0 < max_chars <= 200000" ) target = Path(path).expanduser() _refuse_secrets([target]) if not target.is_file(): if target.is_dir(): raise IsADirectoryError( f"{path} is a directory; files.ls({path!r}) lists it" ) raise FileNotFoundError(_missing(target, path)) last = end_line if limit is not None: limited_last = start_line + limit - 1 last = limited_last if last is None else min(last, limited_last) window = _ReadWindow(start_line, last, max_chars) with target.open("rb") as source: total = _scan_read(source, window, end_line) if start_line > total: return Text(f"[{path} has {total} lines; nothing at line {start_line}]") requested = total if end_line is None else min(end_line, total) last = requested if last is None else min(last, requested) body, used = window.lines, 0 number = window.stopped_line row_chars = window.stopped_row_chars for current, text in enumerate(body, start_line): row = f"{current:>6} | {text}" if used + len(row) + 1 > max_chars: number, row_chars = current, len(row) del body[current - start_line :] break used += len(row) + 1 body[current - start_line] = row if number is not None: if not body: retry = ( f"read it alone with start_line={number}, end_line={number}, max_chars={row_chars + 1}" if row_chars + 1 <= 200_000 else "it exceeds the 200000-character maximum" ) body.append( f"[line {number} is {row_chars} characters with its number, over " f"max_chars={max_chars}; {retry}]" ) else: body.append( f"[stopped at max_chars={max_chars} characters; lines {number}-{last} " f"not shown; read again with start_line={number}, or raise max_chars]" ) elif last < requested: body.append( f"[limit={limit} lines reached; {requested - last} more through line " f"{requested}; read again with start_line={last + 1}]" ) result = "\n".join(body) # Release numbered rows before Text copies the joined Unicode buffer. body.clear() return Text(result)
def ls( self, path: str = ".", pattern: str | None = None, *, hidden: bool = False ) -> Rows: """One directory, directories suffixed with `/`.""" directory = Path(path).expanduser() entries = [] for entry in sorted(directory.iterdir(), key=lambda item: item.name): if not hidden and entry.name.startswith("."): continue if pattern and not fnmatch.fnmatch(entry.name, pattern): continue entries.append(entry.name + ("/" if entry.is_dir() else "")) return Rows(entries)
def find( self, pattern: str, path: str | Sequence[str] = ".", *, glob: str | Sequence[str] | None = None, context: int = 0, max_results: int = 50, literal: bool = False, case_sensitive: bool | None = None, hidden: bool = False, ignored: bool = False, ) -> Search[Rows]: """Content search through ripgrep when it is installed, else pure Python. `path` may be one path or a list; `context=N` adds N lines around each match. A missing path raises. Ignore files and hidden files are skipped unless `ignored=True` / `hidden=True`; an empty result says what was searched and skipped. Await it: `await files.find(pattern)`.""" if not 0 <= context <= 50: raise ValueError("0 <= context <= 50") return Search( "files.find", lambda: albedo_trace.noted( "search", pattern, self._find( pattern, path, glob, context, max_results, literal, case_sensitive, hidden, ignored, ), ), )
async def _find( self, pattern: str, path: str | Sequence[str], glob: str | Sequence[str] | None, context: int, max_results: int, literal: bool, case_sensitive: bool | None, hidden: bool, ignored: bool, ) -> Rows: targets = _targets(path) _require_roots(targets) _refuse_secrets(targets) if not _which("rg"): return _fallback_find( pattern, targets, glob, context, max_results, literal, case_sensitive, hidden, ignored, ) flags = [] if literal: flags.append("-F") if case_sensitive is True: flags.append("-s") elif case_sensitive is False: flags.append("-i") for value in _globs(glob): flags += ["-g", value] visible = [ "--hidden" if hidden else "--no-hidden", "--no-ignore" if ignored else "--ignore", ]
def search(*extra: str) -> str: return " ".join( shlex.quote(part) for part in ["rg", *flags, *extra, "-e", pattern, *map(str, targets)] )
per_match = 4 + 2 * context status, output, clipped = await _run( f"{search('--json', *visible, *(['-C', str(context)] if context else []))}" f" | head -n {max(1, max_results) * per_match}" ) if status not in (0, 1, None) and not output.strip(): raise RuntimeError(f"search failed: {output.strip()[:400]}") results, matched, searched = [], 0, 0 for line in output.splitlines(): try: event = json.loads(line) except ValueError: continue kind = event.get("type") if kind == "summary": searched = event["data"]["stats"]["searches"] if kind not in ("match", "context"): continue data = event["data"] if _secret(data["path"]["text"]): continue if kind == "match" and matched >= max_results: return Rows(results, truncated=True) results.append( Match( data["path"]["text"], data["line_number"], data["lines"]["text"].rstrip("\r\n")[: LINE_WIDTH * 4], context=kind == "context", ) ) matched += kind == "match" if results: return Rows(results, truncated=clipped) skipped = [ *([] if hidden else ["hidden files"]), *([] if ignored else [".gitignore/.ignore rules"]), ] withheld: list[str] = [] if skipped: _, listing, _ = await _run( f"{search('-l', '--hidden', '--no-ignore')} | head -n {WITHHELD}" ) withheld = listing.splitlines() return Rows( truncated=clipped, note=_empty_note(pattern, targets, searched, skipped, glob, withheld), )
def paths( self, pattern: str | None = None, path: str | Sequence[str] = ".", *, glob: str | Sequence[str] | None = None, max_results: int = 100, hidden: bool = False, ) -> Search[Rows]: """File names, not contents; the same ripgrep-or-Python split. A pattern with *, ? or [ is a glob over names, or over the path from the search root when it has a /; other text matches anywhere in the path. `path` may be one path or a list. Await it: `await files.paths(pattern)`.""" return Search( "files.paths", lambda: self._paths(pattern, path, glob, max_results, hidden) )
async def _paths( self, pattern: str | None, path: str | Sequence[str], glob: str | Sequence[str] | None, max_results: int, hidden: bool, ) -> Rows: targets = _targets(path) globs = _globs(glob) if not _which("rg"): return _fallback_paths(pattern, targets, globs, max_results, hidden) flags = ["--files"] if hidden: flags.append("--hidden") for value in globs: flags += ["-g", value] command = " ".join( shlex.quote(part) for part in ["rg", *flags, *map(str, targets)] ) _, output, clipped = await _run(command) matches = _name_matcher(pattern) # rg lists "./x" under ".", where a path from the root is just "x" listed = (line.removeprefix("./") for line in output.splitlines() if line) found = [line for line in listed if matches(line)] return Rows(found[:max_results], truncated=len(found) > max_results or clipped)
@hook.reported( "files.edit", {"path": "file_path", "old_str": "old_string", "new_str": "new_string"}, ) def edit( self, path: str, old_str: str | Sequence[tuple[str, str]], new_str: str | None = None, line_hint: int | None = None, ) -> Text: """Replace one exact, unique string, or apply a list of (old, new) pairs in order: all of them or, when one misses, none. A miss reports what is actually there.""" if isinstance(old_str, str): if new_str is None: raise ValueError("new_str is required with one old_str") edits = [(old_str, new_str)] else: if new_str is not None or line_hint is not None: raise ValueError( "a list of (old, new) pairs takes no new_str or line_hint" ) edits = _pairs(old_str) if line_hint is not None and ( isinstance(line_hint, bool) or not isinstance(line_hint, int) ): raise ValueError( f"line_hint must be a line number, not {type(line_hint).__name__}" ) target = Path(path).expanduser() if not target.exists(): raise FileNotFoundError(_missing(target, path)) target = target.resolve() snapshot = _snapshot(target) content = snapshot[0].decode("utf-8") for number, (old, new) in enumerate(edits, 1): try: content = _replace_one(content, old, new, line_hint, path) except ValueError as error: if len(edits) == 1: raise raise ValueError( f"edit {number} of {len(edits)} failed, so none were made: {error}" ) from None _replace(target, snapshot, content.encode("utf-8")) return Text( f"Edited {target}" + (f" ({len(edits)} replacements)" if len(edits) > 1 else "") )
@hook.reported("files.write", {"path": "file_path", "content": "content"}) def write(self, path: str, content: str) -> Text: """Create or replace a whole file. Use `edit` to change part of one.""" target = Path(path).expanduser() target.parent.mkdir(parents=True, exist_ok=True) target.write_text(content, encoding="utf-8") return Text(f"Wrote {target.resolve()} ({len(content.encode())} bytes)")
def _pairs(edits: object) -> list[tuple[str, str]]: """The (old, new) pairs of a multi-edit, checked before anything is read.""" if not isinstance(edits, (list, tuple)) or not edits: raise ValueError("old_str is a string, or a non-empty list of (old, new) pairs") pairs = [] for number, pair in enumerate(edits, 1): if not ( isinstance(pair, (list, tuple)) and len(pair) == 2 and all(isinstance(part, str) for part in pair) ): raise ValueError(f"edit {number} is not an (old, new) pair of strings") pairs.append((pair[0], pair[1])) return pairs
def _replace_one( content: str, old_str: str, new_str: str, line_hint: int | None, path: str) -> str: """`content` with the one occurrence of `old_str` replaced, or ValueError saying why there is not exactly one.""" if not old_str: raise ValueError("old_str must be non-empty") found = _occurrences(content, old_str) if not found: closest = _closest(content, old_str) raise ValueError( f"string not found in {path}" + ( "; the text is there, but whitespace differs (indentation, " "tabs, trailing spaces or line endings)" if _words(old_str) in _words(content) and _words(old_str) else "" ) + (f"\n{closest}" if closest else "") ) chosen = _choose(content, old_str, found, line_hint, path) return content[: chosen.index] + new_str + content[chosen.index + len(old_str) :]
def _which(name: str) -> str | None: for directory in os.environ.get("PATH", "").split(os.pathsep): candidate = Path(directory) / name if candidate.is_file() and os.access(candidate, os.X_OK): return str(candidate) return None
def _secret(path: str | Path) -> bool: """Whether a path is the daemon's secret store, however it is spelled.""" home = os.environ.get("ALBEDO_HOME") or os.path.expanduser("~/.albedo") return Path(path).expanduser().resolve() == (Path(home) / "creds.json").resolve()
def _refuse_secrets(targets: list[Path]) -> None: if any(_secret(target) for target in targets): raise PermissionError( "creds.json holds long-lived OAuth refresh tokens and API keys; " "await commands.accounts() lists accounts without secrets." )
def _require_roots(targets: list[Path]) -> None: for target in targets: if not target.exists(): raise FileNotFoundError(_missing(target, str(target)))
def _empty_note( pattern: str, targets: list[Path], searched: int, skipped: list[str], glob: str | Sequence[str] | None, withheld: list[str],) -> str: """Why a search came back empty: where it looked, what it left out, and which skipped files would have matched.""" roots = ", ".join(str(target.resolve()) for target in targets) left_out = [*skipped, *([f"files outside glob {_globs(glob)}"] if glob else [])] note = f"no matches for {pattern!r} in {roots}: {searched} files searched" if left_out: note += f"; skipped {', '.join(left_out)}" if withheld: note += ( f"; it matches in skipped files ({', '.join(withheld)}): " "rerun with hidden=True and/or ignored=True" ) return note
def _fallback_find( pattern, targets: list[Path], glob, context, max_results, literal, case_sensitive, hidden, ignored,) -> Rows: rows, searched = _fallback_scan( pattern, targets, glob, context, max_results, literal, case_sensitive, hidden, ignored, ) if rows: return rows withheld = [] if not (hidden and ignored): everything, _ = _fallback_scan( pattern, targets, glob, 0, WITHHELD, literal, case_sensitive, True, True ) withheld = [match.path for match in everything] skipped = [ *([] if hidden else ["hidden files"]), *([] if ignored else [".gitignore/.ignore rules"]), ] return Rows(note=_empty_note(pattern, targets, searched, skipped, glob, withheld))
def _fallback_scan( pattern, targets: list[Path], glob, context, max_results, literal, case_sensitive, hidden, ignored,) -> tuple[Rows, int]: flags = 0 if case_sensitive else re.IGNORECASE matcher = re.compile(re.escape(pattern) if literal else pattern, flags) globs = _globs(glob) results, matched, searched = [], 0, 0 for target in targets: for file in _walk(target, hidden, ignored): if not _globbed(file, globs) or _secret(file): continue try: lines = file.read_text(encoding="utf-8", errors="replace").splitlines() except OSError: continue searched += 1 hits = [ number for number, line in enumerate(lines, 1) if matcher.search(line) ] hit = set(hits) shown = 0 for number in hits: if matched >= max_results: return Rows(results, truncated=True), searched for around in range( max(number - context, shown + 1), min(number + context, len(lines)) + 1, ): results.append( Match( str(file), around, lines[around - 1][: LINE_WIDTH * 4], context=around not in hit, ) ) shown = around matched += 1 return Rows(results), searched
def _targets(path: str | Sequence[str]) -> list[Path]: return [ Path(item).expanduser() for item in ([path] if isinstance(path, str) else path) ]
def _globs(glob: str | Sequence[str] | None) -> list[str]: return [glob] if isinstance(glob, str) else list(glob or [])
def _globbed(file: Path, globs: list[str]) -> bool: """True when there is no glob, or one matches the path or the file name.""" return not globs or any( fnmatch.fnmatch(str(file), value) or fnmatch.fnmatch(file.name, value) for value in globs )
def _name_matcher(pattern: str | None) -> Callable[[str], bool]: """A pattern with *, ? or [ is a glob over the file name (or the whole path when it has a /); anything else matches as case-insensitive text anywhere in the path, so "*.md" and "readme" both mean what they look like.""" if not pattern: return lambda _path: True folded = pattern.lower().removeprefix("./") if any(char in pattern for char in "*?["): return lambda path: fnmatch.fnmatch( (path if "/" in pattern else Path(path).name).lower(), folded ) return lambda path: folded in path.lower()
def _fallback_paths( pattern, targets: list[Path], globs: list[str], max_results, hidden) -> Rows: matches = _name_matcher(pattern) found = [ str(file) for target in targets for file in _walk(target, hidden, ignored=False) if _globbed(file, globs) and matches(str(file)) ] return Rows(found[:max_results], truncated=len(found) > max_results)
def _walk(path: Path, hidden: bool, ignored: bool): """Files under `path` as ripgrep would search them without ripgrep: hidden names and, unless `ignored`, what each directory's .gitignore and .ignore leave out (plain patterns; negations are not supported).""" if path.is_file(): yield path return # ripgrep also honours the ignore files of the directories above the root. rules = { os.path.dirname(path): [ rule for ancestor in reversed(path.absolute().parents) for rule in _ignore_rules(str(ancestor)) ] } for root, directories, names in os.walk(path): if not hidden: directories[:] = [name for name in directories if not name.startswith(".")] names = [name for name in names if not name.startswith(".")] if not ignored: rules[root] = rules[os.path.dirname(root)] + _ignore_rules(root) directories[:] = [ name for name in directories if not _ignores(rules[root], os.path.join(root, name), True) ] names = [ name for name in names if not _ignores(rules[root], os.path.join(root, name), False) ] for name in sorted(names): yield Path(root) / name
def _ignore_rules(directory: str) -> list[tuple[str, str, bool]]: """(base, pattern, directories only) for each rule in a directory's .gitignore and .ignore.""" found = [] for source in (".gitignore", ".ignore"): try: lines = Path(directory, source).read_text(errors="replace").splitlines() except OSError: continue for line in lines: line = line.strip() if not line or line.startswith(("#", "!")): continue found.append((directory, line.rstrip("/"), line.endswith("/"))) return found
def _ignores(rules: list[tuple[str, str, bool]], path: str, directory: bool) -> bool: for base, pattern, directories_only in rules: if directories_only and not directory: continue relative = os.path.relpath(path, base) if "/" in pattern: if fnmatch.fnmatch(relative, pattern.lstrip("/")): return True elif fnmatch.fnmatch(os.path.basename(path), pattern): return True return False
def _snapshot(target: Path) -> tuple[bytes, tuple[int, int, int, int, int]]: """Bytes plus identity: an edit refuses to publish over a concurrent change.""" for _ in range(3): before = target.stat() data = target.read_bytes() after = target.stat() identity = ( after.st_dev, after.st_ino, stat.S_IMODE(after.st_mode), after.st_size, after.st_mtime_ns, ) if identity == ( before.st_dev, before.st_ino, stat.S_IMODE(before.st_mode), before.st_size, before.st_mtime_ns, ): return data, identity raise ValueError(f"file changed while reading {target}; refusing to edit it")
def _replace( target: Path, snapshot: tuple[bytes, tuple[int, int, int, int, int]], replacement: bytes,) -> None: descriptor, name = tempfile.mkstemp( dir=target.parent, prefix=f".{target.name}.", suffix=".tmp" ) temporary = Path(name) try: with os.fdopen(descriptor, "wb") as stream: os.fchmod(stream.fileno(), snapshot[1][2]) stream.write(replacement) stream.flush() os.fsync(stream.fileno()) if _snapshot(target) != snapshot: raise ValueError( f"file changed while editing {target}; refusing to overwrite concurrent changes" ) os.replace(temporary, target) finally: try: temporary.unlink() except FileNotFoundError: pass
@dataclass(frozen=True)class _Occurrence: index: int start_line: int end_line: int
@property def label(self) -> str: return ( str(self.start_line) if self.start_line == self.end_line else f"{self.start_line}-{self.end_line}" )
def _occurrences(content: str, needle: str) -> list[_Occurrence]: found, span, start = [], needle.count("\n") + 1, 0 while True: index = content.find(needle, start) if index < 0: return found first = content.count("\n", 0, index) + 1 found.append(_Occurrence(index, first, first + span - 1)) start = index + len(needle)
def _choose( content: str, old_str: str, found: list[_Occurrence], line_hint: int | None, path: str,) -> _Occurrence: if len(found) == 1: return found[0] listed = ", ".join(one.label for one in found[:LISTED]) + ( f", ... ({len(found)} total)" if len(found) > LISTED else "" ) if line_hint is None: raise ValueError( f"found {len(found)} occurrences in {path}: lines {listed}. Nothing was changed. " "Retry with line_hint=<a line inside the range you want>, or widen old_str until it is unique.\n" + _candidates(content, found) ) inside = next( (one for one in found if one.start_line <= line_hint <= one.end_line), None ) if inside is None: raise ValueError( f"line_hint={line_hint} is inside none of the {len(found)} occurrences in {path}: lines {listed}. " f"Nothing was changed.\n{_candidates(content, found)}" ) return inside
def _candidates(content: str, found: list[_Occurrence]) -> str: lines = content.splitlines() blocks = [] for position, occurrence in enumerate(found[:CANDIDATES], 1): matched = occurrence.end_line - occurrence.start_line + 1 preview = min(matched, PREVIEW_LINES) first = max(1, occurrence.start_line - CONTEXT) last = min(len(lines), occurrence.end_line + CONTEXT) body = [ _numbered(number, lines) for number in range(first, occurrence.start_line) ] body += [ _numbered(number, lines) for number in range(occurrence.start_line, occurrence.start_line + preview) ] if matched > preview: body.append(f"{'...':>6} | ({matched - preview} further matching lines)") body += [ _numbered(number, lines) for number in range(occurrence.end_line + 1, last + 1) ] blocks.append( f"candidate {position} of {len(found)}: lines {occurrence.label}\n" + "\n".join(body) ) if len(found) > CANDIDATES: blocks.append(f"... {len(found) - CANDIDATES} further occurrences not shown") return "\n\n".join(blocks)
def _words(text: str) -> str: """text with every run of whitespace as one space, to see a whitespace-only miss.""" return " ".join(text.split())
def _closest(content: str, needle: str) -> str: lines, needle_lines = content.splitlines(), needle.splitlines() head = next((line.strip() for line in needle_lines if line.strip()), "") if not lines or not head: return "" ranked = sorted( ( ( difflib.SequenceMatcher( None, head, line.strip(), autojunk=False ).ratio(), index, ) for index, line in enumerate(lines) if line.strip() ), reverse=True, )[:12] width = max(1, len(needle_lines)) scored = [] for _, start in ranked: end = min(len(lines), start + width) current = "\n".join(lines[start:end]).strip() scored.append( ( difflib.SequenceMatcher( None, needle.strip(), current, autojunk=False ).ratio(), start, end, ) ) blocks = [] for score, start, end in sorted(scored, reverse=True): if score < 0.25: continue body = "\n".join(_numbered(number + 1, lines) for number in range(start, end)) blocks.append( f"closest candidate lines {start + 1}-{end} (similarity {score:.0%}):\n{body}" ) if len(blocks) == 3: break return "\n\n".join(blocks)
def _missing(target: Path, original: str) -> str: message = f"{original} not found (cwd: {Path.cwd()})" parent = target.parent if not parent.is_dir(): return message close = difflib.get_close_matches( target.name, [entry.name for entry in parent.iterdir()], n=3, cutoff=0.35 ) return message + ( f"; nearby paths: {', '.join(str(parent / name) for name in close)}" if close else "" )
def setup(api: PythonApi) -> dict[str, object]: return {"files": Files()}