diff --git a/bot/telegram.py b/bot/telegram.py index 3faa94b..0a86923 100644 --- a/bot/telegram.py +++ b/bot/telegram.py @@ -56,7 +56,10 @@ def _build_full_text( result += "\n\nšŸ†• *ŠŠ¾Š²Š¾Šµ:*\n" + "\n".join(newest) if responsible: - names = ", ".join(r["user_name"] for r in responsible) + names = ", ".join( + formatter.escape_markdown(r["user_name"]) or r["user_name"] + for r in responsible + ) result += _RESP_SEP + f"\nšŸ™‹ Š’ работе: {names}" return result diff --git a/intradesk/formatter.py b/intradesk/formatter.py index f4d7581..39c0e91 100644 --- a/intradesk/formatter.py +++ b/intradesk/formatter.py @@ -87,6 +87,15 @@ def _shorten_name(name: str | None) -> str | None: 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) @@ -146,6 +155,7 @@ def _parse_single_event(event: dict, username: str = "") -> ParsedEvent | None: 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): @@ -155,6 +165,7 @@ def _parse_single_event(event: dict, username: str = "") -> ParsedEvent | 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}") @@ -187,8 +198,8 @@ def _parse_single_event(event: dict, username: str = "") -> ParsedEvent | None: if entity_data: val_new = entity_data.get("valuenew") or {} val_old = entity_data.get("valueold") - new_str = _extract_name(val_new, blockname) - old_str = _extract_name(val_old, blockname) if val_old else None + 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: @@ -205,11 +216,14 @@ def _parse_single_event(event: dict, username: str = "") -> ParsedEvent | None: 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}*") # ── Добытие без Š·Š½Š°Ń‡ŠµŠ½ŠøŃ ────────────────────────────────────────────────── @@ -230,6 +244,7 @@ def get_description_event(fields: dict) -> str | None: if not description: return None truncated = description[:100] + ('…' if len(description) > 100 else '') + truncated = escape_markdown(truncated) or "" return f"šŸ“ {truncated}" @@ -374,12 +389,12 @@ def format_notification(payload: dict) -> str: op_type = payload.get("Event", {}).get("OperationType", "") fields = payload.get("Fields", {}) number = fields.get("Number", "?") - name = _scalar(fields.get("name")) or "—" - applicant = get_initiator(fields) - executors = get_all_executors(fields) - obj_name = get_object_name(fields) + 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 = _resolve_field(fields, "status", "statuses") + status = escape_markdown(_resolve_field(fields, "status", "statuses")) header = f"šŸ†• *ŠŠ¾Š²Š°Ń #{number}*" if op_type == "Created" \ else f"šŸ”„ *#{number} обновлена*"