Spaces:
Running
Running
File size: 1,526 Bytes
8248385 3cb103a 8248385 3cb103a 8248385 34b2550 8248385 |
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 |
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)
|