#!/usr/bin/env python3 """ H3R7Tech — LeadHunter Scorer ================================ Moteur de scoring des leads restaurants MEL. Critères (ordre de priorité métier) : 1. [+3] Site web absent ← CRITIQUE : raison d'être du produit 2. [+2] Nombre d'avis élevé (≥ 50) : forte activité = bon prospect de vente 3. [+2] Note Google élevée (≥ 4.0) : établissement sérieux 4. [+1] Téléphone présent : facilite la prise de contact 5. [-1] Note faible (< 3.0) : risque reputationnel pour la prestation web Score maximum théorique : 8 Score minimum : 0 (leads avec site web ne doivent pas passer ici) Auteur: H3R7Tech Backend Engineer Issue: HRT-66 """ import logging from logging.handlers import RotatingFileHandler # ─── Logging ──────────────────────────────────────────────────────────────── logger = logging.getLogger("leadhunter.scorer") _handler = RotatingFileHandler( "/home/h3r7/leadhunter.log", maxBytes=5 * 1024 * 1024, backupCount=3, ) _handler.setFormatter( logging.Formatter("%(asctime)s %(levelname)-8s %(name)s — %(message)s") ) logger.setLevel(logging.INFO) if not logger.handlers: logger.addHandler(_handler) logger.addHandler(logging.StreamHandler()) # ─── Scorer ────────────────────────────────────────────────────────────────── class LeadScorer: """ Calcule le score de priorité d'un lead. Le score sert à trier les leads dans le CRM : - Score élevé = prospect chaud (sans site + actif + bien noté) - Score faible = prospect froid (peut être ignoré ou traité en dernier) """ def _calculate_score(self, lead: dict) -> int: """ Calcule le score d'un lead. Args: lead: dict avec les champs normalisés du scraper (name, website, rating, reviews_count, phone, ...) Returns: Score entier (0–8) """ score = 0 # ── Critère 1 : site web absent [CRITIQUE — logique métier centrale] ── # C'est le critère n°1 : on cherche des restaurants SANS site web # pour leur proposer une création de site à 800–1500€. website = lead.get("website", "") if not website or not website.strip(): score += 3 logger.debug(f"{lead.get('name')}: +3 (site web absent)") else: # Si le lead a un site web, score = 0 immédiatement. # Ce cas ne devrait pas se produire (filtre scraper), # mais on reste défensif. logger.warning( f"{lead.get('name')}: site web présent ({website}), " "lead ignoré pour scoring." ) return 0 # ── Critère 2 : nombre d'avis élevé (≥ 50) ────────────────────────── reviews = lead.get("reviews_count") if reviews is not None: try: reviews = int(reviews) if reviews >= 50: score += 2 logger.debug(f"{lead.get('name')}: +2 (avis ≥ 50 : {reviews})") except (TypeError, ValueError) as e: logger.warning(f"reviews_count invalide pour {lead.get('name')}: {e}") # ── Critère 3 : bonne note Google (≥ 4.0) ─────────────────────────── rating = lead.get("rating") if rating is not None: try: rating = float(rating) if rating >= 4.0: score += 2 logger.debug(f"{lead.get('name')}: +2 (note ≥ 4.0 : {rating})") elif rating < 3.0: score -= 1 logger.debug(f"{lead.get('name')}: -1 (note < 3.0 : {rating})") except (TypeError, ValueError) as e: logger.warning(f"rating invalide pour {lead.get('name')}: {e}") # ── Critère 4 : téléphone présent ──────────────────────────────────── phone = lead.get("phone", "") if phone and phone.strip(): score += 1 logger.debug(f"{lead.get('name')}: +1 (téléphone présent)") # Plancher à 0 score = max(0, score) logger.info(f"Score calculé pour '{lead.get('name')}' : {score}/8") return score def score_lead(self, lead: dict) -> dict: """ Enrichit un lead avec son score. Args: lead: dict normalisé du scraper. Returns: Même dict avec le champ 'score' ajouté/mis à jour. """ lead = dict(lead) # copie défensive lead["score"] = self._calculate_score(lead) return lead def score_leads(self, leads: list[dict]) -> list[dict]: """ Score et trie une liste de leads (score décroissant). Args: leads: liste de dicts normalisés. Returns: Liste triée par score décroissant. """ scored = [self.score_lead(lead) for lead in leads] scored.sort(key=lambda l: l.get("score", 0), reverse=True) logger.info( f"score_leads terminé : {len(scored)} leads scorés. " f"Score max = {scored[0]['score'] if scored else 0}, " f"Score min = {scored[-1]['score'] if scored else 0}" ) return scored # ─── CLI (debug) ───────────────────────────────────────────────────────────── if __name__ == "__main__": # Exemple de test rapide sans appel API test_leads = [ { "name": "Restaurant A", "website": "", "rating": 4.5, "reviews_count": 120, "phone": "+33 3 20 00 00 01", }, { "name": "Restaurant B", "website": "", "rating": 3.8, "reviews_count": 30, "phone": "", }, { "name": "Café C", "website": "", "rating": 2.5, "reviews_count": 5, "phone": "+33 3 20 00 00 03", }, { "name": "Bar D avec site", "website": "https://bar-d.fr", "rating": 4.2, "reviews_count": 80, "phone": "+33 3 20 00 00 04", }, ] scorer = LeadScorer() results = scorer.score_leads(test_leads) print("\n=== Résultats scoring ===") for r in results: print(f" [{r['score']:2d}/8] {r['name']}")