diff --git a/app/api/routes/dashboard.py b/app/api/routes/dashboard.py index 274d95f..0e5d8ad 100644 --- a/app/api/routes/dashboard.py +++ b/app/api/routes/dashboard.py @@ -83,6 +83,19 @@ async def dashboard_home(request: Request, db: Session = Depends(get_db)): }) +@router.get("/compose", response_class=HTMLResponse) +async def dashboard_compose(request: Request, db: Session = Depends(get_db)): + """Página para crear nuevas publicaciones.""" + user = require_auth(request, db) + if not user: + return RedirectResponse(url="/login", status_code=302) + + return templates.TemplateResponse("compose.html", { + "request": request, + "user": user.to_dict() + }) + + @router.get("/posts", response_class=HTMLResponse) async def dashboard_posts(request: Request, db: Session = Depends(get_db)): """Página de gestión de posts.""" diff --git a/app/api/routes/publish.py b/app/api/routes/publish.py new file mode 100644 index 0000000..55468eb --- /dev/null +++ b/app/api/routes/publish.py @@ -0,0 +1,262 @@ +""" +API endpoints para publicación en redes sociales. +""" + +from typing import Optional, List +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel + +from app.publishers.manager import publisher_manager, Platform, MultiPublishResult + + +router = APIRouter() + + +# ============================================================ +# Schemas +# ============================================================ + +class PublishRequest(BaseModel): + """Solicitud de publicación simple.""" + platform: str + content: str + image_url: Optional[str] = None + + +class MultiPublishRequest(BaseModel): + """Solicitud de publicación múltiple.""" + platforms: List[str] + content: dict # {platform: content} o string único + image_url: Optional[str] = None + + +class ThreadPublishRequest(BaseModel): + """Solicitud de publicación de hilo.""" + platform: str + posts: List[str] + images: Optional[List[str]] = None + + +class PublishResponse(BaseModel): + """Respuesta de publicación.""" + success: bool + post_id: Optional[str] = None + url: Optional[str] = None + error: Optional[str] = None + + +class ConnectionTestResponse(BaseModel): + """Respuesta de test de conexión.""" + platform: str + configured: bool + connected: bool + error: Optional[str] = None + details: dict = {} + + +# ============================================================ +# Endpoints +# ============================================================ + +@router.post("/single", response_model=PublishResponse) +async def publish_single(request: PublishRequest): + """ + Publicar contenido en una plataforma específica. + + - **platform**: x, threads, facebook, instagram + - **content**: Texto del post + - **image_url**: URL o ruta de imagen (opcional) + """ + try: + platform = Platform(request.platform.lower()) + except ValueError: + raise HTTPException( + status_code=400, + detail=f"Plataforma no válida: {request.platform}. " + f"Opciones: x, threads, facebook, instagram" + ) + + result = await publisher_manager.publish( + platform=platform, + content=request.content, + image_path=request.image_url + ) + + return PublishResponse( + success=result.success, + post_id=result.post_id, + url=result.url, + error=result.error_message + ) + + +@router.post("/multiple") +async def publish_multiple(request: MultiPublishRequest): + """ + Publicar contenido en múltiples plataformas. + + - **platforms**: Lista de plataformas ["x", "threads", "facebook"] + - **content**: Dict con contenido por plataforma o string único + - **image_url**: URL o ruta de imagen (opcional) + """ + platforms = [] + for p in request.platforms: + try: + platforms.append(Platform(p.lower())) + except ValueError: + raise HTTPException( + status_code=400, + detail=f"Plataforma no válida: {p}" + ) + + result = await publisher_manager.publish_to_multiple( + platforms=platforms, + content=request.content, + image_path=request.image_url + ) + + return { + "success": result.success, + "successful_platforms": result.successful_platforms, + "failed_platforms": result.failed_platforms, + "results": { + p: { + "success": r.success, + "post_id": r.post_id, + "url": r.url, + "error": r.error_message + } + for p, r in result.results.items() + }, + "errors": result.errors + } + + +@router.post("/thread", response_model=PublishResponse) +async def publish_thread(request: ThreadPublishRequest): + """ + Publicar un hilo de posts en una plataforma. + + - **platform**: x o threads + - **posts**: Lista de textos para cada post del hilo + - **images**: Lista de URLs de imágenes (opcional) + """ + try: + platform = Platform(request.platform.lower()) + except ValueError: + raise HTTPException( + status_code=400, + detail=f"Plataforma no válida: {request.platform}" + ) + + if platform not in [Platform.X, Platform.THREADS]: + raise HTTPException( + status_code=400, + detail="Los hilos solo están soportados en X y Threads" + ) + + result = await publisher_manager.publish_thread( + platform=platform, + posts=request.posts, + images=request.images + ) + + return PublishResponse( + success=result.success, + post_id=result.post_id, + url=result.url, + error=result.error_message + ) + + +@router.get("/test/{platform}", response_model=ConnectionTestResponse) +async def test_connection(platform: str): + """ + Probar conexión con una plataforma específica. + + Verifica que las credenciales estén configuradas y funcionando. + """ + try: + plat = Platform(platform.lower()) + except ValueError: + raise HTTPException( + status_code=400, + detail=f"Plataforma no válida: {platform}" + ) + + result = await publisher_manager.test_connection(plat) + + return ConnectionTestResponse(**result) + + +@router.get("/test") +async def test_all_connections(): + """ + Probar conexión con todas las plataformas. + + Devuelve el estado de cada plataforma configurada. + """ + results = await publisher_manager.test_all_connections() + + return { + "platforms": results, + "summary": { + "total": len(results), + "configured": sum(1 for r in results.values() if r["configured"]), + "connected": sum(1 for r in results.values() if r["connected"]) + } + } + + +@router.get("/platforms") +async def get_available_platforms(): + """ + Obtener lista de plataformas disponibles. + + Solo devuelve plataformas con credenciales configuradas. + """ + return { + "available": publisher_manager.get_available_platforms(), + "all": [p.value for p in Platform] + } + + +class PreviewRequest(BaseModel): + """Solicitud de previsualización.""" + content: str + platforms: List[str] + + +@router.post("/preview") +async def preview_content(request: PreviewRequest): + """ + Previsualizar cómo se verá el contenido en cada plataforma. + + Valida longitud y muestra advertencias. + """ + previews = {} + + char_limits = { + "x": 280, + "threads": 500, + "instagram": 2200, + "facebook": 63206 + } + + for p in request.platforms: + limit = char_limits.get(p.lower(), 500) + length = len(request.content) + + previews[p] = { + "content": request.content[:limit], + "length": length, + "limit": limit, + "valid": length <= limit, + "truncated": length > limit, + "remaining": max(0, limit - length) + } + + return { + "original_length": len(request.content), + "previews": previews + } diff --git a/app/main.py b/app/main.py index 9958f32..4b055b4 100644 --- a/app/main.py +++ b/app/main.py @@ -11,7 +11,7 @@ from fastapi import FastAPI from fastapi.staticfiles import StaticFiles from fastapi.middleware.cors import CORSMiddleware -from app.api.routes import posts, products, services, calendar, dashboard, interactions, auth +from app.api.routes import posts, products, services, calendar, dashboard, interactions, auth, publish from app.core.config import settings from app.core.database import engine from app.models import Base @@ -49,6 +49,11 @@ app.add_middleware( # Montar archivos estáticos app.mount("/static", StaticFiles(directory="dashboard/static"), name="static") +# Montar directorio de uploads para imágenes públicas +import os +os.makedirs("uploads/images", exist_ok=True) +app.mount("/uploads", StaticFiles(directory="uploads"), name="uploads") + # Registrar rutas app.include_router(auth.router, prefix="", tags=["Auth"]) app.include_router(dashboard.router, prefix="", tags=["Dashboard"]) @@ -57,6 +62,7 @@ app.include_router(products.router, prefix="/api/products", tags=["Products"]) app.include_router(services.router, prefix="/api/services", tags=["Services"]) app.include_router(calendar.router, prefix="/api/calendar", tags=["Calendar"]) app.include_router(interactions.router, prefix="/api/interactions", tags=["Interactions"]) +app.include_router(publish.router, prefix="/api/publish", tags=["Publish"]) @app.get("/api/health") diff --git a/app/publishers/__init__.py b/app/publishers/__init__.py index 8fa5984..7c87669 100644 --- a/app/publishers/__init__.py +++ b/app/publishers/__init__.py @@ -7,6 +7,7 @@ from app.publishers.x_publisher import XPublisher from app.publishers.threads_publisher import ThreadsPublisher from app.publishers.instagram_publisher import InstagramPublisher from app.publishers.facebook_publisher import FacebookPublisher +from app.publishers.manager import PublisherManager, Platform, publisher_manager def get_publisher(platform: str) -> BasePublisher: @@ -31,5 +32,8 @@ __all__ = [ "ThreadsPublisher", "InstagramPublisher", "FacebookPublisher", + "PublisherManager", + "Platform", + "publisher_manager", "get_publisher" ] diff --git a/app/publishers/manager.py b/app/publishers/manager.py new file mode 100644 index 0000000..7098edf --- /dev/null +++ b/app/publishers/manager.py @@ -0,0 +1,343 @@ +""" +PublisherManager - Gestor unificado de publicaciones. +Coordina todos los publishers y proporciona una interfaz única. +""" + +import asyncio +from typing import Optional, List, Dict, Any +from dataclasses import dataclass +from enum import Enum + +from app.publishers.base import PublishResult +from app.publishers.x_publisher import XPublisher +from app.publishers.threads_publisher import ThreadsPublisher +from app.publishers.facebook_publisher import FacebookPublisher +from app.publishers.instagram_publisher import InstagramPublisher +from app.services.image_upload import image_upload + + +class Platform(str, Enum): + """Plataformas soportadas.""" + X = "x" + THREADS = "threads" + FACEBOOK = "facebook" + INSTAGRAM = "instagram" + + +@dataclass +class MultiPublishResult: + """Resultado de publicación en múltiples plataformas.""" + success: bool + results: Dict[str, PublishResult] + errors: List[str] + + @property + def successful_platforms(self) -> List[str]: + """Plataformas donde se publicó exitosamente.""" + return [p for p, r in self.results.items() if r.success] + + @property + def failed_platforms(self) -> List[str]: + """Plataformas donde falló la publicación.""" + return [p for p, r in self.results.items() if not r.success] + + +class PublisherManager: + """ + Gestor unificado de publicaciones. + + Proporciona una interfaz única para publicar contenido + en todas las plataformas soportadas. + """ + + def __init__(self): + self._publishers = {} + self._init_publishers() + + def _init_publishers(self): + """Inicializar publishers disponibles.""" + self._publishers = { + Platform.X: XPublisher(), + Platform.THREADS: ThreadsPublisher(), + Platform.FACEBOOK: FacebookPublisher(), + Platform.INSTAGRAM: InstagramPublisher(), + } + + def get_publisher(self, platform: Platform): + """Obtener publisher para una plataforma.""" + return self._publishers.get(platform) + + def get_available_platforms(self) -> List[str]: + """Obtener lista de plataformas con credenciales configuradas.""" + available = [] + + for platform, publisher in self._publishers.items(): + if self._is_configured(platform): + available.append(platform.value) + + return available + + def _is_configured(self, platform: Platform) -> bool: + """Verificar si una plataforma tiene credenciales configuradas.""" + publisher = self._publishers.get(platform) + + if not publisher: + return False + + if platform == Platform.X: + return publisher.client is not None + elif platform == Platform.THREADS: + return bool(publisher.access_token and publisher.user_id) + elif platform == Platform.FACEBOOK: + return bool(publisher.access_token and publisher.page_id) + elif platform == Platform.INSTAGRAM: + return bool(publisher.access_token and publisher.account_id) + + return False + + async def publish( + self, + platform: Platform, + content: str, + image_path: Optional[str] = None + ) -> PublishResult: + """ + Publicar en una plataforma específica. + + Args: + platform: Plataforma destino + content: Texto del post + image_path: Ruta local a imagen (opcional) + + Returns: + PublishResult con el resultado + """ + publisher = self._publishers.get(platform) + + if not publisher: + return PublishResult( + success=False, + error_message=f"Plataforma no soportada: {platform}" + ) + + # Validar contenido + if not publisher.validate_content(content): + return PublishResult( + success=False, + error_message=f"Contenido excede límite de {publisher.char_limit} caracteres" + ) + + # Subir imagen si es necesario para Meta APIs + public_image_url = None + if image_path and platform in [Platform.THREADS, Platform.FACEBOOK, Platform.INSTAGRAM]: + try: + public_image_url = await image_upload.upload_from_path(image_path) + except Exception as e: + return PublishResult( + success=False, + error_message=f"Error subiendo imagen: {e}" + ) + + # Publicar + try: + if platform == Platform.X: + return await publisher.publish(content, image_path) + else: + return await publisher.publish(content, public_image_url or image_path) + except Exception as e: + return PublishResult( + success=False, + error_message=f"Error al publicar: {e}" + ) + + async def publish_to_multiple( + self, + platforms: List[Platform], + content: Dict[str, str], + image_path: Optional[str] = None, + parallel: bool = True + ) -> MultiPublishResult: + """ + Publicar en múltiples plataformas. + + Args: + platforms: Lista de plataformas + content: Dict con contenido por plataforma {platform: texto} + o string único para todas + image_path: Ruta a imagen (opcional) + parallel: Si True, publica en paralelo + + Returns: + MultiPublishResult con resultados por plataforma + """ + results = {} + errors = [] + + # Normalizar contenido + if isinstance(content, str): + content = {p.value: content for p in platforms} + + # Subir imagen una sola vez si es necesario + public_image_url = None + if image_path: + try: + public_image_url = await image_upload.upload_from_path(image_path) + except Exception as e: + errors.append(f"Error subiendo imagen: {e}") + + async def publish_to_platform(platform: Platform): + platform_content = content.get(platform.value, "") + + if not platform_content: + return platform, PublishResult( + success=False, + error_message="Sin contenido para esta plataforma" + ) + + # X usa ruta local, Meta usa URL pública + img = image_path if platform == Platform.X else public_image_url + + result = await self.publish(platform, platform_content, img) + return platform, result + + if parallel: + # Publicar en paralelo + tasks = [publish_to_platform(p) for p in platforms] + platform_results = await asyncio.gather(*tasks, return_exceptions=True) + + for item in platform_results: + if isinstance(item, Exception): + errors.append(str(item)) + else: + platform, result = item + results[platform.value] = result + else: + # Publicar secuencialmente + for platform in platforms: + try: + _, result = await publish_to_platform(platform) + results[platform.value] = result + except Exception as e: + errors.append(f"{platform.value}: {e}") + results[platform.value] = PublishResult( + success=False, + error_message=str(e) + ) + + # Determinar éxito global + success = any(r.success for r in results.values()) + + return MultiPublishResult( + success=success, + results=results, + errors=errors + ) + + async def publish_thread( + self, + platform: Platform, + posts: List[str], + images: Optional[List[str]] = None + ) -> PublishResult: + """Publicar un hilo en una plataforma.""" + publisher = self._publishers.get(platform) + + if not publisher: + return PublishResult( + success=False, + error_message=f"Plataforma no soportada: {platform}" + ) + + return await publisher.publish_thread(posts, images) + + async def test_connection(self, platform: Platform) -> Dict[str, Any]: + """ + Probar conexión con una plataforma. + + Returns: + Dict con status y detalles + """ + result = { + "platform": platform.value, + "configured": self._is_configured(platform), + "connected": False, + "error": None, + "details": {} + } + + if not result["configured"]: + result["error"] = "Credenciales no configuradas" + return result + + publisher = self._publishers.get(platform) + + try: + if platform == Platform.X: + me = publisher.client.get_me() + if me.data: + result["connected"] = True + result["details"] = { + "username": me.data.username, + "name": me.data.name, + "id": str(me.data.id) + } + + elif platform == Platform.THREADS: + import httpx + async with httpx.AsyncClient() as client: + url = f"{publisher.base_url}/{publisher.user_id}" + params = { + "fields": "id,username", + "access_token": publisher.access_token + } + response = await client.get(url, params=params) + if response.status_code == 200: + data = response.json() + result["connected"] = True + result["details"] = data + + elif platform == Platform.FACEBOOK: + import httpx + async with httpx.AsyncClient() as client: + url = f"{publisher.base_url}/{publisher.page_id}" + params = { + "fields": "id,name,followers_count", + "access_token": publisher.access_token + } + response = await client.get(url, params=params) + if response.status_code == 200: + data = response.json() + result["connected"] = True + result["details"] = data + + elif platform == Platform.INSTAGRAM: + import httpx + async with httpx.AsyncClient() as client: + url = f"{publisher.base_url}/{publisher.account_id}" + params = { + "fields": "id,username,followers_count,media_count", + "access_token": publisher.access_token + } + response = await client.get(url, params=params) + if response.status_code == 200: + data = response.json() + result["connected"] = True + result["details"] = data + + except Exception as e: + result["error"] = str(e) + + return result + + async def test_all_connections(self) -> Dict[str, Dict[str, Any]]: + """Probar conexión con todas las plataformas.""" + results = {} + + for platform in Platform: + results[platform.value] = await self.test_connection(platform) + + return results + + +# Instancia global +publisher_manager = PublisherManager() diff --git a/app/services/image_upload.py b/app/services/image_upload.py new file mode 100644 index 0000000..6d6f270 --- /dev/null +++ b/app/services/image_upload.py @@ -0,0 +1,166 @@ +""" +Servicio de subida y gestión de imágenes. +Sube imágenes locales y devuelve URLs públicas para las APIs de Meta. +""" + +import os +import uuid +import shutil +from pathlib import Path +from typing import Optional +from datetime import datetime + +from app.core.config import settings + + +class ImageUploadService: + """Servicio para subir imágenes y obtener URLs públicas.""" + + def __init__(self): + self.upload_dir = Path("uploads/images") + self.upload_dir.mkdir(parents=True, exist_ok=True) + self._base_url: Optional[str] = None + + @property + def base_url(self) -> str: + """URL base para servir imágenes.""" + if self._base_url: + return self._base_url + + # En producción, usar dominio público + if settings.APP_ENV == "production": + return f"{settings.BUSINESS_WEBSITE}/uploads/images" + + # En desarrollo, usar localhost + return "http://localhost:8000/uploads/images" + + def set_base_url(self, url: str): + """Establecer URL base manualmente.""" + self._base_url = url.rstrip("/") + + def _generate_filename(self, original_name: str) -> str: + """Generar nombre único para archivo.""" + ext = Path(original_name).suffix.lower() or ".png" + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + unique_id = uuid.uuid4().hex[:8] + return f"{timestamp}_{unique_id}{ext}" + + async def upload_from_path(self, file_path: str) -> str: + """ + Subir imagen desde ruta local. + + Args: + file_path: Ruta al archivo de imagen local + + Returns: + URL pública de la imagen + """ + source = Path(file_path) + + if not source.exists(): + raise FileNotFoundError(f"Archivo no encontrado: {file_path}") + + # Generar nombre único + new_name = self._generate_filename(source.name) + dest = self.upload_dir / new_name + + # Copiar archivo + shutil.copy2(source, dest) + + return f"{self.base_url}/{new_name}" + + async def upload_from_bytes( + self, + data: bytes, + filename: str = "image.png" + ) -> str: + """ + Subir imagen desde bytes. + + Args: + data: Contenido del archivo en bytes + filename: Nombre original para extensión + + Returns: + URL pública de la imagen + """ + new_name = self._generate_filename(filename) + dest = self.upload_dir / new_name + + dest.write_bytes(data) + + return f"{self.base_url}/{new_name}" + + async def upload_generated_image(self, image_path: str) -> str: + """ + Subir imagen generada por ImageGenerator. + Alias de upload_from_path para claridad. + """ + return await self.upload_from_path(image_path) + + def get_local_path(self, url: str) -> Optional[Path]: + """Obtener ruta local desde URL.""" + if self.base_url not in url: + return None + + filename = url.split("/")[-1] + path = self.upload_dir / filename + + return path if path.exists() else None + + def delete(self, url: str) -> bool: + """Eliminar imagen por URL.""" + path = self.get_local_path(url) + + if path and path.exists(): + path.unlink() + return True + + return False + + def cleanup_old(self, days: int = 30) -> int: + """ + Limpiar imágenes antiguas. + + Args: + days: Eliminar imágenes más antiguas que estos días + + Returns: + Cantidad de archivos eliminados + """ + import time + + cutoff = time.time() - (days * 86400) + deleted = 0 + + for file in self.upload_dir.iterdir(): + if file.is_file() and file.stat().st_mtime < cutoff: + file.unlink() + deleted += 1 + + return deleted + + def list_images(self, limit: int = 100) -> list: + """Listar imágenes subidas.""" + images = [] + + for file in sorted( + self.upload_dir.iterdir(), + key=lambda f: f.stat().st_mtime, + reverse=True + )[:limit]: + if file.is_file(): + images.append({ + "filename": file.name, + "url": f"{self.base_url}/{file.name}", + "size": file.stat().st_size, + "modified": datetime.fromtimestamp( + file.stat().st_mtime + ).isoformat() + }) + + return images + + +# Instancia global +image_upload = ImageUploadService() diff --git a/dashboard/templates/compose.html b/dashboard/templates/compose.html new file mode 100644 index 0000000..5356da6 --- /dev/null +++ b/dashboard/templates/compose.html @@ -0,0 +1,443 @@ + + + + + + Crear Post - Social Media Automation + + + + + +
+
+

+ Consultoría AS - Social Media +

+ +
+
+ +
+

Crear Nueva Publicación

+ +
+ +
+

Plataformas

+
+ + + + +
+

Selecciona una o más plataformas

+
+ + +
+
+

Contenido

+
+ 0 caracteres +
+
+ + + +
+ +
+
+ + +
+

Asistente IA

+
+ + + +
+

+ Usa IA para generar o mejorar el contenido +

+
+ + +
+

Programación

+
+ + +
+ +
+ + +
+ + + +
+
+ + + + + + +
+ + + + diff --git a/dashboard/templates/index.html b/dashboard/templates/index.html index 5e4c582..a61098a 100644 --- a/dashboard/templates/index.html +++ b/dashboard/templates/index.html @@ -21,11 +21,12 @@ Consultoría AS - Social Media diff --git a/scripts/test_connections.py b/scripts/test_connections.py new file mode 100755 index 0000000..29257a3 --- /dev/null +++ b/scripts/test_connections.py @@ -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)