SkyForces
642 lines
22 KiB
Python
642 lines
22 KiB
Python
import asyncio
|
|
import logging
|
|
import time
|
|
from collections.abc import AsyncGenerator
|
|
from contextlib import asynccontextmanager
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from .bitrix import BitrixClient
|
|
from .domain import (
|
|
DEALS_PER_PAGE,
|
|
Binding,
|
|
ClientInfo,
|
|
DealPage,
|
|
DealStage,
|
|
DealStageAdvance,
|
|
DealStageFilter,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DealAssignmentConflict(RuntimeError):
|
|
"""Ответственный изменился после показа карточки."""
|
|
|
|
|
|
class DealAdvanceForbidden(RuntimeError):
|
|
"""Стадию может менять только ответственный за сделку."""
|
|
|
|
|
|
class DealStageConflict(RuntimeError):
|
|
"""Стадия изменилась после запроса комментария."""
|
|
|
|
|
|
class DealCommentSaveError(RuntimeError):
|
|
"""Стадия изменена, но комментарий не добавлен."""
|
|
|
|
def __init__(self, advance: DealStageAdvance) -> None:
|
|
super().__init__(
|
|
"Стадия изменена, но комментарий не удалось сохранить."
|
|
)
|
|
self.advance = advance
|
|
|
|
|
|
@dataclass
|
|
class _DealLockEntry:
|
|
lock: asyncio.Lock
|
|
users: int = 0
|
|
|
|
|
|
class DealService:
|
|
"""Загрузка и изменение сделок Bitrix."""
|
|
|
|
deal_select = [
|
|
"ID",
|
|
"TITLE",
|
|
"STAGE_ID",
|
|
"IS_NEW",
|
|
"OPPORTUNITY",
|
|
"CURRENCY_ID",
|
|
"DATE_CREATE",
|
|
"ASSIGNED_BY_ID",
|
|
"CONTACT_ID",
|
|
"COMPANY_ID",
|
|
"SOURCE_ID",
|
|
"COMMENTS",
|
|
]
|
|
|
|
def __init__(
|
|
self,
|
|
bitrix: BitrixClient,
|
|
take_to_work_stage_id: str
|
|
) -> None:
|
|
self.bitrix = bitrix
|
|
self.take_to_work_stage_id = take_to_work_stage_id
|
|
self._deal_locks: dict[tuple[str, str], _DealLockEntry] = {}
|
|
self._deal_locks_guard = asyncio.Lock()
|
|
self._stage_cache: dict[
|
|
tuple[str, int, int], tuple[float, tuple[DealStage, ...]]
|
|
] = {}
|
|
|
|
async def list_by_stage(
|
|
self,
|
|
binding: Binding,
|
|
stage_key: str = "new",
|
|
page: int = 0,
|
|
limit: int = DEALS_PER_PAGE
|
|
) -> DealPage:
|
|
stage_filters = await self.stage_filters(binding)
|
|
stage_filter = self._select_stage_filter(stage_filters, stage_key)
|
|
bitrix_filter = {}
|
|
if stage_filter.stage_id:
|
|
bitrix_filter["STAGE_ID"] = stage_filter.stage_id
|
|
if stage_filter.assigned_to_viewer:
|
|
bitrix_filter["ASSIGNED_BY_ID"] = binding.bitrix_user_id
|
|
|
|
page = max(page, 0)
|
|
start_index = page * limit
|
|
end_index = start_index + limit
|
|
loaded_deals = []
|
|
bitrix_start: int | None = 0
|
|
total_deals = 0
|
|
|
|
# Битрикс и Telegram используют страницы разного размера.
|
|
while bitrix_start is not None and len(loaded_deals) < end_index:
|
|
# https://apidocs.bitrix24.ru/api-reference/crm/deals/crm-deal-list.html
|
|
data = await self.bitrix.call(
|
|
binding,
|
|
"crm.deal.list",
|
|
{
|
|
"order": {"DATE_CREATE": "DESC"},
|
|
"filter": bitrix_filter,
|
|
"select": self.deal_select,
|
|
"start": bitrix_start
|
|
}
|
|
)
|
|
loaded_deals.extend(data.get("result", []))
|
|
total_deals = int(data.get("total", len(loaded_deals)))
|
|
bitrix_start = data.get("next")
|
|
|
|
deals = loaded_deals[start_index:end_index]
|
|
total_pages = max(1, (total_deals + limit - 1) // limit)
|
|
return DealPage(
|
|
deals,
|
|
page,
|
|
total_deals,
|
|
total_pages,
|
|
stage_filter,
|
|
stage_filters
|
|
)
|
|
|
|
async def stage_filters(
|
|
self,
|
|
binding: Binding
|
|
) -> tuple[DealStageFilter, ...]:
|
|
stages = await self._stages(binding, category_id=0)
|
|
filters = [
|
|
DealStageFilter(
|
|
key="mine",
|
|
title="Мои сделки",
|
|
stage_id=None,
|
|
assigned_to_viewer=True,
|
|
)
|
|
]
|
|
filters.extend(
|
|
DealStageFilter(stage.stage_id, stage.title, stage.stage_id)
|
|
for stage in stages
|
|
)
|
|
filters.append(DealStageFilter("all", "Все", None))
|
|
return tuple(filters)
|
|
|
|
def _select_stage_filter(
|
|
self,
|
|
filters: tuple[DealStageFilter, ...],
|
|
stage_key: str
|
|
) -> DealStageFilter:
|
|
initial = next(
|
|
(item for item in filters if item.stage_id is not None),
|
|
filters[-1]
|
|
)
|
|
# Callback `new` означает первую стадию, полученную из Битрикса.
|
|
if stage_key == "new":
|
|
return initial
|
|
|
|
selected = next((item for item in filters if item.key == stage_key),
|
|
None)
|
|
return selected or initial
|
|
|
|
async def _stages(
|
|
self,
|
|
binding: Binding,
|
|
category_id: int
|
|
) -> tuple[DealStage, ...]:
|
|
key = (binding.member_id, binding.bitrix_user_id, category_id)
|
|
cached = self._stage_cache.get(key)
|
|
if cached and cached[0] > time.monotonic():
|
|
return cached[1]
|
|
|
|
entity_id = "DEAL_STAGE" if category_id == 0 else (
|
|
f"DEAL_STAGE_{category_id}"
|
|
)
|
|
items = []
|
|
bitrix_start: int | None = 0
|
|
# https://apidocs.bitrix24.ru/api-reference/crm/status/crm-status-list.html
|
|
while bitrix_start is not None:
|
|
data = await self.bitrix.call(
|
|
binding,
|
|
"crm.status.list",
|
|
{
|
|
"order": {"SORT": "ASC"},
|
|
"filter": {"ENTITY_ID": entity_id},
|
|
"start": bitrix_start
|
|
}
|
|
)
|
|
items.extend(data.get("result", []))
|
|
bitrix_start = data.get("next")
|
|
|
|
stages = []
|
|
seen_stage_ids = set()
|
|
for item in items:
|
|
raw_stage_id = str(item.get("STATUS_ID") or "")
|
|
if not raw_stage_id:
|
|
continue
|
|
|
|
stage_id = raw_stage_id
|
|
prefix = f"C{category_id}:"
|
|
if category_id and not stage_id.startswith(prefix):
|
|
stage_id = prefix + stage_id
|
|
if stage_id in seen_stage_ids:
|
|
continue
|
|
|
|
semantics = self._stage_semantics(item)
|
|
stages.append(
|
|
DealStage(
|
|
stage_id=stage_id,
|
|
title=str(item.get("NAME") or raw_stage_id),
|
|
semantics=semantics,
|
|
)
|
|
)
|
|
seen_stage_ids.add(stage_id)
|
|
|
|
result = tuple(stages)
|
|
self._stage_cache[key] = (time.monotonic() + 300, result)
|
|
return result
|
|
|
|
async def _stage_map(
|
|
self,
|
|
binding: Binding,
|
|
category_id: int
|
|
) -> dict[str, str]:
|
|
stages = await self._stages(binding, category_id)
|
|
return {stage.stage_id: stage.title for stage in stages}
|
|
|
|
@staticmethod
|
|
def _stage_semantics(item: dict) -> str:
|
|
semantics = str(item.get("SEMANTICS") or "").upper()
|
|
if semantics in {"P", "S", "F"}:
|
|
return semantics
|
|
|
|
extra_semantics = str(
|
|
(item.get("EXTRA") or {}).get("SEMANTICS") or ""
|
|
).lower()
|
|
return {
|
|
"process": "P",
|
|
"success": "S",
|
|
"failure": "F",
|
|
}.get(extra_semantics, "P")
|
|
|
|
async def stage_name(
|
|
self,
|
|
binding: Binding,
|
|
category_id: int,
|
|
stage_id: str
|
|
) -> str | None:
|
|
stages = await self._stage_map(binding, category_id)
|
|
return stages.get(stage_id)
|
|
|
|
async def get(self, binding: Binding, deal_id: str) -> dict | None:
|
|
# https://apidocs.bitrix24.ru/api-reference/crm/deals/crm-deal-get.html
|
|
data = await self.bitrix.call(binding, "crm.deal.get", {"id": deal_id})
|
|
deal = data.get("result")
|
|
if not deal:
|
|
return None
|
|
|
|
client = await self.get_client_info(binding, deal)
|
|
deal["CLIENT_NAME"] = client.name
|
|
deal["CLIENT_COMPANY"] = client.company
|
|
deal["CLIENT_PHONE"] = client.phone
|
|
deal["SOURCE_NAME"] = await self.get_source_name(
|
|
binding, str(deal.get("SOURCE_ID") or "")
|
|
)
|
|
deal["STAGE_NAME"] = await self.stage_name(
|
|
binding,
|
|
int(deal.get("CATEGORY_ID") or 0),
|
|
str(deal.get("STAGE_ID") or "")
|
|
)
|
|
next_stage = await self._next_stage(binding, deal)
|
|
if next_stage:
|
|
deal["NEXT_STAGE_ID"] = next_stage.stage_id
|
|
deal["NEXT_STAGE_NAME"] = next_stage.title
|
|
deal["NEXT_STAGE_IS_FINAL"] = next_stage.is_final
|
|
return deal
|
|
|
|
async def prepare_stage_advance(
|
|
self,
|
|
binding: Binding,
|
|
deal_id: str
|
|
) -> DealStageAdvance:
|
|
deal = await self._get_raw(binding, deal_id)
|
|
if not deal:
|
|
raise RuntimeError("Сделка не найдена")
|
|
self._ensure_responsible(binding, deal)
|
|
|
|
next_stage = await self._next_stage(binding, deal)
|
|
if not next_stage:
|
|
raise RuntimeError("Сделка уже находится на финальной стадии")
|
|
|
|
return DealStageAdvance(
|
|
deal_id=deal_id,
|
|
current_stage_id=str(deal.get("STAGE_ID") or ""),
|
|
target_stage_id=next_stage.stage_id,
|
|
target_stage_title=next_stage.title,
|
|
is_final=next_stage.is_final,
|
|
)
|
|
|
|
async def advance_stage(
|
|
self,
|
|
binding: Binding,
|
|
deal_id: str,
|
|
expected_stage_id: str,
|
|
expected_target_stage_id: str,
|
|
comment: str | None,
|
|
) -> DealStageAdvance:
|
|
async with self._deal_lock(binding, deal_id):
|
|
current = await self._get_raw(binding, deal_id)
|
|
if not current:
|
|
raise RuntimeError("Сделка не найдена")
|
|
self._ensure_responsible(binding, current)
|
|
|
|
current_stage_id = str(current.get("STAGE_ID") or "")
|
|
if current_stage_id != expected_stage_id:
|
|
raise DealStageConflict(
|
|
"Стадия уже изменилась. Обновите карточку сделки."
|
|
)
|
|
|
|
next_stage = await self._next_stage(binding, current)
|
|
if (
|
|
not next_stage
|
|
or next_stage.stage_id != expected_target_stage_id
|
|
):
|
|
raise DealStageConflict(
|
|
"Набор стадий изменился. Откройте сделку заново."
|
|
)
|
|
|
|
advance = DealStageAdvance(
|
|
deal_id=deal_id,
|
|
current_stage_id=current_stage_id,
|
|
target_stage_id=next_stage.stage_id,
|
|
target_stage_title=next_stage.title,
|
|
is_final=next_stage.is_final,
|
|
)
|
|
# https://apidocs.bitrix24.ru/api-reference/crm/deals/crm-deal-update.html
|
|
await self.bitrix.call(
|
|
binding,
|
|
"crm.deal.update",
|
|
{
|
|
"id": deal_id,
|
|
"fields": {"STAGE_ID": next_stage.stage_id},
|
|
"params": {"REGISTER_HISTORY_EVENT": "Y"},
|
|
},
|
|
)
|
|
|
|
updated = await self._get_raw(binding, deal_id)
|
|
if str((updated or {}).get(
|
|
"STAGE_ID") or "") != next_stage.stage_id:
|
|
raise DealStageConflict(
|
|
"Стадия изменилась одновременно с обновлением."
|
|
)
|
|
|
|
if comment:
|
|
try:
|
|
# Комментарий добавляется в таймлайн, не затирая COMMENTS.
|
|
await self.bitrix.call(
|
|
binding,
|
|
"crm.timeline.comment.add",
|
|
{
|
|
"fields": {
|
|
"ENTITY_ID": int(deal_id),
|
|
"ENTITY_TYPE": "deal",
|
|
"COMMENT": comment,
|
|
}
|
|
},
|
|
)
|
|
except Exception as error:
|
|
raise DealCommentSaveError(advance) from error
|
|
|
|
return advance
|
|
|
|
async def _next_stage(
|
|
self,
|
|
binding: Binding,
|
|
deal: dict
|
|
) -> DealStage | None:
|
|
current_stage_id = str(deal.get("STAGE_ID") or "")
|
|
current_semantics = str(
|
|
deal.get("STAGE_SEMANTIC_ID") or ""
|
|
).upper()
|
|
if current_semantics in {"S", "F"}:
|
|
return None
|
|
|
|
stages = await self._stages(
|
|
binding,
|
|
int(deal.get("CATEGORY_ID") or 0),
|
|
)
|
|
current_index = next(
|
|
(
|
|
index
|
|
for index, stage in enumerate(stages)
|
|
if stage.stage_id == current_stage_id
|
|
),
|
|
None,
|
|
)
|
|
if current_index is None or stages[current_index].is_final:
|
|
return None
|
|
if current_index + 1 >= len(stages):
|
|
return None
|
|
return stages[current_index + 1]
|
|
|
|
@staticmethod
|
|
def _ensure_responsible(binding: Binding, deal: dict) -> None:
|
|
responsible_id = str(deal.get("ASSIGNED_BY_ID") or "")
|
|
if responsible_id != str(binding.bitrix_user_id):
|
|
raise DealAdvanceForbidden(
|
|
"Переводить сделку может только ответственный за нее."
|
|
)
|
|
|
|
async def history(
|
|
self,
|
|
binding: Binding,
|
|
deal_id: str,
|
|
limit: int = 5
|
|
) -> list[dict]:
|
|
# https://apidocs.bitrix24.ru/api-reference/crm/crm-stage-history-list.html
|
|
data = await self.bitrix.call(
|
|
binding,
|
|
"crm.stagehistory.list",
|
|
{
|
|
"entityTypeId": 2,
|
|
"order": {"ID": "DESC"},
|
|
"filter": {"OWNER_ID": int(deal_id)},
|
|
"select": [
|
|
"ID",
|
|
"TYPE_ID",
|
|
"CATEGORY_ID",
|
|
"STAGE_ID",
|
|
"CREATED_TIME"
|
|
],
|
|
"start": 0
|
|
}
|
|
)
|
|
result = data.get("result") or {}
|
|
events = list(result.get("items") or [])[:limit]
|
|
for event in events:
|
|
event["STAGE_NAME"] = await self.stage_name(
|
|
binding,
|
|
int(event.get("CATEGORY_ID") or 0),
|
|
str(event.get("STAGE_ID") or "")
|
|
)
|
|
return events
|
|
|
|
async def remind_to_call(
|
|
self,
|
|
binding: Binding,
|
|
deal_id: str
|
|
) -> datetime:
|
|
deadline = datetime.now(UTC) + timedelta(hours=1)
|
|
# https://apidocs.bitrix24.ru/api-reference/crm/timeline/activities/todo/crm-activity-todo-add.html
|
|
await self.bitrix.call(
|
|
binding,
|
|
"crm.activity.todo.add",
|
|
{
|
|
"ownerTypeId": 2,
|
|
"ownerId": int(deal_id),
|
|
"deadline": deadline.isoformat(),
|
|
"title": "Позвонить клиенту",
|
|
"description": f"Отложенный звонок по сделке #{deal_id}",
|
|
"responsibleId": binding.bitrix_user_id,
|
|
"pingOffsets": [0]
|
|
}
|
|
)
|
|
return deadline
|
|
|
|
async def take_to_work(
|
|
self,
|
|
binding: Binding,
|
|
deal_id: str,
|
|
expected_responsible_id: str
|
|
) -> bool:
|
|
async with self._deal_lock(binding, deal_id):
|
|
current = await self._get_raw(binding, deal_id)
|
|
if not current:
|
|
raise RuntimeError("Сделка не найдена")
|
|
|
|
responsible_id = str(current.get("ASSIGNED_BY_ID") or "")
|
|
target_id = str(binding.bitrix_user_id)
|
|
if responsible_id == target_id:
|
|
return False
|
|
if responsible_id != expected_responsible_id:
|
|
raise DealAssignmentConflict(
|
|
"Ответственный уже изменился. Обновите карточку сделки."
|
|
)
|
|
|
|
target_stage_id = await self._take_to_work_stage_id(
|
|
binding,
|
|
int(current.get("CATEGORY_ID") or 0)
|
|
)
|
|
|
|
# https://apidocs.bitrix24.com/api-reference/crm/deals/crm-deal-update.html
|
|
await self.bitrix.call(
|
|
binding,
|
|
"crm.deal.update",
|
|
{
|
|
"id": deal_id,
|
|
"fields": {
|
|
"ASSIGNED_BY_ID": binding.bitrix_user_id,
|
|
"STAGE_ID": target_stage_id
|
|
},
|
|
"params": {"REGISTER_HISTORY_EVENT": "Y"}
|
|
},
|
|
)
|
|
|
|
# REST Bitrix не поддерживает условный UPDATE, поэтому проверяем результат.
|
|
updated = await self._get_raw(binding, deal_id)
|
|
if str((updated or {}).get("ASSIGNED_BY_ID") or "") != target_id:
|
|
raise DealAssignmentConflict(
|
|
"Ответственный изменился одновременно с назначением."
|
|
)
|
|
return True
|
|
|
|
async def _take_to_work_stage_id(
|
|
self,
|
|
binding: Binding,
|
|
category_id: int
|
|
) -> str:
|
|
stages = await self._stage_map(binding, category_id)
|
|
candidates = [self.take_to_work_stage_id]
|
|
if category_id and ":" not in self.take_to_work_stage_id:
|
|
candidates.append(
|
|
f"C{category_id}:{self.take_to_work_stage_id}"
|
|
)
|
|
|
|
target = next((item for item in candidates if item in stages), None)
|
|
if target:
|
|
return target
|
|
|
|
raise RuntimeError(
|
|
"Стадия для взятия в работу "
|
|
f"{self.take_to_work_stage_id} не найдена в Битриксе"
|
|
)
|
|
|
|
async def _get_raw(self, binding: Binding, deal_id: str) -> dict | None:
|
|
data = await self.bitrix.call(binding, "crm.deal.get", {"id": deal_id})
|
|
return data.get("result") or None
|
|
|
|
@asynccontextmanager
|
|
async def _deal_lock(
|
|
self,
|
|
binding: Binding,
|
|
deal_id: str
|
|
) -> AsyncGenerator[None]:
|
|
key = (binding.member_id, deal_id)
|
|
async with self._deal_locks_guard:
|
|
entry = self._deal_locks.get(key)
|
|
if entry is None:
|
|
entry = _DealLockEntry(asyncio.Lock())
|
|
self._deal_locks[key] = entry
|
|
entry.users += 1
|
|
|
|
await entry.lock.acquire()
|
|
try:
|
|
yield
|
|
finally:
|
|
entry.lock.release()
|
|
async with self._deal_locks_guard:
|
|
entry.users -= 1
|
|
if entry.users == 0:
|
|
self._deal_locks.pop(key, None)
|
|
|
|
async def get_client_info(self, binding: Binding, deal: dict) -> ClientInfo:
|
|
contact = None
|
|
company = None
|
|
|
|
contact_id = str(deal.get("CONTACT_ID") or "")
|
|
if contact_id:
|
|
# https://apidocs.bitrix24.com/api-reference/crm/contacts/crm-contact-get.html
|
|
contact = await self._entity(binding, "crm.contact.get", contact_id)
|
|
|
|
company_id = str(deal.get("COMPANY_ID") or "")
|
|
if company_id:
|
|
# https://apidocs.bitrix24.com/api-reference/crm/companies/crm-company-get.html
|
|
company = await self._entity(binding, "crm.company.get", company_id)
|
|
|
|
return ClientInfo(
|
|
name=self._contact_name(contact),
|
|
company=self._company_name(company),
|
|
phone=(
|
|
self._phone_from_entity(contact) or self._phone_from_entity(
|
|
company)
|
|
)
|
|
)
|
|
|
|
async def get_source_name(self, binding: Binding,
|
|
source_id: str) -> str | None:
|
|
if not source_id:
|
|
return None
|
|
|
|
try:
|
|
# https://apidocs.bitrix24.ru/api-reference/crm/status/crm-status-list.html
|
|
data = await self.bitrix.call(
|
|
binding,
|
|
"crm.status.list",
|
|
{
|
|
"filter": {
|
|
"ENTITY_ID": "SOURCE",
|
|
"STATUS_ID": source_id
|
|
}
|
|
}
|
|
)
|
|
except Exception:
|
|
logger.exception("Failed to load Bitrix source name")
|
|
return None
|
|
|
|
sources = data.get("result", [])
|
|
return str(sources[0].get("NAME") or "") or None if sources else None
|
|
|
|
async def _entity(self, binding: Binding, method: str,
|
|
entity_id: str) -> dict:
|
|
data = await self.bitrix.call(binding, method, {"id": entity_id})
|
|
return data.get("result", {}) or {}
|
|
|
|
@staticmethod
|
|
def _phone_from_entity(entity: dict | None) -> str | None:
|
|
phones = (entity or {}).get("PHONE") or []
|
|
return str(phones[0].get("VALUE") or "") or None if phones else None
|
|
|
|
@staticmethod
|
|
def _contact_name(contact: dict | None) -> str | None:
|
|
if not contact:
|
|
return None
|
|
parts = [
|
|
str(contact.get("LAST_NAME") or "").strip(),
|
|
str(contact.get("NAME") or "").strip(),
|
|
str(contact.get("SECOND_NAME") or "").strip()
|
|
]
|
|
return " ".join(part for part in parts if part) or None
|
|
|
|
@staticmethod
|
|
def _company_name(company: dict | None) -> str | None:
|
|
if not company:
|
|
return None
|
|
return str(company.get("TITLE") or "").strip() or None
|