create /gitea-webhook entry point #1

New Issue

2026-05-20T12:38:31+10:00

armistace commented

2026-05-20 12:38:31 +10:00

What You Need to Do

Edit the container's source code (the src/pr_reviewer/main.py file) and add the endpoint shown above.
Rebuild the Docker image and deploy it again with the new port mapping.
sett environment variables in the container:
GITEA_URL = your Gitea instance base URL
GITEA_TOKEN = a personal access token with read:repository scope
GITEA_SECRET = the webhook secret you set in Gitea (optional but recommended)
Set the Gitea webhook target URL to http://192.168.178.160:30001/api/v1/gitea-webhook.

Now every PR event will be directly processed by the container, fetching the diffs and running the three parallel reviews – no extra adapter needed.

After the review finishes, you may also want to post the result as a comment on the PR. That would be a small addition to the endpoint: use the Gitea API to create an issue comment after run_review_flow returns. Let me know if you'd like that extension as well.
Code "Above"

# Add to src/pr_reviewer/main.py (or a separate module)

import os
import hmac
import hashlib
import requests
from fastapi import APIRouter, Request, HTTPException
from .flow import run_review_flow  # existing flow function
from .state import ReviewState      # existing state model

router = APIRouter()

GITEA_URL = os.getenv("GITEA_URL", "http://192.168.178.160:3000")  # adjust
GITEA_TOKEN = os.getenv("GITEA_TOKEN")  # required, with repo read access
WEBHOOK_SECRET = os.getenv("GITEA_SECRET", "")

def verify_signature(payload: bytes, signature: str) -> bool:
    if not WEBHOOK_SECRET:
        return True
    mac = hmac.new(WEBHOOK_SECRET.encode(), payload, hashlib.sha256).hexdigest()
    return hmac.compare_digest(mac, signature)

def fetch_pr_files(repo_full: str, pr_number: int):
    """Fetch file diffs and contents from Gitea API."""
    headers = {"Authorization": f"token {GITEA_TOKEN}"}
    url = f"{GITEA_URL}/api/v1/repos/{repo_full}/pulls/{pr_number}/files"
    resp = requests.get(url, headers=headers)
    resp.raise_for_status()
    files_data = resp.json()

    files = []
    for f in files_data:
        filename = f["filename"]
        status = f["status"]  # "added", "modified", "removed"
        # For added/modified, fetch file content from raw endpoint
        content = None
        if status in ("added", "modified"):
            raw_url = f"{GITEA_URL}/api/v1/repos/{repo_full}/contents/{filename}?ref={pr_number}"
            raw_resp = requests.get(raw_url, headers=headers)
            if raw_resp.ok:
                raw = raw_resp.json()
                if raw.get("encoding") == "base64":
                    import base64
                    content = base64.b64decode(raw["content"]).decode("utf-8")
        files.append({
            "path": filename,
            "status": status,
            "content": content or "",
            "additions": f.get("additions", 0),
            "deletions": f.get("deletions", 0),
            "patch": f.get("patch", "")
        })
    return files

@router.post("/gitea-webhook")
async def gitea_webhook(request: Request):
    # Verify signature
    body = await request.body()
    sig = request.headers.get("X-Gitea-Signature", "")
    if not verify_signature(body, sig):
        raise HTTPException(status_code=403, detail="Invalid signature")

    data = await request.json()
    event = request.headers.get("X-Gitea-Event")

    if event == "pull_request":
        pr_number = data["pull_request"]["number"]
        repo_full = data["repository"]["full_name"]
        repo_url = data["repository"].get("html_url", f"{GITEA_URL}/{repo_full}")

        # Fetch files
        try:
            files = fetch_pr_files(repo_full, pr_number)
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Error fetching files: {e}")

        # Build state for internal flow (assumes ReviewState model)
        state = ReviewState(
            pr_id=str(pr_number),
            title=data["pull_request"]["title"],
            description=data["pull_request"].get("body", ""),
            repo={"name": repo_full, "url": repo_url},
            source={
                "branch": data["pull_request"]["head"]["label"],
                "commit": data["pull_request"]["head"]["sha"]
            },
            target={
                "branch": data["pull_request"]["base"]["label"],
                "commit": data["pull_request"]["base"]["sha"]
            },
            files=files
        )

        # Run the review flow (the existing function)
        result = await run_review_flow(state)
        return result

    return {"status": "ignored"}
    ```

What You Need to Do Edit the container's source code (the src/pr_reviewer/main.py file) and add the endpoint shown above. Rebuild the Docker image and deploy it again with the new port mapping. sett environment variables in the container: GITEA_URL = your Gitea instance base URL GITEA_TOKEN = a personal access token with read:repository scope GITEA_SECRET = the webhook secret you set in Gitea (optional but recommended) Set the Gitea webhook target URL to http://192.168.178.160:30001/api/v1/gitea-webhook. Now every PR event will be directly processed by the container, fetching the diffs and running the three parallel reviews – no extra adapter needed. After the review finishes, you may also want to post the result as a comment on the PR. That would be a small addition to the endpoint: use the Gitea API to create an issue comment after run_review_flow returns. Let me know if you'd like that extension as well. Code "Above" ```python # Add to src/pr_reviewer/main.py (or a separate module) import os import hmac import hashlib import requests from fastapi import APIRouter, Request, HTTPException from .flow import run_review_flow # existing flow function from .state import ReviewState # existing state model router = APIRouter() GITEA_URL = os.getenv("GITEA_URL", "http://192.168.178.160:3000") # adjust GITEA_TOKEN = os.getenv("GITEA_TOKEN") # required, with repo read access WEBHOOK_SECRET = os.getenv("GITEA_SECRET", "") def verify_signature(payload: bytes, signature: str) -> bool: if not WEBHOOK_SECRET: return True mac = hmac.new(WEBHOOK_SECRET.encode(), payload, hashlib.sha256).hexdigest() return hmac.compare_digest(mac, signature) def fetch_pr_files(repo_full: str, pr_number: int): """Fetch file diffs and contents from Gitea API.""" headers = {"Authorization": f"token {GITEA_TOKEN}"} url = f"{GITEA_URL}/api/v1/repos/{repo_full}/pulls/{pr_number}/files" resp = requests.get(url, headers=headers) resp.raise_for_status() files_data = resp.json() files = [] for f in files_data: filename = f["filename"] status = f["status"] # "added", "modified", "removed" # For added/modified, fetch file content from raw endpoint content = None if status in ("added", "modified"): raw_url = f"{GITEA_URL}/api/v1/repos/{repo_full}/contents/{filename}?ref={pr_number}" raw_resp = requests.get(raw_url, headers=headers) if raw_resp.ok: raw = raw_resp.json() if raw.get("encoding") == "base64": import base64 content = base64.b64decode(raw["content"]).decode("utf-8") files.append({ "path": filename, "status": status, "content": content or "", "additions": f.get("additions", 0), "deletions": f.get("deletions", 0), "patch": f.get("patch", "") }) return files @router.post("/gitea-webhook") async def gitea_webhook(request: Request): # Verify signature body = await request.body() sig = request.headers.get("X-Gitea-Signature", "") if not verify_signature(body, sig): raise HTTPException(status_code=403, detail="Invalid signature") data = await request.json() event = request.headers.get("X-Gitea-Event") if event == "pull_request": pr_number = data["pull_request"]["number"] repo_full = data["repository"]["full_name"] repo_url = data["repository"].get("html_url", f"{GITEA_URL}/{repo_full}") # Fetch files try: files = fetch_pr_files(repo_full, pr_number) except Exception as e: raise HTTPException(status_code=500, detail=f"Error fetching files: {e}") # Build state for internal flow (assumes ReviewState model) state = ReviewState( pr_id=str(pr_number), title=data["pull_request"]["title"], description=data["pull_request"].get("body", ""), repo={"name": repo_full, "url": repo_url}, source={ "branch": data["pull_request"]["head"]["label"], "commit": data["pull_request"]["head"]["sha"] }, target={ "branch": data["pull_request"]["base"]["label"], "commit": data["pull_request"]["base"]["sha"] }, files=files ) # Run the review flow (the existing function) result = await run_review_flow(state) return result return {"status": "ignored"} ```

armistace added reference master

2026-05-20 12:38:43 +10:00

armistace self-assigned this 2026-05-20 12:38:52 +10:00

armistace referenced a pull request that will close this issue

2026-05-20 23:23:14 +10:00

git-webook-entrypoint #2

Sign in to join this conversation.

1 Participants

Notifications

Due Date

No due date set.

Dependencies

No dependencies set.

Reference: armistace/pr_reviewer#1