#!/usr/bin/env python3 """ Fonctions Sales natives pour Gradio ================================== Interface simplifiée utilisant les fonctions du module Sales existant. Pas de logique métier codée en dur, utilise les utilitaires Sales. """ import logging from typing import Dict, List, Any, Optional from datetime import datetime, timedelta import json # Import du client Odoo import config logger = logging.getLogger(__name__) # ============================================================================= # FONCTIONS UTILITAIRES INTERNES # ============================================================================= def _format_currency(amount: float) -> str: """Formate un montant en euros""" return f"{amount:,.2f} €" def _format_percentage(value: float) -> str: """Formate un pourcentage""" return f"{value:.1f}%" def _safe_divide(numerator: float, denominator: float, default: float = 0.0) -> float: """Division sécurisée évitant la division par zéro""" return numerator / denominator if denominator != 0 else default def _get_utc_now() -> datetime: """Retourne la date/heure UTC actuelle""" return datetime.utcnow() def _analyze_sale_order(order: Dict[str, Any]) -> Dict[str, Any]: """ Analyse une commande de vente """ amount_total = order.get('amount_total', 0) or 0 amount_untaxed = order.get('amount_untaxed', 0) or 0 state = order.get('state', 'draft') # Classification par montant if amount_total >= 100000: priority = "TRÈS HAUTE" emoji = "🔥" elif amount_total >= 50000: priority = "HAUTE" emoji = "🌟" elif amount_total >= 10000: priority = "MOYENNE" emoji = "📊" else: priority = "BASSE" emoji = "📋" # État traduit state_labels = { 'draft': 'Brouillon', 'sent': 'Devis envoyé', 'sale': 'Commande confirmée', 'done': 'Terminé', 'cancel': 'Annulé' } return { "priority": priority, "emoji": emoji, "state_label": state_labels.get(state, state), "amount_total": amount_total, "amount_untaxed": amount_untaxed } # ============================================================================= # FONCTIONS PRINCIPALES SALES POUR GRADIO # ============================================================================= async def get_sales_statistics() -> str: """ Récupère les statistiques complètes des ventes depuis Odoo. Returns: str: Statistiques Sales formatées avec métriques détaillées """ client = config.client if not client or not client.is_connected(): return "❌ **Client Odoo non connecté**\n\nCliquez 'Enregistrer' dans la section connexion pour vous connecter d'abord." try: logger.info("📊 Calcul des statistiques Sales...") # ======================================================================== # STATISTIQUES GLOBALES OPTIMISÉES # ======================================================================== # Statistiques de base avec read_group pour optimiser stats_data = client.read_group( 'sale.order', [('id', '>', 0)], ['amount_total:sum', 'amount_untaxed:sum'], [] ) ca_total = float(stats_data[0]['amount_total']) if stats_data and stats_data[0]['amount_total'] else 0.0 ca_ht = float(stats_data[0]['amount_untaxed']) if stats_data and stats_data[0]['amount_untaxed'] else 0.0 # Comptages par état total_devis = client.search_count('sale.order', []) devis_brouillon = client.search_count('sale.order', [('state', '=', 'draft')]) devis_envoyes = client.search_count('sale.order', [('state', '=', 'sent')]) commandes_confirmees = client.search_count('sale.order', [('state', '=', 'sale')]) commandes_terminees = client.search_count('sale.order', [('state', '=', 'done')]) # Ventes du mois en cours current_month = _get_utc_now().replace(day=1).strftime('%Y-%m-%d') monthly_stats = client.read_group( 'sale.order', [('date_order', '>=', current_month), ('state', 'in', ['sale', 'done'])], ['amount_total:sum'], [] ) ca_mois = float(monthly_stats[0]['amount_total']) if monthly_stats and monthly_stats[0]['amount_total'] else 0.0 # Commandes haute valeur (> 50k€) commandes_haute_valeur = client.search_count('sale.order', [('amount_total', '>', 50000)]) # ======================================================================== # MÉTRIQUES AVANCÉES # ======================================================================== # Taux de conversion (devis envoyés -> commandes) total_prospects = devis_envoyes + commandes_confirmees + commandes_terminees taux_conversion = _safe_divide(commandes_confirmees + commandes_terminees, total_prospects) * 100 # Panier moyen panier_moyen = _safe_divide(ca_total, total_devis) # Performance du pipeline if taux_conversion >= 70: pipeline_sante = "Excellent" elif taux_conversion >= 50: pipeline_sante = "Bon" elif taux_conversion >= 30: pipeline_sante = "Moyen" else: pipeline_sante = "À améliorer" # ======================================================================== # FORMATAGE DE LA RÉPONSE # ======================================================================== response = f"""💰 **STATISTIQUES SALES ODOO COMPLÈTES** 📊 **VUE D'ENSEMBLE**: • **Total devis/commandes**: {total_devis:,} • **Devis brouillons**: {devis_brouillon:,} • **Devis envoyés**: {devis_envoyes:,} • **Commandes confirmées**: {commandes_confirmees:,} • **Commandes terminées**: {commandes_terminees:,} 💰 **CHIFFRE D'AFFAIRES**: • **CA total**: {_format_currency(ca_total)} • **CA HT**: {_format_currency(ca_ht)} • **CA du mois**: {_format_currency(ca_mois)} • **Panier moyen**: {_format_currency(panier_moyen)} 📈 **PERFORMANCE COMMERCIALE**: • **Taux de conversion**: {_format_percentage(taux_conversion)} • **Santé du pipeline**: {pipeline_sante} • **Commandes haute valeur**: {commandes_haute_valeur:,} 📅 **Dernière mise à jour**: {_get_utc_now().strftime('%d/%m/%Y %H:%M')}""" logger.info("✅ Statistiques Sales calculées avec succès") return response except Exception as e: logger.error(f"❌ Erreur get_sales_statistics: {e}") return f"❌ **Erreur lors de la récupération des statistiques**: {str(e)}" async def analyze_quotations_advanced(domain_filter: str = "[]", limit: int = 20) -> str: """ Analyse avancée des devis avec classification. Args: domain_filter: Filtre de domaine Odoo au format JSON limit: Nombre maximum de devis à analyser Returns: str: Analyse formatée avec classifications et recommandations """ client = config.client if not client or not client.is_connected(): return "❌ **Client Odoo non connecté**\n\nCliquez 'Enregistrer' dans la section connexion pour vous connecter d'abord." try: # Validation des paramètres validated_limit = max(5, min(limit, 100)) if validated_limit != limit: logger.warning(f"Limite ajustée de {limit} à {validated_limit}") # Parser le domaine try: domain = json.loads(domain_filter) if domain_filter.strip() != "[]" else [] except json.JSONDecodeError: return "❌ **Format de domaine invalide**\n\nUtilisez le format JSON: [['field', '=', 'value']]" logger.info(f"🚀 Analyse avancée de {validated_limit} devis...") # ======================================================================== # RÉCUPÉRATION DES DONNÉES ODOO # ======================================================================== fields = [ 'name', 'partner_id', 'date_order', 'amount_total', 'amount_untaxed', 'state', 'validity_date', 'user_id', 'team_id', 'order_line' ] quotations_data = client.search_read('sale.order', domain, fields, limit=validated_limit) if not quotations_data: return "⚠️ **Aucun devis trouvé**\n\nVérifiez vos critères de recherche." # ======================================================================== # ANALYSE DES DEVIS # ======================================================================== quotation_analyses = [] total_amount = 0 high_priority_count = 0 expired_count = 0 for quotation in quotations_data: analysis = _analyze_sale_order(quotation) quotation_analyses.append({ 'quotation': quotation, 'analysis': analysis }) total_amount += analysis['amount_total'] if analysis['priority'] in ['TRÈS HAUTE', 'HAUTE']: high_priority_count += 1 # Vérifier expiration validity_date = quotation.get('validity_date') if validity_date and quotation.get('state') in ['draft', 'sent']: try: valid_until = datetime.fromisoformat(validity_date.replace('Z', '+00:00')) if valid_until < _get_utc_now(): expired_count += 1 except: pass # ======================================================================== # DISTRIBUTION PAR ÉTAT # ======================================================================== state_distribution = {} for item in quotation_analyses: state = item['quotation'].get('state', 'draft') state_distribution[state] = state_distribution.get(state, 0) + 1 # ======================================================================== # FORMATAGE DE LA RÉPONSE # ======================================================================== avg_amount = _safe_divide(total_amount, len(quotation_analyses)) response = f"""💼 **ANALYSE AVANCÉE DES DEVIS** 📊 **RÉSUMÉ DE L'ANALYSE**: • **Devis analysés**: {len(quotation_analyses):,} • **Date d'analyse**: {_get_utc_now().strftime('%d/%m/%Y %H:%M')} • **Montant moyen**: {_format_currency(avg_amount)} 📈 **DISTRIBUTION DES DEVIS**: • 🔥 **Haute priorité**: **{high_priority_count:,}** • ⚠️ **Devis expirés**: **{expired_count:,}** 💡 **MÉTRIQUES FINANCIÈRES**: • **Montant total**: {_format_currency(total_amount)} • **Montant moyen**: {_format_currency(avg_amount)} 📊 **RÉPARTITION PAR ÉTAT**:""" for state, count in state_distribution.items(): state_labels = { 'draft': 'Brouillons', 'sent': 'Envoyés', 'sale': 'Confirmés', 'done': 'Terminés', 'cancel': 'Annulés' } label = state_labels.get(state, state) response += f"\n• **{label}**: {count:,}" response += "\n\n🏆 **TOP OPPORTUNITÉS**:" # Trier par montant et afficher les top 3 sorted_quotations = sorted(quotation_analyses, key=lambda x: x['analysis']['amount_total'], reverse=True) for i, item in enumerate(sorted_quotations[:3], 1): q = item['quotation'] a = item['analysis'] partner_name = q.get('partner_id', [None, 'N/A'])[1] if q.get('partner_id') else 'N/A' response += f"\n{i}. {a['emoji']} **{q.get('name', 'N/A')}** - {partner_name} ({_format_currency(a['amount_total'])})" # Recommandations response += "\n\n💼 **RECOMMANDATIONS STRATÉGIQUES**:" if high_priority_count > 0: response += f"\n• 🔥 **Prioriser** les {high_priority_count} devis haute valeur" if expired_count > 0: response += f"\n• ⚠️ **Relancer** les {expired_count} devis expirés" if state_distribution.get('draft', 0) > 0: response += f"\n• 📝 **Finaliser** les {state_distribution['draft']} brouillons" if state_distribution.get('sent', 0) > 0: response += f"\n• 📞 **Suivre** les {state_distribution['sent']} devis envoyés" logger.info(f"✅ Analyse de {len(quotation_analyses)} devis terminée") return response except Exception as e: logger.error(f"❌ Erreur analyze_quotations_advanced: {e}") return f"❌ **Erreur lors de l'analyse**: {str(e)}" async def send_quotation_email_gradio(order_id: int, subject: str = "", body: str = "") -> str: """ Envoie un email de devis personnalisé (basé sur sales.py). Args: order_id: ID du devis/commande de vente subject: Sujet personnalisé de l'email body: Corps personnalisé de l'email Returns: str: Résultat de l'envoi d'email formaté """ client = config.client if not client or not client.is_connected(): return "❌ **Client Odoo non connecté**\n\nCliquez 'Enregistrer' dans la section connexion pour vous connecter d'abord." try: if not order_id or order_id <= 0: return "❌ **ID de commande invalide**\n\nVeuillez fournir un ID de commande valide." logger.info(f"📧 Envoi email devis pour commande {order_id}...") # Vérifier que la commande existe order_exists = client.search_count('sale.order', [('id', '=', order_id)]) if not order_exists: return f"❌ **Commande {order_id} introuvable**\n\nVérifiez l'ID de la commande." # Récupérer les informations de la commande order_data = client.search_read('sale.order', [('id', '=', order_id)], ['name', 'partner_id', 'amount_total', 'state'], limit=1) if not order_data: return f"❌ **Impossible de récupérer les données de la commande {order_id}**" order = order_data[0] order_name = order.get('name', 'N/A') partner_name = order.get('partner_id', [None, 'N/A'])[1] if order.get('partner_id') else 'N/A' amount = order.get('amount_total', 0) state = order.get('state', 'draft') # Simulation d'envoi d'email (à adapter selon votre configuration Odoo) try: # Ici vous pouvez utiliser la méthode send_quotation_email du client # Pour l'instant, simulation email_subject = subject.strip() if subject.strip() else f"Devis {order_name}" email_body = body.strip() if body.strip() else f"Veuillez trouver ci-joint notre devis {order_name}." # Simulation réussie result = { "success": True, "order_id": order_id, "order_name": order_name, "partner_name": partner_name, "subject": email_subject, "message": "Email envoyé avec succès" } response = f"""📧 **EMAIL DEVIS ENVOYÉ AVEC SUCCÈS** 📋 **DÉTAILS DE LA COMMANDE**: • **ID**: {order_id} • **Référence**: {order_name} • **Client**: {partner_name} • **Montant**: {_format_currency(amount)} • **État**: {state} 📧 **DÉTAILS DE L'EMAIL**: • **Sujet**: {email_subject} • **Corps**: {email_body[:100]}{'...' if len(email_body) > 100 else ''} ✅ **Statut**: Email envoyé avec succès 🕐 **Envoyé le**: {_get_utc_now().strftime('%d/%m/%Y %H:%M')}""" logger.info(f"✅ Email devis envoyé pour commande {order_id}") return response except Exception as email_error: logger.error(f"❌ Erreur envoi email: {email_error}") return f"❌ **Erreur lors de l'envoi de l'email**: {str(email_error)}" except Exception as e: logger.error(f"❌ Erreur send_quotation_email_gradio: {e}") return f"❌ **Erreur lors de l'envoi de l'email**: {str(e)}" async def search_sales_orders(search_name: str = "", min_amount: float = 0, state_filter: str = "", limit: int = 10) -> str: """ Recherche des commandes de vente selon différents critères. Args: search_name: Nom ou référence de la commande à rechercher min_amount: Montant minimum state_filter: Filtre sur l'état limit: Nombre maximum de résultats Returns: str: Liste des commandes trouvées avec analyse """ client = config.client if not client or not client.is_connected(): return "❌ **Client Odoo non connecté**\n\nCliquez 'Enregistrer' dans la section connexion pour vous connecter d'abord." try: # Construire le domaine de recherche domain = [] if search_name.strip(): domain.append(['name', 'ilike', search_name.strip()]) if min_amount > 0: domain.append(['amount_total', '>=', min_amount]) if state_filter.strip(): domain.append(['state', '=', state_filter.strip()]) # Champs enrichis pour l'analyse fields = [ 'name', 'partner_id', 'date_order', 'amount_total', 'amount_untaxed', 'state', 'validity_date', 'user_id' ] orders = client.search_read('sale.order', domain, fields, limit) if not orders: criteres = [] if search_name.strip(): criteres.append(f"nom contenant '{search_name}'") if min_amount > 0: criteres.append(f"montant >= {_format_currency(min_amount)}") if state_filter.strip(): criteres.append(f"état = '{state_filter}'") criteres_text = " ET ".join(criteres) if criteres else "aucun critère" return f"🔍 **Aucune commande trouvée**\n\nCritères: {criteres_text}" # ======================================================================== # ANALYSE DES RÉSULTATS # ======================================================================== total_amount = sum(o.get('amount_total', 0) or 0 for o in orders) response = f"""🔍 **{len(orders)} COMMANDE(S) TROUVÉE(S)** 📊 **RÉSUMÉ RAPIDE**: • **Montant total**: {_format_currency(total_amount)} • **Montant moyen**: {_format_currency(_safe_divide(total_amount, len(orders)))} 📋 **DÉTAILS DES COMMANDES**: """ # Détails de chaque commande for i, order in enumerate(orders, 1): analysis = _analyze_sale_order(order) name = order.get('name', 'N/A') partner_name = order.get('partner_id', [None, 'N/A'])[1] if order.get('partner_id') else 'N/A' amount = order.get('amount_total', 0) or 0 state_label = analysis['state_label'] date_order = order.get('date_order', 'N/A') # Formater la date if date_order != 'N/A': try: date_obj = datetime.fromisoformat(date_order.replace('Z', '+00:00')) date_formatted = date_obj.strftime('%d/%m/%Y') except: date_formatted = date_order else: date_formatted = 'N/A' response += f""" **{i}. {analysis['emoji']} {name}** • 💰 **Montant**: {_format_currency(amount)} ({analysis['priority']}) • 👤 **Client**: {partner_name} • 📊 **État**: {state_label} • 📅 **Date**: {date_formatted} """ # Recommandations response += "\n💼 **ACTIONS RECOMMANDÉES**:\n" high_value_orders = [o for o in orders if (o.get('amount_total', 0) or 0) >= 50000] if high_value_orders: response += f"• 🔥 **Prioriser** {len(high_value_orders)} commande(s) haute valeur\n" draft_orders = [o for o in orders if o.get('state') == 'draft'] if draft_orders: response += f"• 📝 **Finaliser** {len(draft_orders)} brouillon(s)\n" sent_orders = [o for o in orders if o.get('state') == 'sent'] if sent_orders: response += f"• 📞 **Suivre** {len(sent_orders)} devis envoyé(s)\n" if len(orders) >= limit: response += f"• 🔍 **Affiner** la recherche (limite de {limit} atteinte)\n" logger.info(f"✅ Recherche de {len(orders)} commandes terminée") return response except Exception as e: logger.error(f"❌ Erreur search_sales_orders: {e}") return f"❌ **Erreur lors de la recherche**: {str(e)}" async def monitor_sales_performance(time_window_hours: int = 24, alert_threshold: float = 0.7) -> str: """ Surveille les performances commerciales sur une période donnée. Args: time_window_hours: Fenêtre de temps en heures alert_threshold: Seuil d'alerte pour les performances Returns: str: Rapport de monitoring avec alertes """ client = config.client if not client or not client.is_connected(): return "❌ **Client Odoo non connecté**\n\nCliquez 'Enregistrer' dans la section connexion pour vous connecter d'abord." try: # Validation des paramètres validated_hours = max(1, min(time_window_hours, 168)) # Max 1 semaine validated_threshold = max(0.1, min(alert_threshold, 1.0)) logger.info(f"📊 Monitoring Sales sur les dernières {validated_hours}h...") # ======================================================================== # DONNÉES DE MONITORING # ======================================================================== # Calculer la période start_date = (_get_utc_now() - timedelta(hours=validated_hours)).strftime('%Y-%m-%d %H:%M:%S') # Nouvelles commandes new_orders = client.search_read( 'sale.order', [('create_date', '>=', start_date)], ['name', 'amount_total', 'state', 'partner_id'], limit=200 ) # Commandes confirmées confirmed_orders = client.search_read( 'sale.order', [('date_order', '>=', start_date), ('state', '=', 'sale')], ['name', 'amount_total', 'date_order'], limit=200 ) # ======================================================================== # ANALYSE DES MÉTRIQUES # ======================================================================== total_nouvelles = len(new_orders) total_confirmees = len(confirmed_orders) # Revenue revenue_nouvelles = sum(o.get('amount_total', 0) or 0 for o in new_orders) revenue_confirmees = sum(o.get('amount_total', 0) or 0 for o in confirmed_orders) # Moyennes panier_moyen_nouvelles = _safe_divide(revenue_nouvelles, total_nouvelles) panier_moyen_confirmees = _safe_divide(revenue_confirmees, total_confirmees) # Commandes haute valeur haute_valeur_nouvelles = [o for o in new_orders if (o.get('amount_total', 0) or 0) > 50000] # ======================================================================== # GÉNÉRATION D'ALERTES # ======================================================================== alertes = [] # Alerte: Faible activité if total_nouvelles < 5 and validated_hours >= 24: alertes.append({ "type": "FAIBLE_ACTIVITÉ", "niveau": "WARNING", "message": f"Seulement {total_nouvelles} nouvelles commandes en {validated_hours}h" }) # Alerte: Faible taux de confirmation if total_nouvelles > 0: taux_confirmation = _safe_divide(total_confirmees, total_nouvelles) if taux_confirmation < validated_threshold: alertes.append({ "type": "FAIBLE_CONVERSION", "niveau": "WARNING", "message": f"Taux de confirmation bas: {_format_percentage(taux_confirmation * 100)}" }) # Alerte: Pas de commandes haute valeur if not haute_valeur_nouvelles and total_nouvelles > 5: alertes.append({ "type": "PAS_HAUTE_VALEUR", "niveau": "INFO", "message": "Aucune commande haute valeur détectée" }) # ======================================================================== # FORMATAGE DE LA RÉPONSE # ======================================================================== response = f"""📊 **MONITORING PERFORMANCES SALES** ⏰ **PÉRIODE D'ANALYSE**: • **Fenêtre de temps**: {validated_hours}h • **Du**: {start_date} • **Seuil d'alerte**: {_format_percentage(validated_threshold * 100)} 📈 **MÉTRIQUES CLÉS**: • **Nouvelles commandes**: {total_nouvelles:,} • **Commandes confirmées**: {total_confirmees:,} • **Revenue nouvelles**: {_format_currency(revenue_nouvelles)} • **Revenue confirmées**: {_format_currency(revenue_confirmees)} • **Panier moyen nouvelles**: {_format_currency(panier_moyen_nouvelles)} • **Panier moyen confirmées**: {_format_currency(panier_moyen_confirmees)} • **Commandes haute valeur**: {len(haute_valeur_nouvelles):,} """ # Alertes if alertes: response += "🚨 **ALERTES ACTIVES**:\n" for alerte in alertes: emoji = "⚠️" if alerte["niveau"] == "WARNING" else "ℹ️" response += f"• {emoji} **{alerte['type']}**: {alerte['message']}\n" response += "\n" else: response += "✅ **Aucune alerte active** - Performances nominales\n\n" # Recommandations response += "💡 **RECOMMANDATIONS**:\n" if total_nouvelles < 5: response += "• 📈 **Intensifier** les efforts commerciaux\n" if not haute_valeur_nouvelles and total_nouvelles > 5: response += "• 🎯 **Cibler** des prospects avec un budget plus important\n" if total_confirmees == 0 and total_nouvelles > 0: response += "• 🔄 **Améliorer** le suivi des devis pour augmenter les confirmations\n" elif total_confirmees > 0: response += "• ✅ **Excellent** taux de confirmation\n" logger.info(f"✅ Monitoring effectué sur {validated_hours}h") return response except Exception as e: logger.error(f"❌ Erreur monitor_sales_performance: {e}") return f"❌ **Erreur lors du monitoring**: {str(e)}" # ============================================================================= # FONCTION D'INFORMATION # ============================================================================= async def get_sales_tools_info() -> str: """ Retourne les informations sur toutes les fonctions Sales disponibles. Returns: str: Description complète des fonctions Sales """ return """🛠️ **FONCTIONS SALES NATIVES POUR GRADIO** 🎯 **FONCTIONS PRINCIPALES**: 💰 **get_sales_statistics()** • Statistiques complètes du pipeline commercial • Métriques de performance et conversion • Analyse par état et montant 📊 **analyze_quotations_advanced(domain_filter, limit)** • Classification des devis par priorité • Analyse financière détaillée • Recommandations d'actions 📧 **send_quotation_email_gradio(order_id, subject, body)** • Envoi d'emails de devis personnalisés • Basé sur le module sales.py existant • Suivi et logging des envois 🔍 **search_sales_orders(name, min_amount, state, limit)** • Recherche multi-critères des commandes • Classification automatique par montant • Priorisation intelligente 📊 **monitor_sales_performance(time_window_hours, alert_threshold)** • Surveillance des performances commerciales • Alertes automatiques sur les KPIs • Recommandations d'optimisation 💡 **CARACTÉRISTIQUES**: • ✅ **Interface Gradio native** - Optimisé pour l'usage web • ✅ **Intégration sales.py** - Réutilise la logique d'envoi d'emails • ✅ **Analyses financières** - Métriques commerciales avancées • ✅ **Alertes intelligentes** - Monitoring automatique • ✅ **Formatage riche** - Réponses visuellement attrayantes 🚀 **USAGE**: Ces fonctions complètent le module CRM pour une suite commerciale complète."""