refactor: clean up PR commit validation flow

Author: shenxianpeng

main.py

@@ -1,14 +1,15 @@
 #!/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"
 
 # Environment variables
 MESSAGE = os.getenv("MESSAGE", "false")
@@ -19,9 +20,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,27 +47,74 @@ def log_env_vars():
     print(f"PR_COMMENTS = {PR_COMMENTS}\n")
 
 
+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.rstrip("\n")
+        for message in output.split(COMMIT_MESSAGE_DELIMITER)
+        if message.rstrip("\n")
+    ]
+
+
+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 (pull_request event only).
+    """Get all commit messages for the current PR workflow.
 
-    In a pull_request event, actions/checkout checks out a synthetic merge
+    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 os.getenv("GITHUB_EVENT_NAME", "") != "pull_request":
+    if not is_pr_event():
         return []
+
     try:
-        result = subprocess.run(
-            ["git", "log", "--pretty=format:%B%x00", "HEAD^1..HEAD^2"],
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-            encoding="utf-8",
-            check=False,
-        )
-        if result.returncode == 0 and result.stdout:
-            return [
-                m.rstrip("\n") for m in result.stdout.split("\x00") if m.rstrip("\n")
-            ]
+        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}",
@@ -64,73 +123,72 @@ def get_pr_commit_messages() -> list[str]:
     return []
 
 
-def build_check_args(
-    message: str, branch: str, author_name: str, author_email: str
-) -> list[str]:
-    """Maps 'true'/'false' flag values to CLI argument list."""
-    flags = ["--message", "--branch", "--author-name", "--author-email"]
-    values = [message, branch, author_name, author_email]
-    return [flag for flag, value in zip(flags, values) if value == "true"]
+def run_check_command(
+    args: list[str], result_file: TextIO, input_text: str | None = None
+) -> int:
+    """Run commit-check and write both stdout and stderr to the result file."""
+    command = ["commit-check"] + args
+    print(" ".join(command))
+    result = subprocess.run(
+        command,
+        input=input_text,
+        stdout=result_file,
+        stderr=subprocess.STDOUT,
+        text=True,
+        check=False,
+    )
+    return result.returncode
 
 
-def run_pr_message_checks(pr_messages: list[str], result_file) -> int:  # type: ignore[type-arg]
+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
-    for msg in pr_messages:
-        result = subprocess.run(
-            ["commit-check", "--message"],
-            input=msg,
-            stdout=result_file,
-            stderr=subprocess.PIPE,
-            text=True,
-            check=False,
-        )
-        has_failure = has_failure or (result.returncode != 0)
+    total_messages = len(pr_messages)
+    for index, msg in enumerate(pr_messages, start=1):
+        subject = msg.splitlines()[0] if msg else "<empty commit message>"
+        result_file.write(f"\n--- Commit {index}/{total_messages}: {subject}\n")
+        has_failure = run_check_command(
+            ["--message"], result_file, input_text=msg
+        ) != 0 or has_failure
     return 1 if has_failure else 0
 
 
-def run_other_checks(args: list[str], result_file) -> int:  # type: ignore[type-arg]
+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
-    command = ["commit-check"] + args
-    print(" ".join(command))
-    result = subprocess.run(
-        command, stdout=result_file, stderr=subprocess.PIPE, text=True, check=False
-    )
-    return result.returncode
+    return run_check_command(args, result_file)
 
 
-def run_default_checks(args: list[str], result_file) -> int:  # type: ignore[type-arg]
-    """Runs all checks at once (non-PR context or message disabled)."""
-    command = ["commit-check"] + args
-    print(" ".join(command))
-    result = subprocess.run(
-        command, stdout=result_file, stderr=subprocess.PIPE, text=True, check=False
-    )
-    return result.returncode
+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(MESSAGE, BRANCH, AUTHOR_NAME, AUTHOR_EMAIL)
-    total_rc = 0
+    args = build_check_args()
     with open("result.txt", "w") as result_file:
-        if MESSAGE == "true":
+        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.
-                total_rc += run_pr_message_checks(pr_messages, result_file)
+                message_rc = run_pr_message_checks(pr_messages, result_file)
                 other_args = [a for a in args if a != "--message"]
-                total_rc += run_other_checks(other_args, result_file)
-                return total_rc
+                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
-        total_rc += run_default_checks(args, result_file)
-    return total_rc
+        return 1 if run_check_command(args, result_file) else 0
 
 
 def read_result_file() -> str | None:
@@ -144,21 +202,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
 
@@ -183,7 +242,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;
@@ -199,6 +258,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")
@@ -214,15 +275,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
@@ -282,12 +337,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()