|
from fastapi import FastAPI, Request |
|
import requests |
|
from dotenv import load_dotenv |
|
import os |
|
|
|
load_dotenv() |
|
app = FastAPI() |
|
|
|
|
|
HF_TOKEN = os.getenv("HF_TOKEN") |
|
SPACES = ["OrganizedProgrammers/DocFinder", "OrganizedProgrammers/SpecSplitter"] |
|
|
|
def restart_space(space_id): |
|
"""Redémarre un Space Hugging Face via l'API.""" |
|
api_url = f"https://api.huggingface.co/spaces/{space_id}/restart" |
|
headers = {"Authorization": f"Bearer {HF_TOKEN}"} |
|
response = requests.post(api_url, headers=headers, verify=False) |
|
|
|
if response.status_code == 200: |
|
print(f"Space redémarré avec succès: {space_id}") |
|
else: |
|
print(f"Échec du redémarrage: {space_id} - {response.text}") |
|
|
|
@app.post("/webhook") |
|
async def handle_webhook(request: Request): |
|
payload = await request.json() |
|
|
|
|
|
if (payload.get("repo", {}).get("type") == "dataset" and |
|
payload.get("event", {}).get("action") == "update"): |
|
|
|
print("Dataset mis à jour, redémarrage des Spaces...") |
|
for space_id in SPACES: |
|
restart_space(space_id) |
|
|
|
return {"message": "Spaces mis à jour avec succès"} |
|
|
|
return {"message": "Aucune action nécessaire"} |