"""Git history ingestion."""
import subprocess
import logging
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Optional
from ..models import GitChange
logger = logging.getLogger(__name__)
class GitIngestor:
"""Collect recent git changes."""
def get_recent_commits(
self,
repo_path: str,
since_minutes: int = 1440, # Default 24h
max_commits: int = 20
) -> list[GitChange]:
"""Get recent commits from a git repo."""
repo = Path(repo_path).resolve()
if not repo.exists():
logger.warning("Repo not found: %s", repo)
return []
try:
since = f"{since_minutes} minutes ago"
result = subprocess.run(
[
"git", "--no-pager", "log",
"--since", since,
"--format=%H|%ae|%aI|%s",
"--name-only",
"-n", str(max_commits),
"--no-merges",
],
capture_output=True, text=True, timeout=10,
cwd=str(repo)
)
if result.returncode != 0:
logger.warning("git log failed in %s: %s", repo, result.stderr)
return []
changes = []
current_commit = None
current_files = []
for line in result.stdout.split("\n"):
if not line.strip():
continue
# Check if this is a commit line (contains |)
if "|" in line:
# Save previous commit
if current_commit:
changes.append(current_commit)
parts = line.split("|", 3)
if len(parts) >= 4:
try:
timestamp = datetime.fromisoformat(parts[2]).replace(tzinfo=timezone.utc)
except ValueError:
timestamp = datetime.now(timezone.utc)
current_commit = GitChange(
timestamp=timestamp,
repo_path=str(repo),
commit_hash=parts[0][:8],
message=parts[3],
author=parts[1],
files_changed=[]
)
current_files = []
elif current_commit:
current_files.append(line.strip())
# Don't forget the last commit
if current_commit:
current_commit.files_changed = current_files
changes.append(current_commit)
return changes
except FileNotFoundError:
logger.warning("Git not found")
return []
except subprocess.TimeoutExpired:
logger.warning("Git log timed out")
return []
def get_diff_summary(
self,
repo_path: str,
since_minutes: int = 1440
) -> list[dict]:
"""Get file change statistics for recent commits."""
repo = Path(repo_path).resolve()
if not repo.exists():
return []
try:
since = f"{since_minutes} minutes ago"
result = subprocess.run(
[
"git", "--no-pager", "diff",
"--stat",
f"HEAD~$(git rev-list --count --since='{since}' HEAD)",
"HEAD"
],
capture_output=True, text=True, timeout=10,
cwd=str(repo),
shell=True # Need shell for $() expansion
)
if result.returncode != 0:
return []
return {"summary": result.stdout.strip(), "repo": str(repo)}
except (FileNotFoundError, subprocess.TimeoutExpired):
return []
def get_modified_files(
self,
repo_path: str,
since_minutes: int = 1440
) -> list[str]:
"""Get list of files modified in recent commits."""
changes = self.get_recent_commits(repo_path, since_minutes)
files = set()
for change in changes:
files.update(change.files_changed)
return list(files)