Carlos Florêncio
123 lines
3.8 KiB
Python
123 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Build a GitHub API payload for the owned PR walkthrough callout."""
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import sys
|
|
from typing import Any
|
|
|
|
|
|
START_MARKER = "<!-- kandev-pr-walkthrough-start -->"
|
|
END_MARKER = "<!-- kandev-pr-walkthrough-end -->"
|
|
SHA_PATTERN = re.compile(r"^[0-9a-f]{40}$")
|
|
PUBLIC_SHA_LENGTH = 12
|
|
UNCHANGED = 3
|
|
NO_WALKTHROUGH = 4
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--github-response", type=Path, required=True)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
parser.add_argument("--url", required=True)
|
|
parser.add_argument("--pr-number", type=int, required=True)
|
|
parser.add_argument("--head-sha", required=True)
|
|
parser.add_argument(
|
|
"--require-existing",
|
|
action="store_true",
|
|
help="return a no-op when the PR body has no owned walkthrough block",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def require_mapping(value: Any, name: str) -> dict[str, Any]:
|
|
if not isinstance(value, dict):
|
|
raise ValueError(f"{name} must be a JSON object")
|
|
return value
|
|
|
|
|
|
def load_pr(path: Path, pr_number: int) -> str:
|
|
response = require_mapping(json.loads(path.read_text(encoding="utf-8")), "response")
|
|
if response.get("number") != pr_number:
|
|
raise ValueError("pull request number does not match the workflow event")
|
|
|
|
body = response.get("body")
|
|
if body is None:
|
|
return ""
|
|
if not isinstance(body, str):
|
|
raise ValueError("pull request body must be a string or null")
|
|
return body
|
|
|
|
|
|
def validate_url(url: str, pr_number: int, head_sha: str) -> None:
|
|
if pr_number <= 0:
|
|
raise ValueError("pull request number must be positive")
|
|
if not SHA_PATTERN.fullmatch(head_sha):
|
|
raise ValueError("head SHA must be 40 lowercase hexadecimal characters")
|
|
expected = (
|
|
"https://walkthrough.kandev.ai/"
|
|
f"pr/{pr_number}/{head_sha[:PUBLIC_SHA_LENGTH]}.html"
|
|
)
|
|
if url != expected:
|
|
raise ValueError(f"expected walkthrough URL {expected}")
|
|
|
|
|
|
def walkthrough_block(url: str) -> str:
|
|
return "\n".join(
|
|
(
|
|
START_MARKER,
|
|
"> [!TIP]",
|
|
f"> **PR walkthrough:** [Open the visual walkthrough]({url})",
|
|
END_MARKER,
|
|
)
|
|
)
|
|
|
|
|
|
def update_body(body: str, url: str) -> str:
|
|
start_count = body.count(START_MARKER)
|
|
end_count = body.count(END_MARKER)
|
|
block = walkthrough_block(url)
|
|
|
|
if start_count == 0 and end_count == 0:
|
|
return block if not body else f"{block}\n\n{body}"
|
|
if start_count != 1 or end_count != 1:
|
|
raise ValueError("walkthrough marker block must have one start and one end marker")
|
|
if not body.startswith(START_MARKER):
|
|
raise ValueError("walkthrough marker block must remain at the beginning of the body")
|
|
|
|
end = body.find(END_MARKER)
|
|
if end < len(START_MARKER):
|
|
raise ValueError("walkthrough marker block is malformed")
|
|
suffix = body[end + len(END_MARKER) :]
|
|
return f"{block}{suffix}"
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
try:
|
|
validate_url(args.url, args.pr_number, args.head_sha)
|
|
body = load_pr(args.github_response, args.pr_number)
|
|
if (
|
|
args.require_existing
|
|
and START_MARKER not in body
|
|
and END_MARKER not in body
|
|
):
|
|
return NO_WALKTHROUGH
|
|
updated = update_body(body, args.url)
|
|
if updated == body:
|
|
return UNCHANGED
|
|
args.output.write_text(
|
|
json.dumps({"body": updated}, ensure_ascii=False) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
except (OSError, ValueError, json.JSONDecodeError) as error:
|
|
print(f"error: {error}", file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|