turf_saas/leadhunter_scorer.py

#!/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']}")