pzhang_zywl
ae0ff5d4de
CI / test (pull_request) Successful in 8s
- test-dev → test-code：QE-Agent 一致化 label - Dev-Agent 新增 product-code label + [product] 前缀规则 - agent_poller.py 新增 create-issue action - QE/Dev Agent 轮询改为多轮递进：label → title 前缀 → 无标识分析 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
298 lines
11 KiB
Python
298 lines
11 KiB
Python
"""Helper for QE/Dev agents to interact with Gitea issues and PRs.
|
|
|
|
Usage:
|
|
python scripts/agent_poller.py --action list
|
|
python scripts/agent_poller.py --action list --labels test-code
|
|
python scripts/agent_poller.py --action get --issue 1
|
|
python scripts/agent_poller.py --action comment --issue 1 --body "Working on this"
|
|
python scripts/agent_poller.py --action create-issue --title "My issue" --labels test-code --body "..."
|
|
python scripts/agent_poller.py --action create-pr --issue 1 --branch test/issue-1
|
|
python scripts/agent_poller.py --action pr-status --pr 4
|
|
python scripts/agent_poller.py --action merge-pr --pr 4
|
|
python scripts/agent_poller.py --action close-issue --issue 2 --body "Done"
|
|
python scripts/agent_poller.py --action lifecycle --issue 2
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
GITEA_URL = os.environ.get("GITEA_URL", "http://localhost:3000")
|
|
GITEA_REPO = os.environ.get("GITEA_REPO", "pzhang_zywl/document_analyzer")
|
|
GITEA_TOKEN = os.environ.get("GITEA_API_TOKEN", "")
|
|
DEV_AGENT_ID = os.environ.get("DEV_AGENT_ID", "da-01")
|
|
QE_AGENT_ID = os.environ.get("QE_AGENT_ID", "")
|
|
|
|
# Signature appended to all comments / PR bodies
|
|
if QE_AGENT_ID:
|
|
AGENT_ID = QE_AGENT_ID
|
|
AGENT_SIG = f"\n\n---\n[qe-agent: {QE_AGENT_ID}]"
|
|
else:
|
|
AGENT_ID = DEV_AGENT_ID
|
|
AGENT_SIG = f"\n\n---\n[{DEV_AGENT_ID}]"
|
|
|
|
BASE = f"{GITEA_URL}/api/v1/repos/{GITEA_REPO}"
|
|
|
|
|
|
def _req(method, path, data=None):
|
|
url = f"{BASE}{path}"
|
|
payload = json.dumps(data).encode("utf-8") if data else None
|
|
req = urllib.request.Request(url, data=payload, method=method)
|
|
req.add_header("Authorization", f"token {GITEA_TOKEN}")
|
|
req.add_header("Content-Type", "application/json")
|
|
try:
|
|
with urllib.request.urlopen(req) as resp:
|
|
raw = resp.read()
|
|
if not raw:
|
|
return {} # Gitea merge returns 200 with empty body
|
|
return json.loads(raw)
|
|
except urllib.error.HTTPError as e:
|
|
body = e.read().decode()
|
|
print(f"API Error {e.code}: {body}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
# ── Issue operations ─────────────────────────────────────────────────────────
|
|
|
|
def list_issues(labels: list[str] | None = None):
|
|
url = "/issues?state=open"
|
|
if labels:
|
|
for lb in labels:
|
|
url += f"&labels={lb}"
|
|
issues = _req("GET", url)
|
|
if not issues:
|
|
label_hint = f" (filtered by {labels})" if labels else ""
|
|
print(f"No open issues found{label_hint}.")
|
|
return []
|
|
for i in issues:
|
|
issue_labels = [l["name"] for l in i.get("labels", [])]
|
|
print(f"#{i['number']} [{', '.join(issue_labels) if issue_labels else 'no label'}] {i['title']}")
|
|
return issues
|
|
|
|
|
|
def get_issue(num):
|
|
i = _req("GET", f"/issues/{num}")
|
|
print(f"## #{i['number']}: {i['title']}")
|
|
print(f"State: {i['state']}")
|
|
labels = [l["name"] for l in i.get("labels", [])]
|
|
print(f"Labels: {', '.join(labels) if labels else 'none'}")
|
|
print()
|
|
print(i.get("body", "(no description)"))
|
|
return i
|
|
|
|
|
|
def comment_issue(num, body):
|
|
i = _req("POST", f"/issues/{num}/comments", {"body": body + AGENT_SIG})
|
|
print(f"Comment added to #{num}")
|
|
return i
|
|
|
|
|
|
def close_issue(num, body=None):
|
|
"""Close an issue, optionally with a final comment (signature auto-appended)."""
|
|
if body:
|
|
comment_issue(num, body) # comment_issue already appends AGENT_SIG
|
|
i = _req("PATCH", f"/issues/{num}", {"state": "closed"})
|
|
print(f"Issue #{num} closed")
|
|
return i
|
|
|
|
|
|
def create_issue(title, body=None, labels=None):
|
|
"""Create a new Gitea issue.
|
|
|
|
Labels convention (per project rules):
|
|
- Product/feature issues → product-code
|
|
- Test code issues → test-code
|
|
"""
|
|
payload = {"title": title}
|
|
if body:
|
|
payload["body"] = body + AGENT_SIG
|
|
if labels:
|
|
payload["labels"] = [l.strip() for l in labels.split(",") if l.strip()]
|
|
i = _req("POST", "/issues", payload)
|
|
issue_labels = [l["name"] for l in i.get("labels", [])]
|
|
print(f"Issue #{i['number']} created: {i['title']}")
|
|
if issue_labels:
|
|
print(f"Labels: {', '.join(issue_labels)}")
|
|
print(f"URL: {i.get('html_url', i.get('url', ''))}")
|
|
return i
|
|
|
|
|
|
# ── PR operations ────────────────────────────────────────────────────────────
|
|
|
|
def create_pr(issue_num, branch, body=None):
|
|
"""Create a PR for the given issue and branch."""
|
|
issue = _req("GET", f"/issues/{issue_num}")
|
|
title = f"fix: {issue['title']} - Closes #{issue_num}"
|
|
if body is None:
|
|
body = f"Closes #{issue_num}\n\n{issue.get('body', '')}"
|
|
body += AGENT_SIG
|
|
pr = _req("POST", "/pulls", {
|
|
"title": title,
|
|
"head": branch,
|
|
"base": "main",
|
|
"body": body,
|
|
})
|
|
pr_url = pr.get('html_url', pr.get('url', ''))
|
|
print(f"PR created: {pr_url}")
|
|
return pr
|
|
|
|
|
|
def pr_status(pr_num):
|
|
"""Check PR state and CI status."""
|
|
pr = _req("GET", f"/pulls/{pr_num}")
|
|
print(f"PR #{pr['number']}: {pr['title']}")
|
|
print(f"State: {pr['state']}")
|
|
print(f"Merged: {pr.get('merged', False)}")
|
|
print(f"Mergeable: {pr.get('mergeable', 'unknown')}")
|
|
print(f"URL: {pr['html_url']}")
|
|
|
|
# CI status
|
|
sha = pr.get("head", {}).get("sha", "")
|
|
if sha:
|
|
try:
|
|
status = _req("GET", f"/commits/{sha}/status")
|
|
print(f"CI Status: {status.get('state', 'pending')}")
|
|
for s in status.get('statuses', []):
|
|
desc = s.get('description', '')
|
|
print(f" - {s.get('context')}: {s.get('state')} ({desc})")
|
|
except SystemExit:
|
|
print("CI Status: no statuses found")
|
|
return pr
|
|
|
|
|
|
def merge_pr(pr_num):
|
|
"""Merge a PR. Fails if CI hasn't passed or PR is not mergeable."""
|
|
pr = _req("GET", f"/pulls/{pr_num}")
|
|
|
|
if pr.get("state") == "closed":
|
|
if pr.get("merged"):
|
|
print(f"PR #{pr_num} already merged")
|
|
return pr
|
|
else:
|
|
print(f"PR #{pr_num} is closed (not merged)", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
# Check CI
|
|
sha = pr.get("head", {}).get("sha", "")
|
|
ci_passed = True
|
|
if sha:
|
|
try:
|
|
status = _req("GET", f"/commits/{sha}/status")
|
|
ci_state = status.get("state", "pending")
|
|
if ci_state in ("failure", "error"):
|
|
ci_passed = False
|
|
print(f"CI status: {ci_state} — cannot merge", file=sys.stderr)
|
|
for s in status.get('statuses', []):
|
|
print(f" {s.get('context')}: {s.get('state')}", file=sys.stderr)
|
|
except SystemExit:
|
|
pass # No statuses, proceed
|
|
|
|
if not ci_passed:
|
|
print("Merge blocked: CI has not passed", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
result = _req("POST", f"/pulls/{pr_num}/merge", {"Do": "merge"})
|
|
# Verify merge success by re-checking PR state
|
|
pr_after = _req("GET", f"/pulls/{pr_num}")
|
|
if pr_after.get("merged"):
|
|
print(f"PR #{pr_num} merged successfully")
|
|
elif result.get("merged"):
|
|
print(f"PR #{pr_num} merged successfully")
|
|
else:
|
|
print(f"Merge result: {result.get('message', 'unknown')}")
|
|
return result
|
|
|
|
|
|
# ── Lifecycle management ─────────────────────────────────────────────────────
|
|
|
|
def lifecycle(issue_num):
|
|
"""Print the full lifecycle status for an issue: branch, PR, CI, merge."""
|
|
print(f"=== Issue #{issue_num} Lifecycle ===\n")
|
|
|
|
issue = _req("GET", f"/issues/{issue_num}")
|
|
print(f"Issue: {issue['title']}")
|
|
print(f"State: {issue['state']}\n")
|
|
|
|
# Find associated PRs
|
|
prs = _req("GET", "/pulls?state=all")
|
|
related = [p for p in prs if f"Closes #{issue_num}" in p.get("body", "")
|
|
or f"#{issue_num}" in p.get("title", "")]
|
|
if related:
|
|
for pr in related:
|
|
print(f"PR #{pr['number']}: {pr['state']} (merged={pr.get('merged', False)})")
|
|
pr_status(pr["number"])
|
|
else:
|
|
print("No associated PR found.")
|
|
|
|
|
|
# ── CLI ──────────────────────────────────────────────────────────────────────
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Dev agent Gitea helper")
|
|
parser.add_argument("--action", required=True,
|
|
choices=["list", "get", "comment", "close-issue",
|
|
"create-issue", "create-pr", "pr-status", "merge-pr", "lifecycle"])
|
|
parser.add_argument("--issue", type=int)
|
|
parser.add_argument("--pr", type=int)
|
|
parser.add_argument("--title", help="Issue title (for 'create-issue' action)")
|
|
parser.add_argument("--branch")
|
|
parser.add_argument("--body")
|
|
parser.add_argument("--labels", help="Comma-separated labels (filter for 'list', assign for 'create-issue')")
|
|
args = parser.parse_args()
|
|
|
|
if not GITEA_TOKEN:
|
|
print("Error: GITEA_API_TOKEN environment variable is not set.", file=sys.stderr)
|
|
print("Set it via: export GITEA_API_TOKEN=your-token", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
if args.action == "list":
|
|
label_filter = [l.strip() for l in args.labels.split(",") if l.strip()] if args.labels else None
|
|
list_issues(label_filter)
|
|
elif args.action == "get":
|
|
if not args.issue:
|
|
print("--issue is required for 'get' action", file=sys.stderr)
|
|
sys.exit(1)
|
|
get_issue(args.issue)
|
|
elif args.action == "comment":
|
|
if not args.issue or not args.body:
|
|
print("--issue and --body are required for 'comment' action", file=sys.stderr)
|
|
sys.exit(1)
|
|
comment_issue(args.issue, args.body)
|
|
elif args.action == "close-issue":
|
|
if not args.issue:
|
|
print("--issue is required for 'close-issue' action", file=sys.stderr)
|
|
sys.exit(1)
|
|
close_issue(args.issue, args.body)
|
|
elif args.action == "create-issue":
|
|
if not args.title:
|
|
print("--title is required for 'create-issue' action", file=sys.stderr)
|
|
sys.exit(1)
|
|
create_issue(args.title, args.body, args.labels)
|
|
elif args.action == "create-pr":
|
|
if not args.issue or not args.branch:
|
|
print("--issue and --branch are required for 'create-pr' action", file=sys.stderr)
|
|
sys.exit(1)
|
|
create_pr(args.issue, args.branch, args.body)
|
|
elif args.action == "pr-status":
|
|
if not args.pr:
|
|
print("--pr is required for 'pr-status' action", file=sys.stderr)
|
|
sys.exit(1)
|
|
pr_status(args.pr)
|
|
elif args.action == "merge-pr":
|
|
if not args.pr:
|
|
print("--pr is required for 'merge-pr' action", file=sys.stderr)
|
|
sys.exit(1)
|
|
merge_pr(args.pr)
|
|
elif args.action == "lifecycle":
|
|
if not args.issue:
|
|
print("--issue is required for 'lifecycle' action", file=sys.stderr)
|
|
sys.exit(1)
|
|
lifecycle(args.issue)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|