Fusion
92d0d5b891
Using 'git log --all <branch>' creates ambiguous behavior. Now: - Without branch: uses --all to show all commits from all refs - With branch: shows only commits from that specific branch This ensures consistent commit counts between git CLI and API.
383 lines
12 KiB
Python
383 lines
12 KiB
Python
"""Git history extraction utilities for bare/mirror repositories."""
|
|
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class Commit:
|
|
"""Represents a single git commit."""
|
|
|
|
hash: str
|
|
short_hash: str
|
|
parents: list[str]
|
|
author: str
|
|
email: str
|
|
date: str
|
|
timestamp: int
|
|
message: str
|
|
branches: list[str]
|
|
tags: list[str]
|
|
|
|
|
|
@dataclass
|
|
class FileChange:
|
|
"""Represents a changed file in a commit."""
|
|
|
|
path: str
|
|
change_type: str
|
|
insertions: int
|
|
deletions: int
|
|
diff: str
|
|
|
|
|
|
@dataclass
|
|
class CommitDetail(Commit):
|
|
"""Extended commit info with diff."""
|
|
|
|
body: str
|
|
stats: dict[str, int]
|
|
files: list[FileChange]
|
|
|
|
|
|
def _run_git_command(repo_path: str, args: list[str]) -> str:
|
|
"""Execute a git command in the repository directory."""
|
|
result = subprocess.run(
|
|
["git", "-C", repo_path, *args],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"Git command failed: {result.stderr}")
|
|
return result.stdout
|
|
|
|
|
|
def get_commit_history(repo_path: str, branch: str | None = None, limit: int = 100, offset: int = 0) -> dict[str, Any]:
|
|
"""Extract commit history from a git repository.
|
|
|
|
Returns structured data including commits, branches, and graph information.
|
|
"""
|
|
# Get list of branches
|
|
branches_output = _run_git_command(repo_path, ["branch", "-a", "--format=%(refname:short)"])
|
|
branches = [b.strip() for b in branches_output.strip().split("\n") if b.strip()]
|
|
|
|
# Build git log command - use NULL bytes as separators to avoid parsing issues
|
|
log_args = [
|
|
"log",
|
|
"--format=%H%x00%P%x00%an%x00%ae%x00%at%x00%s",
|
|
f"--max-count={limit}",
|
|
f"--skip={offset}",
|
|
]
|
|
if branch:
|
|
log_args.append(branch)
|
|
else:
|
|
log_args.append("--all")
|
|
|
|
log_output = _run_git_command(repo_path, log_args)
|
|
|
|
# Get branch info for each commit
|
|
branch_map = _get_branch_map(repo_path)
|
|
tag_map = _get_tag_map(repo_path)
|
|
|
|
commits = []
|
|
log_lines = log_output.strip().split("\n") if log_output.strip() else []
|
|
|
|
for line in log_lines:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
|
|
parts = line.split("\x00")
|
|
if len(parts) < 6:
|
|
continue
|
|
|
|
commit_hash = parts[0]
|
|
parents = parts[1].split() if parts[1] else []
|
|
|
|
commits.append(
|
|
Commit(
|
|
hash=commit_hash,
|
|
short_hash=commit_hash[:7],
|
|
parents=parents,
|
|
author=parts[2],
|
|
email=parts[3],
|
|
date=parts[4],
|
|
timestamp=int(parts[4]),
|
|
message=parts[5],
|
|
branches=branch_map.get(commit_hash, []),
|
|
tags=tag_map.get(commit_hash, []),
|
|
)
|
|
)
|
|
|
|
# Get total commit count
|
|
count_output = _run_git_command(repo_path, ["rev-list", "--all", "--count"])
|
|
total_commits = int(count_output.strip()) if count_output.strip() else 0
|
|
|
|
# Build graph data and generate graph symbols
|
|
graph_data = _build_graph_data(commits)
|
|
|
|
# Generate simple graph symbols based on parent count
|
|
commit_dicts = []
|
|
for i, commit in enumerate(commits):
|
|
if len(commit.parents) == 0:
|
|
graph_symbol = "○" # Initial commit
|
|
elif len(commit.parents) > 1:
|
|
graph_symbol = "●" # Merge commit
|
|
else:
|
|
graph_symbol = "○" # Regular commit
|
|
|
|
# Simple depth calculation based on merge status
|
|
graph_depth = min(len(commit.parents), 3)
|
|
|
|
commit_dicts.append(_commit_to_dict(commit, graph_symbol, graph_depth))
|
|
|
|
return {
|
|
"commits": commit_dicts,
|
|
"branches": branches,
|
|
"total_commits": total_commits,
|
|
"graph_data": graph_data,
|
|
}
|
|
|
|
|
|
def get_commit_detail(repo_path: str, commit_hash: str) -> dict[str, Any]:
|
|
"""Get detailed information about a specific commit."""
|
|
# Get commit metadata
|
|
format_str = "%H|%P|%an|%ae|%at|%s|%b"
|
|
log_output = _run_git_command(
|
|
repo_path, ["log", "-1", f"--format={format_str}", commit_hash]
|
|
)
|
|
|
|
parts = log_output.strip().split("|", 6)
|
|
if len(parts) < 6:
|
|
raise ValueError(f"Invalid commit: {commit_hash}")
|
|
|
|
commit_hash = parts[0]
|
|
parents = parts[1].split() if parts[1] else []
|
|
author = parts[2]
|
|
email = parts[3]
|
|
timestamp = int(parts[4])
|
|
message = parts[5]
|
|
body = parts[6] if len(parts) > 6 else ""
|
|
|
|
# Get stats
|
|
stat_output = _run_git_command(
|
|
repo_path, ["show", "--stat", "--format=", commit_hash]
|
|
)
|
|
stats = _parse_stats(stat_output)
|
|
|
|
# Get diff
|
|
diff_output = _run_git_command(
|
|
repo_path, ["show", "--format=", commit_hash]
|
|
)
|
|
files = _parse_diff(diff_output)
|
|
|
|
# Get branch/tag info
|
|
branch_map = _get_branch_map(repo_path)
|
|
tag_map = _get_tag_map(repo_path)
|
|
|
|
return {
|
|
"hash": commit_hash,
|
|
"short_hash": commit_hash[:7],
|
|
"parents": parents,
|
|
"author_name": author,
|
|
"author_email": email,
|
|
"author_date": datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat(),
|
|
"committer_name": author, # TODO: extract committer separately
|
|
"committer_email": email, # TODO: extract committer separately
|
|
"committer_date": datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat(),
|
|
"message": message,
|
|
"body": body,
|
|
"branches": branch_map.get(commit_hash, []),
|
|
"tags": tag_map.get(commit_hash, []),
|
|
"stats": stats,
|
|
"diff": diff_output,
|
|
"files": [_file_change_to_dict(f) for f in files],
|
|
}
|
|
|
|
|
|
def _get_branch_map(repo_path: str) -> dict[str, list[str]]:
|
|
"""Build a mapping of commit hash to branch names."""
|
|
result = {}
|
|
branch_output = _run_git_command(
|
|
repo_path, ["for-each-ref", "--format=%(objectname) %(refname:short)", "refs/heads/"]
|
|
)
|
|
|
|
for line in branch_output.strip().split("\n"):
|
|
if " " in line:
|
|
commit_hash, branch_name = line.split(" ", 1)
|
|
if commit_hash not in result:
|
|
result[commit_hash] = []
|
|
result[commit_hash].append(branch_name)
|
|
|
|
return result
|
|
|
|
|
|
def _get_tag_map(repo_path: str) -> dict[str, list[str]]:
|
|
"""Build a mapping of commit hash to tag names."""
|
|
result = {}
|
|
tag_output = _run_git_command(
|
|
repo_path, ["for-each-ref", "--format=%(objectname) %(refname:short)", "refs/tags/"]
|
|
)
|
|
|
|
for line in tag_output.strip().split("\n"):
|
|
if " " in line:
|
|
commit_hash, tag_name = line.split(" ", 1)
|
|
if commit_hash not in result:
|
|
result[commit_hash] = []
|
|
result[commit_hash].append(tag_name)
|
|
|
|
return result
|
|
|
|
|
|
def _build_graph_data(commits: list[Commit]) -> dict[str, Any]:
|
|
"""Build graph visualization data from commits."""
|
|
if not commits:
|
|
return {"nodes": [], "edges": []}
|
|
|
|
# Create hash to index mapping
|
|
hash_to_idx = {c.hash: i for i, c in enumerate(commits)}
|
|
|
|
nodes = []
|
|
edges = []
|
|
|
|
for i, commit in enumerate(commits):
|
|
# Calculate column based on branch
|
|
column = 0
|
|
if commit.branches:
|
|
# Use first branch as column indicator
|
|
column = hash(commit.branches[0]) % 5
|
|
|
|
nodes.append(
|
|
{
|
|
"hash": commit.hash,
|
|
"x": column * 60 + 30,
|
|
"y": i * 50 + 25,
|
|
"column": column,
|
|
}
|
|
)
|
|
|
|
# Create edges to parents
|
|
for parent_hash in commit.parents:
|
|
if parent_hash in hash_to_idx:
|
|
edges.append(
|
|
{
|
|
"from_hash": commit.hash,
|
|
"to_hash": parent_hash,
|
|
"type": "parent",
|
|
}
|
|
)
|
|
|
|
return {"nodes": nodes, "edges": edges}
|
|
|
|
|
|
def _parse_stats(stat_output: str) -> dict[str, int]:
|
|
"""Parse git show --stat output."""
|
|
lines = stat_output.strip().split("\n")
|
|
stats = {"files_changed": 0, "insertions": 0, "deletions": 0}
|
|
|
|
for line in lines:
|
|
line = line.strip()
|
|
if "files changed" in line or "file changed" in line:
|
|
# Parse summary line like "3 files changed, 45 insertions(+), 12 deletions(-)"
|
|
parts = line.split(",")
|
|
for part in parts:
|
|
part = part.strip()
|
|
if "file" in part:
|
|
try:
|
|
stats["files_changed"] = int(part.split()[0])
|
|
except (ValueError, IndexError):
|
|
pass
|
|
elif "insertion" in part:
|
|
try:
|
|
stats["insertions"] = int(part.split()[0])
|
|
except (ValueError, IndexError):
|
|
pass
|
|
elif "deletion" in part:
|
|
try:
|
|
stats["deletions"] = int(part.split()[0])
|
|
except (ValueError, IndexError):
|
|
pass
|
|
|
|
return stats
|
|
|
|
|
|
def _parse_diff(diff_output: str) -> list[FileChange]:
|
|
"""Parse git diff output into file changes."""
|
|
files = []
|
|
current_file = None
|
|
current_diff = []
|
|
|
|
for line in diff_output.split("\n"):
|
|
if line.startswith("diff --git"):
|
|
# Save previous file
|
|
if current_file:
|
|
current_file.diff = "\n".join(current_diff)
|
|
files.append(current_file)
|
|
|
|
# Start new file
|
|
current_diff = [line]
|
|
current_file = FileChange(
|
|
path="",
|
|
change_type="modified",
|
|
insertions=0,
|
|
deletions=0,
|
|
diff="",
|
|
)
|
|
elif line.startswith("--- ") or line.startswith("+++ "):
|
|
current_diff.append(line)
|
|
if line.startswith("+++ ") and not line.startswith("+++ /dev/null"):
|
|
current_file.path = line[6:]
|
|
elif line.startswith("@@ "):
|
|
current_diff.append(line)
|
|
elif line.startswith("+") and not line.startswith("+++"):
|
|
current_diff.append(line)
|
|
current_file.insertions += 1
|
|
elif line.startswith("-") and not line.startswith("---"):
|
|
current_diff.append(line)
|
|
current_file.deletions += 1
|
|
elif current_file:
|
|
current_diff.append(line)
|
|
|
|
# Save last file
|
|
if current_file:
|
|
current_file.diff = "\n".join(current_diff)
|
|
files.append(current_file)
|
|
|
|
return files
|
|
|
|
|
|
def _commit_to_dict(commit: Commit, graph_symbol: str = "", graph_depth: int = 0) -> dict[str, Any]:
|
|
"""Convert Commit dataclass to dictionary."""
|
|
refs = []
|
|
if commit.branches:
|
|
refs.extend(commit.branches)
|
|
if commit.tags:
|
|
refs.extend(commit.tags)
|
|
|
|
return {
|
|
"hash": commit.hash,
|
|
"short_hash": commit.short_hash,
|
|
"parents": commit.parents,
|
|
"author_name": commit.author,
|
|
"author_email": commit.email,
|
|
"author_date": datetime.fromtimestamp(commit.timestamp, tz=timezone.utc).isoformat(),
|
|
"message": commit.message,
|
|
"refs": refs,
|
|
"graph_symbol": graph_symbol,
|
|
"graph_depth": graph_depth,
|
|
}
|
|
|
|
|
|
def _file_change_to_dict(file_change: FileChange) -> dict[str, Any]:
|
|
"""Convert FileChange dataclass to dictionary."""
|
|
return {
|
|
"path": file_change.path,
|
|
"change_type": file_change.change_type,
|
|
"insertions": file_change.insertions,
|
|
"deletions": file_change.deletions,
|
|
"diff": file_change.diff,
|
|
}
|