Spaces:
Running
Running
| import requests | |
| import json | |
| import os | |
| import torch | |
| import requests | |
| import json | |
| from datetime import datetime, timezone | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| WEBHOOK_URL = os.getenv("WEBHOOK_URL") | |
| WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET") | |
| def get_best_torch_device(): | |
| if torch.cuda.is_available(): | |
| return torch.device("cuda") | |
| elif getattr(torch.backends, "mps", None) and torch.backends.mps.is_available(): | |
| return torch.device("mps") | |
| else: | |
| return torch.device("cpu") | |
| device = get_best_torch_device() | |
| def send_post(payload: dict): | |
| """ | |
| Send a post request to the webhook. | |
| """ | |
| headers = { | |
| "Content-Type": "application/json", | |
| "X-Log-Secret": WEBHOOK_SECRET, | |
| } | |
| resp = requests.post(WEBHOOK_URL, headers=headers, data=json.dumps(payload)) | |
| if resp.status_code == 200: | |
| return True | |
| else: | |
| print(f"Posting to webhook failed: {resp.status_code} {resp.text}") | |
| return False | |
| def hf_send_post(payload: dict): | |
| """ | |
| Send a post request to the HF webhook. | |
| """ | |
| payload["service"] = "hf-ai4data-mcp-server" | |
| payload["level"] = "INFO" | |
| payload["timestamp"] = datetime.now(timezone.utc).isoformat() | |
| return send_post(payload) | |
| # Example usage | |
| if __name__ == "__main__": | |
| from datetime import datetime | |
| post_entry = { | |
| "service": "hf-ai4data-mcp-api-test", | |
| "level": "INFO", | |
| "message": "Test message " + datetime.now().isoformat(), | |
| } | |
| send_post(post_entry) | |