webhook_automation_manager.py

"""
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!")