You've already forked ReleaseNotifiableRepo
219 lines
7.2 KiB
Python
219 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
create_repo_from_issue.py
|
|
|
|
Parses an AllSpice/Gitea issue body (from the new-repo-request template),
|
|
validates the inputs, calls the AllSpice API to generate a repo from a
|
|
template, and posts a comment back on the issue.
|
|
|
|
Environment variables (all required):
|
|
ALLSPICE_TOKEN Personal access token with repo scope
|
|
ALLSPICE_URL Base URL, e.g. https://hub.allspice.io
|
|
ISSUE_BODY Raw body text of the issue
|
|
ISSUE_NUMBER Issue number (for posting comment)
|
|
WORKFLOW_REPO owner/repo of the repo running this workflow
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from urllib.request import Request, urlopen
|
|
from urllib.error import HTTPError, URLError
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
ALLSPICE_TOKEN = os.environ["ALLSPICE_TOKEN"]
|
|
ALLSPICE_URL = os.environ.get("ALLSPICE_URL", "https://hub.allspice.io").rstrip("/")
|
|
ISSUE_BODY = os.environ["ISSUE_BODY"]
|
|
ISSUE_NUMBER = os.environ["ISSUE_NUMBER"]
|
|
WORKFLOW_REPO = os.environ["WORKFLOW_REPO"] # e.g. "my-org/automation-repo"
|
|
|
|
TEMPLATE_OWNER = "EXT-CHG"
|
|
TEMPLATE_REPO = "main-repo-template"
|
|
|
|
VALID_ORGS = {"Entacc", "EXT-CHG", "Patio", "Expat"}
|
|
VALID_PROGRAMS = {"Vegas", "MJ", "Yosemite"}
|
|
|
|
DESIGN_NAME_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_-]*[a-zA-Z0-9]$")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def api(method: str, path: str, payload: dict | None = None) -> dict:
|
|
"""Make an authenticated AllSpice API call, return parsed JSON."""
|
|
url = f"{ALLSPICE_URL}/api/v1{path}"
|
|
body = json.dumps(payload).encode() if payload is not None else None
|
|
req = Request(
|
|
url,
|
|
data=body,
|
|
method=method,
|
|
headers={
|
|
"Authorization": f"token {ALLSPICE_TOKEN}",
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
},
|
|
)
|
|
try:
|
|
with urlopen(req) as resp:
|
|
return json.loads(resp.read())
|
|
except HTTPError as e:
|
|
body_text = e.read().decode(errors="replace")
|
|
raise RuntimeError(f"HTTP {e.code} {method} {path}: {body_text}") from e
|
|
except URLError as e:
|
|
raise RuntimeError(f"Network error {method} {path}: {e.reason}") from e
|
|
|
|
|
|
def post_comment(body: str) -> None:
|
|
owner, repo = WORKFLOW_REPO.split("/", 1)
|
|
api("POST", f"/repos/{owner}/{repo}/issues/{ISSUE_NUMBER}/comments", {"body": body})
|
|
|
|
|
|
def close_issue() -> None:
|
|
owner, repo = WORKFLOW_REPO.split("/", 1)
|
|
api("PATCH", f"/repos/{owner}/{repo}/issues/{ISSUE_NUMBER}", {"state": "closed"})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Parsing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def extract_field(body: str, label: str) -> str:
|
|
"""
|
|
Extract the value that follows a ### {label} header in a Gitea issue form body.
|
|
Gitea renders form fields as:
|
|
|
|
### Field Label
|
|
|
|
Field value
|
|
|
|
Returns the stripped value, or raises if not found / empty.
|
|
"""
|
|
pattern = re.compile(
|
|
rf"^###\s+{re.escape(label)}\s*$\n+(.+?)(?:\n|$)",
|
|
re.MULTILINE | re.IGNORECASE,
|
|
)
|
|
m = pattern.search(body)
|
|
if not m:
|
|
raise ValueError(f"Field '{label}' not found in issue body")
|
|
value = m.group(1).strip()
|
|
if not value or value == "_No response_":
|
|
raise ValueError(f"Field '{label}' is empty")
|
|
return value
|
|
|
|
|
|
def parse_issue(body: str) -> dict:
|
|
program_name = extract_field(body, "Program Name")
|
|
design_name = extract_field(body, "Design Name")
|
|
return {
|
|
"org_name": extract_field(body, "Organization"),
|
|
"program_name": program_name,
|
|
"design_name": design_name,
|
|
"repo_name": f"{program_name}_{design_name}",
|
|
"description": extract_field(body, "Description"),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def validate(fields: dict) -> list[str]:
|
|
errors = []
|
|
|
|
if fields["org_name"] not in VALID_ORGS:
|
|
errors.append(f"Organization `{fields['org_name']}` is not valid. Must be one of: {', '.join(sorted(VALID_ORGS))}.")
|
|
|
|
if fields["program_name"] not in VALID_PROGRAMS:
|
|
errors.append(f"Program `{fields['program_name']}` is not valid. Must be one of: {', '.join(sorted(VALID_PROGRAMS))}.")
|
|
|
|
design = fields["design_name"]
|
|
if not DESIGN_NAME_RE.match(design):
|
|
errors.append(
|
|
f"Design name `{design}` is invalid. "
|
|
"Use letters, numbers, hyphens, and underscores only. "
|
|
"Must start and end with a letter or number."
|
|
)
|
|
|
|
return errors
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def main() -> None:
|
|
# --- Parse ---
|
|
try:
|
|
fields = parse_issue(ISSUE_BODY)
|
|
except ValueError as e:
|
|
msg = f"❌ **Could not parse issue fields:** {e}\n\nPlease edit the issue and ensure all required fields are filled in."
|
|
post_comment(msg)
|
|
sys.exit(1)
|
|
|
|
print("Parsed fields:")
|
|
for k, v in fields.items():
|
|
print(f" {k}: {v}")
|
|
|
|
# --- Validate ---
|
|
errors = validate(fields)
|
|
if errors:
|
|
bullet_errors = "\n".join(f"- {e}" for e in errors)
|
|
msg = f"❌ **Validation failed:**\n\n{bullet_errors}\n\nPlease edit the issue to fix the errors above."
|
|
post_comment(msg)
|
|
sys.exit(1)
|
|
|
|
# --- Create repo from template ---
|
|
payload = {
|
|
"owner": fields["org_name"],
|
|
"name": fields["repo_name"],
|
|
"description": fields["description"],
|
|
"private": True,
|
|
"git_content": True,
|
|
"topics": True,
|
|
}
|
|
|
|
print(f"\nCalling AllSpice API: POST /repos/{TEMPLATE_OWNER}/{TEMPLATE_REPO}/generate")
|
|
print(f" -> {fields['org_name']}/{fields['repo_name']}")
|
|
try:
|
|
result = api(
|
|
"POST",
|
|
f"/repos/{TEMPLATE_OWNER}/{TEMPLATE_REPO}/generate",
|
|
payload,
|
|
)
|
|
except RuntimeError as e:
|
|
msg = (
|
|
f"❌ **Repository creation failed.**\n\n"
|
|
f"```\n{e}\n```\n\n"
|
|
"Please check the workflow logs or contact an admin."
|
|
)
|
|
post_comment(msg)
|
|
sys.exit(1)
|
|
|
|
repo_url = result.get("html_url", "(URL unavailable)")
|
|
print(f"\nRepository created: {repo_url}")
|
|
|
|
# --- Post success comment and close issue ---
|
|
success_msg = (
|
|
f"✅ **Repository created successfully!**\n\n"
|
|
f"🔗 {repo_url}\n\n"
|
|
f"| Field | Value |\n"
|
|
f"|---|---|\n"
|
|
f"| Repository | `{fields['org_name']}/{fields['repo_name']}` |\n"
|
|
f"| Program | `{fields['program_name']}` |\n"
|
|
f"| Design | `{fields['design_name']}` |\n"
|
|
f"| Template | `{TEMPLATE_OWNER}/{TEMPLATE_REPO}` |\n"
|
|
f"| Visibility | 🔒 Private |"
|
|
)
|
|
post_comment(success_msg)
|
|
close_issue()
|
|
print("Issue commented and closed.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |