""" Servicio de generación de contenido por lotes. Genera múltiples posts para el calendario de contenido. """ import asyncio from datetime import datetime, timedelta from typing import List, Dict, Optional from dataclasses import dataclass from enum import Enum from app.services.content_generator import content_generator from app.data.content_templates import ( get_tip_template, get_optimal_times, OPTIMAL_POSTING_TIMES ) from app.core.config import settings class ContentType(str, Enum): TIP = "tip" PRODUCT = "product" SERVICE = "service" ENGAGEMENT = "engagement" THREAD = "thread" @dataclass class GeneratedPost: """Post generado para el calendario.""" content: str content_type: ContentType platform: str scheduled_at: datetime metadata: Dict = None @dataclass class BatchResult: """Resultado de generación por lotes.""" success: bool posts: List[GeneratedPost] errors: List[str] total_requested: int total_generated: int class BatchGenerator: """Generador de contenido por lotes.""" # Distribución de contenido por defecto (porcentaje) DEFAULT_DISTRIBUTION = { ContentType.TIP: 60, ContentType.PRODUCT: 20, ContentType.SERVICE: 15, ContentType.ENGAGEMENT: 5, } # Frecuencia por plataforma (posts por día) DEFAULT_FREQUENCY = { "x": 4, "threads": 3, "instagram": 2, "facebook": 1, } def __init__(self): self.generator = content_generator async def generate_week( self, platforms: List[str], start_date: Optional[datetime] = None, distribution: Optional[Dict[ContentType, int]] = None, frequency: Optional[Dict[str, int]] = None, tip_categories: Optional[List[str]] = None, ) -> BatchResult: """ Generar contenido para una semana. Args: platforms: Lista de plataformas start_date: Fecha de inicio (default: mañana) distribution: Distribución de tipos de contenido frequency: Posts por día por plataforma tip_categories: Categorías para tips Returns: BatchResult con posts generados """ if start_date is None: start_date = datetime.now().replace( hour=0, minute=0, second=0, microsecond=0 ) + timedelta(days=1) return await self._generate_batch( platforms=platforms, start_date=start_date, days=7, distribution=distribution, frequency=frequency, tip_categories=tip_categories ) async def generate_month( self, platforms: List[str], start_date: Optional[datetime] = None, distribution: Optional[Dict[ContentType, int]] = None, frequency: Optional[Dict[str, int]] = None, tip_categories: Optional[List[str]] = None, ) -> BatchResult: """Generar contenido para un mes (30 días).""" if start_date is None: start_date = datetime.now().replace( hour=0, minute=0, second=0, microsecond=0 ) + timedelta(days=1) return await self._generate_batch( platforms=platforms, start_date=start_date, days=30, distribution=distribution, frequency=frequency, tip_categories=tip_categories ) async def _generate_batch( self, platforms: List[str], start_date: datetime, days: int, distribution: Optional[Dict[ContentType, int]] = None, frequency: Optional[Dict[str, int]] = None, tip_categories: Optional[List[str]] = None, ) -> BatchResult: """Generar lote de posts.""" distribution = distribution or self.DEFAULT_DISTRIBUTION frequency = frequency or self.DEFAULT_FREQUENCY tip_categories = tip_categories or [ "productividad", "seguridad", "ia", "programacion", "hardware", "impresion3d", "general" ] posts = [] errors = [] total_requested = 0 # Calcular posts necesarios por plataforma for platform in platforms: posts_per_day = frequency.get(platform, 2) total_posts = posts_per_day * days total_requested += total_posts # Distribuir por tipo posts_by_type = {} for content_type, percentage in distribution.items(): count = int(total_posts * percentage / 100) posts_by_type[content_type] = count # Generar posts para cada tipo current_date = start_date day_counter = 0 for content_type, count in posts_by_type.items(): for i in range(count): # Calcular fecha y hora day_offset = (i * days) // count post_date = start_date + timedelta(days=day_offset) is_weekend = post_date.weekday() >= 5 # Obtener hora óptima times = get_optimal_times(platform, is_weekend) time_index = i % len(times) hour, minute = map(int, times[time_index].split(":")) scheduled_at = post_date.replace(hour=hour, minute=minute) try: # Generar contenido content = await self._generate_content( content_type=content_type, platform=platform, tip_categories=tip_categories ) if content: posts.append(GeneratedPost( content=content, content_type=content_type, platform=platform, scheduled_at=scheduled_at, metadata={ "batch_generated": True, "category": tip_categories[i % len(tip_categories)] if content_type == ContentType.TIP else None } )) except Exception as e: errors.append(f"{platform}/{content_type}: {str(e)}") return BatchResult( success=len(posts) > 0, posts=posts, errors=errors, total_requested=total_requested, total_generated=len(posts) ) async def _generate_content( self, content_type: ContentType, platform: str, tip_categories: List[str] ) -> Optional[str]: """Generar contenido individual.""" import random if content_type == ContentType.TIP: category = random.choice(tip_categories) template = get_tip_template(category) return await self.generator.generate_tip_tech( category=category, platform=platform, template=template ) elif content_type == ContentType.ENGAGEMENT: # Generar pregunta de engagement prompts = [ "¿Cuál es tu mayor reto tecnológico esta semana?", "¿Qué herramienta de productividad no podrías dejar de usar?", "¿Prefieres trabajar en oficina o remoto? ¿Por qué?", "¿Qué tecnología te gustaría aprender este año?", "¿Cuál es tu consejo #1 para mantenerte productivo?", ] return random.choice(prompts) # Para productos y servicios, devolver None # (requieren datos específicos) return None async def generate_tips_batch( self, count: int, platform: str, categories: Optional[List[str]] = None ) -> List[str]: """ Generar múltiples tips de una vez. Args: count: Cantidad de tips a generar platform: Plataforma destino categories: Categorías a usar (se rotarán) Returns: Lista de tips generados """ categories = categories or [ "productividad", "seguridad", "ia", "programacion", "hardware", "general" ] tips = [] for i in range(count): category = categories[i % len(categories)] try: tip = await self.generator.generate_tip_tech( category=category, platform=platform ) tips.append(tip) except Exception: pass # Ignorar errores individuales # Pequeña pausa para no saturar la API if i < count - 1: await asyncio.sleep(0.5) return tips def calculate_costs( self, platforms: List[str], days: int, frequency: Optional[Dict[str, int]] = None ) -> Dict: """ Calcular costos estimados para un lote. Returns: Dict con estimaciones de tokens y costos """ frequency = frequency or self.DEFAULT_FREQUENCY total_posts = sum( frequency.get(p, 2) * days for p in platforms ) # Estimaciones basadas en análisis previo avg_input_tokens = 650 avg_output_tokens = 250 total_input = total_posts * avg_input_tokens total_output = total_posts * avg_output_tokens # Precios DeepSeek input_cost = total_input * 0.14 / 1_000_000 output_cost = total_output * 0.28 / 1_000_000 return { "total_posts": total_posts, "estimated_input_tokens": total_input, "estimated_output_tokens": total_output, "estimated_cost_usd": round(input_cost + output_cost, 4), "platforms": platforms, "days": days, "frequency": frequency } # Instancia global batch_generator = BatchGenerator()