mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-09-28 12:34:50 +02:00
[Haik]: ckpt, nine router now runs on different ports for dev vs prod. Also made a custom linter to enforce my standards
This commit is contained in:
@@ -0,0 +1,187 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Structural linter: enforces file line limits and folder item limits."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import fnmatch
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||||
CONFIG_FILE = SCRIPT_DIR / "structlint.json"
|
||||
|
||||
|
||||
def load_config() -> dict[str, Any]:
|
||||
with open(CONFIG_FILE) as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
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 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: File has {count} lines "
|
||||
f"(limit {max_lines}) [max-file-lines]"
|
||||
)
|
||||
return (msg, count)
|
||||
return None
|
||||
|
||||
|
||||
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_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: Folder '{rel}' has {count} items "
|
||||
f"(limit {max_items}) [max-folder-items]"
|
||||
)
|
||||
return (msg, count)
|
||||
return None
|
||||
|
||||
|
||||
def run_checks(root: Path) -> list[str]:
|
||||
config = load_config()
|
||||
rules: dict[str, int] = config["rules"]
|
||||
excludes: list[str] = config["exclude"]
|
||||
exceptions: dict[str, list[str]] = config["exceptions"]
|
||||
extensions: list[str] = config["include_extensions"]
|
||||
|
||||
max_lines: int = rules["max-file-lines"]
|
||||
max_items: int = rules["max-folder-items"]
|
||||
errors: list[str] = []
|
||||
|
||||
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 rel_dir != "." and not is_excepted(rel_dir, "max-folder-items", exceptions):
|
||||
result = check_folder_items(dp, root, max_items, excludes)
|
||||
if result:
|
||||
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 not is_excepted(rel_file, "max-file-lines", exceptions):
|
||||
result = check_file_lines(fp, root, max_lines)
|
||||
if result:
|
||||
errors.append(result[0])
|
||||
|
||||
return sorted(errors)
|
||||
|
||||
|
||||
def print_results(errors: list[str]) -> None:
|
||||
print("structlint: checking...", flush=True)
|
||||
for err in errors:
|
||||
print(err, flush=True)
|
||||
count = len(errors)
|
||||
print(f"structlint: done. {count} error(s) found.", flush=True)
|
||||
|
||||
|
||||
def watch_loop(root: Path) -> None:
|
||||
from watchfiles import watch, DefaultFilter
|
||||
|
||||
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
|
||||
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="Structural 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:
|
||||
errors = run_checks(root)
|
||||
print_results(errors)
|
||||
sys.exit(1 if errors else 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user