import logging import time import traceback from typing import Optional import requests from django.core.cache import cache from django.db import transaction from core.crm.exceptions import CRMException from core.models import CustomerRelationshipManagementProvider, Order, OrderCrmLink logger = logging.getLogger("django") class AmoCRM: def __init__(self): try: self.instance = CustomerRelationshipManagementProvider.objects.get(name="amo") except CustomerRelationshipManagementProvider.DoesNotExist as dne: logger.warning("AMO CRM provider not found") raise CRMException("AMO CRM provider not found") from dne self.base = f"https://{self.instance.integration_url}" self.client_id = self.instance.authentication.get("client_id") self.client_secret = self.instance.authentication.get("client_secret") self.refresh_token = cache.get("amo_refresh_token") self.redirect_uri = self.instance.attributes.get("redirect_uri") self.pipeline_id = self.instance.attributes.get("pipeline_id") self.stage_map = self.instance.attributes.get("stage_map") self.responsible_user_id = self.instance.attributes.get("responsible_user_id") self.fns_api_key = self.instance.attributes.get("fns_api_key") if not all( [ self.base, self.client_id, self.client_secret, self.redirect_uri, self.fns_api_key, ] ): raise CRMException("AMO CRM provider not configured") def _token(self) -> str: cached = getattr(self, "_cached_token", None) expiry = getattr(self, "_cached_expiry", 0) if cached and time.time() < expiry - 15: return cached payload = { "client_id": self.client_id, "client_secret": self.client_secret, "grant_type": "refresh_token", "refresh_token": self.refresh_token, "redirect_uri": self.redirect_uri, } r = requests.post(f"{self.base}/oauth2/access_token", json=payload, timeout=15) r.raise_for_status() data = r.json() self._cached_token = data["access_token"] self._cached_expiry = time.time() + int(data.get("expires_in", 900)) self.refresh_token = data.get("refresh_token", self.refresh_token) return self._cached_token def _headers(self) -> dict: return {"Authorization": f"Bearer {self._token()}", "Content-Type": "application/json"} def _build_lead_payload(self, order: Order) -> dict: name = f"Заказ #{order.human_readable_id}" price = int(round(order.total_price)) payload: dict = {"name": name, "price": price} contact_id = self._create_contact(order) if contact_id: payload["_embedded"] = {"contacts": [{"id": contact_id}]} if self.responsible_user_id: payload["responsible_user_id"] = self.responsible_user_id return payload def _get_customer_name(self, order: Order) -> str: if not order.attributes.get("business_identificator"): return order.user.get_full_name() or ( f"{order.attributes.get('customer_name')} | " f"{order.attributes.get('customer_phone_number') or order.attributes.get('customer_email')}" ) try: business_identificator = order.attributes.get("business_identificator") r = requests.get( f"https://api-fns.ru/api/egr?req={business_identificator}&key={self.fns_api_key}", timeout=15 ) body = r.json() except requests.exceptions.RequestException as rex: logger.error(f"Unable to get company info with FNS: {rex}") logger.error(traceback.format_exc()) return "" ul = body.get("ЮЛ") ip = body.get("ИП") if ul and not ip: return f"{ul.get('НаимСокрЮЛ')} | {business_identificator}" if ip and not ul: return f"ИП {ip.get('ФИОПолн')} | {business_identificator}" return "" def _create_contact(self, order: Order) -> int | None: try: customer_name = self._get_customer_name(order) if customer_name: r = requests.get( f"{self.base}/api/v4/contacts", headers=self._headers().update({"filter[name]": customer_name, "limit": 1}), timeout=15, ) if r.status_code == 200: body = r.json() return body.get("_embedded", {}).get("contacts", [{}])[0].get("id", None) create_contact_payload = {"name": customer_name} if self.responsible_user_id: create_contact_payload["responsible_user_id"] = self.responsible_user_id if order.user: create_contact_payload["first_name"] = order.user.first_name or "" create_contact_payload["last_name"] = order.user.last_name or "" r = requests.post( f"{self.base}/api/v4/contacts", json={"name": customer_name}, headers=self._headers(), timeout=15 ) if r.status_code == 200: body = r.json() return body.get("_embedded", {}).get("contacts", [{}])[0].get("id", None) else: return None except requests.exceptions.RequestException as rex: logger.error(f"Unable to create a company in AmoCRM: {rex}") logger.error(traceback.format_exc()) raise CRMException("Unable to create a company in AmoCRM") from rex def process_order_changes(self, order: Order) -> str: with transaction.atomic(): try: link: Optional[OrderCrmLink] = OrderCrmLink.objects.get(order=order) except OrderCrmLink.MultipleObjectsReturned: link = OrderCrmLink.objects.filter(order=order).first() except OrderCrmLink.DoesNotExist: link = None if link: lead_id = link.crm_lead_id payload = self._build_lead_payload(order) r = requests.patch( f"{self.base}/api/v4/leads/{lead_id}", json=payload, headers=self._headers(), timeout=15 ) if r.status_code not in (200, 204): r.raise_for_status() return lead_id payload = self._build_lead_payload(order) r = requests.post(f"{self.base}/api/v4/leads", json=[payload], headers=self._headers(), timeout=15) r.raise_for_status() body = r.json() lead_id = str(body["_embedded"]["leads"][0]["id"]) OrderCrmLink.objects.create(order=order, crm_lead_id=lead_id, crm=self.instance) return lead_id def update_order_status(self, crm_lead_id: str, new_status_code: str) -> None: link = OrderCrmLink.objects.filter(crm_lead_id=crm_lead_id).first() if not link: return from core.models import Order order = Order.objects.get(uuid=link.order_uuid) if order.status == new_status_code: return order.status = new_status_code order.save(update_fields=["status"])