""" Priority Actions Helper Functions Generates action-oriented insights for commercial agents """ import pandas as pd from datetime import datetime, timedelta import re def parse_year_month(value): """Parse year/month from mes_col format (e.g., '2024/01' or '01/2024')""" if value is None: return None s = str(value).strip() if not s: return None s = s.replace('-', '/').replace('.', '/') # Try YYYY/MM format m = re.search(r"(\d{4})\D?(\d{1,2})", s) if m: year = int(m.group(1)) month = int(m.group(2)) return datetime(year, month, 1) # Try MM/YYYY format m = re.search(r"(\d{1,2})\D?(\d{4})", s) if m: month = int(m.group(1)) year = int(m.group(2)) return datetime(year, month, 1) return None def get_inactive_clients(df, cliente_col, mes_col, fat_col, comercial_filter=None, days_threshold=60): """ Get clients who haven't ordered in the last X days Args: df: DataFrame with sales data cliente_col: Column name for client mes_col: Column name for month/period fat_col: Column name for revenue comercial_filter: List of comercial names to filter by (for RBAC) days_threshold: Days since last order to consider inactive (default 60) Returns: List of dicts with client info """ if not all([cliente_col, mes_col, fat_col]): return [] # Filter by comercial if needed df_filtered = df.copy() if comercial_filter: comercial_col = None for col in df.columns: if 'comercial' in col.lower(): comercial_col = col break if comercial_col: df_filtered = df_filtered[ df_filtered[comercial_col].astype(str).str.strip().str.lower().isin( [c.strip().lower() for c in comercial_filter] ) ] # Parse dates and revenue df_filtered = df_filtered.copy() df_filtered['__parsed_date'] = df_filtered[mes_col].apply(parse_year_month) df_filtered[fat_col] = pd.to_numeric(df_filtered[fat_col], errors='coerce') # Remove rows without valid dates or clients df_filtered = df_filtered[ df_filtered['__parsed_date'].notna() & df_filtered[cliente_col].notna() & (df_filtered[cliente_col].astype(str).str.strip() != '') ] if df_filtered.empty: return [] # Get last order date per client client_last_orders = df_filtered.groupby(cliente_col).agg({ '__parsed_date': 'max', fat_col: 'sum' }).reset_index() client_last_orders.columns = ['cliente', 'last_order_date', 'total_revenue'] # Calculate days since last order today = datetime.now() client_last_orders['days_since_order'] = (today - client_last_orders['last_order_date']).dt.days # Filter inactive clients inactive = client_last_orders[client_last_orders['days_since_order'] > days_threshold].copy() # Sort by revenue (descending) to prioritize important clients inactive = inactive.sort_values('total_revenue', ascending=False) # Convert to list of dicts results = [] for _, row in inactive.head(10).iterrows(): # Limit to top 10 results.append({ 'client_name': str(row['cliente']), 'last_order_date': row['last_order_date'].strftime('%Y-%m-%d'), 'days_since_order': int(row['days_since_order']), 'total_revenue': float(row['total_revenue']) }) return results def get_revenue_drop_clients(df, cliente_col, mes_col, fat_col, __year_col='__year', comercial_filter=None, drop_threshold=20, limit=10): """ Get clients with significant revenue drop (current year vs previous year, same period) Args: df: DataFrame with sales data cliente_col: Column name for client mes_col: Column name for month/period fat_col: Column name for revenue __year_col: Column name for parsed year comercial_filter: List of comercial names to filter by (for RBAC) drop_threshold: Percentage drop to flag (default 20%) limit: Max number of clients to return. Use None to return all. Returns: List of dicts with client revenue comparison """ if not all([cliente_col, fat_col]): return [] # Filter by comercial if needed df_filtered = df.copy() if comercial_filter: comercial_col = None for col in df.columns: if 'comercial' in col.lower(): comercial_col = col break if comercial_col: df_filtered = df_filtered[ df_filtered[comercial_col].astype(str).str.strip().str.lower().isin( [c.strip().lower() for c in comercial_filter] ) ] # Parse year/month if not already done if __year_col not in df_filtered.columns and mes_col: df_filtered['__parsed_date'] = df_filtered[mes_col].apply(parse_year_month) df_filtered[__year_col] = df_filtered['__parsed_date'].apply( lambda x: str(x.year) if x else None ) df_filtered['__month'] = df_filtered['__parsed_date'].apply( lambda x: str(x.month).zfill(2) if x else None ) # Convert revenue to numeric df_filtered[fat_col] = pd.to_numeric(df_filtered[fat_col], errors='coerce') # Get current year and month current_year = str(datetime.now().year) current_month = datetime.now().month # Filter to same period (Jan-current month) for both years df_current_period = df_filtered[ (df_filtered[__year_col] == current_year) & (df_filtered['__month'].notna()) & (df_filtered['__month'].astype(str).astype(int) <= current_month) ].copy() previous_year = str(int(current_year) - 1) df_previous_period = df_filtered[ (df_filtered[__year_col] == previous_year) & (df_filtered['__month'].notna()) & (df_filtered['__month'].astype(str).astype(int) <= current_month) ].copy() # Calculate revenue per client current_revenue = df_current_period.groupby(cliente_col)[fat_col].sum() previous_revenue = df_previous_period.groupby(cliente_col)[fat_col].sum() # Merge dataframes comparison = pd.DataFrame({ 'current_year_revenue': current_revenue, 'previous_year_revenue': previous_revenue }).fillna(0) # Calculate percentage change comparison['change_pct'] = ((comparison['current_year_revenue'] - comparison['previous_year_revenue']) / comparison['previous_year_revenue'] * 100) # Filter clients with significant drop # Only consider clients who had revenue in previous year (> 0) drops = comparison[ (comparison['previous_year_revenue'] > 0) & (comparison['change_pct'] < -drop_threshold) ].copy() # Sort by absolute revenue loss (descending) drops['revenue_loss'] = drops['previous_year_revenue'] - drops['current_year_revenue'] drops = drops.sort_values('revenue_loss', ascending=False) # Convert to list of dicts results = [] limited_drops = drops if limit is None else drops.head(limit) for client_name, row in limited_drops.iterrows(): results.append({ 'client_name': str(client_name), 'current_year_revenue': float(row['current_year_revenue']), 'previous_year_revenue': float(row['previous_year_revenue']), 'change_pct': float(row['change_pct']) }) return results def get_clients_without_recent_visits(visit_df=None, cliente_col='cliente', date_col='visit_date', comercial_filter=None, days_threshold=45): """ Get clients not visited in X days NOTE: This requires visit tracking data which is not yet implemented. This function is a placeholder for when visit logs are added. Args: visit_df: DataFrame with visit logs (client, date, comercial) cliente_col: Column name for client date_col: Column name for visit date comercial_filter: List of comercial names to filter by days_threshold: Days since last visit to flag (default 45) Returns: List of dicts with client visit info (empty until visit tracking is added) """ # Placeholder - return empty list until visit tracking is implemented # TODO: Implement when visit logging system is added return [] def get_upcoming_visits(visit_df=None, date_col='visit_date', objetivo_col='objective', comercial_filter=None, days_ahead=7): """ Get scheduled visits in the next X days NOTE: This requires visit scheduling data which is not yet implemented. This function is a placeholder for when visit planning is added. Args: visit_df: DataFrame with scheduled visits date_col: Column name for scheduled date objetivo_col: Column name for visit objective comercial_filter: List of comercial names to filter by days_ahead: Number of days ahead to show (default 7) Returns: List of dicts with upcoming visits (empty until visit scheduling is added) """ # Placeholder - return empty list until visit scheduling is implemented # TODO: Implement when visit planning system is added return [] def format_currency(value): """Format value as EUR currency""" return f"€{value:,.2f}".replace(",", " ") def format_percentage(value): """Format value as percentage with sign""" sign = "+" if value >= 0 else "" return f"{sign}{value:.1f}%"