|
| 1 | +import os |
1 | 2 | from datetime import date, timezone |
2 | | -from typing import Any, Dict, Optional |
| 3 | +from pathlib import Path |
| 4 | +from typing import Any, Dict, Optional, Tuple |
3 | 5 |
|
| 6 | +from git_analytics.analyzers import bus_factor_post |
4 | 7 | from git_analytics.entities import AnalyticsResult |
5 | 8 | from git_analytics.interfaces import CommitSource |
6 | 9 |
|
7 | 10 |
|
| 11 | +class FileAnalyticsEngine: |
| 12 | + def __init__(self, repo_path: str = "."): |
| 13 | + self.repo_path = repo_path |
| 14 | + |
| 15 | + def run(self) -> Dict[str, Tuple[int, int]]: |
| 16 | + stats: Dict[str, Tuple[int, int]] = {} |
| 17 | + |
| 18 | + repo_path = Path(self.repo_path).resolve() |
| 19 | + |
| 20 | + ignore_dirs = { |
| 21 | + ".git", |
| 22 | + "__pycache__", |
| 23 | + "node_modules", |
| 24 | + ".venv", |
| 25 | + "venv", |
| 26 | + "htmlcov", |
| 27 | + ".pytest_cache", |
| 28 | + ".mypy_cache", |
| 29 | + "dist", |
| 30 | + "build", |
| 31 | + } |
| 32 | + |
| 33 | + for root, dirs, files in os.walk(repo_path): |
| 34 | + dirs[:] = [d for d in dirs if d not in ignore_dirs] |
| 35 | + |
| 36 | + for file in files: |
| 37 | + file_path = Path(root) / file |
| 38 | + |
| 39 | + extension = file_path.suffix if file_path.suffix else "(no extension)" |
| 40 | + |
| 41 | + try: |
| 42 | + with open(file_path, "r", encoding="utf-8") as f: |
| 43 | + line_count = sum(1 for _ in f) |
| 44 | + except (UnicodeDecodeError, PermissionError, OSError): |
| 45 | + line_count = 0 |
| 46 | + |
| 47 | + if extension in stats: |
| 48 | + file_count, total_lines = stats[extension] |
| 49 | + stats[extension] = (file_count + 1, total_lines + line_count) |
| 50 | + else: |
| 51 | + stats[extension] = (1, line_count) |
| 52 | + |
| 53 | + return dict(sorted(stats.items())) |
| 54 | + |
| 55 | + |
8 | 56 | class CommitAnalyticsEngine: |
9 | 57 | def __init__( |
10 | 58 | self, |
@@ -39,6 +87,11 @@ def run( |
39 | 87 | analyzer.process(commit) |
40 | 88 |
|
41 | 89 | result = {analyzer.name: analyzer.result() for analyzer in analyzers} |
| 90 | + |
| 91 | + # post-process |
| 92 | + result["post_data"] = {} |
| 93 | + result["post_data"]["bus_factor"] = bus_factor_post(result) |
| 94 | + |
42 | 95 | if self._additional_data: |
43 | 96 | result["additional_data"] = self._additional_data |
44 | 97 | return result |
0 commit comments