fegger
4b7ddcb1c1
Integrates with Zed editor history and Git commits to track daily coding activity. Writes idempotent `## Coding Work` sections into Obsidian daily notes with project breakdowns and commit stats.
226 lines
6.4 KiB
Python
226 lines
6.4 KiB
Python
"""Scan git repositories for today's commits."""
|
|
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
|
|
|
|
@dataclass
|
|
class CommitInfo:
|
|
hash: str
|
|
message: str
|
|
author: str
|
|
timestamp: int
|
|
files_changed: int
|
|
insertions: int
|
|
deletions: int
|
|
|
|
|
|
@dataclass
|
|
class RepoSummary:
|
|
name: str
|
|
path: Path
|
|
commits: List[CommitInfo]
|
|
total_insertions: int
|
|
total_deletions: int
|
|
total_files: int
|
|
|
|
|
|
def _run_git(args: List[str], cwd: Path, check: bool = True) -> str:
|
|
"""Run a git command and return stdout."""
|
|
result = subprocess.run(
|
|
["git"] + args,
|
|
cwd=cwd,
|
|
capture_output=True,
|
|
text=True,
|
|
check=check,
|
|
)
|
|
if result.returncode != 0 and check:
|
|
raise subprocess.CalledProcessError(
|
|
result.returncode,
|
|
["git"] + args,
|
|
output=result.stdout,
|
|
stderr=result.stderr,
|
|
)
|
|
return result.stdout
|
|
|
|
|
|
def _parse_git_date(date_str: str) -> int:
|
|
"""Parse a git date string into Unix timestamp.
|
|
|
|
Handles both ISO ('2026-05-26 14:30:00 +0200') and
|
|
RFC ('Tue, 26 May 2026 14:30:00 +0200') formats.
|
|
"""
|
|
date_str = date_str.strip()
|
|
# Try ISO format first
|
|
for fmt in ("%Y-%m-%d %H:%M:%S %z", "%a, %d %b %Y %H:%M:%S %z"):
|
|
try:
|
|
dt = datetime.strptime(date_str, fmt)
|
|
return int(dt.timestamp())
|
|
except ValueError:
|
|
continue
|
|
return 0
|
|
|
|
|
|
def _get_commits_for_date(
|
|
repo_path: Path, start_ts: int, end_ts: int
|
|
) -> List[CommitInfo]:
|
|
"""Get commits in the given timestamp range."""
|
|
# Use --since and --until with ISO format which git understands reliably
|
|
since = datetime.fromtimestamp(start_ts, tz=timezone.utc).isoformat()
|
|
until = datetime.fromtimestamp(end_ts, tz=timezone.utc).isoformat()
|
|
|
|
log_format = "%H%x00%s%x00%an%x00%ad%x00"
|
|
try:
|
|
output = _run_git(
|
|
[
|
|
"log",
|
|
f"--since={since}",
|
|
f"--until={until}",
|
|
f"--format={log_format}",
|
|
"--date=iso",
|
|
"--no-merges",
|
|
],
|
|
cwd=repo_path,
|
|
check=False,
|
|
)
|
|
except subprocess.CalledProcessError:
|
|
return []
|
|
|
|
if not output.strip():
|
|
return []
|
|
|
|
commits = []
|
|
entries = output.strip().split("\n")
|
|
for entry in entries:
|
|
parts = entry.split("\x00")
|
|
if len(parts) < 4:
|
|
continue
|
|
commit_hash, message, author, date_str = parts[0], parts[1], parts[2], parts[3]
|
|
ts = _parse_git_date(date_str)
|
|
|
|
# Get stats for this commit
|
|
try:
|
|
stat_output = _run_git(
|
|
["show", "--stat", "--format=", commit_hash],
|
|
cwd=repo_path,
|
|
check=False,
|
|
)
|
|
except subprocess.CalledProcessError:
|
|
stat_output = ""
|
|
|
|
files_changed, insertions, deletions = 0, 0, 0
|
|
for line in stat_output.splitlines():
|
|
line = line.strip()
|
|
if "file changed" in line or "files changed" in line:
|
|
# Parse line like "3 files changed, 10 insertions(+), 2 deletions(-)"
|
|
parts_stat = line.split(",")
|
|
for part in parts_stat:
|
|
part = part.strip()
|
|
if "changed" in part:
|
|
try:
|
|
files_changed = int(part.split()[0])
|
|
except ValueError:
|
|
pass
|
|
elif "insertion" in part:
|
|
try:
|
|
insertions = int(part.split()[0])
|
|
except ValueError:
|
|
pass
|
|
elif "deletion" in part:
|
|
try:
|
|
deletions = int(part.split()[0])
|
|
except ValueError:
|
|
pass
|
|
|
|
commits.append(
|
|
CommitInfo(
|
|
hash=commit_hash[:8],
|
|
message=message.strip(),
|
|
author=author.strip(),
|
|
timestamp=ts,
|
|
files_changed=files_changed,
|
|
insertions=insertions,
|
|
deletions=deletions,
|
|
)
|
|
)
|
|
|
|
return commits
|
|
|
|
|
|
def _is_git_repo(path: Path) -> bool:
|
|
"""Check if a directory is a git repository."""
|
|
return (path / ".git").is_dir()
|
|
|
|
|
|
def find_git_repos(base_dir: Path, max_depth: int = 3) -> List[Path]:
|
|
"""Find git repositories under base_dir up to max_depth levels."""
|
|
repos = []
|
|
base_dir = base_dir.expanduser().resolve()
|
|
if not base_dir.exists():
|
|
return repos
|
|
|
|
for depth in range(max_depth + 1):
|
|
pattern = "/".join(["*"] * depth) if depth > 0 else "."
|
|
if depth == 0:
|
|
if _is_git_repo(base_dir):
|
|
repos.append(base_dir)
|
|
continue
|
|
|
|
for candidate in base_dir.glob(pattern):
|
|
if candidate.is_dir() and _is_git_repo(candidate):
|
|
repos.append(candidate)
|
|
|
|
return repos
|
|
|
|
|
|
def fetch_git_activity(
|
|
start_ts: int,
|
|
end_ts: int,
|
|
code_dirs: Optional[List[Path]] = None,
|
|
max_depth: int = 3,
|
|
) -> List[RepoSummary]:
|
|
"""Scan code directories and return git activity summaries."""
|
|
if code_dirs is None:
|
|
code_dirs = [Path.home() / "Code"]
|
|
|
|
all_repos = []
|
|
for d in code_dirs:
|
|
all_repos.extend(find_git_repos(d, max_depth=max_depth))
|
|
|
|
# Deduplicate
|
|
seen = set()
|
|
unique_repos = []
|
|
for r in all_repos:
|
|
resolved = r.resolve()
|
|
if resolved not in seen:
|
|
seen.add(resolved)
|
|
unique_repos.append(resolved)
|
|
|
|
summaries = []
|
|
for repo_path in unique_repos:
|
|
commits = _get_commits_for_date(repo_path, start_ts, end_ts)
|
|
if not commits:
|
|
continue
|
|
|
|
total_insertions = sum(c.insertions for c in commits)
|
|
total_deletions = sum(c.deletions for c in commits)
|
|
total_files = sum(c.files_changed for c in commits)
|
|
|
|
summaries.append(
|
|
RepoSummary(
|
|
name=repo_path.name,
|
|
path=repo_path,
|
|
commits=commits,
|
|
total_insertions=total_insertions,
|
|
total_deletions=total_deletions,
|
|
total_files=total_files,
|
|
)
|
|
)
|
|
|
|
# Sort by number of commits descending
|
|
summaries.sort(key=lambda r: -len(r.commits))
|
|
return summaries
|