fix: implement PR commit message retrieval and validation

main.py

@@ -1,14 +1,16 @@
 #!/usr/bin/env python3
 import json
 import os
-import sys
-import subprocess
 import re
-from github import Github, Auth, GithubException  # type: ignore
+import subprocess
+import sys
+from typing import TextIO
 
 # Constants for message titles
 SUCCESS_TITLE = "# Commit-Check ✔️"
 FAILURE_TITLE = "# Commit-Check ❌"
+COMMIT_MESSAGE_DELIMITER = "\x00"
+COMMIT_SECTION_SEPARATOR = "\n---\n"
 
 # Environment variables
 MESSAGE = os.getenv("MESSAGE", "false")
@@ -19,9 +21,20 @@
 JOB_SUMMARY = os.getenv("JOB_SUMMARY", "false")
 PR_COMMENTS = os.getenv("PR_COMMENTS", "false")
 GITHUB_STEP_SUMMARY = os.environ["GITHUB_STEP_SUMMARY"]
-GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
-GITHUB_REPOSITORY = os.getenv("GITHUB_REPOSITORY")
-GITHUB_REF = os.getenv("GITHUB_REF")
+
+
+def env_flag(name: str, default: str = "false") -> bool:
+    """Read a GitHub Action boolean-style environment variable."""
+    return os.getenv(name, default).lower() == "true"
+
+
+MESSAGE_ENABLED = env_flag("MESSAGE")
+BRANCH_ENABLED = env_flag("BRANCH")
+AUTHOR_NAME_ENABLED = env_flag("AUTHOR_NAME")
+AUTHOR_EMAIL_ENABLED = env_flag("AUTHOR_EMAIL")
+DRY_RUN_ENABLED = env_flag("DRY_RUN")
+JOB_SUMMARY_ENABLED = env_flag("JOB_SUMMARY")
+PR_COMMENTS_ENABLED = env_flag("PR_COMMENTS")
 
 
 def log_env_vars():
@@ -35,35 +48,170 @@ def log_env_vars():
     print(f"PR_COMMENTS = {PR_COMMENTS}\n")
 
 
-def run_commit_check() -> int:
-    """Runs the commit-check command and logs the result."""
-    args = [
-        "--message",
-        "--branch",
-        "--author-name",
-        "--author-email",
+def is_pr_event() -> bool:
+    """Return whether the workflow was triggered by a PR-style event."""
+    return os.getenv("GITHUB_EVENT_NAME", "") in {"pull_request", "pull_request_target"}
+
+
+def parse_commit_messages(output: str) -> list[str]:
+    """Split git log output into individual commit messages."""
+    return [
+        message.strip("\n")
+        for message in output.split(COMMIT_MESSAGE_DELIMITER)
+        if message.strip("\n")
     ]
-    args = [
-        arg
-        for arg, value in zip(
-            args,
-            [
-                MESSAGE,
-                BRANCH,
-                AUTHOR_NAME,
-                AUTHOR_EMAIL,
-            ],
+
+
+def get_messages_from_merge_ref() -> list[str]:
+    """Read PR commit messages from GitHub's synthetic merge commit."""
+    result = subprocess.run(
+        ["git", "log", "--pretty=format:%B%x00", "--reverse", "HEAD^1..HEAD^2"],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        encoding="utf-8",
+        check=False,
+    )
+    if result.returncode == 0 and result.stdout:
+        return parse_commit_messages(result.stdout)
+    return []
+
+
+def get_messages_from_head_ref(base_ref: str) -> list[str]:
+    """Read PR commit messages when the workflow checks out the head SHA."""
+    result = subprocess.run(
+        [
+            "git",
+            "log",
+            "--pretty=format:%B%x00",
+            "--reverse",
+            f"origin/{base_ref}..HEAD",
+        ],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        encoding="utf-8",
+        check=False,
+    )
+    if result.returncode == 0 and result.stdout:
+        return parse_commit_messages(result.stdout)
+    return []
+
+
+def get_pr_commit_messages() -> list[str]:
+    """Get all commit messages for the current PR workflow.
+
+    In pull_request-style workflows, actions/checkout checks out a synthetic merge
+    commit (HEAD = merge of PR branch into base). HEAD^1 is the base branch
+    tip, HEAD^2 is the PR branch tip. So HEAD^1..HEAD^2 gives all PR commits.
+    If the workflow explicitly checks out the PR head SHA instead, fall back to
+    diffing against origin/<base-ref> when that ref is available locally.
+    """
+    if not is_pr_event():
+        return []
+
+    try:
+        messages = get_messages_from_merge_ref()
+        if messages:
+            return messages
+
+        base_ref = os.getenv("GITHUB_BASE_REF", "")
+        if base_ref:
+            return get_messages_from_head_ref(base_ref)
+    except Exception as e:
+        print(
+            f"::warning::Failed to retrieve PR commit messages: {e}",
+            file=sys.stderr,
         )
-        if value == "true"
-    ]
+    return []
 
+
+def run_check_command(
+    args: list[str],
+    result_file: TextIO,
+    input_text: str | None = None,
+    output_prefix: str | None = None,
+) -> int:
+    """Run commit-check and write both stdout and stderr to the result file."""
     command = ["commit-check"] + args
     print(" ".join(command))
-    with open("result.txt", "w") as result_file:
-        result = subprocess.run(
-            command, stdout=result_file, stderr=subprocess.PIPE, check=False
+    result = subprocess.run(
+        command,
+        input=input_text,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        check=False,
+    )
+    if result.stdout:
+        if output_prefix:
+            result_file.write(output_prefix)
+        result_file.write(result.stdout.rstrip("\n"))
+        result_file.write("\n")
+    return result.returncode
+
+
+def run_pr_message_checks(pr_messages: list[str], result_file: TextIO) -> int:
+    """Checks each PR commit message individually via commit-check --message.
+
+    Returns 1 if any message fails, 0 if all pass.
+    """
+    has_failure = False
+    emitted_failure_output = False
+    total = len(pr_messages)
+    for index, msg in enumerate(pr_messages, start=1):
+        command_args = ["--message"]
+        if emitted_failure_output:
+            command_args.append("--no-banner")
+
+        if emitted_failure_output:
+            output_prefix = f"\n--- Commit {index}/{total}:\n"
+        else:
+            output_prefix = None
+
+        return_code = run_check_command(
+            command_args,
+            result_file,
+            input_text=msg,
+            output_prefix=output_prefix,
         )
-        return result.returncode
+        if return_code != 0:
+            has_failure = True
+            emitted_failure_output = True
+    return 1 if has_failure else 0
+
+
+def run_other_checks(args: list[str], result_file: TextIO) -> int:
+    """Runs non-message checks (branch, author) once. Returns 0 if args is empty."""
+    if not args:
+        return 0
+    return run_check_command(args, result_file)
+
+
+def build_check_args() -> list[str]:
+    """Map enabled validation switches to commit-check CLI arguments."""
+    flags = [
+        ("--message", MESSAGE_ENABLED),
+        ("--branch", BRANCH_ENABLED),
+        ("--author-name", AUTHOR_NAME_ENABLED),
+        ("--author-email", AUTHOR_EMAIL_ENABLED),
+    ]
+    return [flag for flag, enabled in flags if enabled]
+
+
+def run_commit_check() -> int:
+    """Runs the commit-check command and logs the result."""
+    args = build_check_args()
+    with open("result.txt", "w") as result_file:
+        if MESSAGE_ENABLED:
+            pr_messages = get_pr_commit_messages()
+            if pr_messages:
+                # In PR context: check each commit message individually to avoid
+                # only validating the synthetic merge commit at HEAD.
+                message_rc = run_pr_message_checks(pr_messages, result_file)
+                other_args = [a for a in args if a != "--message"]
+                other_rc = run_other_checks(other_args, result_file)
+                return 1 if message_rc or other_rc else 0
+        # Non-PR context or message disabled: run all checks at once
+        return 1 if run_check_command(args, result_file) else 0
 
 
 def read_result_file() -> str | None:
@@ -77,21 +225,22 @@ def read_result_file() -> str | None:
     return None
 
 
+def build_result_body(result_text: str | None) -> str:
+    """Create the human-readable result body used in summaries and PR comments."""
+    if result_text is None:
+        return SUCCESS_TITLE
+    return f"{FAILURE_TITLE}\n```\n{result_text}\n```"
+
+
 def add_job_summary() -> int:
     """Adds the commit check result to the GitHub job summary."""
-    if JOB_SUMMARY == "false":
+    if not JOB_SUMMARY_ENABLED:
         return 0
 
     result_text = read_result_file()
 
-    summary_content = (
-        SUCCESS_TITLE
-        if result_text is None
-        else f"{FAILURE_TITLE}\n```\n{result_text}\n```"
-    )
-
     with open(GITHUB_STEP_SUMMARY, "a") as summary_file:
-        summary_file.write(summary_content)
+        summary_file.write(build_result_body(result_text))
 
     return 0 if result_text is None else 1
 
@@ -116,7 +265,7 @@ def is_fork_pr() -> bool:
 
 def add_pr_comments() -> int:
     """Posts the commit check result as a comment on the pull request."""
-    if PR_COMMENTS == "false":
+    if not PR_COMMENTS_ENABLED:
         return 0
 
     # Fork PRs triggered by the pull_request event receive a read-only token;
@@ -132,6 +281,8 @@ def add_pr_comments() -> int:
         return 0
 
     try:
+        from github import Auth, Github, GithubException  # type: ignore
+
         token = os.getenv("GITHUB_TOKEN")
         repo_name = os.getenv("GITHUB_REPOSITORY")
         pr_number = os.getenv("GITHUB_REF")
@@ -147,15 +298,9 @@ def add_pr_comments() -> int:
         repo = g.get_repo(repo_name)
         pull_request = repo.get_issue(int(pr_number))
 
-        # Prepare comment content
         result_text = read_result_file()
-        pr_comment_body = (
-            SUCCESS_TITLE
-            if result_text is None
-            else f"{FAILURE_TITLE}\n```\n{result_text}\n```"
-        )
+        pr_comment_body = build_result_body(result_text)
 
-        # Fetch all existing comments on the PR
         comments = pull_request.get_comments()
         matching_comments = [
             c
@@ -215,12 +360,9 @@ def main():
     """Main function to run commit-check, add job summary and post PR comments."""
     log_env_vars()
 
-    # Combine return codes
-    ret_code = run_commit_check()
-    ret_code += add_job_summary()
-    ret_code += add_pr_comments()
+    ret_code = max(run_commit_check(), add_job_summary(), add_pr_comments())
 
-    if DRY_RUN == "true":
+    if DRY_RUN_ENABLED:
         ret_code = 0
 
     result_text = read_result_file()