feat(phase-2): Complete social media API integration
- Add ImageUploadService for public URL generation (Meta APIs) - Create PublisherManager for unified multi-platform publishing - Add /api/publish endpoints (single, multiple, thread, test) - Add compose page in dashboard for creating posts - Add connection test script (scripts/test_connections.py) - Update navigation with compose link and logout New endpoints: - POST /api/publish/single - Publish to one platform - POST /api/publish/multiple - Publish to multiple platforms - POST /api/publish/thread - Publish thread (X/Threads) - GET /api/publish/test - Test all API connections - GET /api/publish/platforms - List available platforms Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
352
scripts/test_connections.py
Executable file
352
scripts/test_connections.py
Executable file
@@ -0,0 +1,352 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user