Files
factory/lib/interns-mcp/interns_mcp/interns/grep_audit.py
vitya ca669d96e1 feat(lib): bundle MCP servers from .common/lib into factory/lib/
Copies source (no node_modules, dist, .tasks, .wiki, __pycache__) for:
- projects-meta-mcp v2.25.0 (TypeScript/Node)
- wiki-graph v0.3.1 (TypeScript/Node)
- interns-mcp v0.3.3 (Python/FastMCP)

.gitignore: exclude lib build artefacts (node_modules, dist, .venv, __pycache__, *.pyc)
bootstrap.ps1: add MCP build step — npm install+build for TS servers, venv+pip for Python

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-11 13:17:29 +03:00

105 lines
3.4 KiB
Python

"""Grep audit intern — deterministic matrix of N paths x M patterns. No LLM call."""
from __future__ import annotations
import json
import re
from pathlib import Path
from typing import Any, Callable, Literal
from .base import Intern, InternResponse
class GrepAudit(Intern):
id = "grep_audit"
description = (
"Deterministic grep matrix over N paths x M patterns (substring or regex). "
"No LLM call, no endpoint cost. Returns markdown table (default) or JSON."
)
def run(
self,
paths: list[str],
patterns: list[str | dict[str, Any]],
output: Literal["table", "json"] = "table",
case_sensitive: bool = True,
**_kwargs: Any,
) -> InternResponse:
compiled = _compile_patterns(patterns, case_sensitive)
rows: list[dict[str, Any]] = []
matches_total = 0
for p in paths:
try:
text = Path(p).read_text(encoding="utf-8", errors="replace")
except (FileNotFoundError, PermissionError, IsADirectoryError, OSError) as exc:
rows.append(
{
"path": p,
"matches": {c["name"]: None for c in compiled},
"error": str(exc),
}
)
continue
row_matches: dict[str, bool] = {}
for c in compiled:
hit = bool(c["matcher"](text))
row_matches[c["name"]] = hit
if hit:
matches_total += 1
rows.append({"path": p, "matches": row_matches})
if output == "json":
rendered = json.dumps({"rows": rows}, ensure_ascii=False, indent=2)
else:
rendered = _render_table(rows, [c["name"] for c in compiled])
return InternResponse(
text=rendered,
usage={
"files_scanned": len(paths),
"patterns_evaluated": len(compiled),
"matches_total": matches_total,
},
)
def _compile_patterns(
patterns: list[str | dict[str, Any]], case_sensitive: bool
) -> list[dict[str, Any]]:
out: list[dict[str, Any]] = []
for p in patterns:
if isinstance(p, str):
out.append({"name": p, "matcher": _substring_matcher(p, case_sensitive)})
continue
name = p.get("name", p["pattern"])
if p.get("regex"):
flags = 0 if case_sensitive else re.IGNORECASE
compiled = re.compile(p["pattern"], flags)
out.append({"name": name, "matcher": compiled.search})
else:
out.append(
{"name": name, "matcher": _substring_matcher(p["pattern"], case_sensitive)}
)
return out
def _substring_matcher(needle: str, case_sensitive: bool) -> Callable[[str], bool]:
if case_sensitive:
return lambda text: needle in text
lowered = needle.lower()
return lambda text: lowered in text.lower()
def _render_table(rows: list[dict[str, Any]], names: list[str]) -> str:
header = "| Path | " + " | ".join(names) + " |"
sep = "|" + "---|" * (len(names) + 1)
lines = [header, sep]
for row in rows:
cells: list[str] = []
for n in names:
v = row["matches"].get(n)
cells.append("⚠️" if v is None else ("" if v else ""))
lines.append(f"| `{row['path']}` | " + " | ".join(cells) + " |")
return "\n".join(lines)