feat: add generic gitea issue-drive skills
This commit is contained in:
zfc
@@ -0,0 +1,269 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Create one or more Gitea issues from a JSON spec."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
|
||||
SSH_RE = re.compile(r"^git@(?P<host>[^:]+):(?P<path>.+?)(?:\.git)?/?$")
|
||||
REPO_PATH_RE = re.compile(r"^(?P<owner>[^/]+)/(?P<repo>[^/]+)$")
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Batch-create Gitea issues from a JSON spec."
|
||||
)
|
||||
parser.add_argument("spec", help="Path to JSON spec file")
|
||||
parser.add_argument(
|
||||
"--repo-url",
|
||||
help=(
|
||||
"Override target repo. Accepts full https URL, SSH git origin, "
|
||||
"or owner/repo with GITEA_BASE_URL."
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--repo",
|
||||
help="Shorthand owner/repo. Requires GITEA_BASE_URL.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--dry-run",
|
||||
action="store_true",
|
||||
help="Validate and print the payload without creating issues",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def load_token(required: bool = True) -> str:
|
||||
token = os.getenv("GITEA_ISSUE") or os.getenv("GITEA_TOKEN")
|
||||
if not token and required:
|
||||
raise SystemExit(
|
||||
"Missing GITEA_ISSUE (or fallback GITEA_TOKEN). "
|
||||
"Export the token before creating issues."
|
||||
)
|
||||
return token or ""
|
||||
|
||||
|
||||
def load_spec(path: str) -> dict:
|
||||
with open(path, "r", encoding="utf-8") as handle:
|
||||
data = json.load(handle)
|
||||
if not isinstance(data, dict) or not isinstance(data.get("issues"), list):
|
||||
raise SystemExit("Spec must be an object with an 'issues' array.")
|
||||
if not data["issues"]:
|
||||
raise SystemExit("Spec contains no issues.")
|
||||
return data
|
||||
|
||||
|
||||
def normalize_base_url(base_url: str) -> str:
|
||||
return base_url.rstrip("/")
|
||||
|
||||
|
||||
def parse_http_repo_url(repo_url: str) -> tuple[str, str, str]:
|
||||
parsed = urllib.parse.urlsplit(repo_url)
|
||||
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
||||
raise SystemExit("Invalid repo URL. Expected format: https://host/owner/repo")
|
||||
|
||||
path = parsed.path.rstrip("/")
|
||||
if path.endswith(".git"):
|
||||
path = path[:-4]
|
||||
|
||||
parts = [part for part in path.split("/") if part]
|
||||
if len(parts) < 2:
|
||||
raise SystemExit("Invalid repo URL. Expected format: https://host/owner/repo")
|
||||
|
||||
owner, repo = parts[-2], parts[-1]
|
||||
prefix = "/".join(parts[:-2])
|
||||
origin = f"{parsed.scheme}://{parsed.netloc}"
|
||||
if prefix:
|
||||
origin = f"{origin}/{prefix}"
|
||||
return origin, owner, repo
|
||||
|
||||
|
||||
def parse_repo_target(repo_target: str, base_url: str | None = None) -> tuple[str, str, str, str]:
|
||||
value = repo_target.strip()
|
||||
if not value:
|
||||
raise SystemExit("Repo target cannot be empty.")
|
||||
|
||||
if value.startswith("http://") or value.startswith("https://"):
|
||||
origin, owner, repo = parse_http_repo_url(value)
|
||||
repo_url = f"{origin}/{owner}/{repo}"
|
||||
return origin, owner, repo, repo_url
|
||||
|
||||
if value.startswith("ssh://"):
|
||||
parsed = urllib.parse.urlsplit(value)
|
||||
path = parsed.path.rstrip("/")
|
||||
if path.endswith(".git"):
|
||||
path = path[:-4]
|
||||
parts = [part for part in path.split("/") if part]
|
||||
if len(parts) < 2:
|
||||
raise SystemExit("Invalid SSH repo URL. Expected ssh://git@host/owner/repo.git")
|
||||
owner, repo = parts[-2], parts[-1]
|
||||
host = parsed.hostname
|
||||
if not host:
|
||||
raise SystemExit("Invalid SSH repo URL. Missing host.")
|
||||
origin = normalize_base_url(base_url or f"https://{host}")
|
||||
return origin, owner, repo, f"{origin}/{owner}/{repo}"
|
||||
|
||||
ssh_match = SSH_RE.match(value)
|
||||
if ssh_match:
|
||||
path = ssh_match.group("path").rstrip("/")
|
||||
if path.endswith(".git"):
|
||||
path = path[:-4]
|
||||
parts = [part for part in path.split("/") if part]
|
||||
if len(parts) < 2:
|
||||
raise SystemExit("Invalid SSH repo target. Expected git@host:owner/repo.git")
|
||||
owner, repo = parts[-2], parts[-1]
|
||||
origin = normalize_base_url(base_url or f"https://{ssh_match.group('host')}")
|
||||
return origin, owner, repo, f"{origin}/{owner}/{repo}"
|
||||
|
||||
path_match = REPO_PATH_RE.match(value)
|
||||
if path_match:
|
||||
if not base_url:
|
||||
raise SystemExit(
|
||||
"Repo shorthand owner/repo requires GITEA_BASE_URL or --repo-url."
|
||||
)
|
||||
origin = normalize_base_url(base_url)
|
||||
owner = path_match.group("owner")
|
||||
repo = path_match.group("repo")
|
||||
return origin, owner, repo, f"{origin}/{owner}/{repo}"
|
||||
|
||||
raise SystemExit(
|
||||
"Invalid repo target. Use https://host/owner/repo, git@host:owner/repo.git, "
|
||||
"ssh://git@host/owner/repo.git, or owner/repo with GITEA_BASE_URL."
|
||||
)
|
||||
|
||||
|
||||
def get_origin_repo_target() -> str:
|
||||
try:
|
||||
return subprocess.check_output(
|
||||
["git", "remote", "get-url", "origin"],
|
||||
text=True,
|
||||
stderr=subprocess.STDOUT,
|
||||
).strip()
|
||||
except subprocess.CalledProcessError as exc:
|
||||
raise SystemExit(f"Failed to read git origin: {exc.output.strip()}") from exc
|
||||
|
||||
|
||||
def resolve_repo(spec: dict, args: argparse.Namespace) -> tuple[str, str, str, str]:
|
||||
base_url = os.getenv("GITEA_BASE_URL")
|
||||
repo_target = (
|
||||
args.repo_url
|
||||
or args.repo
|
||||
or spec.get("repo_url")
|
||||
or spec.get("repo")
|
||||
or get_origin_repo_target()
|
||||
)
|
||||
return parse_repo_target(repo_target, base_url=base_url)
|
||||
|
||||
|
||||
def request_json(
|
||||
url: str,
|
||||
token: str,
|
||||
method: str = "GET",
|
||||
payload: dict | None = None,
|
||||
) -> dict | list:
|
||||
data = None
|
||||
headers = {
|
||||
"Authorization": f"token {token}",
|
||||
"Accept": "application/json",
|
||||
}
|
||||
if payload is not None:
|
||||
data = json.dumps(payload).encode("utf-8")
|
||||
headers["Content-Type"] = "application/json"
|
||||
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
try:
|
||||
with urllib.request.urlopen(req) as response:
|
||||
return json.load(response)
|
||||
except urllib.error.HTTPError as exc:
|
||||
body = exc.read().decode("utf-8", errors="replace")
|
||||
raise SystemExit(
|
||||
f"Gitea API request failed: {exc.code} {exc.reason} | {body}"
|
||||
) from exc
|
||||
except urllib.error.URLError as exc:
|
||||
raise SystemExit(f"Failed to reach Gitea API: {exc.reason}") from exc
|
||||
|
||||
|
||||
def fetch_label_map(api_base: str, token: str) -> dict[str, int]:
|
||||
data = request_json(f"{api_base}/labels?page=1&limit=100", token)
|
||||
if not isinstance(data, list):
|
||||
raise SystemExit("Unexpected labels API response.")
|
||||
return {
|
||||
item["name"]: item["id"]
|
||||
for item in data
|
||||
if "name" in item and "id" in item
|
||||
}
|
||||
|
||||
|
||||
def resolve_label_ids(label_names: list[str], label_map: dict[str, int]) -> tuple[list[int], list[str]]:
|
||||
ids: list[int] = []
|
||||
missing: list[str] = []
|
||||
for name in label_names:
|
||||
if name in label_map:
|
||||
ids.append(label_map[name])
|
||||
else:
|
||||
missing.append(name)
|
||||
return ids, missing
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
token = load_token(required=not args.dry_run)
|
||||
spec = load_spec(args.spec)
|
||||
origin, owner, repo, repo_url = resolve_repo(spec, args)
|
||||
api_base = f"{origin}/api/v1/repos/{owner}/{repo}"
|
||||
label_map = fetch_label_map(api_base, token) if token else {}
|
||||
|
||||
issues = spec["issues"]
|
||||
for issue in issues:
|
||||
if not isinstance(issue, dict):
|
||||
raise SystemExit("Each issue entry must be an object.")
|
||||
if not issue.get("title") or not issue.get("body"):
|
||||
raise SystemExit("Each issue must include non-empty 'title' and 'body'.")
|
||||
|
||||
for issue in issues:
|
||||
label_names = issue.get("labels") or []
|
||||
if not isinstance(label_names, list) or not all(
|
||||
isinstance(name, str) for name in label_names
|
||||
):
|
||||
raise SystemExit("'labels' must be an array of strings.")
|
||||
|
||||
label_ids, missing_labels = resolve_label_ids(label_names, label_map)
|
||||
payload = {
|
||||
"title": issue["title"],
|
||||
"body": issue["body"],
|
||||
"labels": label_ids,
|
||||
}
|
||||
|
||||
if args.dry_run:
|
||||
print(f"repo_url={repo_url}")
|
||||
print(f"DRY-RUN\t{issue['title']}")
|
||||
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
||||
if missing_labels:
|
||||
print(f"missing_labels={missing_labels}")
|
||||
if not token:
|
||||
print("warning\tno token provided, remote label validation skipped")
|
||||
continue
|
||||
|
||||
created = request_json(f"{api_base}/issues", token, method="POST", payload=payload)
|
||||
if not isinstance(created, dict):
|
||||
raise SystemExit("Unexpected issue creation response.")
|
||||
print(
|
||||
f"#{created.get('number')}\t{created.get('title')}\t{created.get('html_url')}"
|
||||
)
|
||||
if missing_labels:
|
||||
print(f"warning\tmissing labels skipped: {', '.join(missing_labels)}")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user