agent: implement issue #5 - Feat: 实现 agent 与 human 在 PR 中的交互

This commit is contained in:
2026-05-06 16:43:52 +08:00
parent aa8afa1a63
commit d227443e90
9 changed files with 596 additions and 76 deletions

View File

@@ -1,6 +1,7 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from typing import Any
import httpx
@@ -27,6 +28,16 @@ class GiteaPullRequest:
merged: bool
@dataclass(frozen=True)
class GiteaComment:
id: int
body: str
author: str
html_url: str
created_at: datetime | None
updated_at: datetime | None
@dataclass(frozen=True)
class GiteaRepository:
owner: str
@@ -140,12 +151,32 @@ class GiteaClient:
response.raise_for_status()
return pull_request_from_payload(response.json())
def post_issue_comment(self, *, owner: str, name: str, issue_number: int, body: str) -> None:
def list_issue_comments(self, *, owner: str, name: str, issue_number: int) -> list[GiteaComment]:
comments: list[GiteaComment] = []
page = 1
limit = 50
while True:
response = self.client.get(
f"/repos/{owner}/{name}/issues/{issue_number}/comments",
params={"page": page, "limit": limit},
)
response.raise_for_status()
payload = response.json()
if not payload:
break
comments.extend(comment_from_payload(item) for item in payload)
if len(payload) < limit:
break
page += 1
return comments
def post_issue_comment(self, *, owner: str, name: str, issue_number: int, body: str) -> GiteaComment:
response = self.client.post(
f"/repos/{owner}/{name}/issues/{issue_number}/comments",
json={"body": body},
)
response.raise_for_status()
return comment_from_payload(response.json())
def close_issue(self, *, owner: str, name: str, issue_number: int) -> None:
response = self.client.patch(
@@ -192,3 +223,24 @@ def pull_request_from_payload(payload: dict[str, Any]) -> GiteaPullRequest:
state=payload.get("state") or "",
merged=merged,
)
def comment_from_payload(payload: dict[str, Any]) -> GiteaComment:
user_payload = payload.get("user") or payload.get("poster") or {}
author = user_payload.get("login") or user_payload.get("username") or ""
return GiteaComment(
id=int(payload["id"]),
body=payload.get("body") or "",
author=str(author),
html_url=payload.get("html_url") or payload.get("url") or "",
created_at=parse_gitea_dt(payload.get("created_at")),
updated_at=parse_gitea_dt(payload.get("updated_at")),
)
def parse_gitea_dt(value: str | None) -> datetime | None:
if not value:
return None
if value.endswith("Z"):
value = value[:-1] + "+00:00"
return datetime.fromisoformat(value)