social-media-automation/scripts/test_connections.py

#!/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)