141 lines
6.9 KiB
Python
141 lines
6.9 KiB
Python
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from archaeology.report import ArchaeologyReport, InferredRequirement
|
|
from archaeology.scanner import RepositoryScanner
|
|
from control_plane.knowledge.models import KnowledgeEdge, KnowledgeNode
|
|
from control_plane.projects.models import Artifact, Finding, Project
|
|
|
|
|
|
class ProjectArchaeologist:
|
|
def __init__(self) -> None:
|
|
self.scanner = RepositoryScanner()
|
|
|
|
def inspect(self, project: Project, repository_path: Path) -> ArchaeologyReport:
|
|
scan = self.scanner.scan(repository_path)
|
|
git = self._git_metadata(scan.root)
|
|
django = self._django_metadata(scan.root)
|
|
todos = self._todos(scan.root)
|
|
observed = {
|
|
"root": str(scan.root),
|
|
"readmes": [path.relative_to(scan.root).as_posix() for path in scan.readmes],
|
|
"docs": [path.relative_to(scan.root).as_posix() for path in scan.docs],
|
|
"source_files": [path.relative_to(scan.root).as_posix() for path in scan.source_files[:200]],
|
|
"tests": [path.relative_to(scan.root).as_posix() for path in scan.tests[:200]],
|
|
"git": git,
|
|
"django": django,
|
|
"todos": todos[:100],
|
|
}
|
|
implemented: list[str] = []
|
|
likely_intended: list[str] = []
|
|
gaps: list[dict[str, object]] = []
|
|
inferred: list[InferredRequirement] = []
|
|
if django["models"]:
|
|
implemented.append("Django models exist")
|
|
inferred.append(InferredRequirement("Project uses Django persistence models", 0.9, django["models"], "implemented"))
|
|
if django["urls"]:
|
|
implemented.append("Django URL routing exists")
|
|
inferred.append(InferredRequirement("Project exposes HTTP routes", 0.8, django["urls"], "implemented"))
|
|
if scan.tests:
|
|
implemented.append("Automated tests exist")
|
|
inferred.append(InferredRequirement("Project expects automated test verification", 0.85, observed["tests"], "implemented"))
|
|
else:
|
|
gaps.append({"requirement": "Automated tests", "status": "missing", "evidence": ["No test files found"]})
|
|
if todos:
|
|
likely_intended.append("Repository contains TODO/FIXME markers")
|
|
gaps.append({"requirement": "Resolve incomplete implementation markers", "status": "partial", "evidence": todos[:20]})
|
|
likely_product = "Django/Python software project" if django["settings"] or django["models"] else "Python software project"
|
|
report = ArchaeologyReport(
|
|
likely_product=likely_product,
|
|
confidence=0.85 if django["settings"] else 0.65,
|
|
implemented=implemented,
|
|
partially_implemented=[gap["requirement"] for gap in gaps if gap["status"] == "partial"],
|
|
likely_intended=likely_intended,
|
|
recommended_next_steps=["Review gap analysis", "Ask Sol to interpret evidence", "Create target specification"],
|
|
observed_specification=observed,
|
|
inferred_specification=inferred,
|
|
gap_analysis=gaps,
|
|
)
|
|
self._persist(project, report)
|
|
self._populate_knowledge_graph(project, scan.root, observed)
|
|
return report
|
|
|
|
def _git_metadata(self, root: Path) -> dict[str, object]:
|
|
def git(args: list[str]) -> str:
|
|
result = subprocess.run(["git", *args], cwd=root, capture_output=True, text=True, check=False)
|
|
return result.stdout.strip() if result.returncode == 0 else ""
|
|
|
|
return {
|
|
"status": git(["status", "--short"]),
|
|
"branches": git(["branch", "--list", "--all"]),
|
|
"recent_commits": git(["log", "--oneline", "-10"]),
|
|
}
|
|
|
|
def _django_metadata(self, root: Path) -> dict[str, list[str]]:
|
|
py_files = [path for path in root.rglob("*.py") if ".git" not in path.parts]
|
|
return {
|
|
"settings": [path.relative_to(root).as_posix() for path in py_files if path.name == "settings.py"],
|
|
"models": [path.relative_to(root).as_posix() for path in py_files if path.name == "models.py"],
|
|
"urls": [path.relative_to(root).as_posix() for path in py_files if path.name == "urls.py"],
|
|
"admin": [path.relative_to(root).as_posix() for path in py_files if path.name == "admin.py"],
|
|
"migrations": [path.relative_to(root).as_posix() for path in py_files if "migrations" in path.parts],
|
|
"templates": [path.relative_to(root).as_posix() for path in root.rglob("*.html") if ".git" not in path.parts],
|
|
}
|
|
|
|
def _todos(self, root: Path) -> list[str]:
|
|
findings: list[str] = []
|
|
for path in root.rglob("*"):
|
|
if not path.is_file() or ".git" in path.parts or path.suffix not in {".py", ".md", ".txt", ".html"}:
|
|
continue
|
|
try:
|
|
for number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
|
|
if "TODO" in line or "FIXME" in line:
|
|
findings.append(f"{path.relative_to(root).as_posix()}:{number}: {line.strip()[:200]}")
|
|
except UnicodeDecodeError:
|
|
continue
|
|
return findings
|
|
|
|
def _persist(self, project: Project, report: ArchaeologyReport) -> None:
|
|
Artifact.objects.create(
|
|
project=project,
|
|
artifact_type="report",
|
|
name="archaeology_report",
|
|
content=report.as_dict(),
|
|
generated_by="project_archaeologist",
|
|
)
|
|
for gap in report.gap_analysis:
|
|
Finding.objects.create(
|
|
project=project,
|
|
source="archaeologist",
|
|
finding_type="gap",
|
|
severity="INFO",
|
|
title=str(gap["requirement"]),
|
|
description=str(gap.get("status", "")),
|
|
evidence=gap.get("evidence", []),
|
|
)
|
|
|
|
def _populate_knowledge_graph(self, project: Project, root: Path, observed: dict[str, object]) -> None:
|
|
project_node, _ = KnowledgeNode.objects.get_or_create(
|
|
project=project,
|
|
node_type="Project",
|
|
external_id=str(project.id),
|
|
defaults={"title": project.name},
|
|
)
|
|
for relative in observed.get("source_files", []):
|
|
file_node, _ = KnowledgeNode.objects.get_or_create(
|
|
project=project,
|
|
node_type="File",
|
|
external_id=str(relative),
|
|
defaults={"title": str(relative)},
|
|
)
|
|
KnowledgeEdge.objects.get_or_create(project=project, source=project_node, target=file_node, edge_type="contains")
|
|
for relative in observed.get("tests", []):
|
|
test_node, _ = KnowledgeNode.objects.get_or_create(
|
|
project=project,
|
|
node_type="Test",
|
|
external_id=str(relative),
|
|
defaults={"title": str(relative)},
|
|
)
|
|
KnowledgeEdge.objects.get_or_create(project=project, source=project_node, target=test_node, edge_type="contains")
|