CRM Pipeline и автоматизация

Цель

CRM Pipeline превращает хаотичный поток лидов из Problem Detection Engine в предсказуемую машину продаж. Автоматизация устраняет ручную работу, а скоринг гарантирует, что лучшие лиды получают приоритет.

---

Стадии Pipeline

graph LR
    A["🔍 Discovery\n(авто)"] --> B["✅ Qualified\n(авто)"]
    B --> C["✉️ Outreach\n(авто)"]
    C --> D["💬 Response\n(ручное)"]
    D --> E["📑 Audit\n(авто+ручное)"]
    E --> F["💼 Proposal\n(ручное)"]
    F --> G["🤝 Negotiation\n(ручное)"]
    G --> H{"Результат"}
    H -->|"Да"| I["🏆 Won"]
    H -->|"Нет"| J["❌ Lost"]
    J --> K["📌 Nurturing"]
    K -.->|"через 30–90 дн"| C

    style A fill:#e3f2fd
    style B fill:#bbdefb
    style C fill:#fff3e0
    style D fill:#ffe0b2
    style E fill:#e8f5e9
    style F fill:#c8e6c9
    style G fill:#f3e5f5
    style I fill:#4caf50,color:#fff
    style J fill:#f44336,color:#fff
    style K fill:#9e9e9e,color:#fff

Описание стадий

#	Стадия	Тип	Триггер входа	Триггер выхода	SLA
1	Discovery	Авто	Category Scanner нашёл проблемную карточку	PotentialScore рассчитан	—
2	Qualified	Авто	Score ≥ 30, контакт найден	Контакт верифицирован	—
3	Outreach	Авто	Compliance OK, канал выбран	Сообщение отправлено	24ч
4	Response	Ручное	Лид ответил	Квалификация ответа	2ч
5	Audit	Авто+ручное	Лид запросил аудит	PDF-отчёт отправлен	48ч
6	Proposal	Ручное	Аудит просмотрен, лид заинтересован	КП отправлено	24ч
7	Negotiation	Ручное	КП обсуждается	Финальное решение	5 дн
8	Won	Ручное	Договор подписан	—	—
9	Lost	Ручное	Отказ на любой стадии	—	—
10	Nurturing	Авто	Lost → ожидание re-engagement	Повторный аутрич через 30–90 дн	—

---

Lead Scoring

Комбинированный скоринг

LeadScore = PotentialScore × 0.6 + EngagementScore × 0.4

PotentialScore (0–100): из Problem Detection Engine
  — Качество карточки, масштаб проблемы, упущенная выручка

EngagementScore (0–100): сигналы вовлечённости
  — Ответил на сообщение: +30
  — Запросил аудит: +25
  — Открыл PDF: +15
  — Позвонил/написал первым: +20
  — Задал вопрос о цене: +10
  — Запросил кейсы: +5
  — Несколько SKU (>5): +10
  — Оборот > 5M/мес: +10

Python-реализация

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional


class PipelineStage(Enum):
    DISCOVERY = "discovery"
    QUALIFIED = "qualified"
    OUTREACH = "outreach"
    RESPONSE = "response"
    AUDIT = "audit"
    PROPOSAL = "proposal"
    NEGOTIATION = "negotiation"
    WON = "won"
    LOST = "lost"
    NURTURING = "nurturing"


class LeadTemperature(Enum):
    HOT = "hot"         # Score ≥ 70, требует немедленного внимания
    WARM = "warm"       # Score 50–69, стандартная обработка
    COOL = "cool"       # Score 30–49, низкий приоритет
    COLD = "cold"       # Score < 30, nurturing


@dataclass
class Lead:
    """Лид в CRM pipeline."""
    id: int
    seller_id: str
    seller_name: str
    marketplace: str                    # wb | ozon
    contact_telegram: str = ""
    contact_email: str = ""
    contact_phone: str = ""
    contact_wb_dm: bool = False

    # Скоринг
    potential_score: float = 0.0        # из Problem Detection (0–100)
    engagement_score: float = 0.0       # из вовлечённости (0–100)
    lead_score: float = 0.0            # комбинированный (0–100)
    temperature: LeadTemperature = LeadTemperature.COLD

    # Pipeline
    stage: PipelineStage = PipelineStage.DISCOVERY
    stage_entered_at: datetime = field(default_factory=datetime.utcnow)
    assigned_manager: Optional[str] = None

    # Данные
    top_skus: list[int] = field(default_factory=list)
    total_lost_revenue: float = 0.0
    total_sku_count: int = 0
    estimated_monthly_revenue: float = 0.0

    # Engagement history
    outreach_count: int = 0
    response_count: int = 0
    audit_requested: bool = False
    audit_sent: bool = False
    audit_opened: bool = False
    proposal_sent: bool = False

    # Timestamps
    created_at: datetime = field(default_factory=datetime.utcnow)
    last_activity_at: datetime = field(default_factory=datetime.utcnow)
    won_at: Optional[datetime] = None
    lost_at: Optional[datetime] = None
    lost_reason: str = ""


class LeadScorer:
    """Рассчитывает комбинированный LeadScore."""

    ENGAGEMENT_SIGNALS = {
        "responded": 30,
        "audit_requested": 25,
        "audit_opened": 15,
        "initiated_contact": 20,
        "asked_price": 10,
        "asked_cases": 5,
        "multiple_skus": 10,       # > 5 SKU
        "high_revenue": 10,        # оборот > 5M/мес
    }

    POTENTIAL_WEIGHT = 0.6
    ENGAGEMENT_WEIGHT = 0.4

    def calculate_score(self, lead: Lead) -> tuple[float, LeadTemperature]:
        """Рассчитывает LeadScore и температуру."""

        # Engagement score
        engagement = 0.0
        if lead.response_count > 0:
            engagement += self.ENGAGEMENT_SIGNALS["responded"]
        if lead.audit_requested:
            engagement += self.ENGAGEMENT_SIGNALS["audit_requested"]
        if lead.audit_opened:
            engagement += self.ENGAGEMENT_SIGNALS["audit_opened"]
        if lead.total_sku_count > 5:
            engagement += self.ENGAGEMENT_SIGNALS["multiple_skus"]
        if lead.estimated_monthly_revenue > 5_000_000:
            engagement += self.ENGAGEMENT_SIGNALS["high_revenue"]

        engagement = min(engagement, 100.0)

        # Комбинированный score
        lead_score = (
            lead.potential_score * self.POTENTIAL_WEIGHT
            + engagement * self.ENGAGEMENT_WEIGHT
        )

        # Температура
        if lead_score >= 70:
            temp = LeadTemperature.HOT
        elif lead_score >= 50:
            temp = LeadTemperature.WARM
        elif lead_score >= 30:
            temp = LeadTemperature.COOL
        else:
            temp = LeadTemperature.COLD

        return round(lead_score, 1), temp

    def rescore_lead(self, lead: Lead) -> Lead:
        """Пересчитывает score и обновляет лида."""
        lead.engagement_score = self._calc_engagement(lead)
        lead.lead_score, lead.temperature = self.calculate_score(lead)
        return lead

    def _calc_engagement(self, lead: Lead) -> float:
        engagement = 0.0
        if lead.response_count > 0:
            engagement += 30
        if lead.audit_requested:
            engagement += 25
        if lead.audit_opened:
            engagement += 15
        if lead.total_sku_count > 5:
            engagement += 10
        if lead.estimated_monthly_revenue > 5_000_000:
            engagement += 10
        return min(engagement, 100.0)

---

Database Schema

Таблицы лидогенерации

from sqlalchemy import (
    Column, Integer, String, Float, Boolean, DateTime,
    JSON, ForeignKey, Text, Index, Enum as SAEnum,
)
from sqlalchemy.orm import declarative_base, relationship
from datetime import datetime

Base = declarative_base()


class LeadDB(Base):
    """Основная таблица лидов."""
    __tablename__ = "leads"

    id = Column(Integer, primary_key=True, autoincrement=True)
    seller_id = Column(String(50), nullable=False, index=True)
    seller_name = Column(String(200), default="")
    marketplace = Column(String(10), nullable=False)  # wb | ozon

    # Контакты
    contact_telegram = Column(String(100), default="")
    contact_email = Column(String(200), default="")
    contact_phone = Column(String(20), default="")
    contact_wb_dm = Column(Boolean, default=False)

    # Скоринг
    potential_score = Column(Float, default=0.0, index=True)
    engagement_score = Column(Float, default=0.0)
    lead_score = Column(Float, default=0.0, index=True)
    temperature = Column(String(10), default="cold", index=True)

    # Pipeline
    stage = Column(String(20), default="discovery", index=True)
    stage_entered_at = Column(DateTime, default=datetime.utcnow)
    assigned_manager = Column(String(100))

    # Бизнес-данные
    top_skus = Column(JSON, default=list)
    total_lost_revenue = Column(Float, default=0.0)
    total_sku_count = Column(Integer, default=0)
    estimated_monthly_revenue = Column(Float, default=0.0)
    primary_pattern = Column(String(30), default="")

    # Engagement
    outreach_count = Column(Integer, default=0)
    response_count = Column(Integer, default=0)
    audit_requested = Column(Boolean, default=False)
    audit_sent = Column(Boolean, default=False)
    audit_opened = Column(Boolean, default=False)
    proposal_sent = Column(Boolean, default=False)

    # Timestamps
    created_at = Column(DateTime, default=datetime.utcnow, index=True)
    last_activity_at = Column(DateTime, default=datetime.utcnow)
    won_at = Column(DateTime)
    lost_at = Column(DateTime)
    lost_reason = Column(Text, default="")

    # Мета
    source_scan_id = Column(String(50))  # ID скана, где нашли лида
    notes = Column(Text, default="")
    metadata = Column(JSON, default=dict)

    # Relations
    outreach_logs = relationship("OutreachLogDB", back_populates="lead")
    audit_requests = relationship("AuditRequestDB", back_populates="lead")
    deals = relationship("DealDB", back_populates="lead")

    __table_args__ = (
        Index("ix_leads_stage_score", "stage", "lead_score"),
        Index("ix_leads_temperature_created", "temperature", "created_at"),
    )


class OutreachLogDB(Base):
    """Лог аутрич-сообщений."""
    __tablename__ = "outreach_log"

    id = Column(Integer, primary_key=True, autoincrement=True)
    lead_id = Column(Integer, ForeignKey("leads.id"), nullable=False, index=True)
    channel = Column(String(20), nullable=False)
    template = Column(String(50), nullable=False)
    variant = Column(String(5), default="A")
    message_body = Column(Text, nullable=False)
    subject = Column(String(200))
    data_points_used = Column(Integer, default=0)

    sent_at = Column(DateTime, default=datetime.utcnow)
    delivered = Column(Boolean, default=False)
    opened = Column(Boolean, default=False)
    opened_at = Column(DateTime)
    responded = Column(Boolean, default=False)
    response_at = Column(DateTime)
    response_text = Column(Text)
    response_sentiment = Column(String(20))  # positive, neutral, negative

    ab_test_name = Column(String(50))
    followup_step = Column(Integer, default=0)  # 0=initial, 1=day3, 2=day7...
    metadata = Column(JSON, default=dict)

    lead = relationship("LeadDB", back_populates="outreach_logs")


class AuditRequestDB(Base):
    """Запросы на аудит."""
    __tablename__ = "audit_requests"

    id = Column(Integer, primary_key=True, autoincrement=True)
    lead_id = Column(Integer, ForeignKey("leads.id"), nullable=False, index=True)
    report_id = Column(String(50), unique=True)

    skus = Column(JSON, nullable=False)  # [sku1, sku2, sku3]
    marketplace = Column(String(10), nullable=False)

    status = Column(String(20), default="pending")
    # pending, generating, generated, sent, opened, feedback

    generated_at = Column(DateTime)
    sent_at = Column(DateTime)
    opened_at = Column(DateTime)
    feedback = Column(Text)

    report_path = Column(String(500))  # путь к PDF
    report_data = Column(JSON)         # данные аудита для аналитики

    total_lost_revenue = Column(Float, default=0.0)
    recommended_package = Column(String(50))
    estimated_roi = Column(Float, default=0.0)

    created_at = Column(DateTime, default=datetime.utcnow)
    metadata = Column(JSON, default=dict)

    lead = relationship("LeadDB", back_populates="audit_requests")


class DealDB(Base):
    """Сделки."""
    __tablename__ = "deals"

    id = Column(Integer, primary_key=True, autoincrement=True)
    lead_id = Column(Integer, ForeignKey("leads.id"), nullable=False, index=True)
    deal_number = Column(String(20), unique=True)

    package = Column(String(50), nullable=False)  # AUDIT, OPTIMIZATION, GROWTH, SCALE
    monthly_value = Column(Float, nullable=False)
    contract_months = Column(Integer, default=3)
    total_value = Column(Float, default=0.0)

    status = Column(String(20), default="draft")
    # draft, proposal_sent, negotiation, signed, active, churned, completed

    signed_at = Column(DateTime)
    start_date = Column(DateTime)
    end_date = Column(DateTime)
    churned_at = Column(DateTime)
    churn_reason = Column(Text)

    # Platrum integration
    platrum_task_id = Column(Integer)  # ID задачи в Platrum

    created_at = Column(DateTime, default=datetime.utcnow)
    metadata = Column(JSON, default=dict)

    lead = relationship("LeadDB", back_populates="deals")

---

Pipeline Manager

Управление переходами между стадиями

from datetime import datetime, timedelta
import logging

logger = logging.getLogger(__name__)


class LeadPipeline:
    """
    Управляет движением лидов по pipeline.

    Правила перехода:
    - discovery → qualified: Score ≥ 30 + контакт найден
    - qualified → outreach: Compliance OK
    - outreach → response: Лид ответил
    - response → audit: Лид запросил аудит
    - audit → proposal: Аудит отправлен и просмотрен
    - proposal → negotiation: КП обсуждается
    - negotiation → won/lost: Финальное решение
    """

    ALLOWED_TRANSITIONS = {
        "discovery": ["qualified"],
        "qualified": ["outreach", "lost"],
        "outreach": ["response", "lost"],
        "response": ["audit", "proposal", "lost"],
        "audit": ["proposal", "lost"],
        "proposal": ["negotiation", "lost"],
        "negotiation": ["won", "lost"],
        "lost": ["nurturing"],
        "nurturing": ["outreach"],
        "won": [],  # финальная стадия
    }

    SLA_HOURS = {
        "outreach": 24,      # отправить аутрич за 24ч
        "response": 2,       # ответить на сообщение лида за 2ч
        "audit": 48,         # отправить аудит за 48ч
        "proposal": 24,      # отправить КП за 24ч
        "negotiation": 120,  # 5 дней на закрытие
    }

    MANAGER_ASSIGNMENT = {
        "hot": "senior_manager",      # HOT → старший менеджер
        "warm": "any_manager",        # WARM → любой свободный
        "cool": "auto_only",          # COOL → только автоматика
        "cold": "nurturing_bot",      # COLD → nurturing бот
    }

    def __init__(self, db_session, scorer: LeadScorer, notifier=None):
        self.db = db_session
        self.scorer = scorer
        self.notifier = notifier  # Telegram-бот для уведомлений

    async def advance_stage(
        self,
        lead_id: int,
        new_stage: str,
        metadata: dict = None,
    ) -> Lead:
        """
        Перемещает лида на следующую стадию pipeline.

        Args:
            lead_id: ID лида
            new_stage: новая стадия
            metadata: доп. данные перехода

        Returns:
            обновлённый Lead
        """
        lead = await self._get_lead(lead_id)

        # Проверяем допустимость перехода
        allowed = self.ALLOWED_TRANSITIONS.get(lead.stage.value, [])
        if new_stage not in allowed:
            raise ValueError(
                f"Cannot transition from {lead.stage.value} to {new_stage}. "
                f"Allowed: {allowed}"
            )

        old_stage = lead.stage.value
        lead.stage = PipelineStage(new_stage)
        lead.stage_entered_at = datetime.utcnow()
        lead.last_activity_at = datetime.utcnow()

        # Специфические действия при переходе
        if new_stage == "won":
            lead.won_at = datetime.utcnow()
            await self._on_deal_won(lead)
        elif new_stage == "lost":
            lead.lost_at = datetime.utcnow()
            lead.lost_reason = (metadata or {}).get("reason", "")

        # Пересчитываем score
        lead = self.scorer.rescore_lead(lead)

        # Назначаем менеджера при необходимости
        if new_stage in ("response", "proposal", "negotiation"):
            await self._ensure_manager_assigned(lead)

        # Сохраняем
        await self._save_lead(lead)

        # Уведомления
        if self.notifier:
            await self.notifier.notify_stage_change(
                lead, old_stage, new_stage
            )

        logger.info(
            f"Lead {lead_id} ({lead.seller_name}): "
            f"{old_stage} → {new_stage} (score: {lead.lead_score})"
        )

        return lead

    async def score_lead(self, lead_id: int) -> Lead:
        """Пересчитывает score лида."""
        lead = await self._get_lead(lead_id)
        lead = self.scorer.rescore_lead(lead)
        await self._save_lead(lead)
        return lead

    async def assign_to_manager(self, lead_id: int, manager_id: str = None):
        """Назначает менеджера на лида."""
        lead = await self._get_lead(lead_id)

        if manager_id:
            lead.assigned_manager = manager_id
        else:
            # Авто-назначение по температуре
            assignment_rule = self.MANAGER_ASSIGNMENT.get(
                lead.temperature.value, "auto_only"
            )
            if assignment_rule == "senior_manager":
                lead.assigned_manager = await self._get_least_loaded("senior")
            elif assignment_rule == "any_manager":
                lead.assigned_manager = await self._get_least_loaded("any")
            # auto_only и nurturing_bot — без менеджера

        await self._save_lead(lead)

    async def check_sla_violations(self) -> list[dict]:
        """Проверяет нарушения SLA по всем активным лидам."""
        violations = []
        now = datetime.utcnow()

        active_leads = await self._get_active_leads()
        for lead in active_leads:
            stage = lead.stage.value
            if stage in self.SLA_HOURS:
                hours_in_stage = (now - lead.stage_entered_at).total_seconds() / 3600
                sla_hours = self.SLA_HOURS[stage]

                if hours_in_stage > sla_hours:
                    violations.append({
                        "lead_id": lead.id,
                        "seller_name": lead.seller_name,
                        "stage": stage,
                        "sla_hours": sla_hours,
                        "actual_hours": round(hours_in_stage, 1),
                        "overdue_hours": round(hours_in_stage - sla_hours, 1),
                        "assigned_manager": lead.assigned_manager,
                        "temperature": lead.temperature.value,
                    })

        return sorted(violations, key=lambda v: v["overdue_hours"], reverse=True)

    async def _on_deal_won(self, lead: Lead):
        """Действия при закрытии сделки."""
        # 1. Создаём задачу в Platrum
        # (интеграция через Platrum Spider CLI)
        pass

    async def _ensure_manager_assigned(self, lead: Lead):
        """Убеждаемся, что менеджер назначен."""
        if not lead.assigned_manager:
            await self.assign_to_manager(lead.id)

    async def _get_lead(self, lead_id: int) -> Lead:
        raise NotImplementedError

    async def _save_lead(self, lead: Lead):
        raise NotImplementedError

    async def _get_active_leads(self) -> list[Lead]:
        raise NotImplementedError

    async def _get_least_loaded(self, manager_type: str) -> str:
        raise NotImplementedError

---

Правила автоматизации

Триггеры и действия

#	Триггер	Условие	Действие	Приоритет
1	Новый лид	Score ≥ 50, контакт есть	Авто-квалификация → Outreach	HIGH
2	Нет ответа 3 дня	stage=outreach, 3 дня без ответа	Follow-up #1	MEDIUM
3	Нет ответа 7 дней	stage=outreach, 7 дней без ответа	Follow-up #2 (кейс)	MEDIUM
4	Лид ответил	Входящее сообщение	Уведомить менеджера, stage→response	HIGH
5	Лид запросил аудит	Слово «аудит»/«да» в ответе	Генерировать аудит, stage→audit	HIGH
6	Аудит отправлен	PDF сгенерирован	Отправить PDF, запланировать follow-up	MEDIUM
7	SLA нарушен	Время на стадии > SLA	Уведомить менеджера + руководителя	HIGH
8	HOT лид создан	Temperature=HOT	Уведомить senior менеджера	CRITICAL
9	Deal won	Стадия → Won	Создать задачу в Platrum, поздравить	HIGH
10	Deal lost	Стадия → Lost	Запланировать nurturing через 30 дней	LOW

Реализация автоматизации

import asyncio
from dataclasses import dataclass
from typing import Callable, Awaitable


@dataclass
class AutomationRule:
    """Правило автоматизации."""
    name: str
    trigger: str                           # событие-триггер
    condition: Callable[..., bool]         # условие срабатывания
    action: Callable[..., Awaitable]       # действие
    priority: str                          # CRITICAL, HIGH, MEDIUM, LOW
    enabled: bool = True


class AutomationEngine:
    """Движок автоматизации pipeline."""

    def __init__(self, pipeline: LeadPipeline, notifier, audit_generator):
        self.pipeline = pipeline
        self.notifier = notifier
        self.audit_gen = audit_generator
        self.rules: list[AutomationRule] = self._init_rules()

    def _init_rules(self) -> list[AutomationRule]:
        return [
            AutomationRule(
                name="auto_qualify",
                trigger="lead_created",
                condition=lambda lead: (
                    lead.potential_score >= 50
                    and any([lead.contact_telegram, lead.contact_email,
                             lead.contact_phone, lead.contact_wb_dm])
                ),
                action=self._auto_qualify,
                priority="HIGH",
            ),
            AutomationRule(
                name="hot_lead_alert",
                trigger="lead_scored",
                condition=lambda lead: lead.temperature == LeadTemperature.HOT,
                action=self._alert_hot_lead,
                priority="CRITICAL",
            ),
            AutomationRule(
                name="auto_followup_day3",
                trigger="daily_check",
                condition=lambda lead: (
                    lead.stage == PipelineStage.OUTREACH
                    and lead.response_count == 0
                    and (datetime.utcnow() - lead.stage_entered_at).days >= 3
                    and lead.outreach_count == 1  # только после первого касания
                ),
                action=self._send_followup,
                priority="MEDIUM",
            ),
            AutomationRule(
                name="auto_generate_audit",
                trigger="audit_requested",
                condition=lambda lead: lead.audit_requested and not lead.audit_sent,
                action=self._generate_and_send_audit,
                priority="HIGH",
            ),
            AutomationRule(
                name="sla_violation_alert",
                trigger="hourly_check",
                condition=lambda violations: len(violations) > 0,
                action=self._alert_sla_violation,
                priority="HIGH",
            ),
            AutomationRule(
                name="deal_won_actions",
                trigger="stage_changed",
                condition=lambda lead, new_stage: new_stage == "won",
                action=self._on_deal_won,
                priority="HIGH",
            ),
            AutomationRule(
                name="schedule_nurturing",
                trigger="stage_changed",
                condition=lambda lead, new_stage: new_stage == "lost",
                action=self._schedule_nurturing,
                priority="LOW",
            ),
        ]

    async def _auto_qualify(self, lead: Lead):
        """Автоматическая квалификация лида."""
        await self.pipeline.advance_stage(lead.id, "qualified")
        await self.pipeline.advance_stage(lead.id, "outreach")
        # Аутрич отправляется через OutreachSystem

    async def _alert_hot_lead(self, lead: Lead):
        """Уведомление о горячем лиде."""
        await self.notifier.send_alert(
            channel="senior_manager",
            message=(
                f"🔥 HOT LEAD: {lead.seller_name}\n"
                f"Score: {lead.lead_score} | "
                f"Потери: {lead.total_lost_revenue:,.0f} ₽/мес\n"
                f"Marketplace: {lead.marketplace}\n"
                f"SKU: {len(lead.top_skus)} товаров\n"
                f"Контакт: {lead.contact_telegram or lead.contact_email}"
            ),
        )

    async def _send_followup(self, lead: Lead):
        """Автоматический follow-up."""
        # Делегируем в FollowUpScheduler
        pass

    async def _generate_and_send_audit(self, lead: Lead):
        """Генерирует и отправляет аудит."""
        report = await self.audit_gen.generate_audit(
            lead_id=lead.id,
            skus=lead.top_skus[:3],
            marketplace=lead.marketplace,
            seller_name=lead.seller_name,
        )
        # Рендерим PDF и отправляем
        pass

    async def _alert_sla_violation(self, violations: list[dict]):
        """Уведомление о нарушении SLA."""
        for v in violations:
            await self.notifier.send_alert(
                channel=v["assigned_manager"] or "sales_general",
                message=(
                    f"⚠️ SLA НАРУШЕН: {v['seller_name']}\n"
                    f"Стадия: {v['stage']} | "
                    f"Просрочка: {v['overdue_hours']:.1f}ч\n"
                    f"Temperature: {v['temperature']}"
                ),
            )

    async def _on_deal_won(self, lead: Lead):
        """Действия при закрытии сделки."""
        # 1. Создаём задачу в Platrum
        # python spider.py tasks create --name "Контент для {seller}" ...
        # 2. Уведомляем команду
        await self.notifier.send_alert(
            channel="sales_general",
            message=(
                f"🏆 СДЕЛКА ЗАКРЫТА: {lead.seller_name}\n"
                f"MRR: {lead.estimated_monthly_revenue:,.0f} ₽\n"
                f"Менеджер: {lead.assigned_manager}"
            ),
        )

    async def _schedule_nurturing(self, lead: Lead):
        """Планирует nurturing для lost лидов."""
        await self.pipeline.advance_stage(lead.id, "nurturing")
        # Планируем повторный аутрич через 30–90 дней

---

Telegram-бот для менеджеров

Уведомления

import asyncio
from aiogram import Bot, Dispatcher, types


class SalesNotifierBot:
    """Telegram-бот для уведомления менеджеров о лидах."""

    def __init__(self, token: str, manager_chat_ids: dict[str, int]):
        self.bot = Bot(token=token)
        self.managers = manager_chat_ids
        # {"senior_manager": 123456789, "manager_1": 987654321, ...}

    async def send_alert(self, channel: str, message: str):
        """Отправляет уведомление в нужный чат."""
        chat_id = self.managers.get(channel)
        if not chat_id:
            # Отправляем в общий канал продаж
            chat_id = self.managers.get("sales_general")

        if chat_id:
            await self.bot.send_message(
                chat_id=chat_id,
                text=message,
                parse_mode="HTML",
            )

    async def notify_stage_change(
        self, lead: Lead, old_stage: str, new_stage: str
    ):
        """Уведомление о смене стадии."""
        emoji_map = {
            "discovery": "🔍",
            "qualified": "✅",
            "outreach": "✉️",
            "response": "💬",
            "audit": "📑",
            "proposal": "💼",
            "negotiation": "🤝",
            "won": "🏆",
            "lost": "❌",
            "nurturing": "📌",
        }

        msg = (
            f"{emoji_map.get(new_stage, '📋')} Pipeline Update\n\n"
            f"<b>{lead.seller_name}</b>\n"
            f"{old_stage} → <b>{new_stage}</b>\n"
            f"Score: {lead.lead_score} ({lead.temperature.value})\n"
            f"Потери: {lead.total_lost_revenue:,.0f} ₽/мес"
        )

        # HOT лиды — в личку senior менеджеру
        if lead.temperature == LeadTemperature.HOT:
            await self.send_alert("senior_manager", msg)

        # Все смены стадий — в общий канал
        await self.send_alert("sales_general", msg)

---

Интеграция с Platrum

Автосоздание задач при конвертации

async def create_platrum_task_for_client(lead: Lead, deal: DealDB):
    """Создаёт задачу в Platrum при конвертации лида в клиента."""

    task_name = f"Контент-продакшн: {lead.seller_name} ({lead.marketplace.upper()})"
    description = (
        f"Новый клиент из лидогенерации.\n\n"
        f"Бренд: {lead.seller_name}\n"
        f"Площадка: {lead.marketplace.upper()}\n"
        f"Пакет: {deal.package}\n"
        f"Сумма: {deal.monthly_value:,.0f} ₽/мес\n"
        f"SKU: {', '.join(str(s) for s in lead.top_skus)}\n"
        f"Контакт: {lead.contact_telegram or lead.contact_email}\n\n"
        f"PotentialScore: {lead.potential_score}\n"
        f"Упущенная выручка: {lead.total_lost_revenue:,.0f} ₽/мес\n"
    )

    # Через Platrum Spider CLI
    # python spider.py tasks create --name "..." --description "..." --real
    pass

---

Дашборд

Воронка продаж

╔══════════════════════════════════════════════════════════════╗
║  LEADGEN PIPELINE DASHBOARD                                  ║
║  Период: Март 2026                                           ║
╠══════════════════════════════════════════════════════════════╣
║                                                               ║
║  ВОРОНКА:                                                    ║
║  ════════                                                    ║
║  Discovery:    482  ████████████████████████████████████████  ║
║  Qualified:    312  ██████████████████████████                ║
║  Outreach:     245  █████████████████████                     ║
║  Response:      29  ███                                       ║
║  Audit:         17  ██                                        ║
║  Proposal:       8  █                                         ║
║  Negotiation:    4  █                                         ║
║  Won:            2  ▌                                         ║
║  Lost:          12  █                                         ║
║                                                               ║
║  КОНВЕРСИИ:                                                  ║
║  Discovery → Qualified:    64.7%                              ║
║  Qualified → Outreach:     78.5%                              ║
║  Outreach → Response:      11.8%  ← ✓ цель ≥ 8%             ║
║  Response → Audit:         58.6%  ← ✓ цель ≥ 50%            ║
║  Audit → Proposal:         47.1%                              ║
║  Proposal → Won:           25.0%                              ║
║                                                               ║
║  МЕТРИКИ:                                                    ║
║  MRR текущий:     150 000 ₽                                  ║
║  Pipeline value:  840 000 ₽                                   ║
║  Avg deal size:    75 000 ₽/мес                               ║
║  Avg cycle time:  18 дней                                     ║
║  CAC:             28 750 ₽                                    ║
║  LTV (прогноз):   450 000 ₽                                  ║
║  LTV/CAC:         15.7x                                       ║
║                                                               ║
║  SLA:                                                        ║
║  ⚠️ 3 нарушения (Outreach: 2, Audit: 1)                      ║
║                                                               ║
║  ТОП ЛИДЫ:                                                   ║
║  #1 KnifeKing      Score:87  HOT  1.2M потерь  → Audit       ║
║  #2 SteelPro       Score:82  HOT  890K потерь  → Response     ║
║  #3 ChefBlade      Score:76  HOT  1.5M потерь  → Outreach    ║
║                                                               ║
╚══════════════════════════════════════════════════════════════╝

Еженедельный отчёт

Метрика	На этой неделе	Прошлая неделя	Изменение
Новых лидов	127	95	+33.7%
Аутричей отправлено	82	64	+28.1%
Ответов получено	11	7	+57.1%
Аудитов отправлено	6	4	+50.0%
КП отправлено	3	2	+50.0%
Deals Won	1	0	+1
Новый MRR	75 000 ₽	0 ₽	+75K
Response rate	13.4%	10.9%	+2.5pp

---

CLI команды

# Статус pipeline
python spider.py leadgen pipeline \
    --json

# Лиды по стадии
python spider.py leadgen leads \
    --stage response \
    --temperature hot \
    --limit 20 \
    --json

# Переместить лида
python spider.py leadgen advance \
    --lead-id 42 \
    --stage audit \
    --real

# Назначить менеджера
python spider.py leadgen assign \
    --lead-id 42 \
    --manager "ivan_petrov"

# Проверка SLA
python spider.py leadgen sla-check \
    --json

# Дашборд
python spider.py leadgen dashboard

# Еженедельный отчёт
python spider.py leadgen weekly-report \
    --output reports/leadgen_week_12.md

# Воронка конверсий
python spider.py leadgen funnel \
    --period "2026-03" \
    --json

---

SLA трекинг

Целевые показатели

Стадия	SLA	Эскалация Level 1	Эскалация Level 2
Outreach	24ч	+12ч: менеджер + Telegram	+24ч: руководитель
Response	2ч	+1ч: менеджер	+2ч: руководитель
Audit	48ч	+24ч: менеджер	+48ч: руководитель
Proposal	24ч	+12ч: менеджер	+24ч: руководитель
Negotiation	5 дн	+2 дня: менеджер	+3 дня: руководитель

Нарушение SLA

Каждое нарушение SLA логируется и учитывается в KPI менеджера. При 3+ нарушениях в неделю — автоматическое уведомление руководителю.