loop/scripts/commit_message.py

#!/usr/bin/env python3

"""
commit_message lints, formats, and rewrites Git commit messages.

Goals:
- Enforce a commit subject in "<package>: <summary>" form.
- Wrap subjects and bodies using repository defaults (69/72 columns).
- Preserve markdown-like body structure:
  - Paragraph breaks.
  - Bullet/numbered list item indentation.
  - Fenced and indented code blocks.
  - Git trailers.
- Support linting existing commit ranges and rewording commits in-place.
"""

from __future__ import annotations

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


DEFAULT_SUBJECT_WIDTH = 69
DEFAULT_BODY_WIDTH = 72

SUBJECT_PATTERN = re.compile(
    r"^(?P<pkg>[A-Za-z0-9_][A-Za-z0-9+_.\-/]*): (?P<summary>.+)$"
)
LIST_ITEM_PATTERN = re.compile(
    r"^(?P<indent>\s*)(?P<marker>[-+*]|\d+[.)])(?P<space>\s+)(?P<text>.*)$"
)
TRAILER_PATTERN = re.compile(r"^[A-Za-z0-9-]+:\s+\S.*$")
FENCE_PATTERN = re.compile(r"^\s*(```|~~~)")
QUOTE_PATTERN = re.compile(r"^(?P<indent>\s*)>\s?(?P<text>.*)$")
NEWLINE_ESCAPE_PATTERN = re.compile(r"\\n")


@dataclass
class LintIssue:
    """LintIssue captures one formatting violation."""

    line: int
    message: str


@dataclass
class LintResult:
    """LintResult stores issues and whether a message passed lint."""

    issues: list[LintIssue]

    def ok(self) -> bool:
        return len(self.issues) == 0


def run_git(args: list[str]) -> str:
    """run_git executes git and returns trimmed stdout."""

    return subprocess.check_output(
        ["git", *args],
        stderr=subprocess.STDOUT,
        text=True,
    ).rstrip("\n")


def get_commit_message(rev: str) -> str:
    """get_commit_message returns the full commit message for a revision."""

    return run_git(["show", "-s", "--format=%B", rev])


def commit_subject(rev: str) -> str:
    """commit_subject returns only the subject line for a revision."""

    return run_git(["show", "-s", "--format=%s", rev])


def commit_is_merge(rev: str) -> bool:
    """commit_is_merge reports whether rev has more than one parent."""

    parents = run_git(["rev-list", "--parents", "-n", "1", rev]).split()
    return len(parents) > 2


def split_subject_body(message: str) -> tuple[str, list[str]]:
    """
    split_subject_body returns (subject, body_lines) from a commit message.

    The first line is always treated as subject. Any following lines are body.
    """

    lines = message.splitlines()
    if not lines:
        return "", []
    return lines[0], lines[1:]


def _count_columns(text: str) -> int:
    """_count_columns counts characters for width checks."""

    return len(text)


def _is_list_item(line: str) -> bool:
    """_is_list_item checks whether a line starts a markdown-like list item."""

    return LIST_ITEM_PATTERN.match(line) is not None


def _is_trailer(line: str) -> bool:
    """_is_trailer checks whether a line is a git trailer."""

    return TRAILER_PATTERN.match(line) is not None


def _is_indented_code(line: str) -> bool:
    """_is_indented_code checks whether a line looks like indented code."""

    if "\t" in line[:1]:
        return True

    spaces = len(line) - len(line.lstrip(" "))
    return spaces >= 4 and line.strip() != ""


def _is_quote_line(line: str) -> bool:
    """_is_quote_line checks whether a line is markdown quote text."""

    return QUOTE_PATTERN.match(line) is not None


def _normalize_body_leading_blank(body_lines: list[str]) -> list[str]:
    """
    _normalize_body_leading_blank removes body-leading blank lines.

    The formatter emits exactly one blank separator between subject and body.
    """

    idx = 0
    while idx < len(body_lines) and body_lines[idx].strip() == "":
        idx += 1
    return body_lines[idx:]


def _normalize_body_trailing_blank(body_lines: list[str]) -> list[str]:
    """_normalize_body_trailing_blank trims trailing blank lines."""

    idx = len(body_lines)
    while idx > 0 and body_lines[idx - 1].strip() == "":
        idx -= 1
    return body_lines[:idx]


def _collapse_redundant_blank_lines(lines: list[str]) -> list[str]:
    """_collapse_redundant_blank_lines keeps at most one consecutive blank."""

    out: list[str] = []
    blank = False

    for line in lines:
        is_blank = line.strip() == ""
        if is_blank and blank:
            continue
        out.append(line)
        blank = is_blank

    return out


def _decode_escaped_newlines(lines: list[str]) -> list[str]:
    """
    _decode_escaped_newlines expands literal "\\n" sequences into real lines.

    This targets the common shell-escaping mistake from `git commit -m`.
    """

    out: list[str] = []
    split_re = re.compile(r"(?<!\\)\\n")

    for line in lines:
        if "\\n" not in line:
            out.append(line)
            continue

        out.extend(split_re.split(line))

    return out


def wrap_text(
    text: str,
    width: int,
    initial_indent: str = "",
    subsequent_indent: str = "",
) -> list[str]:
    """wrap_text wraps a paragraph without splitting long tokens."""

    wrapper = textwrap.TextWrapper(
        width=width,
        initial_indent=initial_indent,
        subsequent_indent=subsequent_indent,
        break_long_words=False,
        break_on_hyphens=False,
        replace_whitespace=True,
        drop_whitespace=True,
    )

    wrapped = wrapper.wrap(text)
    return wrapped if wrapped else [initial_indent.rstrip()]


def _consume_paragraph(body: list[str], start: int) -> tuple[int, list[str]]:
    """
    _consume_paragraph consumes plain paragraph lines from body[start:].

    It stops before blank lines and before other structured blocks.
    """

    parts: list[str] = []
    i = start

    while i < len(body):
        line = body[i]
        if line.strip() == "":
            break
        if FENCE_PATTERN.match(line):
            break
        if _is_list_item(line):
            break
        if _is_indented_code(line):
            break
        if _is_trailer(line):
            break
        if _is_quote_line(line):
            break
        parts.append(line.strip())
        i += 1

    return i, parts


def _format_paragraph(parts: list[str], width: int) -> list[str]:
    """_format_paragraph normalizes spacing and wraps plain text paragraphs."""

    text = " ".join(part for part in parts if part)
    if not text:
        return []
    return wrap_text(text, width)


def _consume_quote_block(
    body: list[str],
    start: int,
    width: int,
) -> tuple[int, list[str]]:
    """_consume_quote_block consumes consecutive markdown quote lines."""

    i = start
    quoted: list[str] = []
    indent = ""

    while i < len(body):
        line = body[i]
        if line.strip() == "":
            break
        match = QUOTE_PATTERN.match(line)
        if not match:
            break
        indent = match.group("indent")
        quoted.append(match.group("text").strip())
        i += 1

    text = " ".join(fragment for fragment in quoted if fragment)
    if not text:
        return i, [f"{indent}>"]

    prefix = f"{indent}> "
    return i, wrap_text(
        text,
        width,
        initial_indent=prefix,
        subsequent_indent=prefix,
    )


def _consume_list_item(
    body: list[str],
    start: int,
    width: int,
) -> tuple[int, list[str]]:
    """
    _consume_list_item consumes a single list item with continuation lines.

    Continuation lines are treated as part of the item until a new peer list
    item starts or a blank line is reached.
    """

    first = body[start]
    match = LIST_ITEM_PATTERN.match(first)
    if match is None:
        return start + 1, [first.rstrip()]

    indent = match.group("indent")
    marker = match.group("marker")
    content = [match.group("text").strip()]
    i = start + 1

    while i < len(body):
        line = body[i]
        if line.strip() == "":
            break
        next_match = LIST_ITEM_PATTERN.match(line)
        if next_match is not None and len(next_match.group("indent")) <= len(indent):
            break
        if FENCE_PATTERN.match(line):
            break
        if _is_trailer(line):
            break
        content.append(line.strip())
        i += 1

    text = " ".join(part for part in content if part)
    prefix = f"{indent}{marker} "
    continuation = " " * len(prefix)

    if not text:
        return i, [prefix.rstrip()]

    return i, wrap_text(
        text,
        width,
        initial_indent=prefix,
        subsequent_indent=continuation,
    )


def _consume_fence_block(body: list[str], start: int) -> tuple[int, list[str]]:
    """_consume_fence_block preserves fenced code blocks verbatim."""

    lines: list[str] = []
    i = start
    open_line = body[i].rstrip()
    lines.append(open_line)
    i += 1

    while i < len(body):
        line = body[i].rstrip()
        lines.append(line)
        if FENCE_PATTERN.match(line):
            i += 1
            break
        i += 1

    return i, lines


def _consume_indented_block(body: list[str], start: int) -> tuple[int, list[str]]:
    """_consume_indented_block preserves indented code blocks verbatim."""

    lines: list[str] = []
    i = start

    while i < len(body):
        line = body[i]
        if line.strip() == "":
            lines.append("")
            i += 1
            continue
        if _is_indented_code(line):
            lines.append(line.rstrip())
            i += 1
            continue
        break

    while lines and lines[-1] == "":
        lines.pop()

    return i, lines


def format_body(body_lines: list[str], body_width: int) -> list[str]:
    """format_body returns a wrapped body while preserving markdown structure."""

    body = _normalize_body_leading_blank(body_lines)
    body = _normalize_body_trailing_blank(body)

    if not body:
        return []

    out: list[str] = []
    i = 0
    while i < len(body):
        line = body[i]

        if line.strip() == "":
            out.append("")
            i += 1
            continue

        if FENCE_PATTERN.match(line):
            i, block = _consume_fence_block(body, i)
            out.extend(block)
            continue

        if _is_indented_code(line):
            i, block = _consume_indented_block(body, i)
            out.extend(block)
            continue

        if _is_list_item(line):
            while i < len(body):
                current = body[i]
                if current.strip() == "":
                    break
                if not _is_list_item(current):
                    break
                i, item = _consume_list_item(body, i, body_width)
                out.extend(item)
            continue

        if _is_quote_line(line):
            i, quoted = _consume_quote_block(body, i, body_width)
            out.extend(quoted)
            continue

        if _is_trailer(line):
            out.append(line.rstrip())
            i += 1
            continue

        i, para = _consume_paragraph(body, i)
        out.extend(_format_paragraph(para, body_width))

    return _collapse_redundant_blank_lines(out)


def format_message(
    message: str,
    subject_width: int = DEFAULT_SUBJECT_WIDTH,
    body_width: int = DEFAULT_BODY_WIDTH,
    decode_escaped_newlines: bool = False,
) -> str:
    """format_message normalizes spacing and wraps subject/body."""

    subject, body_lines = split_subject_body(message)
    subject = subject.strip()
    if decode_escaped_newlines:
        body_lines = _decode_escaped_newlines(body_lines)

    match = SUBJECT_PATTERN.match(subject)
    if match is None:
        wrapped_subject = wrap_text(subject, subject_width)
        subject_out = " ".join(wrapped_subject).strip()
    else:
        pkg = match.group("pkg")
        summary = match.group("summary").strip()
        prefix = f"{pkg}: "
        available = max(10, subject_width - len(prefix))
        summary_wrapped = wrap_text(summary, available)
        subject_out = prefix + " ".join(
            line.strip() for line in summary_wrapped if line.strip()
        )

    body_lines = format_body(body_lines, body_width)

    if not body_lines:
        return f"{subject_out}\n"

    return f"{subject_out}\n\n" + "\n".join(body_lines) + "\n"


def lint_message(
    message: str,
    subject_width: int = DEFAULT_SUBJECT_WIDTH,
    body_width: int = DEFAULT_BODY_WIDTH,
) -> LintResult:
    """lint_message validates message structure and line widths."""

    issues: list[LintIssue] = []
    subject, body_lines = split_subject_body(message)

    if not subject:
        issues.append(LintIssue(1, "empty commit message subject"))
        return LintResult(issues)

    if _count_columns(subject) > subject_width:
        issues.append(
            LintIssue(
                1,
                f"subject length {_count_columns(subject)} exceeds "
                f"{subject_width} chars",
            )
        )

    match = SUBJECT_PATTERN.match(subject)
    if match is None:
        issues.append(
            LintIssue(
                1,
                'subject must match "<package>: <summary>"',
            )
        )
    else:
        summary = match.group("summary")
        if summary.endswith("."):
            issues.append(
                LintIssue(1, "subject summary should not end with period")
            )

    if body_lines:
        if body_lines[0].strip() != "":
            issues.append(
                LintIssue(
                    2,
                    "body must be separated from subject by one blank line",
                )
            )
        else:
            leading_blanks = 0
            while (
                leading_blanks < len(body_lines)
                and body_lines[leading_blanks].strip() == ""
            ):
                leading_blanks += 1
            if leading_blanks > 1 and leading_blanks < len(body_lines):
                issues.append(
                    LintIssue(
                        2,
                        "use exactly one blank line between subject and body",
                    )
                )

    in_fence = False
    for idx, line in enumerate(body_lines, start=2):
        if NEWLINE_ESCAPE_PATTERN.search(line):
            issues.append(
                LintIssue(
                    idx,
                    r'found literal "\n"; use real newlines in commit body',
                )
            )

        if FENCE_PATTERN.match(line):
            in_fence = not in_fence
            continue

        if in_fence:
            continue

        if line.strip() == "":
            continue

        if _is_indented_code(line):
            continue
        if _is_trailer(line):
            continue

        width = _count_columns(line)
        if width > body_width:
            issues.append(
                LintIssue(
                    idx,
                    f"body line length {width} exceeds {body_width} chars",
                )
            )

    return LintResult(issues)


def read_input_message(args: argparse.Namespace) -> str:
    """read_input_message reads commit message input from file/stdin/commit."""

    if args.file:
        return Path(args.file).read_text(encoding="utf-8")

    if args.commit:
        return get_commit_message(args.commit)

    if not sys.stdin.isatty():
        return sys.stdin.read()

    raise ValueError("no input provided: use --file, --commit, or stdin")


def write_output_message(args: argparse.Namespace, message: str) -> None:
    """write_output_message writes formatted output to stdout or file."""

    if args.in_place and args.file:
        Path(args.file).write_text(message, encoding="utf-8")
        return

    sys.stdout.write(message)


def _lint_one(
    label: str,
    message: str,
    subject_width: int,
    body_width: int,
) -> bool:
    """_lint_one lints one message and prints issues. Returns success."""

    result = lint_message(
        message,
        subject_width=subject_width,
        body_width=body_width,
    )
    if result.ok():
        print(f"{label}: OK")
        return True

    print(f"{label}: FAIL")
    for issue in result.issues:
        print(f"  L{issue.line}: {issue.message}")
    return False


def _collect_revs(range_expr: str, include_merges: bool) -> list[str]:
    """_collect_revs returns commits in a revision range."""

    args = ["rev-list", "--reverse", range_expr]
    if not include_merges:
        args.insert(1, "--no-merges")
    out = run_git(args)
    return [line for line in out.splitlines() if line.strip()]


def lint_command(args: argparse.Namespace) -> int:
    """lint_command runs linting for a message source or commit range."""

    all_ok = True

    if args.range:
        revs = _collect_revs(args.range, include_merges=args.include_merges)
        if not revs:
            print(f"no commits found in range: {args.range}")
            return 0

        for rev in revs:
            if not args.include_merges and commit_is_merge(rev):
                continue
            subject = commit_subject(rev)
            label = f"{rev[:7]} {subject}"
            all_ok &= _lint_one(
                label,
                get_commit_message(rev),
                args.subject_width,
                args.body_width,
            )

        return 0 if all_ok else 1

    message = read_input_message(args)
    label = args.file or args.commit or "<stdin>"
    ok = _lint_one(label, message, args.subject_width, args.body_width)
    return 0 if ok else 1


def fmt_command(args: argparse.Namespace) -> int:
    """fmt_command formats a message from file/stdin/commit."""

    message = read_input_message(args)
    formatted = format_message(
        message,
        subject_width=args.subject_width,
        body_width=args.body_width,
        decode_escaped_newlines=args.decode_escaped_newlines,
    )

    if args.check:
        if message == formatted:
            return 0
        print("message is not properly formatted")
        return 1

    write_output_message(args, formatted)
    return 0


def _assert_clean_worktree() -> None:
    """_assert_clean_worktree ensures there are no staged or unstaged changes."""

    status = run_git(["status", "--porcelain"])
    if status.strip():
        raise ValueError(
            "worktree is not clean; commit rewriting requires a clean state"
        )


def _reword_head(new_message: str, no_verify: bool) -> None:
    """_reword_head amends HEAD with new_message."""

    with tempfile.NamedTemporaryFile("w", delete=False) as tmp:
        tmp.write(new_message)
        tmp_path = tmp.name

    try:
        cmd = ["git", "commit", "--amend", "-F", tmp_path]
        if no_verify:
            cmd.append("--no-verify")
        subprocess.check_call(cmd)
    finally:
        Path(tmp_path).unlink(missing_ok=True)


def _reword_non_head(target_rev: str, new_message: str, no_verify: bool) -> None:
    """_reword_non_head rewrites a non-HEAD commit via non-interactive rebase."""

    full = run_git(["rev-parse", target_rev])
    head = run_git(["rev-parse", "HEAD"])
    if full == head:
        _reword_head(new_message, no_verify=no_verify)
        return

    merge_count = run_git(["rev-list", "--count", "--merges", f"{full}..HEAD"])
    if int(merge_count) > 0:
        raise ValueError(
            "range from target commit to HEAD contains merge commits; "
            "automatic reword is disabled for this case"
        )

    is_ancestor = subprocess.call(
        ["git", "merge-base", "--is-ancestor", full, "HEAD"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    if is_ancestor != 0:
        raise ValueError(f"target commit {target_rev} is not an ancestor of HEAD")

    parent = run_git(["rev-parse", f"{full}^"])

    with tempfile.TemporaryDirectory() as tmpdir:
        tmp = Path(tmpdir)
        msg_file = tmp / "new-message.txt"
        seq_editor = tmp / "sequence-editor.py"
        msg_editor = tmp / "message-editor.py"

        msg_file.write_text(new_message, encoding="utf-8")
        seq_editor.write_text(
            textwrap.dedent(
                """\
                #!/usr/bin/env python3
                import os
                import pathlib
                import re
                import sys

                target = os.environ["COMMIT_REWORD_TARGET"]
                todo_path = pathlib.Path(sys.argv[1])
                lines = todo_path.read_text(encoding="utf-8").splitlines()

                out = []
                replaced = False
                for line in lines:
                    match = re.match(r"^(pick|p)\\s+([0-9a-f]+)(\\s+.*)$", line)
                    if match and not replaced:
                        commit = match.group(2)
                        if target.startswith(commit) or commit.startswith(target):
                            out.append(f"reword {commit}{match.group(3)}")
                            replaced = True
                            continue
                    out.append(line)

                if not replaced:
                    raise SystemExit(
                        f"could not find commit {target} in rebase todo"
                    )

                todo_path.write_text(
                    "\\n".join(out) + "\\n",
                    encoding="utf-8",
                )
                """
            ),
            encoding="utf-8",
        )
        msg_editor.write_text(
            textwrap.dedent(
                """\
                #!/usr/bin/env python3
                import pathlib
                import shutil
                import sys

                src = pathlib.Path(__import__("os").environ["COMMIT_REWORD_FILE"])
                dst = pathlib.Path(sys.argv[1])
                shutil.copyfile(src, dst)
                """
            ),
            encoding="utf-8",
        )
        seq_editor.chmod(0o755)
        msg_editor.chmod(0o755)

        env = os.environ.copy()
        env["COMMIT_REWORD_TARGET"] = full
        env["COMMIT_REWORD_FILE"] = str(msg_file)
        env["GIT_SEQUENCE_EDITOR"] = str(seq_editor)
        env["GIT_EDITOR"] = str(msg_editor)

        cmd = ["git", "rebase", "-i", parent]
        if no_verify:
            cmd.append("--no-verify")

        subprocess.check_call(cmd, env=env)


def reword_command(args: argparse.Namespace) -> int:
    """reword_command rewrites a commit message using formatted content."""

    original = get_commit_message(args.commit)
    formatted = format_message(
        original,
        subject_width=args.subject_width,
        body_width=args.body_width,
        decode_escaped_newlines=args.decode_escaped_newlines,
    )

    if original == formatted and not args.force:
        print("commit message already formatted; nothing to do")
        return 0

    if args.dry_run:
        print(formatted, end="")
        return 0

    _assert_clean_worktree()

    try:
        _reword_non_head(
            args.commit,
            formatted,
            no_verify=args.no_verify,
        )
    except subprocess.CalledProcessError as err:
        print(f"git command failed with exit status {err.returncode}")
        return err.returncode or 1
    except ValueError as err:
        print(f"error: {err}")
        return 1

    print(f"reworded commit {args.commit} with formatted message")
    return 0


def build_parser() -> argparse.ArgumentParser:
    """build_parser creates the CLI parser."""

    parser = argparse.ArgumentParser(
        description=(
            "Lint, format, and reword commit messages with markdown-aware "
            "body wrapping."
        )
    )
    parser.add_argument(
        "--subject-width",
        type=int,
        default=DEFAULT_SUBJECT_WIDTH,
        help=f"max subject width (default: {DEFAULT_SUBJECT_WIDTH})",
    )
    parser.add_argument(
        "--body-width",
        type=int,
        default=DEFAULT_BODY_WIDTH,
        help=f"max body width (default: {DEFAULT_BODY_WIDTH})",
    )

    sub = parser.add_subparsers(dest="cmd", required=True)

    lint = sub.add_parser("lint", help="lint a message or commit range")
    lint.add_argument("--file", help="path to commit message file")
    lint.add_argument("--commit", help="commit revision to lint")
    lint.add_argument(
        "--range",
        help="lint all commits in revision range (example: origin/main..HEAD)",
    )
    lint.add_argument(
        "--include-merges",
        action="store_true",
        help="include merge commits when linting --range",
    )
    lint.set_defaults(func=lint_command)

    fmt = sub.add_parser("fmt", help="format a message from file/stdin/commit")
    fmt.add_argument("--file", help="path to commit message file")
    fmt.add_argument("--commit", help="commit revision to read as input")
    fmt.add_argument(
        "--in-place",
        action="store_true",
        help="write formatted output back to --file",
    )
    fmt.add_argument(
        "--check",
        action="store_true",
        help="exit non-zero if formatting changes would be applied",
    )
    fmt.add_argument(
        "--decode-escaped-newlines",
        action="store_true",
        help='decode literal "\\n" body sequences before formatting',
    )
    fmt.set_defaults(func=fmt_command)

    reword = sub.add_parser(
        "reword",
        help="reword an existing commit with its formatted message",
    )
    reword.add_argument(
        "--commit",
        default="HEAD",
        help="commit revision to reword (default: HEAD)",
    )
    reword.add_argument(
        "--force",
        action="store_true",
        help="rewrite even if message is already formatted",
    )
    reword.add_argument(
        "--no-verify",
        action="store_true",
        help="pass --no-verify to git amend/rebase reword operations",
    )
    reword.add_argument(
        "--decode-escaped-newlines",
        action="store_true",
        help='decode literal "\\n" body sequences before rewording',
    )
    reword.add_argument(
        "--dry-run",
        action="store_true",
        help="print the formatted message instead of rewriting commits",
    )
    reword.set_defaults(func=reword_command)

    return parser


def main(argv: Iterable[str] | None = None) -> int:
    """main parses arguments and dispatches subcommands."""

    parser = build_parser()
    args = parser.parse_args(argv)

    try:
        return int(args.func(args))
    except ValueError as err:
        print(f"error: {err}")
        return 1


if __name__ == "__main__":
    sys.exit(main())