""" Trend analysis for Sales Bot Aggregates and analyzes sales data over time """ import logging from datetime import datetime, timedelta, timezone from typing import List, Dict, Optional from collections import defaultdict logger = logging.getLogger(__name__) # Mexico timezone TZ_MEXICO = timezone(timedelta(hours=-6)) class TrendAnalyzer: """ Analyzes sales trends over different time periods. """ def __init__(self, nocodb_client): self.nocodb = nocodb_client def get_daily_trends(self, dias: int = 30, vendedor: str = None) -> Dict: """ Obtiene tendencias diarias de ventas. Args: dias: Número de días a analizar vendedor: Username del vendedor (opcional, None para todos) Returns: dict con labels, ventas, tubos, y predicciones """ try: # Obtener ventas del período ventas = self._obtener_ventas_periodo(dias, vendedor) # Agrupar por día ventas_por_dia = self._agrupar_por_dia(ventas, dias) # Generar labels (fechas) hoy = datetime.now(TZ_MEXICO).date() labels = [] valores_ventas = [] valores_tubos = [] for i in range(dias - 1, -1, -1): fecha = hoy - timedelta(days=i) fecha_str = fecha.strftime('%Y-%m-%d') labels.append(fecha.strftime('%d/%m')) dia_data = ventas_por_dia.get(fecha_str, {'monto': 0, 'tubos': 0}) valores_ventas.append(dia_data['monto']) valores_tubos.append(dia_data['tubos']) # Generar predicciones si hay datos prediccion = [] if valores_ventas: from .predictions import prediccion_basica pred_result = prediccion_basica(valores_ventas, 7) if pred_result.get('predicciones'): # Añadir None para los días históricos y luego las predicciones prediccion = [None] * len(valores_ventas) prediccion.extend(pred_result['predicciones']) return { 'labels': labels, 'ventas': valores_ventas, 'tubos': valores_tubos, 'prediccion': prediccion, 'periodo': { 'inicio': (hoy - timedelta(days=dias-1)).strftime('%Y-%m-%d'), 'fin': hoy.strftime('%Y-%m-%d'), 'dias': dias } } except Exception as e: logger.error(f"Error obteniendo tendencias diarias: {str(e)}") return { 'labels': [], 'ventas': [], 'tubos': [], 'prediccion': [], 'error': str(e) } def _obtener_ventas_periodo(self, dias: int, vendedor: str = None) -> List[Dict]: """Obtiene ventas de los últimos N días.""" try: # Usar método existente del cliente NocoDB if hasattr(self.nocodb, 'get_ventas_historico'): return self.nocodb.get_ventas_historico(dias, vendedor) # Fallback: obtener ventas del mes y filtrar ventas = self.nocodb.get_ventas_mes(vendedor) if vendedor else self.nocodb.get_ventas_mes() # Filtrar por días hoy = datetime.now(TZ_MEXICO).date() fecha_inicio = hoy - timedelta(days=dias) ventas_filtradas = [] for venta in ventas: fecha_str = venta.get('fecha_venta', '') try: if 'T' in fecha_str: fecha = datetime.fromisoformat(fecha_str.replace('Z', '+00:00')) else: fecha = datetime.strptime(fecha_str[:10], '%Y-%m-%d') if fecha.tzinfo is None: fecha = fecha.replace(tzinfo=timezone.utc) fecha_local = fecha.astimezone(TZ_MEXICO).date() if fecha_local >= fecha_inicio: ventas_filtradas.append(venta) except: continue return ventas_filtradas except Exception as e: logger.error(f"Error obteniendo ventas del período: {str(e)}") return [] def _agrupar_por_dia(self, ventas: List[Dict], dias: int) -> Dict[str, Dict]: """Agrupa ventas por día.""" ventas_por_dia = defaultdict(lambda: {'monto': 0, 'tubos': 0, 'count': 0}) for venta in ventas: try: fecha_str = venta.get('fecha_venta', '') if 'T' in fecha_str: fecha = datetime.fromisoformat(fecha_str.replace('Z', '+00:00')) else: fecha = datetime.strptime(fecha_str[:10], '%Y-%m-%d') if fecha.tzinfo is None: fecha = fecha.replace(tzinfo=timezone.utc) fecha_local = fecha.astimezone(TZ_MEXICO).date() fecha_key = fecha_local.strftime('%Y-%m-%d') monto = float(venta.get('monto', 0) or 0) ventas_por_dia[fecha_key]['monto'] += monto ventas_por_dia[fecha_key]['count'] += 1 # Contar tubos si hay información de productos tubos = self._contar_tubos_venta(venta) ventas_por_dia[fecha_key]['tubos'] += tubos except Exception as e: continue return dict(ventas_por_dia) def _contar_tubos_venta(self, venta: Dict) -> int: """Cuenta tubos de tinte en una venta.""" # Esto depende de la estructura de datos # Por ahora retorna un valor estimado basado en el monto # En producción, debería consultar la tabla de detalles tubos = venta.get('tubos', 0) if tubos: return int(tubos) # Estimación: ~$50 por tubo en promedio monto = float(venta.get('monto', 0) or 0) return max(1, int(monto / 50)) if monto > 0 else 0 def get_weekly_summary(self, semanas: int = 4) -> Dict: """ Obtiene resumen semanal de ventas. Args: semanas: Número de semanas a analizar Returns: dict con datos semanales """ try: dias_totales = semanas * 7 ventas = self._obtener_ventas_periodo(dias_totales) # Agrupar por semana hoy = datetime.now(TZ_MEXICO).date() semanas_data = [] for i in range(semanas): fin_semana = hoy - timedelta(days=i * 7) inicio_semana = fin_semana - timedelta(days=6) # Filtrar ventas de esta semana ventas_semana = [] for venta in ventas: try: fecha_str = venta.get('fecha_venta', '') if 'T' in fecha_str: fecha = datetime.fromisoformat(fecha_str.replace('Z', '+00:00')) else: fecha = datetime.strptime(fecha_str[:10], '%Y-%m-%d') if fecha.tzinfo is None: fecha = fecha.replace(tzinfo=timezone.utc) fecha_local = fecha.astimezone(TZ_MEXICO).date() if inicio_semana <= fecha_local <= fin_semana: ventas_semana.append(venta) except: continue monto_total = sum(float(v.get('monto', 0) or 0) for v in ventas_semana) semanas_data.append({ 'semana': i + 1, 'inicio': inicio_semana.strftime('%Y-%m-%d'), 'fin': fin_semana.strftime('%Y-%m-%d'), 'label': f"{inicio_semana.strftime('%d/%m')} - {fin_semana.strftime('%d/%m')}", 'monto': round(monto_total, 2), 'ventas': len(ventas_semana) }) # Invertir para orden cronológico semanas_data.reverse() return { 'semanas': semanas_data, 'total_semanas': semanas } except Exception as e: logger.error(f"Error obteniendo resumen semanal: {str(e)}") return {'semanas': [], 'error': str(e)} def get_monthly_summary(self, meses: int = 6) -> Dict: """ Obtiene resumen mensual de ventas. Args: meses: Número de meses a analizar Returns: dict con datos mensuales """ try: # Esto requeriría acceso a más datos históricos # Por ahora retornamos estructura básica hoy = datetime.now(TZ_MEXICO).date() meses_data = [] for i in range(meses): # Calcular mes mes_actual = hoy.month - i anio_actual = hoy.year while mes_actual < 1: mes_actual += 12 anio_actual -= 1 mes_str = f"{anio_actual}-{mes_actual:02d}" mes_nombre = datetime(anio_actual, mes_actual, 1).strftime('%B %Y') meses_data.append({ 'mes': mes_str, 'nombre': mes_nombre, 'monto': 0, # Se llenaría con datos reales 'ventas': 0 }) meses_data.reverse() return { 'meses': meses_data, 'total_meses': meses } except Exception as e: logger.error(f"Error obteniendo resumen mensual: {str(e)}") return {'meses': [], 'error': str(e)}