262 lines
9.9 KiB
Python
262 lines
9.9 KiB
Python
"""
|
||
Клиент Intradesk API.
|
||
|
||
Базовые URL:
|
||
GET задачи: https://apigw.intradesk.ru/tasklist/odata/v3/tasks
|
||
PUT задачи: https://apigw.intradesk.ru/changes/v3/tasks/
|
||
GET история: https://apigw.intradesk.ru/taskhistory/api/v2.0/lifetime/{task_id}/full
|
||
|
||
Аутентификация: ?ApiKey=<ключ> в строке запроса.
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
import requests
|
||
|
||
import config
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
_TASKS_URL = "https://apigw.intradesk.ru/tasklist/odata/v3/tasks"
|
||
_CHANGES_URL = "https://apigw.intradesk.ru/changes/v3/tasks/"
|
||
_HISTORY_URL = "https://apigw.intradesk.ru/taskhistory/api/v2.0/lifetime"
|
||
|
||
|
||
# ── вспомогательные ────────────────────────────────────────────────────────────
|
||
|
||
def _get(url: str, params: dict = None, timeout: int = 30) -> dict | None:
|
||
params = {**(params or {}), "ApiKey": config.API_KEY}
|
||
try:
|
||
r = requests.get(url, params=params, timeout=timeout)
|
||
r.raise_for_status()
|
||
return r.json()
|
||
except requests.exceptions.RequestException as e:
|
||
log.error(f"GET {url} failed: {e}")
|
||
return None
|
||
|
||
|
||
def _put(url: str, body: dict, timeout: int = 30) -> bool:
|
||
params = {"ApiKey": config.API_KEY}
|
||
try:
|
||
r = requests.put(url, json=body, params=params, timeout=timeout)
|
||
r.raise_for_status()
|
||
return True
|
||
except requests.exceptions.RequestException as e:
|
||
log.error(f"PUT {url} failed: {e}")
|
||
return False
|
||
|
||
|
||
def _jv(value: Any) -> str:
|
||
"""Сериализует значение в строку-JSON (формат, которого ожидает API изменений)."""
|
||
return json.dumps({"value": value}, ensure_ascii=False)
|
||
|
||
|
||
# ── задачи ─────────────────────────────────────────────────────────────────────
|
||
|
||
def get_task(task_number: int) -> dict | None:
|
||
"""
|
||
Возвращает полный объект задачи по её номеру или None при ошибке.
|
||
|
||
GET /tasklist/odata/v3/tasks?$filter=tasknumber eq {number}
|
||
"""
|
||
data = _get(_TASKS_URL, {"$filter": f"tasknumber eq {task_number}"})
|
||
if not data:
|
||
return None
|
||
items = data.get("value", [])
|
||
return items[0] if items else None
|
||
|
||
|
||
def get_tasks(filter_expr: str = "", top: int = 150, order_by: str = "updatedat desc") -> list[dict]:
|
||
"""
|
||
Возвращает список задач с поддержкой пагинации.
|
||
|
||
Примеры filter_expr:
|
||
"(status ne 68045) and (executorgroup eq 243025)"
|
||
"tasknumber eq 31"
|
||
"""
|
||
params: dict | None = {"$filter": filter_expr, "$top": top, "$orderby": order_by}
|
||
url: str | None = _TASKS_URL
|
||
all_tasks: list[dict] = []
|
||
|
||
while url:
|
||
data = _get(url, params)
|
||
if not data:
|
||
break
|
||
all_tasks.extend(data.get("value", []))
|
||
url = data.get("@odata.nextLink")
|
||
params = None # nextLink уже содержит все параметры
|
||
|
||
return all_tasks
|
||
|
||
|
||
# ── история (lifetime) ─────────────────────────────────────────────────────────
|
||
|
||
def get_task_lifetime(
|
||
task_id: int,
|
||
events: list[int] | None = None,
|
||
top: int = 50,
|
||
sort_desc: bool = True,
|
||
date_start: datetime | None = None,
|
||
date_end: datetime | None = None,
|
||
) -> list[dict]:
|
||
"""
|
||
Возвращает историю событий задачи.
|
||
|
||
GET /taskhistory/api/v2.0/lifetime/{task_id}/full
|
||
|
||
Параметры:
|
||
events — список кодов событий для фильтрации (None = все)
|
||
top — максимум записей
|
||
sort_desc — по убыванию даты
|
||
date_start — фильтр «с» (UTC datetime)
|
||
date_end — фильтр «по» (UTC datetime)
|
||
|
||
Коды событий: см. reference.EVENT_TYPES
|
||
"""
|
||
url = f"{_HISTORY_URL}/{task_id}/full"
|
||
params: dict = {
|
||
"sortDirection": "Desc" if sort_desc else "Asc",
|
||
"top": top,
|
||
}
|
||
|
||
if date_start:
|
||
params["dateStart"] = date_start.strftime("%Y-%m-%dT%H:%M:%S.0000000Z")
|
||
if date_end:
|
||
params["dateEnd"] = date_end.strftime("%Y-%m-%dT%H:%M:%S.0000000Z")
|
||
|
||
# events передаются как повторяющийся параметр: &events=10&events=50&events=30
|
||
base_params = {**params, "ApiKey": config.API_KEY}
|
||
try:
|
||
r = requests.get(
|
||
url,
|
||
params=base_params if not events else {**base_params},
|
||
timeout=30,
|
||
)
|
||
if events:
|
||
# requests не поддерживает дублирующиеся ключи напрямую — строим вручную
|
||
from urllib.parse import urlencode
|
||
qs = urlencode({**base_params}) + "".join(f"&events={e}" for e in events)
|
||
r = requests.get(f"{url}?{qs}", timeout=30)
|
||
r.raise_for_status()
|
||
return r.json().get("data", [])
|
||
except requests.exceptions.RequestException as e:
|
||
log.error(f"GET lifetime/{task_id} failed: {e}")
|
||
return []
|
||
|
||
|
||
# ── изменение задачи ───────────────────────────────────────────────────────────
|
||
|
||
def update_task(task_number: int, blocks: dict, channel: str = "api") -> bool:
|
||
"""
|
||
Изменяет поля задачи.
|
||
|
||
PUT /changes/v3/tasks/
|
||
|
||
blocks — словарь {поле: уже_сериализованное_значение_json_строкой}.
|
||
Для удобства используй вспомогательные функции ниже.
|
||
|
||
Пример:
|
||
update_task(272, {
|
||
"status": '{"value": 68051}',
|
||
"comment": '{"value":"Принято в работу"}',
|
||
})
|
||
"""
|
||
body = {"number": task_number, "blocks": blocks, "Channel": channel}
|
||
return _put(_CHANGES_URL, body)
|
||
|
||
|
||
# ── удобные обёртки ────────────────────────────────────────────────────────────
|
||
|
||
def change_status(task_number: int, status_id: int) -> bool:
|
||
"""Меняет статус задачи. status_id из reference.STATUSES."""
|
||
return update_task(task_number, {"status": _jv(status_id)})
|
||
|
||
|
||
def add_comment(task_number: int, text: str) -> bool:
|
||
"""Добавляет публичный комментарий к задаче."""
|
||
return update_task(task_number, {"comment": _jv(text)})
|
||
|
||
|
||
def add_private_comment(task_number: int, text: str) -> bool:
|
||
"""Добавляет приватный (скрытый) комментарий к задаче."""
|
||
return update_task(task_number, {"privatecomment": _jv(text)})
|
||
|
||
|
||
def assign_executor(
|
||
task_number: int,
|
||
user_id: int | None = None,
|
||
group_id: int | None = None,
|
||
) -> bool:
|
||
"""
|
||
Назначает исполнителя и/или группу исполнителей.
|
||
|
||
assign_executor(272, user_id=109747) # только сотрудник
|
||
assign_executor(272, group_id=243025) # только группа
|
||
assign_executor(272, group_id=243025, user_id=109747) # группа + сотрудник
|
||
"""
|
||
value: dict = {}
|
||
if user_id:
|
||
value["userid"] = user_id
|
||
if group_id:
|
||
value["groupid"] = group_id
|
||
if not value:
|
||
log.warning("assign_executor: нужно указать user_id или group_id")
|
||
return False
|
||
return update_task(task_number, {"executor": _jv(value)})
|
||
|
||
|
||
def set_resolution_date(task_number: int, dt: datetime) -> bool:
|
||
"""
|
||
Устанавливает плановый срок выполнения.
|
||
dt должен быть UTC-datetime.
|
||
|
||
Пример: set_resolution_date(272, datetime(2025, 3, 15, 15, 0, tzinfo=timezone.utc))
|
||
"""
|
||
iso = dt.strftime("%Y-%m-%dT%H:%M:%S.0000000Z")
|
||
return update_task(task_number, {"resolutiondateplan": _jv(iso)})
|
||
|
||
|
||
def set_service(task_number: int, service_path: str) -> bool:
|
||
"""
|
||
Устанавливает сервис задачи.
|
||
service_path — путь сервиса вида "84681/84683/" (из карточки сервиса).
|
||
"""
|
||
return update_task(task_number, {"service": _jv(service_path)})
|
||
|
||
|
||
def set_task_type(task_number: int, tasktype_id: int) -> bool:
|
||
"""Устанавливает тип задачи."""
|
||
return update_task(task_number, {"tasktype": _jv(tasktype_id)})
|
||
|
||
|
||
def set_tags(task_number: int, tag_ids: list[int]) -> bool:
|
||
"""Устанавливает теги задачи (заменяет все существующие)."""
|
||
return update_task(task_number, {"tags": _jv(tag_ids)})
|
||
|
||
|
||
def set_priority(task_number: int, priority_id: int) -> bool:
|
||
"""Устанавливает приоритет задачи. priority_id из reference.PRIORITIES."""
|
||
return update_task(task_number, {"priority": _jv(priority_id)})
|
||
|
||
|
||
def set_initiator(
|
||
task_number: int,
|
||
user_id: int | None = None,
|
||
group_id: int | None = None,
|
||
) -> bool:
|
||
"""
|
||
Устанавливает инициатора (заявителя).
|
||
|
||
set_initiator(272, user_id=109747) # сотрудник
|
||
set_initiator(272, group_id=109747) # клиент
|
||
set_initiator(272, group_id=109747, user_id=31974) # клиент + пользователь
|
||
"""
|
||
value: dict = {}
|
||
if user_id:
|
||
value["userid"] = user_id
|
||
if group_id:
|
||
value["groupid"] = group_id
|
||
return update_task(task_number, {"initiator": _jv(value)})
|