Files
Aestimare/generators/report_generator.py
james.maes 80729108c7 Add developer activity analysis across all repositories
- Collect developer activity data for past 12 months:
  * Commits per developer across all repos
  * Pull requests (created, merged, stats)
  * Issues (created, closed)
  * Lines of code (added, deleted, net)
  * Repository and organization participation

- Generate comprehensive developer reports:
  * Individual developer reports showing all activity
  * Summary reports with rankings (most/least productive)
  * Activity breakdown by repository
  * Statistics and metrics

- Rankings by:
  * Commits
  * Pull requests
  * Total contributions
  * Lines of code
  * Repository count

- Reports include:
  * Per-developer detailed activity
  * Top/bottom performers
  * Activity statistics
  * Repository participation
2025-12-14 16:11:36 -06:00

1362 lines
48 KiB
Python

#!/usr/bin/env python3
"""
Report Generator - Generates markdown reports from assessment data.
Creates a structured report directory similar to the existing assessment format.
"""
import json
import os
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Any
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class ReportGenerator:
"""Generates assessment reports in markdown format."""
def __init__(self, config: Dict):
self.config = config
self.output_dir = Path(config.get('output_dir', 'reports'))
self.assessment_name = config.get('assessment', {}).get('name', 'Aestimare Assessment')
self.organization = config.get('assessment', {}).get('organization', 'Organization')
self.timestamp = datetime.now().strftime('%Y-%m-%d')
def generate_all_reports(self, executive_summary: Dict, security: Dict,
infrastructure: Dict, cost: Dict, repos: List[Dict],
developer_activity: Dict, aws_data: Dict, github_data: Dict):
"""Generate all reports from assessment data."""
logger.info(f"Generating reports to: {self.output_dir}")
# Create directory structure
self._create_directory_structure()
# Generate reports
self.generate_executive_summary(executive_summary)
self.generate_security_report(security)
self.generate_security_issues(security)
self.generate_infrastructure_report(infrastructure, aws_data)
self.generate_cost_report(cost)
self.generate_repository_reports(repos)
self.generate_developer_activity_reports(developer_activity)
self.generate_appendices(aws_data, github_data)
logger.info("Report generation complete")
def _create_directory_structure(self):
"""Create the report directory structure."""
dirs = [
'Security/Issues',
'Infrastructure/EC2',
'Infrastructure/RDS',
'Infrastructure/Lambda',
'Infrastructure/S3',
'Code-Repositories/Repos',
'Code-Repositories/Product-Lines',
'Cost-Analysis',
'Product-Lines',
'Developer-Activity/Developers',
'Developer-Activity/Summaries',
'Appendices',
]
for dir_path in dirs:
(self.output_dir / dir_path).mkdir(parents=True, exist_ok=True)
def generate_executive_summary(self, summary: Dict):
"""Generate the executive summary report."""
md = f"""# {self.assessment_name}
**Organization:** {self.organization}
**Assessment Date:** {self.timestamp}
**Generated By:** Aestimare
---
## Executive Summary
{summary.get('narrative', 'Assessment summary not available.')}
---
## Overall Scores
| Category | Score | Rating |
|----------|-------|--------|
| **Overall** | {summary.get('overall_score', 'N/A')}/100 | {summary.get('overall_rating', 'N/A')} |
| **Infrastructure** | {summary.get('infrastructure_score', 'N/A')}/100 | |
| **Security** | {summary.get('security_score', 'N/A')}/100 | |
| **Code Quality** | {summary.get('code_score', 'N/A')}/100 | |
---
## Risk Summary
| Severity | Count |
|----------|-------|
| Critical | {summary.get('risks', {}).get('critical', 0)} |
| High | {summary.get('risks', {}).get('high', 0)} |
| Medium | {summary.get('risks', {}).get('medium', 0)} |
| Low | {summary.get('risks', {}).get('low', 0)} |
---
## Key Findings
"""
for finding in summary.get('key_findings', []):
md += f"- {finding}\n"
md += """
---
## Critical Issues Requiring Immediate Attention
"""
for issue in summary.get('critical_issues', []):
md += f"1. **{issue}**\n"
md += """
---
## Strengths
"""
for strength in summary.get('strengths', []):
md += f"- {strength}\n"
md += """
---
## Cost Summary
"""
cost_summary = summary.get('cost_summary', {})
md += f"""| Metric | Value |
|--------|-------|
| Monthly Cost | ${cost_summary.get('monthly_cost', 0):,.2f} |
| Potential Savings | ${cost_summary.get('potential_savings', 0):,.2f}/month |
| Optimization Score | {cost_summary.get('optimization_score', 'N/A')}/100 |
---
## Recommendations
### Immediate (24-48 hours)
"""
for rec in summary.get('recommendations', {}).get('immediate', []):
md += f"1. {rec}\n"
md += """
### Week 1
"""
for rec in summary.get('recommendations', {}).get('week_1', []):
md += f"1. {rec}\n"
md += """
### Month 1
"""
for rec in summary.get('recommendations', {}).get('month_1', []):
md += f"1. {rec}\n"
md += """
### Quarter 1
"""
for rec in summary.get('recommendations', {}).get('quarter_1', []):
md += f"1. {rec}\n"
md += """
---
## Related Documents
- [[Security/Security-Assessment|Security Assessment]]
- [[Infrastructure/Overview|Infrastructure Overview]]
- [[Cost-Analysis/Overview|Cost Analysis]]
- [[Code-Repositories/Overall-Assessment|Code Repository Assessment]]
"""
self._write_file('Executive-Summary.md', md)
def generate_security_report(self, security: Dict):
"""Generate the security assessment report."""
summary = security.get('summary', {})
md = f"""# Security Assessment
**Assessment Date:** {self.timestamp}
**Overall Security Score:** {summary.get('overall_score', 'N/A')}/100
**Risk Level:** {summary.get('risk_level', 'N/A')}
---
## Summary
| Severity | Count |
|----------|-------|
| Critical | {summary.get('critical', 0)} |
| High | {summary.get('high', 0)} |
| Medium | {summary.get('medium', 0)} |
| Low | {summary.get('low', 0)} |
| **Total** | **{summary.get('total_issues', 0)}** |
---
## Security Issues
"""
# Group issues by severity
issues = security.get('issues', [])
for severity in ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW']:
severity_issues = [i for i in issues if i.get('severity') == severity]
if severity_issues:
md += f"### {severity} ({len(severity_issues)})\n\n"
for issue in severity_issues:
md += f"- **[{issue.get('id')}]** [[Security/Issues/{issue.get('id')}|{issue.get('title')}]]\n"
md += "\n"
md += """
---
## Positive Findings
"""
for finding in security.get('positive_findings', []):
md += f"- {finding}\n"
md += """
---
## Recommendations
### Immediate Actions
"""
for rec in security.get('recommendations', {}).get('immediate', []):
md += f"1. {rec}\n"
md += """
### Short-term (1 week)
"""
for rec in security.get('recommendations', {}).get('short_term', []):
md += f"1. {rec}\n"
md += """
### Long-term
"""
for rec in security.get('recommendations', {}).get('long_term', []):
md += f"1. {rec}\n"
md += """
---
## Related Documents
- [[Executive-Summary|Executive Summary]]
- [[Infrastructure/VPC-Network|Network Security]]
- [[Infrastructure/IAM|IAM Configuration]]
"""
self._write_file('Security/Security-Assessment.md', md)
def generate_security_issues(self, security: Dict):
"""Generate individual security issue files."""
issues = security.get('issues', [])
for issue in issues:
issue_id = issue.get('id', 'SEC-000')
title = issue.get('title', 'Unknown Issue')
md = f"""# {issue_id}: {title}
**Severity:** {issue.get('severity', 'N/A')}
**Category:** {issue.get('category', 'N/A')}
**Resource Type:** {issue.get('resource_type', 'N/A')}
---
## Description
{issue.get('description', 'No description available.')}
---
## Risk
{issue.get('risk', 'Risk not assessed.')}
---
## Affected Resources
"""
for resource in issue.get('affected_resources', []):
md += f"- `{resource}`\n"
md += f"""
---
## Evidence
{issue.get('evidence', 'No evidence provided.')}
---
## Remediation
{issue.get('remediation', 'No remediation guidance provided.')}
---
## References
"""
for ref in issue.get('references', []):
md += f"- {ref}\n"
md += """
---
## Related Documents
- [[../Security-Assessment|Security Assessment]]
"""
self._write_file(f'Security/Issues/{issue_id}.md', md)
def generate_infrastructure_report(self, infrastructure: Dict, aws_data: Dict):
"""Generate infrastructure overview report."""
summary = infrastructure.get('summary', {})
inventory = infrastructure.get('inventory', {})
md = f"""# Infrastructure Overview
**Assessment Date:** {self.timestamp}
---
## Summary Scores
| Category | Score |
|----------|-------|
| Architecture | {summary.get('architecture_score', 'N/A')}/100 |
| Reliability | {summary.get('reliability_score', 'N/A')}/100 |
| Performance | {summary.get('performance_score', 'N/A')}/100 |
| Operational Excellence | {summary.get('operational_score', 'N/A')}/100 |
| **Overall** | **{summary.get('overall_score', 'N/A')}/100** |
---
## Resource Inventory
### Compute
| Resource | Count |
|----------|-------|
| EC2 Instances | {inventory.get('ec2_instances', {}).get('total', 0)} |
| - Running | {inventory.get('ec2_instances', {}).get('running', 0)} |
| - Stopped | {inventory.get('ec2_instances', {}).get('stopped', 0)} |
| Lambda Functions | {inventory.get('lambda_functions', {}).get('total', 0)} |
### Database
| Resource | Count |
|----------|-------|
| RDS Instances | {inventory.get('rds_instances', {}).get('total', 0)} |
| DynamoDB Tables | {inventory.get('dynamodb_tables', 0) if isinstance(inventory.get('dynamodb_tables'), int) else inventory.get('dynamodb_tables', {}).get('total', 0)} |
### Networking
| Resource | Count |
|----------|-------|
| VPCs | {inventory.get('vpcs', 0)} |
| Subnets | {inventory.get('subnets', 0)} |
| S3 Buckets | {inventory.get('s3_buckets', 0)} |
---
## Architecture Review
### Strengths
"""
for strength in infrastructure.get('architecture_review', {}).get('strengths', []):
md += f"- {strength}\n"
md += """
### Weaknesses
"""
for weakness in infrastructure.get('architecture_review', {}).get('weaknesses', []):
md += f"- {weakness}\n"
md += """
### Modernization Opportunities
"""
for opp in infrastructure.get('architecture_review', {}).get('modernization_opportunities', []):
md += f"- {opp}\n"
md += """
---
## Infrastructure Findings
"""
findings = infrastructure.get('findings', [])
for severity in ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW']:
severity_findings = [f for f in findings if f.get('severity') == severity]
if severity_findings:
md += f"### {severity}\n\n"
for finding in severity_findings:
md += f"- **{finding.get('id')}**: {finding.get('title')}\n"
md += f" - Category: {finding.get('category')}\n"
md += f" - Recommendation: {finding.get('recommendation')}\n\n"
md += """
---
## Deprecated Resources
"""
for dep in infrastructure.get('deprecated_resources', []):
md += f"- **{dep.get('resource')}**: {dep.get('issue')}\n"
md += f" - Migration: {dep.get('migration_path')}\n"
md += """
---
## Related Documents
- [[Executive-Summary|Executive Summary]]
- [[EC2/Overview|EC2 Instances]]
- [[RDS/Overview|RDS Databases]]
- [[Lambda/Overview|Lambda Functions]]
- [[VPC-Network|VPC Networking]]
"""
self._write_file('Infrastructure/Overview.md', md)
def generate_cost_report(self, cost: Dict):
"""Generate cost analysis report."""
summary = cost.get('summary', {})
md = f"""# Cost Analysis
**Assessment Date:** {self.timestamp}
---
## Summary
| Metric | Value |
|--------|-------|
| Monthly Cost | ${summary.get('monthly_cost', 0):,.2f} |
| Projected Annual Cost | ${summary.get('projected_annual_cost', 0):,.2f} |
| Potential Monthly Savings | ${summary.get('potential_monthly_savings', 0):,.2f} |
| Potential Annual Savings | ${summary.get('potential_annual_savings', 0):,.2f} |
| Savings Percentage | {summary.get('savings_percentage', 0):.1f}% |
| Cost Efficiency Score | {summary.get('cost_efficiency_score', 'N/A')}/100 |
---
## Cost Breakdown by Service
| Service | Monthly Cost | % of Total |
|---------|-------------|------------|
"""
for service, data in cost.get('cost_breakdown', {}).get('by_service', {}).items():
if isinstance(data, dict):
md += f"| {service} | ${data.get('cost', 0):,.2f} | {data.get('percentage', 0):.1f}% |\n"
md += """
---
## Savings Opportunities
"""
for opp in cost.get('savings_opportunities', []):
md += f"""### {opp.get('id')}: {opp.get('title')}
- **Category:** {opp.get('category')}
- **Priority:** {opp.get('priority')}
- **Estimated Monthly Savings:** ${opp.get('estimated_monthly_savings', 0):,.2f}
- **Estimated Annual Savings:** ${opp.get('estimated_annual_savings', 0):,.2f}
- **Implementation Effort:** {opp.get('implementation_effort')}
**Current State:** {opp.get('current_state', 'N/A')}
**Recommended Action:** {opp.get('recommended_action', 'N/A')}
---
"""
md += """
## Commitment Analysis
### Reserved Instances
"""
ri = cost.get('commitment_analysis', {}).get('reserved_instances', {})
md += f"""- Active: {ri.get('active', 0)}
- Expired: {ri.get('expired', 0)}
- Utilization: {ri.get('utilization', 'N/A')}%
"""
md += """
### Savings Plans
"""
sp = cost.get('commitment_analysis', {}).get('savings_plans', {})
md += f"""- Active: {sp.get('active', 0)}
- Coverage: {sp.get('coverage', 'N/A')}%
---
## Unused Resources
"""
for resource in cost.get('unused_resources', []):
md += f"- **{resource.get('resource_type')}** `{resource.get('resource_id')}` - ${resource.get('monthly_cost', 0):,.2f}/month\n"
md += f" - Reason: {resource.get('reason')}\n"
md += f" - Recommendation: {resource.get('recommendation')}\n\n"
md += """
---
## Right-Sizing Recommendations
"""
for rec in cost.get('right_sizing', []):
md += f"- **{rec.get('resource_type')}** `{rec.get('resource_id')}`\n"
md += f" - Current: {rec.get('current_size')} -> Recommended: {rec.get('recommended_size')}\n"
md += f" - Savings: ${rec.get('monthly_savings', 0):,.2f}/month\n\n"
md += """
---
## Recommendations
### Quick Wins
"""
for rec in cost.get('recommendations', {}).get('quick_wins', []):
md += f"1. {rec}\n"
md += """
### Short-term
"""
for rec in cost.get('recommendations', {}).get('short_term', []):
md += f"1. {rec}\n"
md += """
### Long-term
"""
for rec in cost.get('recommendations', {}).get('long_term', []):
md += f"1. {rec}\n"
md += """
---
## Related Documents
- [[Executive-Summary|Executive Summary]]
- [[Infrastructure/Overview|Infrastructure Overview]]
"""
self._write_file('Cost-Analysis/Overview.md', md)
def generate_repository_reports(self, repos: List[Dict]):
"""Generate code repository reports."""
# Generate overall assessment
self._generate_repo_overall(repos)
# Generate individual repo reports
for repo in repos:
if 'error' not in repo:
self._generate_repo_individual(repo)
def _generate_repo_overall(self, repos: List[Dict]):
"""Generate overall repository assessment."""
valid_repos = [r for r in repos if 'error' not in r and 'summary' in r]
if not valid_repos:
avg_score = 0
total_loc = 0
else:
avg_score = sum(r['summary'].get('overall_score', 0) for r in valid_repos) / len(valid_repos)
total_loc = sum(r.get('metrics', {}).get('lines_of_code', 0) for r in valid_repos)
md = f"""# Code Repository Assessment
**Assessment Date:** {self.timestamp}
**Total Repositories:** {len(repos)}
**Average Score:** {avg_score:.1f}/100
---
## Summary
| Metric | Value |
|--------|-------|
| Total Repositories | {len(repos)} |
| Assessed Successfully | {len(valid_repos)} |
| Total Lines of Code | {total_loc:,} |
| Average Score | {avg_score:.1f}/100 |
---
## Repository Scores
| # | Repository | Overall | Security | Code Quality | Architecture | Performance | DevOps | Activity |
|---|------------|---------|----------|--------------|--------------|-------------|--------|----------|
"""
# Sort by score
sorted_repos = sorted(valid_repos, key=lambda x: x.get('summary', {}).get('overall_score', 0), reverse=True)
for i, repo in enumerate(sorted_repos, 1):
summary = repo.get('summary', {})
name = summary.get('repository_name', repo.get('name', 'Unknown'))
md += f"| {i} | [[Repos/{name}\\|{name}]] | "
md += f"{summary.get('overall_score', 'N/A')} | "
md += f"{summary.get('security_score', 'N/A')} | "
md += f"{summary.get('code_quality_score', 'N/A')} | "
md += f"{summary.get('architecture_score', 'N/A')} | "
md += f"{summary.get('performance_score', 'N/A')} | "
md += f"{summary.get('devops_score', 'N/A')} | "
md += f"{summary.get('activity_level', 'N/A')} |\n"
md += """
---
## Score Distribution
"""
# Count by rating
ratings = {}
for repo in valid_repos:
rating = repo.get('summary', {}).get('rating', 'Unknown')
ratings[rating] = ratings.get(rating, 0) + 1
for rating, count in sorted(ratings.items()):
md += f"- **{rating}:** {count} repositories\n"
md += """
---
## Related Documents
- [[Executive-Summary|Executive Summary]]
- [[Product-Lines/Overview|Product Lines]]
"""
self._write_file('Code-Repositories/Overall-Assessment.md', md)
def _generate_repo_individual(self, repo: Dict):
"""Generate individual repository report with comprehensive assessments."""
summary = repo.get('summary', {})
metrics = repo.get('metrics', {})
name = summary.get('repository_name', repo.get('name', 'Unknown'))
md = f"""# Repository Assessment: {name}
**Assessment Date:** {self.timestamp}
**Overall Score:** {summary.get('overall_score', 'N/A')}/100
**Rating:** {summary.get('rating', 'N/A')}
**Activity Level:** {summary.get('activity_level', 'N/A')}
---
## Scores
| Category | Score |
|----------|-------|
| **Overall** | **{summary.get('overall_score', 'N/A')}/100** |
| Completeness | {summary.get('completeness_score', 'N/A')}/100 |
| Security | {summary.get('security_score', 'N/A')}/100 |
| Code Quality | {summary.get('code_quality_score', 'N/A')}/100 |
| Architecture | {summary.get('architecture_score', 'N/A')}/100 |
| Performance | {summary.get('performance_score', 'N/A')}/100 |
| Maintainability | {summary.get('maintainability_score', 'N/A')}/100 |
| DevOps | {summary.get('devops_score', 'N/A')}/100 |
| Compliance | {summary.get('compliance_score', 'N/A')}/100 |
| Best Practices | {summary.get('best_practices_score', 'N/A')}/100 |
| Coding Standards | {summary.get('coding_standards_score', 'N/A')}/100 |
---
## Metrics
| Metric | Value |
|--------|-------|
| Lines of Code | {metrics.get('lines_of_code', 0):,} |
| Files | {metrics.get('files_count', 0):,} |
| Primary Language | {metrics.get('primary_language', 'Unknown')} |
| Dependencies | {metrics.get('dependencies_count', 0)} |
| Contributors | {metrics.get('contributors_count', 0)} |
| Commits (30d) | {metrics.get('commits_30d', 0)} |
---
## Critical Findings
"""
# Group findings by severity
critical_findings = [f for f in repo.get('findings', []) if f.get('severity') == 'CRITICAL']
high_findings = [f for f in repo.get('findings', []) if f.get('severity') == 'HIGH']
if critical_findings:
md += "### CRITICAL Issues\n\n"
for finding in critical_findings:
md += f"**{finding.get('title')}** [{finding.get('category')}]\n\n"
md += f"{finding.get('description')}\n\n"
md += f"*Recommendation:* {finding.get('recommendation')}\n\n"
md += "---\n\n"
if high_findings:
md += "### HIGH Priority Issues\n\n"
for finding in high_findings[:5]: # Limit to top 5
md += f"**{finding.get('title')}** [{finding.get('category')}]\n\n"
md += f"{finding.get('description')}\n\n"
md += f"*Recommendation:* {finding.get('recommendation')}\n\n"
md += "---\n\n"
# Code Quality Assessment
code_quality = repo.get('code_quality_assessment', {})
if code_quality:
md += "## Code Quality Assessment\n\n"
complexity = code_quality.get('complexity_indicators', {})
tech_debt = code_quality.get('technical_debt', {})
md += f"**Complexity:** {complexity.get('complexity_rating', 'unknown').upper()}\n"
md += f"- Large files: {complexity.get('large_files_count', 0)}\n"
md += f"- High complexity: {'Yes' if complexity.get('high_complexity') else 'No'}\n\n"
md += f"**Technical Debt:** {tech_debt.get('debt_level', 'unknown').upper()}\n"
md += f"- TODO comments: {tech_debt.get('todo_count', 0)}\n"
md += f"- FIXME comments: {tech_debt.get('fixme_count', 0)}\n"
md += f"- HACK comments: {tech_debt.get('hack_count', 0)}\n"
md += f"- Deprecated code: {tech_debt.get('deprecated_count', 0)}\n\n"
if code_quality.get('recommendations'):
md += "**Recommendations:**\n"
for rec in code_quality.get('recommendations', [])[:3]:
md += f"- {rec}\n"
md += "\n---\n\n"
# Security Deep Dive
security = repo.get('security_assessment', {})
if security:
md += "## Security Assessment\n\n"
md += f"- .gitignore: {'✅' if security.get('gitignore_present') else '❌'}\n"
md += f"- .env.example: {'✅' if security.get('env_example_present') else '❌'}\n"
md += f"- Dependency Lock: {'✅' if security.get('dependency_lock_present') else '❌'}\n"
md += f"- Security Policy: {'✅' if security.get('security_policy_present') else '❌'}\n"
md += f"- Dependabot: {'✅ Enabled' if security.get('dependabot_enabled') else '❌ Not Enabled'}\n"
md += f"- Branch Protection: {'✅' if security.get('branch_protection') else '❌'}\n"
md += f"- Vulnerabilities: {security.get('vulnerabilities_count', 0)}\n"
md += f"- Potential Secrets: {security.get('potential_secrets_count', 0)}\n\n"
insecure = security.get('insecure_patterns', {})
if insecure.get('sql_injection_risk') or insecure.get('xss_risk') or insecure.get('eval_usage'):
md += "**⚠️ Security Risks Detected:**\n"
if insecure.get('sql_injection_risk'):
md += "- SQL Injection risk\n"
if insecure.get('xss_risk'):
md += "- XSS risk\n"
if insecure.get('eval_usage'):
md += "- eval() usage detected\n"
if insecure.get('dangerous_functions'):
md += f"- Dangerous functions: {', '.join([f['function'] for f in insecure.get('dangerous_functions', [])[:5]])}\n"
md += "\n---\n\n"
# Performance Assessment
performance = repo.get('performance_assessment', {})
if performance:
md += "## Performance Assessment\n\n"
perf_tests = performance.get('performance_tests', {})
md += f"- Performance Tests: {'✅' if perf_tests.get('has_perf_tests') else '❌'}\n"
monitoring = performance.get('monitoring_tools', {})
if monitoring.get('apm_tools'):
md += f"- APM Tools: {', '.join(monitoring.get('apm_tools', []))}\n"
else:
md += "- APM Tools: None detected\n"
caching = performance.get('caching_indicators', {})
if caching.get('redis_usage') or caching.get('memcached_usage'):
md += f"- Caching: Redis={'✅' if caching.get('redis_usage') else '❌'}, Memcached={'✅' if caching.get('memcached_usage') else '❌'}\n"
if performance.get('recommendations'):
md += "\n**Recommendations:**\n"
for rec in performance.get('recommendations', [])[:3]:
md += f"- {rec}\n"
md += "\n---\n\n"
# Architecture Assessment
architecture = repo.get('architecture_assessment', {})
if architecture:
md += "## Architecture Assessment\n\n"
structure = architecture.get('project_structure', {})
md += f"**Structure Quality:** {structure.get('structure_quality', 'unknown').upper()}\n"
md += f"- Common directories: {len(structure.get('common_directories', []))}\n\n"
patterns = architecture.get('design_patterns', {})
detected_patterns = [k.upper() for k, v in patterns.items() if v]
if detected_patterns:
md += f"**Design Patterns:** {', '.join(detected_patterns)}\n\n"
api = architecture.get('api_design', {})
md += f"**API Design:**\n"
md += f"- OpenAPI/Swagger: {'✅' if api.get('has_openapi') or api.get('has_swagger') else '❌'}\n"
md += f"- GraphQL: {'✅' if api.get('has_graphql') else '❌'}\n"
md += f"- Versioning: {'✅' if api.get('versioning') else '❌'}\n\n"
if architecture.get('recommendations'):
md += "**Recommendations:**\n"
for rec in architecture.get('recommendations', [])[:3]:
md += f"- {rec}\n"
md += "\n---\n\n"
# Maintainability Assessment
maintainability = repo.get('maintainability_assessment', {})
if maintainability:
md += "## Maintainability Assessment\n\n"
comments = maintainability.get('code_comments', {})
md += f"**Code Comments:** {comments.get('comment_quality', 'unknown').upper()}\n"
md += f"- Comment ratio: {comments.get('comment_ratio', 0)}%\n"
md += f"- Docstring ratio: {comments.get('docstring_ratio', 0)}%\n\n"
ownership = maintainability.get('code_ownership', {})
md += f"**Code Ownership:**\n"
md += f"- CODEOWNERS: {'✅' if ownership.get('codeowners_present') else '❌'}\n"
md += f"- Maintainers file: {'✅' if ownership.get('maintainers_file') else '❌'}\n\n"
legacy = maintainability.get('legacy_indicators', {})
if legacy.get('deprecated_libraries'):
md += f"**Legacy Code:** {len(legacy.get('deprecated_libraries', []))} deprecated libraries detected\n\n"
if maintainability.get('recommendations'):
md += "**Recommendations:**\n"
for rec in maintainability.get('recommendations', [])[:3]:
md += f"- {rec}\n"
md += "\n---\n\n"
# DevOps Assessment
devops = repo.get('devops_assessment', {})
if devops:
md += "## DevOps & Operations Assessment\n\n"
iac = devops.get('infrastructure_as_code', {})
iac_tools = [k.upper() for k, v in iac.items() if v]
if iac_tools:
md += f"**Infrastructure as Code:** {', '.join(iac_tools)}\n"
else:
md += "**Infrastructure as Code:** None detected\n"
deployment = devops.get('deployment_config', {})
md += f"**Deployment:**\n"
md += f"- Dockerfile: {'✅' if deployment.get('dockerfile') else '❌'}\n"
md += f"- Kubernetes: {'✅' if deployment.get('kubernetes') else '❌'}\n"
md += f"- Helm: {'✅' if deployment.get('helm') else '❌'}\n\n"
monitoring = devops.get('monitoring_logging', {})
md += f"**Monitoring:**\n"
md += f"- Structured logging: {'✅' if monitoring.get('structured_logging') else '❌'}\n"
md += f"- Health checks: {'✅' if devops.get('health_checks', {}).get('has_health_endpoint') else '❌'}\n\n"
if devops.get('recommendations'):
md += "**Recommendations:**\n"
for rec in devops.get('recommendations', [])[:3]:
md += f"- {rec}\n"
md += "\n---\n\n"
# Compliance Assessment
compliance = repo.get('compliance_assessment', {})
if compliance:
md += "## Compliance Assessment\n\n"
licenses = compliance.get('licenses', {})
md += f"**Licenses:**\n"
md += f"- Main license: {licenses.get('main_license', 'Unknown')}\n"
md += f"- License file: {'✅' if licenses.get('license_file') else '❌'}\n"
md += f"- Compliance: {licenses.get('compliance_status', 'unknown')}\n\n"
privacy = compliance.get('privacy_compliance', {})
md += f"**Privacy:**\n"
md += f"- Privacy policy: {'✅' if privacy.get('privacy_policy') else '❌'}\n"
md += f"- GDPR indicators: {'✅' if privacy.get('gdpr_indicators') else '❌'}\n\n"
a11y = compliance.get('accessibility', {})
md += f"**Accessibility:**\n"
md += f"- A11y testing: {'✅' if a11y.get('a11y_testing') else '❌'}\n"
if a11y.get('a11y_libraries'):
md += f"- A11y libraries: {', '.join(a11y.get('a11y_libraries', []))}\n"
md += "\n---\n\n"
# Testing Assessment
test = repo.get('testing_assessment', {})
if test:
md += "## Testing Assessment\n\n"
md += f"- Has Tests: {'✅' if test.get('has_tests') else '❌'}\n"
test_types = test.get('test_types', {})
md += f"**Test Types:**\n"
md += f"- Unit tests: {'✅' if test_types.get('unit_tests') else '❌'}\n"
md += f"- Integration tests: {'✅' if test_types.get('integration_tests') else '❌'}\n"
md += f"- E2E tests: {'✅' if test_types.get('e2e_tests') else '❌'}\n"
md += f"- Performance tests: {'✅' if test_types.get('performance_tests') else '❌'}\n\n"
frameworks = test.get('test_frameworks', [])
if frameworks:
md += f"**Frameworks:** {', '.join(frameworks)}\n\n"
md += f"- CI/CD: {', '.join(test.get('ci_cd_systems', [])) or 'None'}\n"
md += f"- Coverage tools: {'✅' if test.get('test_coverage_tools', {}).get('coverage_tools') else '❌'}\n\n"
if test.get('recommendations'):
md += "**Recommendations:**\n"
for rec in test.get('recommendations', [])[:3]:
md += f"- {rec}\n"
md += "\n---\n\n"
# Developer Experience
dev_ex = repo.get('developer_experience_assessment', {})
if dev_ex:
md += "## Developer Experience\n\n"
onboarding = dev_ex.get('onboarding', {})
md += f"**Onboarding:**\n"
md += f"- Getting started guide: {'✅' if onboarding.get('getting_started') else '❌'}\n"
md += f"- Setup instructions: {'✅' if onboarding.get('setup_instructions') else '❌'}\n\n"
dev_env = dev_ex.get('dev_environment', {})
md += f"**Dev Environment:**\n"
md += f"- Docker dev: {'✅' if dev_env.get('docker_dev') else '❌'}\n"
md += f"- DevContainer: {'✅' if dev_env.get('devcontainer') else '❌'}\n"
md += f"- Setup scripts: {'✅' if dev_env.get('setup_scripts') else '❌'}\n\n"
md += f"- Pre-commit hooks: {'✅' if dev_ex.get('pre_commit_hooks', {}).get('husky') or dev_ex.get('pre_commit_hooks', {}).get('pre_commit') else '❌'}\n"
md += "\n---\n\n"
# Code Review
code_review = repo.get('code_review_assessment', {})
if code_review:
md += "## Code Review Practices\n\n"
md += f"- PR template: {'✅' if code_review.get('pr_template') else '❌'}\n"
md += f"- Issue templates: {'✅' if code_review.get('issue_templates') else '❌'}\n"
md += f"- Review requirements: {'✅' if code_review.get('review_requirements', {}).get('review_docs') else '❌'}\n"
md += "\n---\n\n"
# Error Handling
error_handling = repo.get('error_handling_assessment', {})
if error_handling:
md += "## Error Handling\n\n"
md += f"**Quality:** {error_handling.get('error_handling_quality', 'unknown').upper()}\n"
md += f"- Try-catch usage: {'✅' if error_handling.get('try_catch_usage') else '❌'}\n"
md += f"- Error boundaries: {'✅' if error_handling.get('error_boundaries') else '❌'}\n"
md += f"- Logging: {'✅' if error_handling.get('logging_present') else '❌'}\n"
if error_handling.get('recommendations'):
md += "\n**Recommendations:**\n"
for rec in error_handling.get('recommendations', [])[:3]:
md += f"- {rec}\n"
md += "\n---\n\n"
# Documentation
doc = repo.get('documentation_assessment', {})
if doc:
md += "## Documentation\n\n"
md += f"- README: {doc.get('readme_quality', 'N/A')}\n"
api_docs = doc.get('api_docs', {})
if isinstance(api_docs, dict):
md += f"- API Documentation: OpenAPI={'✅' if api_docs.get('openapi') else '❌'}, Swagger={'✅' if api_docs.get('swagger') else '❌'}, GraphQL={'✅' if api_docs.get('graphql_schema') else '❌'}\n"
else:
md += f"- API Documentation: {'Yes' if api_docs else 'No'}\n"
md += f"- Code examples: {'✅' if doc.get('code_examples', {}).get('examples_dir') or doc.get('code_examples', {}).get('usage_examples') else '❌'}\n"
md += f"- ADR: {'✅' if doc.get('adr_present') else '❌'}\n"
md += f"- Runbooks: {'✅' if doc.get('runbooks') else '❌'}\n"
md += f"- Contributing Guide: {'✅' if doc.get('contributing_guide') else '❌'}\n"
md += f"- Changelog: {'✅' if doc.get('changelog') else '❌'}\n"
md += f"- License: {doc.get('license', 'Missing')}\n"
md += "\n---\n\n"
# Dependencies
deps = repo.get('dependencies_assessment', {})
if deps:
md += "## Dependencies\n\n"
md += f"- Outdated: {deps.get('outdated_dependencies', 0)}\n"
md += f"- Security vulnerabilities: {deps.get('security_vulnerabilities', 0)}\n"
if deps.get('deprecated_packages'):
md += f"- Deprecated: {', '.join(deps.get('deprecated_packages', [])[:5])}\n"
md += "\n---\n\n"
# Coding Standards
coding_standards = repo.get('coding_standards_assessment', {})
if coding_standards:
md += "## Coding Standards\n\n"
md += f"- Linter: {'✅' if coding_standards.get('linter_configured') else '❌'}\n"
md += f"- Formatter: {'✅' if coding_standards.get('formatter_configured') else '❌'}\n"
md += f"- TypeScript: {'✅' if coding_standards.get('typescript_configured') else '❌'}\n"
md += f"- EditorConfig: {'✅' if coding_standards.get('editorconfig_present') else '❌'}\n"
md += "\n---\n\n"
# All Findings
all_findings = repo.get('findings', [])
if all_findings:
md += "## All Findings\n\n"
for finding in all_findings:
severity = finding.get('severity', 'INFO')
md += f"### [{severity}] {finding.get('title')}\n\n"
md += f"**Category:** {finding.get('category')}\n\n"
md += f"{finding.get('description')}\n\n"
md += f"**Impact:** {finding.get('impact', 'N/A')}\n\n"
md += f"**Recommendation:** {finding.get('recommendation')}\n\n"
md += f"**Effort:** {finding.get('effort', 'N/A')}\n\n"
md += "---\n\n"
# Recommendations
recommendations = repo.get('recommendations', {})
if recommendations:
md += "## Recommendations\n\n"
if recommendations.get('immediate'):
md += "### Immediate (24-48 hours)\n\n"
for rec in recommendations.get('immediate', []):
md += f"1. {rec}\n"
md += "\n"
if recommendations.get('short_term'):
md += "### Short-term (1-4 weeks)\n\n"
for rec in recommendations.get('short_term', []):
md += f"1. {rec}\n"
md += "\n"
if recommendations.get('long_term'):
md += "### Long-term (Strategic)\n\n"
for rec in recommendations.get('long_term', []):
md += f"1. {rec}\n"
self._write_file(f'Code-Repositories/Repos/{name}.md', md)
def generate_appendices(self, aws_data: Dict, github_data: Dict):
"""Generate appendix documents."""
# Data reference
md = f"""# Data Reference
**Assessment Date:** {self.timestamp}
This document references the raw data files collected during the assessment.
---
## AWS Data
"""
for account_name in aws_data.keys():
md += f"### {account_name}\n\n"
md += f"- Full data: `.data/aws/{account_name.lower().replace(' ', '_')}/full_data.json`\n"
md += f"- Regional data: `.data/aws/{account_name.lower().replace(' ', '_')}/[region]/`\n"
md += f"- Global services: `.data/aws/{account_name.lower().replace(' ', '_')}/global/`\n\n"
md += """
## GitHub Data
"""
for org_name in github_data.keys():
md += f"### {org_name}\n\n"
md += f"- Full data: `.data/github/{org_name.lower()}/full_data.json`\n"
md += f"- Repository list: `.data/github/{org_name.lower()}/repo_list.json`\n\n"
md += """
---
## Report Generation
This report was generated using Aestimare.
"""
self._write_file('Appendices/Data-Reference.md', md)
def generate_developer_activity_reports(self, developer_activity: Dict):
"""Generate developer activity reports."""
if not developer_activity or 'error' in developer_activity or not developer_activity.get('developers'):
logger.warning("No developer activity data available - skipping developer reports")
return
# Generate summary report
self._generate_developer_summary(developer_activity)
# Generate individual developer reports
developers = developer_activity.get('developers', {})
for login, dev_data in developers.items():
self._generate_developer_individual(login, dev_data, developer_activity)
def _generate_developer_summary(self, activity: Dict):
"""Generate developer activity summary report."""
rankings = activity.get('rankings', {})
summary = activity.get('summary', {})
md = f"""# Developer Activity Summary
**Assessment Period:** {activity.get('period_start', 'N/A')} to {activity.get('period_end', 'N/A')}
**Total Developers:** {activity.get('total_developers', 0)}
---
## Top Performers
### Most Commits
"""
if rankings.get('by_commits'):
top_commits = rankings['by_commits'][:10]
md += "| Rank | Developer | Commits | Repositories | Lines Added |\n"
md += "|------|-----------|---------|--------------|-------------|\n"
for i, dev in enumerate(top_commits, 1):
md += f"| {i} | [[Developers/{dev['login']}\\|{dev['login']}]] | {dev['total_commits']} | {dev['repo_count']} | {dev['lines_added']:,} |\n"
md += "\n### Most Pull Requests\n\n"
if rankings.get('by_prs'):
top_prs = rankings['by_prs'][:10]
md += "| Rank | Developer | PRs Created | PRs Merged | Merge Rate |\n"
md += "|------|-----------|-------------|------------|------------|\n"
for i, dev in enumerate(top_prs, 1):
merge_rate = (dev['total_prs_merged'] / dev['total_prs_created'] * 100) if dev['total_prs_created'] > 0 else 0
md += f"| {i} | [[Developers/{dev['login']}\\|{dev['login']}]] | {dev['total_prs_created']} | {dev['total_prs_merged']} | {merge_rate:.1f}% |\n"
md += "\n### Most Contributions (Commits + PRs + Issues)\n\n"
if rankings.get('by_contributions'):
top_contrib = rankings['by_contributions'][:10]
md += "| Rank | Developer | Total Contributions | Commits | PRs | Issues |\n"
md += "|------|-----------|---------------------|---------|-----|--------|\n"
for i, dev in enumerate(top_contrib, 1):
md += f"| {i} | [[Developers/{dev['login']}\\|{dev['login']}]] | {dev['total_contributions']} | {dev['total_commits']} | {dev['total_prs_created']} | {dev['total_issues_created']} |\n"
md += "\n### Most Lines of Code Added\n\n"
if rankings.get('by_lines_added'):
top_lines = rankings['by_lines_added'][:10]
md += "| Rank | Developer | Lines Added | Lines Deleted | Net Lines |\n"
md += "|------|-----------|-------------|---------------|-----------|\n"
for i, dev in enumerate(top_lines, 1):
md += f"| {i} | [[Developers/{dev['login']}\\|{dev['login']}]] | {dev['lines_added']:,} | {dev['lines_deleted']:,} | {dev['net_lines']:,} |\n"
md += "\n### Most Active Across Repositories\n\n"
if rankings.get('by_repositories'):
top_repos = rankings['by_repositories'][:10]
md += "| Rank | Developer | Repositories | Organizations |\n"
md += "|------|-----------|--------------|---------------|\n"
for i, dev in enumerate(top_repos, 1):
md += f"| {i} | [[Developers/{dev['login']}\\|{dev['login']}]] | {dev['repo_count']} | {dev['org_count']} |\n"
md += "\n---\n\n## Activity Statistics\n\n"
developers_list = list(activity.get('developers', {}).values())
if developers_list:
total_commits = sum(d['total_commits'] for d in developers_list)
total_prs = sum(d['total_prs_created'] for d in developers_list)
total_issues = sum(d['total_issues_created'] for d in developers_list)
total_lines = sum(d['lines_added'] for d in developers_list)
avg_commits = total_commits / len(developers_list) if developers_list else 0
avg_prs = total_prs / len(developers_list) if developers_list else 0
md += f"| Metric | Total | Average per Developer |\n"
md += f"|--------|-------|----------------------|\n"
md += f"| Commits | {total_commits:,} | {avg_commits:.1f} |\n"
md += f"| Pull Requests | {total_prs:,} | {avg_prs:.1f} |\n"
md += f"| Issues | {total_issues:,} | {total_issues / len(developers_list):.1f} |\n"
md += f"| Lines Added | {total_lines:,} | {total_lines / len(developers_list):,.0f} |\n"
md += "\n---\n\n## Related Documents\n\n"
md += "- [[Developers/|Individual Developer Reports]]\n"
self._write_file('Developer-Activity/Summary.md', md)
def _generate_developer_individual(self, login: str, dev_data: Dict, activity: Dict):
"""Generate individual developer activity report."""
md = f"""# Developer Activity: {login}
**Name:** {dev_data.get('name', 'N/A')}
**Email:** {dev_data.get('email', 'N/A')}
**Assessment Period:** {activity.get('period_start', 'N/A')} to {activity.get('period_end', 'N/A')}
---
## Overview
| Metric | Value |
|--------|-------|
| **Total Contributions** | **{dev_data.get('total_contributions', 0)}** |
| Commits | {dev_data.get('total_commits', 0)} |
| Pull Requests Created | {dev_data.get('total_prs_created', 0)} |
| Pull Requests Merged | {dev_data.get('total_prs_merged', 0)} |
| Issues Created | {dev_data.get('total_issues_created', 0)} |
| Issues Closed | {dev_data.get('total_issues_closed', 0)} |
| Lines Added | {dev_data.get('lines_added', 0):,} |
| Lines Deleted | {dev_data.get('lines_deleted', 0):,} |
| Net Lines | {dev_data.get('net_lines', 0):,} |
| Repositories | {dev_data.get('repo_count', 0)} |
| Organizations | {dev_data.get('org_count', 0)} |
| First Activity | {dev_data.get('first_activity', 'N/A')} |
| Last Activity | {dev_data.get('last_activity', 'N/A')} |
---
## Repositories Worked On
"""
repos = dev_data.get('repositories', [])
if repos:
md += f"**Total:** {len(repos)} repositories\n\n"
for repo in sorted(repos):
md += f"- {repo}\n"
else:
md += "No repositories found.\n"
md += "\n---\n\n## Organizations\n\n"
orgs = dev_data.get('organizations', [])
for org in sorted(orgs):
md += f"- {org}\n"
# Recent commits
commits = dev_data.get('commits', [])
if commits:
md += "\n---\n\n## Recent Commits (Last 20)\n\n"
md += "| Repository | SHA | Message | Date |\n"
md += "|------------|-----|---------|------|\n"
for commit in sorted(commits, key=lambda x: x.get('date', ''), reverse=True)[:20]:
md += f"| {commit.get('repo', 'N/A')} | [{commit.get('sha', '')[:7]}]({commit.get('url', '')}) | {commit.get('message', '')[:60]}... | {commit.get('date', '')[:10]} |\n"
# Pull requests
prs = dev_data.get('pull_requests', [])
if prs:
md += "\n---\n\n## Pull Requests\n\n"
md += f"**Total:** {len(prs)} pull requests\n\n"
md += "| Repository | # | Title | State | Merged | Additions | Deletions |\n"
md += "|------------|---|-------|-------|--------|-----------|-----------|\n"
for pr in sorted(prs, key=lambda x: x.get('created_at', ''), reverse=True)[:30]:
merged = '✅' if pr.get('merged') else '❌'
md += f"| {pr.get('repo', 'N/A')} | [{pr.get('number')}]({pr.get('url', '')}) | {pr.get('title', '')[:40]}... | {pr.get('state', 'N/A')} | {merged} | {pr.get('additions', 0):,} | {pr.get('deletions', 0):,} |\n"
# Issues
issues = dev_data.get('issues', [])
if issues:
md += "\n---\n\n## Issues\n\n"
md += f"**Total:** {len(issues)} issues\n\n"
md += "| Repository | # | Title | State | Created |\n"
md += "|------------|---|-------|-------|---------|\n"
for issue in sorted(issues, key=lambda x: x.get('created_at', ''), reverse=True)[:20]:
md += f"| {issue.get('repo', 'N/A')} | [{issue.get('number')}]({issue.get('url', '')}) | {issue.get('title', '')[:50]}... | {issue.get('state', 'N/A')} | {issue.get('created_at', '')[:10]} |\n"
# Activity breakdown by repository
md += "\n---\n\n## Activity by Repository\n\n"
repo_activity = {}
for commit in commits:
repo = commit.get('repo')
if repo:
if repo not in repo_activity:
repo_activity[repo] = {'commits': 0, 'prs': 0, 'issues': 0}
repo_activity[repo]['commits'] += 1
for pr in prs:
repo = pr.get('repo')
if repo:
if repo not in repo_activity:
repo_activity[repo] = {'commits': 0, 'prs': 0, 'issues': 0}
repo_activity[repo]['prs'] += 1
for issue in issues:
repo = issue.get('repo')
if repo:
if repo not in repo_activity:
repo_activity[repo] = {'commits': 0, 'prs': 0, 'issues': 0}
repo_activity[repo]['issues'] += 1
if repo_activity:
md += "| Repository | Commits | PRs | Issues | Total |\n"
md += "|------------|---------|-----|--------|-------|\n"
for repo, stats in sorted(repo_activity.items(), key=lambda x: sum(x[1].values()), reverse=True):
total = stats['commits'] + stats['prs'] + stats['issues']
md += f"| {repo} | {stats['commits']} | {stats['prs']} | {stats['issues']} | {total} |\n"
md += "\n---\n\n## Related Documents\n\n"
md += "- [[../Summary|Developer Activity Summary]]\n"
self._write_file(f'Developer-Activity/Developers/{login}.md', md)
def _write_file(self, relative_path: str, content: str):
"""Write content to a file."""
file_path = self.output_dir / relative_path
file_path.parent.mkdir(parents=True, exist_ok=True)
with open(file_path, 'w') as f:
f.write(content)
logger.debug(f" Written: {relative_path}")
def main():
"""Test the report generator."""
import yaml
config_path = Path(__file__).parent.parent / 'config' / 'accounts.yaml'
if not config_path.exists():
config_path = Path(__file__).parent.parent / 'config' / 'accounts.example.yaml'
with open(config_path, 'r') as f:
config = yaml.safe_load(f)
generator = ReportGenerator(config)
print("Report Generator initialized successfully")
print(f"Output directory: {generator.output_dir}")
if __name__ == '__main__':
main()