feat: gitea webhook listener — instant reply on rockachopa comments

Listens on port 7777 for issue_comment events.
Spawns hermes-agent to reply when rockachopa comments.
Webhooks registered on Timmy-time-dashboard and alexanderwhitestone.com.
This commit is contained in:
Alexander Whitestone
2026-03-15 13:34:24 -04:00
parent b70cf27dab
commit b42220aa79

135
bin/gitea-webhook-listener.py Executable file
View File

@@ -0,0 +1,135 @@
#!/usr/bin/env python3
"""
Gitea Webhook Listener — The Real Watcher
Listens for issue_comment events from Gitea webhooks.
When rockachopa comments, spawns Hermes to reply instantly.
"""
import json
import os
import subprocess
import sys
from http.server import HTTPServer, BaseHTTPRequestHandler
from datetime import datetime
PORT = 7777
TOKEN_PATH = os.path.expanduser("~/.hermes/gitea_token")
LOG_PATH = os.path.expanduser("~/.hermes/logs/webhook.log")
os.makedirs(os.path.dirname(LOG_PATH), exist_ok=True)
def log(msg):
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
line = f"[{ts}] {msg}"
print(line, flush=True)
with open(LOG_PATH, "a") as f:
f.write(line + "\n")
def spawn_hermes_reply(issue_number, issue_title, comment_body, comment_user, repo_full_name):
"""Spawn a hermes-agent to reply to the comment."""
prompt = f"""You are Hermes Agent. Rockachopa (Alexander) just commented on Gitea issue #{issue_number} in {repo_full_name}.
ISSUE: #{issue_number}{issue_title}
HIS COMMENT:
{comment_body}
YOUR TASK:
1. Read the full issue and all comments for context:
API: http://localhost:3000/api/v1/repos/{repo_full_name}/issues/{issue_number}
Comments: http://localhost:3000/api/v1/repos/{repo_full_name}/issues/{issue_number}/comments
Token: read from ~/.hermes/gitea_token
2. Write a thoughtful reply that addresses what Alexander said specifically.
- If he asked a question, answer it.
- If he gave direction, acknowledge and adapt the plan.
- If he made an observation, engage with it genuinely.
- If he's being mythical or philosophical, meet him there.
- Keep it concise. Don't over-explain. Don't pad.
3. Post your reply to the issue comments endpoint.
You are Hermes. Be genuine, not performative. Match his energy."""
# Use hermes CLI to handle this
try:
subprocess.Popen(
["hermes", "-q", prompt],
stdout=open(os.path.expanduser("~/.hermes/logs/webhook-reply.log"), "a"),
stderr=subprocess.STDOUT,
start_new_session=True
)
log(f" → Spawned hermes to reply to #{issue_number}")
except FileNotFoundError:
# Fallback: use the API directly with a simple reply
log(f" → hermes CLI not found, trying direct API reply")
try:
import urllib.request
token = open(TOKEN_PATH).read().strip()
url = f"http://localhost:3000/api/v1/repos/{repo_full_name}/issues/{issue_number}/comments?token={token}"
data = json.dumps({"body": f"Heard you, Alexander. Reading the full thread now and will follow up.\n\n> {comment_body[:200]}"}).encode()
req = urllib.request.Request(url, data, headers={"Content-Type": "application/json"})
urllib.request.urlopen(req)
log(f" → Posted acknowledgment to #{issue_number}")
except Exception as e:
log(f" → Failed to reply: {e}")
class WebhookHandler(BaseHTTPRequestHandler):
def do_POST(self):
content_length = int(self.headers.get('Content-Length', 0))
body = self.rfile.read(content_length)
self.send_response(200)
self.send_header('Content-Type', 'text/plain')
self.end_headers()
self.wfile.write(b'ok')
try:
payload = json.loads(body)
except json.JSONDecodeError:
log("Received non-JSON payload, ignoring")
return
action = payload.get("action", "")
comment = payload.get("comment", {})
issue = payload.get("issue", {})
repo = payload.get("repository", {})
commenter = comment.get("user", {}).get("login", "")
issue_number = issue.get("number", 0)
issue_title = issue.get("title", "")
comment_body = comment.get("body", "")
repo_full_name = repo.get("full_name", "")
# Only react to new comments from rockachopa
if action != "created":
log(f"Ignoring action={action} on #{issue_number}")
return
if commenter != "rockachopa":
log(f"Ignoring comment from {commenter} on #{issue_number}")
return
log(f"rockachopa commented on #{issue_number} ({issue_title})")
log(f" Comment: {comment_body[:100]}...")
spawn_hermes_reply(issue_number, issue_title, comment_body, commenter, repo_full_name)
def do_GET(self):
self.send_response(200)
self.send_header('Content-Type', 'text/plain')
self.end_headers()
self.wfile.write(b'Gitea webhook listener alive\n')
def log_message(self, format, *args):
# Suppress default HTTP logging, we use our own
pass
if __name__ == "__main__":
log(f"Starting Gitea webhook listener on port {PORT}")
server = HTTPServer(("127.0.0.1", PORT), WebhookHandler)
try:
server.serve_forever()
except KeyboardInterrupt:
log("Shutting down")
server.server_close()