Интеграция с amoCRM API: полное руководство по автоматизации продаж

 1import httpx
 2from datetime import datetime, timedelta
 3
 4class AmoCRMAPI:
 5    BASE_URL = "https://{subdomain}.amocrm.ru"
 6    
 7    def __init__(self, subdomain: str, client_id: str, client_secret: str, refresh_token: str):
 8        self.subdomain = subdomain
 9        self.client_id = client_id
10        self.client_secret = client_secret
11        self.refresh_token = refresh_token
12        self.access_token = None
13        self.token_expires = None
14    
15    async def _refresh_access_token(self):
16        """Обновить access_token"""
17        async with httpx.AsyncClient() as client:
18            response = await client.post(
19                f"{self.BASE_URL}/oauth2/access_token",
20                json={
21                    "client_id": self.client_id,
22                    "client_secret": self.client_secret,
23                    "grant_type": "refresh_token",
24                    "refresh_token": self.refresh_token,
25                }
26            )
27            data = response.json()
28            self.access_token = data["access_token"]
29            self.refresh_token = data["refresh_token"]
30            self.token_expires = datetime.now() + timedelta(seconds=data["expires_in"] - 60)
31    
32    async def _get_headers(self) -> dict:
33        if not self.access_token or datetime.now() >= self.token_expires:
34            await self._refresh_access_token()
35        return {
36            "Authorization": f"Bearer {self.access_token}",
37            "Content-Type": "application/json",
38        }
39    
40    async def create_lead(self, name: str, contact_data: dict, pipeline_id: int = None) -> dict:
41        """Создать сделку с контактом"""
42        headers = await self._get_headers()
43        
44        # Сначала создаём или находим контакт
45        contact_id = await self._find_or_create_contact(contact_data)
46        
47        lead = {
48            "name": name,
49            "contact_id": contact_id,
50            "custom_fields_values": [
51                {"field_id": 1, "values": [{"value": contact_data.get("source", "API")}]}
52            ],
53        }
54        if pipeline_id:
55            lead["pipeline_id"] = pipeline_id
56        
57        async with httpx.AsyncClient() as client:
58            response = await client.post(
59                f"{self.BASE_URL}/api/v4/leads",
60                headers=headers,
61                json=[lead],
62            )
63            response.raise_for_status()
64            return response.json()["_embedded"]["leads"][0]
65    
66    async def _find_or_create_contact(self, data: dict) -> int:
67        """Найти существующий контакт или создать новый"""
68        headers = await self._get_headers()
69        
70        # Поиск по телефону
71        async with httpx.AsyncClient() as client:
72            response = await client.get(
73                f"{self.BASE_URL}/api/v4/contacts",
74                headers=headers,
75                params={"query": data["phone"]},
76            )
77            result = response.json()
78        
79        if result.get("_embedded", {}).get("contacts"):
80            return result["_embedded"]["contacts"][0]["id"]
81        
82        # Создаём новый контакт
83        async with httpx.AsyncClient() as client:
84            response = await client.post(
85                f"{self.BASE_URL}/api/v4/contacts",
86                headers=headers,
87                json=[{
88                    "name": data["name"],
89                    "custom_fields_values": [
90                        {"field_id": 2, "values": [{"value": data["phone"], "enum_code": "WORK"}]},
91                        {"field_id": 3, "values": [{"value": data.get("email", ""), "enum_code": "WORK"}]},
92                    ],
93                }],
94            )
95            response.raise_for_status()
96            return response.json()["_embedded"]["contacts"][0]["id"]

 1from fastapi import FastAPI, Request
 2
 3app = FastAPI()
 4
 5@app.post("/webhook/amocrm")
 6async def handle_amocrm_webhook(request: Request):
 7    """Обработка webhook от amoCRM"""
 8    data = await request.json()
 9    
10    for lead in data.get("leads", {}).get("add", []):
11        # Новая сделка — запускаем автоматизацию
12        await process_new_lead(lead)
13    
14    for lead in data.get("leads", {}).get("update", []):
15        # Обновление сделки — проверяем смену этапа
16        await check_stage_change(lead)
17    
18    for contact in data.get("contacts", {}).get("add", []):
19        # Новый контакт — обогащаем данные
20        await enrich_contact(contact)
21    
22    return {"status": "ok"}

 1async def qualify_lead(amocrm: AmoCRMAPI, lead_id: int) -> dict:
 2    """Автоматическая квалификация лида по набору правил"""
 3    
 4    # Получаем полные данные сделки
 5    lead = await amocrm.get_lead(lead_id)
 6    contact = await amocrm.get_contact(lead["contact_id"])
 7    
 8    # Скоринг по правилам
 9    score = 0
10    reasons = []
11    
12    # Правило 1: Есть телефон
13    phones = [f["values"][0]["value"] for f in contact.get("custom_fields_values", []) 
14              if f.get("field_code") == "PHONE"]
15    if phones:
16        score += 20
17        reasons.append("Есть телефон")
18    
19    # Правило 2: Есть email
20    emails = [f["values"][0]["value"] for f in contact.get("custom_fields_values", []) 
21              if f.get("field_code") == "EMAIL"]
22    if emails:
23        score += 10
24        reasons.append("Есть email")
25    
26    # Правило 3: Источник — горячий канал
27    hot_sources = ["CALL", "WEB_FORM", "CHAT"]
28    source = lead.get("source", "")
29    if source in hot_sources:
30        score += 25
31        reasons.append(f"Горячий источник: {source}")
32    
33    # Правило 4: Бюджет указан
34    budget = lead.get("price", 0)
35    if budget > 50000:
36        score += 30
37        reasons.append(f"Бюджет: {budget} ₽")
38    elif budget > 10000:
39        score += 15
40        reasons.append(f"Бюджет: {budget} ₽")
41    
42    # Правило 5: Комментарий содержит ключевые слова
43    comments = lead.get("notes", [])
44    hot_keywords = ["срочно", "сегодня", "бюджет", "оплата", "договор"]
45    for note in comments:
46        text = note.get("text", "").lower()
47        if any(kw in text for kw in hot_keywords):
48            score += 15
49            reasons.append("Горячие ключевые слова в комментарии")
50            break
51    
52    # Определяем этап воронки по скору
53    if score >= 70:
54        stage = "PREPARATION"  # Презентация
55        priority = "HIGH"
56    elif score >= 40:
57        stage = "NEGOTIATION"  # Переговоры
58        priority = "MEDIUM"
59    else:
60        stage = "NEW_LEAD"  # Остаётся на квалификации
61        priority = "LOW"
62    
63    # Обновляем сделку
64    await amocrm.update_lead(lead_id, {
65        "status_id": stage,
66        "custom_fields_values": [
67            {"field_id": 100, "values": [{"value": str(score)}]},
68            {"field_id": 101, "values": [{"value": priority}]},
69        ],
70    })
71    
72    # Создаём задачу для менеджера
73    deadline_hours = {"HIGH": 1, "MEDIUM": 4, "LOW": 24}
74    await amocrm.create_task({
75        "entity_id": lead_id,
76        "entity_type": "leads",
77        "text": f"Квалификация: скор {score}. {', '.join(reasons)}",
78        "complete_till": int((datetime.now() + timedelta(hours=deadline_hours[priority])).timestamp()),
79        "responsible_user_id": await get_best_manager(amocrm, priority),
80    })
81    
82    return {"lead_id": lead_id, "score": score, "stage": stage, "priority": priority}

 1async def handle_whatsapp_message(amocrm: AmoCRMAPI, webhook_data: dict):
 2    """Обработка входящего сообщения из WhatsApp"""
 3    
 4    message = webhook_data["message"]
 5    contact_phone = message["from"]
 6    text = message["text"]
 7    
 8    # Ищем контакт по телефону
 9    contact = await amocrm.find_contact_by_phone(contact_phone)
10    
11    if not contact:
12        # Новый контакт — создаём лид
13        lead = await amocrm.create_lead(
14            name=f"WhatsApp: {contact_phone}",
15            contact_data={
16                "name": message.get("contact_name", contact_phone),
17                "phone": contact_phone,
18                "source": "WHATSAPP",
19            }
20        )
21        contact = {"id": lead["contact_id"]}
22    
23    # Ищем активную сделку
24    leads = await amocrm.get_leads_by_contact(contact["id"])
25    active_lead = next((l for l in leads if l["status_id"] != "142"), None)  # 142 = закрыта
26    
27    if not active_lead:
28        # Нет активной сделки — создаём
29        active_lead = await amocrm.create_lead(
30            name=f"WhatsApp: {text[:50]}",
31            contact_data={"phone": contact_phone, "name": contact.get("name", contact_phone)}
32        )
33    
34    # Добавляем сообщение в воронку чатов
35    await amocrm.add_chat_message(
36        lead_id=active_lead["id"],
37        message=text,
38        source="whatsapp",
39        contact_phone=contact_phone,
40    )
41    
42    # Проверяем на горячие ключевые слова
43    hot_keywords = ["купить", "заказать", "цена", "стоимость", "когда", "срочно"]
44    if any(kw in text.lower() for kw in hot_keywords):
45        # Горячий лид — уведомляем менеджера
46        await amocrm.create_task({
47            "entity_id": active_lead["id"],
48            "entity_type": "leads",
49            "text": f"🔥 Горячий лид из WhatsApp: {text[:100]}",
50            "complete_till": int((datetime.now() + timedelta(minutes=15)).timestamp()),
51        })
52    
53    return {"lead_id": active_lead["id"], "action": "processed"}