292 lines
11 KiB
Python
292 lines
11 KiB
Python
|
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
Capability Discovery Tool
|
||
|
|
|
||
|
|
Provides automated discovery and validation of included capabilities
|
||
|
|
to prevent code duplication and ensure proper capability usage.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import os
|
||
|
|
import json
|
||
|
|
import subprocess
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Dict, List, Optional, Tuple
|
||
|
|
|
||
|
|
|
||
|
|
class CapabilityDiscovery:
|
||
|
|
"""Tool for discovering and validating included capabilities."""
|
||
|
|
|
||
|
|
def __init__(self, project_root: str = "."):
|
||
|
|
self.project_root = Path(project_root).resolve()
|
||
|
|
self.capabilities = {}
|
||
|
|
self._discover_capabilities()
|
||
|
|
|
||
|
|
def _discover_capabilities(self):
|
||
|
|
"""Discover all available capabilities."""
|
||
|
|
# Submodule capabilities
|
||
|
|
self._discover_submodules()
|
||
|
|
|
||
|
|
# Local capabilities
|
||
|
|
self._discover_local_capabilities()
|
||
|
|
|
||
|
|
# External dependencies
|
||
|
|
self._discover_external_dependencies()
|
||
|
|
|
||
|
|
def _discover_submodules(self):
|
||
|
|
"""Discover git submodule capabilities."""
|
||
|
|
gitmodules_path = self.project_root / ".gitmodules"
|
||
|
|
if not gitmodules_path.exists():
|
||
|
|
return
|
||
|
|
|
||
|
|
with open(gitmodules_path, 'r') as f:
|
||
|
|
content = f.read()
|
||
|
|
|
||
|
|
# Parse submodules
|
||
|
|
submodules = []
|
||
|
|
current_submodule = {}
|
||
|
|
|
||
|
|
for line in content.split('\n'):
|
||
|
|
line = line.strip()
|
||
|
|
if line.startswith('[submodule'):
|
||
|
|
if current_submodule:
|
||
|
|
submodules.append(current_submodule)
|
||
|
|
current_submodule = {'name': line.split('"')[1]}
|
||
|
|
elif '=' in line and current_submodule:
|
||
|
|
key, value = line.split('=', 1)
|
||
|
|
current_submodule[key.strip()] = value.strip()
|
||
|
|
|
||
|
|
if current_submodule:
|
||
|
|
submodules.append(current_submodule)
|
||
|
|
|
||
|
|
# Add to capabilities
|
||
|
|
for submodule in submodules:
|
||
|
|
path = submodule.get('path', '')
|
||
|
|
if path:
|
||
|
|
self.capabilities[path] = {
|
||
|
|
'type': 'submodule',
|
||
|
|
'name': submodule['name'],
|
||
|
|
'path': path,
|
||
|
|
'url': submodule.get('url', ''),
|
||
|
|
'status': self._check_submodule_status(path)
|
||
|
|
}
|
||
|
|
|
||
|
|
def _discover_local_capabilities(self):
|
||
|
|
"""Discover local capability directories."""
|
||
|
|
capabilities_dir = self.project_root / "capabilities"
|
||
|
|
if not capabilities_dir.exists():
|
||
|
|
return
|
||
|
|
|
||
|
|
for cap_dir in capabilities_dir.iterdir():
|
||
|
|
if cap_dir.is_dir() and not cap_dir.name.startswith('.'):
|
||
|
|
readme_path = cap_dir / "README.md"
|
||
|
|
self.capabilities[f"capabilities/{cap_dir.name}"] = {
|
||
|
|
'type': 'local',
|
||
|
|
'name': cap_dir.name,
|
||
|
|
'path': f"capabilities/{cap_dir.name}",
|
||
|
|
'has_readme': readme_path.exists(),
|
||
|
|
'status': 'available'
|
||
|
|
}
|
||
|
|
|
||
|
|
def _discover_external_dependencies(self):
|
||
|
|
"""Discover external package dependencies."""
|
||
|
|
pyproject_path = self.project_root / "pyproject.toml"
|
||
|
|
if not pyproject_path.exists():
|
||
|
|
return
|
||
|
|
|
||
|
|
# Basic parsing of dependencies (could be enhanced with toml library)
|
||
|
|
with open(pyproject_path, 'r') as f:
|
||
|
|
content = f.read()
|
||
|
|
|
||
|
|
# Extract key dependencies
|
||
|
|
key_deps = ['click', 'pytest', 'sqlalchemy', 'requests']
|
||
|
|
for dep in key_deps:
|
||
|
|
if dep in content:
|
||
|
|
self.capabilities[f"external/{dep}"] = {
|
||
|
|
'type': 'external',
|
||
|
|
'name': dep,
|
||
|
|
'path': f"external/{dep}",
|
||
|
|
'status': 'package_dependency'
|
||
|
|
}
|
||
|
|
|
||
|
|
def _check_submodule_status(self, path: str) -> str:
|
||
|
|
"""Check if submodule is properly initialized."""
|
||
|
|
submodule_path = self.project_root / path
|
||
|
|
|
||
|
|
if not submodule_path.exists():
|
||
|
|
return 'missing'
|
||
|
|
|
||
|
|
# Check if it's an empty directory
|
||
|
|
if submodule_path.is_dir() and not any(submodule_path.iterdir()):
|
||
|
|
return 'uninitialized'
|
||
|
|
|
||
|
|
return 'available'
|
||
|
|
|
||
|
|
def get_capability_info(self, name: str) -> Optional[Dict]:
|
||
|
|
"""Get information about a specific capability."""
|
||
|
|
for cap_path, info in self.capabilities.items():
|
||
|
|
if info['name'] == name or cap_path == name:
|
||
|
|
return info
|
||
|
|
return None
|
||
|
|
|
||
|
|
def search_functionality(self, search_term: str) -> List[Tuple[str, str]]:
|
||
|
|
"""Search for functionality across capabilities."""
|
||
|
|
results = []
|
||
|
|
|
||
|
|
for cap_path, info in self.capabilities.items():
|
||
|
|
if info['type'] == 'submodule' and info['status'] == 'available':
|
||
|
|
results.extend(self._search_in_submodule(cap_path, search_term))
|
||
|
|
elif info['type'] == 'local':
|
||
|
|
results.extend(self._search_in_local_capability(cap_path, search_term))
|
||
|
|
|
||
|
|
return results
|
||
|
|
|
||
|
|
def _search_in_submodule(self, path: str, search_term: str) -> List[Tuple[str, str]]:
|
||
|
|
"""Search for functionality in a submodule."""
|
||
|
|
results = []
|
||
|
|
submodule_path = self.project_root / path
|
||
|
|
|
||
|
|
if not submodule_path.exists():
|
||
|
|
return results
|
||
|
|
|
||
|
|
# Search README files
|
||
|
|
for readme in submodule_path.glob("**/README.md"):
|
||
|
|
try:
|
||
|
|
with open(readme, 'r') as f:
|
||
|
|
content = f.read().lower()
|
||
|
|
if search_term.lower() in content:
|
||
|
|
results.append((path, f"README: {readme.relative_to(submodule_path)}"))
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
|
||
|
|
# Search Python files for functions/classes
|
||
|
|
for py_file in submodule_path.glob("**/*.py"):
|
||
|
|
try:
|
||
|
|
with open(py_file, 'r') as f:
|
||
|
|
content = f.read()
|
||
|
|
if search_term in content:
|
||
|
|
results.append((path, f"Code: {py_file.relative_to(submodule_path)}"))
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
|
||
|
|
return results
|
||
|
|
|
||
|
|
def _search_in_local_capability(self, path: str, search_term: str) -> List[Tuple[str, str]]:
|
||
|
|
"""Search for functionality in a local capability."""
|
||
|
|
results = []
|
||
|
|
cap_path = self.project_root / path
|
||
|
|
|
||
|
|
if not cap_path.exists():
|
||
|
|
return results
|
||
|
|
|
||
|
|
# Search Python files
|
||
|
|
for py_file in cap_path.glob("**/*.py"):
|
||
|
|
try:
|
||
|
|
with open(py_file, 'r') as f:
|
||
|
|
content = f.read()
|
||
|
|
if search_term in content:
|
||
|
|
results.append((path, f"Code: {py_file.relative_to(cap_path)}"))
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
|
||
|
|
return results
|
||
|
|
|
||
|
|
def generate_report(self) -> str:
|
||
|
|
"""Generate a capability discovery report."""
|
||
|
|
report = ["# Capability Discovery Report", ""]
|
||
|
|
|
||
|
|
# Summary
|
||
|
|
total_caps = len(self.capabilities)
|
||
|
|
submodules = len([c for c in self.capabilities.values() if c['type'] == 'submodule'])
|
||
|
|
local = len([c for c in self.capabilities.values() if c['type'] == 'local'])
|
||
|
|
external = len([c for c in self.capabilities.values() if c['type'] == 'external'])
|
||
|
|
|
||
|
|
report.extend([
|
||
|
|
f"**Total Capabilities**: {total_caps}",
|
||
|
|
f"- Submodules: {submodules}",
|
||
|
|
f"- Local: {local}",
|
||
|
|
f"- External: {external}",
|
||
|
|
""
|
||
|
|
])
|
||
|
|
|
||
|
|
# Details by type
|
||
|
|
for cap_type in ['submodule', 'local', 'external']:
|
||
|
|
caps_of_type = {k: v for k, v in self.capabilities.items() if v['type'] == cap_type}
|
||
|
|
if caps_of_type:
|
||
|
|
report.append(f"## {cap_type.title()} Capabilities")
|
||
|
|
for path, info in caps_of_type.items():
|
||
|
|
status_emoji = "✅" if info['status'] == 'available' else "❌"
|
||
|
|
report.append(f"- {status_emoji} **{info['name']}** (`{path}`) - {info['status']}")
|
||
|
|
report.append("")
|
||
|
|
|
||
|
|
return '\n'.join(report)
|
||
|
|
|
||
|
|
def validate_capability_usage(self, file_path: str) -> List[str]:
|
||
|
|
"""Validate that a file properly uses existing capabilities."""
|
||
|
|
warnings = []
|
||
|
|
|
||
|
|
if not Path(file_path).exists():
|
||
|
|
return ["File not found"]
|
||
|
|
|
||
|
|
with open(file_path, 'r') as f:
|
||
|
|
content = f.read()
|
||
|
|
|
||
|
|
# Check for potential duplication
|
||
|
|
duplication_patterns = {
|
||
|
|
'issue management': ['create_issue', 'list_issues', 'close_issue'],
|
||
|
|
'content parsing': ['parse_markdown', 'extract_content', 'ContentParser'],
|
||
|
|
'utilities': ['helper_function', 'utility_function']
|
||
|
|
}
|
||
|
|
|
||
|
|
for capability, patterns in duplication_patterns.items():
|
||
|
|
for pattern in patterns:
|
||
|
|
if pattern in content:
|
||
|
|
# Check if proper capability import is used
|
||
|
|
if capability == 'issue management' and 'issue-facade' not in content:
|
||
|
|
warnings.append(f"Potential issue management duplication: {pattern} found but no issue-facade usage")
|
||
|
|
elif capability == 'content parsing' and 'markitect_content' not in content:
|
||
|
|
warnings.append(f"Potential content parsing duplication: {pattern} found but no markitect-content usage")
|
||
|
|
|
||
|
|
return warnings
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
"""CLI interface for capability discovery."""
|
||
|
|
import sys
|
||
|
|
|
||
|
|
discovery = CapabilityDiscovery()
|
||
|
|
|
||
|
|
if len(sys.argv) < 2:
|
||
|
|
print("Usage: capability_discovery.py [report|search|validate] [args...]")
|
||
|
|
return
|
||
|
|
|
||
|
|
command = sys.argv[1]
|
||
|
|
|
||
|
|
if command == "report":
|
||
|
|
print(discovery.generate_report())
|
||
|
|
|
||
|
|
elif command == "search" and len(sys.argv) > 2:
|
||
|
|
search_term = sys.argv[2]
|
||
|
|
results = discovery.search_functionality(search_term)
|
||
|
|
if results:
|
||
|
|
print(f"Found '{search_term}' in:")
|
||
|
|
for cap, location in results:
|
||
|
|
print(f" - {cap}: {location}")
|
||
|
|
else:
|
||
|
|
print(f"No results found for '{search_term}'")
|
||
|
|
|
||
|
|
elif command == "validate" and len(sys.argv) > 2:
|
||
|
|
file_path = sys.argv[2]
|
||
|
|
warnings = discovery.validate_capability_usage(file_path)
|
||
|
|
if warnings:
|
||
|
|
print("Validation warnings:")
|
||
|
|
for warning in warnings:
|
||
|
|
print(f" ⚠️ {warning}")
|
||
|
|
else:
|
||
|
|
print("✅ No capability usage issues found")
|
||
|
|
|
||
|
|
else:
|
||
|
|
print("Invalid command or missing arguments")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|