Upload webhook_automation_manager.py with huggingface_hub

webhook_automation_manager.py

@@ -0,0 +1,90 @@
+"""
+Hugging Face Webhooks Automation Substrate
+Automates real-time event listening and CI/CD triggers on repository changes.
+Supports `create_webhook`, payload signature validation, and event handling.
+"""
+
+from huggingface_hub import HfApi, get_token
+import os
+import json
+import hmac
+import hashlib
+
+class WebhookAutomationManager:
+    def __init__(self, token: str = None):
+        self.token = token or get_token()
+        self.api = HfApi(token=self.token)
+
+    def register_repo_webhook(
+        self,
+        watched_repos: list[dict],
+        job_id: str = None,
+        url: str = None,
+        secret: str = None,
+        domains: list[str] = None
+    ) -> dict:
+        """
+        Create a webhook on Hugging Face to monitor repository or discussion changes.
+        watched: list of dicts e.g. [{"type": "user", "name": "bbkdevops"}]
+        domains: ["repo", "discussion"]
+        """
+        domains = domains or ["repo", "discussion"]
+        secret = secret or os.environ.get("HF_WEBHOOK_SECRET", "fiber-moe-webhook-secret-2026")
+        
+        try:
+            # Check if create_webhook exists in HfApi
+            if hasattr(self.api, "create_webhook"):
+                webhook = self.api.create_webhook(
+                    watched=watched_repos,
+                    domains=domains,
+                    secret=secret,
+                    url=url,
+                    job_id=job_id
+                )
+                return {"status": "created", "webhook": str(webhook)}
+            else:
+                return {
+                    "status": "simulated",
+                    "watched": watched_repos,
+                    "domains": domains,
+                    "secret": "***"
+                }
+        except Exception as e:
+            return {"status": "error", "error": str(e)}
+
+    @staticmethod
+    def verify_webhook_signature(payload_bytes: bytes, signature_header: str, secret: str) -> bool:
+        """Verify the HMAC SHA-256 webhook payload signature."""
+        if not signature_header or not secret:
+            return False
+        expected = hmac.new(secret.encode("utf-8"), payload_bytes, hashlib.sha256).hexdigest()
+        return hmac.compare_digest(signature_header, expected)
+
+    @staticmethod
+    def parse_event_payload(payload_json: str | dict) -> dict:
+        """Parse incoming webhook event metadata."""
+        if isinstance(payload_json, str):
+            data = json.loads(payload_json)
+        else:
+            data = payload_json
+
+        event = data.get("event", {})
+        repo = data.get("repo", {})
+        return {
+            "action": event.get("action"),
+            "scope": event.get("scope"),
+            "repo_id": f"{repo.get('owner')}/{repo.get('name')}" if repo else None,
+            "repo_type": repo.get("type"),
+            "head_sha": repo.get("headSha")
+        }
+
+if __name__ == "__main__":
+    manager = WebhookAutomationManager()
+    print("Testing WebhookAutomationManager...")
+    sample_payload = {
+        "event": {"action": "update", "scope": "repo.content"},
+        "repo": {"owner": "bbkdevops", "name": "Fiber-MoE-Symplectic-Gating-Research", "type": "model", "headSha": "a1b2c3d4"}
+    }
+    parsed = manager.parse_event_payload(sample_payload)
+    print("Parsed Event:", json.dumps(parsed, indent=2))
+    print("Webhook manager ready!")