#!/usr/bin/env python3 """ Script para probar conexiones con las APIs de redes sociales. Verifica que las credenciales estén correctamente configuradas. Uso: python scripts/test_connections.py # Probar todas python scripts/test_connections.py x # Solo X/Twitter python scripts/test_connections.py threads # Solo Threads """ import sys import asyncio from pathlib import Path # Agregar directorio raíz al path sys.path.insert(0, str(Path(__file__).parent.parent)) from app.core.config import settings def print_header(text: str): """Imprimir encabezado formateado.""" print() print("=" * 60) print(f" {text}") print("=" * 60) def print_status(label: str, value: str, ok: bool = True): """Imprimir estado con color.""" status = "✅" if ok else "❌" print(f" {status} {label}: {value}") def print_config(label: str, value: str, is_secret: bool = False): """Imprimir configuración.""" if is_secret and value: display = value[:8] + "..." + value[-4:] if len(value) > 12 else "****" else: display = value or "(no configurado)" print(f" {label}: {display}") async def test_x(): """Probar conexión con X (Twitter).""" print_header("X (Twitter)") # Verificar configuración print("\n Configuración:") print_config("API Key", settings.X_API_KEY, is_secret=True) print_config("API Secret", settings.X_API_SECRET, is_secret=True) print_config("Access Token", settings.X_ACCESS_TOKEN, is_secret=True) print_config("Access Secret", settings.X_ACCESS_TOKEN_SECRET, is_secret=True) print_config("Bearer Token", settings.X_BEARER_TOKEN, is_secret=True) if not all([ settings.X_API_KEY, settings.X_API_SECRET, settings.X_ACCESS_TOKEN, settings.X_ACCESS_TOKEN_SECRET ]): print_status("Estado", "Credenciales incompletas", ok=False) return False print("\n Probando conexión...") try: import tweepy client = tweepy.Client( consumer_key=settings.X_API_KEY, consumer_secret=settings.X_API_SECRET, access_token=settings.X_ACCESS_TOKEN, access_token_secret=settings.X_ACCESS_TOKEN_SECRET, bearer_token=settings.X_BEARER_TOKEN ) me = client.get_me(user_fields=["public_metrics"]) if me.data: print_status("Conexión", "OK") print(f"\n Cuenta verificada:") print(f" Usuario: @{me.data.username}") print(f" Nombre: {me.data.name}") print(f" ID: {me.data.id}") if me.data.public_metrics: metrics = me.data.public_metrics print(f" Seguidores: {metrics.get('followers_count', 0):,}") print(f" Siguiendo: {metrics.get('following_count', 0):,}") print(f" Tweets: {metrics.get('tweet_count', 0):,}") return True else: print_status("Conexión", "Sin datos de usuario", ok=False) return False except Exception as e: print_status("Error", str(e), ok=False) return False async def test_threads(): """Probar conexión con Threads.""" print_header("Threads") print("\n Configuración:") print_config("Access Token", settings.META_ACCESS_TOKEN, is_secret=True) print_config("User ID", settings.THREADS_USER_ID) if not settings.META_ACCESS_TOKEN or not settings.THREADS_USER_ID: print_status("Estado", "Credenciales incompletas", ok=False) return False print("\n Probando conexión...") try: import httpx async with httpx.AsyncClient() as client: url = f"https://graph.threads.net/v1.0/{settings.THREADS_USER_ID}" params = { "fields": "id,username,threads_profile_picture_url", "access_token": settings.META_ACCESS_TOKEN } response = await client.get(url, params=params) if response.status_code == 200: data = response.json() print_status("Conexión", "OK") print(f"\n Cuenta verificada:") print(f" Usuario: @{data.get('username', 'N/A')}") print(f" ID: {data.get('id', 'N/A')}") return True else: error = response.json().get("error", {}) print_status("Error", error.get("message", response.text), ok=False) return False except Exception as e: print_status("Error", str(e), ok=False) return False async def test_facebook(): """Probar conexión con Facebook Page.""" print_header("Facebook Page") print("\n Configuración:") print_config("Access Token", settings.META_ACCESS_TOKEN, is_secret=True) print_config("Page ID", settings.FACEBOOK_PAGE_ID) if not settings.META_ACCESS_TOKEN or not settings.FACEBOOK_PAGE_ID: print_status("Estado", "Credenciales incompletas", ok=False) return False print("\n Probando conexión...") try: import httpx async with httpx.AsyncClient() as client: url = f"https://graph.facebook.com/v18.0/{settings.FACEBOOK_PAGE_ID}" params = { "fields": "id,name,followers_count,fan_count,link", "access_token": settings.META_ACCESS_TOKEN } response = await client.get(url, params=params) if response.status_code == 200: data = response.json() print_status("Conexión", "OK") print(f"\n Página verificada:") print(f" Nombre: {data.get('name', 'N/A')}") print(f" ID: {data.get('id', 'N/A')}") print(f" Seguidores: {data.get('followers_count', 0):,}") print(f" Fans: {data.get('fan_count', 0):,}") if data.get("link"): print(f" URL: {data['link']}") return True else: error = response.json().get("error", {}) print_status("Error", error.get("message", response.text), ok=False) return False except Exception as e: print_status("Error", str(e), ok=False) return False async def test_instagram(): """Probar conexión con Instagram Business.""" print_header("Instagram Business") print("\n Configuración:") print_config("Access Token", settings.META_ACCESS_TOKEN, is_secret=True) print_config("Account ID", settings.INSTAGRAM_ACCOUNT_ID) if not settings.META_ACCESS_TOKEN or not settings.INSTAGRAM_ACCOUNT_ID: print_status("Estado", "Credenciales incompletas", ok=False) return False print("\n Probando conexión...") try: import httpx async with httpx.AsyncClient() as client: url = f"https://graph.facebook.com/v18.0/{settings.INSTAGRAM_ACCOUNT_ID}" params = { "fields": "id,username,name,followers_count,follows_count,media_count", "access_token": settings.META_ACCESS_TOKEN } response = await client.get(url, params=params) if response.status_code == 200: data = response.json() print_status("Conexión", "OK") print(f"\n Cuenta verificada:") print(f" Usuario: @{data.get('username', 'N/A')}") print(f" Nombre: {data.get('name', 'N/A')}") print(f" ID: {data.get('id', 'N/A')}") print(f" Seguidores: {data.get('followers_count', 0):,}") print(f" Siguiendo: {data.get('follows_count', 0):,}") print(f" Publicaciones: {data.get('media_count', 0):,}") return True else: error = response.json().get("error", {}) print_status("Error", error.get("message", response.text), ok=False) return False except Exception as e: print_status("Error", str(e), ok=False) return False async def test_deepseek(): """Probar conexión con DeepSeek API.""" print_header("DeepSeek API") print("\n Configuración:") print_config("API Key", settings.DEEPSEEK_API_KEY, is_secret=True) print_config("Base URL", settings.DEEPSEEK_BASE_URL) if not settings.DEEPSEEK_API_KEY: print_status("Estado", "API Key no configurada", ok=False) return False print("\n Probando conexión...") try: from openai import OpenAI client = OpenAI( api_key=settings.DEEPSEEK_API_KEY, base_url=settings.DEEPSEEK_BASE_URL ) # Hacer una llamada mínima para verificar response = client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": "Responde solo: OK"}], max_tokens=10 ) if response.choices: print_status("Conexión", "OK") print(f"\n API verificada:") print(f" Modelo: {response.model}") print(f" Tokens usados: {response.usage.total_tokens}") return True else: print_status("Error", "Sin respuesta", ok=False) return False except Exception as e: print_status("Error", str(e), ok=False) return False async def main(): """Ejecutar pruebas de conexión.""" print() print("╔════════════════════════════════════════════════════════════╗") print("║ TEST DE CONEXIONES - Social Media Automation ║") print("╚════════════════════════════════════════════════════════════╝") # Determinar qué probar platforms = sys.argv[1:] if len(sys.argv) > 1 else ["all"] tests = { "x": test_x, "twitter": test_x, "threads": test_threads, "facebook": test_facebook, "fb": test_facebook, "instagram": test_instagram, "ig": test_instagram, "deepseek": test_deepseek, "ai": test_deepseek, } results = {} if "all" in platforms: # Probar todas las plataformas for name, test_func in [ ("X", test_x), ("Threads", test_threads), ("Facebook", test_facebook), ("Instagram", test_instagram), ("DeepSeek", test_deepseek), ]: results[name] = await test_func() else: # Probar solo las especificadas for platform in platforms: test_func = tests.get(platform.lower()) if test_func: results[platform] = await test_func() else: print(f"\n⚠️ Plataforma desconocida: {platform}") print(f" Opciones: x, threads, facebook, instagram, deepseek") # Resumen print_header("RESUMEN") total = len(results) passed = sum(1 for r in results.values() if r) for name, success in results.items(): status = "✅ OK" if success else "❌ FALLO" print(f" {name}: {status}") print() print(f" Total: {passed}/{total} conexiones exitosas") if passed < total: print() print(" 💡 Revisa docs/API_KEYS_SETUP.md para configurar las credenciales") print() return passed == total if __name__ == "__main__": success = asyncio.run(main()) sys.exit(0 if success else 1)