Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
112 changes: 110 additions & 2 deletions src/cocoindex_code/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
ProjectStatusResponse,
SearchResponse,
)
from .textsearch import FileMatches as TextFileMatches
from .textsearch import TextSearchWarning

from .settings import (
DEFAULT_ST_MODEL,
Expand Down Expand Up @@ -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,
)

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think if we stopped after hitting limit, we may output a message to hint about this, and may say something like if want to see more output, retry with a higher limit.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added — when it stops at the limit it prints Stopped at --limit N. Re-run with a higher --limit to see more. (run returns whether it stopped early, so the CLI knows when to show it.) (e6476c8)


_GITIGNORE_COMMENT = "# CocoIndex Code (ccc)"
_GITIGNORE_ENTRY = "/.cocoindex_code/"

Expand Down Expand Up @@ -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)."
),
Comment on lines +729 to +731

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

They're redundant to each other. We only need one.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Dropped -s/--case-sensitive and kept just -i/--ignore-case. Smart-case already gives case-sensitivity when a term has an uppercase letter, so the only override worth a flag is force-insensitive (-i). (1a1c7c0)

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
)

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

--offset is also not supported for --text, so we should also echo an error if offset is not 0.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done — --text now errors on --offset, same as --refresh: both are index concepts with no meaning for a live scan. (1a1c7c0) (same as the reply for copilot)

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)

Expand Down
90 changes: 86 additions & 4 deletions src/cocoindex_code/file_walk.py
Original file line number Diff line number Diff line change
@@ -1,22 +1,32 @@
"""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

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]:
Expand Down Expand Up @@ -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)
78 changes: 13 additions & 65 deletions src/cocoindex_code/grep.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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)


# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -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,
Expand All @@ -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(
Expand Down Expand Up @@ -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)
Expand Down
Loading