diff --git a/src/cocoindex_code/cli.py b/src/cocoindex_code/cli.py index a58a8c4..6dabf17 100644 --- a/src/cocoindex_code/cli.py +++ b/src/cocoindex_code/cli.py @@ -21,6 +21,8 @@ ProjectStatusResponse, SearchResponse, ) + from .textsearch import FileMatches as TextFileMatches + from .textsearch import TextSearchWarning from .settings import ( DEFAULT_ST_MODEL, @@ -294,6 +296,69 @@ def _on_waiting() -> None: return resp +def _run_text_search( + terms: list[str], + *, + path_glob: str | None, + languages: list[str], + limit: int, + ignore_case: bool, + no_color: bool, +) -> None: + """Run the local literal/regex text search (``ccc search --text``). + + Fully local — no project init or daemon required — walking files like + ``ccc grep``. Results stream per file as they complete. + """ + from . import textsearch as _ts + + # -i forces case-insensitive; otherwise smart-case. + case_flag: bool | None = False if ignore_case else None + + try: + compiled = _ts.compile_terms(terms, case_sensitive=case_flag) + except _ts.TermSyntaxError as e: + _typer.echo(f"Error: {e}", err=True) + raise _typer.Exit(code=1) + + req = _ts.TextSearchRequest( + terms=tuple(compiled), + root=Path.cwd(), + languages=frozenset(name.lower() for name in languages) or None, + path_glob=path_glob, + ) + use_color = not no_color and sys.stdout.isatty() and not os.environ.get("NO_COLOR") + + matched = 0 + # `run` calls `_emit` from several worker threads at once; the lock keeps one + # file's output (and the `matched` bookkeeping) from interleaving with another's. + output_lock = threading.Lock() + + def _emit(item: TextFileMatches | TextSearchWarning) -> None: + nonlocal matched + if isinstance(item, _ts.TextSearchWarning): + with output_lock: + _typer.echo(f"warning: {item.message}", err=True) + return + block = _ts.render_file(item, color=use_color) # render outside the lock + with output_lock: + if matched: + _typer.echo() # blank line between files + _typer.echo(block) + matched += 1 + + # The engine caps emitted files at `limit` and reports whether it stopped early. + hit_limit = _ts.TextSearch(req).run(_emit, limit=limit) + + if matched == 0: + _typer.echo("No matches found.") + elif hit_limit: + _typer.echo( + f"\nStopped at --limit {limit}. Re-run with a higher --limit to see more.", + err=True, + ) + + _GITIGNORE_COMMENT = "# CocoIndex Code (ccc)" _GITIGNORE_ENTRY = "/.cocoindex_code/" @@ -648,15 +713,58 @@ def index() -> None: @app.command() @_catch_daemon_start_error def search( - query: list[str] = _typer.Argument(..., help="Search query"), + query: list[str] = _typer.Argument(..., help="Search query (or terms for --text)"), lang: list[str] = _typer.Option([], "--lang", help="Filter by language"), path: str | None = _typer.Option(None, "--path", help="Filter by file path glob"), offset: int = _typer.Option(0, "--offset", help="Number of results to skip"), limit: int = _typer.Option(10, "--limit", help="Maximum results to return"), refresh: bool = _typer.Option(False, "--refresh", help="Refresh index before searching"), json_output: bool = _typer.Option(False, "--json", help="Output results as JSON"), + text: bool = _typer.Option( + False, + "--text", + "-t", + help="Literal/regex full-text search — local, no index or daemon.", + ), + ignore_case: bool = _typer.Option( + False, "-i", "--ignore-case", help="[--text] Case-insensitive (default: smart-case)." + ), + no_color: bool = _typer.Option(False, "--no-color", help="[--text] Disable colored output."), ) -> None: - """Semantic search across the codebase.""" + """Semantic search across the codebase (or literal/regex search with --text).""" + # --text is a local scan (like `grep`): branch out *before* the project/daemon + # gate below, so it runs anywhere with no index and no daemon. + if text: + # --refresh / --offset are index concepts — meaningless for a live scan. + if refresh: + _typer.echo( + "Error: --refresh does not apply to --text (live scan, no index).", err=True + ) + raise _typer.Exit(code=1) + if offset: + _typer.echo("Error: --offset does not apply to --text (live scan, no index).", err=True) + raise _typer.Exit(code=1) + # A JSON shape for line-level text matches isn't defined yet — reject rather + # than silently ignore the flag. + if json_output: + _typer.echo("Error: --json is not supported with --text yet.", err=True) + raise _typer.Exit(code=1) + _run_text_search( + query, + path_glob=path, + languages=lang, + limit=limit, + ignore_case=ignore_case, + no_color=no_color, + ) + return + + # --ignore-case / --no-color only make sense for the --text local scan. + invalid = [n for n, v in (("--ignore-case", ignore_case), ("--no-color", no_color)) if v] + if invalid: + _typer.echo(f"Error: {', '.join(invalid)} only apply with --text.", err=True) + raise _typer.Exit(code=1) + project_root = str(require_project_root()) query_str = " ".join(query) diff --git a/src/cocoindex_code/file_walk.py b/src/cocoindex_code/file_walk.py index 62ebc3c..8c0cc9f 100644 --- a/src/cocoindex_code/file_walk.py +++ b/src/cocoindex_code/file_walk.py @@ -1,10 +1,13 @@ """Shared source-file walking: pattern + .gitignore matching, reused by the -indexer, the daemon's doctor file-walk, and ``ccc grep``. +indexer, the daemon's doctor file-walk, ``ccc grep``, and ``ccc search --text``. The matcher (include/exclude globs + nested ``.gitignore`` awareness) is the single source of truth for "which files count as part of the project". The -indexer feeds it to CocoIndex's incremental file source; the daemon and ``ccc -grep`` drive a plain :func:`os.walk` over it via :func:`iter_included_files`. +indexer feeds it to CocoIndex's incremental file source; the daemon drives a +plain :func:`os.walk` over it via :func:`iter_included_files`. The local searches +(``grep`` / ``--text``) share the higher-level :func:`iter_project_files`, which +resolves a root to its included files and yields display paths — leaving per-file +language gating to each caller. """ from __future__ import annotations @@ -12,11 +15,18 @@ import os from collections.abc import Iterable, Iterator from pathlib import Path, PurePath +from typing import NamedTuple from cocoindex.resources.file import FilePathMatcher, PatternFilePathMatcher from pathspec import GitIgnoreSpec -from .settings import load_gitignore_spec +from .settings import ( + DEFAULT_EXCLUDED_PATTERNS, + DEFAULT_INCLUDED_PATTERNS, + find_project_root, + load_gitignore_spec, + load_project_settings, +) def _normalize_gitignore_lines(lines: Iterable[str], directory: PurePath) -> list[str]: @@ -174,3 +184,75 @@ def iter_included_files( rel_path = rel_dir / fname if rel_dir != PurePath(".") else PurePath(fname) if matcher.is_file_included(rel_path): yield dirpath / fname, rel_path + + +class ProjectFiles(NamedTuple): + """A search root resolved to the files the project includes, plus the + extension→language overrides callers need for their own per-file language + decisions. + + Shared by ``ccc grep`` and ``ccc search --text``: the *walk* is common, while + per-file gating differs (grep skips files with no matchable code language; + text search keeps every included file). So this yields language-agnostic files + and hands the overrides back for the caller to decide. + """ + + files: Iterator[tuple[Path, str]] + """``(absolute_path, display_path)`` for every included file, in sorted walk + order — code, docs, and config alike.""" + + ext_overrides: dict[str, str] + """Project extension→language overrides (e.g. ``{".inc": "php"}``); ``{}`` + outside an initialized project.""" + + +def _language_overrides(project_root: Path | None) -> dict[str, str]: + """Extension→language overrides from project settings; ``{}`` if no project.""" + if project_root is None: + return {} + ps = load_project_settings(project_root) + return {f".{lo.ext}": lo.lang for lo in ps.language_overrides} + + +def iter_project_files(root: Path, path_glob: str | None = None) -> ProjectFiles: + """Resolve ``root`` to the project's included files — the single source of + truth shared by ``ccc grep`` and ``ccc search --text``. + + Honors the project's include/exclude + nested ``.gitignore`` (or the default + source-file patterns outside a project) and an optional ``path_glob``. ``root`` + may be a file (searched on its own) or a directory (walked recursively). + ``files`` yields display paths mirroring ``root`` (e.g. ``src/a.py``); language + gating is left to the caller. + """ + resolved = root.resolve() + + if resolved.is_file(): + # A single file: just it, anchored to its enclosing project (if any). + overrides = _language_overrides(find_project_root(resolved.parent)) + return ProjectFiles(files=iter([(resolved, root.as_posix())]), ext_overrides=overrides) + + project_root = find_project_root(resolved) + if project_root is not None: + ps = load_project_settings(project_root) + included, excluded = ps.include_patterns, ps.exclude_patterns + ext_overrides = {f".{lo.ext}": lo.lang for lo in ps.language_overrides} + base = project_root + else: + included = list(DEFAULT_INCLUDED_PATTERNS) + excluded = list(DEFAULT_EXCLUDED_PATTERNS) + ext_overrides = {} + # Anchor .gitignore at the enclosing git repo so a subdirectory search + # still honors the repo-root rules; fall back to the target itself. + base = find_git_root(resolved) or resolved + + matcher = build_matcher(base, included, excluded) + path_filter = PatternFilePathMatcher(included_patterns=[path_glob]) if path_glob else None + + def _walk() -> Iterator[tuple[Path, str]]: + for abs_path, rel in iter_included_files(resolved, base, matcher): + if path_filter is not None and not path_filter.is_file_included(rel): + continue + # Display paths mirror the root the user gave (e.g. "src/a.py"). + yield abs_path, (root / abs_path.relative_to(resolved)).as_posix() + + return ProjectFiles(files=_walk(), ext_overrides=ext_overrides) diff --git a/src/cocoindex_code/grep.py b/src/cocoindex_code/grep.py index d769410..71e84d6 100644 --- a/src/cocoindex_code/grep.py +++ b/src/cocoindex_code/grep.py @@ -17,18 +17,11 @@ from dataclasses import dataclass from pathlib import Path -import click from cocoindex.ops.code import CodeMatch, CodePattern from cocoindex.ops.text import detect_code_language -from cocoindex.resources.file import PatternFilePathMatcher -from .file_walk import build_matcher, find_git_root, iter_included_files -from .settings import ( - DEFAULT_EXCLUDED_PATTERNS, - DEFAULT_INCLUDED_PATTERNS, - find_project_root, - load_project_settings, -) +from .file_walk import iter_project_files +from .render import gutter, gutter_width, paint, path_header # A trivial, always-valid pattern (a bare identifier) used to probe whether a # language is structurally matchable, independent of the user's pattern. @@ -151,13 +144,6 @@ def _detect_language(path: Path, ext_overrides: dict[str, str]) -> str | None: return ext_overrides.get(path.suffix) or detect_code_language(filename=path.name) -def _ext_overrides(project_root: Path | None) -> dict[str, str]: - if project_root is None: - return {} - ps = load_project_settings(project_root) - return {f".{lo.ext}": lo.lang for lo in ps.language_overrides} - - def _target_for_file( abs_path: Path, display: str, @@ -201,41 +187,9 @@ def _iter_targets(req: GrepRequest, compiler: _PatternCompiler) -> Iterator[_Tar ``.gitignore`` rules decide which files belong to the project. Outside a project we fall back to the default source-file patterns. """ - root = req.root.resolve() - - if root.is_file(): - project_root = find_project_root(root.parent) - yield from _resolve_file( - root, req.root.as_posix(), _ext_overrides(project_root), req, compiler - ) - return - - project_root = find_project_root(root) - if project_root is not None: - ps = load_project_settings(project_root) - included, excluded = ps.include_patterns, ps.exclude_patterns - ext_overrides = {f".{lo.ext}": lo.lang for lo in ps.language_overrides} - base = project_root - else: - included = list(DEFAULT_INCLUDED_PATTERNS) - excluded = list(DEFAULT_EXCLUDED_PATTERNS) - ext_overrides = {} - # Anchor at the enclosing git repo so grepping a subdirectory still honors the - # repo-root (and intervening) .gitignore; fall back to the target dir itself. - base = find_git_root(root) or root - - matcher = build_matcher(base, included, excluded) - path_filter = ( - PatternFilePathMatcher(included_patterns=[req.path_glob]) if req.path_glob else None - ) - - for abs_path, rel in iter_included_files(root, base, matcher): - if path_filter is not None and not path_filter.is_file_included(rel): - continue - # Display paths mirror the root the user gave (e.g. "src/a.py", "/tmp/x/a.py"), - # rather than always being cwd-relative. - display = (req.root / abs_path.relative_to(root)).as_posix() - yield from _resolve_file(abs_path, display, ext_overrides, req, compiler) + walk = iter_project_files(req.root, req.path_glob) + for abs_path, display in walk.files: + yield from _resolve_file(abs_path, display, walk.ext_overrides, req, compiler) # --------------------------------------------------------------------------- @@ -333,12 +287,6 @@ def _line_char_offsets(source: str) -> list[int]: return offsets -def _paint(text: str, color: bool, **style: object) -> str: - if not color or not text: - return text - return click.style(text, **style) # type: ignore[arg-type] - - def _render_code_line( line_no: int, text: str, @@ -352,13 +300,13 @@ def _render_code_line( and the matched span shown normally.""" pre = max(0, min(dim_pre_end, len(text))) post = max(pre, min(dim_post_start, len(text))) - gutter = _paint(f"{line_no:>{width}}| ", color, fg="bright_black") + prefix = gutter(line_no, width, color=color) if not color: - return f"{gutter}{text}" - before = _paint(text[:pre], color, dim=True) + return f"{prefix}{text}" + before = paint(text[:pre], color, dim=True) matched = text[pre:post] - after = _paint(text[post:], color, dim=True) - return f"{gutter}{before}{matched}{after}" + after = paint(text[post:], color, dim=True) + return f"{prefix}{before}{matched}{after}" def _render_match( @@ -390,15 +338,15 @@ def render_file(fm: FileMatches, *, color: bool) -> str: src_lines = [line.rstrip("\r") for line in fm.source.split("\n")] offsets = _line_char_offsets(fm.source) max_line = max((m.chunks[0].end.line for m in fm.matches if m.chunks), default=1) - width = len(str(max_line)) + width = gutter_width(max_line) - parts = [_paint(fm.path, color, fg="magenta", bold=True)] + parts = [path_header(fm.path, color=color)] emitted = False for match in fm.matches: if not match.chunks: continue if emitted: - parts.append(_paint("---", color, fg="bright_black")) + parts.append(paint("---", color, fg="bright_black")) parts.extend(_render_match(src_lines, offsets, match, width, color)) emitted = True return "\n".join(parts) diff --git a/src/cocoindex_code/render.py b/src/cocoindex_code/render.py new file mode 100644 index 0000000..d40eb5a --- /dev/null +++ b/src/cocoindex_code/render.py @@ -0,0 +1,35 @@ +"""Shared terminal rendering for the local searches — ``ccc grep`` and +``ccc search --text``. + +Both print the same shape: a bold path header, then ``| `` rows behind +a dim gutter. Those styling primitives live here so the two stay in sync. What each +search draws *inside* a line differs — grep dims the context around a structural +match, text search highlights the matched spans — so that part stays with each +caller. +""" + +from __future__ import annotations + +import click + + +def paint(text: str, color: bool, **style: object) -> str: + """``click.style(text, **style)`` when ``color`` is on, else ``text`` unchanged.""" + if not color or not text: + return text + return click.style(text, **style) # type: ignore[arg-type] + + +def path_header(path: str, *, color: bool) -> str: + """The bold path line that opens one file's matches.""" + return paint(path, color, fg="magenta", bold=True) + + +def gutter_width(max_line: int) -> int: + """Right-align width for line numbers, so a file's gutters line up.""" + return len(str(max_line)) + + +def gutter(line_no: int, width: int, *, color: bool) -> str: + """The dim ``| `` prefix — number, pipe, then exactly one space.""" + return paint(f"{line_no:>{width}}| ", color, fg="bright_black") diff --git a/src/cocoindex_code/textsearch.py b/src/cocoindex_code/textsearch.py new file mode 100644 index 0000000..cd173f2 --- /dev/null +++ b/src/cocoindex_code/textsearch.py @@ -0,0 +1,337 @@ +r"""``ccc search --text`` — literal / regex full-text search over files. + +Unlike ``ccc search`` (semantic: needs the index + daemon + embeddings), text +search runs entirely locally, like ``ccc grep``: it walks the project's source +files — the *same* include/exclude + ``.gitignore`` rules the indexer uses — and +matches each file's *content* against literal terms and/or regexes. No index or +daemon is required, and results are always fresh. + +Query model (a small GitHub-code-search subset): + +* One or more **terms**, AND-combined: a file matches only if *every* term + appears in it (on some line). +* A term wrapped in slashes — ``/expr/`` — is a **regex**, *unless* its body has + an unescaped ``/`` (then it stays a **literal**, e.g. ``/blobs/v1/``) — matching + GitHub code search. Any other term is a literal substring. To regex-match a + literal slash, escape it inside the body: ``/\/foo\//``. +* **Smart-case:** matching is case-insensitive unless the term contains an + uppercase letter (decided per term), matching ripgrep's default. ``-i`` / ``-s`` + force the choice. + +Structure mirrors ``grep.py``: walk → per-file match → stream results as each +file completes. +""" + +from __future__ import annotations + +import re +import threading +from collections.abc import Callable, Iterator +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass +from pathlib import Path + +from cocoindex.ops.text import detect_code_language + +from .file_walk import iter_project_files +from .render import gutter, gutter_width, paint, path_header + + +class TermSyntaxError(ValueError): + """A term could not be parsed / compiled (e.g. an invalid regex).""" + + +@dataclass(frozen=True, slots=True) +class TextSearchWarning: + """A non-fatal problem surfaced during a run — e.g. a file that couldn't be + read. The search keeps going; the CLI prints these to stderr.""" + + message: str + + +@dataclass(frozen=True, slots=True) +class LineMatch: + """One matching line and the character spans matched within it.""" + + line_no: int + """1-based line number.""" + + text: str + """The line's text, without the trailing newline.""" + + spans: list[tuple[int, int]] + """``(start, end)`` char offsets of matched substrings — sorted and merged.""" + + +@dataclass(frozen=True, slots=True) +class FileMatches: + """Every matching line found in one file (at least one).""" + + path: str + """Display path, mirroring the search root (e.g. ``src/a.py``).""" + + matches: list[LineMatch] + + +@dataclass(frozen=True, slots=True) +class Term: + """One compiled search term. + + Keeps the metadata a future index needs to decide *per term* whether it can + use an inverted-index lookup (literal) or must fall back to a scan (regex) — + the raw text, whether it's a regex, and its case sensitivity — alongside the + compiled ``pattern`` used for matching today. + """ + + raw: str + """The term as typed (e.g. ``password`` or ``/def \\w+\\(/``).""" + + is_regex: bool + """True for a ``/…/`` regex term; False for a literal substring.""" + + case_sensitive: bool + """The resolved case mode (after smart-case / ``-i`` / ``-s``).""" + + pattern: re.Pattern[str] + """Compiled matcher — the case flag is already baked in.""" + + +@dataclass(frozen=True, slots=True) +class TextSearchRequest: + """A text-search invocation.""" + + terms: tuple[Term, ...] + """Search terms, AND-combined — a file matches only if every term does.""" + + root: Path + """Directory to search.""" + + languages: frozenset[str] | None = None + """Restrict to files of these languages (lowercased canonical names); + ``None`` = every included file, regardless of language.""" + + path_glob: str | None = None + """Extra include glob (globset syntax) on the project-relative path.""" + + +# --------------------------------------------------------------------------- +# Term parsing +# --------------------------------------------------------------------------- + + +def _looks_like_regex(raw: str) -> bool: + r"""Whether ``raw`` is a ``/…/`` regex term (GitHub-code-search rule). + + A term is a regex only if it is wrapped in slashes *and* its body has no + *free* (unescaped) ``/``. So ``/def \w+\(/`` is a regex, but ``/blobs/v1/`` + stays a literal (the middle ``/`` is free) — matching GitHub. To regex-match + a literal slash, escape it inside the body: ``/\/foo\//``. + """ + if len(raw) < 3 or raw[0] != "/" or raw[-1] != "/": + return False + # Drop escaped ``\\`` then ``\/``; any ``/`` still left is a free slash. + stripped = raw[1:-1].replace("\\\\", "").replace("\\/", "") + return "/" not in stripped + + +def compile_terms(raw_terms: list[str], *, case_sensitive: bool | None) -> list[Term]: + """Compile each raw term into a :class:`Term`. + + ``case_sensitive`` forces the case handling for every term; ``None`` selects + smart-case per term (case-sensitive iff the term contains an uppercase + letter). Raises :class:`TermSyntaxError` on an invalid regex. + """ + terms: list[Term] = [] + for raw in raw_terms: + is_regex = _looks_like_regex(raw) + source = raw[1:-1] if is_regex else re.escape(raw) + for_case = raw[1:-1] if is_regex else raw + cs = case_sensitive if case_sensitive is not None else any(c.isupper() for c in for_case) + flags = 0 if cs else re.IGNORECASE + try: + pattern = re.compile(source, flags) + except re.error as e: + raise TermSyntaxError(f"invalid regex {raw!r}: {e}") from e + terms.append(Term(raw=raw, is_regex=is_regex, case_sensitive=cs, pattern=pattern)) + return terms + + +# --------------------------------------------------------------------------- +# Matching +# --------------------------------------------------------------------------- + + +def _merge_spans(spans: list[tuple[int, int]]) -> list[tuple[int, int]]: + """Sort and merge overlapping/adjacent spans.""" + if not spans: + return spans + spans.sort() + merged = [spans[0]] + for start, end in spans[1:]: + last_start, last_end = merged[-1] + if start <= last_end: + merged[-1] = (last_start, max(last_end, end)) + else: + merged.append((start, end)) + return merged + + +def _match_file( + path: Path, display: str, terms: tuple[Term, ...] +) -> FileMatches | TextSearchWarning | None: + """Match one file. + + File-level AND, line-oriented: the file qualifies only if *every* term + matches on at least one line; the returned lines are those matching *any* + term. Returns ``None`` for a binary/undecodable file or one that doesn't + satisfy the AND, and a :class:`TextSearchWarning` for an unreadable file. + """ + try: + content = path.read_text(encoding="utf-8") + except UnicodeDecodeError: + return None # binary / non-UTF-8 → silent skip, like `grep -I` + except OSError as e: + return TextSearchWarning(f"cannot read {display}: {e}") + + seen_terms: set[int] = set() + line_matches: list[LineMatch] = [] + for line_no, raw_line in enumerate(content.split("\n"), start=1): + line = raw_line.rstrip("\r") + spans: list[tuple[int, int]] = [] + for i, term in enumerate(terms): + matched_here = False + for m in term.pattern.finditer(line): + if m.end() > m.start(): # ignore zero-width matches + spans.append((m.start(), m.end())) + matched_here = True + if matched_here: + seen_terms.add(i) + if spans: + line_matches.append(LineMatch(line_no, line, _merge_spans(spans))) + + if len(seen_terms) != len(terms): + return None # AND not satisfied — some term never appeared + return FileMatches(path=display, matches=line_matches) if line_matches else None + + +# --------------------------------------------------------------------------- +# File selection (single source of truth with the indexer, via file_walk) +# --------------------------------------------------------------------------- + + +def _iter_files(req: TextSearchRequest) -> Iterator[tuple[Path, str]]: + """Yield ``(absolute_path, display_path)`` for every file to search, using the + same include/exclude + ``.gitignore`` rules as the indexer and ``ccc grep``. + + Unlike ``grep``, a file is *not* skipped for lacking a matchable code + language — text search covers every included file (source, docs, config). + ``--lang`` optionally restricts by detected language. + """ + walk = iter_project_files(req.root, req.path_glob) + for abs_path, display in walk.files: + if req.languages is not None: + language = walk.ext_overrides.get(abs_path.suffix) or detect_code_language( + filename=abs_path.name + ) + if language is None or language.lower() not in req.languages: + continue + yield abs_path, display + + +class TextSearch: + """A single text-search run. :meth:`run` matches each file as it's listed and + streams results as they complete.""" + + def __init__(self, req: TextSearchRequest) -> None: + self._req = req + + def run( + self, + emit: Callable[[FileMatches | TextSearchWarning], object], + *, + limit: int | None = None, + ) -> bool: + """Match the request, calling ``emit`` with each file's matches (and any + read warning) the moment it's ready — while the walk is still running. + + The walk runs on the calling thread and submits each file to a pool. + Python regex matching holds the GIL, so the pool mainly overlaps file + I/O rather than the matching itself; still, results stream as they + finish. ``emit`` is called from worker threads, so a consumer doing I/O + must serialize itself. + + With ``limit`` set, at most ``limit`` matching files are emitted and the + walk stops as soon as that many are found (warnings are never capped). + Returns ``True`` if the walk was cut short by ``limit`` — i.e. more files + may match. + """ + terms = self._req.terms + matched = 0 + hit_limit = False + lock = threading.Lock() + stop = threading.Event() + + def _match(item: tuple[Path, str]) -> None: + nonlocal matched, hit_limit + if stop.is_set(): + return + result = _match_file(item[0], item[1], terms) + if result is None: # binary / undecodable / AND not met + return + if isinstance(result, TextSearchWarning): + emit(result) + return + with lock: + if limit is not None and matched >= limit: + return + matched += 1 + if limit is not None and matched >= limit: + hit_limit = True + stop.set() # enough matches — stop feeding the walk + emit(result) + + with ThreadPoolExecutor() as pool: + for abs_path, display in _iter_files(self._req): + if stop.is_set(): + break + pool.submit(_match, (abs_path, display)) + # ThreadPoolExecutor.__exit__ waits for every submitted match. + + return hit_limit + + +# --------------------------------------------------------------------------- +# Rendering +# --------------------------------------------------------------------------- + + +def _highlight_line(text: str, spans: list[tuple[int, int]], color: bool) -> str: + """Render a line with its matched spans emphasized.""" + if not color or not spans: + return text + out: list[str] = [] + cursor = 0 + for start, end in spans: + out.append(text[cursor:start]) + out.append(paint(text[start:end], color, fg="red", bold=True)) + cursor = end + out.append(text[cursor:]) + return "".join(out) + + +def render_file(fm: FileMatches, *, color: bool) -> str: + """Render one file's matches: the path header, then ``line| text`` per + matching line with matched spans emphasized — mirroring ``ccc grep``.""" + max_line = max((lm.line_no for lm in fm.matches), default=1) + width = gutter_width(max_line) + parts = [path_header(fm.path, color=color)] + for lm in fm.matches: + prefix = gutter(lm.line_no, width, color=color) + parts.append(f"{prefix}{_highlight_line(lm.text, lm.spans, color)}") + return "\n".join(parts) + + +def render_results(results: list[FileMatches], *, color: bool) -> str: + """Render a list of per-file matches, files separated by a blank line. The + CLI streams with :func:`render_file` instead; this is the batch form (tests).""" + return "\n\n".join(render_file(fm, color=color) for fm in results) diff --git a/tests/test_textsearch.py b/tests/test_textsearch.py new file mode 100644 index 0000000..463a6a8 --- /dev/null +++ b/tests/test_textsearch.py @@ -0,0 +1,423 @@ +"""Tests for ``ccc search --text`` — literal / regex full-text search. + +These run entirely locally (no daemon, no index, no embeddings): the engine +walks files on disk and matches their content against literal terms / regexes. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from typer.testing import CliRunner + +from cocoindex_code import textsearch as ts +from cocoindex_code.cli import app + +runner = CliRunner() + + +def req_for( + root: Path, + *raw_terms: str, + case_sensitive: bool | None = None, + languages: frozenset[str] | None = None, + path_glob: str | None = None, +) -> ts.TextSearchRequest: + """Build a request, compiling ``raw_terms`` the way the CLI does.""" + return ts.TextSearchRequest( + terms=tuple(ts.compile_terms(list(raw_terms), case_sensitive=case_sensitive)), + root=root, + languages=languages, + path_glob=path_glob, + ) + + +def collect_ts(req: ts.TextSearchRequest) -> list[ts.FileMatches | ts.TextSearchWarning]: + """Drain a run into a list (matches + read warnings), completion order.""" + items: list[ts.FileMatches | ts.TextSearchWarning] = [] + ts.TextSearch(req).run(items.append) + return items + + +def run_ts(req: ts.TextSearchRequest) -> list[ts.FileMatches]: + """Just the file matches (dropping warnings), sorted by path for deterministic + assertions (the engine itself yields in completion order).""" + files = [it for it in collect_ts(req) if isinstance(it, ts.FileMatches)] + files.sort(key=lambda fm: fm.path) + return files + + +def names_of(files: list[ts.FileMatches]) -> set[str]: + return {Path(fm.path).name for fm in files} + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def corpus(tmp_path: Path) -> Path: + """A small mixed tree (code + docs + config) with no cocoindex project marker.""" + (tmp_path / "src").mkdir() + (tmp_path / "src" / "auth.py").write_text( + "def authenticate(user, password):\n" + " # TODO: add rate limiting\n" + " return check(password)\n" + ) + (tmp_path / "src" / "db.py").write_text( + "def connect(dsn):\n password = dsn.get('password')\n return password\n" + ) + (tmp_path / "src" / "util").mkdir() + (tmp_path / "src" / "util" / "log.py").write_text( + "def log_error(msg):\n # TODO: structured logging\n print(msg)\n" + ) + (tmp_path / "README.md").write_text( + "# Project\n\nHandles authentication and passwords.\n\nTODO: write docs\n" + ) + (tmp_path / "config.yaml").write_text("db:\n host: localhost\n password: s3cret\n") + (tmp_path / "notes.txt").write_text("Remember the Password policy.\nNothing else here.\n") + # Non-UTF-8 bytes, but a whitelisted extension: must be skipped silently. + (tmp_path / "src" / "blob.py").write_bytes(b"\x00\x01\xff\xfe password \x80\x81 TODO\x00") + return tmp_path + + +# --------------------------------------------------------------------------- +# Term parsing +# --------------------------------------------------------------------------- + + +def _one(raw: str, *, case_sensitive: bool | None = None) -> ts.Term: + (term,) = ts.compile_terms([raw], case_sensitive=case_sensitive) + return term + + +def test_literal_matches_substring_smart_case() -> None: + term = _one("password") + assert not term.is_regex + assert term.pattern.search("has a password here") + assert term.pattern.search("PASSWORD") # lowercase term → smart-case insensitive + + +def test_uppercase_term_is_case_sensitive() -> None: + term = _one("Password") + assert term.case_sensitive + assert term.pattern.search("Password") + assert not term.pattern.search("password") # an uppercase letter → case-sensitive + + +def test_case_override() -> None: + assert _one("Password", case_sensitive=False).pattern.search("password") + assert not _one("password", case_sensitive=True).pattern.search("PASSWORD") + + +def test_literal_metachars_are_escaped() -> None: + # A literal with regex metacharacters matches literally, not as a pattern. + term = _one("a.b") + assert not term.is_regex + assert term.pattern.search("a.b") + assert not term.pattern.search("axb") # '.' is a literal dot, not "any char" + + +def test_regex_term() -> None: + term = _one(r"/def \w+\(/") + assert term.is_regex + assert term.pattern.search("def foo(") + assert not term.pattern.search("definitely typed") + + +def test_slash_wrapped_with_free_slash_stays_literal() -> None: + # GitHub rule: a free (unescaped) `/` in the body → literal, not a regex. + term = _one("/blobs/docs-v1/") + assert not term.is_regex + assert term.pattern.search("see /blobs/docs-v1/ path") + assert not term.pattern.search("blobs docs-v1") + + +def test_regex_can_match_literal_slashes_when_escaped() -> None: + # Escaping the slashes inside a regex matches a literal `/foo/`. + term = _one(r"/\/foo\//") + assert term.is_regex + assert term.pattern.search("x /foo/ y") + assert not term.pattern.search("foo") + + +def test_bare_double_slash_is_literal() -> None: + term = _one("//") # too short to be a regex body + assert not term.is_regex + assert term.pattern.search("a // comment") + + +def test_invalid_regex_raises() -> None: + with pytest.raises(ts.TermSyntaxError): + ts.compile_terms(["/def(/"], case_sensitive=None) + + +# --------------------------------------------------------------------------- +# Engine +# --------------------------------------------------------------------------- + + +def test_single_literal_across_files(corpus: Path) -> None: + files = run_ts(req_for(corpus, "password")) + # Every file containing "password" (case-insensitive); the binary blob skipped. + assert names_of(files) == {"auth.py", "db.py", "README.md", "config.yaml", "notes.txt"} + + +def test_and_semantics(corpus: Path) -> None: + # A file qualifies only if *every* term appears. "TODO" (uppercase → case- + # sensitive) + "password" → only auth.py and README.md carry both. + files = run_ts(req_for(corpus, "TODO", "password")) + assert names_of(files) == {"auth.py", "README.md"} + + +def test_smart_case_sensitive(corpus: Path) -> None: + files = run_ts(req_for(corpus, "Password")) # uppercase → case-sensitive + assert names_of(files) == {"notes.txt"} + + +def test_case_insensitive_override(corpus: Path) -> None: + files = run_ts(req_for(corpus, "Password", case_sensitive=False)) + assert names_of(files) == {"auth.py", "db.py", "README.md", "config.yaml", "notes.txt"} + + +def test_regex_across_files(corpus: Path) -> None: + files = run_ts(req_for(corpus, r"/def \w+\(/")) + assert names_of(files) == {"auth.py", "db.py", "log.py"} + + +def test_language_filter(corpus: Path) -> None: + files = run_ts(req_for(corpus, "password", languages=frozenset({"python"}))) + assert names_of(files) == {"auth.py", "db.py"} + + +def test_path_glob(corpus: Path) -> None: + files = run_ts(req_for(corpus, "password", path_glob="src/**")) + assert names_of(files) == {"auth.py", "db.py"} + + +def test_covers_non_code_files(corpus: Path) -> None: + # Unlike `grep`, text search matches docs/config that have no code language. + files = run_ts(req_for(corpus, "localhost")) + assert names_of(files) == {"config.yaml"} + + +def test_binary_file_skipped_silently(corpus: Path) -> None: + items = collect_ts(req_for(corpus, "password")) + matched = [it for it in items if isinstance(it, ts.FileMatches)] + assert not any(Path(fm.path).name == "blob.py" for fm in matched) + # A non-UTF-8 file is a silent skip (like `grep -I`), not a warning. + assert not any(isinstance(it, ts.TextSearchWarning) for it in items) + + +def test_no_matches(corpus: Path) -> None: + assert run_ts(req_for(corpus, "zzz_absent_zzz")) == [] + + +def test_matched_lines_are_reported(corpus: Path) -> None: + # auth.py matches "password" on lines 1 and 3 (not the TODO line 2). + (auth,) = [fm for fm in run_ts(req_for(corpus, "password")) if fm.path.endswith("auth.py")] + assert [lm.line_no for lm in auth.matches] == [1, 3] + + +def test_many_files(tmp_path: Path) -> None: + # Many files exercise the thread pool: every file matched exactly once, no + # duplicates and no lost results. + n = 300 + for i in range(n): + (tmp_path / f"f{i:04d}.py").write_text(f"value_{i} = {i}\n# marker line\n") + files = run_ts(req_for(tmp_path, "marker")) + assert len(files) == n + assert len({fm.path for fm in files}) == n + + +def test_run_respects_limit(corpus: Path) -> None: + # "password" matches 5 files; limit=2 emits at most 2 and reports it stopped early. + got: list[ts.FileMatches | ts.TextSearchWarning] = [] + hit = ts.TextSearch(req_for(corpus, "password")).run(got.append, limit=2) + files = [it for it in got if isinstance(it, ts.FileMatches)] + assert len(files) <= 2 + assert hit is True + + +def test_run_no_limit_returns_all(corpus: Path) -> None: + got: list[ts.FileMatches | ts.TextSearchWarning] = [] + hit = ts.TextSearch(req_for(corpus, "password")).run(got.append) + files = [it for it in got if isinstance(it, ts.FileMatches)] + assert len(files) == 5 + assert hit is False + + +# --------------------------------------------------------------------------- +# Rendering +# --------------------------------------------------------------------------- + + +def test_render_plain_format(tmp_path: Path) -> None: + (tmp_path / "a.py").write_text("x = 1\nfoo = 2\n") + rendered = ts.render_results(run_ts(req_for(tmp_path, "foo")), color=False) + lines = rendered.split("\n") + assert lines[0] == (tmp_path / "a.py").as_posix() # path header + assert lines[1] == "2| foo = 2" # "| ", one space after the pipe + + +def test_render_strips_crlf(tmp_path: Path) -> None: + (tmp_path / "crlf.py").write_text("a = 1\r\nfoo = 2\r\n", newline="") + rendered = ts.render_results(run_ts(req_for(tmp_path, "foo")), color=False) + assert "\r" not in rendered + assert "2| foo = 2" in rendered + + +def test_render_line_number_width(tmp_path: Path) -> None: + (tmp_path / "w.py").write_text("".join(f"mark {i}\n" for i in range(1, 14))) # 13 lines + rendered = ts.render_results(run_ts(req_for(tmp_path, "mark")), color=False) + assert "\n 1| mark 1" in rendered # single-digit line, padded to width 2 + assert "\n13| mark 13" in rendered + + +def test_render_color_highlights_match(tmp_path: Path) -> None: + (tmp_path / "a.py").write_text("the password is here\n") + rendered = ts.render_results(run_ts(req_for(tmp_path, "password")), color=True) + assert "\x1b[" in rendered # ANSI escapes present when color is on + + +# --------------------------------------------------------------------------- +# Project- and gitignore-awareness (single source of truth with the indexer) +# --------------------------------------------------------------------------- + + +def test_respects_project_exclude_patterns(tmp_path: Path) -> None: + (tmp_path / ".cocoindex_code").mkdir() + (tmp_path / ".cocoindex_code" / "settings.yml").write_text( + "include_patterns:\n - '**/*.py'\nexclude_patterns:\n - '**/.*'\n - '**/skip'\n" + ) + (tmp_path / "keep.py").write_text("a token here\n") + (tmp_path / "skip").mkdir() + (tmp_path / "skip" / "hidden.py").write_text("a token here\n") + + files = run_ts(req_for(tmp_path, "token")) + assert "keep.py" in names_of(files) + assert not any("skip" in fm.path for fm in files) + + +def test_respects_gitignore(tmp_path: Path) -> None: + (tmp_path / ".cocoindex_code").mkdir() + (tmp_path / ".cocoindex_code" / "settings.yml").write_text("include_patterns:\n - '**/*.py'\n") + (tmp_path / ".gitignore").write_text("ignored.py\n") + (tmp_path / "kept.py").write_text("a token\n") + (tmp_path / "ignored.py").write_text("a token\n") + + names = names_of(run_ts(req_for(tmp_path, "token"))) + assert "kept.py" in names + assert "ignored.py" not in names + + +# --------------------------------------------------------------------------- +# CLI end-to-end (via CliRunner — no daemon needed) +# --------------------------------------------------------------------------- + + +def test_cli_basic(corpus: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(corpus) + result = runner.invoke(app, ["search", "--text", "password"], catch_exceptions=False) + assert result.exit_code == 0 + assert "auth.py" in result.output + assert "db.py" in result.output + + +def test_cli_and(corpus: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(corpus) + result = runner.invoke(app, ["search", "--text", "TODO", "password"], catch_exceptions=False) + assert result.exit_code == 0 + assert "auth.py" in result.output + assert "db.py" not in result.output # db.py lacks TODO + + +def test_cli_no_matches(corpus: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(corpus) + result = runner.invoke(app, ["search", "--text", "zzz_absent"], catch_exceptions=False) + assert result.exit_code == 0 + assert "No matches found." in result.output + + +def test_cli_invalid_regex(corpus: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(corpus) + result = runner.invoke(app, ["search", "--text", "/def(/"], catch_exceptions=False) + assert result.exit_code == 1 + assert "invalid regex" in result.output + + +def test_cli_refresh_rejected(corpus: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(corpus) + result = runner.invoke( + app, ["search", "--text", "--refresh", "password"], catch_exceptions=False + ) + assert result.exit_code == 1 + assert "--refresh does not apply" in result.output + + +def test_cli_offset_rejected(corpus: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(corpus) + result = runner.invoke( + app, ["search", "--text", "--offset", "3", "password"], catch_exceptions=False + ) + assert result.exit_code == 1 + assert "--offset does not apply" in result.output + + +def test_cli_json_rejected(corpus: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(corpus) + # --json has no defined shape for text matches yet; reject rather than ignore. + result = runner.invoke(app, ["search", "--text", "--json", "password"], catch_exceptions=False) + assert result.exit_code == 1 + assert "--json is not supported with --text" in result.output + + +def test_cli_text_flags_rejected_without_text( + corpus: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(corpus) + # -i is a --text flag; using it for semantic search is an error, not a silent no-op. + result = runner.invoke(app, ["search", "-i", "password"], catch_exceptions=False) + assert result.exit_code == 1 + assert "only apply with --text" in result.output + + +def test_cli_lang_filter(corpus: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(corpus) + result = runner.invoke( + app, ["search", "--text", "--lang", "python", "password"], catch_exceptions=False + ) + assert result.exit_code == 0 + assert "auth.py" in result.output + assert "config.yaml" not in result.output + + +def test_cli_limit(corpus: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(corpus) + # "password" matches 5 files; --limit 2 caps the printed files to 2. + result = runner.invoke( + app, ["search", "--text", "--limit", "2", "password"], catch_exceptions=False + ) + assert result.exit_code == 0 + candidates = ["src/auth.py", "src/db.py", "README.md", "config.yaml", "notes.txt"] + assert sum(1 for f in candidates if f in result.output) == 2 + + +def test_cli_limit_hint(corpus: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(corpus) + # 5 files match; --limit 2 truncates → a hint to raise the limit is shown. + result = runner.invoke( + app, ["search", "--text", "--limit", "2", "password"], catch_exceptions=False + ) + assert result.exit_code == 0 + assert "higher --limit" in result.output + + +def test_cli_no_hint_under_limit(corpus: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(corpus) + # Only 1 file matches "localhost" — well under the limit, so no truncation hint. + result = runner.invoke(app, ["search", "--text", "localhost"], catch_exceptions=False) + assert result.exit_code == 0 + assert "higher --limit" not in result.output