Spaces:
Running
Running
File size: 2,442 Bytes
28b2d15 6cd757a 28b2d15 6cd757a 28b2d15 a4ed0ef 28b2d15 6f66537 28b2d15 363ac35 28b2d15 dba72fc 6cd757a dba72fc 4cfb17a dba72fc 6cd757a dba72fc f644dfd 6cd757a 5e8c10b 6f66537 6cd757a a4ed0ef 6cd757a a4ed0ef 6cd757a dd21b8c 6cd757a abbb598 6cd757a 4cfb17a 5e8c10b |
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 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 |
import os
from dotenv import load_dotenv
import time
import base64
import requests
import threading
import gradio as gr
from fastapi.responses import JSONResponse, RedirectResponse
from fastapi import FastAPI, Request, Response, status
from src.utils import (
manage_urls,
wake_up_space,
get_wake_up_action_name,
get_url_list,
)
# Charger les variables d'environnement
load_dotenv()
app = FastAPI()
USERNAME = os.getenv("USERNAME")
PASSWORD = os.getenv("PASSWORD")
@app.middleware("http")
async def basic_auth(request: Request, call_next):
auth = request.headers.get("Authorization")
expected = "Basic " + base64.b64encode(f"{USERNAME}:{PASSWORD}".encode()).decode()
if USERNAME and PASSWORD and auth != expected:
return Response(
status_code=status.HTTP_401_UNAUTHORIZED,
headers={"WWW-Authenticate": "Basic"},
content="Unauthorized",
)
return await call_next(request)
# Interface Gradio
interface = gr.Interface(
fn=manage_urls,
inputs=[
gr.Radio(choices=["Ajouter", "Supprimer"], label="Action"),
gr.Textbox(label="URL"),
],
outputs=[
gr.Textbox(label="Message"),
gr.Textbox(label="Liste actuelle des URLs", lines=10, interactive=False),
],
title="🔐 Gestion des URLs à pinger",
description="Ajoutez ou supprimez dynamiquement des URLs à surveiller. La liste actuelle est toujours affichée.",
flagging_dir="/tmp/flags"
)
@app.get("/")
def root_redirect():
return RedirectResponse(url="/gradio")
@app.get("/check_health")
def check_health():
return JSONResponse(content={"status": "ok", "nb_urls": len(get_url_list())})
# Background thread pour pinger les URLs toutes les heures
def ping_loop():
while True:
for url in get_url_list():
try:
if get_wake_up_action_name(url):
print(f"[wake-up] 🚀 {url}")
wake_up_space(url)
else:
requests.get(url, timeout=5)
print(f"[ping] ✅ {url}")
except Exception as e:
print(f"[ping] ❌ {url} | {e}")
# Cooldown de 1 seconde entre les pings
time.sleep(1)
time.sleep(3600) # 1h
threading.Thread(target=ping_loop, daemon=True).start()
# Monter Gradio sur /gradio
gr.mount_gradio_app(app, interface, path="/gradio")
|