CRM на автопилоте — HubSpot + AI, сделки без менеджера

# HubSpot через Claude + API
import os
import time
import anthropic
import requests

HUBSPOT_TOKEN = os.environ["HUBSPOT_ACCESS_TOKEN"]

def update_crm_from_email(email_text: str, contact_id: str):
    client = anthropic.Anthropic()

    # Шаг 1: Claude извлекает структурированные данные из письма
    extraction = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"""Извлеки из письма клиента следующие данные в JSON:
            - contact_name: имя клиента
            - company: компания
            - interest: что конкретно интересует
            - budget_hint: намёк на бюджет (если есть)
            - next_step: что нужно сделать дальше
            - urgency: срочность (low/medium/high)

            Письмо: {email_text}

            Верни только JSON, без пояснений."""
        }]
    )

    import json
    data = json.loads("".join(b.text for b in extraction.content if b.type == "text"))

    # Шаг 2: Обновляем контакт в HubSpot
    headers = {
        "Authorization": f"Bearer {HUBSPOT_TOKEN}",
        "Content-Type": "application/json"
    }

    # Обновляем свойства контакта.
    # Поле notes_last_contacted HubSpot заполняет сам (только чтение), поэтому
    # резюме письма сохраняем заметкой (см. практику ниже), а не в свойстве.
    update_payload = {
        "properties": {
            "hs_lead_status": "IN_PROGRESS"
        }
    }

    response = requests.patch(
        f"https://api.hubapi.com/crm/v3/objects/contacts/{contact_id}",
        headers=headers,
        json=update_payload
    )

    # Шаг 3: Создаём задачу на follow-up если высокая срочность
    if data.get("urgency") == "high":
        task_payload = {
            "properties": {
                "hs_task_subject": f"Срочный follow-up: {data.get('contact_name')}",
                "hs_timestamp": str(int(time.time() * 1000)),  # срок задачи, обязательное поле
                "hs_task_status": "NOT_STARTED",
                "hs_task_priority": "HIGH",
                "hubspot_owner_id": "owner_id_здесь"
            }
        }
        requests.post(
            "https://api.hubapi.com/crm/v3/objects/tasks",
            headers=headers,
            json=task_payload
        )

    return data

def score_lead(contact_history: str) -> dict:
    client = anthropic.Anthropic()

    response = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=300,
        messages=[{
            "role": "user",
            "content": f"""Оцени готовность лида к покупке по шкале 0-100.

            История взаимодействий:
            {contact_history}

            Верни JSON:
            {{
                "score": число от 0 до 100,
                "reasoning": "краткое объяснение",
                "recommended_action": "что делать прямо сейчас",
                "hot_signals": ["список сигналов готовности"],
                "risk_signals": ["список тревожных сигналов"]
            }}"""
        }]
    )

    import json
    return json.loads("".join(b.text for b in response.content if b.type == "text"))

# Пример использования
history = """
- 15 апреля: первый звонок, спросили про функционал
- 18 апреля: прислали список требований
- 22 апреля: попросили демо
- 25 апреля: после демо спросили "какие минимальные сроки внедрения?"
- 28 апреля: написали "наш директор хочет встретиться на следующей неделе"
"""

result = score_lead(history)
# score: 87
# recommended_action: "Немедленно подтвердить встречу с директором, подготовить КП"

def generate_followup(
    contact_name: str,
    last_interaction: str,
    days_since_contact: int,
    deal_stage: str
) -> str:
    client = anthropic.Anthropic()

    response = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=400,
        messages=[{
            "role": "user",
            "content": f"""Напиши follow-up письмо клиенту.

            Контекст:
            - Имя: {contact_name}
            - Последний контакт: {last_interaction}
            - Дней без ответа: {days_since_contact}
            - Стадия сделки: {deal_stage}

            Требования:
            - Короткое (3-4 предложения)
            - Не давящее, не умоляющее
            - Добавь ценность (статья, инсайт, кейс)
            - Один чёткий призыв к действию
            - Русский язык, деловой но живой тон"""
        }]
    )

    return "".join(b.text for b in response.content if b.type == "text")

def audit_pipeline(deals: list) -> list:
    client = anthropic.Anthropic()

    deals_text = "\n".join([
        f"- {d['name']}: стадия '{d['stage']}', "
        f"последнее действие {d['days_stale']} дней назад, "
        f"сумма {d['amount']}"
        for d in deals
    ])

    response = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=800,
        messages=[{
            "role": "user",
            "content": f"""Проанализируй пайплайн продаж и выдели проблемные сделки.

            Сделки:
            {deals_text}

            Для каждой проблемной сделки (застряла >14 дней) дай:
            1. Диагноз: почему застряла
            2. План выхода: конкретные 2-3 шага
            3. Рекомендацию: продолжать или закрыть как потерянную

            Формат: список с заголовками по каждой сделке."""
        }]
    )

    return "".join(b.text for b in response.content if b.type == "text")

# .env
HUBSPOT_ACCESS_TOKEN=<токен из HubSpot>
ANTHROPIC_API_KEY=sk-ant-...

# email_processor.py
import os
import json
import time
import anthropic
import requests
from dotenv import load_dotenv

load_dotenv()

client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
HUBSPOT_TOKEN = os.environ["HUBSPOT_ACCESS_TOKEN"]

def process_incoming_email(
    email_from: str,
    email_subject: str,
    email_body: str
):
    """Обрабатывает входящее письмо и обновляет CRM"""

    print(f"Обрабатываю письмо от: {email_from}")

    # 1. Claude анализирует письмо
    analysis = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=600,
        messages=[{
            "role": "user",
            "content": f"""Проанализируй входящее письмо клиента.

Тема: {email_subject}
От: {email_from}
Текст: {email_body}

Верни JSON без дополнительного текста:
{{
    "intent": "inquiry/complaint/purchase_ready/follow_up/other",
    "summary": "краткое резюме в 1-2 предложения",
    "key_points": ["список важных моментов"],
    "urgency": "low/medium/high",
    "sentiment": "positive/neutral/negative",
    "next_action": "конкретное следующее действие",
    "follow_up_days": число дней когда написать если нет ответа
}}"""
        }]
    )

    data = json.loads("".join(b.text for b in analysis.content if b.type == "text"))
    print(f"Анализ: {data['intent']}, срочность: {data['urgency']}")

    # 2. Ищем контакт в HubSpot по email
    headers = {
        "Authorization": f"Bearer {HUBSPOT_TOKEN}",
        "Content-Type": "application/json"
    }

    search_response = requests.post(
        "https://api.hubapi.com/crm/v3/objects/contacts/search",
        headers=headers,
        json={
            "filterGroups": [{
                "filters": [{
                    "propertyName": "email",
                    "operator": "EQ",
                    "value": email_from
                }]
            }],
            "properties": ["email", "firstname", "lastname", "hs_lead_status"]
        }
    )

    contacts = search_response.json().get("results", [])

    if contacts:
        contact_id = contacts[0]["id"]
        contact_name = (
            contacts[0]["properties"].get("firstname", "") + " " +
            contacts[0]["properties"].get("lastname", "")
        ).strip() or email_from

        # 3. Обновляем контакт
        update_response = requests.patch(
            f"https://api.hubapi.com/crm/v3/objects/contacts/{contact_id}",
            headers=headers,
            json={
                "properties": {
                    "hs_lead_status": "IN_PROGRESS" if data["intent"] == "purchase_ready" else "OPEN"
                }
            }
        )

        # 4. Добавляем заметку
        note_response = requests.post(
            "https://api.hubapi.com/crm/v3/objects/notes",
            headers=headers,
            json={
                "properties": {
                    "hs_note_body": f"Email анализ:\n{data['summary']}\n\nКлючевые моменты:\n" +
                                   "\n".join(f"• {p}" for p in data["key_points"]),
                    "hs_timestamp": str(int(time.time() * 1000))
                },
                "associations": [{
                    "to": {"id": contact_id},
                    "types": [{"associationCategory": "HUBSPOT_DEFINED", "associationTypeId": 202}]
                }]
            }
        )

        # 5. Создаём задачу если высокая срочность
        if data["urgency"] == "high" or data["intent"] == "purchase_ready":
            task_response = requests.post(
                "https://api.hubapi.com/crm/v3/objects/tasks",
                headers=headers,
                json={
                    "properties": {
                        "hs_task_subject": f"[ГОРЯЧИЙ] Ответить: {contact_name} — {email_subject}",
                        "hs_timestamp": str(int(time.time() * 1000)),
                        "hs_task_status": "NOT_STARTED",
                        "hs_task_priority": "HIGH",
                        "hs_task_body": data["next_action"]
                    }
                }
            )
            print(f"Создана срочная задача для {contact_name}")

        print(f"CRM обновлён для {contact_name}")
        return {"status": "updated", "contact": contact_name, "analysis": data}

    else:
        # Контакт не найден — создаём новый
        print(f"Новый контакт: {email_from}")
        create_response = requests.post(
            "https://api.hubapi.com/crm/v3/objects/contacts",
            headers=headers,
            json={
                "properties": {
                    "email": email_from,
                    "hs_lead_status": "NEW"
                }
            }
        )
        return {"status": "created", "contact": email_from, "analysis": data}


# Тест
if __name__ == "__main__":
    result = process_incoming_email(
        email_from="[email protected]",
        email_subject="Хотим внедрить вашу систему",
        email_body="""
        Добрый день!

        Мы рассматриваем несколько решений для автоматизации продаж.
        Ваш продукт нас заинтересовал. У нас команда 15 менеджеров,
        бюджет обсуждаем, хотелось бы начать уже в этом квартале.

        Можем ли мы провести демо на следующей неделе?

        Иван Петров, Директор по продажам
        """
    )
    print(json.dumps(result, ensure_ascii=False, indent=2))

python email_processor.py