Исходный код amocrm.models.common

from __future__ import annotations

import dataclasses
from dataclasses import dataclass, field
from typing import Any, ClassVar, TypeVar




class CustomFieldsMixin:
    """Mixin для типизированного доступа к кастомным полям AmoCRM.

    Предоставляет методы для чтения значений кастомных полей по их ID.
    Должен использоваться совместно с датаклассами, имеющими поле
    ``custom_fields_values: list[CustomFieldValue] | None``.
    """

    custom_fields_values: list[CustomFieldValue] | None



    def get_cf_raw(self, field_id: int) -> list[dict[str, Any]] | None:
        """Вернуть список значений кастомного поля по его ID или ``None``."""
        if self.custom_fields_values is None:
            return None
        for cf in self.custom_fields_values:
            if cf.field_id == field_id:
                return cf.values
        return None




    def get_cf_value(self, field_id: int) -> Any:
        """Вернуть первое значение поля (``values[0]["value"]``) или ``None``."""
        values = self.get_cf_raw(field_id)
        if not values:
            return None
        return values[0].get("value")




    def get_cf_values(self, field_id: int) -> list[Any]:
        """Вернуть все значения поля (``values[i]["value"]``)."""
        values = self.get_cf_raw(field_id)
        if not values:
            return []
        return [v.get("value") for v in values]




    def get_cf_enum_id(self, field_id: int) -> int | None:
        """Вернуть ``enum_id`` первого значения поля или ``None``."""
        values = self.get_cf_raw(field_id)
        if not values:
            return None
        return values[0].get("enum_id")




    def set_cf_value(self, field_id: int, value: Any) -> None:
        """Установить единственное значение кастомного поля."""
        if self.custom_fields_values is None:
            self.custom_fields_values = []
        for cf in self.custom_fields_values:
            if cf.field_id == field_id:
                cf.values = [] if value is None else [{"value": value}]
                return
        if value is not None:
            self.custom_fields_values.append(
                CustomFieldValue(field_id=field_id, values=[{"value": value}])
            )




    def set_cf_values(self, field_id: int, values: list[Any]) -> None:
        """Установить несколько значений кастомного поля (multiselect)."""
        if self.custom_fields_values is None:
            self.custom_fields_values = []
        new_values = [{"value": v} for v in values]
        for cf in self.custom_fields_values:
            if cf.field_id == field_id:
                cf.values = new_values
                return
        self.custom_fields_values.append(
            CustomFieldValue(field_id=field_id, values=new_values)
        )




    def set_cf_raw(self, field_id: int, values: list[dict[str, Any]] | None) -> None:
        """Установить raw-значения кастомного поля (например, smart_address)."""
        if self.custom_fields_values is None:
            self.custom_fields_values = []
        for cf in self.custom_fields_values:
            if cf.field_id == field_id:
                cf.values = values if values is not None else []
                return
        if values is not None:
            self.custom_fields_values.append(
                CustomFieldValue(field_id=field_id, values=values)
            )






@dataclass(kw_only=True)
class Tag:
    """Тег сущности AmoCRM.

    Attributes:
        id: Идентификатор тега.
        name: Название тега.
    """

    id: int | None = None
    name: str | None = None



    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> Tag:
        """Создать экземпляр из словаря API."""
        return cls(id=data.get("id"), name=data.get("name"))




    def to_dict(self) -> dict[str, Any]:
        """Сериализовать в словарь, исключая поля со значением ``None``."""
        return {k: v for k, v in dataclasses.asdict(self).items() if v is not None}






@dataclass(kw_only=True)
class CustomFieldValue:
    """Значение кастомного поля AmoCRM.

    Attributes:
        field_id: Идентификатор кастомного поля.
        values: Список значений поля (каждое — словарь с ключами ``value``
            и опционально ``enum_id`` / ``enum_code``).
    """

    field_id: int
    values: list[dict[str, Any]] = field(default_factory=list)



    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> CustomFieldValue:
        """Создать экземпляр из словаря API."""
        return cls(field_id=data["field_id"], values=data.get("values", []))




    def to_dict(self) -> dict[str, Any]:
        """Сериализовать в словарь для отправки в API."""
        return {"field_id": self.field_id, "values": self.values}




_M = TypeVar("_M", bound="BaseModel")




class BaseModel(CustomFieldsMixin):
    """Базовый класс DTO-моделей AmoCRM с общей логикой ``from_dict``/``to_dict``.

    Подклассы должны быть датаклассами и определять ``_scalar_fields`` —
    кортеж имён полей, которые берутся напрямую из словаря API.
    """

    _scalar_fields: ClassVar[tuple[str, ...]] = ()
    tags: list[Tag] | None
    custom_fields_values: list[CustomFieldValue] | None



    @classmethod
    def from_dict(cls: type[_M], data: dict[str, Any]) -> _M:
        """Создать экземпляр из словаря API AmoCRM."""
        kwargs: dict[str, Any] = {k: data.get(k) for k in cls._scalar_fields}
        tags_raw = data.get("_embedded", {}).get("tags")
        cf_raw = data.get("custom_fields_values")
        if tags_raw is not None:
            kwargs["tags"] = [Tag.from_dict(t) for t in tags_raw]
        if cf_raw is not None:
            kwargs["custom_fields_values"] = [
                CustomFieldValue.from_dict(cf) for cf in cf_raw
            ]
        return cls(**kwargs)




    def to_dict(self) -> dict[str, Any]:
        """Сериализовать в словарь для API, исключая поля со значением ``None``."""
        result: dict[str, Any] = {
            k: getattr(self, k)
            for k in self._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
            ]
        return result