9dd462b764
* test(pipeline): remove adr and project-map test files * refactor(pipeline): remove DOCS phase from pipeline-status * test(pipeline): update tests for 6-phase pipeline * docs(handoff): add handoff and ADR for remove-docs-phase * docs(handoff): set PR number * fix(pipeline): parse PR body via json.loads in check_implement --------- Co-authored-by: opencode-agent <agent@opencode.local>
652 lines
23 KiB
Python
652 lines
23 KiB
Python
#!/usr/bin/env python3
|
||
"""Pipeline-status oracle: determine PR phase in 6-phase PR pipeline.
|
||
|
||
Reads facts from GitHub (gh CLI), git, and memory files to deterministically
|
||
derive the current pipeline phase of a PR — no state file, like ``git status``
|
||
for the PR pipeline. CI gate via ``gh pr view --json statusCheckRollup`` (read-only):
|
||
CI ❌ blocks MERGE (transitive guard — first ❌ phase blocks all subsequent phases).
|
||
|
||
Usage:
|
||
python3 config/scripts/pipeline-status.py <PR_NUMBER> # single PR status
|
||
python3 config/scripts/pipeline-status.py # table of open PRs
|
||
|
||
Six phases:
|
||
1. ISSUE — GitHub issue exists and linked via Closes/Fixes #N
|
||
2. IMPLEMENT — PR exists + PR body has 4 required headings
|
||
3. CI — all checks on PR head SHA completed & success (statusCheckRollup)
|
||
4. REVIEW — APPROVE found in PR comments
|
||
5. MERGE — PR state is MERGED
|
||
6. MEMORY — PR#N distilled into repos/{host}/{org}/{repo}.md
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import functools
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from dataclasses import dataclass
|
||
from enum import StrEnum
|
||
from pathlib import Path
|
||
|
||
|
||
def _resolve_repo_root() -> Path:
|
||
"""Resolve repo root via git (cwd-aware), fallback to script location."""
|
||
result = subprocess.run(
|
||
["git", "rev-parse", "--show-toplevel"], capture_output=True, text=True, check=False
|
||
)
|
||
if result.returncode == 0 and result.stdout.strip():
|
||
return Path(result.stdout.strip()).resolve()
|
||
return Path(__file__).resolve().parent.parent.parent
|
||
|
||
|
||
REPO_ROOT = _resolve_repo_root()
|
||
_MEMORY_BASE = os.environ.get(
|
||
"OPENCODE_MEMORY_DIR",
|
||
str(REPO_ROOT / "app_data" / "opencode-memory"),
|
||
)
|
||
MEMORY_DIR = Path(_MEMORY_BASE) / "repos"
|
||
|
||
PHASE_NAMES = ["ISSUE", "IMPLEMENT", "CI", "REVIEW", "MERGE", "MEMORY"]
|
||
|
||
CI_WAIT_TIMEOUT = 300
|
||
CI_POLL_INTERVAL = 10
|
||
CI_NO_RUNS_RETRY = 3
|
||
CI_NO_RUNS_INTERVAL = 5
|
||
|
||
CLOSURE_RE = re.compile(r"(?:Closes|Fixes|Resolves)\s+#(\d+)", re.IGNORECASE)
|
||
REVIEW_APPROVE_RE = re.compile(
|
||
r"## Code Review Summary.*?###\s*Verdict:\s*APPROVE\b",
|
||
re.IGNORECASE | re.DOTALL,
|
||
)
|
||
REVIEW_VERDICT_RE = re.compile(
|
||
r"## Code Review Summary.*?###\s*Verdict:\s*(\w+)",
|
||
re.IGNORECASE | re.DOTALL,
|
||
)
|
||
JSON_FIELD_RE = re.compile(r'"(\w+)"\s*:\s*"?([^",}]*)"?', re.IGNORECASE)
|
||
|
||
|
||
class PhaseStatus(StrEnum):
|
||
"""Phase check result."""
|
||
|
||
DONE = "DONE"
|
||
NOT_DONE = "NOT_DONE"
|
||
AMBIGUOUS = "AMBIGUOUS"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PhaseResult:
|
||
"""Result of a single phase check."""
|
||
|
||
status: PhaseStatus
|
||
detail: str
|
||
|
||
|
||
def run_cmd(args: list[str]) -> tuple[int, str, str]:
|
||
"""Run a command, return (returncode, stdout, stderr)."""
|
||
result = subprocess.run(args, capture_output=True, text=True, check=False)
|
||
return result.returncode, result.stdout, result.stderr
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CiPollConfig:
|
||
"""CI polling parameters.
|
||
|
||
Priority: CLI flag > env var > constant in code.
|
||
"""
|
||
|
||
wait_timeout: int
|
||
poll_interval: int
|
||
|
||
|
||
def _env_int(name: str, default: int) -> int:
|
||
"""Read int from env var, fallback to default on missing/invalid."""
|
||
raw = os.environ.get(name)
|
||
if raw is None:
|
||
return default
|
||
try:
|
||
return int(raw)
|
||
except ValueError:
|
||
return default
|
||
|
||
|
||
def _load_ci_config(argv: list[str] | None = None) -> CiPollConfig:
|
||
"""Build CiPollConfig from CLI flags + env vars + constants.
|
||
|
||
Priority: CLI flag > env var > constant. CLI flags ``--ci-wait-timeout N``
|
||
and ``--ci-poll-interval N`` parsed manually from ``argv`` (last wins).
|
||
"""
|
||
args = argv if argv is not None else sys.argv[1:]
|
||
cli_timeout: int | None = None
|
||
cli_interval: int | None = None
|
||
i = 0
|
||
while i < len(args):
|
||
if args[i] == "--ci-wait-timeout" and i + 1 < len(args):
|
||
try:
|
||
cli_timeout = int(args[i + 1])
|
||
except ValueError:
|
||
cli_timeout = None
|
||
i += 2
|
||
continue
|
||
if args[i] == "--ci-poll-interval" and i + 1 < len(args):
|
||
try:
|
||
cli_interval = int(args[i + 1])
|
||
except ValueError:
|
||
cli_interval = None
|
||
i += 2
|
||
continue
|
||
i += 1
|
||
|
||
timeout = (
|
||
cli_timeout
|
||
if cli_timeout is not None
|
||
else _env_int("OPENCODE_CI_WAIT_TIMEOUT", CI_WAIT_TIMEOUT)
|
||
)
|
||
interval = (
|
||
cli_interval
|
||
if cli_interval is not None
|
||
else _env_int("OPENCODE_CI_POLL_INTERVAL", CI_POLL_INTERVAL)
|
||
)
|
||
return CiPollConfig(wait_timeout=timeout, poll_interval=interval)
|
||
|
||
|
||
def parse_remote_url(url: str) -> tuple[str, str, str]:
|
||
"""Parse git remote URL into (host, org, repo).
|
||
|
||
Supports both HTTPS and SSH formats:
|
||
https://github.com/org/repo.git -> (github.com, org, repo)
|
||
git@github.com:org/repo.git -> (github.com, org, repo)
|
||
"""
|
||
ssh_match = re.match(r"git@([^:]+):([^/]+)/(.+?)(?:\.git)?$", url)
|
||
if ssh_match:
|
||
return ssh_match.group(1), ssh_match.group(2), ssh_match.group(3)
|
||
# `(?:[^/@]*@)?` optionally skips `user:password@` userinfo before host.
|
||
# Needed because `git config url.insteadOf` rewrites `https://github.com/`
|
||
# to `https://x-access-token:TOKEN@github.com/`, and `git remote get-url
|
||
# origin` returns the rewritten URL (see ADR-022 / ADR-023).
|
||
https_match = re.match(r"https?://(?:[^/@]*@)?([^/]+)/([^/]+)/(.+?)(?:\.git)?$", url)
|
||
if https_match:
|
||
return https_match.group(1), https_match.group(2), https_match.group(3)
|
||
raise ValueError(f"Cannot parse remote URL: {url}")
|
||
|
||
|
||
def get_memory_file_path() -> Path:
|
||
"""Derive memory file path from ``git remote get-url origin``."""
|
||
rc, out, err = run_cmd(["git", "remote", "get-url", "origin"])
|
||
if rc != 0:
|
||
raise RuntimeError(f"Cannot get git remote URL: {err.strip()}")
|
||
host, org, repo = parse_remote_url(out.strip())
|
||
return MEMORY_DIR / host / org / f"{repo}.md"
|
||
|
||
|
||
@functools.cache
|
||
def get_repo_full_name() -> str:
|
||
"""Return ``org/repo`` from git remote (cached, one call per run).
|
||
|
||
Used for GitHub Actions API URL: ``repos/{org}/{repo}/actions/runs``.
|
||
Cached via ``functools.cache`` (one git call per process); tests reset
|
||
via ``get_repo_full_name.cache_clear()``.
|
||
"""
|
||
rc, out, err = run_cmd(["git", "remote", "get-url", "origin"])
|
||
if rc != 0:
|
||
raise RuntimeError(f"Cannot get git remote URL: {err.strip()}")
|
||
_host, org, repo = parse_remote_url(out.strip())
|
||
return f"{org}/{repo}"
|
||
|
||
|
||
def extract_json_field(json_str: str, field: str) -> str | None:
|
||
"""Extract a string field value from JSON (simple regex, no json import)."""
|
||
match = re.search(rf'"{field}"\s*:\s*"([^"]*)"', json_str)
|
||
return match.group(1) if match else None
|
||
|
||
|
||
def check_gh_auth() -> str | None:
|
||
"""Check if gh CLI is authenticated. Returns error message or None."""
|
||
rc, _, err = run_cmd(["gh", "auth", "status"])
|
||
if rc != 0:
|
||
return f"gh CLI не авторизован: {err.strip()}"
|
||
return None
|
||
|
||
|
||
def check_issue(pr_number: int) -> PhaseResult:
|
||
"""Phase 1: ISSUE — issue exists and linked via Closes/Fixes #N."""
|
||
rc, out, _ = run_cmd(
|
||
["gh", "pr", "view", str(pr_number), "--json", "body", "--repo", get_repo_full_name()]
|
||
)
|
||
if rc != 0:
|
||
return PhaseResult(PhaseStatus.NOT_DONE, f"PR #{pr_number} не существует")
|
||
|
||
matches = CLOSURE_RE.findall(out)
|
||
if not matches:
|
||
return PhaseResult(PhaseStatus.NOT_DONE, "Closes/Fixes #N не найден в body")
|
||
|
||
issue_numbers = list({int(m) for m in matches})
|
||
if len(issue_numbers) > 1:
|
||
return PhaseResult(
|
||
PhaseStatus.AMBIGUOUS,
|
||
f"несколько issue в body: {', '.join(f'#{n}' for n in issue_numbers)}",
|
||
)
|
||
|
||
issue_num = issue_numbers[0]
|
||
rc2, _, err2 = run_cmd(["gh", "issue", "view", str(issue_num), "--repo", get_repo_full_name()])
|
||
if rc2 != 0:
|
||
return PhaseResult(
|
||
PhaseStatus.NOT_DONE,
|
||
f"issue #{issue_num} не существует: {err2.strip()}",
|
||
)
|
||
|
||
return PhaseResult(PhaseStatus.DONE, f"#{issue_num} связан через Closes #{issue_num}")
|
||
|
||
|
||
def check_implement(pr_number: int) -> PhaseResult:
|
||
"""Phase 2: IMPLEMENT — PR exists + PR body has 4 required headings."""
|
||
rc, out, _ = run_cmd(
|
||
["gh", "pr", "view", str(pr_number), "--json", "body", "--repo", get_repo_full_name()]
|
||
)
|
||
if rc != 0:
|
||
return PhaseResult(PhaseStatus.NOT_DONE, f"PR #{pr_number} не существует")
|
||
try:
|
||
pr_body = json.loads(out).get("body", "") or ""
|
||
except (json.JSONDecodeError, ValueError):
|
||
return PhaseResult(
|
||
PhaseStatus.NOT_DONE, f"PR #{pr_number}: не удалось распарсить JSON body"
|
||
)
|
||
required = ["## Что сделано", "## Почему", "## Watch out", "## Pending"]
|
||
missing = [h for h in required if h not in pr_body]
|
||
if missing:
|
||
return PhaseResult(
|
||
PhaseStatus.NOT_DONE, f"PR body не содержит heading'и: {', '.join(missing)}"
|
||
)
|
||
return PhaseResult(PhaseStatus.DONE, "PR body: 4 heading'а валидны")
|
||
|
||
|
||
def check_ci(pr_number: int) -> PhaseResult:
|
||
"""Phase 3: CI — all checks on PR head SHA completed & success.
|
||
|
||
Uses ``gh pr view --json statusCheckRollup`` which aggregates ALL
|
||
workflows for the PR head SHA (CI, CI (always), ADR check, etc.).
|
||
This handles docs-only PRs where ``ci.yml`` has ``paths-ignore`` and
|
||
only ``always-ci.yml`` runs.
|
||
"""
|
||
config = _load_ci_config()
|
||
return _run_ci_loop(pr_number, config)
|
||
|
||
|
||
def _run_ci_loop(pr_number: int, config: CiPollConfig) -> PhaseResult:
|
||
"""Initial CI query + edge-case dispatch + delegate to poll/no-checks helpers."""
|
||
kind, json_str, err = _query_ci_rollup(pr_number)
|
||
if kind == "error":
|
||
return PhaseResult(PhaseStatus.AMBIGUOUS, err)
|
||
if kind == "no_checks":
|
||
return _retry_no_checks(pr_number, config)
|
||
return _classify_rollup_with_poll(pr_number, config, json_str)
|
||
|
||
|
||
def _retry_no_checks(pr_number: int, config: CiPollConfig) -> PhaseResult:
|
||
"""Retry CI query when no checks registered yet (CI may lag after push).
|
||
|
||
Up to ``CI_NO_RUNS_RETRY`` total attempts, sleeping ``CI_NO_RUNS_INTERVAL``
|
||
between attempts. On success → classify/poll; exhausted → AMBIGUOUS.
|
||
"""
|
||
for attempt in range(CI_NO_RUNS_RETRY):
|
||
if attempt > 0:
|
||
time.sleep(CI_NO_RUNS_INTERVAL)
|
||
kind, json_str, err = _query_ci_rollup(pr_number)
|
||
if kind == "error":
|
||
return PhaseResult(PhaseStatus.AMBIGUOUS, err)
|
||
if kind == "rollup":
|
||
return _classify_rollup_with_poll(pr_number, config, json_str)
|
||
return PhaseResult(
|
||
PhaseStatus.AMBIGUOUS,
|
||
f"нет CI checks на PR #{pr_number} — возможна проблема триггера",
|
||
)
|
||
|
||
|
||
def _classify_rollup_with_poll(pr_number: int, config: CiPollConfig, json_str: str) -> PhaseResult:
|
||
"""Classify rollup; if in_progress → poll until completed or timeout."""
|
||
result = _classify_rollup(json_str)
|
||
if result.status != PhaseStatus.AMBIGUOUS or "in progress" not in result.detail.lower():
|
||
return result
|
||
elapsed = 0
|
||
while elapsed < config.wait_timeout:
|
||
if elapsed + config.poll_interval > config.wait_timeout:
|
||
break
|
||
time.sleep(config.poll_interval)
|
||
elapsed += config.poll_interval
|
||
kind, json_str_new, err = _query_ci_rollup(pr_number)
|
||
if kind == "error":
|
||
return PhaseResult(PhaseStatus.AMBIGUOUS, err)
|
||
if kind == "no_checks":
|
||
return PhaseResult(
|
||
PhaseStatus.AMBIGUOUS,
|
||
f"нет CI checks на PR #{pr_number} — возможна проблема триггера",
|
||
)
|
||
result = _classify_rollup(json_str_new)
|
||
if result.status != PhaseStatus.AMBIGUOUS or "in progress" not in result.detail.lower():
|
||
return result
|
||
return PhaseResult(
|
||
PhaseStatus.AMBIGUOUS,
|
||
f"CI ещё идёт после {config.wait_timeout}s — проверь вручную: gh pr checks {pr_number}",
|
||
)
|
||
|
||
|
||
def _query_ci_rollup(pr_number: int) -> tuple[str, str, str]:
|
||
"""Query PR statusCheckRollup via gh CLI.
|
||
|
||
GitHub aggregates ALL checks (CI, CI (always), ADR check) for PR head SHA.
|
||
Return ``(kind, json_str, error)`` where ``kind`` is:
|
||
- ``"error"`` — API call failed (rc != 0).
|
||
- ``"no_checks"`` — rollup array is empty (no checks registered yet).
|
||
- ``"rollup"`` — JSON with statusCheckRollup array, ``json_str`` set.
|
||
"""
|
||
rc, out, err = run_cmd(
|
||
[
|
||
"gh",
|
||
"pr",
|
||
"view",
|
||
str(pr_number),
|
||
"--json",
|
||
"statusCheckRollup",
|
||
"--repo",
|
||
get_repo_full_name(),
|
||
]
|
||
)
|
||
if rc != 0:
|
||
return "error", "", f"PR API error: {err.strip()}"
|
||
json_str = out.strip()
|
||
if not json_str:
|
||
return "no_checks", "", ""
|
||
if re.search(r'"statusCheckRollup"\s*:\s*\[\s*\]', json_str):
|
||
return "no_checks", "", ""
|
||
return "rollup", json_str, ""
|
||
|
||
|
||
def _classify_rollup(json_str: str) -> PhaseResult:
|
||
"""Classify CI status from statusCheckRollup JSON.
|
||
|
||
All checks COMPLETED + SUCCESS/SKIPPED/NEUTRAL → DONE.
|
||
Any check COMPLETED + non-success conclusion → NOT_DONE.
|
||
Any check IN_PROGRESS/QUEUED/PENDING → AMBIGUOUS (poll).
|
||
"""
|
||
statuses = re.findall(r'"status"\s*:\s*"([^"]*)"', json_str, re.IGNORECASE)
|
||
if not statuses:
|
||
return PhaseResult(PhaseStatus.AMBIGUOUS, "no checks found in rollup")
|
||
|
||
in_progress = [s for s in statuses if s.upper() in ("IN_PROGRESS", "QUEUED", "PENDING")]
|
||
if in_progress:
|
||
return PhaseResult(PhaseStatus.AMBIGUOUS, "CI in progress")
|
||
|
||
conclusions = re.findall(r'"conclusion"\s*:\s*"([^"]*)"', json_str, re.IGNORECASE)
|
||
null_conclusions = re.findall(r'"conclusion"\s*:\s*null', json_str, re.IGNORECASE)
|
||
|
||
for c in conclusions:
|
||
if c.upper() not in ("SUCCESS", "SKIPPED", "NEUTRAL"):
|
||
return PhaseResult(PhaseStatus.NOT_DONE, f"CI {c.lower()} — fix needed")
|
||
|
||
if null_conclusions:
|
||
return PhaseResult(PhaseStatus.AMBIGUOUS, "CI completed but conclusion missing")
|
||
|
||
return PhaseResult(PhaseStatus.DONE, "CI green")
|
||
|
||
|
||
def _extract_comment_bodies(json_str: str) -> list[str]:
|
||
"""Extract 'body' fields from gh pr view --json comments output.
|
||
|
||
Handles JSON string escaping (\\n, \\", \\\\).
|
||
"""
|
||
bodies = []
|
||
for match in re.finditer(r'"body"\s*:\s*"((?:[^"\\]|\\.)*)"', json_str):
|
||
raw = match.group(1)
|
||
body = raw.encode().decode("unicode_escape")
|
||
bodies.append(body)
|
||
return bodies
|
||
|
||
|
||
def check_review(pr_number: int) -> PhaseResult:
|
||
"""Phase 5: REVIEW — APPROVE found in PR comments from code reviewer.
|
||
|
||
Looks for '## Code Review Summary' heading (NOT '## Docs Review Summary')
|
||
with '### Verdict: APPROVE'. Only the latest reviewer comment counts —
|
||
if reviewer changed from APPROVE to REQUEST_CHANGES, NOT_DONE.
|
||
"""
|
||
rc, out, _ = run_cmd(
|
||
["gh", "pr", "view", str(pr_number), "--json", "comments", "--repo", get_repo_full_name()]
|
||
)
|
||
if rc != 0:
|
||
return PhaseResult(PhaseStatus.NOT_DONE, "не удалось получить комментарии PR")
|
||
|
||
comment_bodies = _extract_comment_bodies(out)
|
||
if not comment_bodies:
|
||
return PhaseResult(PhaseStatus.NOT_DONE, "нет комментариев PR")
|
||
|
||
reviewer_verdict = None
|
||
for body in comment_bodies:
|
||
if REVIEW_VERDICT_RE.search(body):
|
||
match = REVIEW_VERDICT_RE.search(body)
|
||
reviewer_verdict = match.group(1).upper() if match else None
|
||
|
||
if reviewer_verdict is None:
|
||
return PhaseResult(PhaseStatus.NOT_DONE, "Code Review Summary не найден в комментариях")
|
||
|
||
if reviewer_verdict == "APPROVE":
|
||
return PhaseResult(PhaseStatus.DONE, "APPROVE найден в Code Review Summary")
|
||
|
||
return PhaseResult(PhaseStatus.NOT_DONE, f"последний verdict reviewer'а: {reviewer_verdict}")
|
||
|
||
|
||
def check_merge(pr_number: int) -> PhaseResult:
|
||
"""Phase 6: MERGE — PR state is MERGED."""
|
||
rc, out, _ = run_cmd(
|
||
["gh", "pr", "view", str(pr_number), "--json", "state", "--repo", get_repo_full_name()]
|
||
)
|
||
if rc != 0:
|
||
return PhaseResult(PhaseStatus.NOT_DONE, f"PR #{pr_number} не существует")
|
||
|
||
state = extract_json_field(out, "state")
|
||
if state is None:
|
||
return PhaseResult(PhaseStatus.AMBIGUOUS, "не удалось распарсить state PR")
|
||
|
||
if state == "MERGED":
|
||
return PhaseResult(PhaseStatus.DONE, "merged")
|
||
|
||
return PhaseResult(PhaseStatus.NOT_DONE, f"state={state}")
|
||
|
||
|
||
def check_memory(pr_number: int) -> PhaseResult:
|
||
"""Phase 7: MEMORY — PR#N distilled into memory file."""
|
||
try:
|
||
memory_file = get_memory_file_path()
|
||
except (RuntimeError, ValueError) as exc:
|
||
return PhaseResult(PhaseStatus.NOT_DONE, str(exc))
|
||
|
||
if not memory_file.exists():
|
||
return PhaseResult(
|
||
PhaseStatus.NOT_DONE,
|
||
f"memory file не существует: {memory_file.name}",
|
||
)
|
||
|
||
content = memory_file.read_text()
|
||
pattern = f"PR#{pr_number}"
|
||
if pattern in content:
|
||
return PhaseResult(PhaseStatus.DONE, f"{pattern} в {memory_file.name}")
|
||
|
||
return PhaseResult(
|
||
PhaseStatus.NOT_DONE,
|
||
f"{pattern} не найден в {memory_file.name}",
|
||
)
|
||
|
||
|
||
def get_pr_title(pr_number: int) -> str:
|
||
"""Get PR title via gh CLI."""
|
||
rc, out, _ = run_cmd(
|
||
["gh", "pr", "view", str(pr_number), "--json", "title", "--repo", get_repo_full_name()]
|
||
)
|
||
if rc != 0:
|
||
return f"PR #{pr_number}"
|
||
title = extract_json_field(out, "title")
|
||
return title if title else f"PR #{pr_number}"
|
||
|
||
|
||
NEXT_ACTIONS: dict[str, str] = {
|
||
"ISSUE": "dispatch subagent (subagent_type=general, template=A) for PR #N",
|
||
"IMPLEMENT": "dispatch subagent (subagent_type=general, template=A) for PR #N",
|
||
"CI": "проверь статус CI вручную (gh run view)",
|
||
"REVIEW": "dispatch subagent (subagent_type=reviewer, template=C) for PR #N",
|
||
"MERGE": "call merge_pr tool with pr_number=N",
|
||
"MEMORY": "dispatch subagent (subagent_type=memory-syncer, template=E) for PR #N",
|
||
}
|
||
|
||
REVIEW_NEXT_REQUEST_CHANGES = (
|
||
"dispatch subagent (subagent_type=general) with prompt 'fix reviewer comments: <list>', "
|
||
"commit, push → re-loop (pipeline_status проверит CI автоматически)"
|
||
)
|
||
REVIEW_NEXT_NEEDS_DISCUSSION = "уточни вопросы с автором PR (verdict: NEEDS_DISCUSSION)"
|
||
REVIEW_NEXT_DEFAULT = NEXT_ACTIONS["REVIEW"]
|
||
|
||
STATUS_ICONS: dict[PhaseStatus, str] = {
|
||
PhaseStatus.DONE: "✅",
|
||
PhaseStatus.NOT_DONE: "❌",
|
||
PhaseStatus.AMBIGUOUS: "⚠️",
|
||
}
|
||
|
||
|
||
def get_next_action(phase_name: str, pr_number: int) -> str:
|
||
"""Get NEXT action description for a not-done phase."""
|
||
action = NEXT_ACTIONS.get(phase_name, "уточнить статус")
|
||
return action.replace("N", str(pr_number))
|
||
|
||
|
||
def get_next_action_review(result: PhaseResult) -> str:
|
||
"""REVIEW-фаза: NEXT зависит от вердикта reviewer'а в result.detail."""
|
||
detail = result.detail.upper()
|
||
if "REQUEST_CHANGES" in detail:
|
||
return REVIEW_NEXT_REQUEST_CHANGES
|
||
if "NEEDS_DISCUSSION" in detail:
|
||
return REVIEW_NEXT_NEEDS_DISCUSSION
|
||
return REVIEW_NEXT_DEFAULT
|
||
|
||
|
||
def run_all_checks(pr_number: int) -> list[PhaseResult]:
|
||
"""Run all 6 phase checks, return results in order."""
|
||
return [
|
||
check_issue(pr_number),
|
||
check_implement(pr_number),
|
||
check_ci(pr_number),
|
||
check_review(pr_number),
|
||
check_merge(pr_number),
|
||
check_memory(pr_number),
|
||
]
|
||
|
||
|
||
def find_current_phase(results: list[PhaseResult]) -> int | None:
|
||
"""Return index of first not-done phase, or None if all done."""
|
||
for i, result in enumerate(results):
|
||
if result.status != PhaseStatus.DONE:
|
||
return i
|
||
return None
|
||
|
||
|
||
def format_single_pr(pr_number: int, results: list[PhaseResult]) -> str:
|
||
"""Format detailed output for a single PR."""
|
||
title = get_pr_title(pr_number)
|
||
lines = [f"PR #{pr_number}: {title}", ""]
|
||
|
||
for i, (name, result) in enumerate(zip(PHASE_NAMES, results, strict=True)):
|
||
icon = STATUS_ICONS[result.status]
|
||
lines.append(f"{icon} {i + 1}. {name:<12} {result.detail}")
|
||
|
||
lines.append("")
|
||
|
||
current = find_current_phase(results)
|
||
if current is None:
|
||
lines.append("Status: COMPLETE")
|
||
else:
|
||
result = results[current]
|
||
if result.status == PhaseStatus.AMBIGUOUS:
|
||
lines.append(f"AMBIGUOUS: {result.detail}")
|
||
lines.append("NEXT: уточните статус вручную")
|
||
else:
|
||
if PHASE_NAMES[current] == "REVIEW":
|
||
action = get_next_action_review(result)
|
||
else:
|
||
action = get_next_action(PHASE_NAMES[current], pr_number)
|
||
lines.append(f"NEXT: {action}")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
def list_open_pr_numbers() -> list[int]:
|
||
"""List numbers of all open PRs."""
|
||
rc, out, _ = run_cmd(
|
||
["gh", "pr", "list", "--state", "open", "--json", "number", "--repo", get_repo_full_name()]
|
||
)
|
||
if rc != 0:
|
||
return []
|
||
numbers = re.findall(r'"number"\s*:\s*(\d+)', out)
|
||
return sorted(int(n) for n in numbers)
|
||
|
||
|
||
def format_pr_row(pr_number: int) -> str:
|
||
"""Format a single row for the open-PRs table."""
|
||
results = run_all_checks(pr_number)
|
||
title = get_pr_title(pr_number)
|
||
icon_str = "".join(STATUS_ICONS[r.status] for r in results)
|
||
|
||
current = find_current_phase(results)
|
||
if current is None:
|
||
next_action = "COMPLETE"
|
||
elif PHASE_NAMES[current] == "REVIEW":
|
||
next_action = get_next_action_review(results[current])
|
||
else:
|
||
next_action = get_next_action(PHASE_NAMES[current], pr_number)
|
||
|
||
short_title = title[:40] + "..." if len(title) > 40 else title
|
||
return f"PR#{pr_number:<5} {short_title:<43} {icon_str} NEXT: {next_action}"
|
||
|
||
|
||
def format_table(pr_numbers: list[int]) -> str:
|
||
"""Format table of all open PRs."""
|
||
if not pr_numbers:
|
||
return "Нет открытых PR"
|
||
rows = [format_pr_row(n) for n in pr_numbers]
|
||
return "\n".join(rows)
|
||
|
||
|
||
def pr_exists(pr_number: int) -> bool:
|
||
"""Check if PR exists via gh CLI."""
|
||
rc, _, _ = run_cmd(
|
||
["gh", "pr", "view", str(pr_number), "--json", "number", "--repo", get_repo_full_name()]
|
||
)
|
||
return rc == 0
|
||
|
||
|
||
def main() -> None:
|
||
"""Entry point: parse args and dispatch to single-PR or table mode."""
|
||
auth_error = check_gh_auth()
|
||
if auth_error:
|
||
print(auth_error, file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
if len(sys.argv) > 1:
|
||
try:
|
||
pr_number = int(sys.argv[1])
|
||
except ValueError:
|
||
print(f"Некорректный номер PR: {sys.argv[1]}", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
if not pr_exists(pr_number):
|
||
print(f"PR #{pr_number} не существует", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
results = run_all_checks(pr_number)
|
||
print(format_single_pr(pr_number, results))
|
||
else:
|
||
pr_numbers = list_open_pr_numbers()
|
||
print(format_table(pr_numbers))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|