gahow/agent-manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

代理实现：Enhancement: 检查现有的代码框架，确认是否需要优化 #4

Merged
gahow merged 1 commits from agent/issue-3-enhancement into main 2026-05-06 08:16:06 +00:00
Conversation 2 Commits 1 Files Changed 4 +90 -18
4 changed files with 90 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
src/agent_gitea/gitea.py
View File

@@ -85,13 +85,19 @@ class GiteaClient:
return repositories
def list_open_issues(self, owner: str, name: str) -> list[GiteaIssue]:
issues: list[GiteaIssue] = []
page = 1
limit = 50
while True:
response = self.client.get(
f"/repos/{owner}/{name}/issues",
params={"state": "open", "type": "issues", "limit": 50},
params={"state": "open", "type": "issues", "page": page, "limit": limit},
)
response.raise_for_status()
issues: list[GiteaIssue] = []
for item in response.json():
payload = response.json()
if not payload:
break
for item in payload:
if item.get("pull_request"):
continue
issues.append(
@@ -104,6 +110,9 @@ class GiteaClient:
html_url=item.get("html_url") or item.get("url") or "",
)
)
if len(payload) < limit:
break
page += 1
return issues
def create_pull_request(

10
src/agent_gitea/rendering.py
View File

@@ -86,11 +86,13 @@ def render_pr_body(issue: IssueRecord, implementation_report: str) -> str:
def parse_review_report(raw: str) -> ReviewReport:
verdict_match = re.search(r"(?im)^\s*(?:##\s*)?Verdict\s*:?\s*`?([A-Z_]+)`?", raw)
verdict_match = re.search(r"(?im)^\s*(?:[-*]\s*)?(?:##\s*)?Verdict\s*:?\s*`?([A-Z_]+)`?", raw)
verdict = verdict_match.group(1) if verdict_match else "NEEDS_HUMAN_DECISION"
if verdict not in VALID_VERDICTS:
verdict = "NEEDS_HUMAN_DECISION"
suggested = extract_section(raw, "Suggested PR Comment").strip()
if not suggested:
suggested = extract_inline_field(raw, "Suggested PR Comment").strip()
if not suggested:
suggested = raw.strip()
return ReviewReport(verdict=verdict, raw=raw, suggested_pr_comment=suggested)
@@ -104,6 +106,12 @@ def extract_section(raw: str, title: str) -> str:
return match.group("body") if match else ""
def extract_inline_field(raw: str, title: str) -> str:
pattern = re.compile(rf"(?im)^\s*(?:[-*]\s*)?{re.escape(title)}\s*:\s*(?P<body>.+?)\s*$")
match = pattern.search(raw)
return match.group("body") if match else ""
def render_human_review_summary(review: ReviewReport) -> str:
return f"""## 代理评审摘要

43
tests/test_gitea_service.py
View File

@@ -119,6 +119,49 @@ def test_list_open_issues_keeps_normal_issues_with_null_pull_request():
assert [issue.number for issue in issues] == [1]
def test_list_open_issues_reads_all_pages():
seen_pages: list[int] = []
def handler(request: httpx.Request) -> httpx.Response:
assert request.url.path == "/api/v1/repos/acme/service/issues"
page = int(request.url.params["page"])
limit = int(request.url.params["limit"])
seen_pages.append(page)
if page == 1:
return httpx.Response(
200,
json=[
{
"number": number,
"title": f"Issue {number}",
"body": "",
"state": "open",
"labels": [],
}
for number in range(1, limit + 1)
],
)
if page == 2:
return httpx.Response(
200,
json=[
{
"number": 51,
"title": "Issue 51",
"body": "",
"state": "open",
"labels": [],
}
],
)
return httpx.Response(500)
issues = make_client(handler).list_open_issues("acme", "service")
assert [issue.number for issue in issues] == list(range(1, 52))
assert seen_pages == [1, 2]
class FakeWorkspaceManager:
def __init__(self, root: Path, *, diff: bool = True):
self.root = root

12
tests/test_rendering_workspace.py
View File

@@ -58,3 +58,15 @@ Please add tests.
assert parsed.verdict == "REQUEST_CHANGES"
assert parsed.suggested_pr_comment == "Please add tests."
def test_review_report_parsing_accepts_prompt_bullet_contract():
report = """- Verdict: APPROVE
- Summary: Done
- Suggested PR Comment: Looks good.
"""
parsed = parse_review_report(report)
assert parsed.verdict == "APPROVE"
assert parsed.suggested_pr_comment == "Looks good."