File size: 3,331 Bytes
bf9f684 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | """
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!")
|