diff --git a/apps/api/src/api/git_repositories.py b/apps/api/src/api/git_repositories.py index cb2ead1..957e84b 100644 --- a/apps/api/src/api/git_repositories.py +++ b/apps/api/src/api/git_repositories.py @@ -14,6 +14,7 @@ from src.config import Settings from src.models.git_repository import GitRepository from src.models.project import Project from src.models.user import User +from src.utils.git_history import get_commit_detail, get_commit_history from src.utils.git_url_parser import parse_git_url router = APIRouter(prefix="/projects", tags=["git-repositories"]) @@ -209,3 +210,58 @@ async def create_repository( await session.commit() await session.refresh(repo) return repo + + +@router.get("/{project_id}/repositories/{repo_id}/history") +async def get_repository_history( + project_id: uuid.UUID, + repo_id: uuid.UUID, + view: str = "graph", + branch: str | None = None, + limit: int = 100, + offset: int = 0, + user_id: uuid.UUID = Depends(get_current_user_id), + session: AsyncSession = Depends(get_db_session), +) -> dict: + """Get commit history for a repository.""" + _user = await _get_user(session, user_id) + _project = await _get_owned_project(project_id, user_id, session) + + repo = await session.get(GitRepository, repo_id) + if repo is None or repo.project_id != project_id: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found") + + if not os.path.exists(repo.path): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found on disk") + + try: + history = get_commit_history(repo.path, branch=branch, limit=limit, offset=offset) + return history + except RuntimeError as e: + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)) + + +@router.get("/{project_id}/repositories/{repo_id}/commits/{commit_hash}") +async def get_repository_commit( + project_id: uuid.UUID, + repo_id: uuid.UUID, + commit_hash: str, + user_id: uuid.UUID = Depends(get_current_user_id), + session: AsyncSession = Depends(get_db_session), +) -> dict: + """Get detailed information about a specific commit.""" + _user = await _get_user(session, user_id) + _project = await _get_owned_project(project_id, user_id, session) + + repo = await session.get(GitRepository, repo_id) + if repo is None or repo.project_id != project_id: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found") + + if not os.path.exists(repo.path): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found on disk") + + try: + detail = get_commit_detail(repo.path, commit_hash) + return detail + except (RuntimeError, ValueError) as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) diff --git a/apps/api/src/utils/git_history.py b/apps/api/src/utils/git_history.py new file mode 100644 index 0000000..9ac61c2 --- /dev/null +++ b/apps/api/src/utils/git_history.py @@ -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, + }