intra_max_chatbot/external/intradesk_api.py

760 lines
28 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
Intradesk REST API: создание заявок и чтение истории/комментариев.
Эндпоинты:
POST /changes/v3/tasks — создать заявку
PUT /changes/v3/tasks — изменить заявку / добавить комментарий
GET /taskhistory/api/v2.0/lifetime/{id}/full — история (комментарии, смена статуса...)
Кэширование:
Все GET-запросы кэшируются в памяти с TTL, чтобы не дёргать API при каждом
переходе между экранами. Мутации (create/comment) сбрасывают нужные ключи.
"""
from __future__ import annotations
import asyncio
import json
import logging
import re
import time
from typing import Any
import httpx
import config
logger = logging.getLogger(__name__)
# ── HTML-очистка ──────────────────────────────────────────────────────────────
_TAG_RE = re.compile(r"<[^>]+>")
def _strip_html(text: str) -> str:
"""Убрать HTML-теги и лишние пробелы из текста комментария."""
if not text:
return text
cleaned = _TAG_RE.sub("", text)
cleaned = re.sub(r"\s{2,}", " ", cleaned).strip()
return cleaned
# ── TTL-кэш ───────────────────────────────────────────────────────────────────
# Структура: ключ → (expires_at: float, данные: Any)
_cache: dict[str, tuple[float, Any]] = {}
# TTL по типу данных
_TTL_COUNT = 60.0 # count_client_open_tasks
_TTL_LAST_COMMENT = 60.0 # get_client_last_comment
_TTL_OPEN_TASKS = 30.0 # get_client_open_tasks (список страниц)
_TTL_TASK_COMMENT = 30.0 # get_task_last_comment (история одной задачи)
def _cache_get(key: str) -> tuple[bool, Any]:
"""Вернуть (True, значение) если ключ свежий, иначе (False, None)."""
entry = _cache.get(key)
if entry is None:
return False, None
expires_at, data = entry
if time.monotonic() > expires_at:
del _cache[key]
return False, None
return True, data
def _cache_set(key: str, value: Any, ttl: float) -> None:
_cache[key] = (time.monotonic() + ttl, value)
def _cache_delete_prefix(prefix: str) -> None:
"""Удалить все ключи начинающиеся с prefix."""
for k in [k for k in _cache if k.startswith(prefix)]:
del _cache[k]
def invalidate_client_cache(client_id: int) -> None:
"""
Сбросить кэш клиента после мутации (создание заявки и т.п.).
Очищает: счётчик, последний комментарий, все страницы списка заявок.
"""
_cache_delete_prefix(f"count:{client_id}")
_cache_delete_prefix(f"last_comment:{client_id}")
_cache_delete_prefix(f"open_tasks:{client_id}:")
logger.debug("cache invalidated for client_id=%s", client_id)
def invalidate_task_comment(task_id: int) -> None:
"""Сбросить кэш истории конкретной задачи после добавления комментария."""
_cache.pop(f"task_comment:{task_id}", None)
_cache_delete_prefix(f"task_comments:{task_id}:")
logger.debug("cache invalidated for task_id=%s", task_id)
# ── Загрузка файлов ───────────────────────────────────────────────────────────
def _default_filename(att_type: str) -> str:
t = str(att_type).upper()
if "IMAGE" in t or "PHOTO" in t:
return "image.jpg"
if "VIDEO" in t:
return "video.mp4"
if "AUDIO" in t:
return "audio.wav"
return "file.bin"
async def upload_file_to_intradesk(
file_bytes: bytes,
filename: str,
content_type: str,
task_id: int = 0,
target: str = "Description",
) -> dict | None:
"""
Загрузить файл в Intradesk.
task_id=0 — для новой заявки (описание).
task_id=N — для существующей заявки (комментарий).
target — "Description" (30) или "Comment" (20).
Возвращает metadata файла {"id", "name", "contentType", ...} или None.
"""
key = config.INTRADESK_CHANGES_API_KEY
if not key:
logger.warning("upload_file_to_intradesk: INTRADESK_CHANGES_API_KEY не задан")
return None
url = (
f"{config.INTRADESK_BASE_URL.rstrip('/')}"
f"/files/api/tasks/{task_id}/files/target/{target}"
f"?ApiKey={key}"
)
if config.DRY_RUN:
print(
f"\n[DRY-RUN] upload_file_to_intradesk — {filename} ({len(file_bytes)} bytes)\n"
f" POST {url}\n"
)
return None
try:
async with httpx.AsyncClient() as client:
resp = await client.post(
url,
files={"file": (filename, file_bytes, content_type)},
timeout=30.0,
)
if resp.status_code not in (200, 201):
logger.error(
"upload_file_to_intradesk → HTTP %s: %s",
resp.status_code, resp.text[:300],
)
return None
data = resp.json()
meta = data[0] if isinstance(data, list) and data else data
if isinstance(meta, dict):
logger.info("upload_file_to_intradesk ok: %s → id=%s", filename, meta.get("id"))
return meta if isinstance(meta, dict) else None
except Exception as exc:
logger.exception("upload_file_to_intradesk error: %s", exc)
return None
async def download_and_upload(
file_url: str,
filename: str | None,
content_type_hint: str | None,
task_id: int = 0,
target: str = "Description",
) -> dict | None:
"""Скачать файл из MAX CDN и загрузить в Intradesk."""
try:
async with httpx.AsyncClient() as client:
resp = await client.get(file_url, timeout=30.0, follow_redirects=True)
if resp.status_code != 200:
logger.warning("download_and_upload: GET %s → HTTP %s", file_url, resp.status_code)
return None
file_bytes = resp.content
ct_raw = resp.headers.get("content-type", content_type_hint or "application/octet-stream")
ct = ct_raw.split(";")[0].strip()
fname = filename or "attachment"
except Exception as exc:
logger.warning("download_and_upload download error: %s", exc)
return None
return await upload_file_to_intradesk(file_bytes, fname, ct, task_id=task_id, target=target)
async def upload_attachments(
attachments: list[dict],
task_id: int = 0,
target: str = "Description",
) -> list[dict]:
"""
Скачать и загрузить все вложения в Intradesk параллельно.
Возвращает те же вложения, обогащённые полем ``intradesk_meta``.
Если загрузка файла не удалась — поле будет None, вложение остаётся в списке.
"""
async def _process(att: dict) -> dict:
url = att.get("url")
if not url:
return att
filename = att.get("filename") or _default_filename(att.get("type", ""))
meta = await download_and_upload(url, filename, None, task_id=task_id, target=target)
return {**att, "intradesk_meta": meta}
results = await asyncio.gather(*[_process(a) for a in attachments], return_exceptions=True)
enriched = []
for orig, result in zip(attachments, results):
enriched.append(result if isinstance(result, dict) else orig)
return enriched
# ── API-функции ───────────────────────────────────────────────────────────────
async def get_task_last_comment(task_id: int) -> dict | None:
"""
Получить последний комментарий к заявке из Intradesk.
Возвращает dict: {text, author, date} или None если нет комментариев.
Результат кэшируется на _TTL_TASK_COMMENT секунд.
"""
ck = f"task_comment:{task_id}"
hit, cached = _cache_get(ck)
if hit:
return cached
api_key = config.INTRADESK_TASKS_API_KEY
if not api_key:
return None
url = (
f"{config.INTRADESK_BASE_URL.rstrip('/')}"
f"/taskhistory/api/v2.0/lifetime/{task_id}/full"
f"?ApiKey={api_key}&sortDirection=Desc&top=10&events=50"
)
try:
async with httpx.AsyncClient() as client:
resp = await client.get(url, timeout=15.0)
if resp.status_code != 200:
logger.warning(
"get_task_last_comment task=%s → HTTP %s", task_id, resp.status_code
)
return None
data = resp.json()
except Exception as exc:
logger.warning("get_task_last_comment error: %s", exc)
return None
result: dict | None = None
for entry in data.get("data") or []:
for ev in (entry.get("events") or {}).get("data") or []:
if ev.get("blockname") == "comment" and ev.get("type") == 50:
raw_date = entry.get("eventat") or ""
result = {
"text": _strip_html(ev.get("stringvalue") or ""),
"author": entry.get("username") or "",
"date": raw_date[:16].replace("T", " "),
}
break
if result:
break
_cache_set(ck, result, _TTL_TASK_COMMENT)
return result
async def get_task_comments(task_id: int, count: int = 10) -> list[dict]:
"""
Получить последние комментарии к заявке.
Возвращает список {text, author, date}, до count штук, новые сначала.
Результат кэшируется на _TTL_TASK_COMMENT секунд.
"""
ck = f"task_comments:{task_id}:{count}"
hit, cached = _cache_get(ck)
if hit:
return cached
api_key = config.INTRADESK_TASKS_API_KEY
if not api_key:
return []
url = (
f"{config.INTRADESK_BASE_URL.rstrip('/')}"
f"/taskhistory/api/v2.0/lifetime/{task_id}/full"
f"?ApiKey={api_key}&sortDirection=Desc&top={count * 3}&events=50"
)
try:
async with httpx.AsyncClient() as client:
resp = await client.get(url, timeout=15.0)
if resp.status_code != 200:
logger.warning("get_task_comments task=%s → HTTP %s", task_id, resp.status_code)
return []
data = resp.json()
except Exception as exc:
logger.warning("get_task_comments error: %s", exc)
return []
result: list[dict] = []
for entry in data.get("data") or []:
for ev in (entry.get("events") or {}).get("data") or []:
if ev.get("blockname") == "comment" and ev.get("type") == 50:
raw_date = entry.get("eventat") or ""
result.append({
"text": _strip_html(ev.get("stringvalue") or ""),
"author": entry.get("username") or "",
"date": raw_date[:16].replace("T", " "),
})
break # одна запись = одно событие комментария
if len(result) >= count:
break
_cache_set(ck, result, _TTL_TASK_COMMENT)
return result
async def count_client_open_tasks(client_id: int) -> int:
"""
Количество незакрытых заявок клиента.
Intradesk не возвращает @odata.count при $top=0, поэтому берём $top=1.
Результат кэшируется на _TTL_COUNT секунд.
"""
ck = f"count:{client_id}"
hit, cached = _cache_get(ck)
if hit:
return cached
api_key = config.INTRADESK_TASKS_API_KEY
if not api_key:
return 0
url = (
f"{config.INTRADESK_BASE_URL.rstrip('/')}/tasklist/odata/v3/tasks"
f"?ApiKey={api_key}"
f"&$filter=(clientid eq {client_id}) and (closedat eq null) and (isarchived eq false)"
f"&$top=1&$count=true"
)
try:
async with httpx.AsyncClient() as client:
resp = await client.get(url, timeout=10.0)
if resp.status_code != 200:
return 0
data = resp.json()
count = data.get("@odata.count")
result = int(count) if count is not None else len(data.get("value") or [])
except Exception:
return 0
_cache_set(ck, result, _TTL_COUNT)
return result
async def get_client_last_comment(client_id: int) -> dict | None:
"""
Найти последний комментарий по любой незакрытой заявке клиента.
Берём задачу с самым свежим updatedat → ищем комментарий в её истории.
Возвращает {text, author, date, task_number} или None.
Результат кэшируется на _TTL_LAST_COMMENT секунд.
"""
ck = f"last_comment:{client_id}"
hit, cached = _cache_get(ck)
if hit:
return cached
api_key = config.INTRADESK_TASKS_API_KEY
if not api_key:
return None
url = (
f"{config.INTRADESK_BASE_URL.rstrip('/')}/tasklist/odata/v3/tasks"
f"?ApiKey={api_key}"
f"&$filter=(clientid eq {client_id}) and (closedat eq null) and (isarchived eq false)"
f"&$orderby=updatedat desc&$top=1"
)
try:
async with httpx.AsyncClient() as client:
resp = await client.get(url, timeout=15.0)
if resp.status_code != 200:
return None
data = resp.json()
except Exception:
return None
tasks = data.get("value") or []
if not tasks:
_cache_set(ck, None, _TTL_LAST_COMMENT)
return None
task = tasks[0]
task_id = task.get("id")
task_number = task.get("tasknumber")
if not task_id:
_cache_set(ck, None, _TTL_LAST_COMMENT)
return None
# get_task_last_comment имеет собственный кэш — двойного запроса не будет
comment = await get_task_last_comment(task_id)
if comment:
comment = {**comment, "task_number": task_number}
_cache_set(ck, comment, _TTL_LAST_COMMENT)
return comment
async def get_client_open_tasks(
client_id: int,
page: int = 0,
page_size: int = 5,
) -> dict:
"""
Получить незакрытые заявки клиента из Intradesk API.
Возвращает {"tasks": [...], "total": int}.
Последний комментарий загружается параллельно через history API.
Результат кэшируется на _TTL_OPEN_TASKS секунд.
"""
import asyncio
ck = f"open_tasks:{client_id}:{page}:{page_size}"
hit, cached = _cache_get(ck)
if hit:
return cached
api_key = config.INTRADESK_TASKS_API_KEY
if not api_key:
return {"tasks": [], "total": 0}
skip = page * page_size
url = (
f"{config.INTRADESK_BASE_URL.rstrip('/')}/tasklist/odata/v3/tasks"
f"?ApiKey={api_key}"
f"&$filter=(clientid eq {client_id}) and (closedat eq null) and (isarchived eq false)"
f"&$orderby=updatedat desc"
f"&$top={page_size}&$skip={skip}"
f"&$count=true"
)
try:
async with httpx.AsyncClient() as client:
resp = await client.get(url, timeout=20.0)
if resp.status_code != 200:
logger.warning("get_client_open_tasks client=%s → HTTP %s", client_id, resp.status_code)
return {"tasks": [], "total": 0}
data = resp.json()
except Exception as exc:
logger.warning("get_client_open_tasks error: %s", exc)
return {"tasks": [], "total": 0}
total = data.get("@odata.count") or 0
raw_tasks = data.get("value") or []
# Словари для разрешения имён статусов
dicts_map: dict[int, str] = {}
for d in data.get("dictionaries") or []:
did = d.get("id")
dname = d.get("name") or ""
if isinstance(dname, dict):
dname = dname.get("ru") or dname.get("en") or ""
if did:
dicts_map[did] = dname
base_tasks = []
for t in raw_tasks:
status_id = t.get("status")
base_tasks.append({
"id": t.get("id"),
"task_number": t.get("tasknumber"),
"name": t.get("name") or "",
"description": _strip_html(t.get("description") or ""),
"status_name": dicts_map.get(status_id, "") if status_id else "",
"created_at": t.get("createdat") or "",
"updated_at": t.get("updatedat") or "",
})
# Параллельно загружаем последний комментарий (используют собственный кэш)
task_ids = [t["id"] for t in base_tasks if t["id"]]
comments = await asyncio.gather(
*[get_task_last_comment(tid) for tid in task_ids],
return_exceptions=True,
)
tasks = []
for task, comment in zip(base_tasks, comments):
task["last_comment"] = comment if isinstance(comment, dict) else None
tasks.append(task)
result = {"tasks": tasks, "total": total}
_cache_set(ck, result, _TTL_OPEN_TASKS)
return result
async def fetch_all_open_tasks_lightweight() -> list[dict]:
"""
Получить ВСЕ открытые задачи (по всем клиентам) одним запросом.
Используется нотифайером — без кэша, без комментариев.
Возвращает [{id, task_number, client_id, updatedat}].
Нотифайер сам группирует результат по client_id.
"""
api_key = config.INTRADESK_TASKS_API_KEY
if not api_key:
return []
url = (
f"{config.INTRADESK_BASE_URL.rstrip('/')}/tasklist/odata/v3/tasks"
f"?ApiKey={api_key}"
f"&$filter=(closedat eq null) and (isarchived eq false)"
f"&$orderby=updatedat desc"
f"&$top=500"
)
try:
async with httpx.AsyncClient() as client:
resp = await client.get(url, timeout=30.0)
if resp.status_code != 200:
logger.warning("fetch_all_open_tasks_lightweight → HTTP %s", resp.status_code)
return []
data = resp.json()
except Exception as exc:
logger.warning("fetch_all_open_tasks_lightweight error: %s", exc)
return []
return [
{
"id": t.get("id"),
"task_number": t.get("tasknumber"),
"client_id": t.get("clientid"),
"updatedat": t.get("updatedat") or "",
}
for t in (data.get("value") or [])
if t.get("id") and t.get("clientid")
]
def _extract_last_comment_from_lifetime(lifetime: dict | None) -> dict | None:
"""Извлечь последний комментарий из поля lifetime задачи."""
if not lifetime:
return None
for entry in (lifetime.get("data") or []):
for ev in (entry.get("events") or {}).get("data") or []:
if ev.get("blockname") == "comment" and ev.get("type") == 50:
raw_date = entry.get("eventat") or ""
return {
"text": _strip_html(ev.get("stringvalue") or ""),
"author": entry.get("username") or "",
"date": raw_date[:16].replace("T", " "),
}
return None
async def add_task_comment(
task_number: int,
text: str,
contact_person_id: int | None = None,
attachments: list[dict] | None = None,
) -> bool:
"""
Добавить комментарий к заявке (PUT /changes/v3/tasks).
attachments — список вложений с полем ``intradesk_meta`` (заполняется
через upload_attachments с target="Comment" перед вызовом этой функции).
Возвращает True при успехе, False при ошибке.
Кэш задачи сбрасывается вызывающей стороной через invalidate_task_comment.
"""
key = config.INTRADESK_CHANGES_API_KEY
if not key:
logger.warning("add_task_comment: INTRADESK_CHANGES_API_KEY не задан")
return False
blocks: dict[str, str] = {
"comment": json.dumps({"value": text}, ensure_ascii=False),
}
add_files = [a["intradesk_meta"] for a in (attachments or []) if a.get("intradesk_meta")]
if add_files:
blocks["attachments"] = json.dumps(
{"value": {"addFiles": add_files, "deleteFileIds": []}},
ensure_ascii=False,
)
body = {"number": task_number, "blocks": blocks, "Channel": "api"}
url = f"{config.INTRADESK_BASE_URL.rstrip('/')}/changes/v3/tasks?ApiKey={key}"
if contact_person_id is not None:
url += f"&userid={contact_person_id}"
if config.DRY_RUN:
print(
"\n[DRY-RUN] add_task_comment — запрос НЕ отправлен\n"
f" PUT {url}\n"
f" Body: {json.dumps(body, ensure_ascii=False, indent=2)}\n"
)
return False
try:
async with httpx.AsyncClient() as client:
resp = await client.put(
url,
json=body,
headers={"Content-Type": "application/json"},
timeout=20.0,
)
if resp.status_code not in (200, 201):
logger.error(
"add_task_comment → HTTP %s: %s",
resp.status_code,
resp.text[:500],
)
return False
logger.info("add_task_comment ok: task_number=%s", task_number)
return True
except Exception as exc:
logger.exception("add_task_comment error: %s", exc)
return False
async def close_task(
task_number: int,
contact_person_id: int | None = None,
) -> bool:
"""
Закрыть заявку — установить статус INTRADESK_CLOSE_STATUS_ID (PUT /changes/v3/tasks).
Возвращает True при успехе, False при ошибке или незаданном статусе.
"""
status_id = config.INTRADESK_CLOSE_STATUS_ID
if not status_id:
logger.warning("close_task: INTRADESK_CLOSE_STATUS_ID не задан")
return False
key = config.INTRADESK_CHANGES_API_KEY
if not key:
logger.warning("close_task: INTRADESK_CHANGES_API_KEY не задан")
return False
blocks: dict[str, str] = {
"status": json.dumps({"value": int(status_id)}, ensure_ascii=False),
}
body = {"number": task_number, "blocks": blocks, "Channel": "api"}
url = f"{config.INTRADESK_BASE_URL.rstrip('/')}/changes/v3/tasks?ApiKey={key}"
if contact_person_id is not None:
url += f"&userid={contact_person_id}"
if config.DRY_RUN:
print(
"\n[DRY-RUN] close_task — запрос НЕ отправлен\n"
f" PUT {url}\n"
f" Body: {json.dumps(body, ensure_ascii=False, indent=2)}\n"
)
return False
try:
async with httpx.AsyncClient() as client:
resp = await client.put(
url,
json=body,
headers={"Content-Type": "application/json"},
timeout=20.0,
)
if resp.status_code not in (200, 201):
logger.error("close_task → HTTP %s: %s", resp.status_code, resp.text[:500])
return False
logger.info("close_task ok: task_number=%s", task_number)
return True
except Exception as exc:
logger.exception("close_task error: %s", exc)
return False
async def create_intradesk_task(
name: str,
description: str,
client_intradesk_id: int | None = None,
contact_person_id: int | None = None,
attachments: list[dict] | None = None,
) -> dict | None:
"""
Создать заявку в Intradesk.
attachments — список вложений с полем ``intradesk_meta`` (заполняется
через upload_attachments с target="Description" перед вызовом этой функции).
Возвращает ответ API (dict с полями id, tasknumber и др.) или None при ошибке.
Кэш клиента сбрасывается вызывающей стороной через invalidate_client_cache.
"""
key = config.INTRADESK_CHANGES_API_KEY
if not key:
logger.warning("create_intradesk_task: INTRADESK_CHANGES_API_KEY не задан")
return None
blocks: dict[str, str] = {
"name": json.dumps({"value": name}, ensure_ascii=False),
"description": json.dumps({"value": description}, ensure_ascii=False),
}
add_files = [a["intradesk_meta"] for a in (attachments or []) if a.get("intradesk_meta")]
if add_files:
blocks["attachments"] = json.dumps(
{"value": {"addFiles": add_files, "deleteFileIds": []}},
ensure_ascii=False,
)
if client_intradesk_id is not None and contact_person_id is not None:
blocks["initiator"] = json.dumps(
{"value": {"groupid": client_intradesk_id, "userid": contact_person_id}},
ensure_ascii=False,
)
elif client_intradesk_id is not None:
blocks["initiator"] = json.dumps(
{"value": {"groupid": client_intradesk_id}}, ensure_ascii=False
)
if config.INTRADESK_DEFAULT_SERVICE:
blocks["service"] = json.dumps(
{"value": config.INTRADESK_DEFAULT_SERVICE}, ensure_ascii=False
)
if config.INTRADESK_DEFAULT_TASK_TYPE:
blocks["tasktype"] = json.dumps(
{"value": int(config.INTRADESK_DEFAULT_TASK_TYPE)}, ensure_ascii=False
)
body = {"blocks": blocks, "Channel": "web"}
url = f"{config.INTRADESK_BASE_URL.rstrip('/')}/changes/v3/tasks?ApiKey={key}"
if contact_person_id is not None:
url += f"&userid={contact_person_id}"
if config.DRY_RUN:
print(
"\n[DRY-RUN] create_intradesk_task — запрос НЕ отправлен\n"
f" POST {url}\n"
f" Body: {json.dumps(body, ensure_ascii=False, indent=2)}\n"
)
return None
try:
async with httpx.AsyncClient() as client:
resp = await client.post(
url,
json=body,
headers={"Content-Type": "application/json"},
timeout=20.0,
)
if resp.status_code not in (200, 201):
logger.error(
"create_intradesk_task → HTTP %s: %s",
resp.status_code,
resp.text[:500],
)
return None
result = resp.json()
task_id = result.get("id") or result.get("Id") or result.get("ID")
task_number = (
result.get("tasknumber")
or result.get("number")
or result.get("Number")
or result.get("TaskNumber")
)
logger.info("create_intradesk_task ok: id=%s number=%s", task_id, task_number)
return result
except Exception as exc:
logger.exception("create_intradesk_task error: %s", exc)
return None