Move directory

2026-03-06 17:08:31 +09:00
parent 4d99258344
commit 21addb6e88
73 changed files with 0 additions and 0 deletions
@@ -0,0 +1,49 @@
+---
+name: create_pr
+description: Create a GitHub PR for the chart upgrade.
+user-invokable: false
+---
+
+# create_pr (Skill)
+
+브랜치 생성 → 커밋 → 푸시 → GitHub PR 생성을 수행한다.
+breaking=true이면 `needs-review` 레이블, 아니면 `auto-update` 레이블을 붙인다.
+실행 후 `main` 브랜치로 복귀하여 다음 cronjob 실행을 위한 클린 상태를 유지한다.
+
+## 브랜치 명명 규칙
+
+`update-{chart}/{to_version}` — 예: `update-airflow/1.19.0`
+
+## 재실행 안전성 (Idempotent)
+
+- 브랜치가 이미 존재하면 체크아웃 후 추가 커밋
+- PR이 이미 존재하면 URL만 반환 (중복 PR 생성 안 함)
+
+## Input schema
+```json
+{
+  "repo_path": "string",
+  "chart": "string",
+  "from_version": "string",
+  "to_version": "string",
+  "breaking": "boolean",
+  "severity": "critical|high|medium|warning",
+  "reasons": "array (breaking_change_check.reasons)",
+  "warnings": "array (breaking_change_check.warnings)"
+}
+```
+
+## Output schema
+```json
+{
+  "pr_url": "string",
+  "branch_name": "string",
+  "labels": ["string"],
+  "committed": "boolean"
+}
+```
+
+## 의존 도구
+
+- `git` — 브랜치/커밋/푸시
+- `gh` (GitHub CLI) — PR 생성 및 레이블 관리. `gh auth login` 완료 필요.
@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+import argparse
+import json
+import os
+import sys
+from pathlib import Path
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--repo-path", required=True, help="dip-catalog git repo 경로")
+    parser.add_argument("--chart", required=True)
+    parser.add_argument("--from-version", required=True)
+    parser.add_argument("--to-version", required=True)
+    parser.add_argument("--breaking-file", required=True, help="breaking_change_check 출력 JSON 경로")
+    args = parser.parse_args()
+
+    repo_root = os.environ.get("UPDATE_CATALOG_ROOT")
+    if repo_root:
+        sys.path.insert(0, str(Path(repo_root) / "src"))
+    else:
+        sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+    from update_catalog.create_pr import create_pr  # type: ignore
+
+    with open(args.breaking_file, "r", encoding="utf-8") as f:
+        breaking_data = json.load(f)
+
+    out = create_pr({
+        "repo_path": args.repo_path,
+        "chart": args.chart,
+        "from_version": args.from_version,
+        "to_version": args.to_version,
+        "breaking": breaking_data.get("breaking", False),
+        "severity": breaking_data.get("severity", "warning"),
+        "reasons": breaking_data.get("reasons", []),
+        "warnings": breaking_data.get("warnings", []),
+    })
+    print(json.dumps(out, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,164 @@
+"""create_pr skill implementation.
+
+Branch → Commit → Push → GitHub PR 생성.
+"""
+from __future__ import annotations
+
+import json
+import subprocess
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from .skill_interface import CreatePRInput, CreatePROutput, BreakingReason  # type: ignore
+
+
+def _git(args: List[str], cwd: Path) -> str:
+    res = subprocess.run(["git"] + args, cwd=cwd, capture_output=True, text=True)
+    if res.returncode != 0:
+        raise RuntimeError(f"git {' '.join(args)} failed: {res.stderr.strip()}")
+    return res.stdout.strip()
+
+
+def _gh(args: List[str], cwd: Path) -> str:
+    res = subprocess.run(["gh"] + args, cwd=cwd, capture_output=True, text=True)
+    if res.returncode != 0:
+        raise RuntimeError(f"gh {' '.join(args)} failed: {res.stderr.strip()}")
+    return res.stdout.strip()
+
+
+def _get_existing_pr_url(branch_name: str, cwd: Path) -> Optional[str]:
+    try:
+        url = _gh(
+            ["pr", "list", "--head", branch_name, "--json", "url", "--jq", ".[0].url"],
+            cwd=cwd,
+        )
+        return url.strip() or None
+    except Exception:
+        return None
+
+
+def _ensure_label(label: str, cwd: Path) -> None:
+    """레이블이 없으면 생성한다."""
+    label_colors = {
+        "needs-review": "e11d48",   # 빨강
+        "auto-update": "16a34a",    # 초록
+    }
+    try:
+        _gh(["label", "list", "--json", "name", "--jq", f'.[] | select(.name == "{label}") | .name'], cwd=cwd)
+    except Exception:
+        pass
+    color = label_colors.get(label, "0075ca")
+    try:
+        _gh(["label", "create", label, "--color", color, "--force"], cwd=cwd)
+    except Exception:
+        pass  # 레이블 생성 실패해도 PR 생성은 계속
+
+
+def _build_pr_body(
+    chart: str,
+    from_version: str,
+    to_version: str,
+    breaking: bool,
+    severity: str,
+    reasons: List[BreakingReason],
+    warnings: List[BreakingReason],
+) -> str:
+    lines = [
+        f"## Helm Chart Update: {chart} `{from_version}` → `{to_version}`",
+        "",
+        f"**Severity**: `{severity}`  ",
+        f"**Breaking**: {'✅ Human review required before merge' if breaking else '❌ No breaking changes'}",
+        "",
+    ]
+    if reasons:
+        lines.append("### Breaking Changes")
+        for r in reasons:
+            key = r.get("key") or r.get("resource") or ""
+            lines.append(f"- `[{r['type']}]` {key} — {r.get('detail', '')}")
+        lines.append("")
+    if warnings:
+        lines.append(f"### Warnings ({len(warnings)} removed keys not used in custom-values.yaml)")
+        for w in warnings:
+            lines.append(f"- `[{w['type']}]` {w.get('key', '')} — {w.get('detail', '')}")
+        lines.append("")
+    if not reasons:
+        lines += ["No breaking changes detected.", ""]
+    lines.append("---")
+    lines.append("*Generated by update-catalog automation*")
+    return "\n".join(lines)
+
+
+def create_pr(payload: Dict[str, Any]) -> Dict[str, Any]:
+    inp = CreatePRInput(**payload)
+    repo_path = Path(inp.repo_path)
+    chart = inp.chart
+    from_version = inp.from_version
+    to_version = inp.to_version
+    breaking = inp.breaking
+    severity = inp.severity
+    reasons = inp.reasons
+    warnings = inp.warnings
+
+    branch_name = f"update-{chart}/{to_version}"
+    chart_dir = f"manifests/helm/{chart}/{to_version}"
+
+    # 1. main으로 이동 후 최신화
+    _git(["checkout", "main"], cwd=repo_path)
+    _git(["pull"], cwd=repo_path)
+
+    # 2. 브랜치 생성 또는 기존 브랜치 체크아웃
+    local_exists = bool(_git(["branch", "--list", branch_name], cwd=repo_path).strip())
+    remote_exists = bool(_git(["branch", "-r", "--list", f"origin/{branch_name}"], cwd=repo_path).strip())
+
+    if local_exists:
+        _git(["checkout", branch_name], cwd=repo_path)
+    elif remote_exists:
+        _git(["checkout", "-b", branch_name, f"origin/{branch_name}"], cwd=repo_path)
+    else:
+        _git(["checkout", "-b", branch_name], cwd=repo_path)
+
+    # 3. 변경 파일 스테이징
+    _git(["add", chart_dir], cwd=repo_path)
+
+    # 4. 변경사항이 있으면 커밋
+    status = _git(["status", "--porcelain", chart_dir], cwd=repo_path)
+    committed = False
+    if status.strip():
+        breaking_tag = " [BREAKING]" if breaking else ""
+        commit_msg = f"update {chart}/{to_version}{breaking_tag}"
+        _git(["commit", "-m", commit_msg], cwd=repo_path)
+        committed = True
+
+    # 5. 원격 브랜치에 푸시
+    _git(["push", "-u", "origin", branch_name], cwd=repo_path)
+
+    # 6. 레이블 준비
+    labels = ["needs-review"] if breaking else ["auto-update"]
+    for label in labels:
+        _ensure_label(label, cwd=repo_path)
+
+    # 7. PR 생성 (이미 존재하면 URL만 반환)
+    pr_url = _get_existing_pr_url(branch_name, repo_path)
+    if not pr_url:
+        breaking_tag = " [BREAKING]" if breaking else ""
+        title = f"update {chart}: {from_version} → {to_version}{breaking_tag}"
+        body = _build_pr_body(chart, from_version, to_version, breaking, severity, reasons, warnings)
+        label_args: List[str] = []
+        for label in labels:
+            label_args += ["--label", label]
+        pr_url = _gh(
+            ["pr", "create", "--title", title, "--body", body, "--base", "main"] + label_args,
+            cwd=repo_path,
+        )
+
+    # 8. main으로 복귀 (다음 cronjob 실행을 위해 클린 상태 유지)
+    _git(["checkout", "main"], cwd=repo_path)
+
+    return json.loads(
+        CreatePROutput(
+            pr_url=pr_url,
+            branch_name=branch_name,
+            labels=labels,
+            committed=committed,
+        ).model_dump_json()
+    )
@@ -0,0 +1,28 @@
+"""Skill Interface for create_pr."""
+from __future__ import annotations
+
+from typing import Any, Dict, List, Literal, Optional
+from pydantic import BaseModel, Field
+
+Severity = Literal["critical", "high", "medium", "warning"]
+
+# BreakingReason은 dict로 처리 (breaking_change_check 출력 그대로 수신)
+BreakingReason = Dict[str, Any]
+
+
+class CreatePRInput(BaseModel):
+    chart: str
+    from_version: str
+    to_version: str
+    repo_path: str
+    breaking: bool
+    severity: Severity
+    reasons: List[BreakingReason] = Field(default_factory=list)
+    warnings: List[BreakingReason] = Field(default_factory=list)
+
+
+class CreatePROutput(BaseModel):
+    pr_url: str
+    branch_name: str
+    labels: List[str]
+    committed: bool = False