Implementación inicial del sistema de automatización de redes sociales

- Estructura completa del proyecto con FastAPI
- Modelos de base de datos (productos, servicios, posts, calendario, interacciones)
- Publishers para X, Threads, Instagram, Facebook
- Generador de contenido con DeepSeek API
- Worker de Celery con tareas programadas
- Dashboard básico con templates HTML
- Docker Compose para despliegue
- Documentación completa

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-01-28 01:11:44 +00:00
commit 049d2133f9
53 changed files with 5876 additions and 0 deletions

134
app/models/post.py Normal file
View File

@@ -0,0 +1,134 @@
"""
Modelo de Post - Posts generados y programados.
"""
from datetime import datetime
from sqlalchemy import Column, Integer, String, Text, Boolean, DateTime, ForeignKey, JSON, Enum
from sqlalchemy.dialects.postgresql import ARRAY
import enum
from app.core.database import Base
class PostStatus(enum.Enum):
"""Estados posibles de un post."""
DRAFT = "draft"
PENDING_APPROVAL = "pending_approval"
APPROVED = "approved"
SCHEDULED = "scheduled"
PUBLISHING = "publishing"
PUBLISHED = "published"
FAILED = "failed"
CANCELLED = "cancelled"
class ContentType(enum.Enum):
"""Tipos de contenido."""
TIP_TECH = "tip_tech"
DATO_CURIOSO = "dato_curioso"
FRASE_MOTIVACIONAL = "frase_motivacional"
EFEMERIDE = "efemeride"
PRODUCTO = "producto"
SERVICIO = "servicio"
HILO_EDUCATIVO = "hilo_educativo"
CASO_EXITO = "caso_exito"
PROMOCION = "promocion"
ANUNCIO = "anuncio"
MANUAL = "manual"
class Post(Base):
"""Modelo para posts de redes sociales."""
__tablename__ = "posts"
id = Column(Integer, primary_key=True, index=True)
# Contenido
content = Column(Text, nullable=False)
content_type = Column(String(50), nullable=False, index=True)
# Contenido adaptado por plataforma (opcional)
content_x = Column(Text, nullable=True) # Versión para X (280 chars)
content_threads = Column(Text, nullable=True)
content_instagram = Column(Text, nullable=True)
content_facebook = Column(Text, nullable=True)
# Plataformas destino
platforms = Column(ARRAY(String), nullable=False)
# Ejemplo: ["x", "threads", "instagram", "facebook"]
# Estado y programación
status = Column(String(50), default="draft", index=True)
scheduled_at = Column(DateTime, nullable=True, index=True)
published_at = Column(DateTime, nullable=True)
# Imagen
image_url = Column(String(500), nullable=True)
image_template_id = Column(Integer, ForeignKey("image_templates.id"), nullable=True)
# IDs de publicación en cada plataforma
platform_post_ids = Column(JSON, nullable=True)
# Ejemplo: {"x": "123456", "instagram": "789012", ...}
# Errores de publicación
error_message = Column(Text, nullable=True)
retry_count = Column(Integer, default=0)
# Aprobación
approval_required = Column(Boolean, default=False)
approved_by = Column(String(100), nullable=True)
approved_at = Column(DateTime, nullable=True)
# Relaciones con contenido fuente
product_id = Column(Integer, ForeignKey("products.id"), nullable=True)
service_id = Column(Integer, ForeignKey("services.id"), nullable=True)
tip_template_id = Column(Integer, ForeignKey("tip_templates.id"), nullable=True)
# Metadatos
hashtags = Column(ARRAY(String), nullable=True)
mentions = Column(ARRAY(String), nullable=True)
# Métricas (actualizadas después de publicar)
metrics = Column(JSON, nullable=True)
# Ejemplo: {"likes": 10, "retweets": 5, "comments": 3}
# Timestamps
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
def __repr__(self):
return f"<Post {self.id} - {self.status}>"
def to_dict(self):
"""Convertir a diccionario."""
return {
"id": self.id,
"content": self.content,
"content_type": self.content_type,
"content_x": self.content_x,
"content_threads": self.content_threads,
"content_instagram": self.content_instagram,
"content_facebook": self.content_facebook,
"platforms": self.platforms,
"status": self.status,
"scheduled_at": self.scheduled_at.isoformat() if self.scheduled_at else None,
"published_at": self.published_at.isoformat() if self.published_at else None,
"image_url": self.image_url,
"platform_post_ids": self.platform_post_ids,
"error_message": self.error_message,
"approval_required": self.approval_required,
"hashtags": self.hashtags,
"metrics": self.metrics,
"created_at": self.created_at.isoformat() if self.created_at else None
}
def get_content_for_platform(self, platform: str) -> str:
"""Obtener contenido adaptado para una plataforma específica."""
platform_content = {
"x": self.content_x,
"threads": self.content_threads,
"instagram": self.content_instagram,
"facebook": self.content_facebook
}
return platform_content.get(platform) or self.content