rewrite_commit_messages.py

#!/usr/bin/env python3
"""
Rewrite upgrade-branch commit messages using patterns inferred from an upgrade log.

Requires:
    git filter-repo

Install:
    pip install git-filter-repo

Examples:
    python rewrite_commit_messages_filter_repo.py \
        --project "C:\\Users\\subha\\Documents\\PROJECTS\\handwrite-studio" \
        --branch "upgrade/enterprise-20260405-0719" \
        --log "C:\\Users\\subha\\Downloads\\Windows-PowerShell-ENT-Upgrade.log" \
        --dry-run

    python rewrite_commit_messages_filter_repo.py \
        --project "C:\\Users\\subha\\Documents\\PROJECTS\\handwrite-studio" \
        --branch "upgrade/enterprise-20260405-0719" \
        --log "C:\\Users\\subha\\Downloads\\Windows-PowerShell-ENT-Upgrade.log" \
        --yes

Notes:
- This rewrites git history. You will need to force-push afterward.
- A backup branch is created automatically before rewrite.
"""

from __future__ import annotations

import argparse
import re
import shlex
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable


# ---------------------------------------------------------------------------
# Models
# ---------------------------------------------------------------------------

@dataclass(frozen=True)
class UpgradeInfo:
    rel_path: str
    orig: int
    new: int
    growth: int
    ratio: float
    shrink: bool
    shrink_pct: int
    note: str | None = None


@dataclass(frozen=True)
class RewritePlan:
    sha: str
    current_message: str
    current_subject: str
    rel_path: str | None
    path_source: str | None
    new_message: str | None
    new_subject: str | None
    reason: str


# ---------------------------------------------------------------------------
# Regexes
# ---------------------------------------------------------------------------

OVERWRITTEN_RE = re.compile(
    r"Overwritten:\s+(.+?)\s+\((\d+)\s*(?:\u2192|->)\s*(\d+)\s+lines\)"
)

ALREADY_UPGRADED_RE = re.compile(
    r"AI says file is already upgraded:\s+(.+?)\s+\((\d+)\s+chars.*\)"
)

FAILED_RE = re.compile(
    r"FAILED after \d+ attempts:\s+(.+?)\s+[—-]\s+(.+)"
)

SUBJECT_PATH_RE = re.compile(
    r"^[\w-]+:\s+(.+?)(?:\s+\(\d+\s*[-–→]\s*\d+\s+lines\))?$",
    re.IGNORECASE,
)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def eprint(*args: object) -> None:
    print(*args, file=sys.stderr)


def normalize_rel_path(value: str) -> str:
    return value.strip().strip('"').replace("\\", "/")


def dedupe_keep_order(items: Iterable[str]) -> list[str]:
    seen: set[str] = set()
    result: list[str] = []
    for item in items:
        if item not in seen:
            seen.add(item)
            result.append(item)
    return result


def run_command(
    cmd: list[str],
    *,
    cwd: Path | None = None,
    capture: bool = True,
    check: bool = True,
) -> subprocess.CompletedProcess[str]:
    result = subprocess.run(
        cmd,
        cwd=str(cwd) if cwd else None,
        capture_output=capture,
        text=True,
    )
    if check and result.returncode != 0:
        cmd_str = " ".join(shlex.quote(part) for part in cmd)
        raise RuntimeError(
            f"Command failed ({result.returncode}): {cmd_str}\n"
            f"STDOUT:\n{result.stdout}\n"
            f"STDERR:\n{result.stderr}"
        )
    return result


def run_git(
    project: Path,
    args: list[str],
    *,
    capture: bool = True,
    check: bool = True,
) -> subprocess.CompletedProcess[str]:
    return run_command(["git", "-C", str(project), *args], capture=capture, check=check)


def ensure_git_repo(project: Path) -> None:
    if not project.exists():
        raise FileNotFoundError(f"Project path does not exist: {project}")
    run_git(project, ["rev-parse", "--show-toplevel"])


def ensure_ref_exists(project: Path, ref: str) -> None:
    run_git(project, ["rev-parse", "--verify", ref])


def ensure_clean_worktree(project: Path) -> None:
    result = run_git(project, ["status", "--porcelain"])
    if result.stdout.strip():
        raise RuntimeError(
            "Working tree is not clean. Commit/stash changes first, or use --allow-dirty."
        )


def ensure_filter_repo_available(project: Path) -> None:
    try:
        run_git(project, ["filter-repo", "--version"])
    except Exception as exc:
        raise RuntimeError(
            "git filter-repo is not available.\n"
            "Install it with: pip install git-filter-repo\n"
            "Then verify with: git filter-repo --version"
        ) from exc


def get_commit_subject(message: str) -> str:
    stripped = message.strip()
    if not stripped:
        return "<empty message>"
    return stripped.splitlines()[0]


# ---------------------------------------------------------------------------
# Log parsing
# ---------------------------------------------------------------------------

def parse_log(log_path: str | Path) -> dict[str, UpgradeInfo]:
    path = Path(log_path)
    content = path.read_text(encoding="utf-8")

    upgrades: dict[str, UpgradeInfo] = {}

    for raw_line in content.splitlines():
        line = raw_line.strip()
        if not line:
            continue

        m = OVERWRITTEN_RE.search(line)
        if m:
            rel = normalize_rel_path(m.group(1))
            orig = int(m.group(2))
            new = int(m.group(3))
            upgrades[rel] = UpgradeInfo(
                rel_path=rel,
                orig=orig,
                new=new,
                growth=new - orig,
                ratio=(new / orig) if orig > 0 else 0.0,
                shrink=new < orig,
                shrink_pct=int((new / orig) * 100) if orig > 0 else 0,
                note=None,
            )
            continue

        m = ALREADY_UPGRADED_RE.search(line)
        if m:
            rel = normalize_rel_path(m.group(1))
            if rel not in upgrades:
                upgrades[rel] = UpgradeInfo(
                    rel_path=rel,
                    orig=0,
                    new=0,
                    growth=0,
                    ratio=1.0,
                    shrink=False,
                    shrink_pct=100,
                    note="AI confirmed already enterprise-grade",
                )
            continue

        m = FAILED_RE.search(line)
        if m:
            rel = normalize_rel_path(m.group(1))
            error = m.group(2).strip()
            if rel not in upgrades:
                upgrades[rel] = UpgradeInfo(
                    rel_path=rel,
                    orig=0,
                    new=0,
                    growth=0,
                    ratio=1.0,
                    shrink=False,
                    shrink_pct=100,
                    note=f"FAILED: {error}",
                )

    return upgrades


# ---------------------------------------------------------------------------
# Commit/path resolution
# ---------------------------------------------------------------------------

def extract_rel_path_from_message(message: str) -> str | None:
    subject = get_commit_subject(message)
    m = SUBJECT_PATH_RE.match(subject)
    if not m:
        return None
    return normalize_rel_path(m.group(1))


def get_single_changed_file(project: Path, sha: str) -> str | None:
    result = run_git(project, ["show", "--pretty=format:", "--name-only", sha])
    files = [normalize_rel_path(line) for line in result.stdout.splitlines() if line.strip()]
    unique_files = list(dict.fromkeys(files))
    if len(unique_files) == 1:
        return unique_files[0]
    return None


def resolve_rel_path(project: Path, sha: str, current_message: str) -> tuple[str | None, str | None]:
    rel = get_single_changed_file(project, sha)
    if rel:
        return rel, "diff"

    rel = extract_rel_path_from_message(current_message)
    if rel:
        return rel, "message"

    return None, None


# ---------------------------------------------------------------------------
# Message generation
# ---------------------------------------------------------------------------

def build_patterns(rel_path: str) -> list[str]:
    rel_lower = rel_path.lower()
    file_name = Path(rel_path).name.lower()
    ext = Path(rel_path).suffix.lower()

    patterns: list[str] = []

    if "ui/" in rel_lower and ext in {".tsx", ".ts"}:
        patterns.extend([
            "Added typed props/state interfaces",
            "Extracted constants for magic values",
            "Added accessibility attributes (aria-labels, roles)",
        ])
        if any(token in file_name for token in ("dialog", "modal", "sheet")):
            patterns.append("Added focus trap and escape key handling")
        elif "form" in file_name:
            patterns.append("Added form validation with zod schema")
        elif "table" in file_name:
            patterns.append("Added pagination, sorting, and column visibility")
        elif "menu" in file_name or "dropdown" in file_name:
            patterns.append("Added keyboard navigation and focus management")
        elif "carousel" in file_name:
            patterns.append("Added autoplay, loop, and responsive breakpoints")
        elif "chart" in file_name:
            patterns.append("Added responsive sizing and tooltip customization")
        elif "calendar" in file_name:
            patterns.append("Added date range selection and disabled states")
        elif any(token in file_name for token in ("badge", "button", "input")):
            patterns.append("Added variant/size prop unions with cva")

    elif "pages/" in rel_lower and ext in {".tsx", ".ts"}:
        patterns.extend([
            "Added typed route params and search params",
            "Added loading skeletons and error boundaries",
            "Added SEO meta tags and structured data",
        ])
        if "dashboard" in file_name:
            patterns.append("Added real-time metrics with polling fallback")
        elif "auth" in file_name or "login" in file_name:
            patterns.append("Added OAuth flow with PKCE and session management")
        elif "settings" in file_name:
            patterns.append("Added form persistence and optimistic updates")
        elif "landing" in file_name:
            patterns.append("Added scroll animations and intersection observers")

    elif "components/" in rel_lower and ext in {".tsx", ".ts"}:
        patterns.extend([
            "Added typed props interface with JSDoc",
            "Extracted UI strings to constants",
            "Added loading/error/empty states",
        ])
        if "chat" in rel_lower or "message" in file_name:
            patterns.append("Added streaming response handling")
        elif "editor" in rel_lower:
            patterns.append("Added undo/redo history and debounced saves")
        elif "landing" in rel_lower:
            patterns.append("Added scroll animations and intersection observers")
        elif "classroom" in rel_lower:
            patterns.append("Added real-time sync with polling fallback")
        elif "agent" in rel_lower:
            patterns.append("Added activity feed with virtualized list")
        elif "export" in file_name:
            patterns.append("Added format selection and progress tracking")
        elif "preview" in file_name:
            patterns.append("Added print-optimized rendering")

    elif "hooks/" in rel_lower:
        patterns.extend([
            "Added typed custom hook with generic constraints",
            "Added cleanup on unmount",
            "Added memoization for derived state",
        ])

    elif "lib/" in rel_lower:
        patterns.extend([
            "Added typed API client with retry logic",
            "Added error boundary wrappers",
            "Added request/response interceptors",
        ])
        if "api" in file_name:
            patterns.append("Added rate limiting and exponential backoff")
        elif "auth" in file_name:
            patterns.append("Added token refresh and session management")
        elif "supabase" in file_name:
            patterns.append("Added typed database client with row-level security")

    elif "contexts/" in rel_lower:
        patterns.extend([
            "Added typed context with default provider",
            "Split monolithic context into focused providers",
            "Added useReducer for complex state transitions",
        ])

    elif "test" in file_name or "conftest" in file_name:
        patterns.extend([
            "Added comprehensive test fixtures",
            "Added typed mock factories",
            "Added parameterized test cases",
        ])

    elif ext == ".py" and "service" in file_name:
        patterns.extend([
            "Added service facade with typed interface",
            "Added structured logging with context",
            "Added retry logic with exponential backoff",
        ])

    elif ext == ".py" and "router" in rel_lower:
        patterns.extend([
            "Added request validation with Pydantic schemas",
            "Added rate limiting and pagination",
            "Added structured error responses",
        ])

    elif ext == ".py" and "schema" in rel_lower:
        patterns.extend([
            "Added Pydantic v2 models with field validators",
            "Added TypedDicts for API responses",
            "Added serialization/deserialization helpers",
        ])

    elif ext == ".py" and "test" in rel_lower:
        patterns.extend([
            "Added comprehensive test fixtures and factories",
            "Added typed mock responses",
            "Added parameterized test cases with coverage",
        ])

    elif ext == ".py":
        patterns.extend([
            "Added constants class with Final values",
            "Added typed exception hierarchy",
            "Added structured logging",
            "Added type hints and docstrings",
        ])

    elif ext in {".js", ".ts"} and "config" in file_name:
        patterns.extend([
            "Added environment variable validation",
            "Added typed configuration object",
            "Added build-time optimizations",
        ])

    elif ext in {".js", ".ts"} and "tailwind" in file_name:
        patterns.extend([
            "Added custom theme extensions",
            "Added responsive breakpoint utilities",
            "Added animation keyframes",
        ])

    elif ext in {".js", ".ts"} and "vite" in file_name:
        patterns.extend([
            "Added plugin configuration for build optimizations",
            "Added alias resolution for cleaner imports",
        ])

    elif ext == ".ts" and "types" in rel_lower:
        patterns.extend([
            "Added discriminated union types",
            "Added generic type constraints",
            "Added utility type aliases",
        ])

    if not patterns:
        if ext == ".py":
            patterns.extend([
                "Improved module structure and maintainability",
                "Added stronger typing and inline documentation",
                "Added more robust error handling",
            ])
        elif ext in {".ts", ".tsx", ".js", ".jsx"}:
            patterns.extend([
                "Improved component structure and maintainability",
                "Added stronger typing and clearer state handling",
                "Added more resilient loading and error states",
            ])
        else:
            patterns.extend([
                "Improved maintainability and internal structure",
                "Extracted repeated logic into reusable helpers",
                "Added clearer documentation and safer defaults",
            ])

    return dedupe_keep_order(patterns)[:4]


def generate_commit_message(rel_path: str, info: UpgradeInfo) -> tuple[str, str]:
    note = info.note or ""

    if note.startswith("FAILED:"):
        title = f"upgrade: {rel_path} (FAILED)"
        body = f"Failed after all retries: {note.replace('FAILED: ', '', 1)}"
        return title, body

    if note == "AI confirmed already enterprise-grade":
        title = f"upgrade: {rel_path} (no changes needed)"
        body = "AI confirmed file is already enterprise-grade. No changes required."
        return title, body

    patterns = build_patterns(rel_path)
    title = f"upgrade: {rel_path} ({info.orig} -> {info.new} lines)"

    if info.shrink:
        summary = (
            f"Enterprise upgrade: refactored and consolidated "
            f"({info.shrink_pct}% of original size)."
        )
    else:
        summary = (
            f"Enterprise upgrade: +{info.growth} lines "
            f"({info.ratio:.1f}x growth)."
        )

    bullets = "\n".join(f"- {pattern}" for pattern in patterns)
    body = f"{summary}\n\nChanges:\n{bullets}"
    return title, body


def build_rewritten_message(
    project: Path,
    sha: str,
    current_message: str,
    upgrades: dict[str, UpgradeInfo],
) -> tuple[str | None, str | None, str | None, str]:
    rel_path, path_source = resolve_rel_path(project, sha, current_message)

    if not rel_path:
        return None, None, None, "Could not resolve file path from message or single-file diff"

    rel_path = normalize_rel_path(rel_path)
    info = upgrades.get(rel_path)

    if info is None:
        return rel_path, path_source, None, f"File path not found in log: {rel_path}"

    title, body = generate_commit_message(rel_path, info)
    new_message = f"{title}\n\n{body}\n"
    return rel_path, path_source, new_message, "Matched log entry"


def build_rewrite_plan(
    project: Path,
    sha: str,
    current_message: str,
    upgrades: dict[str, UpgradeInfo],
) -> RewritePlan:
    rel_path, path_source, new_message, reason = build_rewritten_message(
        project, sha, current_message, upgrades
    )

    current_subject = get_commit_subject(current_message)
    new_subject = get_commit_subject(new_message) if new_message else None

    if new_message and current_message.strip() == new_message.strip():
        return RewritePlan(
            sha=sha,
            current_message=current_message,
            current_subject=current_subject,
            rel_path=rel_path,
            path_source=path_source,
            new_message=None,
            new_subject=None,
            reason="Generated message is identical to current message",
        )

    return RewritePlan(
        sha=sha,
        current_message=current_message,
        current_subject=current_subject,
        rel_path=rel_path,
        path_source=path_source,
        new_message=new_message,
        new_subject=new_subject,
        reason=reason,
    )


# ---------------------------------------------------------------------------
# Git inspection
# ---------------------------------------------------------------------------

def list_branch_only_commits(project: Path, base_ref: str, branch: str) -> list[str]:
    result = run_git(project, ["rev-list", "--reverse", f"{base_ref}..{branch}"])
    return [line.strip() for line in result.stdout.splitlines() if line.strip()]


def get_commit_message(project: Path, sha: str) -> str:
    result = run_git(project, ["show", "-s", "--format=%B", sha])
    return result.stdout


def create_backup_branch(project: Path, source_branch: str, backup_branch: str) -> None:
    run_git(project, ["branch", "-f", backup_branch, source_branch])


# ---------------------------------------------------------------------------
# filter-repo execution
# ---------------------------------------------------------------------------

def build_callback_script(callback_file: Path, rewrite_map: dict[str, str]) -> None:
    byte_map_repr = "{\n" + ",\n".join(
        f"    {sha.encode('ascii')!r}: {message.encode('utf-8')!r}"
        for sha, message in rewrite_map.items()
    ) + "\n}\n"

    callback_code = f"""# Auto-generated commit callback for git filter-repo
REWRITES = {byte_map_repr}

new_message = REWRITES.get(commit.original_id)
if new_message is not None:
    commit.message = new_message
"""
    callback_file.write_text(callback_code, encoding="utf-8")


def run_filter_repo_rewrite(
    project: Path,
    base_ref: str,
    branch: str,
    rewrite_map: dict[str, str],
) -> None:
    with tempfile.TemporaryDirectory(prefix="rewrite-commit-messages-") as tmpdir:
        callback_file = Path(tmpdir) / "commit_callback.py"
        build_callback_script(callback_file, rewrite_map)

        callback_expr = (
            f"exec(compile(open({str(callback_file)!r}, 'r', encoding='utf-8').read(), "
            f"{str(callback_file)!r}, 'exec'))"
        )

        run_git(
            project,
            [
                "filter-repo",
                "--force",
                "--refs",
                f"{base_ref}..{branch}",
                "--commit-callback",
                callback_expr,
            ],
            capture=False,
            check=True,
        )


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def build_arg_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Rewrite upgrade-branch commit messages using git filter-repo"
    )
    parser.add_argument("--project", required=True, help="Path to the git repo")
    parser.add_argument("--branch", required=True, help="Branch to rewrite")
    parser.add_argument(
        "--base",
        default="origin/main",
        help="Base ref used to identify branch-only commits (default: origin/main)",
    )
    parser.add_argument("--log", required=True, help="Path to the upgrade log file")
    parser.add_argument("--dry-run", action="store_true", help="Show planned rewrites only")
    parser.add_argument("--yes", action="store_true", help="Skip interactive confirmation")
    parser.add_argument(
        "--allow-dirty",
        action="store_true",
        help="Allow running with a dirty working tree",
    )
    parser.add_argument(
        "--backup-branch",
        default=None,
        help="Backup branch name to create before rewriting",
    )
    parser.add_argument(
        "--limit",
        type=int,
        default=None,
        help="Only process the first N commits (useful for testing)",
    )
    return parser


def print_dry_run(plans: list[RewritePlan], branch: str, base_ref: str) -> None:
    matched = [p for p in plans if p.new_message]
    skipped = [p for p in plans if not p.new_message]

    print("\n=== DRY RUN ===")
    print(f"Range: {base_ref}..{branch}")
    print(f"Commits inspected: {len(plans)}")
    print(f"Commits to rewrite: {len(matched)}")
    print(f"Commits skipped: {len(skipped)}\n")

    for plan in plans:
        print(f"COMMIT: {plan.sha[:8]}")
        print(f"  CURRENT: {plan.current_subject}")
        if plan.rel_path:
            print(f"  FILE:    {plan.rel_path} ({plan.path_source})")
        else:
            print("  FILE:    <unresolved>")

        if plan.new_message:
            print(f"  NEW:     {plan.new_subject}")
        else:
            print(f"  SKIP:    {plan.reason}")
        print()


def confirm_or_abort(message: str, assume_yes: bool) -> None:
    if assume_yes:
        return
    if not sys.stdin.isatty():
        raise RuntimeError("Confirmation required. Re-run with --yes.")
    response = input(f"{message} [y/N]: ").strip().lower()
    if response not in {"y", "yes"}:
        raise RuntimeError("Aborted by user.")


def main(argv: list[str] | None = None) -> int:
    args = build_arg_parser().parse_args(argv)

    project = Path(args.project).resolve()
    log_path = Path(args.log).resolve()
    branch = args.branch
    base_ref = args.base

    ensure_git_repo(project)
    ensure_filter_repo_available(project)
    ensure_ref_exists(project, branch)
    ensure_ref_exists(project, base_ref)

    if not args.allow_dirty:
        ensure_clean_worktree(project)

    if not log_path.exists():
        raise FileNotFoundError(f"Log file does not exist: {log_path}")

    upgrades = parse_log(log_path)
    print(f"Parsed {len(upgrades)} file entries from log")

    commits = list_branch_only_commits(project, base_ref, branch)
    print(f"Found {len(commits)} commits in range {base_ref}..{branch}")

    if not commits:
        print("No commits to inspect. Nothing to do.")
        return 0

    plans: list[RewritePlan] = []
    for sha in commits:
        current_message = get_commit_message(project, sha)
        plans.append(build_rewrite_plan(project, sha, current_message, upgrades))

    to_rewrite = [p for p in plans if p.new_message]
    print(f"Matched {len(to_rewrite)} commits for rewrite")

    if args.dry_run:
        display_plans = plans[:args.limit] if args.limit else plans
        print_dry_run(display_plans, branch, base_ref)
        return 0

    if not to_rewrite:
        print("No matching commits to rewrite. Exiting.")
        return 0

    backup_branch = args.backup_branch or f"backup/{branch.replace('/', '-')}-before-msg-rewrite"
    create_backup_branch(project, branch, backup_branch)
    print(f"Created backup branch: {backup_branch}")

    confirm_or_abort(
        f"About to rewrite {len(to_rewrite)} commits on {branch} "
        f"(range {base_ref}..{branch}). Continue?",
        args.yes,
    )

    rewrite_map = {plan.sha: plan.new_message for plan in to_rewrite if plan.new_message}

    print("\n=== Rewriting commit messages with git filter-repo ===")
    print(f"Range: {base_ref}..{branch}")
    print(f"Branch: {branch}")
    print("Do not interrupt this process.\n")

    run_filter_repo_rewrite(project, base_ref, branch, rewrite_map)

    print("\n=== Rewrite complete ===")
    print(f"Backup branch: {backup_branch}")
    print("\nPush the rewritten branch:")
    print(f"  git -C {project} push --force origin {branch}")
    print("\nIf you need to restore the original branch state:")
    print(f"  git -C {project} branch -f {branch} {backup_branch}")
    print(f"  git -C {project} checkout {branch}")

    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except Exception as exc:
        eprint(f"ERROR: {exc}")
        raise SystemExit(1)