from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from .common import CustomFieldsMixin, CustomFieldValue, Tag
from .companies import Company
from .contacts import Contact
_LEAD_SCALAR_FIELDS = (
"id",
"name",
"price",
"status_id",
"pipeline_id",
"responsible_user_id",
"group_id",
"created_by",
"updated_by",
"created_at",
"updated_at",
"closed_at",
"closest_task_at",
"is_deleted",
"loss_reason_id",
"score",
"account_id",
"labor_cost",
)
[документация]
@dataclass(kw_only=True)
class Lead(CustomFieldsMixin):
"""DTO-модель сделки AmoCRM.
Attributes:
id: Идентификатор сделки.
name: Название сделки.
price: Бюджет сделки.
status_id: Идентификатор статуса воронки.
pipeline_id: Идентификатор воронки.
responsible_user_id: Идентификатор ответственного пользователя.
group_id: Идентификатор группы пользователей.
created_by: Идентификатор пользователя, создавшего сделку.
updated_by: Идентификатор пользователя, обновившего сделку.
created_at: Дата создания (Unix timestamp).
updated_at: Дата последнего изменения (Unix timestamp).
closed_at: Дата закрытия сделки (Unix timestamp).
closest_task_at: Дата ближайшей задачи (Unix timestamp).
is_deleted: Признак удалённой сделки.
loss_reason_id: Идентификатор причины закрытия (проигрыша).
score: Оценка сделки.
account_id: Идентификатор аккаунта AmoCRM.
labor_cost: Затраченное время (в минутах).
tags: Список тегов сделки.
custom_fields_values: Список значений кастомных полей.
"""
id: int | None = None
name: str | None = None
price: int | None = None
status_id: int | None = None
pipeline_id: int | None = None
responsible_user_id: int | None = None
group_id: int | None = None
created_by: int | None = None
updated_by: int | None = None
created_at: int | None = None
updated_at: int | None = None
closed_at: int | None = None
closest_task_at: int | None = None
is_deleted: bool | None = None
loss_reason_id: int | None = None
score: int | None = None
account_id: int | None = None
labor_cost: int | None = None
tags: list[Tag] | None = None
custom_fields_values: list[CustomFieldValue] | None = None
contacts: list[Contact] | None = None
company: Company | None = None
[документация]
@classmethod
def from_dict(cls, data: dict[str, Any]) -> Lead:
"""Создать экземпляр из словаря API AmoCRM."""
embedded = data.get("_embedded", {})
tags_raw = embedded.get("tags")
contacts_raw = embedded.get("contacts")
companies_raw = embedded.get("companies")
cf_raw = data.get("custom_fields_values")
return cls(
id=data.get("id"),
name=data.get("name"),
price=data.get("price"),
status_id=data.get("status_id"),
pipeline_id=data.get("pipeline_id"),
responsible_user_id=data.get("responsible_user_id"),
group_id=data.get("group_id"),
created_by=data.get("created_by"),
updated_by=data.get("updated_by"),
created_at=data.get("created_at"),
updated_at=data.get("updated_at"),
closed_at=data.get("closed_at"),
closest_task_at=data.get("closest_task_at"),
is_deleted=data.get("is_deleted"),
loss_reason_id=data.get("loss_reason_id"),
score=data.get("score"),
account_id=data.get("account_id"),
labor_cost=data.get("labor_cost"),
tags=[Tag.from_dict(t) for t in tags_raw] if tags_raw is not None else None,
custom_fields_values=(
[CustomFieldValue.from_dict(cf) for cf in cf_raw]
if cf_raw is not None
else None
),
contacts=(
[Contact.from_dict(c) for c in contacts_raw] if contacts_raw else None
),
company=Company.from_dict(companies_raw[0]) if companies_raw else None,
)
[документация]
def to_dict(self) -> dict[str, Any]:
"""Сериализовать в словарь для API, исключая поля со значением ``None``."""
result: dict[str, Any] = {
k: getattr(self, k)
for k in _LEAD_SCALAR_FIELDS
if getattr(self, k) is not None
}
if self.tags is not None:
result["tags"] = [t.to_dict() for t in self.tags]
if self.custom_fields_values is not None:
result["custom_fields_values"] = [
cf.to_dict() for cf in self.custom_fields_values
]
embedded: dict[str, Any] = {}
if self.contacts is not None:
embedded["contacts"] = [c.to_dict() for c in self.contacts]
if self.company is not None:
embedded["companies"] = [self.company.to_dict()]
if embedded:
result["_embedded"] = embedded
return result