generated from coulomb/repo-seed
Milestone 2’s core deterministic scanner path
This commit is contained in:
1
src/repo_registry/repo_scanning/__init__.py
Normal file
1
src/repo_registry/repo_scanning/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Deterministic repository scanning."""
|
||||
271
src/repo_registry/repo_scanning/scanner.py
Normal file
271
src/repo_registry/repo_scanning/scanner.py
Normal file
@@ -0,0 +1,271 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
IGNORED_DIRS = {
|
||||
".git",
|
||||
".hg",
|
||||
".mypy_cache",
|
||||
".pytest_cache",
|
||||
".ruff_cache",
|
||||
".tox",
|
||||
".venv",
|
||||
"__pycache__",
|
||||
"build",
|
||||
"dist",
|
||||
"node_modules",
|
||||
"target",
|
||||
"vendor",
|
||||
}
|
||||
|
||||
LANGUAGE_BY_EXTENSION = {
|
||||
".go": "Go",
|
||||
".java": "Java",
|
||||
".js": "JavaScript",
|
||||
".jsx": "JavaScript",
|
||||
".kt": "Kotlin",
|
||||
".php": "PHP",
|
||||
".py": "Python",
|
||||
".rb": "Ruby",
|
||||
".rs": "Rust",
|
||||
".ts": "TypeScript",
|
||||
".tsx": "TypeScript",
|
||||
}
|
||||
|
||||
MANIFEST_FRAMEWORK_HINTS = {
|
||||
"pyproject.toml": {
|
||||
"fastapi": "FastAPI",
|
||||
"django": "Django",
|
||||
"flask": "Flask",
|
||||
"typer": "Typer",
|
||||
"click": "Click",
|
||||
"pytest": "pytest",
|
||||
},
|
||||
"requirements.txt": {
|
||||
"fastapi": "FastAPI",
|
||||
"django": "Django",
|
||||
"flask": "Flask",
|
||||
"typer": "Typer",
|
||||
"click": "Click",
|
||||
"pytest": "pytest",
|
||||
},
|
||||
"package.json": {
|
||||
"next": "Next.js",
|
||||
"react": "React",
|
||||
"express": "Express",
|
||||
"vite": "Vite",
|
||||
"jest": "Jest",
|
||||
"vitest": "Vitest",
|
||||
},
|
||||
"Cargo.toml": {
|
||||
"axum": "Axum",
|
||||
"actix-web": "Actix Web",
|
||||
"clap": "Clap",
|
||||
"tokio": "Tokio",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FactCandidate:
|
||||
kind: str
|
||||
name: str
|
||||
path: str = ""
|
||||
value: str = ""
|
||||
metadata: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ScanResult:
|
||||
source_path: str
|
||||
commit_hash: str
|
||||
branch: str
|
||||
file_count: int
|
||||
facts: list[FactCandidate]
|
||||
|
||||
|
||||
class DeterministicScanner:
|
||||
version = "deterministic-v0.1"
|
||||
|
||||
def scan(self, source_path: str | Path) -> ScanResult:
|
||||
root = Path(source_path).expanduser().resolve()
|
||||
if not root.exists() or not root.is_dir():
|
||||
raise ValueError(f"source path does not exist or is not a directory: {root}")
|
||||
|
||||
files = list(self._iter_files(root))
|
||||
facts: list[FactCandidate] = []
|
||||
facts.extend(self._language_facts(files, root))
|
||||
facts.extend(self._classified_file_facts(files, root))
|
||||
facts.extend(self._framework_facts(files, root))
|
||||
facts.extend(self._interface_facts(files, root))
|
||||
|
||||
return ScanResult(
|
||||
source_path=str(root),
|
||||
commit_hash=self._git_value(root, "rev-parse", "HEAD") or "working-tree",
|
||||
branch=self._git_value(root, "branch", "--show-current") or "unknown",
|
||||
file_count=len(files),
|
||||
facts=sorted(facts, key=lambda fact: (fact.kind, fact.path, fact.name)),
|
||||
)
|
||||
|
||||
def _iter_files(self, root: Path) -> list[Path]:
|
||||
files: list[Path] = []
|
||||
for path in root.rglob("*"):
|
||||
if not path.is_file():
|
||||
continue
|
||||
relative_parts = path.relative_to(root).parts
|
||||
if any(part in IGNORED_DIRS for part in relative_parts):
|
||||
continue
|
||||
files.append(path)
|
||||
return files
|
||||
|
||||
def _language_facts(self, files: list[Path], root: Path) -> list[FactCandidate]:
|
||||
counts: dict[str, int] = {}
|
||||
for path in files:
|
||||
language = LANGUAGE_BY_EXTENSION.get(path.suffix)
|
||||
if language is None:
|
||||
continue
|
||||
counts[language] = counts.get(language, 0) + 1
|
||||
|
||||
return [
|
||||
FactCandidate(
|
||||
kind="language",
|
||||
name=language,
|
||||
value=str(count),
|
||||
metadata={"file_count": count},
|
||||
)
|
||||
for language, count in counts.items()
|
||||
]
|
||||
|
||||
def _classified_file_facts(
|
||||
self, files: list[Path], root: Path
|
||||
) -> list[FactCandidate]:
|
||||
facts: list[FactCandidate] = []
|
||||
for path in files:
|
||||
relative = path.relative_to(root).as_posix()
|
||||
lower = relative.lower()
|
||||
name = path.name.lower()
|
||||
|
||||
if name.startswith("readme"):
|
||||
facts.append(FactCandidate("documentation", "README", relative))
|
||||
elif lower.startswith("docs/") or lower.startswith("doc/"):
|
||||
facts.append(FactCandidate("documentation", path.name, relative))
|
||||
|
||||
if lower.startswith("examples/") or lower.startswith("example/"):
|
||||
facts.append(FactCandidate("example", path.name, relative))
|
||||
|
||||
if (
|
||||
lower.startswith("tests/")
|
||||
or lower.startswith("test/")
|
||||
or name.startswith("test_")
|
||||
or name.endswith("_test.py")
|
||||
or name.endswith(".test.ts")
|
||||
or name.endswith(".spec.ts")
|
||||
):
|
||||
facts.append(FactCandidate("test", path.name, relative))
|
||||
|
||||
if name in MANIFEST_FRAMEWORK_HINTS or name in {
|
||||
"requirements.txt",
|
||||
"poetry.lock",
|
||||
"package-lock.json",
|
||||
"pnpm-lock.yaml",
|
||||
"yarn.lock",
|
||||
"go.mod",
|
||||
}:
|
||||
facts.append(FactCandidate("manifest", path.name, relative))
|
||||
|
||||
if lower.endswith((".yaml", ".yml", ".toml", ".ini", ".env.example")):
|
||||
facts.append(FactCandidate("config", path.name, relative))
|
||||
|
||||
return facts
|
||||
|
||||
def _framework_facts(self, files: list[Path], root: Path) -> list[FactCandidate]:
|
||||
facts: list[FactCandidate] = []
|
||||
seen: set[tuple[str, str]] = set()
|
||||
for path in files:
|
||||
hints = MANIFEST_FRAMEWORK_HINTS.get(path.name)
|
||||
if hints is None:
|
||||
continue
|
||||
try:
|
||||
text = path.read_text(encoding="utf-8", errors="ignore").lower()
|
||||
except OSError:
|
||||
continue
|
||||
for needle, framework in hints.items():
|
||||
if needle not in text:
|
||||
continue
|
||||
key = (framework, path.relative_to(root).as_posix())
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
facts.append(
|
||||
FactCandidate(
|
||||
kind="framework",
|
||||
name=framework,
|
||||
path=path.relative_to(root).as_posix(),
|
||||
metadata={"source": "manifest_hint", "needle": needle},
|
||||
)
|
||||
)
|
||||
return facts
|
||||
|
||||
def _interface_facts(self, files: list[Path], root: Path) -> list[FactCandidate]:
|
||||
facts: list[FactCandidate] = []
|
||||
for path in files:
|
||||
relative = path.relative_to(root).as_posix()
|
||||
lower = relative.lower()
|
||||
if path.suffix == ".py":
|
||||
facts.extend(self._python_interface_facts(path, relative))
|
||||
if "cli" in lower or lower.endswith("/commands.py"):
|
||||
facts.append(FactCandidate("interface", "possible CLI", relative))
|
||||
if "routes" in lower or "api" in lower:
|
||||
facts.append(FactCandidate("interface", "possible API surface", relative))
|
||||
return facts
|
||||
|
||||
def _python_interface_facts(self, path: Path, relative: str) -> list[FactCandidate]:
|
||||
facts: list[FactCandidate] = []
|
||||
try:
|
||||
lines = path.read_text(encoding="utf-8", errors="ignore").splitlines()
|
||||
except OSError:
|
||||
return facts
|
||||
|
||||
for line_number, line in enumerate(lines, start=1):
|
||||
stripped = line.strip()
|
||||
if stripped.startswith("@app.") or stripped.startswith("@router."):
|
||||
facts.append(
|
||||
FactCandidate(
|
||||
kind="interface",
|
||||
name="python route decorator",
|
||||
path=relative,
|
||||
value=stripped,
|
||||
metadata={"line": line_number},
|
||||
)
|
||||
)
|
||||
elif stripped.startswith("@click.command") or stripped.startswith("@app.command"):
|
||||
facts.append(
|
||||
FactCandidate(
|
||||
kind="interface",
|
||||
name="python CLI command decorator",
|
||||
path=relative,
|
||||
value=stripped,
|
||||
metadata={"line": line_number},
|
||||
)
|
||||
)
|
||||
return facts
|
||||
|
||||
def _git_value(self, root: Path, *args: str) -> str | None:
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", *args],
|
||||
cwd=root,
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5,
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
return None
|
||||
if result.returncode != 0:
|
||||
return None
|
||||
return result.stdout.strip() or None
|
||||
Reference in New Issue
Block a user