""" Клиент 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)})