Python Code Style Review — cultiva-lead-scorer

# lead_scorer.py
import pandas as pd
import numpy as np
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
import os, sys   # ❌ import múltiple en una línea
from typing import *   # ❌ wildcard import

API_url = "https://api.cultivaia.com/leads"   # ❌ naming
maxRetries = 3   # ❌ camelCase en constante

class lead_model:   # ❌ snake_case en clase
    def __init__(self, mdl_path, thresh=0.5):
        # ❌ sin type annotations
        self.mdl_path = mdl_path
        self.thresh = thresh
        self.mdl = None

    def load(self):
        import pickle   # ❌ import tardío
        with open(self.mdl_path, 'rb') as f:
            self.mdl = pickle.load(f)

    def score(self, X, ret_proba=False):
        if self.mdl == None:   # ❌ usar is None
            raise Exception("Model not loaded")  # ❌ genérico
        p = self.mdl.predict_proba(X)[:, 1]
        if ret_proba == True:   # ❌ comparación redundante
            return p
        return (p >= self.thresh).astype(int)

def process_leads(leads_list, output_p, model, skip_err=False):
    # ❌ sin docstring, sin type hints
    res = []
    for l in leads_list:
        try:
            score = model.score(pd.DataFrame([l]))
            res.append({'id': l['id'], 'score': int(score[0])})
        except Exception as e:
            if skip_err == False:   # ❌
                raise e
    pd.DataFrame(res).to_csv(output_p, index=False)
    return res

"""Módulo de puntuación de leads para CULTIVA IA.

Integra modelos de ML para clasificar leads de marketing
en tiempo real con soporte de procesamiento por lotes.
"""
# Stdlib
import pickle   # ✓ import al inicio
from pathlib import Path
from collections.abc import Callable

# Terceros
import numpy as np
import pandas as pd

API_BASE_URL = "https://api.cultivaia.com/leads"   # ✓
MAX_RETRIES = 3   # ✓ SCREAMING_SNAKE_CASE


class ModelNotLoadedError(RuntimeError):   # ✓ excepción específica
    """Se lanza al intentar usar el modelo antes de cargarlo."""


class LeadScoringModel:   # ✓ PascalCase
    """Wrapper de modelo ML para puntuación de leads en producción.

    Attributes:
        model_path: Ruta al archivo pickle del modelo entrenado.
        threshold: Umbral de decisión para clasificación binaria.
    """

    def __init__(self, model_path: Path, threshold: float = 0.5) -> None:
        """Inicializa el modelo de puntuación.

        Args:
            model_path: Ruta al .pkl del modelo entrenado.
            threshold: Probabilidad mínima para score=1. Default 0.5.
        """
        self.model_path = model_path
        self.threshold = threshold
        self._model: object | None = None

    def load(self) -> None:
        """Carga el modelo desde disco."""
        with open(self.model_path, "rb") as f:
            self._model = pickle.load(f)

    def score(self, features: pd.DataFrame, return_proba: bool = False) -> np.ndarray:
        """Puntúa un DataFrame de features.

        Args:
            features: DataFrame con las features del lead.
            return_proba: Si True, retorna probabilidades crudas.

        Returns:
            Array de probabilidades (float) o etiquetas binarias (int).

        Raises:
            ModelNotLoadedError: Si no se ha llamado a load() antes.
        """
        if self._model is None:   # ✓ is None
            raise ModelNotLoadedError("Llama a load() antes de score()")  # ✓
        probas = self._model.predict_proba(features)[:, 1]
        if return_proba:   # ✓ sin == True
            return probas
        return (probas >= self.threshold).astype(int)


def process_leads(   # ✓ con type hints y docstring
    leads: list[dict],
    output_path: Path,
    model: LeadScoringModel,
    on_error: Callable[[Exception], None] | None = None,
) -> list[dict]:
    """Puntúa un lote de leads y los guarda en CSV.

    Args:
        leads: Lista de dicts con datos de cada lead (requiere 'id').
        output_path: Ruta de destino del CSV de resultados.
        model: Instancia de LeadScoringModel ya cargada.
        on_error: Callback opcional para errores por lead. Si None,
            los errores se propagan.

    Returns:
        Lista de dicts {id, score} con los resultados.

    Raises:
        ModelNotLoadedError: Si el modelo no ha sido cargado.
    """
    results: list[dict] = []
    for lead in leads:
        try:
            score = model.score(pd.DataFrame([lead]))
            results.append({"id": lead["id"], "score": int(score[0])})
        except Exception as exc:
            if on_error is not None:   # ✓ callback vs flag booleano
                on_error(exc)
            else:
                raise
    pd.DataFrame(results).to_csv(output_path, index=False)
    return results