100 lines
3.4 KiB
Python
100 lines
3.4 KiB
Python
import os
|
|
import subprocess
|
|
import re
|
|
from datetime import datetime
|
|
from typing import Optional, List, Dict, Any
|
|
from pathlib import Path
|
|
|
|
from .config import Config
|
|
from .models import Project
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ProjectDetector:
|
|
def __init__(self, config: Optional[Config] = None):
|
|
self.config = config or Config()
|
|
self._project_cache: Dict[str, Project] = {}
|
|
|
|
def detect(self, path: Optional[str] = None) -> Optional[Project]:
|
|
search_path = Path(path or os.getcwd()).resolve()
|
|
cache_key = str(search_path)
|
|
if cache_key in self._project_cache:
|
|
return self._project_cache[cache_key]
|
|
|
|
for parent in [search_path] + list(search_path.parents):
|
|
git_path = parent / ".git"
|
|
if git_path.exists() and git_path.is_dir():
|
|
project = self._create_project_from_git(parent)
|
|
if project:
|
|
self._project_cache[cache_key] = project
|
|
return project
|
|
|
|
if self.config.get("project.auto_detect_git", True):
|
|
return None
|
|
return self._create_project(search_path)
|
|
|
|
def _create_project_from_git(self, path: Path) -> Optional[Project]:
|
|
project = Project(
|
|
name=path.name,
|
|
path=str(path),
|
|
git_remote=self._get_git_remote(path),
|
|
tech_stack=self._detect_tech_stack(path),
|
|
created_at=datetime.utcnow(),
|
|
updated_at=datetime.utcnow(),
|
|
)
|
|
return project
|
|
|
|
def _create_project(self, path: Path) -> Project:
|
|
return Project(
|
|
name=path.name,
|
|
path=str(path),
|
|
tech_stack=self._detect_tech_stack(path),
|
|
created_at=datetime.utcnow(),
|
|
updated_at=datetime.utcnow(),
|
|
)
|
|
|
|
def _get_git_remote(self, path: Path) -> Optional[str]:
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "remote", "get-url", "origin"],
|
|
cwd=str(path),
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5,
|
|
)
|
|
if result.returncode == 0:
|
|
remote = result.stdout.strip()
|
|
remote = re.sub(r"\.git$", "", remote)
|
|
return remote
|
|
except (subprocess.SubprocessError, FileNotFoundError):
|
|
pass
|
|
return None
|
|
|
|
def _detect_tech_stack(self, path: Path) -> List[str]:
|
|
tech_stack = []
|
|
files = list(path.iterdir())
|
|
file_names = [f.name for f in files]
|
|
|
|
if "requirements.txt" in file_names or "setup.py" in file_names or "pyproject.toml" in file_names:
|
|
tech_stack.append("Python")
|
|
if "package.json" in file_names or "yarn.lock" in file_names:
|
|
tech_stack.append("Node.js")
|
|
if "Cargo.toml" in file_names:
|
|
tech_stack.append("Rust")
|
|
if "go.mod" in file_names:
|
|
tech_stack.append("Go")
|
|
if "pom.xml" in file_names or "build.gradle" in file_names:
|
|
tech_stack.append("Java")
|
|
if "Dockerfile" in file_names or "docker-compose.yml" in file_names:
|
|
tech_stack.append("Docker")
|
|
if "kubectl" in file_names or any("kube" in f for f in file_names):
|
|
tech_stack.append("Kubernetes")
|
|
if ".gitignore" in file_names:
|
|
tech_stack.append("Git")
|
|
|
|
return tech_stack
|
|
|
|
def clear_cache(self) -> None:
|
|
self._project_cache.clear()
|