feat: add git history backend implementation
- Add git_history.py utility for extracting commit history and details - Add API endpoints for repository history and commit details - Integrate with existing git_repositories router This completes the backend for git history visualization.
This commit is contained in:
Fusion
@@ -0,0 +1,356 @@
|
||||
"""Git history extraction utilities for bare/mirror repositories."""
|
||||
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
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
|
||||
log_args = [
|
||||
"log",
|
||||
"--all",
|
||||
"--graph",
|
||||
f"--format=%H|%P|%an|%ae|%at|%s",
|
||||
f"--max-count={limit}",
|
||||
f"--skip={offset}",
|
||||
]
|
||||
if branch:
|
||||
log_args.append(branch)
|
||||
|
||||
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 = []
|
||||
graph_lines = log_output.strip().split("\n") if log_output.strip() else []
|
||||
|
||||
for line in graph_lines:
|
||||
if "|" not in line:
|
||||
continue
|
||||
|
||||
parts = line.split("|")
|
||||
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
|
||||
graph_data = _build_graph_data(commits)
|
||||
|
||||
return {
|
||||
"commits": [_commit_to_dict(c) for c in commits],
|
||||
"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": author,
|
||||
"email": email,
|
||||
"date": str(timestamp),
|
||||
"timestamp": timestamp,
|
||||
"message": message,
|
||||
"body": body,
|
||||
"branches": branch_map.get(commit_hash, []),
|
||||
"tags": tag_map.get(commit_hash, []),
|
||||
"stats": stats,
|
||||
"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) -> dict[str, Any]:
|
||||
"""Convert Commit dataclass to dictionary."""
|
||||
return {
|
||||
"hash": commit.hash,
|
||||
"short_hash": commit.short_hash,
|
||||
"parents": commit.parents,
|
||||
"author": commit.author,
|
||||
"email": commit.email,
|
||||
"date": commit.date,
|
||||
"timestamp": commit.timestamp,
|
||||
"message": commit.message,
|
||||
"branches": commit.branches,
|
||||
"tags": commit.tags,
|
||||
}
|
||||
|
||||
|
||||
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,
|
||||
}
|
||||
Reference in New Issue
Block a user