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")