mirror of
https://github.com/marcogll/soul23_placeholder_site_server.git
synced 2026-01-13 13:25:18 +00:00
188 lines
6.8 KiB
Python
Executable File
188 lines
6.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import requests
|
|
import json
|
|
from datetime import datetime, timezone
|
|
import os
|
|
import subprocess # <-- CAMBIO: Importar para ejecutar comandos del sistema
|
|
|
|
# --- Variables y Configuración ---
|
|
|
|
webhook_env_var = os.getenv('WEBHOOK_URLS', '')
|
|
WEBHOOK_URLS = [url.strip() for url in webhook_env_var.split(',') if url.strip()]
|
|
|
|
INTERNOS = {
|
|
"vps_soul23": "31.97.41.188", # <-- CAMBIO: Añadido el VPS para el ping
|
|
"coolify": "https://aperture.soul23.cloud",
|
|
"formbricks": "https://feedback.soul23.cloud",
|
|
"vikunja": "https://tasks.soul23.cloud",
|
|
"gokapi": "https://wetrans.soul23.cloud",
|
|
"n8n_flows": "https://flows.soul23.cloud",
|
|
"ap_pos": "https://apos.soul23.cloud"
|
|
}
|
|
|
|
EXTERNOS = {
|
|
"openai": "https://status.openai.com/",
|
|
"google_gemini": "https://aistudio.google.com/status",
|
|
"canva": "https://www.canvastatus.com/",
|
|
"tiktok": "https://www.tiktok.com",
|
|
"facebook": "https://www.facebook.com",
|
|
"instagram": "https://www.instagram.com",
|
|
"telegram": "https://core.telegram.org",
|
|
"whatsapp": "https://www.whatsapp.com",
|
|
"x": "https://www.x.com",
|
|
"youtube": "https://www.youtube.com"
|
|
}
|
|
|
|
STATUSPAGE_SERVICES = ["openai", "canva"]
|
|
|
|
# --- Funciones de Verificación de Estado ---
|
|
|
|
def check_url(url):
|
|
try:
|
|
r = requests.get(url, timeout=8)
|
|
return r.status_code
|
|
except:
|
|
return 0
|
|
|
|
# <-- CAMBIO: Nueva función para hacer ping al VPS
|
|
def check_vps_ping(ip_address):
|
|
"""
|
|
Envía un único paquete ICMP (ping) a la IP especificada.
|
|
Devuelve un estado basado en si el host respondió.
|
|
"""
|
|
try:
|
|
# Comando de ping para Linux (que es lo que usan los runners de GitHub):
|
|
# -c 1: Enviar solo 1 paquete.
|
|
# -W 2: Esperar un máximo de 2 segundos por una respuesta.
|
|
command = ["ping", "-c", "1", "-W", "2", ip_address]
|
|
|
|
# Ejecuta el comando, ocultando la salida para mantener los logs limpios.
|
|
result = subprocess.run(command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
|
|
# Un código de retorno 0 significa que el ping fue exitoso.
|
|
if result.returncode == 0:
|
|
return f"🟢 OK (VPS Reachable)"
|
|
else:
|
|
return f"🔴 Caído (VPS Unreachable)"
|
|
except Exception:
|
|
# Esto podría pasar si el comando 'ping' no estuviera disponible.
|
|
return f"🔴 Error (Ping command failed)"
|
|
|
|
def check_formbricks_health(base_url):
|
|
health_url = f"{base_url.rstrip('/')}/health"
|
|
try:
|
|
response = requests.get(health_url, timeout=8)
|
|
if response.status_code == 200:
|
|
try:
|
|
data = response.json()
|
|
if data.get("status") == "ok":
|
|
return f"🟢 OK (API Health: ok)"
|
|
else:
|
|
return f"🟡 Advertencia (API Health: {data.get('status', 'unknown')})"
|
|
except json.JSONDecodeError:
|
|
return "🟡 Advertencia (Respuesta no es JSON válido)"
|
|
else:
|
|
return f"🔴 Caído (Health Endpoint: {response.status_code})"
|
|
except requests.RequestException as e:
|
|
return f"🔴 Caído (Error de red: {e})"
|
|
|
|
|
|
def get_statuspage_status(base_url):
|
|
api_url = f"{base_url.rstrip('/')}/api/v2/summary.json"
|
|
try:
|
|
response = requests.get(api_url, timeout=8)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
status_description = data.get('status', {}).get('description')
|
|
if status_description:
|
|
status_indicator = data.get('status', {}).get('indicator')
|
|
if status_indicator == 'none':
|
|
return f"🟢 OK ({status_description})"
|
|
else:
|
|
return f"🟡 Advertencia ({status_description})"
|
|
return "🟡 Advertencia (Respuesta JSON inesperada)"
|
|
else:
|
|
return f"🔴 Caído (API Status: {response.status_code})"
|
|
except requests.RequestException as e:
|
|
return f"🔴 Caído (Error de red: {e})"
|
|
except json.JSONDecodeError:
|
|
return f"🔴 Caído (No se pudo decodificar JSON)"
|
|
|
|
|
|
def get_gemini_status(status_url):
|
|
incidents_url = "https://status.cloud.google.com/incidents.json"
|
|
try:
|
|
response = requests.get(incidents_url, timeout=8)
|
|
if response.status_code == 200:
|
|
incidents = response.json()
|
|
gemini_incident = any('Vertex AI' in i.get('service_name', '') and i.get('end') is None for i in incidents)
|
|
if not gemini_incident:
|
|
return "🟢 OK (Sin incidentes reportados para Vertex AI)"
|
|
else:
|
|
return "🟡 Advertencia (Incidente activo en Vertex AI)"
|
|
else:
|
|
return f"🔴 Caído ({response.status_code})"
|
|
except requests.RequestException as e:
|
|
return f"🔴 Caído (Error: {e})"
|
|
|
|
|
|
def human_state(code):
|
|
if code == 200:
|
|
return f"🟢 OK ({code})"
|
|
if code in (301, 302, 307, 308, 401, 403, 404):
|
|
return f"🟡 Advertencia ({code})"
|
|
return f"🔴 Caído ({code})"
|
|
|
|
# --- Lógica Principal ---
|
|
|
|
def build_section(diccionario):
|
|
salida = {}
|
|
for nombre, url_or_ip in diccionario.items():
|
|
# <-- CAMBIO: Añadida la lógica para el ping del VPS
|
|
if nombre == 'vps_soul23':
|
|
status_message = check_vps_ping(url_or_ip)
|
|
salida[f"{nombre}_url"] = url_or_ip # Guardamos la IP
|
|
elif nombre in STATUSPAGE_SERVICES:
|
|
status_message = get_statuspage_status(url_or_ip)
|
|
elif nombre == 'google_gemini':
|
|
status_message = get_gemini_status(url_or_ip)
|
|
elif nombre == 'formbricks':
|
|
status_message = check_formbricks_health(url_or_ip)
|
|
else:
|
|
status = check_url(url_or_ip)
|
|
salida[f"{nombre}_url"] = url_or_ip
|
|
salida[f"{nombre}_status"] = status
|
|
salida[f"{nombre}_state"] = human_state(status)
|
|
continue
|
|
|
|
# Asignación para los casos especiales
|
|
salida[f"{nombre}_url"] = url_or_ip
|
|
salida[f"{nombre}_status"] = status_message
|
|
salida[f"{nombre}_state"] = status_message
|
|
return salida
|
|
|
|
|
|
def main():
|
|
if not WEBHOOK_URLS:
|
|
print("⚠️ Advertencia: La variable de entorno WEBHOOK_URLS no está configurada o está vacía.")
|
|
|
|
resultado = {
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"internos": build_section(INTERNOS),
|
|
"externos": build_section(EXTERNOS)
|
|
}
|
|
|
|
print("\n--- Enviando a Webhooks ---")
|
|
for url in WEBHOOK_URLS:
|
|
try:
|
|
requests.post(url, json=resultado, timeout=10)
|
|
print(f"✅ Resultado enviado exitosamente a: {url}")
|
|
except requests.RequestException as e:
|
|
print(f"❌ Error al enviar al webhook {url}: {e}")
|
|
|
|
print("\n--- Payload Enviado ---")
|
|
print(json.dumps(resultado, indent=4))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |