NextERP-scripts/scripts/interpreter/gen_commit.py

import asyncio
import os
import sys
from dotenv import load_dotenv
from interpreter import OpenInterpreter

load_dotenv()


def get_git_credentials():
    if len(sys.argv) != 4:
        print("Usage: python gen_commit.py <GIT_USER> <GIT_PASS> <PROJECT_PATH>")
        sys.exit(1)
    return sys.argv[1], sys.argv[2], sys.argv[3]


def get_git_diff(project_path):
    """Run git diff and return the output."""
    result = os.popen(f'cd "{project_path}" && git diff --compact-summary').read()
    return result.strip()


def gen_commit(project_path):
    """Generate commit message using OpenInterpreter based on git diff output."""
    agent = OpenInterpreter()
    agent.llm.model = "gpt-4o"
    agent.auto_run = True

    convention_path = os.path.join(os.path.dirname(__file__), "resources/commit_convention.md")
    with open(convention_path, "r") as f:
        convention_content = f.read()

    agent.system_message = f"""
You are a helpful assistant that generates concise, conventional commit messages based on code diffs.

Guidelines:
- Use English.
- Do not include explanations—only output commit message in under 100 characters.
- Follow this convention:
{convention_content}
"""

    diff_output = get_git_diff(project_path)
    if not diff_output:
        return "chore: no changes to commit"

    prompt = f"""
Analyze the following git diff and generate a commit message:


{diff_output}
"""

    response = agent.chat(prompt)
    if isinstance(response, list) and response and isinstance(response[-1], dict) and "content" in response[-1]:
        return response[-1]["content"].strip()
    return str(response).strip()


async def has_submodules():
    """Check if the repository has any submodules."""
    proc = await asyncio.create_subprocess_exec(
        "git", "config", "--file", ".gitmodules", "--get-regexp", r"^submodule\.",
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    return bool(stdout.decode().strip())


async def get_submodule_info():
    if not await has_submodules():
        return {}

    proc = await asyncio.create_subprocess_exec(
        "git", "config", "--file", ".gitmodules", "--get-regexp", r"^submodule\..*\.path$",
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()

    if stderr:
        print(f"Error getting submodule paths:\n{stderr.decode()}")
        return {}

    submodule_info = {}
    for line in stdout.decode().splitlines():
        parts = line.split()
        if len(parts) == 2:
            name = parts[0].split('.')[1]
            path = parts[1]
            abs_path = os.path.abspath(path)
            submodule_info[abs_path] = name
    return submodule_info


async def commit_and_push_submodules():
    if not await has_submodules():
        print("No submodules found in this repository.")
        return

    print("Checking for submodule changes...")
    submodule_info = await get_submodule_info()

    proc = await asyncio.create_subprocess_shell(
        "git submodule foreach --quiet 'if [ -n \"$(git status --porcelain)\" ]; then echo $path; fi'",
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    if stderr:
        print(f"Error checking submodules:\n{stderr.decode()}")
        return

    submodule_paths = stdout.decode().strip().split('\n')
    git_user, git_pass, _ = get_git_credentials()

    for submodule_path in submodule_paths:
        if not submodule_path:
            continue

        abs_path = os.path.abspath(submodule_path)
        name = submodule_info.get(abs_path, submodule_path)
        print(f"Processing submodule: {name} at {abs_path}")

        commit_msg = gen_commit(abs_path)

        proc = await asyncio.create_subprocess_exec(
            "git", "remote", "get-url", "origin",
            cwd=abs_path, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await proc.communicate()
        if stderr:
            print(f"Error getting remote URL for {name}:\n{stderr.decode()}")
            continue

        remote_url = stdout.decode().strip().replace(".git", "").replace("https://", "")
        remote_url = f"https://{git_user}:{git_pass}@{remote_url}"

        commands = [
            f"cd {abs_path} && git add .",
            f'cd {abs_path} && git commit -m "{commit_msg}" || true',
            f"cd {abs_path} && git push {remote_url} || true",
        ]

        for cmd in commands:
            proc = await asyncio.create_subprocess_shell(
                cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
            )
            stdout, stderr = await proc.communicate()
            if stdout:
                print(stdout.decode())
            if stderr:
                print(f"Submodule '{name}' stderr:\n{stderr.decode()}")


async def push_code():
    git_user, git_pass, project_path = get_git_credentials()
    os.chdir(project_path)
    print(f"Changed to project path: {project_path}")

    proc = await asyncio.create_subprocess_exec(
        "git", "remote", "get-url", "origin",
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    if stderr:
        print(f"Failed to get remote URL: {stderr.decode()}")
    git_repo = stdout.decode().strip().replace(".git", "").replace("https://", "")
    remote_url = f"https://{git_user}:{git_pass}@{git_repo}"

    await commit_and_push_submodules()

    commit_msg = gen_commit(project_path)
    print(f"Generated commit message:\n{commit_msg}\n")

    commands = [
        f"git pull {remote_url} || true",
        "git add .",
        f'git commit -m "{commit_msg}" || true',
        f"git push {remote_url} || true",
    ]

    for cmd in commands:
        proc = await asyncio.create_subprocess_shell(
            cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
        )
        stdout, stderr = await proc.communicate()
    if proc.returncode != 0:
        print(f"Command failed: {cmd}")
        print(stderr.decode())
    elif stderr:
        # Not an error; just info output
        print(stderr.decode())


if __name__ == "__main__":
    asyncio.run(push_code())