mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-08-17 18:25:42 +02:00
[eric] lint: vendor structural linter + config
This commit is contained in:
@@ -9,3 +9,7 @@
|
||||
|
||||
pytest==8.3.4
|
||||
pytest-asyncio==0.25.2
|
||||
|
||||
# Used by linter/lint.py (the vulture dead-code check). watchfiles, also
|
||||
# needed by lint.py, already comes in transitively via uvicorn[standard].
|
||||
vulture==2.16
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
__pycache__/
|
||||
*.pyc
|
||||
@@ -0,0 +1,68 @@
|
||||
"""Check infrastructure: shared filter/match utilities."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import fnmatch
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
LINTIGNORE_PREFIX = ".lintignore"
|
||||
|
||||
|
||||
def _matches_any(text: str, patterns: list[str]) -> bool:
|
||||
return any(fnmatch.fnmatch(text, p) for p in patterns)
|
||||
|
||||
|
||||
def is_excluded(path: Path, root: Path, excludes: list[str]) -> bool:
|
||||
rel = path.relative_to(root)
|
||||
for part in rel.parts:
|
||||
if _matches_any(part, excludes):
|
||||
return True
|
||||
return _matches_any(str(rel), excludes)
|
||||
|
||||
|
||||
def is_excepted(rel_path: str, rule: str, exceptions: dict[str, list[str]]) -> bool:
|
||||
return _matches_any(rel_path, exceptions.get(rule, []))
|
||||
|
||||
|
||||
def collect_lintignores(root: Path, excludes: list[str]) -> dict[Path, set[str]]:
|
||||
"""Scan *root* for ``.lintignore*`` sentinel files.
|
||||
|
||||
Returns ``{directory: set_of_ignored_rules}``.
|
||||
The special token ``"__all__"`` means every rule is ignored.
|
||||
"""
|
||||
ignores: dict[Path, set[str]] = {}
|
||||
for dirpath_str, dirnames, filenames in os.walk(root):
|
||||
dp = Path(dirpath_str)
|
||||
if is_excluded(dp, root, excludes):
|
||||
dirnames.clear()
|
||||
continue
|
||||
for fname in filenames:
|
||||
if fname == LINTIGNORE_PREFIX:
|
||||
ignores.setdefault(dp, set()).add("__all__")
|
||||
elif fname.startswith(f"{LINTIGNORE_PREFIX}-"):
|
||||
rule = fname[len(LINTIGNORE_PREFIX) + 1 :]
|
||||
ignores.setdefault(dp, set()).add(rule)
|
||||
return ignores
|
||||
|
||||
|
||||
def is_lintignored(
|
||||
path: Path,
|
||||
root: Path,
|
||||
rule: str,
|
||||
ignores: dict[Path, set[str]],
|
||||
) -> bool:
|
||||
"""Return True if *path* is covered by a ``.lintignore`` file for *rule*.
|
||||
|
||||
Walks from *path* up to *root* checking each ancestor directory.
|
||||
"""
|
||||
current = path if path.is_dir() else path.parent
|
||||
root = root.resolve()
|
||||
while True:
|
||||
rules = ignores.get(current)
|
||||
if rules and ("__all__" in rules or rule in rules):
|
||||
return True
|
||||
if current == root:
|
||||
break
|
||||
current = current.parent
|
||||
return False
|
||||
@@ -0,0 +1,61 @@
|
||||
"""Class-level dead code detection with framework awareness.
|
||||
|
||||
Pydantic BaseModel subclasses are auto-whitelisted: every annotated field is
|
||||
part of the serialization schema and therefore intentionally "used".
|
||||
|
||||
Non-framework classes are skipped for now (tier 2 — future cross-referencing).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
from pathlib import Path
|
||||
|
||||
from . import is_excepted, is_excluded, is_lintignored
|
||||
|
||||
FRAMEWORK_BASES = {"BaseModel"}
|
||||
|
||||
|
||||
def _is_framework_model(cls: ast.ClassDef) -> bool:
|
||||
return any(
|
||||
(isinstance(b, ast.Name) and b.id in FRAMEWORK_BASES)
|
||||
or (isinstance(b, ast.Attribute) and b.attr in FRAMEWORK_BASES)
|
||||
for b in cls.bases
|
||||
)
|
||||
|
||||
|
||||
def run_class_check(
|
||||
root: Path,
|
||||
exceptions: dict[str, list[str]],
|
||||
excludes: list[str],
|
||||
ignores: dict[Path, set[str]] | None = None,
|
||||
) -> list[str]:
|
||||
"""Analyse classes in backend Python files and return errors."""
|
||||
errors: list[str] = []
|
||||
backend = root / "backend"
|
||||
if not backend.is_dir():
|
||||
return errors
|
||||
|
||||
for pyfile in sorted(backend.rglob("*.py")):
|
||||
if is_excluded(pyfile, root, excludes):
|
||||
continue
|
||||
rel = str(pyfile.relative_to(root))
|
||||
if is_excepted(rel, "classes", exceptions):
|
||||
continue
|
||||
if ignores and is_lintignored(pyfile, root, "classes", ignores):
|
||||
continue
|
||||
try:
|
||||
source = pyfile.read_text()
|
||||
tree = ast.parse(source, filename=rel)
|
||||
except (OSError, SyntaxError):
|
||||
continue
|
||||
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.ClassDef):
|
||||
continue
|
||||
if _is_framework_model(node):
|
||||
continue
|
||||
# Tier 2 placeholder: non-framework classes are skipped until
|
||||
# cross-reference analysis is implemented.
|
||||
|
||||
return errors
|
||||
@@ -0,0 +1,210 @@
|
||||
"""Orphaned endpoint detection — cross-references backend routes with usage.
|
||||
|
||||
Extracts all registered API routes from the backend (decorator and add_api_route
|
||||
patterns) and checks whether each route's static path segments appear in the
|
||||
frontend source or in other backend files (e.g. MCP servers that call endpoints
|
||||
internally). Routes with no matching reference anywhere are flagged.
|
||||
|
||||
Limitations (v1):
|
||||
- Routes that end with a path parameter (e.g. /{id}) and have no trailing
|
||||
static segment are skipped — they're too ambiguous to match.
|
||||
- WebSocket routes in main.py are not checked.
|
||||
- Backend-only endpoints (health checks, OAuth callbacks) should be excluded
|
||||
via the exceptions list or endpoint-ignore-routes in config.json.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import fnmatch
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from . import is_excepted, is_lintignored
|
||||
|
||||
_DECORATOR_RE = re.compile(
|
||||
r"@(\w+)\.router\.\w+\(\s*[\"']([^\"']+)[\"']"
|
||||
)
|
||||
_ADD_ROUTE_RE = re.compile(
|
||||
r"(\w+)\.router\.add_api_route\(\s*[\"']([^\"']+)[\"']"
|
||||
)
|
||||
_SUBAPP_RE = re.compile(
|
||||
r"(\w+)\s*=\s*SubApp\(\s*[\"']([^\"']+)[\"']"
|
||||
)
|
||||
_FUNC_DEF_RE = re.compile(r"\s*(?:async\s+)?def\s+(\w+)")
|
||||
_ADD_ROUTE_FUNC_RE = re.compile(r"add_api_route\([^,]+,\s*(?:\w+\.)*(\w+)")
|
||||
|
||||
_TEMPLATE_ASSIGN_RE = re.compile(
|
||||
r"""(?:export\s+)?(?:const|let|var)\s+(\w+)\s*=\s*`([^`]*)`"""
|
||||
)
|
||||
_STRING_ASSIGN_RE = re.compile(
|
||||
r"""(?:export\s+)?(?:const|let|var)\s+(\w+)\s*=\s*(['"])(.*?)\2"""
|
||||
)
|
||||
_TEMPLATE_REF_RE = re.compile(r"\$\{(\w+)\}")
|
||||
|
||||
|
||||
def _static_tail(route_path: str) -> str:
|
||||
"""Return the trailing contiguous static segments of a route path.
|
||||
|
||||
>>> _static_tail("/sessions/{id}/message")
|
||||
'/message'
|
||||
>>> _static_tail("/usage-summary")
|
||||
'/usage-summary'
|
||||
>>> _static_tail("/{id}")
|
||||
''
|
||||
"""
|
||||
parts = route_path.strip("/").split("/")
|
||||
tail: list[str] = []
|
||||
for part in reversed(parts):
|
||||
if part.startswith("{"):
|
||||
break
|
||||
tail.append(part)
|
||||
tail.reverse()
|
||||
return "/" + "/".join(tail) if tail else ""
|
||||
|
||||
|
||||
def _resolve_frontend_vars(files: list[tuple[str, str]]) -> dict[str, str]:
|
||||
"""Collect const/let/var string assignments across files and resolve refs.
|
||||
|
||||
Handles patterns like:
|
||||
const API_BASE = "/api";
|
||||
const WORKSPACE_API = `${API_BASE}/outputs/workspace`;
|
||||
"""
|
||||
raw: dict[str, str] = {}
|
||||
for _, text in files:
|
||||
for m in _STRING_ASSIGN_RE.finditer(text):
|
||||
raw.setdefault(m.group(1), m.group(3))
|
||||
for m in _TEMPLATE_ASSIGN_RE.finditer(text):
|
||||
raw.setdefault(m.group(1), m.group(2))
|
||||
resolved = dict(raw)
|
||||
for _ in range(5):
|
||||
changed = False
|
||||
for name, val in list(resolved.items()):
|
||||
new_val = _TEMPLATE_REF_RE.sub(
|
||||
lambda m: resolved.get(m.group(1), m.group(0)), val
|
||||
)
|
||||
if new_val != val:
|
||||
resolved[name] = new_val
|
||||
changed = True
|
||||
if not changed:
|
||||
break
|
||||
return resolved
|
||||
|
||||
|
||||
def _expand_template_refs(text: str, resolved: dict[str, str]) -> str:
|
||||
"""Replace ``${VAR}`` references in *text* with resolved values."""
|
||||
return _TEMPLATE_REF_RE.sub(
|
||||
lambda m: resolved.get(m.group(1), m.group(0)), text
|
||||
)
|
||||
|
||||
|
||||
def _find_func_name(lines: list[str], decorator_idx: int) -> str:
|
||||
for j in range(decorator_idx + 1, min(decorator_idx + 5, len(lines))):
|
||||
m = _FUNC_DEF_RE.match(lines[j])
|
||||
if m:
|
||||
return m.group(1)
|
||||
return ""
|
||||
|
||||
|
||||
def run_endpoint_check(
|
||||
root: Path,
|
||||
exceptions: dict[str, list[str]],
|
||||
ignore_routes: list[str] | None = None,
|
||||
ignores: dict[Path, set[str]] | None = None,
|
||||
) -> list[str]:
|
||||
"""Find backend API endpoints with no matching frontend or backend reference."""
|
||||
backend_dir = root / "backend"
|
||||
frontend_dir = root / "frontend" / "src"
|
||||
if not backend_dir.exists() or not frontend_dir.exists():
|
||||
return []
|
||||
|
||||
_ignore_routes = ignore_routes or []
|
||||
var_to_name: dict[str, str] = {}
|
||||
for py in backend_dir.rglob("*.py"):
|
||||
if ".venv" in py.parts:
|
||||
continue
|
||||
for m in _SUBAPP_RE.finditer(py.read_text(errors="ignore")):
|
||||
var_to_name[m.group(1)] = m.group(2)
|
||||
|
||||
routes: list[tuple[str, str, str, int, str]] = []
|
||||
|
||||
for py in backend_dir.rglob("*.py"):
|
||||
if ".venv" in py.parts:
|
||||
continue
|
||||
text = py.read_text(errors="ignore")
|
||||
lines = text.splitlines()
|
||||
rel = str(py.relative_to(root))
|
||||
|
||||
for i, line in enumerate(lines):
|
||||
m = _DECORATOR_RE.search(line)
|
||||
if m:
|
||||
var, path = m.group(1), m.group(2)
|
||||
name = var_to_name.get(var)
|
||||
if name:
|
||||
func = _find_func_name(lines, i)
|
||||
routes.append((name, path, rel, i + 1, func))
|
||||
|
||||
m2 = _ADD_ROUTE_RE.search(line)
|
||||
if m2:
|
||||
var, path = m2.group(1), m2.group(2)
|
||||
name = var_to_name.get(var)
|
||||
if name:
|
||||
fm = _ADD_ROUTE_FUNC_RE.search(line)
|
||||
func = fm.group(1) if fm else ""
|
||||
routes.append((name, path, rel, i + 1, func))
|
||||
|
||||
frontend_files: list[tuple[str, str]] = []
|
||||
for ext in ("*.ts", "*.tsx"):
|
||||
for f in frontend_dir.rglob(ext):
|
||||
frontend_files.append((str(f.relative_to(root)), f.read_text(errors="ignore")))
|
||||
|
||||
backend_files: list[tuple[str, str]] = []
|
||||
for py in backend_dir.rglob("*.py"):
|
||||
if ".venv" in py.parts:
|
||||
continue
|
||||
backend_files.append((str(py.relative_to(root)), py.read_text(errors="ignore")))
|
||||
|
||||
resolved_vars = _resolve_frontend_vars(frontend_files)
|
||||
|
||||
errors: list[str] = []
|
||||
for subapp_name, route_path, filepath, lineno, func_name in routes:
|
||||
if is_excepted(filepath, "endpoints", exceptions):
|
||||
continue
|
||||
if ignores and is_lintignored(root / filepath, root, "endpoints", ignores):
|
||||
continue
|
||||
|
||||
full_path = f"{subapp_name}{route_path}"
|
||||
|
||||
if any(fnmatch.fnmatch(full_path, p) for p in _ignore_routes):
|
||||
continue
|
||||
tail = _static_tail(route_path)
|
||||
|
||||
if not tail:
|
||||
continue
|
||||
|
||||
found = False
|
||||
for _fe_path, fe_text in frontend_files:
|
||||
expanded = _expand_template_refs(fe_text, resolved_vars)
|
||||
if full_path in expanded:
|
||||
found = True
|
||||
break
|
||||
if subapp_name in expanded and tail in expanded:
|
||||
found = True
|
||||
break
|
||||
|
||||
if not found:
|
||||
for be_path, be_text in backend_files:
|
||||
if be_path == filepath:
|
||||
continue
|
||||
if full_path in be_text:
|
||||
found = True
|
||||
break
|
||||
|
||||
if not found:
|
||||
label = func_name or route_path
|
||||
errors.append(
|
||||
f"{filepath}:{lineno}:1: warning: "
|
||||
f"[endpoints] orphaned endpoint '{label}' "
|
||||
f"(/api/{full_path}) — no frontend or backend reference found"
|
||||
)
|
||||
|
||||
return sorted(errors)
|
||||
@@ -0,0 +1,50 @@
|
||||
"""ESLint runner for the TypeScript frontend."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from . import is_lintignored
|
||||
|
||||
|
||||
def run_eslint(root: Path, ignores: dict[Path, set[str]] | None = None) -> list[str]:
|
||||
"""Run ESLint on the TypeScript frontend and return errors."""
|
||||
frontend_dir = root / "frontend"
|
||||
eslint_bin = frontend_dir / "node_modules" / ".bin" / "eslint"
|
||||
if not eslint_bin.exists():
|
||||
return []
|
||||
|
||||
cmd = [str(eslint_bin), "src/", "--format", "json", "--no-warn-ignored"]
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd, capture_output=True, text=True,
|
||||
cwd=str(frontend_dir), timeout=60,
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
return []
|
||||
|
||||
try:
|
||||
data = json.loads(result.stdout)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
return []
|
||||
|
||||
errors: list[str] = []
|
||||
for entry in data:
|
||||
try:
|
||||
filepath = Path(entry["filePath"])
|
||||
rel = str(filepath.relative_to(root))
|
||||
except (ValueError, KeyError):
|
||||
continue
|
||||
if ignores and is_lintignored(filepath, root, "eslint", ignores):
|
||||
continue
|
||||
for msg in entry.get("messages", []):
|
||||
sev = "error" if msg.get("severity", 0) >= 2 else "warning"
|
||||
text = msg.get("message", "").replace("\n", " ").strip()
|
||||
rule = msg.get("ruleId") or "unknown"
|
||||
errors.append(
|
||||
f"{rel}:{msg.get('line', 1)}:{msg.get('column', 1)}: "
|
||||
f"{sev}: [eslint] {text} ({rule})"
|
||||
)
|
||||
return errors
|
||||
@@ -0,0 +1,66 @@
|
||||
"""Knip unused-code runner for the TypeScript frontend."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from . import is_lintignored
|
||||
|
||||
KIND_LABELS = {
|
||||
"dependencies": "Unused dependency",
|
||||
"devDependencies": "Unused devDependency",
|
||||
"exports": "Unused export",
|
||||
"types": "Unused exported type",
|
||||
"unlisted": "Unlisted dependency",
|
||||
"binaries": "Unused binary",
|
||||
"files": "Unused file",
|
||||
"duplicates": "Duplicate export",
|
||||
}
|
||||
|
||||
|
||||
def run_knip(root: Path, ignores: dict[Path, set[str]] | None = None) -> list[str]:
|
||||
"""Run Knip on the TypeScript frontend and return errors."""
|
||||
frontend_dir = root / "frontend"
|
||||
knip_bin = frontend_dir / "node_modules" / ".bin" / "knip"
|
||||
if not knip_bin.exists():
|
||||
return []
|
||||
|
||||
cmd = [str(knip_bin), "--reporter", "json"]
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd, capture_output=True, text=True,
|
||||
cwd=str(frontend_dir), timeout=60,
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
return []
|
||||
|
||||
try:
|
||||
data = json.loads(result.stdout)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
return []
|
||||
|
||||
errors: list[str] = []
|
||||
for entry in data.get("issues", []):
|
||||
filepath = entry.get("file", "")
|
||||
rel = f"frontend/{filepath}"
|
||||
abs_path = root / rel
|
||||
if ignores and is_lintignored(abs_path, root, "knip", ignores):
|
||||
continue
|
||||
for kind, label in KIND_LABELS.items():
|
||||
for item in entry.get(kind, []):
|
||||
if isinstance(item, dict):
|
||||
name = item.get("name", "")
|
||||
line = item.get("line", 1)
|
||||
col = item.get("col", 1)
|
||||
elif isinstance(item, str):
|
||||
name = item
|
||||
line, col = 1, 1
|
||||
else:
|
||||
continue
|
||||
errors.append(
|
||||
f"{rel}:{line}:{col}: error: "
|
||||
f"[knip] {label} '{name}'"
|
||||
)
|
||||
return errors
|
||||
@@ -0,0 +1,104 @@
|
||||
"""Structural checks: file length, folder size, and nested imports."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
from pathlib import Path
|
||||
|
||||
from . import _matches_any
|
||||
|
||||
ANCHOR_FILES = ("__init__.py", "index.ts", "index.tsx", "index.js")
|
||||
|
||||
|
||||
def _find_anchor_file(dirpath: Path, root: Path) -> str:
|
||||
"""Find a real file inside the folder to attach the diagnostic to.
|
||||
|
||||
Prefers common entry-point files (__init__.py, index.ts, etc.) so the
|
||||
error shows up inline when you open that file. Falls back to the first
|
||||
file alphabetically, then the directory path itself.
|
||||
"""
|
||||
for name in ANCHOR_FILES:
|
||||
candidate = dirpath / name
|
||||
if candidate.exists():
|
||||
return str(candidate.relative_to(root))
|
||||
try:
|
||||
first = sorted(
|
||||
f for f in dirpath.iterdir()
|
||||
if f.is_file() and not f.name.startswith(".")
|
||||
)
|
||||
if first:
|
||||
return str(first[0].relative_to(root))
|
||||
except OSError:
|
||||
pass
|
||||
return str(dirpath.relative_to(root))
|
||||
|
||||
|
||||
def check_file_lines(
|
||||
filepath: Path, root: Path, max_lines: int,
|
||||
) -> tuple[str, int] | None:
|
||||
try:
|
||||
count = len(filepath.read_text(errors="ignore").splitlines())
|
||||
except OSError:
|
||||
return None
|
||||
if count >= max_lines:
|
||||
rel = filepath.relative_to(root)
|
||||
msg = (
|
||||
f"{rel}:1:1: error: "
|
||||
f"[max-file-lines] File has {count} lines (limit {max_lines})"
|
||||
)
|
||||
return (msg, count)
|
||||
return None
|
||||
|
||||
|
||||
def check_folder_items(
|
||||
dirpath: Path, root: Path, max_items: int, excludes: list[str],
|
||||
) -> tuple[str, int] | None:
|
||||
try:
|
||||
items = [
|
||||
i for i in dirpath.iterdir()
|
||||
if not i.name.startswith(".") and not _matches_any(i.name, excludes)
|
||||
]
|
||||
except OSError:
|
||||
return None
|
||||
count = len(items)
|
||||
if count >= max_items:
|
||||
anchor = _find_anchor_file(dirpath, root)
|
||||
rel = dirpath.relative_to(root)
|
||||
msg = (
|
||||
f"{anchor}:1:1: error: "
|
||||
f"[max-folder-items] Folder '{rel}' has {count} items (limit {max_items})"
|
||||
)
|
||||
return (msg, count)
|
||||
return None
|
||||
|
||||
|
||||
def check_nested_imports(filepath: Path, root: Path) -> list[str]:
|
||||
"""Detect import statements inside function or method bodies."""
|
||||
if filepath.suffix != ".py":
|
||||
return []
|
||||
try:
|
||||
source = filepath.read_text(errors="ignore")
|
||||
tree = ast.parse(source, filename=str(filepath))
|
||||
except (OSError, SyntaxError):
|
||||
return []
|
||||
|
||||
errors: list[str] = []
|
||||
rel = str(filepath.relative_to(root))
|
||||
|
||||
def _visit(node: ast.AST, in_function: bool) -> None:
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
in_function = True
|
||||
if in_function and isinstance(node, (ast.Import, ast.ImportFrom)):
|
||||
if isinstance(node, ast.ImportFrom):
|
||||
name = node.module or ""
|
||||
else:
|
||||
name = ", ".join(a.name for a in node.names)
|
||||
errors.append(
|
||||
f"{rel}:{node.lineno}:1: error: "
|
||||
f"[no-nested-imports] Nested import '{name}'"
|
||||
)
|
||||
for child in ast.iter_child_nodes(node):
|
||||
_visit(child, in_function)
|
||||
|
||||
_visit(tree, False)
|
||||
return errors
|
||||
@@ -0,0 +1,95 @@
|
||||
"""Vulture dead-code detection runner.
|
||||
|
||||
Class-body findings (fields, methods inside a class) are filtered out here
|
||||
and handled separately by checks/classes.py which understands Pydantic.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
from functools import lru_cache
|
||||
from pathlib import Path
|
||||
|
||||
from . import is_excepted, is_lintignored
|
||||
|
||||
CONFIG_DIR = Path(__file__).resolve().parent.parent / "config"
|
||||
|
||||
|
||||
@lru_cache(maxsize=64)
|
||||
def _class_line_ranges(filepath: str) -> list[tuple[int, int]]:
|
||||
"""Return (start, end) line ranges for all class bodies in *filepath*."""
|
||||
try:
|
||||
tree = ast.parse(Path(filepath).read_text())
|
||||
except (OSError, SyntaxError):
|
||||
return []
|
||||
ranges: list[tuple[int, int]] = []
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.ClassDef):
|
||||
end = max(getattr(n, "lineno", node.lineno) for n in ast.walk(node))
|
||||
ranges.append((node.lineno, end))
|
||||
return ranges
|
||||
|
||||
|
||||
def _is_inside_class(filepath: str, lineno: int) -> bool:
|
||||
"""True when *lineno* is strictly inside a class body.
|
||||
|
||||
The class declaration line itself (``class Foo:``) is *not* considered
|
||||
inside, so vulture's "unused class" findings still pass through.
|
||||
"""
|
||||
return any(start < lineno <= end for start, end in _class_line_ranges(filepath))
|
||||
|
||||
|
||||
def run_vulture(
|
||||
root: Path, min_confidence: int, error_threshold: int,
|
||||
exceptions: dict[str, list[str]],
|
||||
ignores: dict[Path, set[str]] | None = None,
|
||||
) -> list[str]:
|
||||
"""Run vulture on the Python backend and return errors."""
|
||||
vulture_bin = root / "backend" / ".venv" / "bin" / "vulture"
|
||||
if not vulture_bin.exists():
|
||||
found = shutil.which("vulture")
|
||||
if not found:
|
||||
return []
|
||||
vulture_bin = Path(found)
|
||||
|
||||
whitelist = CONFIG_DIR / "vulture_whitelist.py"
|
||||
targets = ["backend"]
|
||||
if (root / "debug.py").exists():
|
||||
targets.append("debug.py")
|
||||
cmd = [str(vulture_bin), *targets]
|
||||
if whitelist.exists():
|
||||
cmd.append(str(whitelist))
|
||||
cmd.extend([
|
||||
"--min-confidence", str(min_confidence),
|
||||
"--exclude", ".venv,__pycache__,data,uv-bin",
|
||||
"--ignore-decorators", "@*.router.*,@*.websocket,@app.*,@pytest.fixture,@pytest.fixture*",
|
||||
"--ignore-names", "cls",
|
||||
])
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd, capture_output=True, text=True, cwd=str(root), timeout=30,
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
return []
|
||||
|
||||
errors: list[str] = []
|
||||
for line in result.stdout.strip().splitlines():
|
||||
m = re.match(r"^(.+):(\d+): (.+)$", line)
|
||||
if not m:
|
||||
continue
|
||||
filepath, lineno, message = m.groups()
|
||||
if is_excepted(filepath, "vulture", exceptions):
|
||||
continue
|
||||
if ignores and is_lintignored(root / filepath, root, "vulture", ignores):
|
||||
continue
|
||||
if _is_inside_class(str(root / filepath), int(lineno)):
|
||||
continue
|
||||
conf = re.search(r"\((\d+)% confidence\)", message)
|
||||
confidence = int(conf.group(1)) if conf else 0
|
||||
severity = "error" if confidence >= error_threshold else "warning"
|
||||
errors.append(f"{filepath}:{lineno}:1: {severity}: [vulture] {message}")
|
||||
return errors
|
||||
@@ -0,0 +1,139 @@
|
||||
{
|
||||
"enabled": {
|
||||
"max-file-lines": true,
|
||||
"max-folder-items": true,
|
||||
"no-nested-imports": false,
|
||||
"vulture": true,
|
||||
"eslint": false,
|
||||
"knip": false,
|
||||
"endpoints": false,
|
||||
"classes": false
|
||||
},
|
||||
"_notes": {
|
||||
"no-nested-imports": "Off on purpose: this codebase uses function-level/lazy imports to break import cycles (400+ sites). Flagging them all is wrong for us.",
|
||||
"eslint-knip": "Node tooling deferred to a later pass.",
|
||||
"classes": "Placeholder check, not wired up. endpoints: orphaned-endpoint triage deferred."
|
||||
},
|
||||
"rules": {
|
||||
"max-file-lines": 300,
|
||||
"max-folder-items": 7,
|
||||
"vulture-min-confidence": 1,
|
||||
"vulture-error-threshold": 1,
|
||||
"no-nested-imports": true,
|
||||
"endpoint-ignore-routes": ["*/callback", "*/callback/*"]
|
||||
},
|
||||
"include_extensions": [".py", ".ts", ".tsx", ".js", ".jsx"],
|
||||
"exclude": [
|
||||
"node_modules",
|
||||
".venv",
|
||||
"dist",
|
||||
"build",
|
||||
"__pycache__",
|
||||
".git",
|
||||
".cursor",
|
||||
".vscode",
|
||||
".claude",
|
||||
".pytest_cache",
|
||||
".mypy_cache",
|
||||
".ruff_cache",
|
||||
".account-factory",
|
||||
".local-stash",
|
||||
".DS_Store",
|
||||
"python-env",
|
||||
"build-staging",
|
||||
"webapp_template_cache",
|
||||
"openswarm-cloud",
|
||||
"uv-bin",
|
||||
"data",
|
||||
"public",
|
||||
"openswarm_debug.egg-info"
|
||||
],
|
||||
"exceptions": {
|
||||
"max-file-lines": [
|
||||
"backend/apps/agents/agent_manager.py",
|
||||
"backend/apps/agents/agents.py",
|
||||
"backend/apps/agents/anthropic_proxy.py",
|
||||
"backend/apps/agents/anthropic_to_openai.py",
|
||||
"backend/apps/agents/browser_agent.py",
|
||||
"backend/apps/agents/browser_agent_mcp_server.py",
|
||||
"backend/apps/agents/browser_schema.py",
|
||||
"backend/apps/agents/prompt_context.py",
|
||||
"backend/apps/agents/providers/pricing.py",
|
||||
"backend/apps/agents/providers/registry.py",
|
||||
"backend/apps/dashboards/dashboards.py",
|
||||
"backend/apps/discord_mcp_shim/server.py",
|
||||
"backend/apps/mcp_registry/mcp_registry.py",
|
||||
"backend/apps/nine_router/oauth.py",
|
||||
"backend/apps/nine_router/process.py",
|
||||
"backend/apps/nine_router/sync_custom.py",
|
||||
"backend/apps/outputs/outputs.py",
|
||||
"backend/apps/outputs/runtime.py",
|
||||
"backend/apps/outputs/view_builder_templates.py",
|
||||
"backend/apps/outputs/webapp_template/frontend/src/shared/styles/ThemeContext.tsx",
|
||||
"backend/apps/service/client.py",
|
||||
"backend/apps/service/service.py",
|
||||
"backend/apps/settings/settings.py",
|
||||
"backend/apps/skills/skills.py",
|
||||
"backend/apps/subscription/router.py",
|
||||
"backend/apps/tools_lib/tools_lib.py",
|
||||
"backend/apps/web/web.py",
|
||||
"backend/main.py",
|
||||
"backend/tests/test_disconnect_resilience.py",
|
||||
"backend/tests/test_outputs_runtime_cleanup.py",
|
||||
"backend/tests/test_service.py",
|
||||
"backend/tests/test_v2_invariants.py",
|
||||
"backend/tests/test_v2_label_logic.py",
|
||||
"electron/affiliateTracking.test.js",
|
||||
"electron/main.js",
|
||||
"frontend/src/app/Main.tsx",
|
||||
"frontend/src/app/components/CommandPicker.tsx",
|
||||
"frontend/src/app/components/DirectoryBrowser.tsx",
|
||||
"frontend/src/app/components/DynamicIsland.tsx",
|
||||
"frontend/src/app/components/GlobalSearchPalette.tsx",
|
||||
"frontend/src/app/components/Layout/AppShell.tsx",
|
||||
"frontend/src/app/components/Onboarding/OnboardingPanel.tsx",
|
||||
"frontend/src/app/components/Onboarding/ac/AgenticCursor.tsx",
|
||||
"frontend/src/app/components/Onboarding/ac/acRuntime.ts",
|
||||
"frontend/src/app/components/PlanPicker.tsx",
|
||||
"frontend/src/app/components/RichPromptEditor.tsx",
|
||||
"frontend/src/app/components/SignInGate.tsx",
|
||||
"frontend/src/app/components/useDomElementSelector.ts",
|
||||
"frontend/src/app/pages/AgentChat/AgentChat.tsx",
|
||||
"frontend/src/app/pages/AgentChat/ApprovalBar.tsx",
|
||||
"frontend/src/app/pages/AgentChat/BrowserAgentInlineFeed.tsx",
|
||||
"frontend/src/app/pages/AgentChat/ChatInput.tsx",
|
||||
"frontend/src/app/pages/AgentChat/MessageBubble.tsx",
|
||||
"frontend/src/app/pages/AgentChat/ToolCallBubble.tsx",
|
||||
"frontend/src/app/pages/AgentChat/toolLabels.ts",
|
||||
"frontend/src/app/pages/Analytics/PixelChart.tsx",
|
||||
"frontend/src/app/pages/Commands/Commands.tsx",
|
||||
"frontend/src/app/pages/Dashboard/AgentCard.tsx",
|
||||
"frontend/src/app/pages/Dashboard/BrowserAgentOverlay.tsx",
|
||||
"frontend/src/app/pages/Dashboard/BrowserCard.tsx",
|
||||
"frontend/src/app/pages/Dashboard/Dashboard.tsx",
|
||||
"frontend/src/app/pages/Dashboard/DashboardHeader.tsx",
|
||||
"frontend/src/app/pages/Dashboard/DashboardToolbar.tsx",
|
||||
"frontend/src/app/pages/Dashboard/DashboardViewCard.tsx",
|
||||
"frontend/src/app/pages/Dashboard/NoteCard.tsx",
|
||||
"frontend/src/app/pages/Dashboard/useCanvasControls.ts",
|
||||
"frontend/src/app/pages/DashboardSelection/DashboardSelection.tsx",
|
||||
"frontend/src/app/pages/Modes/Modes.tsx",
|
||||
"frontend/src/app/pages/Settings/Settings.tsx",
|
||||
"frontend/src/app/pages/Skills/SkillBuilderChat.tsx",
|
||||
"frontend/src/app/pages/Skills/Skills.tsx",
|
||||
"frontend/src/app/pages/Tools/Tools.tsx",
|
||||
"frontend/src/app/pages/Views/ViewEditor.tsx",
|
||||
"frontend/src/app/pages/Views/ViewPreview.tsx",
|
||||
"frontend/src/app/pages/Views/useIframeElementSelector.ts",
|
||||
"frontend/src/shared/browserCommandHandler.ts",
|
||||
"frontend/src/shared/state/agentsSlice.ts",
|
||||
"frontend/src/shared/state/dashboardLayoutSlice.ts",
|
||||
"frontend/src/shared/ws/WebSocketManager.ts"
|
||||
],
|
||||
"max-folder-items": [],
|
||||
"no-nested-imports": [],
|
||||
"vulture": ["backend/tests/*"],
|
||||
"endpoints": [],
|
||||
"classes": []
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
{
|
||||
"include": ["../../backend"],
|
||||
"exclude": [
|
||||
"../../backend/.venv",
|
||||
"../../backend/uv-bin",
|
||||
"../../backend/data",
|
||||
"../../backend/tests"
|
||||
],
|
||||
"typeCheckingMode": "strict",
|
||||
"pythonVersion": "3.11",
|
||||
"venvPath": "../../backend",
|
||||
"venv": ".venv",
|
||||
"reportMissingTypeStubs": false
|
||||
}
|
||||
+170
@@ -0,0 +1,170 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Unified linter: orchestrates structural checks, dead-code detection, and lint tools."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from checks import is_excluded, is_excepted, is_lintignored, collect_lintignores
|
||||
from checks.structural import check_file_lines, check_folder_items, check_nested_imports
|
||||
from checks.vulture import run_vulture
|
||||
from checks.eslint import run_eslint
|
||||
from checks.knip import run_knip
|
||||
from checks.endpoints import run_endpoint_check
|
||||
from checks.classes import run_class_check
|
||||
from watchfiles import watch, DefaultFilter
|
||||
|
||||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||||
CONFIG_FILE = SCRIPT_DIR / "config" / "config.json"
|
||||
|
||||
|
||||
def load_config() -> dict[str, Any]:
|
||||
with open(CONFIG_FILE) as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
def run_checks(root: Path) -> tuple[list[str], list[str], list[str], list[str], list[str], list[str]]:
|
||||
config = load_config()
|
||||
enabled: dict[str, bool] = config.get("enabled", {})
|
||||
rules: dict[str, int] = config["rules"]
|
||||
excludes: list[str] = config["exclude"]
|
||||
exceptions: dict[str, list[str]] = config["exceptions"]
|
||||
extensions: list[str] = config["include_extensions"]
|
||||
ignores = collect_lintignores(root, excludes)
|
||||
|
||||
max_lines: int = rules["max-file-lines"]
|
||||
max_items: int = rules["max-folder-items"]
|
||||
check_imports: bool = rules.get("no-nested-imports", False)
|
||||
structural_errors: list[str] = []
|
||||
|
||||
file_lines_on = enabled.get("max-file-lines", True)
|
||||
folder_items_on = enabled.get("max-folder-items", True)
|
||||
nested_imports_on = enabled.get("no-nested-imports", True)
|
||||
|
||||
for dirpath_str, dirnames, filenames in os.walk(root):
|
||||
dp = Path(dirpath_str)
|
||||
|
||||
if is_excluded(dp, root, excludes):
|
||||
dirnames.clear()
|
||||
continue
|
||||
|
||||
rel_dir = str(dp.relative_to(root))
|
||||
if (
|
||||
folder_items_on
|
||||
and rel_dir != "."
|
||||
and not is_excepted(rel_dir, "max-folder-items", exceptions)
|
||||
and not is_lintignored(dp, root, "max-folder-items", ignores)
|
||||
):
|
||||
result = check_folder_items(dp, root, max_items, excludes)
|
||||
if result:
|
||||
structural_errors.append(result[0])
|
||||
|
||||
for fname in filenames:
|
||||
fp = dp / fname
|
||||
if fp.suffix not in extensions:
|
||||
continue
|
||||
if is_excluded(fp, root, excludes):
|
||||
continue
|
||||
rel_file = str(fp.relative_to(root))
|
||||
if (
|
||||
file_lines_on
|
||||
and not is_excepted(rel_file, "max-file-lines", exceptions)
|
||||
and not is_lintignored(fp, root, "max-file-lines", ignores)
|
||||
):
|
||||
result = check_file_lines(fp, root, max_lines)
|
||||
if result:
|
||||
structural_errors.append(result[0])
|
||||
if (
|
||||
nested_imports_on
|
||||
and check_imports
|
||||
and not is_excepted(rel_file, "no-nested-imports", exceptions)
|
||||
and not is_lintignored(fp, root, "no-nested-imports", ignores)
|
||||
):
|
||||
structural_errors.extend(check_nested_imports(fp, root))
|
||||
|
||||
vulture_errors: list[str] = []
|
||||
if enabled.get("vulture", True):
|
||||
vulture_confidence = rules.get("vulture-min-confidence")
|
||||
if vulture_confidence is not None:
|
||||
vulture_error_threshold = rules.get("vulture-error-threshold", 100)
|
||||
vulture_errors = run_vulture(
|
||||
root, vulture_confidence, vulture_error_threshold, exceptions, ignores,
|
||||
)
|
||||
|
||||
eslint_errors = run_eslint(root, ignores) if enabled.get("eslint", True) else []
|
||||
knip_errors = run_knip(root, ignores) if enabled.get("knip", True) else []
|
||||
endpoint_ignore_routes: list[str] = rules.get("endpoint-ignore-routes", [])
|
||||
endpoint_errors = run_endpoint_check(root, exceptions, endpoint_ignore_routes, ignores) if enabled.get("endpoints", True) else []
|
||||
class_errors = run_class_check(root, exceptions, excludes, ignores) if enabled.get("classes", True) else []
|
||||
|
||||
return sorted(structural_errors), sorted(vulture_errors), sorted(eslint_errors), sorted(knip_errors), sorted(endpoint_errors), sorted(class_errors)
|
||||
|
||||
|
||||
def _print_section(name: str, errors: list[str]) -> None:
|
||||
print(f"{name}: checking...", flush=True)
|
||||
for e in errors:
|
||||
print(e, flush=True)
|
||||
print(f"{name}: done. {len(errors)} error(s) found.", flush=True)
|
||||
|
||||
|
||||
def print_results(
|
||||
structural_errors: list[str], vulture_errors: list[str],
|
||||
eslint_errors: list[str], knip_errors: list[str],
|
||||
endpoint_errors: list[str], class_errors: list[str],
|
||||
) -> None:
|
||||
_print_section("structural", structural_errors)
|
||||
_print_section("vulture", vulture_errors)
|
||||
_print_section("eslint", eslint_errors)
|
||||
_print_section("knip", knip_errors)
|
||||
_print_section("endpoints", endpoint_errors)
|
||||
_print_section("classes", class_errors)
|
||||
|
||||
|
||||
def watch_loop(root: Path) -> None:
|
||||
|
||||
config_dir = SCRIPT_DIR / "config"
|
||||
|
||||
print_results(*run_checks(root))
|
||||
|
||||
class SourceFilter(DefaultFilter):
|
||||
allowed_extensions = (".py", ".ts", ".tsx", ".js", ".jsx")
|
||||
|
||||
def __call__(self, change: Any, path: str) -> bool:
|
||||
if not super().__call__(change, path):
|
||||
return False
|
||||
if Path(path).suffix in self.allowed_extensions:
|
||||
return True
|
||||
p = Path(path)
|
||||
if p.suffix == ".json" and (p.parent == SCRIPT_DIR or p.parent == config_dir):
|
||||
return True
|
||||
if p.name.startswith(".lintignore"):
|
||||
return True
|
||||
return Path(path).is_dir()
|
||||
|
||||
for _changes in watch(root, watch_filter=SourceFilter()):
|
||||
print_results(*run_checks(root))
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="Unified linter")
|
||||
parser.add_argument("--watch", action="store_true", help="Watch for changes")
|
||||
parser.add_argument("--root", type=str, default=".", help="Root directory")
|
||||
args = parser.parse_args()
|
||||
|
||||
root = Path(args.root).resolve()
|
||||
|
||||
if args.watch:
|
||||
watch_loop(root)
|
||||
else:
|
||||
results = run_checks(root)
|
||||
print_results(*results)
|
||||
sys.exit(1 if any(results) else 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Executable
+64
@@ -0,0 +1,64 @@
|
||||
#!/bin/bash
|
||||
# Print lint violations to stdout with colored formatting.
|
||||
# Usage: bash linter/print_errors.sh [ROOT_DIR]
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
ROOT_DIR="${1:-$(dirname "$SCRIPT_DIR")}"
|
||||
|
||||
YELLOW='\033[33m'
|
||||
CYAN='\033[36m'
|
||||
BOLD='\033[1m'
|
||||
RESET='\033[0m'
|
||||
|
||||
LINT_OUTPUT=$(python3 "$SCRIPT_DIR/lint.py" --root "$ROOT_DIR" 2>&1)
|
||||
LINT_EXIT=$?
|
||||
|
||||
if [ $LINT_EXIT -ne 0 ]; then
|
||||
STRUCT_LINES=$(echo "$LINT_OUTPUT" | grep -v "^structural:" | grep -v "^vulture:" | grep -v "^eslint:" | grep -v "^knip:" | grep -v '\[vulture\]' | grep -v '\[eslint\]' | grep -v '\[knip\]')
|
||||
VULTURE_LINES=$(echo "$LINT_OUTPUT" | grep '\[vulture\]')
|
||||
ESLINT_LINES=$(echo "$LINT_OUTPUT" | grep '\[eslint\]')
|
||||
KNIP_LINES=$(echo "$LINT_OUTPUT" | grep '\[knip\]')
|
||||
|
||||
STRUCT_COUNT=$(echo "$STRUCT_LINES" | grep -cE ':\s+(error|warning):\s+')
|
||||
VULTURE_COUNT=$(echo "$VULTURE_LINES" | grep -cE ':\s+(error|warning):\s+')
|
||||
ESLINT_COUNT=$(echo "$ESLINT_LINES" | grep -cE ':\s+(error|warning):\s+')
|
||||
KNIP_COUNT=$(echo "$KNIP_LINES" | grep -cE ':\s+(error|warning):\s+')
|
||||
|
||||
if [ "$STRUCT_COUNT" -gt 0 ]; then
|
||||
echo ""
|
||||
echo -e "${YELLOW}${BOLD}[structural] Violations found:${RESET}"
|
||||
echo "$STRUCT_LINES" | while IFS= read -r line; do
|
||||
[ -n "$line" ] && echo -e "${YELLOW} $line${RESET}"
|
||||
done
|
||||
echo -e "${YELLOW}${BOLD} ${STRUCT_COUNT} violation(s) — fix or add exceptions in linter/config/config.json${RESET}"
|
||||
fi
|
||||
|
||||
if [ "$VULTURE_COUNT" -gt 0 ]; then
|
||||
echo ""
|
||||
echo -e "${CYAN}${BOLD}[vulture] Dead code found:${RESET}"
|
||||
echo "$VULTURE_LINES" | while IFS= read -r line; do
|
||||
[ -n "$line" ] && echo -e "${CYAN} $line${RESET}"
|
||||
done
|
||||
echo -e "${CYAN}${BOLD} ${VULTURE_COUNT} finding(s) — fix or add to linter/config/vulture_whitelist.py${RESET}"
|
||||
fi
|
||||
|
||||
if [ "$ESLINT_COUNT" -gt 0 ]; then
|
||||
echo ""
|
||||
echo -e "${YELLOW}${BOLD}[eslint] Lint errors found:${RESET}"
|
||||
echo "$ESLINT_LINES" | while IFS= read -r line; do
|
||||
[ -n "$line" ] && echo -e "${YELLOW} $line${RESET}"
|
||||
done
|
||||
echo -e "${YELLOW}${BOLD} ${ESLINT_COUNT} error(s) — fix or disable rules in frontend/eslint.config.mjs${RESET}"
|
||||
fi
|
||||
|
||||
if [ "$KNIP_COUNT" -gt 0 ]; then
|
||||
echo ""
|
||||
echo -e "${CYAN}${BOLD}[knip] Unused code/dependencies found:${RESET}"
|
||||
echo "$KNIP_LINES" | while IFS= read -r line; do
|
||||
[ -n "$line" ] && echo -e "${CYAN} $line${RESET}"
|
||||
done
|
||||
echo -e "${CYAN}${BOLD} ${KNIP_COUNT} finding(s) — remove unused code or update frontend/knip.json${RESET}"
|
||||
fi
|
||||
|
||||
echo ""
|
||||
fi
|
||||
Reference in New Issue
Block a user