Files
social-media-automation/app/main.py
Consultoría AS edc0e5577b feat(phase-4): Complete scheduling and automation system
- Add Celery worker with 5 scheduled tasks (Beat)
- Create ContentScheduler for optimal posting times
- Add calendar endpoints for scheduled posts management
- Implement Telegram notification service
- Add notification API with setup guide

Celery Beat Schedule:
- check_scheduled_posts: Every minute
- generate_daily_content: Daily at 6 AM
- sync_interactions: Every 15 minutes
- send_daily_summary: Daily at 9 PM
- cleanup_old_data: Weekly on Sundays

New endpoints:
- GET /api/calendar/posts/scheduled - List scheduled posts
- GET /api/calendar/posts/view - Calendar view
- GET /api/calendar/posts/slots - Available time slots
- POST /api/calendar/posts/{id}/schedule - Schedule post
- POST /api/calendar/posts/{id}/publish-now - Publish immediately
- GET /api/notifications/status - Check notification config
- POST /api/notifications/test - Send test notification
- GET /api/notifications/setup-guide - Configuration guide

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-28 01:56:10 +00:00

92 lines
3.0 KiB
Python

"""
Social Media Automation - Consultoría AS
=========================================
Sistema automatizado para la creación y publicación de contenido
en redes sociales (X, Threads, Instagram, Facebook).
"""
from contextlib import asynccontextmanager
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, publish, generate, notifications
from app.core.config import settings
from app.core.database import engine
from app.models import Base
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Lifecycle manager - crea tablas al iniciar si hay conexión a BD."""
try:
Base.metadata.create_all(bind=engine)
except Exception as e:
print(f"⚠️ No se pudo conectar a BD: {e}")
print(" La app funcionará pero sin persistencia hasta configurar BD")
yield
# Inicializar aplicación FastAPI
app = FastAPI(
title="Social Media Automation",
description="Sistema de automatización de redes sociales para Consultoría AS",
version="1.0.0",
docs_url="/api/docs",
redoc_url="/api/redoc",
lifespan=lifespan,
)
# Configurar CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # En producción, especificar dominios
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 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"])
app.include_router(posts.router, prefix="/api/posts", tags=["Posts"])
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.include_router(generate.router, prefix="/api/generate", tags=["AI Generation"])
app.include_router(notifications.router, prefix="/api/notifications", tags=["Notifications"])
@app.get("/api/health")
async def health_check():
"""Verificar estado del sistema."""
return {
"status": "healthy",
"app": settings.APP_NAME,
"version": "1.0.0"
}
@app.get("/api/stats")
async def get_stats():
"""Obtener estadísticas generales del sistema."""
# TODO: Implementar estadísticas reales desde la BD
return {
"posts_today": 0,
"posts_week": 0,
"posts_month": 0,
"pending_approval": 0,
"scheduled": 0,
"interactions_pending": 0
}