feat: add recorder, project detector, and pattern detection
This commit is contained in:
99
cli_memory/project.py
Normal file
99
cli_memory/project.py
Normal file
@@ -0,0 +1,99 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user