Developer
5ed5e1c84b
- Fix ProjectsPage tests by wrapping renders in MemoryRouter (9 passing) - Improve session auto-naming to 'project / repo / tool' format - Add missing /users/me/sessions endpoint for sidebar session loading - Handle git history 500s: catch RuntimeError in endpoints, graceful empty repo handling - Add git status badge and discard-changes button to FileEditor toolbar Quality gates: tsc pass, build pass, Python syntax pass
398 lines
12 KiB
Python
398 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 (may fail for empty repos)
|
|
try:
|
|
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()]
|
|
except RuntimeError:
|
|
branches = []
|
|
|
|
# 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")
|
|
|
|
try:
|
|
log_output = _run_git_command(repo_path, log_args)
|
|
except RuntimeError:
|
|
# Empty repo or no commits
|
|
return {
|
|
"commits": [],
|
|
"branches": branches,
|
|
"total_commits": 0,
|
|
"graph_data": {"nodes": [], "edges": []},
|
|
}
|
|
|
|
# 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
|
|
try:
|
|
count_output = _run_git_command(repo_path, ["rev-list", "--all", "--count"])
|
|
total_commits = int(count_output.strip()) if count_output.strip() else 0
|
|
except RuntimeError:
|
|
total_commits = 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,
|
|
}
|