418 lines
17 KiB
Python
418 lines
17 KiB
Python
"""
|
||
Парсинг и форматирование данных из вебхуков и API Intradesk.
|
||
"""
|
||
import json
|
||
import logging
|
||
import re
|
||
from dataclasses import dataclass
|
||
from typing import Literal
|
||
|
||
import database
|
||
from intradesk import reference
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
# ── Маппинги ──────────────────────────────────────────────────────────────────
|
||
|
||
_ADDFIELD_BOOL_LABELS: dict[str, str] = {
|
||
"addfield_677b45atoBoolean": "ТО",
|
||
"addfield_ea5d305sozvon_pered_vyezdomBoolean": "Созвон перед выездом",
|
||
"addfield_2684bdczamena_oborudovaniiaBoolean": "Замена оборудования",
|
||
}
|
||
|
||
BLOCK_TO_ENTITY: dict[str, str] = {
|
||
"status": "statuses",
|
||
"priority": "priorities",
|
||
"executorgroup": "groups",
|
||
"executor": "employees",
|
||
"service": "services",
|
||
"tasktype": "tasktypes",
|
||
"initiator": "employees",
|
||
"workflow": "workflows",
|
||
"sla": "sla",
|
||
}
|
||
|
||
FIELD_LABELS: dict[str, str] = {
|
||
"status": "Статус",
|
||
"priority": "Приоритет",
|
||
"executor": "Исполнитель",
|
||
"executorgroup": "Группа",
|
||
"service": "Услуга",
|
||
"tasktype": "Тип заявки",
|
||
"initiator": "Инициатор",
|
||
"resolutiondateplan": "Плановый срок",
|
||
"closedat": "Закрыта",
|
||
"description": "Описание",
|
||
"name": "Название",
|
||
"workflow": "Бизнес-процесс",
|
||
"sla": "SLA",
|
||
"tags": "Теги",
|
||
}
|
||
|
||
IGNORED_FIELDS: set[str] = {
|
||
"lifetime", "updatedat", "updatedby", "updatedchannel",
|
||
"reactiondatefact", "resolutiondatefact", "resolutionleftminutes",
|
||
"reactiondateplan", "ismassincidentactive",
|
||
}
|
||
|
||
_QUOTE_RE = re.compile(
|
||
r'\[quote[^\]]*\].*?\[/quote\]'
|
||
r'|\{quote[^\}]*\}.*?\{/quote\}'
|
||
r'|<blockquote[^>]*>.*?</blockquote>'
|
||
r'|<quote[^>]*>.*?</quote>',
|
||
re.DOTALL | re.IGNORECASE,
|
||
)
|
||
|
||
_FILE_EXT_RE = re.compile(r'\.([a-zA-Z0-9]{1,10})$')
|
||
|
||
|
||
# ── Типы событий ──────────────────────────────────────────────────────────────
|
||
|
||
@dataclass
|
||
class ParsedEvent:
|
||
"""Нормализованное событие из истории заявки."""
|
||
kind: Literal["comment", "change"]
|
||
text: str
|
||
|
||
|
||
# ── Утилиты ───────────────────────────────────────────────────────────────────
|
||
|
||
def _shorten_name(name: str | None) -> str | None:
|
||
"""'Иванов Иван Петрович' → 'Иванов И.'"""
|
||
if not name:
|
||
return name
|
||
parts = name.strip().split()
|
||
if len(parts) >= 2:
|
||
return f"{parts[0]} {parts[1][0]}."
|
||
return name
|
||
|
||
|
||
def escape_markdown(text: str | None) -> str | None:
|
||
"""Escape user-provided text for Telegram legacy Markdown parse mode."""
|
||
if text is None:
|
||
return None
|
||
escaped = str(text).replace("\\", "\\\\")
|
||
for ch in ("_", "*", "`", "["):
|
||
escaped = escaped.replace(ch, f"\\{ch}")
|
||
return escaped
|
||
|
||
def _strip_quote(text: str) -> tuple[str, bool]:
|
||
"""Убирает блоки цитирования. Возвращает (cleaned_text, had_quote)."""
|
||
cleaned, n = _QUOTE_RE.subn("", text)
|
||
return cleaned.strip(), n > 0
|
||
|
||
|
||
def _parse_json_field(raw):
|
||
if raw is None:
|
||
return None
|
||
if isinstance(raw, (dict, list)):
|
||
return raw
|
||
if isinstance(raw, str):
|
||
try:
|
||
return json.loads(raw)
|
||
except Exception:
|
||
return raw
|
||
return raw
|
||
|
||
|
||
def _scalar(raw):
|
||
parsed = _parse_json_field(raw)
|
||
if isinstance(parsed, dict):
|
||
return parsed.get("Value") or parsed.get("Id")
|
||
return parsed
|
||
|
||
|
||
def _extract_name(v: dict | None, blockname: str = "") -> str | None:
|
||
if not v:
|
||
return None
|
||
raw_name = v.get("currentname") or v.get("name") or v.get("Name")
|
||
if isinstance(raw_name, dict):
|
||
raw_name = raw_name.get("ru") or raw_name.get("en") or None
|
||
if raw_name:
|
||
name_str = str(raw_name).strip() or None
|
||
if name_str and blockname in ("executor", "initiator"):
|
||
return _shorten_name(name_str)
|
||
return name_str
|
||
item_id = v.get("id") or v.get("Id")
|
||
entity = BLOCK_TO_ENTITY.get(blockname)
|
||
if item_id and entity:
|
||
resolved = reference.resolve(entity, item_id) or str(item_id)
|
||
if blockname in ("executor", "initiator"):
|
||
return _shorten_name(resolved)
|
||
return resolved
|
||
return str(item_id) if item_id else None
|
||
|
||
|
||
# ── Ядро: разбор одного события ───────────────────────────────────────────────
|
||
|
||
def _parse_single_event(event: dict, username: str = "") -> ParsedEvent | None:
|
||
"""
|
||
Разбирает одно сырое событие из lifetime в нормализованный ParsedEvent.
|
||
|
||
Используется как в parse_lifetime() (вебхук), так и в format_lifetime_api_events()
|
||
(прямой API). Единственный источник правды для логики разбора событий.
|
||
"""
|
||
etype = event.get("type")
|
||
blockname = event.get("blockname", "")
|
||
short_user = _shorten_name(username) or "Неизвестный"
|
||
short_user = escape_markdown(short_user) or short_user
|
||
|
||
# ── Комментарии ──────────────────────────────────────────────────────────
|
||
if etype in (50, 55):
|
||
text = event.get("stringvalue") or ""
|
||
if not text:
|
||
return None
|
||
text, had_quote = _strip_quote(text)
|
||
if not text:
|
||
return None
|
||
text = escape_markdown(text) or ""
|
||
prefix = "Ответ на прошлое: " if had_quote else ""
|
||
lock = " 🔒" if etype == 55 else ""
|
||
return ParsedEvent("comment", f"*{short_user}*{lock}: {prefix}{text}")
|
||
|
||
if blockname in IGNORED_FIELDS:
|
||
return None
|
||
|
||
# ── Доп. поля Boolean (ТО, Созвон и т.д.) ────────────────────────────────
|
||
addfield = event.get("additionalfieldvalue")
|
||
if addfield is None and blockname.startswith("addfield_"):
|
||
str_val_add = event.get("stringvalue")
|
||
if str_val_add is not None:
|
||
addfield = {"value": str_val_add}
|
||
|
||
if addfield is not None and blockname.startswith("addfield_"):
|
||
bool_label = _ADDFIELD_BOOL_LABELS.get(blockname)
|
||
if bool_label:
|
||
raw_new = addfield.get("value") if isinstance(addfield, dict) else addfield
|
||
raw_old = addfield.get("oldvalue") if isinstance(addfield, dict) else None
|
||
_bool_str = lambda v: "*да*" if str(v).lower() == "true" else "нет"
|
||
if raw_old is not None:
|
||
return ParsedEvent("change", f"• {bool_label}: {_bool_str(raw_old)} → {_bool_str(raw_new)}")
|
||
return ParsedEvent("change", f"• {bool_label}: {_bool_str(raw_new)}")
|
||
return None # Неизвестное доп. поле — пропускаем
|
||
|
||
label = FIELD_LABELS.get(blockname, reference.event_name(etype))
|
||
|
||
# ── Изменения с entityvalue / entitylanguagevalue ─────────────────────────
|
||
entity_data = event.get("entityvalue") or event.get("entitylanguagevalue")
|
||
if entity_data:
|
||
val_new = entity_data.get("valuenew") or {}
|
||
val_old = entity_data.get("valueold")
|
||
new_str = escape_markdown(_extract_name(val_new, blockname))
|
||
old_str = escape_markdown(_extract_name(val_old, blockname)) if val_old else None
|
||
if not new_str and not old_str:
|
||
return None
|
||
if new_str and old_str:
|
||
return ParsedEvent("change", f"• {label}: {old_str} → *{new_str}*")
|
||
if new_str:
|
||
return ParsedEvent("change", f"• {label}: *{new_str}*")
|
||
return ParsedEvent("change", f"• {label}: {old_str} → *снят*")
|
||
|
||
# ── Строковые изменения ───────────────────────────────────────────────────
|
||
str_val = event.get("stringvalue")
|
||
if str_val:
|
||
if blockname in ("file", "files", "taskfiles", "attachment", "attachments"):
|
||
attach = _parse_json_field(str_val)
|
||
if isinstance(attach, dict):
|
||
fname = attach.get("Name", "")
|
||
uploader = _shorten_name(attach.get("UploadedBy", "").strip() or None) or "?"
|
||
fname = escape_markdown(fname) or ""
|
||
uploader = escape_markdown(uploader) or "?"
|
||
return ParsedEvent("change", f"• 📎 {uploader} прикрепил: {fname}")
|
||
m = _FILE_EXT_RE.search(str_val.strip())
|
||
ext = m.group(1).upper() if m else "файл"
|
||
return ParsedEvent("change", f"• 📎 Добавлен файл .{ext}")
|
||
truncated = str_val[:80] + ("…" if len(str_val) > 80 else "")
|
||
truncated = escape_markdown(truncated) or ""
|
||
return ParsedEvent("change", f"• {label}: *{truncated}*")
|
||
|
||
# ── Событие без значения ──────────────────────────────────────────────────
|
||
if etype not in (185, 250):
|
||
return ParsedEvent("change", f"• {label}")
|
||
|
||
return None
|
||
|
||
|
||
# ── Публичные функции ─────────────────────────────────────────────────────────
|
||
|
||
def get_description_event(fields: dict) -> str | None:
|
||
description = _scalar(fields.get("description"))
|
||
if not description:
|
||
return None
|
||
description = re.sub(r'<[^>]+>', ' ', description).strip()
|
||
description = re.sub(r'\s+', ' ', description).strip()
|
||
if not description:
|
||
return None
|
||
truncated = description[:100] + ('…' if len(description) > 100 else '')
|
||
truncated = escape_markdown(truncated) or ""
|
||
return f"📝 {truncated}"
|
||
|
||
|
||
def parse_lifetime(fields: dict, max_events: int = 5) -> tuple[list[str], list[str]]:
|
||
"""
|
||
Разбирает lifetime-поле вебхука (вложенная структура).
|
||
Возвращает (changes, comments) — последние max_events событий.
|
||
"""
|
||
all_events: list[ParsedEvent] = []
|
||
|
||
lifetime = _parse_json_field(fields.get("lifetime", "{}"))
|
||
if not isinstance(lifetime, dict):
|
||
return [], []
|
||
|
||
for entry in lifetime.get("Data", []):
|
||
if entry.get("lifetimetype") == 10:
|
||
continue
|
||
username = entry.get("username", "Неизвестный")
|
||
for event in entry.get("events", {}).get("Data", []):
|
||
parsed = _parse_single_event(event, username)
|
||
if parsed:
|
||
all_events.append(parsed)
|
||
|
||
recent = all_events[-max_events:] if max_events and len(all_events) > max_events else all_events
|
||
changes = [e.text for e in recent if e.kind == "change"]
|
||
comments = [e.text for e in recent if e.kind == "comment"]
|
||
return changes, comments
|
||
|
||
|
||
def format_lifetime_api_events(events_list: list[dict], max_events: int = 5) -> str:
|
||
"""
|
||
Форматирует события из прямого API lifetime (плоский список).
|
||
events_list — результат intradesk_api.get_task_lifetime().
|
||
"""
|
||
lines: list[str] = []
|
||
for event in events_list[:max_events]:
|
||
username = event.get("username", "")
|
||
parsed = _parse_single_event(event, username)
|
||
if parsed:
|
||
# Для API-формата обрезаем комментарии до 100 символов
|
||
text = parsed.text
|
||
if parsed.kind == "comment" and len(text) > 110:
|
||
text = text[:110] + "…"
|
||
lines.append(text)
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _resolve_field(fields: dict, field: str, entity: str) -> str | None:
|
||
raw = _parse_json_field(fields.get(field, {}))
|
||
if not isinstance(raw, dict):
|
||
return None
|
||
item_id = raw.get("Id") or raw.get("Value")
|
||
if not item_id:
|
||
return None
|
||
return reference.resolve(entity, item_id) or str(item_id)
|
||
|
||
|
||
def get_object_name(fields: dict) -> str | None:
|
||
raw = _parse_json_field(fields.get("taskusergroups", {}))
|
||
if not isinstance(raw, dict):
|
||
return None
|
||
for group in raw.get("TaskUserGroups", []):
|
||
if group.get("Type") == 50:
|
||
return group.get("Name", "").strip() or None
|
||
return None
|
||
|
||
|
||
def get_executor_group(fields: dict) -> str | None:
|
||
raw = _parse_json_field(fields.get("taskusergroups", {}))
|
||
if not isinstance(raw, dict):
|
||
return None
|
||
for group in raw.get("TaskUserGroups", []):
|
||
if group.get("Type") == 10:
|
||
return group.get("Name", "").strip() or None
|
||
return None
|
||
|
||
|
||
def get_executor_group_id(fields: dict) -> int:
|
||
raw = _parse_json_field(fields.get("taskusergroups", {}))
|
||
if not isinstance(raw, dict):
|
||
return 0
|
||
for group in raw.get("TaskUserGroups", []):
|
||
if group.get("Type") == 10:
|
||
return int(group.get("Id", 0))
|
||
return 0
|
||
|
||
|
||
def get_initiator(fields: dict) -> str | None:
|
||
raw = _parse_json_field(fields.get("taskusers", {}))
|
||
if not isinstance(raw, dict):
|
||
return None
|
||
for user in raw.get("TaskUsers", []):
|
||
if user.get("Type") == 10:
|
||
return _shorten_name((user.get("Name") or "").strip() or None)
|
||
return None
|
||
|
||
|
||
def get_executor(fields: dict) -> str | None:
|
||
raw = _parse_json_field(fields.get("taskusers", {}))
|
||
if not isinstance(raw, dict):
|
||
return None
|
||
for user in raw.get("TaskUsers", []):
|
||
if user.get("Type") == 30:
|
||
return _shorten_name((user.get("Name") or "").strip() or None)
|
||
return None
|
||
|
||
|
||
def is_final(fields: dict) -> bool:
|
||
raw = _parse_json_field(fields.get("status", {}))
|
||
if isinstance(raw, dict):
|
||
return bool(raw.get("IsFinal", False))
|
||
return False
|
||
|
||
|
||
def is_to(fields: dict) -> bool:
|
||
raw = _parse_json_field(fields.get("addfield_677b45atoBoolean"))
|
||
if isinstance(raw, dict):
|
||
val = raw.get("Value")
|
||
return str(val).lower() == "true" if val is not None else False
|
||
return False
|
||
|
||
|
||
def get_all_executors(fields: dict) -> list[str]:
|
||
result: list[str] = []
|
||
seen: set[str] = set()
|
||
raw_users = _parse_json_field(fields.get("taskusers", {}))
|
||
if isinstance(raw_users, dict):
|
||
for user in raw_users.get("TaskUsers", []):
|
||
if user.get("Type") == 30:
|
||
name = _shorten_name((user.get("Name") or "").strip() or None)
|
||
if name and name not in seen:
|
||
seen.add(name)
|
||
result.append(name)
|
||
group = get_executor_group(fields)
|
||
if group and group not in seen:
|
||
result.append(group)
|
||
return result
|
||
|
||
|
||
def format_notification(payload: dict) -> str:
|
||
"""Полная карточка задачи для основного поста в канале."""
|
||
op_type = payload.get("Event", {}).get("OperationType", "")
|
||
fields = payload.get("Fields", {})
|
||
number = fields.get("Number", "?")
|
||
name = escape_markdown(_scalar(fields.get("name")) or "—") or "-"
|
||
applicant = escape_markdown(get_initiator(fields))
|
||
executors = [escape_markdown(x) or "" for x in get_all_executors(fields)]
|
||
obj_name = escape_markdown(get_object_name(fields))
|
||
to_flag = is_to(fields)
|
||
status = escape_markdown(_resolve_field(fields, "status", "statuses"))
|
||
|
||
header = f"🆕 *Новая #{number}*" if op_type == "Created" \
|
||
else f"🔄 *#{number} обновлена*"
|
||
lines = [header, name, ""]
|
||
if to_flag: lines.append("🔧 *ТО*")
|
||
if status: lines.append(f"📌 Статус: {status}")
|
||
if obj_name: lines.append(f"🏢 Объект: {obj_name}")
|
||
if applicant: lines.append(f"👤 Автор: {applicant}")
|
||
if executors: lines.append(f"👔 Исполнитель: {', '.join(executors)}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def format_update_summary(changes: list[str], comments: list[str]) -> str | None:
|
||
if not changes and not comments:
|
||
return None
|
||
lines = ["🔄 *Изменено:*"]
|
||
lines.extend(changes)
|
||
if comments:
|
||
lines.append("💬 *Комментарии:*")
|
||
lines.extend(comments)
|
||
return "\n".join(lines)
|