Consultoría AS
049d2133f9
- 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>
135 lines
4.6 KiB
Python
135 lines
4.6 KiB
Python
"""
|
|
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
|