gen_commit_msg

#!/usr/bin/env python3
"""Generate commit messages using a language model,
or only print prompt if --stdout is given script"""

import os
import subprocess
import sys
import argparse
import logging
from subprocess import Popen, PIPE

# Constants
DEFAULT_MODEL = "qwen2.5-coder:14b"
DEFAULT_ATTEMPTS = 1
MAX_LINE_LENGTH = 74

# Configure logging
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger(__name__)


def get_git_diff() -> str:
    """Returns the staged git diff as a string.

    Raises:
        subprocess.CalledProcessError: If the git diff command fails.
    """
    custom_env = os.environ.copy()
    custom_env["GIT_PAGER"] = "cat"
    result = subprocess.run(
        ["git", "diff", "--cached"],
        capture_output=True,
        text=True,
        check=True,
        env=custom_env,
    )
    return result.stdout.strip()


def prompt_llm(
    prompt: str, model: str = DEFAULT_MODEL, attempts: int = DEFAULT_ATTEMPTS
) -> list[str]:
    """
    Sends a prompt to the specified language model and returns the generated messages.

    Args:
        prompt (str): The input prompt for the language model.
        model (str): The name of the language model to use. Defaults to DEFAULT_MODEL.
        attempts (int): The number of times to attempt generating a message.
        Defaults to DEFAULT_ATTEMPTS.

    Returns:
        list[str]: A list of generated messages from the language model.

    Raises:
        RuntimeError: If the LLM process fails.
    """
    messages = []
    for i in range(attempts):
        logger.debug("=> LLM generation attempt '%s'", i)
        with Popen(
            ["ask", prompt], stdout=PIPE, bufsize=1, universal_newlines=True
        ) as p:
            if p.stdout is None:
                continue

            output = []
            for line in p.stdout:
                logger.debug(line.rstrip())
                output.append(line)

            captured_output = "".join(output).strip()
            messages.append(f"{captured_output}\n")

    return messages


def commit(messages: list[str]) -> int:
    """
    Commits the staged changes with the provided commit messages.

    Args:
        messages (list[str]): A list of commit messages.

    Returns:
        int: The return code of the git commit command.
    """
    commit_message = ""
    for i, message in enumerate(messages):
        if DEFAULT_ATTEMPTS > 1:
            commit_message += f"#===Commit message generated #{i} ===\n"
        commit_message += message
        if DEFAULT_ATTEMPTS > 1:
            commit_message += f"#{'=' * 40}\n"
    result = subprocess.run(
        ["git", "commit", "-s", "-e", "-m", commit_message], check=False
    )
    return result.returncode


def generate_prompt(diff: str, context: str = "", use_emoji: bool = False) -> str:
    """
    Generate a high-quality prompt for commit message generation.

    Args:
        diff (str): The staged git diff output.
        context (str): Any extra context for the LLM.
        use_emoji (bool): Whether to request GitMoji convention.

    Returns:
        str: The final prompt for the language model.

    Raises:
        ValueError: If diff is empty.
    """
    if not diff:
        raise ValueError("The diff is empty. Please stage changes before running.")

    type_instructions = (
        "Use GitMoji to preface the commit, e.g.: 🐛 fix, ✨ feat, 📝 docs, ♻️ refactor, 🔧 config, etc.\n"
        if use_emoji
        else "Do not preface with emoji. Use one of: feat, fix, docs, style, refactor, test, chore, ci, perf, build.\n"
    )

    commit_rules = (
        "Follow Conventional Commits: <type>[optional scope]: <description>.\n"
        "Types: feat, fix, docs, style, refactor, perf, test, build, chore, ci.\n"
        "Scope (in parentheses) only if relevant.\n"
        "For breaking changes, add ! or BREAKING CHANGE: footer.\n"
        "Description must be imperative, lower case, no period.\n"
        "Body explains what and why. Include footers as needed.\n"
        f"Line length <= {MAX_LINE_LENGTH}.\n"
        "Output commit message only—no extra commentary, code blocks, or 'this commit ...' statements. Use English."
    )

    context_block = (
        f"Extra context:\n<context>\n{context}\n</context>\n" if context else ""
    )

    prompt = (
        "You are a commit message generator powered by the Conventional Commits specification.\n"
        "Convert the following git diff into a concise, best-practice commit message.\n"
        f"{type_instructions}"
        f"{commit_rules}\n"
        f"{context_block}"
        "<diff>\n"
        f"{diff}\n"
        "</diff>"
    )
    return prompt.strip()


def main() -> None:
    """
    Main function to execute the script. Parses command-line arguments, generates a prompt,
    and either prints it or sends it to a language model for processing.
    """
    parser = argparse.ArgumentParser(
        description=(
            "Generate commit messages using a language model,"
            " or only print prompt if --stdout is given."
        )
    )
    parser.add_argument(
        "-c", "--context", help="Extra context to feed to prompt.", type=str, default=""
    )
    parser.add_argument(
        "-n",
        "--attempts",
        help="Number of generations asked.",
        type=int,
        default=DEFAULT_ATTEMPTS,
    )
    parser.add_argument(
        "-m",
        "--model",
        help=f"Set model to use, default is {DEFAULT_MODEL}",
        type=str,
        default=DEFAULT_MODEL,
    )
    parser.add_argument(
        "--stdout",
        help="If given, echoes prompt to stdout instead of sending it to the model.",
        action="store_true",
    )
    parser.add_argument(
        "-e", "--emoji", help="Uses emojis if the flag is given.", action="store_true"
    )
    args = parser.parse_args()

    try:
        diff = get_git_diff()
        prompt = generate_prompt(diff, args.context, use_emoji=args.emoji)

        if args.stdout:
            print(prompt)
            sys.exit(0)

        commit_messages = prompt_llm(prompt, attempts=args.attempts, model=args.model)
        ret = commit(commit_messages)
        sys.exit(ret)

    except (subprocess.CalledProcessError, ValueError, RuntimeError) as e:
        logger.error("Error: %s", e)
        sys.exit(1)


if __name__ == "__main__":
    main()