""" ContentGeneratorV2 - Generador de contenido mejorado con DeepSeek. Este módulo maneja: - Generación de contenido usando prompts de la biblioteca - Integración con Context Engine para anti-repetición - Soporte para few-shot learning - Parámetros dinámicos por tipo de contenido """ import json from typing import Optional, List, Dict, Any from openai import OpenAI from sqlalchemy.orm import Session from app.core.config import settings from app.services.ai.prompt_library import prompt_library, PromptLibrary from app.services.ai.context_engine import context_engine, ContextEngine from app.services.ai.platform_adapter import platform_adapter, PlatformAdapter class ContentGeneratorV2: """ Generador de contenido v2.0 con motor modular. Mejoras sobre v1: - Prompts externalizados en YAML - Anti-repetición via Context Engine - Few-shot learning con posts exitosos - Adaptación automática por plataforma """ def __init__( self, prompt_lib: Optional[PromptLibrary] = None, ctx_engine: Optional[ContextEngine] = None, plt_adapter: Optional[PlatformAdapter] = None ): """ Inicializar el generador. Args: prompt_lib: PromptLibrary (usa global si no se especifica) ctx_engine: ContextEngine (usa global si no se especifica) plt_adapter: PlatformAdapter (usa global si no se especifica) """ self._client = None self.model = "deepseek-chat" self.prompt_lib = prompt_lib or prompt_library self.ctx_engine = ctx_engine or context_engine self.plt_adapter = plt_adapter or platform_adapter @property def client(self) -> OpenAI: """Lazy initialization del cliente OpenAI/DeepSeek.""" if self._client is None: if not settings.DEEPSEEK_API_KEY: raise ValueError( "DEEPSEEK_API_KEY no configurada. " "Configura la variable de entorno." ) self._client = OpenAI( api_key=settings.DEEPSEEK_API_KEY, base_url=settings.DEEPSEEK_BASE_URL ) return self._client # === Generación Principal === async def generate( self, template_name: str, variables: Dict[str, Any], platform: str, db: Optional[Session] = None, use_context: bool = True, use_few_shot: bool = True, personality: Optional[str] = None, temperature_override: Optional[float] = None, ) -> Dict[str, Any]: """ Generar contenido usando un template. Args: template_name: Nombre del template (tip_tech, product_post, etc.) variables: Variables para el template platform: Plataforma destino db: Sesión de DB (requerida para context) use_context: Si usar anti-repetición use_few_shot: Si usar ejemplos de posts exitosos personality: Override de personalidad temperature_override: Override de temperature Returns: Dict con: - content: Contenido generado - adapted_content: Contenido adaptado a plataforma - metadata: Info de generación """ # 1. Renderizar template rendered = self.prompt_lib.render_template( template_name, variables, personality or "default" ) system_prompt = rendered["system_prompt"] user_prompt = rendered["user_prompt"] params = rendered["parameters"] # 2. Agregar contexto de exclusión si hay DB if db and use_context: exclusion_context = self.ctx_engine.build_exclusion_context( db, template_name, variables.get("category") ) if exclusion_context: user_prompt += f"\n\n{exclusion_context}" # 3. Agregar few-shot examples si hay DB if db and use_few_shot: few_shot_context = self.ctx_engine.build_few_shot_context( db, template_name, platform ) if few_shot_context: user_prompt += f"\n\n{few_shot_context}" # También agregar ejemplos del YAML si hay categoría category = variables.get("category") if category: yaml_examples = self.prompt_lib.get_few_shot_examples( template_name, category, max_examples=2 ) if yaml_examples: user_prompt += "\n\nEJEMPLOS DE REFERENCIA:\n" for ex in yaml_examples: user_prompt += f"\n---\n{ex}\n---" # 4. Agregar límites de plataforma limits = self.plt_adapter.get_limits(platform) user_prompt += f"\n\nPLATAFORMA: {platform}" user_prompt += f"\nLÍMITE DE CARACTERES: {limits.get('max_characters', 500)}" user_prompt += f"\nMÁXIMO HASHTAGS: {limits.get('max_hashtags', 5)}" # 5. Llamar a DeepSeek temperature = temperature_override or params.get("temperature", 0.7) max_tokens = params.get("max_tokens", 400) response = self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], max_tokens=max_tokens, temperature=temperature ) content = response.choices[0].message.content.strip() # 6. Detectar si es un hilo (no adaptar, devolver posts separados) is_thread = template_name == "thread" if is_thread: # Parsear posts del hilo thread_posts = self._parse_thread_posts(content) return { "content": content, "adapted_content": content, # No truncar hilos "is_thread": True, "thread_posts": thread_posts, "truncated": False, "metadata": { "template": template_name, "personality": rendered["personality"], "platform": platform, "temperature": temperature, "tokens_used": response.usage.total_tokens if response.usage else None, "changes_made": [], "num_posts": len(thread_posts), } } # 7. Adaptar a plataforma (solo para contenido normal) adapted = self.plt_adapter.adapt(content, platform) return { "content": content, "adapted_content": adapted.content, "is_thread": False, "truncated": adapted.truncated, "metadata": { "template": template_name, "personality": rendered["personality"], "platform": platform, "temperature": temperature, "tokens_used": response.usage.total_tokens if response.usage else None, "changes_made": adapted.changes_made, } } def _parse_thread_posts(self, content: str) -> List[Dict[str, Any]]: """ Parsear el contenido de un hilo en posts individuales. Args: content: Contenido completo del hilo Returns: Lista de dicts con cada post y sus metadatos """ import re posts = [] # Intentar separar por patrones de numeración: "1/", "2/", "1)", "2)" # Buscar patrones como "\n1/" o inicio con "1/" numbered_pattern = r'(?:^|\n)(\d+)[/\)]\s*' # Encontrar todas las posiciones de inicio de posts numerados matches = list(re.finditer(numbered_pattern, content)) if matches and len(matches) >= 2: # Extraer posts basándose en las posiciones for i, match in enumerate(matches): start = match.start() # El fin es el inicio del siguiente match o el final del contenido end = matches[i + 1].start() if i + 1 < len(matches) else len(content) post_content = content[start:end].strip() post_num = int(match.group(1)) posts.append({ "number": post_num, "content": post_content, "char_count": len(post_content) }) else: # Fallback: separar por doble salto de línea raw_posts = [p.strip() for p in content.split("\n\n") if p.strip()] for i, post_content in enumerate(raw_posts, 1): posts.append({ "number": i, "content": post_content, "char_count": len(post_content) }) return posts # === Métodos de Conveniencia === async def generate_tip( self, category: str, platform: str, db: Optional[Session] = None, difficulty: str = "principiante", **kwargs ) -> Dict[str, Any]: """ Generar un tip de tecnología. Args: category: Categoría del tip (ia, productividad, seguridad, etc.) platform: Plataforma destino db: Sesión de DB difficulty: Nivel de dificultad **kwargs: Argumentos adicionales para generate() Returns: Dict con contenido generado """ return await self.generate( template_name="tip_tech", variables={ "category": category, "difficulty_level": difficulty, "target_audience": "profesionales tech" }, platform=platform, db=db, **kwargs ) async def generate_product_post( self, product: Dict[str, Any], platform: str, db: Optional[Session] = None, **kwargs ) -> Dict[str, Any]: """ Generar post para un producto. Args: product: Dict con datos del producto platform: Plataforma destino db: Sesión de DB **kwargs: Argumentos adicionales Returns: Dict con contenido generado """ return await self.generate( template_name="product_post", variables={ "product_name": product["name"], "product_description": product.get("description", ""), "price": product["price"], "category": product["category"], "specs": json.dumps(product.get("specs", {}), ensure_ascii=False), "highlights": ", ".join(product.get("highlights", [])), }, platform=platform, db=db, personality="promotional", **kwargs ) async def generate_service_post( self, service: Dict[str, Any], platform: str, db: Optional[Session] = None, **kwargs ) -> Dict[str, Any]: """ Generar post para un servicio. Args: service: Dict con datos del servicio platform: Plataforma destino db: Sesión de DB **kwargs: Argumentos adicionales Returns: Dict con contenido generado """ return await self.generate( template_name="service_post", variables={ "service_name": service["name"], "service_description": service.get("description", ""), "category": service["category"], "target_sectors": ", ".join(service.get("target_sectors", [])), "benefits": ", ".join(service.get("benefits", [])), "call_to_action": service.get("call_to_action", "Contáctanos"), }, platform=platform, db=db, personality="promotional", **kwargs ) async def generate_thread( self, topic: str, num_posts: int = 5, db: Optional[Session] = None, **kwargs ) -> List[str]: """ Generar un hilo educativo. Args: topic: Tema del hilo num_posts: Número de posts en el hilo db: Sesión de DB **kwargs: Argumentos adicionales Returns: Lista de posts del hilo """ result = await self.generate( template_name="thread", variables={ "topic": topic, "num_posts": num_posts, "depth": "intermedio", }, platform="x", # Threads son para X principalmente db=db, personality="educational", **kwargs ) # Parsear posts del hilo content = result["content"] posts = [p.strip() for p in content.split("\n\n") if p.strip()] # Si no se separaron bien, intentar por números if len(posts) < num_posts - 1: import re posts = re.split(r"\n(?=\d+[/\)])", content) posts = [p.strip() for p in posts if p.strip()] return posts async def generate_response( self, interaction_content: str, interaction_type: str, sentiment: str = "neutral", context: Optional[str] = None, **kwargs ) -> List[str]: """ Generar sugerencias de respuesta. Args: interaction_content: Contenido de la interacción interaction_type: Tipo (comment, mention, reply, dm) sentiment: Sentimiento detectado context: Contexto adicional **kwargs: Argumentos adicionales Returns: Lista de 3 opciones de respuesta """ result = await self.generate( template_name="response", variables={ "interaction_content": interaction_content, "interaction_type": interaction_type, "sentiment": sentiment, "context": context or "Sin contexto adicional", }, platform="x", use_context=False, use_few_shot=False, **kwargs ) # Parsear las 3 opciones content = result["content"] lines = [l.strip() for l in content.split("\n") if l.strip()] # Limpiar numeración responses = [] for line in lines: if line and line[0].isdigit() and len(line) > 2: # Remover "1.", "1)", "1:" etc. clean = line[2:].strip() if line[1] in ".):,-" else line if clean: responses.append(clean) elif line and not line[0].isdigit(): responses.append(line) return responses[:3] async def adapt_content( self, content: str, source_platform: str, target_platform: str ) -> str: """ Adaptar contenido existente a otra plataforma usando IA. Args: content: Contenido original source_platform: Plataforma de origen target_platform: Plataforma destino Returns: Contenido adaptado """ # Generar prompt de adaptación adaptation_prompt = self.plt_adapter.get_adaptation_prompt( content, source_platform, target_platform ) system_prompt = self.prompt_lib.get_system_prompt() response = self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": adaptation_prompt} ], max_tokens=500, temperature=0.6 ) return response.choices[0].message.content.strip() async def generate_for_all_platforms( self, template_name: str, variables: Dict[str, Any], platforms: List[str], db: Optional[Session] = None, **kwargs ) -> Dict[str, Dict[str, Any]]: """ Generar contenido para múltiples plataformas. Genera una versión base y la adapta para cada plataforma. Args: template_name: Nombre del template variables: Variables del template platforms: Lista de plataformas db: Sesión de DB **kwargs: Argumentos adicionales Returns: Dict de plataforma -> resultado de generación """ results = {} # Generar versión base (para la plataforma con más espacio) base_platform = "instagram" # Más espacio = más contexto if "facebook" in platforms: base_platform = "facebook" base_result = await self.generate( template_name=template_name, variables=variables, platform=base_platform, db=db, **kwargs ) base_content = base_result["content"] results[base_platform] = base_result # Adaptar para otras plataformas for platform in platforms: if platform == base_platform: continue # Usar adaptación con IA para mejor calidad adapted_content = await self.adapt_content( base_content, base_platform, platform ) # Validar adaptación adapted = self.plt_adapter.adapt(adapted_content, platform) results[platform] = { "content": adapted_content, "adapted_content": adapted.content, "truncated": adapted.truncated, "metadata": { "template": template_name, "platform": platform, "adapted_from": base_platform, "changes_made": adapted.changes_made, } } return results # Instancia global content_generator_v2 = ContentGeneratorV2()