119 lines
3.9 KiB
Python
119 lines
3.9 KiB
Python
"""Helper for dev agent to interact with Gitea issues.
|
|
|
|
Usage:
|
|
python scripts/agent_poller.py --action list
|
|
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-pr --issue 1 --branch fix/issue-1
|
|
"""
|
|
|
|
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", "")
|
|
|
|
BASE = f"{GITEA_URL}/api/v1/repos/{GITEA_REPO}"
|
|
TARGET_LABELS = {"qe-feedback", "ci-failure"}
|
|
|
|
|
|
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:
|
|
return json.loads(resp.read())
|
|
except urllib.error.HTTPError as e:
|
|
body = e.read().decode()
|
|
print(f"API Error {e.code}: {body}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def list_issues():
|
|
issues = _req("GET", "/issues?state=open")
|
|
if not issues:
|
|
print("No open issues found.")
|
|
return []
|
|
for i in issues:
|
|
labels = [l["name"] for l in i.get("labels", [])]
|
|
if TARGET_LABELS & set(labels):
|
|
print(f"#{i['number']} [{', '.join(labels)}] {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})
|
|
print(f"Comment added to #{num}")
|
|
return i
|
|
|
|
|
|
def create_pr(issue_num, branch):
|
|
# Get issue title for PR description
|
|
issue = _req("GET", f"/issues/{issue_num}")
|
|
title = f"Fix #{issue_num}: {issue['title'].replace('CI Failure: ', '')}"
|
|
body = f"Closes #{issue_num}\n\n{issue.get('body', '')}\n\n🤖 Generated by dev agent"
|
|
pr = _req("POST", "/pulls", {
|
|
"title": title,
|
|
"head": branch,
|
|
"base": "main",
|
|
"body": body,
|
|
})
|
|
print(f"PR created: {pr.get('html_url', pr.get('url', ''))}")
|
|
return pr
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Dev agent Gitea helper")
|
|
parser.add_argument("--action", required=True,
|
|
choices=["list", "get", "comment", "create-pr"])
|
|
parser.add_argument("--issue", type=int)
|
|
parser.add_argument("--branch")
|
|
parser.add_argument("--body")
|
|
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":
|
|
list_issues()
|
|
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 == "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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|