| | 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"} |