refactor_batch_reapply_diffs.py

#!/usr/bin/env python3

"""
This script attempts to find files changed in a specific commit on a source branch, produces a patch for each file,
and applies those patches one by one to a corresponding file under a path prefix in the target branch.

This was created to help transfer changes from an old branch to a new branch after a refactor moved the app to a new path.
"""


import subprocess
from pathlib import Path

SOURCE_BRANCH = "old-branch"
TARGET_BRANCH = "new-branch"
DIFF_COMMIT = "commit-id-with-changes-missing-in-target"
TARGET_PATH_PREFIX = "apps/web"
PATCH_DIR = "patches"


def run_git(args: list[str]) -> subprocess.CompletedProcess:
    return subprocess.run(["git"] + args, capture_output=True, text=True)


def get_changed_files() -> list[str]:
    result = run_git(["diff-tree", "--no-commit-id", "--name-only", "-r", DIFF_COMMIT])
    return [f for f in result.stdout.strip().split("\n") if f]


def get_patch(filepath: str) -> str:
    result = run_git(["diff", f"{DIFF_COMMIT}^..{DIFF_COMMIT}", "--", filepath])
    return result.stdout


def get_added_lines(patch: str) -> list[str]:
    lines = []
    for line in patch.split("\n"):
        if line.startswith("+") and not line.startswith("+++"):
            content = line[1:].strip()
            if content:
                lines.append(content)
    return lines


def changes_already_applied(patch: str, target_content: str) -> bool:
    added_lines = get_added_lines(patch)
    if not added_lines:
        return True

    target_normalized = target_content.replace(" ", "").replace("\t", "")
    for line in added_lines:
        line_normalized = line.replace(" ", "").replace("\t", "")
        if line_normalized not in target_normalized:
            return False
    return True


def is_deletion_patch(patch: str) -> bool:
    return "deleted file mode" in patch


def is_new_file_patch(patch: str) -> bool:
    return "new file mode" in patch


def get_pending_files() -> list[str]:
    pending = []
    for source_path in get_changed_files():
        target_path = Path(TARGET_PATH_PREFIX) / source_path
        patch = get_patch(source_path)
        if not patch:
            continue

        if not target_path.exists():
            if is_deletion_patch(patch):
                continue
            if is_new_file_patch(patch):
                pending.append(source_path)
                continue
            continue

        if is_deletion_patch(patch):
            pending.append(source_path)
            continue

        if not changes_already_applied(patch, target_path.read_text()):
            pending.append(source_path)

    return pending


def process_file(source_path: str) -> bool:
    target_path = Path(TARGET_PATH_PREFIX) / source_path
    filename = Path(source_path).name
    patch_dir = Path(PATCH_DIR)
    patch_file = patch_dir / f"{filename}.patch"

    patch = get_patch(source_path)
    if not patch:
        print(f"  No changes in commit")
        return True

    patch_dir.mkdir(exist_ok=True)
    patch_file.write_text(patch)

    is_new = is_new_file_patch(patch)
    is_del = is_deletion_patch(patch)

    if is_new:
        print(f"  New file (will create)")
    elif is_del:
        print(f"  Delete file")

    response = input(f"  Apply? [Y/n/q]: ").strip().lower()
    if response == "q":
        return False
    if response not in ("", "y"):
        print("  Skipped")
        return True

    if is_del:
        target_path.unlink()
        print(f"  Deleted")
        return True

    if is_new:
        target_path.parent.mkdir(parents=True, exist_ok=True)

    result = run_git(
        ["apply", f"--directory={TARGET_PATH_PREFIX}", "--3way", str(patch_file)]
    )

    if result.returncode == 0:
        print(f"  Applied successfully")
    else:
        print(f"  Conflicts - resolve manually")
        if result.stderr:
            print(f"  {result.stderr.strip()}")
        input("  Press Enter to continue...")

    return True


def main():
    pending = get_pending_files()

    if not pending:
        print("All files up to date")
        return

    print(f"Found {len(pending)} file(s) with changes:\n")

    for i, source_path in enumerate(pending, 1):
        filename = Path(source_path).name
        print(f"[{i}/{len(pending)}] {PATCH_DIR}/{filename}.patch")
        if not process_file(source_path):
            print("Quit")
            break
        print()


if __name__ == "__main__":
    main()