Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
ZacharyZhang-NYandClaude Sonnet 4.6 committed 2026-03-30 22:02:28 -04:00
1 parent 16d8426119
commit 66158e222e
398 files changed
+85259 -270637

No files matched your search

@@ -0,0 +1,660 @@
#!/usr/bin/env python3
"""
Architecture Diagram Generator
Generates architecture diagrams from project structure in multiple formats:
- Mermaid (default)
- PlantUML
- ASCII
Supports diagram types:
- component: Shows modules and their relationships
- layer: Shows architectural layers
- deployment: Shows deployment topology
"""
import os
import sys
import json
import argparse
import re
from pathlib import Path
from typing import Dict, List, Set, Tuple, Optional
from collections import defaultdict
class ProjectScanner:
"""Scans project structure to detect components and relationships."""
# Common architectural layer patterns
LAYER_PATTERNS = {
'presentation': ['controller', 'handler', 'view', 'page', 'component', 'ui'],
'api': ['api', 'route', 'endpoint', 'rest', 'graphql'],
'business': ['service', 'usecase', 'domain', 'logic', 'core'],
'data': ['repository', 'dao', 'model', 'entity', 'schema', 'migration'],
'infrastructure': ['config', 'util', 'helper', 'middleware', 'plugin'],
}
# File patterns for different technologies
TECH_PATTERNS = {
'react': ['jsx', 'tsx', 'package.json'],
'vue': ['vue', 'nuxt.config'],
'angular': ['component.ts', 'module.ts', 'angular.json'],
'node': ['package.json', 'express', 'fastify'],
'python': ['requirements.txt', 'pyproject.toml', 'setup.py'],
'go': ['go.mod', 'go.sum'],
'rust': ['Cargo.toml'],
'java': ['pom.xml', 'build.gradle'],
'docker': ['Dockerfile', 'docker-compose'],
'kubernetes': ['deployment.yaml', 'service.yaml', 'k8s'],
}
def __init__(self, project_path: Path):
self.project_path = project_path
self.components: Dict[str, Dict] = {}
self.relationships: List[Tuple[str, str, str]] = [] # (from, to, type)
self.layers: Dict[str, List[str]] = defaultdict(list)
self.technologies: Set[str] = set()
self.external_deps: Set[str] = set()
def scan(self) -> Dict:
"""Scan the project and return structure information."""
self._scan_directories()
self._detect_technologies()
self._detect_relationships()
self._classify_layers()
return {
'components': self.components,
'relationships': self.relationships,
'layers': dict(self.layers),
'technologies': list(self.technologies),
'external_deps': list(self.external_deps),
}
def _scan_directories(self):
"""Scan directory structure for components."""
ignore_dirs = {'.git', 'node_modules', '__pycache__', '.venv', 'venv',
'dist', 'build', '.next', '.nuxt', 'coverage', '.pytest_cache'}
for item in self.project_path.iterdir():
if item.is_dir() and item.name not in ignore_dirs and not item.name.startswith('.'):
component_info = self._analyze_directory(item)
if component_info['files'] > 0:
self.components[item.name] = component_info
def _analyze_directory(self, dir_path: Path) -> Dict:
"""Analyze a directory to understand its role."""
files = list(dir_path.rglob('*'))
code_files = [f for f in files if f.is_file() and f.suffix in
['.py', '.js', '.ts', '.jsx', '.tsx', '.go', '.rs', '.java', '.vue']]
# Count imports/dependencies within the directory
imports = set()
for f in code_files[:50]: # Limit to avoid large projects
imports.update(self._extract_imports(f))
return {
'path': str(dir_path.relative_to(self.project_path)),
'files': len(code_files),
'imports': list(imports)[:20], # Top 20 imports
'type': self._guess_component_type(dir_path.name),
}
def _extract_imports(self, file_path: Path) -> Set[str]:
"""Extract import statements from a file."""
imports = set()
try:
content = file_path.read_text(encoding='utf-8', errors='ignore')
# Python imports
py_imports = re.findall(r'^(?:from|import)\s+([\w.]+)', content, re.MULTILINE)
imports.update(py_imports)
# JS/TS imports
js_imports = re.findall(r'(?:import|require)\s*\(?[\'"]([^\'"\s]+)[\'"]', content)
imports.update(js_imports)
# Go imports
go_imports = re.findall(r'import\s+(?:\(\s*)?["\']([^"\']+)["\']', content)
imports.update(go_imports)
except Exception:
pass
return imports
def _guess_component_type(self, name: str) -> str:
"""Guess component type from directory name."""
name_lower = name.lower()
for layer, patterns in self.LAYER_PATTERNS.items():
for pattern in patterns:
if pattern in name_lower:
return layer
return 'unknown'
def _detect_technologies(self):
"""Detect technologies used in the project."""
for tech, patterns in self.TECH_PATTERNS.items():
for pattern in patterns:
matches = list(self.project_path.rglob(f'*{pattern}*'))
if matches:
self.technologies.add(tech)
break
# Detect external dependencies from package files
self._parse_package_json()
self._parse_requirements_txt()
self._parse_go_mod()
def _parse_package_json(self):
"""Parse package.json for dependencies."""
pkg_path = self.project_path / 'package.json'
if pkg_path.exists():
try:
data = json.loads(pkg_path.read_text())
deps = list(data.get('dependencies', {}).keys())[:10]
self.external_deps.update(deps)
except Exception:
pass
def _parse_requirements_txt(self):
"""Parse requirements.txt for dependencies."""
req_path = self.project_path / 'requirements.txt'
if req_path.exists():
try:
content = req_path.read_text()
deps = re.findall(r'^([a-zA-Z0-9_-]+)', content, re.MULTILINE)[:10]
self.external_deps.update(deps)
except Exception:
pass
def _parse_go_mod(self):
"""Parse go.mod for dependencies."""
mod_path = self.project_path / 'go.mod'
if mod_path.exists():
try:
content = mod_path.read_text()
deps = re.findall(r'^\s+([^\s]+)\s+v', content, re.MULTILINE)[:10]
self.external_deps.update([d.split('/')[-1] for d in deps])
except Exception:
pass
def _detect_relationships(self):
"""Detect relationships between components."""
component_names = set(self.components.keys())
for comp_name, comp_info in self.components.items():
for imp in comp_info.get('imports', []):
# Check if import references another component
for other_comp in component_names:
if other_comp != comp_name and other_comp.lower() in imp.lower():
self.relationships.append((comp_name, other_comp, 'uses'))
def _classify_layers(self):
"""Classify components into architectural layers."""
for comp_name, comp_info in self.components.items():
layer = comp_info.get('type', 'unknown')
if layer != 'unknown':
self.layers[layer].append(comp_name)
else:
self.layers['other'].append(comp_name)
class DiagramGenerator:
"""Base class for diagram generators."""
def __init__(self, scan_result: Dict):
self.components = scan_result['components']
self.relationships = scan_result['relationships']
self.layers = scan_result['layers']
self.technologies = scan_result['technologies']
self.external_deps = scan_result['external_deps']
def generate(self, diagram_type: str) -> str:
"""Generate diagram based on type."""
if diagram_type == 'component':
return self._generate_component_diagram()
elif diagram_type == 'layer':
return self._generate_layer_diagram()
elif diagram_type == 'deployment':
return self._generate_deployment_diagram()
else:
return self._generate_component_diagram()
def _generate_component_diagram(self) -> str:
raise NotImplementedError
def _generate_layer_diagram(self) -> str:
raise NotImplementedError
def _generate_deployment_diagram(self) -> str:
raise NotImplementedError
class MermaidGenerator(DiagramGenerator):
"""Generate Mermaid diagrams."""
def _generate_component_diagram(self) -> str:
lines = ['graph TD']
# Add components
for name, info in self.components.items():
safe_name = self._safe_id(name)
file_count = info.get('files', 0)
lines.append(f' {safe_name}["{name}<br/>{file_count} files"]')
# Add relationships
seen = set()
for src, dst, rel_type in self.relationships:
key = (src, dst)
if key not in seen:
seen.add(key)
lines.append(f' {self._safe_id(src)} --> {self._safe_id(dst)}')
# Add external dependencies if any
if self.external_deps:
lines.append('')
lines.append(' subgraph External')
for dep in list(self.external_deps)[:5]:
safe_dep = self._safe_id(dep)
lines.append(f' {safe_dep}(("{dep}"))')
lines.append(' end')
return '\n'.join(lines)
def _generate_layer_diagram(self) -> str:
lines = ['graph TB']
layer_order = ['presentation', 'api', 'business', 'data', 'infrastructure', 'other']
for layer in layer_order:
components = self.layers.get(layer, [])
if components:
lines.append(f' subgraph {layer.title()} Layer')
for comp in components:
safe_comp = self._safe_id(comp)
lines.append(f' {safe_comp}["{comp}"]')
lines.append(' end')
lines.append('')
# Add layer relationships (top-down)
prev_layer = None
for layer in layer_order:
if self.layers.get(layer):
if prev_layer and self.layers.get(prev_layer):
first_prev = self._safe_id(self.layers[prev_layer][0])
first_curr = self._safe_id(self.layers[layer][0])
lines.append(f' {first_prev} -.-> {first_curr}')
prev_layer = layer
return '\n'.join(lines)
def _generate_deployment_diagram(self) -> str:
lines = ['graph LR']
# Client
lines.append(' subgraph Client')
lines.append(' browser["Browser/Mobile"]')
lines.append(' end')
lines.append('')
# Determine if we have typical deployment components
has_api = any('api' in t for t in self.technologies)
has_docker = 'docker' in self.technologies
has_k8s = 'kubernetes' in self.technologies
# Application tier
lines.append(' subgraph Application')
if has_k8s:
lines.append(' k8s["Kubernetes Cluster"]')
elif has_docker:
lines.append(' docker["Docker Container"]')
else:
lines.append(' app["Application Server"]')
lines.append(' end')
lines.append('')
# Data tier
lines.append(' subgraph Data')
lines.append(' db[("Database")]')
if self.external_deps:
lines.append(' cache[("Cache")]')
lines.append(' end')
lines.append('')
# Connections
if has_k8s:
lines.append(' browser --> k8s')
lines.append(' k8s --> db')
elif has_docker:
lines.append(' browser --> docker')
lines.append(' docker --> db')
else:
lines.append(' browser --> app')
lines.append(' app --> db')
return '\n'.join(lines)
def _safe_id(self, name: str) -> str:
"""Convert name to safe Mermaid ID."""
return re.sub(r'[^a-zA-Z0-9]', '_', name)
class PlantUMLGenerator(DiagramGenerator):
"""Generate PlantUML diagrams."""
def _generate_component_diagram(self) -> str:
lines = ['@startuml', 'skinparam componentStyle rectangle', '']
# Add components
for name, info in self.components.items():
file_count = info.get('files', 0)
lines.append(f'component "{name}\\n({file_count} files)" as {self._safe_id(name)}')
lines.append('')
# Add relationships
seen = set()
for src, dst, rel_type in self.relationships:
key = (src, dst)
if key not in seen:
seen.add(key)
lines.append(f'{self._safe_id(src)} --> {self._safe_id(dst)}')
# External dependencies
if self.external_deps:
lines.append('')
lines.append('package "External Dependencies" {')
for dep in list(self.external_deps)[:5]:
lines.append(f' [{dep}]')
lines.append('}')
lines.append('')
lines.append('@enduml')
return '\n'.join(lines)
def _generate_layer_diagram(self) -> str:
lines = ['@startuml', 'skinparam packageStyle rectangle', '']
layer_order = ['presentation', 'api', 'business', 'data', 'infrastructure', 'other']
for layer in layer_order:
components = self.layers.get(layer, [])
if components:
lines.append(f'package "{layer.title()} Layer" {{')
for comp in components:
lines.append(f' [{comp}]')
lines.append('}')
lines.append('')
lines.append('@enduml')
return '\n'.join(lines)
def _generate_deployment_diagram(self) -> str:
lines = ['@startuml', '']
lines.append('node "Client" {')
lines.append(' [Browser/Mobile] as browser')
lines.append('}')
lines.append('')
has_docker = 'docker' in self.technologies
has_k8s = 'kubernetes' in self.technologies
lines.append('node "Application Server" {')
if has_k8s:
lines.append(' [Kubernetes Cluster] as app')
elif has_docker:
lines.append(' [Docker Container] as app')
else:
lines.append(' [Application] as app')
lines.append('}')
lines.append('')
lines.append('database "Data Store" {')
lines.append(' [Database] as db')
lines.append('}')
lines.append('')
lines.append('browser --> app')
lines.append('app --> db')
lines.append('')
lines.append('@enduml')
return '\n'.join(lines)
def _safe_id(self, name: str) -> str:
"""Convert name to safe PlantUML ID."""
return re.sub(r'[^a-zA-Z0-9]', '_', name)
class ASCIIGenerator(DiagramGenerator):
"""Generate ASCII diagrams."""
def _generate_component_diagram(self) -> str:
lines = []
lines.append('=' * 60)
lines.append('COMPONENT DIAGRAM')
lines.append('=' * 60)
lines.append('')
# Components
lines.append('Components:')
lines.append('-' * 40)
for name, info in self.components.items():
file_count = info.get('files', 0)
comp_type = info.get('type', 'unknown')
lines.append(f' [{name}]')
lines.append(f' Files: {file_count}')
lines.append(f' Type: {comp_type}')
lines.append('')
# Relationships
if self.relationships:
lines.append('Relationships:')
lines.append('-' * 40)
seen = set()
for src, dst, rel_type in self.relationships:
key = (src, dst)
if key not in seen:
seen.add(key)
lines.append(f' {src} --> {dst}')
lines.append('')
# External dependencies
if self.external_deps:
lines.append('External Dependencies:')
lines.append('-' * 40)
for dep in list(self.external_deps)[:10]:
lines.append(f' - {dep}')
lines.append('')
lines.append('=' * 60)
return '\n'.join(lines)
def _generate_layer_diagram(self) -> str:
lines = []
lines.append('=' * 60)
lines.append('LAYERED ARCHITECTURE')
lines.append('=' * 60)
lines.append('')
layer_order = ['presentation', 'api', 'business', 'data', 'infrastructure', 'other']
for layer in layer_order:
components = self.layers.get(layer, [])
if components:
lines.append(f'+{"-" * 56}+')
lines.append(f'| {layer.upper():^54} |')
lines.append(f'+{"-" * 56}+')
for comp in components:
lines.append(f'| [{comp:^48}] |')
lines.append(f'+{"-" * 56}+')
lines.append(' |')
lines.append(' v')
# Remove last arrow
if lines[-2:] == [' |', ' v']:
lines = lines[:-2]
lines.append('')
lines.append('=' * 60)
return '\n'.join(lines)
def _generate_deployment_diagram(self) -> str:
lines = []
lines.append('=' * 60)
lines.append('DEPLOYMENT DIAGRAM')
lines.append('=' * 60)
lines.append('')
has_docker = 'docker' in self.technologies
has_k8s = 'kubernetes' in self.technologies
# Client tier
lines.append('+----------------------+')
lines.append('| CLIENT |')
lines.append('| [Browser/Mobile] |')
lines.append('+----------+-----------+')
lines.append(' |')
lines.append(' v')
# Application tier
lines.append('+----------------------+')
lines.append('| APPLICATION |')
if has_k8s:
lines.append('| [Kubernetes Cluster] |')
elif has_docker:
lines.append('| [Docker Container] |')
else:
lines.append('| [App Server] |')
lines.append('+----------+-----------+')
lines.append(' |')
lines.append(' v')
# Data tier
lines.append('+----------------------+')
lines.append('| DATA |')
lines.append('| [(Database)] |')
lines.append('+----------------------+')
lines.append('')
# Technologies detected
if self.technologies:
lines.append('Technologies detected:')
lines.append('-' * 40)
for tech in sorted(self.technologies):
lines.append(f' - {tech}')
lines.append('')
lines.append('=' * 60)
return '\n'.join(lines)
def main():
parser = argparse.ArgumentParser(
description='Generate architecture diagrams from project structure',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog='''
Examples:
%(prog)s ./my-project --format mermaid
%(prog)s ./my-project --format plantuml --type layer
%(prog)s ./my-project --format ascii -o architecture.txt
Diagram types:
component - Shows modules and their relationships (default)
layer - Shows architectural layers
deployment - Shows deployment topology
Output formats:
mermaid - Mermaid.js format (default)
plantuml - PlantUML format
ascii - ASCII art format
'''
)
parser.add_argument(
'project_path',
help='Path to the project directory'
)
parser.add_argument(
'--format', '-f',
choices=['mermaid', 'plantuml', 'ascii'],
default='mermaid',
help='Output format (default: mermaid)'
)
parser.add_argument(
'--type', '-t',
choices=['component', 'layer', 'deployment'],
default='component',
help='Diagram type (default: component)'
)
parser.add_argument(
'--output', '-o',
help='Output file path (prints to stdout if not specified)'
)
parser.add_argument(
'--verbose', '-v',
action='store_true',
help='Enable verbose output'
)
parser.add_argument(
'--json',
action='store_true',
help='Output raw scan results as JSON'
)
args = parser.parse_args()
project_path = Path(args.project_path).resolve()
if not project_path.exists():
print(f"Error: Project path does not exist: {project_path}", file=sys.stderr)
sys.exit(1)
if not project_path.is_dir():
print(f"Error: Project path is not a directory: {project_path}", file=sys.stderr)
sys.exit(1)
if args.verbose:
print(f"Scanning project: {project_path}")
# Scan project
scanner = ProjectScanner(project_path)
scan_result = scanner.scan()
if args.verbose:
print(f"Found {len(scan_result['components'])} components")
print(f"Found {len(scan_result['relationships'])} relationships")
print(f"Technologies: {', '.join(scan_result['technologies']) or 'none detected'}")
# Output raw JSON if requested
if args.json:
output = json.dumps(scan_result, indent=2)
if args.output:
Path(args.output).write_text(output)
print(f"Results written to {args.output}")
else:
print(output)
return
# Generate diagram
generators = {
'mermaid': MermaidGenerator,
'plantuml': PlantUMLGenerator,
'ascii': ASCIIGenerator,
}
generator = generators[args.format](scan_result)
diagram = generator.generate(args.type)
# Output
if args.output:
Path(args.output).write_text(diagram)
print(f"Diagram written to {args.output}")
else:
print(diagram)
if __name__ == '__main__':
main()
@@ -0,0 +1,615 @@
#!/usr/bin/env python3
"""
Dependency Analyzer
Analyzes project dependencies for:
- Dependency tree (direct and transitive)
- Circular dependencies between modules
- Coupling score (0-100)
- Outdated packages (basic detection)
Supports:
- npm/yarn (package.json)
- Python (requirements.txt, pyproject.toml)
- Go (go.mod)
- Rust (Cargo.toml)
"""
import os
import sys
import json
import argparse
import re
from pathlib import Path
from typing import Dict, List, Set, Tuple, Optional
from collections import defaultdict
class DependencyAnalyzer:
"""Analyzes project dependencies and module coupling."""
def __init__(self, project_path: Path, verbose: bool = False):
self.project_path = project_path
self.verbose = verbose
# Results
self.direct_deps: Dict[str, str] = {} # name -> version
self.dev_deps: Dict[str, str] = {}
self.internal_modules: Dict[str, Set[str]] = defaultdict(set) # module -> imports
self.circular_deps: List[List[str]] = []
self.coupling_score: float = 0
self.issues: List[Dict] = []
self.recommendations: List[str] = []
self.package_manager: Optional[str] = None
def analyze(self) -> Dict:
"""Run full dependency analysis."""
self._detect_package_manager()
self._parse_dependencies()
self._scan_internal_modules()
self._detect_circular_dependencies()
self._calculate_coupling_score()
self._generate_recommendations()
return self._build_report()
def _detect_package_manager(self):
"""Detect which package manager is used."""
if (self.project_path / 'package.json').exists():
self.package_manager = 'npm'
elif (self.project_path / 'requirements.txt').exists():
self.package_manager = 'pip'
elif (self.project_path / 'pyproject.toml').exists():
self.package_manager = 'poetry'
elif (self.project_path / 'go.mod').exists():
self.package_manager = 'go'
elif (self.project_path / 'Cargo.toml').exists():
self.package_manager = 'cargo'
else:
self.package_manager = 'unknown'
if self.verbose:
print(f"Detected package manager: {self.package_manager}")
def _parse_dependencies(self):
"""Parse dependencies based on detected package manager."""
parsers = {
'npm': self._parse_npm,
'pip': self._parse_pip,
'poetry': self._parse_poetry,
'go': self._parse_go,
'cargo': self._parse_cargo,
}
parser = parsers.get(self.package_manager)
if parser:
parser()
def _parse_npm(self):
"""Parse package.json for npm dependencies."""
pkg_path = self.project_path / 'package.json'
try:
data = json.loads(pkg_path.read_text())
# Direct dependencies
for name, version in data.get('dependencies', {}).items():
self.direct_deps[name] = self._clean_version(version)
# Dev dependencies
for name, version in data.get('devDependencies', {}).items():
self.dev_deps[name] = self._clean_version(version)
if self.verbose:
print(f"Found {len(self.direct_deps)} direct deps, "
f"{len(self.dev_deps)} dev deps")
except Exception as e:
self.issues.append({
'type': 'parse_error',
'severity': 'error',
'message': f"Failed to parse package.json: {e}"
})
def _parse_pip(self):
"""Parse requirements.txt for Python dependencies."""
req_path = self.project_path / 'requirements.txt'
try:
content = req_path.read_text()
for line in content.strip().split('\n'):
line = line.strip()
if not line or line.startswith('#') or line.startswith('-'):
continue
# Parse name and version
match = re.match(r'^([a-zA-Z0-9_-]+)(?:[=<>!~]+(.+))?', line)
if match:
name = match.group(1)
version = match.group(2) or 'any'
self.direct_deps[name] = version
if self.verbose:
print(f"Found {len(self.direct_deps)} dependencies")
except Exception as e:
self.issues.append({
'type': 'parse_error',
'severity': 'error',
'message': f"Failed to parse requirements.txt: {e}"
})
def _parse_poetry(self):
"""Parse pyproject.toml for Poetry dependencies."""
toml_path = self.project_path / 'pyproject.toml'
try:
content = toml_path.read_text()
# Simple TOML parsing for dependencies section
in_deps = False
in_dev_deps = False
for line in content.split('\n'):
line = line.strip()
if line == '[tool.poetry.dependencies]':
in_deps = True
in_dev_deps = False
continue
elif line == '[tool.poetry.dev-dependencies]' or \
line == '[tool.poetry.group.dev.dependencies]':
in_deps = False
in_dev_deps = True
continue
elif line.startswith('['):
in_deps = False
in_dev_deps = False
continue
if (in_deps or in_dev_deps) and '=' in line:
match = re.match(r'^([a-zA-Z0-9_-]+)\s*=\s*["\']?([^"\']+)', line)
if match:
name = match.group(1)
version = match.group(2)
if name != 'python':
if in_deps:
self.direct_deps[name] = version
else:
self.dev_deps[name] = version
if self.verbose:
print(f"Found {len(self.direct_deps)} direct deps, "
f"{len(self.dev_deps)} dev deps")
except Exception as e:
self.issues.append({
'type': 'parse_error',
'severity': 'error',
'message': f"Failed to parse pyproject.toml: {e}"
})
def _parse_go(self):
"""Parse go.mod for Go dependencies."""
mod_path = self.project_path / 'go.mod'
try:
content = mod_path.read_text()
# Find require block
in_require = False
for line in content.split('\n'):
line = line.strip()
if line.startswith('require ('):
in_require = True
continue
elif line == ')' and in_require:
in_require = False
continue
elif line.startswith('require ') and '(' not in line:
# Single-line require
match = re.match(r'require\s+([^\s]+)\s+([^\s]+)', line)
if match:
self.direct_deps[match.group(1)] = match.group(2)
continue
if in_require:
match = re.match(r'([^\s]+)\s+([^\s]+)', line)
if match:
self.direct_deps[match.group(1)] = match.group(2)
if self.verbose:
print(f"Found {len(self.direct_deps)} dependencies")
except Exception as e:
self.issues.append({
'type': 'parse_error',
'severity': 'error',
'message': f"Failed to parse go.mod: {e}"
})
def _parse_cargo(self):
"""Parse Cargo.toml for Rust dependencies."""
cargo_path = self.project_path / 'Cargo.toml'
try:
content = cargo_path.read_text()
in_deps = False
in_dev_deps = False
for line in content.split('\n'):
line = line.strip()
if line == '[dependencies]':
in_deps = True
in_dev_deps = False
continue
elif line == '[dev-dependencies]':
in_deps = False
in_dev_deps = True
continue
elif line.startswith('['):
in_deps = False
in_dev_deps = False
continue
if (in_deps or in_dev_deps) and '=' in line:
match = re.match(r'^([a-zA-Z0-9_-]+)\s*=\s*["\']?([^"\']+)', line)
if match:
name = match.group(1)
version = match.group(2)
if in_deps:
self.direct_deps[name] = version
else:
self.dev_deps[name] = version
if self.verbose:
print(f"Found {len(self.direct_deps)} direct deps, "
f"{len(self.dev_deps)} dev deps")
except Exception as e:
self.issues.append({
'type': 'parse_error',
'severity': 'error',
'message': f"Failed to parse Cargo.toml: {e}"
})
def _clean_version(self, version: str) -> str:
"""Clean version string."""
return version.lstrip('^~>=<!')
def _scan_internal_modules(self):
"""Scan internal module imports for coupling analysis."""
ignore_dirs = {'.git', 'node_modules', '__pycache__', '.venv', 'venv',
'dist', 'build', '.next', 'coverage'}
# Find all code files
extensions = ['.py', '.js', '.ts', '.jsx', '.tsx', '.go', '.rs']
for ext in extensions:
for file_path in self.project_path.rglob(f'*{ext}'):
# Skip ignored directories
if any(ignored in file_path.parts for ignored in ignore_dirs):
continue
# Get module name (directory relative to project root)
try:
rel_path = file_path.relative_to(self.project_path)
module = rel_path.parts[0] if len(rel_path.parts) > 1 else 'root'
# Extract imports
imports = self._extract_imports(file_path)
self.internal_modules[module].update(imports)
except Exception:
continue
if self.verbose:
print(f"Scanned {len(self.internal_modules)} internal modules")
def _extract_imports(self, file_path: Path) -> Set[str]:
"""Extract import statements from a file."""
imports = set()
try:
content = file_path.read_text(encoding='utf-8', errors='ignore')
# Python imports
for match in re.finditer(r'^(?:from|import)\s+([\w.]+)', content, re.MULTILINE):
imports.add(match.group(1).split('.')[0])
# JS/TS imports
for match in re.finditer(r'(?:import|require)\s*\(?[\'"]([^\'"\s]+)[\'"]', content):
imp = match.group(1)
if imp.startswith('.') or imp.startswith('@/') or imp.startswith('~/'):
# Relative import - extract first path component
parts = imp.lstrip('./~@').split('/')
if parts:
imports.add(parts[0])
except Exception:
pass
return imports
def _detect_circular_dependencies(self):
"""Detect circular dependencies between internal modules."""
# Build dependency graph
graph = defaultdict(set)
modules = set(self.internal_modules.keys())
for module, imports in self.internal_modules.items():
for imp in imports:
# Check if import is an internal module
for internal_module in modules:
if internal_module.lower() in imp.lower() and internal_module != module:
graph[module].add(internal_module)
# Find cycles using DFS
visited = set()
rec_stack = set()
cycles = []
def find_cycles(node: str, path: List[str]):
visited.add(node)
rec_stack.add(node)
path.append(node)
for neighbor in graph.get(node, []):
if neighbor not in visited:
find_cycles(neighbor, path)
elif neighbor in rec_stack:
# Found cycle
cycle_start = path.index(neighbor)
cycle = path[cycle_start:] + [neighbor]
if cycle not in cycles:
cycles.append(cycle)
path.pop()
rec_stack.remove(node)
for module in modules:
if module not in visited:
find_cycles(module, [])
self.circular_deps = cycles
if cycles:
for cycle in cycles:
self.issues.append({
'type': 'circular_dependency',
'severity': 'warning',
'message': f"Circular dependency: {' -> '.join(cycle)}"
})
if self.verbose:
print(f"Found {len(self.circular_deps)} circular dependencies")
def _calculate_coupling_score(self):
"""Calculate coupling score (0-100, lower is better)."""
if not self.internal_modules:
self.coupling_score = 0
return
# Count connections between modules
total_modules = len(self.internal_modules)
total_connections = 0
modules = set(self.internal_modules.keys())
for module, imports in self.internal_modules.items():
for imp in imports:
for internal_module in modules:
if internal_module.lower() in imp.lower() and internal_module != module:
total_connections += 1
# Max possible connections (complete graph)
max_connections = total_modules * (total_modules - 1) if total_modules > 1 else 1
# Coupling score as percentage of max connections
self.coupling_score = min(100, int((total_connections / max_connections) * 100))
# Add penalty for circular dependencies
self.coupling_score = min(100, self.coupling_score + len(self.circular_deps) * 10)
if self.verbose:
print(f"Coupling score: {self.coupling_score}/100")
def _generate_recommendations(self):
"""Generate actionable recommendations."""
# Circular dependency recommendations
if self.circular_deps:
self.recommendations.append(
"Extract shared interfaces or create a common module to break circular dependencies"
)
# High coupling recommendations
if self.coupling_score > 70:
self.recommendations.append(
"High coupling detected. Consider applying SOLID principles and "
"introducing abstraction layers"
)
# Too many dependencies
if len(self.direct_deps) > 50:
self.recommendations.append(
f"Large dependency count ({len(self.direct_deps)}). "
"Review for unused dependencies and consider bundle size impact"
)
# Check for known problematic packages (simplified check)
problematic = {
'lodash': 'Consider lodash-es or native methods for smaller bundle',
'moment': 'Consider day.js or date-fns for smaller bundle',
'request': 'Deprecated. Use axios, node-fetch, or native fetch',
}
for pkg, suggestion in problematic.items():
if pkg in self.direct_deps:
self.recommendations.append(f"{pkg}: {suggestion}")
def _build_report(self) -> Dict:
"""Build the analysis report."""
return {
'project_path': str(self.project_path),
'package_manager': self.package_manager,
'summary': {
'direct_dependencies': len(self.direct_deps),
'dev_dependencies': len(self.dev_deps),
'internal_modules': len(self.internal_modules),
'coupling_score': self.coupling_score,
'circular_dependencies': len(self.circular_deps),
'issues': len(self.issues),
},
'dependencies': {
'direct': self.direct_deps,
'dev': self.dev_deps,
},
'internal_modules': {k: list(v) for k, v in self.internal_modules.items()},
'circular_dependencies': self.circular_deps,
'issues': self.issues,
'recommendations': self.recommendations,
}
def print_human_report(report: Dict):
"""Print human-readable report."""
print("\n" + "=" * 60)
print("DEPENDENCY ANALYSIS REPORT")
print("=" * 60)
print(f"\nProject: {report['project_path']}")
print(f"Package Manager: {report['package_manager']}")
summary = report['summary']
print("\n--- Summary ---")
print(f"Direct dependencies: {summary['direct_dependencies']}")
print(f"Dev dependencies: {summary['dev_dependencies']}")
print(f"Internal modules: {summary['internal_modules']}")
print(f"Coupling score: {summary['coupling_score']}/100 ", end='')
if summary['coupling_score'] < 30:
print("(low - good)")
elif summary['coupling_score'] < 70:
print("(moderate)")
else:
print("(high - consider refactoring)")
if report['circular_dependencies']:
print(f"\n--- Circular Dependencies ({len(report['circular_dependencies'])}) ---")
for cycle in report['circular_dependencies']:
print(f" {' -> '.join(cycle)}")
if report['issues']:
print(f"\n--- Issues ({len(report['issues'])}) ---")
for issue in report['issues']:
severity = issue['severity'].upper()
print(f" [{severity}] {issue['message']}")
if report['recommendations']:
print(f"\n--- Recommendations ---")
for i, rec in enumerate(report['recommendations'], 1):
print(f" {i}. {rec}")
# Show top dependencies
deps = report['dependencies']['direct']
if deps:
print(f"\n--- Top Dependencies (of {len(deps)}) ---")
for name, version in list(deps.items())[:10]:
print(f" {name}: {version}")
if len(deps) > 10:
print(f" ... and {len(deps) - 10} more")
print("\n" + "=" * 60)
def main():
parser = argparse.ArgumentParser(
description='Analyze project dependencies and module coupling',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog='''
Examples:
%(prog)s ./my-project
%(prog)s ./my-project --output json
%(prog)s ./my-project --check circular
%(prog)s ./my-project --verbose
Supported package managers:
- npm/yarn (package.json)
- pip (requirements.txt)
- poetry (pyproject.toml)
- go (go.mod)
- cargo (Cargo.toml)
'''
)
parser.add_argument(
'project_path',
help='Path to the project directory'
)
parser.add_argument(
'--output', '-o',
choices=['human', 'json'],
default='human',
help='Output format (default: human)'
)
parser.add_argument(
'--check',
choices=['all', 'circular', 'coupling'],
default='all',
help='What to check (default: all)'
)
parser.add_argument(
'--verbose', '-v',
action='store_true',
help='Enable verbose output'
)
parser.add_argument(
'--save', '-s',
help='Save report to file'
)
args = parser.parse_args()
project_path = Path(args.project_path).resolve()
if not project_path.exists():
print(f"Error: Project path does not exist: {project_path}", file=sys.stderr)
sys.exit(1)
if not project_path.is_dir():
print(f"Error: Project path is not a directory: {project_path}", file=sys.stderr)
sys.exit(1)
# Run analysis
analyzer = DependencyAnalyzer(project_path, verbose=args.verbose)
report = analyzer.analyze()
# Filter report based on --check option
if args.check == 'circular':
if report['circular_dependencies']:
print("Circular dependencies found:")
for cycle in report['circular_dependencies']:
print(f" {' -> '.join(cycle)}")
sys.exit(1)
else:
print("No circular dependencies found.")
sys.exit(0)
elif args.check == 'coupling':
score = report['summary']['coupling_score']
print(f"Coupling score: {score}/100")
if score > 70:
print("WARNING: High coupling detected")
sys.exit(1)
sys.exit(0)
# Output report
if args.output == 'json':
output = json.dumps(report, indent=2)
if args.save:
Path(args.save).write_text(output)
print(f"Report saved to {args.save}")
else:
print(output)
else:
print_human_report(report)
if args.save:
Path(args.save).write_text(json.dumps(report, indent=2))
print(f"\nJSON report saved to {args.save}")
if __name__ == '__main__':
main()
@@ -0,0 +1,749 @@
#!/usr/bin/env python3
"""
Project Architect
Analyzes project structure and detects:
- Architectural patterns (MVC, layered, hexagonal, microservices)
- Code organization issues (god classes, mixed concerns)
- Layer violations
- Missing architectural components
Provides architecture assessment and improvement recommendations.
"""
import os
import sys
import json
import argparse
import re
from pathlib import Path
from typing import Dict, List, Set, Tuple, Optional
from collections import defaultdict
class PatternDetector:
"""Detects architectural patterns in a project."""
# Pattern signatures
PATTERNS = {
'layered': {
'indicators': ['controller', 'service', 'repository', 'dao', 'model', 'entity'],
'structure': ['controllers', 'services', 'repositories', 'models'],
'weight': 0,
},
'mvc': {
'indicators': ['model', 'view', 'controller'],
'structure': ['models', 'views', 'controllers'],
'weight': 0,
},
'hexagonal': {
'indicators': ['port', 'adapter', 'domain', 'infrastructure', 'application'],
'structure': ['ports', 'adapters', 'domain', 'infrastructure'],
'weight': 0,
},
'clean': {
'indicators': ['entity', 'usecase', 'interface', 'framework', 'adapter'],
'structure': ['entities', 'usecases', 'interfaces', 'frameworks'],
'weight': 0,
},
'microservices': {
'indicators': ['service', 'api', 'gateway', 'docker', 'kubernetes'],
'structure': ['services', 'api-gateway', 'docker-compose'],
'weight': 0,
},
'modular_monolith': {
'indicators': ['module', 'feature', 'bounded'],
'structure': ['modules', 'features'],
'weight': 0,
},
'feature_based': {
'indicators': ['feature', 'component', 'page'],
'structure': ['features', 'components', 'pages'],
'weight': 0,
},
}
# Layer definitions for violation detection
LAYER_HIERARCHY = {
'presentation': ['controller', 'handler', 'view', 'page', 'component', 'ui', 'route'],
'application': ['service', 'usecase', 'application', 'facade'],
'domain': ['domain', 'entity', 'model', 'aggregate', 'valueobject'],
'infrastructure': ['repository', 'dao', 'adapter', 'gateway', 'client', 'config'],
}
LAYER_ORDER = ['presentation', 'application', 'domain', 'infrastructure']
def __init__(self, project_path: Path):
self.project_path = project_path
self.directories: Set[str] = set()
self.files: Dict[str, List[str]] = defaultdict(list) # dir -> files
self.detected_pattern: Optional[str] = None
self.confidence: float = 0
self.layer_assignments: Dict[str, str] = {} # dir -> layer
def scan(self) -> Dict:
"""Scan project and detect patterns."""
self._scan_structure()
self._detect_pattern()
self._assign_layers()
return {
'detected_pattern': self.detected_pattern,
'confidence': self.confidence,
'directories': list(self.directories),
'layer_assignments': self.layer_assignments,
'pattern_scores': {p: d['weight'] for p, d in self.PATTERNS.items()},
}
def _scan_structure(self):
"""Scan directory structure."""
ignore_dirs = {'.git', 'node_modules', '__pycache__', '.venv', 'venv',
'dist', 'build', '.next', 'coverage', '.pytest_cache'}
for item in self.project_path.iterdir():
if item.is_dir() and item.name not in ignore_dirs and not item.name.startswith('.'):
self.directories.add(item.name.lower())
# Scan files in directory
try:
for f in item.rglob('*'):
if f.is_file():
self.files[item.name.lower()].append(f.name.lower())
except PermissionError:
pass
def _detect_pattern(self):
"""Detect the primary architectural pattern."""
for pattern, config in self.PATTERNS.items():
score = 0
# Check directory structure
for struct in config['structure']:
if struct.lower() in self.directories:
score += 2
# Check indicator presence in directory names
for indicator in config['indicators']:
for dir_name in self.directories:
if indicator in dir_name:
score += 1
# Check file patterns
all_files = [f for files in self.files.values() for f in files]
for indicator in config['indicators']:
matching_files = sum(1 for f in all_files if indicator in f)
score += min(matching_files // 5, 3) # Cap contribution
config['weight'] = score
# Find best match
best_pattern = max(self.PATTERNS.items(), key=lambda x: x[1]['weight'])
if best_pattern[1]['weight'] > 3:
self.detected_pattern = best_pattern[0]
max_possible = len(best_pattern[1]['structure']) * 2 + len(best_pattern[1]['indicators']) * 2
self.confidence = min(100, int((best_pattern[1]['weight'] / max(max_possible, 1)) * 100))
else:
self.detected_pattern = 'unstructured'
self.confidence = 0
def _assign_layers(self):
"""Assign directories to architectural layers."""
for dir_name in self.directories:
for layer, indicators in self.LAYER_HIERARCHY.items():
for indicator in indicators:
if indicator in dir_name:
self.layer_assignments[dir_name] = layer
break
if dir_name in self.layer_assignments:
break
if dir_name not in self.layer_assignments:
self.layer_assignments[dir_name] = 'unknown'
class CodeAnalyzer:
"""Analyzes code for architectural issues."""
# Thresholds
MAX_FILE_LINES = 500
MAX_CLASS_LINES = 300
MAX_FUNCTION_LINES = 50
MAX_IMPORTS_PER_FILE = 30
def __init__(self, project_path: Path, verbose: bool = False):
self.project_path = project_path
self.verbose = verbose
self.issues: List[Dict] = []
self.metrics: Dict = {}
def analyze(self) -> Dict:
"""Run code analysis."""
self._analyze_file_sizes()
self._analyze_imports()
self._detect_god_classes()
self._check_naming_conventions()
return {
'issues': self.issues,
'metrics': self.metrics,
}
def _analyze_file_sizes(self):
"""Check for oversized files."""
extensions = ['.py', '.js', '.ts', '.jsx', '.tsx', '.go', '.rs', '.java']
large_files = []
total_lines = 0
file_count = 0
ignore_dirs = {'.git', 'node_modules', '__pycache__', '.venv', 'venv',
'dist', 'build', '.next', 'coverage'}
for ext in extensions:
for file_path in self.project_path.rglob(f'*{ext}'):
if any(ignored in file_path.parts for ignored in ignore_dirs):
continue
try:
content = file_path.read_text(encoding='utf-8', errors='ignore')
lines = len(content.split('\n'))
total_lines += lines
file_count += 1
if lines > self.MAX_FILE_LINES:
large_files.append({
'path': str(file_path.relative_to(self.project_path)),
'lines': lines,
})
self.issues.append({
'type': 'large_file',
'severity': 'warning',
'file': str(file_path.relative_to(self.project_path)),
'message': f"File has {lines} lines (threshold: {self.MAX_FILE_LINES})",
'suggestion': "Consider splitting into smaller, focused modules",
})
except Exception:
pass
self.metrics['total_lines'] = total_lines
self.metrics['file_count'] = file_count
self.metrics['avg_file_lines'] = total_lines // file_count if file_count > 0 else 0
self.metrics['large_files'] = large_files
def _analyze_imports(self):
"""Analyze import patterns."""
extensions = ['.py', '.js', '.ts', '.jsx', '.tsx']
high_import_files = []
ignore_dirs = {'.git', 'node_modules', '__pycache__', '.venv', 'venv',
'dist', 'build', '.next', 'coverage'}
for ext in extensions:
for file_path in self.project_path.rglob(f'*{ext}'):
if any(ignored in file_path.parts for ignored in ignore_dirs):
continue
try:
content = file_path.read_text(encoding='utf-8', errors='ignore')
# Count imports
py_imports = len(re.findall(r'^(?:from|import)\s+', content, re.MULTILINE))
js_imports = len(re.findall(r'^import\s+', content, re.MULTILINE))
imports = py_imports + js_imports
if imports > self.MAX_IMPORTS_PER_FILE:
high_import_files.append({
'path': str(file_path.relative_to(self.project_path)),
'imports': imports,
})
self.issues.append({
'type': 'high_imports',
'severity': 'info',
'file': str(file_path.relative_to(self.project_path)),
'message': f"File has {imports} imports (threshold: {self.MAX_IMPORTS_PER_FILE})",
'suggestion': "Consider if all imports are necessary or if the file has too many responsibilities",
})
except Exception:
pass
self.metrics['high_import_files'] = high_import_files
def _detect_god_classes(self):
"""Detect potential god classes (oversized classes)."""
extensions = ['.py', '.js', '.ts', '.java']
god_classes = []
ignore_dirs = {'.git', 'node_modules', '__pycache__', '.venv', 'venv',
'dist', 'build', '.next', 'coverage'}
for ext in extensions:
for file_path in self.project_path.rglob(f'*{ext}'):
if any(ignored in file_path.parts for ignored in ignore_dirs):
continue
try:
content = file_path.read_text(encoding='utf-8', errors='ignore')
lines = content.split('\n')
# Simple class detection
class_pattern = r'^\s*(?:export\s+)?(?:abstract\s+)?class\s+(\w+)'
in_class = False
class_name = None
class_start = 0
brace_count = 0
for i, line in enumerate(lines):
match = re.match(class_pattern, line)
if match:
if in_class and class_name:
# End previous class
class_lines = i - class_start
if class_lines > self.MAX_CLASS_LINES:
god_classes.append({
'file': str(file_path.relative_to(self.project_path)),
'class': class_name,
'lines': class_lines,
})
class_name = match.group(1)
class_start = i
in_class = True
# Check last class
if in_class and class_name:
class_lines = len(lines) - class_start
if class_lines > self.MAX_CLASS_LINES:
god_classes.append({
'file': str(file_path.relative_to(self.project_path)),
'class': class_name,
'lines': class_lines,
})
self.issues.append({
'type': 'god_class',
'severity': 'warning',
'file': str(file_path.relative_to(self.project_path)),
'message': f"Class '{class_name}' has ~{class_lines} lines (threshold: {self.MAX_CLASS_LINES})",
'suggestion': "Consider applying Single Responsibility Principle and splitting into smaller classes",
})
except Exception:
pass
self.metrics['god_classes'] = god_classes
def _check_naming_conventions(self):
"""Check for naming convention issues."""
ignore_dirs = {'.git', 'node_modules', '__pycache__', '.venv', 'venv',
'dist', 'build', '.next', 'coverage'}
naming_issues = []
# Check directory naming
for dir_path in self.project_path.rglob('*'):
if not dir_path.is_dir():
continue
if any(ignored in dir_path.parts for ignored in ignore_dirs):
continue
dir_name = dir_path.name
# Check for mixed case in directories (should be kebab-case or snake_case)
if re.search(r'[A-Z]', dir_name) and '-' not in dir_name and '_' not in dir_name:
rel_path = str(dir_path.relative_to(self.project_path))
if len(rel_path.split('/')) <= 3: # Only check top-level dirs
naming_issues.append({
'type': 'directory',
'path': rel_path,
'issue': 'PascalCase directory name',
})
if naming_issues:
self.issues.append({
'type': 'naming_convention',
'severity': 'info',
'message': f"Found {len(naming_issues)} naming convention inconsistencies",
'details': naming_issues[:5], # Show first 5
})
self.metrics['naming_issues'] = naming_issues
class LayerViolationDetector:
"""Detects architectural layer violations."""
LAYER_ORDER = ['presentation', 'application', 'domain', 'infrastructure']
# Valid dependency directions (key can depend on values)
VALID_DEPENDENCIES = {
'presentation': ['application', 'domain'],
'application': ['domain', 'infrastructure'],
'domain': [], # Domain should not depend on other layers
'infrastructure': ['domain'],
}
def __init__(self, project_path: Path, layer_assignments: Dict[str, str]):
self.project_path = project_path
self.layer_assignments = layer_assignments
self.violations: List[Dict] = []
def detect(self) -> List[Dict]:
"""Detect layer violations."""
self._analyze_imports()
return self.violations
def _analyze_imports(self):
"""Analyze imports for layer violations."""
extensions = ['.py', '.js', '.ts', '.jsx', '.tsx']
ignore_dirs = {'.git', 'node_modules', '__pycache__', '.venv', 'venv',
'dist', 'build', '.next', 'coverage'}
for ext in extensions:
for file_path in self.project_path.rglob(f'*{ext}'):
if any(ignored in file_path.parts for ignored in ignore_dirs):
continue
try:
rel_path = file_path.relative_to(self.project_path)
if len(rel_path.parts) < 2:
continue
source_dir = rel_path.parts[0].lower()
source_layer = self.layer_assignments.get(source_dir)
if not source_layer or source_layer == 'unknown':
continue
# Extract imports
content = file_path.read_text(encoding='utf-8', errors='ignore')
imports = self._extract_imports(content)
# Check each import for layer violations
for imp in imports:
target_dir = self._get_import_directory(imp)
if not target_dir:
continue
target_layer = self.layer_assignments.get(target_dir.lower())
if not target_layer or target_layer == 'unknown':
continue
if self._is_violation(source_layer, target_layer):
self.violations.append({
'type': 'layer_violation',
'severity': 'warning',
'file': str(rel_path),
'source_layer': source_layer,
'target_layer': target_layer,
'import': imp,
'message': f"{source_layer} layer should not depend on {target_layer} layer",
})
except Exception:
pass
def _extract_imports(self, content: str) -> List[str]:
"""Extract import statements."""
imports = []
# Python imports
imports.extend(re.findall(r'^(?:from|import)\s+([\w.]+)', content, re.MULTILINE))
# JS/TS imports
imports.extend(re.findall(r'(?:import|require)\s*\(?[\'"]([^\'"\s]+)[\'"]', content))
return imports
def _get_import_directory(self, imp: str) -> Optional[str]:
"""Get the directory from an import path."""
# Handle relative imports
if imp.startswith('.'):
return None # Skip relative imports
parts = imp.replace('@/', '').replace('~/', '').split('/')
if parts:
return parts[0].split('.')[0]
return None
def _is_violation(self, source_layer: str, target_layer: str) -> bool:
"""Check if the dependency is a violation."""
if source_layer == target_layer:
return False
valid_deps = self.VALID_DEPENDENCIES.get(source_layer, [])
return target_layer not in valid_deps and target_layer != source_layer
class ProjectArchitect:
"""Main class that orchestrates architecture analysis."""
def __init__(self, project_path: Path, verbose: bool = False):
self.project_path = project_path
self.verbose = verbose
def analyze(self) -> Dict:
"""Run full architecture analysis."""
if self.verbose:
print(f"Analyzing project: {self.project_path}")
# Pattern detection
pattern_detector = PatternDetector(self.project_path)
pattern_result = pattern_detector.scan()
if self.verbose:
print(f"Detected pattern: {pattern_result['detected_pattern']} "
f"(confidence: {pattern_result['confidence']}%)")
# Code analysis
code_analyzer = CodeAnalyzer(self.project_path, self.verbose)
code_result = code_analyzer.analyze()
if self.verbose:
print(f"Found {len(code_result['issues'])} code issues")
# Layer violation detection
violation_detector = LayerViolationDetector(
self.project_path,
pattern_result['layer_assignments']
)
violations = violation_detector.detect()
if self.verbose:
print(f"Found {len(violations)} layer violations")
# Generate recommendations
recommendations = self._generate_recommendations(
pattern_result, code_result, violations
)
return {
'project_path': str(self.project_path),
'architecture': {
'detected_pattern': pattern_result['detected_pattern'],
'confidence': pattern_result['confidence'],
'layer_assignments': pattern_result['layer_assignments'],
'pattern_scores': pattern_result['pattern_scores'],
},
'structure': {
'directories': pattern_result['directories'],
},
'code_quality': {
'metrics': code_result['metrics'],
'issues': code_result['issues'],
},
'layer_violations': violations,
'recommendations': recommendations,
'summary': {
'pattern': pattern_result['detected_pattern'],
'confidence': pattern_result['confidence'],
'total_issues': len(code_result['issues']) + len(violations),
'code_issues': len(code_result['issues']),
'layer_violations': len(violations),
},
}
def _generate_recommendations(self, pattern_result: Dict, code_result: Dict,
violations: List[Dict]) -> List[str]:
"""Generate actionable recommendations."""
recommendations = []
# Pattern recommendations
pattern = pattern_result['detected_pattern']
confidence = pattern_result['confidence']
if pattern == 'unstructured' or confidence < 30:
recommendations.append(
"Consider adopting a clear architectural pattern (Layered, Clean, or Hexagonal) "
"to improve code organization and maintainability"
)
# Layer violation recommendations
if violations:
recommendations.append(
f"Fix {len(violations)} layer violation(s) to maintain proper separation of concerns. "
"Dependencies should flow from presentation → application → domain ← infrastructure"
)
# God class recommendations
god_classes = code_result['metrics'].get('god_classes', [])
if god_classes:
recommendations.append(
f"Split {len(god_classes)} large class(es) into smaller, focused classes "
"following the Single Responsibility Principle"
)
# Large file recommendations
large_files = code_result['metrics'].get('large_files', [])
if large_files:
recommendations.append(
f"Consider refactoring {len(large_files)} large file(s) into smaller modules"
)
# Missing layer recommendations
assigned_layers = set(pattern_result['layer_assignments'].values())
if pattern in ['layered', 'clean', 'hexagonal']:
expected_layers = {'presentation', 'application', 'domain', 'infrastructure'}
missing = expected_layers - assigned_layers - {'unknown'}
if missing:
recommendations.append(
f"Consider adding missing architectural layer(s): {', '.join(missing)}"
)
return recommendations
def print_human_report(report: Dict):
"""Print human-readable report."""
print("\n" + "=" * 60)
print("ARCHITECTURE ASSESSMENT")
print("=" * 60)
print(f"\nProject: {report['project_path']}")
arch = report['architecture']
print(f"\n--- Architecture Pattern ---")
print(f"Detected: {arch['detected_pattern'].replace('_', ' ').title()}")
print(f"Confidence: {arch['confidence']}%")
if arch['layer_assignments']:
print(f"\nLayer Assignments:")
for dir_name, layer in sorted(arch['layer_assignments'].items()):
if layer != 'unknown':
status = "OK"
else:
status = "?"
print(f" {status} {dir_name:20} -> {layer}")
summary = report['summary']
print(f"\n--- Summary ---")
print(f"Total issues: {summary['total_issues']}")
print(f" Code issues: {summary['code_issues']}")
print(f" Layer violations: {summary['layer_violations']}")
if report['code_quality']['issues']:
print(f"\n--- Code Issues ---")
for issue in report['code_quality']['issues'][:10]:
severity = issue['severity'].upper()
print(f" [{severity}] {issue.get('file', 'N/A')}")
print(f" {issue['message']}")
if 'suggestion' in issue:
print(f" Suggestion: {issue['suggestion']}")
if report['layer_violations']:
print(f"\n--- Layer Violations ---")
for v in report['layer_violations'][:5]:
print(f" {v['file']}")
print(f" {v['message']}")
if report['recommendations']:
print(f"\n--- Recommendations ---")
for i, rec in enumerate(report['recommendations'], 1):
print(f" {i}. {rec}")
metrics = report['code_quality']['metrics']
print(f"\n--- Metrics ---")
print(f" Total lines: {metrics.get('total_lines', 'N/A')}")
print(f" File count: {metrics.get('file_count', 'N/A')}")
print(f" Avg lines/file: {metrics.get('avg_file_lines', 'N/A')}")
print("\n" + "=" * 60)
def main():
parser = argparse.ArgumentParser(
description='Analyze project architecture and detect patterns and issues',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog='''
Examples:
%(prog)s ./my-project
%(prog)s ./my-project --verbose
%(prog)s ./my-project --output json
%(prog)s ./my-project --check layers
Detects:
- Architectural patterns (Layered, MVC, Hexagonal, Clean, Microservices)
- Code organization issues (large files, god classes)
- Layer violations (incorrect dependencies between layers)
- Missing architectural components
'''
)
parser.add_argument(
'project_path',
help='Path to the project directory'
)
parser.add_argument(
'--output', '-o',
choices=['human', 'json'],
default='human',
help='Output format (default: human)'
)
parser.add_argument(
'--check',
choices=['all', 'pattern', 'layers', 'code'],
default='all',
help='What to check (default: all)'
)
parser.add_argument(
'--verbose', '-v',
action='store_true',
help='Enable verbose output'
)
parser.add_argument(
'--save', '-s',
help='Save report to file'
)
args = parser.parse_args()
project_path = Path(args.project_path).resolve()
if not project_path.exists():
print(f"Error: Project path does not exist: {project_path}", file=sys.stderr)
sys.exit(1)
if not project_path.is_dir():
print(f"Error: Project path is not a directory: {project_path}", file=sys.stderr)
sys.exit(1)
# Run analysis
architect = ProjectArchitect(project_path, verbose=args.verbose)
report = architect.analyze()
# Handle specific checks
if args.check == 'pattern':
arch = report['architecture']
print(f"Pattern: {arch['detected_pattern']} (confidence: {arch['confidence']}%)")
sys.exit(0)
elif args.check == 'layers':
violations = report['layer_violations']
if violations:
print(f"Found {len(violations)} layer violation(s):")
for v in violations:
print(f" {v['file']}: {v['message']}")
sys.exit(1)
else:
print("No layer violations found.")
sys.exit(0)
elif args.check == 'code':
issues = report['code_quality']['issues']
if issues:
print(f"Found {len(issues)} code issue(s):")
for issue in issues[:10]:
print(f" [{issue['severity'].upper()}] {issue['message']}")
sys.exit(1 if any(i['severity'] == 'warning' for i in issues) else 0)
else:
print("No code issues found.")
sys.exit(0)
# Output report
if args.output == 'json':
output = json.dumps(report, indent=2)
if args.save:
Path(args.save).write_text(output)
print(f"Report saved to {args.save}")
else:
print(output)
else:
print_human_report(report)
if args.save:
Path(args.save).write_text(json.dumps(report, indent=2))
print(f"\nJSON report saved to {args.save}")
if __name__ == '__main__':
main()