Add initial implementation of serverless function for Gitea webhook handling

.gitea/workflows/lint.yml

@@ -0,0 +1,33 @@
+name: Lint and Syntax Check
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: '3.14'
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+          # Install flake8 for syntax checking
+          pip install flake8
+
+      - name: Lint with flake8
+        run: |
+          # stop the build if there are Python syntax errors or undefined names
+          flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
+          # exit-zero treats all errors as warnings.
+          flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics

README.md

@@ -0,0 +1,41 @@
+# Assign Me Openclaw (<a href="https://github.com/Space-Banane/shsf"><img src="https://github.com/Space-Banane/Space-Banane/blob/main/WORKS%20WITH%20SHSF%20-%20WHITE%20TEXT.png?raw=true" height="30" alt="Works with SHSF" style="vertical-align: middle;"></a>)
+
+This project is a serverless function (designed for SHSF) that handles Gitea webhooks for issue and pull request assignments. It notifies an AI agent (via OpenClaw) when they are assigned or unassigned, even if the unassignment event lacks assignee data. The function uses a database to track assignees across events, ensuring accurate notifications.
+
+## Features
+
+- **Multiple Assignee Support**: Correctly handles Gitea events with multiple assignees.
+- **State Persistence**: Uses a database to track assignees across events, ensuring "unassigned" events have the necessary context even when the payload is sparse.
+- **AI Integration**: Sends formatted Markdown notifications to an OpenClaw-compatible API.
+- **Assignment Logic**: Specifically identifies when the configured `AGENT_USERNAME` is involved in an assignment change.
+
+## Environment Variables
+
+The following environment variables are required for the service to function:
+
+| Variable | Description |
+| :--- | :--- |
+| `OPENCLAW_URL` | The endpoint URL for the OpenClaw API. (with hook path) |
+| `OPENCLAW_TOKEN` | The authentication token for OpenClaw. (from hooks, not gateway ui) |
+| `OPENCLAW_PROXY_AUTH` | (Optional) Proxy credentials in `username:password` format. |
+| `DATABASE_STORAGE_NAME` | The name of the database storage to use. |
+| `AGENT_USERNAME` | The Gitea username of the AI agent (e.g., `whateveryouragentisnamed`). |
+
+## How it works
+
+1. **Webhook Reception**: The `main(args)` function receives a POST request from Gitea.
+2. **Event Filtering**: It filters for `issues` and `pull_request` events with `assigned` or `unassigned` actions.
+3. **Database Lookups**: If an `unassigned` event lacks assignee data, the script retrieves the last known assignees from the database.
+4. **Agent Notification**: If the configured `AGENT_USERNAME` is found in the assignee list, a detailed Markdown message is constructed and sent to OpenClaw.
+5. **Database Sync**: The current state of assignees for the specific issue/PR is updated in the database for future reference.
+
+## Development
+
+The project consists of:
+- `main.py`: The entry point and logic for the webhook handler.
+- `_db_com.py`: (Internal dependency at SHSF Runner level) Database communication module.
+- `requirements.txt`: Python dependencies.
+
+## Usage in SHSF
+
+This script is intended to be deployed as a serverless function using SHSF. Ensure that all required environment variables are set in your Function.

main.py

@@ -0,0 +1,297 @@
+import json
+import requests
+import os
+from _db_com import database
+
+OPENCLAW_URL = os.getenv("OPENCLAW_URL")
+OPENCLAW_TOKEN = os.getenv("OPENCLAW_TOKEN")
+OPENCLAW_PROXY_AUTH = os.getenv("OPENCLAW_PROXY_AUTH")
+DATABASE_STORAGE = os.getenv("DATABASE_STORAGE_NAME")
+AGENT_USERNAME = os.getenv("AGENT_USERNAME")
+
+eventsToHandle = ["pull_request", "issues"]
+actionsToHandle = ["assigned", "unassigned"]
+
+
+def get_assignee_key(event_type, repository_id, number):
+    if event_type == "issues" or event_type == "issue":
+        return f"assignees_issue_{repository_id}_{number}"
+    elif event_type == "pull_request":
+        return f"assignees_pr_{repository_id}_{number}"
+    return None
+
+
+def get_stored_assignees(db, event_type, repository_id, number):
+    key = get_assignee_key(event_type, repository_id, number)
+    if not key:
+        return []
+    try:
+        assignees_str = db.get(DATABASE_STORAGE, key)
+        if assignees_str:
+            return [a.strip() for a in assignees_str.split(",") if a.strip()]
+    except Exception as e:
+        print(f"Failed to retrieve assignees from DB: {e}")
+    return []
+
+
+def set_stored_assignees(db, event_type, repository_id, number, assignees_list):
+    key = get_assignee_key(event_type, repository_id, number)
+    if not key:
+        return
+    try:
+        assignees_str = ",".join(assignees_list)
+        if assignees_str:
+            db.set(DATABASE_STORAGE, key, assignees_str)
+        else:
+            db.delete_item(DATABASE_STORAGE, key)
+    except Exception as e:
+        print(f"Failed to store assignees in DB: {e}")
+
+
+def delete_stored_assignees(db, event_type, repository_id, number):
+    key = get_assignee_key(event_type, repository_id, number)
+    if not key:
+        return
+    try:
+        db.delete_item(DATABASE_STORAGE, key)
+    except Exception as e:
+        print(f"Failed to delete assignees from DB: {e}")
+
+
+def sendToAgent(event_object):
+    headers = {"x-openclaw-token": OPENCLAW_TOKEN, "Content-Type": "application/json"}
+    action_str = "assigned" if event_object["action"] == "assigned" else "unassigned"
+    type_str = "issue" if event_object["type"] == "issue" else "pull request"
+
+    assignees = event_object.get("assignees", [event_object.get("assignee", "Unknown")])
+
+    if AGENT_USERNAME in assignees:
+        message = f"""# GITEA STATUS UPDATE
+- You were {action_str} to an {type_str} in {event_object['repository']} by {event_object['sender']}.
+raw event data:
+```json{json.dumps(event_object, indent=2)}```
+Please check the details and take necessary actions.
+
+DM your human in the main session and tell them about this change and if you should do anything about it.
+Inform them that you have to be told the exact details on what to do next, as you won't know what you sent them once they respond, which is a restriction of your system."""
+
+    try:
+        if OPENCLAW_PROXY_AUTH:
+            auth = tuple(OPENCLAW_PROXY_AUTH.split(":"))
+        else:
+            auth = None
+        response = requests.post(
+            OPENCLAW_URL, headers=headers, json={"message": message, "thinking": "low", "timeoutSeconds": 45, "model": "openrouter/google/gemini-3-flash-preview"}, auth=auth
+        )
+        if response.status_code == 200:
+            print(
+                f"Successfully notified OpenClaw with ({len(message)} chars) about {type_str} {action_str}."
+            )
+        else:
+            print(
+                f"Failed to notify OpenClaw. Status code: {response.status_code}, Response: {response.text}"
+            )
+    except Exception as e:
+        print(f"Error notifying OpenClaw: {e}")
+
+
+# SHSF Handler - Serverless
+def main(args):
+    required_data = ["OPENCLAW_URL", "OPENCLAW_TOKEN", "DATABASE_STORAGE_NAME", "AGENT_USERNAME"]
+    for var in required_data:
+        if not os.getenv(var):
+            return {
+                "_shsf": "v2",
+                "_code": 500,
+                "_res": {"error": f"Missing required environment variable: {var}"},
+                "_headers": {"Content-Type": "application/json"},
+            }
+    
+    headers = args.get("headers", {})
+    route = args.get("route")
+    event_type = headers.get("X-Gitea-Event") or headers.get("x-gitea-event")
+
+    if not route or not event_type:
+        return {
+            "_shsf": "v2",
+            "_code": 400,
+            "_res": {"error": "Missing route or X-Gitea-Event header"},
+            "_headers": {"Content-Type": "application/json"},
+        }
+
+    if event_type not in eventsToHandle:
+        return {
+            "_shsf": "v2",
+            "_code": 200,
+            "_res": {"status": f"Ignored event type: {event_type}"},
+            "_headers": {"Content-Type": "application/json"},
+        }
+
+    try:
+        db = database()
+    except:
+        print("Failed to connect to database")
+        return {
+            "_shsf": "v2",
+            "_code": 500,
+            "_res": {"error": "Database connection failed"},
+            "_headers": {"Content-Type": "application/json"},
+        }
+    data = args.get("body")
+    data = json.loads(data)
+
+    if not data:
+        return {
+            "_shsf": "v2",
+            "_code": 400,
+            "_res": {"error": "No JSON payload"},
+            "_headers": {"Content-Type": "application/json"},
+        }
+
+    if route == "webhook":
+        # Prepare DB
+        try:
+            db.create_storage(DATABASE_STORAGE, purpose="For fast key-value data")
+        except Exception as e:
+            print(f"Failed to create storage: {e}")
+            # Probably already exists, continue anyway
+
+        action = data.get("action")
+        print(f"Action: {action}, Event Type: {event_type}")
+
+        if not action:
+            return {
+                "_shsf": "v2",
+                "_code": 400,
+                "_res": {"error": "Missing action in payload"},
+                "_headers": {"Content-Type": "application/json"},
+            }
+
+        if action not in actionsToHandle:
+            return {
+                "_shsf": "v2",
+                "_code": 200,
+                "_res": {"status": f"Ignored action: {action}"},
+                "_headers": {"Content-Type": "application/json"},
+            }
+
+        if action in ["assigned", "unassigned"]:
+            assignee_data = data.get("assignee")
+            assignees_data = data.get("assignees")
+
+            # If "assignee" is missing in the root, look for it in the issue/pull_request object
+            if not assignee_data:
+                assignee_data = (
+                    data.get("issue") or data.get("pull_request") or {}
+                ).get("assignee")
+
+            if not assignees_data:
+                assignees_data = (
+                    data.get("issue") or data.get("pull_request") or {}
+                ).get("assignees")
+
+            # Extract logins from the list of assignees, or single assignee
+            assignees_list = []
+            if assignees_data:
+                assignees_list = [
+                    a.get("login") for a in assignees_data if a.get("login")
+                ]
+            elif assignee_data:
+                assignees_list = (
+                    [assignee_data.get("login")] if assignee_data.get("login") else []
+                )
+
+            repository = data.get("repository", {}).get("full_name", "Unknown")
+            repo_id = data.get("repository", {}).get("id")
+            sender = data.get("sender", {}).get("login", "Unknown")
+
+            event_object = None
+
+            # Use helpers to get stored assignees if needed
+            issue_number = (data.get("issue") or data.get("pull_request") or {}).get(
+                "number"
+            )
+
+            # Get assignees from DB (previous events) if the current payload has none
+            if not assignees_list:
+                assignees_list = get_stored_assignees(
+                    db, event_type, repo_id, issue_number
+                )
+
+            if not assignees_list:
+                assignees_list = ["Unknown"]
+
+            if event_type == "issues":
+                issue_data = data.get("issue", {})
+                event_object = {
+                    "type": "issue",
+                    "action": action,
+                    "repository": repository,
+                    "number": issue_data.get("number"),
+                    "title": issue_data.get("title"),
+                    "assignees": assignees_list,
+                    "sender": sender,
+                    "url": issue_data.get("html_url"),
+                    "state": issue_data.get("state"),
+                }
+
+            elif event_type == "pull_request":
+                pr_data = data.get("pull_request", {})
+                event_object = {
+                    "type": "pull_request",
+                    "action": action,
+                    "repository": repository,
+                    "number": pr_data.get("number"),
+                    "title": pr_data.get("title"),
+                    "assignees": assignees_list,
+                    "sender": sender,
+                    "url": pr_data.get("html_url"),
+                    "state": pr_data.get("state"),
+                    "merged": pr_data.get("merged", False),
+                }
+
+            if event_object:
+                # Use helpers to store/delete assignees
+                if action == "assigned":
+                    set_stored_assignees(
+                        db, event_type, repo_id, event_object["number"], assignees_list
+                    )
+
+                if action == "unassigned":
+                    # If the payload STILL has assignees, it means someone is still assigned
+                    # If it's empty (or only contains "Unknown"), delete it.
+                    if not assignees_data or len(assignees_data) == 0:
+                        delete_stored_assignees(
+                            db, event_type, repo_id, event_object["number"]
+                        )
+                    else:
+                        set_stored_assignees(
+                            db,
+                            event_type,
+                            repo_id,
+                            event_object["number"],
+                            assignees_list,
+                        )
+
+                print(f"--- {event_object['type'].upper()} {action.capitalize()} ---")
+                print(f"Title: {event_object['title']}")
+                print(f"Assignees: {', '.join(event_object['assignees'])}")
+                print(f"Repo: {event_object['repository']}")
+                print(f"By: {event_object['sender']}")
+
+                # Send to OpenClaw
+                sendToAgent(event_object)
+
+        return {
+            "_shsf": "v2",
+            "_code": 200,
+            "_res": {"status": "received"},
+            "_headers": {"Content-Type": "application/json"},
+        }
+    else:
+        return {
+            "_shsf": "v2",
+            "_code": 404,
+            "_res": {"error": "Not Found"},
+            "_headers": {"Content-Type": "application/json"},
+        }

requirements.txt

@@ -0,0 +1 @@
+requests