code-review-agent/scripts/gitea_api.py

"""Gitea API helpers for the code-review-agent.

Fetches PR data (metadata, diff, files), linked issues, and posts reviews.
"""

import json
import re
import sys
import urllib.error
import urllib.request


# ── Low-level helper ────────────────────────────────────────────────────

def _api_req(gitea_url, api_token, method, path, body=None):
    """Send a Gitea API request, return parsed JSON or raw bytes."""
    url = f"{gitea_url}/api/v1/repos{path}"
    data = json.dumps(body).encode("utf-8") if body else None
    req = urllib.request.Request(url, method=method, data=data)
    req.add_header("Authorization", f"token {api_token}")
    if data:
        req.add_header("Content-Type", "application/json")

    try:
        with urllib.request.urlopen(req) as resp:
            raw = resp.read()
            if not raw:
                return {} if path.endswith("/merge") else None
            content_type = resp.headers.get("Content-Type", "")
            if "application/json" in content_type:
                return json.loads(raw)
            return raw  # raw bytes for .diff endpoints
    except urllib.error.HTTPError as e:
        err_body = e.read().decode(errors="replace")
        print(f"API Error {e.code} on {method} {path}: {err_body}", file=sys.stderr)
        sys.exit(1)


# ── PR endpoints ────────────────────────────────────────────────────────

def fetch_pr_metadata(gitea_url, api_token, target_repo, pr_num):
    """Fetch PR details: title, description, head SHA."""
    pr = _api_req(gitea_url, api_token, "GET", f"/{target_repo}/pulls/{pr_num}")
    return {
        "number": pr["number"],
        "title": pr["title"],
        "body": pr.get("body", ""),
        "head_sha": pr.get("head", {}).get("sha", ""),
        "html_url": pr.get("html_url", ""),
    }


def fetch_pr_diff(gitea_url, api_token, target_repo, pr_num):
    """Fetch raw unified diff for a PR."""
    raw = _api_req(gitea_url, api_token, "GET", f"/{target_repo}/pulls/{pr_num}.diff")
    return raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else ""


def fetch_pr_files(gitea_url, api_token, target_repo, pr_num):
    """Fetch changed file list with stats."""
    files = _api_req(gitea_url, api_token, "GET", f"/{target_repo}/pulls/{pr_num}/files")
    result = []
    for f in files:
        result.append({
            "filename": f["filename"],
            "status": f["status"],
            "additions": f.get("additions", 0),
            "deletions": f.get("deletions", 0),
        })
    return result


def post_review(gitea_url, api_token, target_repo, pr_num, review):
    """Post a PR review to Gitea."""
    comments = review.get("comments", [])
    gitea_comments = []
    for c in comments:
        gc = {"path": c["path"], "body": c["body"]}
        if c.get("line"):
            gc["new_line"] = c["line"]
        if c.get("old_line"):
            gc["old_line"] = c["old_line"]
        gitea_comments.append(gc)

    payload = {
        "body": review["body"],
        "event": review.get("event", "COMMENT"),
    }
    if gitea_comments:
        payload["comments"] = gitea_comments

    _api_req(gitea_url, api_token, "POST",
             f"/{target_repo}/pulls/{pr_num}/reviews", body=payload)
    print(f"Review posted to PR #{pr_num}")


# ── Issue endpoints ─────────────────────────────────────────────────────

_ISSUE_REF_RE = re.compile(
    r"(?:fixe?s?|close?s?|resolve?s?|refs?|see)\s+"
    r"((?:[\w.-]+/[\w.-]+)?#\d+)",
    re.IGNORECASE,
)
_BARE_ISSUE_RE = re.compile(r"(?:^|\s)(#\d+)")


def _extract_issue_refs(pr_body):
    """Extract issue references from PR description.

    Returns list of ``(owner, repo, issue_num)`` tuples.
    ``owner/repo`` is ``None`` when using shorthand ``#123``.
    """
    refs = []
    for m in _ISSUE_REF_RE.finditer(pr_body):
        ref = m.group(1)
        match = re.match(r"(?:([\w.-]+)/([\w.-]+))?#(\d+)", ref)
        if match:
            refs.append((match.group(1), match.group(2), int(match.group(3))))

    # Add bare #123 refs not already captured
    for m in _BARE_ISSUE_RE.finditer(pr_body):
        num = int(m.group(1)[1:])
        if not any(r[2] == num for r in refs):
            refs.append((None, None, num))

    return refs


def fetch_issue(gitea_url, api_token, target_repo, issue_num):
    """Fetch a single issue from Gitea.

    Returns dict with keys: number, title, body, state, labels, html_url.
    """
    data = _api_req(gitea_url, api_token, "GET",
                    f"/{target_repo}/issues/{issue_num}")
    return {
        "number": data["number"],
        "title": data["title"],
        "body": data.get("body", ""),
        "state": data.get("state", "open"),
        "labels": [lb["name"] for lb in data.get("labels", [])],
        "html_url": data.get("html_url", ""),
    }


def fetch_linked_issues(gitea_url, api_token, target_repo, pr_body):
    """Extract issue refs from PR body and fetch each issue.

    Returns list of issue dicts (see ``fetch_issue``).
    """
    refs = _extract_issue_refs(pr_body)
    issues = []
    for owner, repo, num in refs:
        if owner and repo:
            repo_path = f"{owner}/{repo}"
        else:
            repo_path = target_repo
        try:
            issues.append(fetch_issue(gitea_url, api_token, repo_path, num))
        except SystemExit:
            # issue fetch failed (404, etc.) — skip and continue
            print(f"[gitea] Warning: could not fetch issue #{num}, skipping",
                  file=sys.stderr)
    return issues


def format_issues_markdown(issues):
    """Format a list of issue dicts as a Markdown string."""
    if not issues:
        return ""
    parts = ["# Linked Issues\n"]
    for iss in issues:
        labels = ", ".join(iss["labels"]) if iss["labels"] else "none"
        parts.append(
            f"## #{iss['number']}: {iss['title']}\n"
            f"- **State**: {iss['state']}\n"
            f"- **Labels**: {labels}\n"
            f"\n{iss['body']}\n"
        )
    return "\n".join(parts)