From e2efcf70eef72ddf058a47ee9d682f2f9954c497 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 18 Jun 2026 09:46:19 +0000 Subject: [PATCH] Improve Telegram notification formatting for Max messages Add configurable message styles (plain, enhanced, compact) with sender-change separators, timestamps, and blockquote body text to make notifications from different users easier to scan in Telegram. Co-authored-by: eGoR --- .env.example | 7 ++ README.md | 6 ++ app/config.py | 21 ++++ app/main.py | 9 ++ app/max_listener.py | 200 ++++++++++++++++++++++++++++++++++--- tests/test_config.py | 22 ++++ tests/test_max_listener.py | 121 +++++++++++++++++++++- 7 files changed, 372 insertions(+), 14 deletions(-) diff --git a/.env.example b/.env.example index 63e4ee2..1ae0e89 100644 --- a/.env.example +++ b/.env.example @@ -33,5 +33,12 @@ PLUGINS_ENABLED=true # Skip muted chats (same as «без звука» / не беспокоить in Max) # SKIP_MUTED=true +# Telegram message formatting (plain | enhanced | compact) +# TG_FORMAT_STYLE=enhanced +# Separator line when sender changes (true/false, default true) +# TG_FORMAT_SEPARATOR=true +# Show message time in header HH:MM (true/false, default true) +# TG_FORMAT_TIMESTAMP=true + # Log directory (default: logs) # LOG_DIR=logs diff --git a/README.md b/README.md index 7353162..f4bf08b 100644 --- a/README.md +++ b/README.md @@ -79,6 +79,9 @@ cp .env.example .env | `UNREAD_ONLY` | нет | `true` — пересылать только непрочитанные (если прочитали в Max — в TG не придёт) | | `UNREAD_DELAY_SEC` | нет | Задержка в секундах перед проверкой прочитанности (по умолчанию `2`) | | `SKIP_MUTED` | нет | `true` — не пересылать из заглушённых чатов Max («без звука») | +| `TG_FORMAT_STYLE` | нет | Стиль оформления: `plain` (как раньше), `enhanced` (по умолчанию), `compact` (enhanced + без повторного заголовка у того же автора) | +| `TG_FORMAT_SEPARATOR` | нет | `true` — линия-разделитель при смене автора (по умолчанию `true`) | +| `TG_FORMAT_TIMESTAMP` | нет | `true` — время сообщения в заголовке `HH:MM` (по умолчанию `true`) | | `LOG_DIR` | нет | Путь к директории логов (по умолчанию `logs`) | | `TG_PROXY` | нет | SOCKS5-прокси для Telegram (`socks5://host:port`) | | `TG_READ_TIMEOUT` | нет | Таймаут чтения HTTP-ответа от Telegram, в секундах | @@ -396,6 +399,9 @@ cp .env.example .env | `UNREAD_ONLY` | no | `true` — forward only unread messages (skip if read in Max) | | `UNREAD_DELAY_SEC` | no | Delay before read check in seconds (default `2`) | | `SKIP_MUTED` | no | `true` — skip muted / do-not-disturb chats in Max | +| `TG_FORMAT_STYLE` | no | Message style: `plain`, `enhanced` (default), `compact` (enhanced + hide repeated header for same sender) | +| `TG_FORMAT_SEPARATOR` | no | `true` — separator line when sender changes (default `true`) | +| `TG_FORMAT_TIMESTAMP` | no | `true` — show message time in header as `HH:MM` (default `true`) | | `LOG_DIR` | no | Log directory path (default: `logs`) | | `TG_PROXY` | no | SOCKS5 proxy for Telegram (`socks5://host:port`) | | `TG_READ_TIMEOUT` | no | HTTP read timeout for Telegram responses, in seconds | diff --git a/app/config.py b/app/config.py index a2bcc31..5453c45 100644 --- a/app/config.py +++ b/app/config.py @@ -1,8 +1,13 @@ +import logging import os from dataclasses import dataclass from dotenv import load_dotenv +log = logging.getLogger(__name__) + +VALID_TG_FORMAT_STYLES = frozenset({"plain", "enhanced", "compact"}) + @dataclass(frozen=True) class Settings: @@ -21,6 +26,9 @@ class Settings: unread_only: bool = False unread_delay_sec: float = 2.0 skip_muted: bool = False + tg_format_style: str = "enhanced" + tg_format_separator: bool = True + tg_format_timestamp: bool = True def load_settings() -> Settings: @@ -42,6 +50,14 @@ def load_settings() -> Settings: f"TG_CHAT_ID must be a valid integer, got: {tg_chat_id!r}" ) + raw_format_style = os.environ.get("TG_FORMAT_STYLE", "enhanced").strip().lower() + if raw_format_style not in VALID_TG_FORMAT_STYLES: + log.warning( + "Unknown TG_FORMAT_STYLE=%r, falling back to 'enhanced'", + raw_format_style, + ) + raw_format_style = "enhanced" + return Settings( max_token=os.environ["MAX_TOKEN"].strip(), max_device_id=os.environ["MAX_DEVICE_ID"].strip(), @@ -58,4 +74,9 @@ def load_settings() -> Settings: unread_only=os.environ.get("UNREAD_ONLY", "").lower() in ("1", "true", "yes"), unread_delay_sec=float(os.environ.get("UNREAD_DELAY_SEC", "2") or "2"), skip_muted=os.environ.get("SKIP_MUTED", "").lower() in ("1", "true", "yes"), + tg_format_style=raw_format_style, + tg_format_separator=os.environ.get("TG_FORMAT_SEPARATOR", "true").lower() + not in ("0", "false", "no"), + tg_format_timestamp=os.environ.get("TG_FORMAT_TIMESTAMP", "true").lower() + not in ("0", "false", "no"), ) diff --git a/app/main.py b/app/main.py index 82cb08d..4897db9 100644 --- a/app/main.py +++ b/app/main.py @@ -93,6 +93,9 @@ async def main(): debug=settings.debug, reply_enabled=settings.reply_enabled, unread_only=settings.unread_only, unread_delay_sec=settings.unread_delay_sec, skip_muted=settings.skip_muted, + tg_format_style=settings.tg_format_style, + tg_format_separator=settings.tg_format_separator, + tg_format_timestamp=settings.tg_format_timestamp, ) if settings.unread_only: @@ -102,6 +105,12 @@ async def main(): ) if settings.skip_muted: log.info("Skip-muted mode: ON (no forwards from muted Max chats)") + log.info( + "Message format: style=%s separator=%s timestamp=%s", + settings.tg_format_style, + settings.tg_format_separator, + settings.tg_format_timestamp, + ) tg_app = None if settings.reply_enabled: diff --git a/app/max_listener.py b/app/max_listener.py index 198520a..ea71245 100644 --- a/app/max_listener.py +++ b/app/max_listener.py @@ -13,9 +13,36 @@ PHOTO_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"} VIDEO_EXTENSIONS = {".mp4", ".mov", ".avi", ".mkv", ".webm"} +SEPARATOR_LINE = "━━━━━━━━━━━━━━━━" + + +def _format_time(timestamp: Any) -> str: + if timestamp is None: + return "" + try: + ts = int(timestamp) + except (TypeError, ValueError): + return "" + if ts > 1_000_000_000_000: + ts //= 1000 + try: + return datetime.fromtimestamp(ts).strftime("%H:%M") + except (OSError, OverflowError, ValueError): + return "" + + +def _needs_separator( + last_key: tuple | None, + chat_id: Any, + sender_id: Any, + separator_enabled: bool, +) -> bool: + if not separator_enabled or last_key is None: + return False + return last_key != (chat_id, sender_id) -def _header(msg: MaxMessage, sender_label: str, chat_label: str, is_dm: bool) -> str: +def _header_plain(sender_label: str, chat_label: str, is_dm: bool) -> str: if is_dm: return f"✉ {sender_label}" if chat_label: @@ -23,6 +50,125 @@ def _header(msg: MaxMessage, sender_label: str, chat_label: str, is_dm: bool) -> return f"💬 {sender_label}" +def _build_header( + sender_label: str, + chat_label: str, + is_dm: bool, + style: str, + timestamp: Any, + show_timestamp: bool, +) -> str: + if style == "plain": + return _header_plain(sender_label, chat_label, is_dm) + + time_suffix = "" + if show_timestamp: + time_str = _format_time(timestamp) + if time_str: + time_suffix = f" · {time_str}" + + if is_dm: + return f"✉ {sender_label}{time_suffix}" + if chat_label: + return f"💬 {chat_label}\n👤 {sender_label}{time_suffix}" + return f"💬 {sender_label}{time_suffix}" + + +def _format_body_text(text: str, style: str, *, use_blockquote: bool = True) -> str: + if not text: + return "" + escaped = escape(text) + if style == "plain" or not use_blockquote: + return escaped + return f"
{escaped}
" + + +def _join_header_body(header: str, body: str, *, gap: str = "\n\n") -> str: + if header and body: + return f"{header}{gap}{body}" + return header or body + + +class MessageFormatter: + def __init__( + self, + style: str = "enhanced", + separator_enabled: bool = True, + show_timestamp: bool = True, + ) -> None: + self.style = style + self.separator_enabled = separator_enabled + self.show_timestamp = show_timestamp + self.last_sender_key: tuple | None = None + + def begin_message( + self, + chat_id: Any, + sender_id: Any, + sender_label: str, + chat_label: str, + is_dm: bool, + timestamp: Any, + ) -> str: + sender_key = (chat_id, sender_id) + same_sender = self.last_sender_key == sender_key + need_sep = _needs_separator( + self.last_sender_key, + chat_id, + sender_id, + self.separator_enabled and self.style != "plain", + ) + self.last_sender_key = sender_key + + parts: list[str] = [] + if need_sep: + parts.append(SEPARATOR_LINE) + + show_header = self.style == "plain" or self.style == "enhanced" or not same_sender + if show_header: + header_style = "enhanced" if self.style == "compact" else self.style + parts.append( + _build_header( + sender_label, + chat_label, + is_dm, + header_style, + timestamp, + self.show_timestamp, + ) + ) + + if self.style == "plain": + return "\n".join(parts) + return "\n\n".join(parts) + + def format_content(self, text: str, *, use_blockquote: bool = True) -> str: + return _format_body_text(text, self.style, use_blockquote=use_blockquote) + + def join_header_body(self, header: str, body: str) -> str: + gap = "\n" if self.style == "plain" else "\n\n" + return _join_header_body(header, body, gap=gap) + + def format_text_message( + self, + chat_id: Any, + sender_id: Any, + sender_label: str, + chat_label: str, + is_dm: bool, + text: str, + timestamp: Any, + ) -> str: + header = self.begin_message( + chat_id, sender_id, sender_label, chat_label, is_dm, timestamp + ) + return self.join_header_body(header, self.format_content(text)) + + +def _header(msg: MaxMessage, sender_label: str, chat_label: str, is_dm: bool) -> str: + return _header_plain(sender_label, chat_label, is_dm) + + def _extract_photo_url(attach: dict) -> str | None: """Extract the best available URL for a PHOTO attachment.""" return attach.get("baseUrl") or attach.get("url") @@ -181,6 +327,7 @@ async def _handle_forward_message( sender: TelegramSender, resolver: ContactResolver, kb=None, + formatter: MessageFormatter | None = None, ) -> None: """Handle FORWARD link inside a message.""" fwd_meaningful, fwd_sender_label, fwd_text = await _parse_link(link, resolver) @@ -189,21 +336,25 @@ async def _handle_forward_message( if fwd_sender_label: prefix = f"↩️ Переслано от {fwd_sender_label}" - full_header = f"{header_text}\n{prefix}" + full_header = f"{header_text}\n{prefix}" if header_text else prefix + join = formatter.join_header_body if formatter else lambda h, b: _join_header_body(h, b, gap="\n") if fwd_meaningful: text_sent = False for i, attach in enumerate(fwd_meaningful): if i == 0 and fwd_text: - cap = f"{full_header}\n{escape(fwd_text)}" + body = formatter.format_content(fwd_text) if formatter else escape(fwd_text) + cap = join(full_header, body) text_sent = True else: cap = full_header await _send_attach(attach, client, sender, cap, None, None, kb=kb) if fwd_text and not text_sent: - await sender.send(f"{full_header}\n{escape(fwd_text)}", reply_markup=kb) + body = formatter.format_content(fwd_text) if formatter else escape(fwd_text) + await sender.send(join(full_header, body), reply_markup=kb) elif fwd_text: - await sender.send(f"{full_header}\n{escape(fwd_text)}", reply_markup=kb) + body = formatter.format_content(fwd_text) if formatter else escape(fwd_text) + await sender.send(join(full_header, body), reply_markup=kb) else: await sender.send(f"{full_header}\n[без содержимого]", reply_markup=kb) @@ -219,7 +370,7 @@ async def _handle_reply_message( if fwd_sender_label: prefix = f"↩ Ответ на {fwd_sender_label}" - full_header = f"{header_text}\n{prefix}" + full_header = f"{header_text}\n{prefix}" if header_text else prefix attaches_str = "" if fwd_meaningful: for fwd_attach in fwd_meaningful: @@ -256,12 +407,20 @@ def create_max_client( max_token: str, max_device_id: str, sender: TelegramSender, max_chat_ids: str | None = None, debug: bool = False, reply_enabled: bool = False, unread_only: bool = False, unread_delay_sec: float = 2.0, skip_muted: bool = False, + tg_format_style: str = "enhanced", + tg_format_separator: bool = True, + tg_format_timestamp: bool = True, ) -> MaxClient: client = MaxClient( token=max_token, device_id=max_device_id, debug=debug, chat_ids=max_chat_ids, unread_only=unread_only, skip_muted=skip_muted, ) resolver = ContactResolver(client=client) + formatter = MessageFormatter( + style=tg_format_style, + separator_enabled=tg_format_separator, + show_timestamp=tg_format_timestamp, + ) _first_connect = True _notif_count = 0 @@ -371,16 +530,21 @@ async def _message_sent() -> None: chat_label = "" else: chat_label = escape(resolver.chat_name(msg.chat_id)) - header_text = _header(msg, sender_label, chat_label, is_dm) + header_text = formatter.begin_message( + msg.chat_id, msg.sender_id, sender_label, chat_label, is_dm, msg.timestamp + ) kb = reply_keyboard(msg.chat_id) if reply_enabled else None link = msg.link link_type = link.get("type") if isinstance(link, dict) else None if link_type == "FORWARD": - await _handle_forward_message(link, header_text, client, sender, resolver, kb=kb) + await _handle_forward_message(link, header_text, client, sender, resolver, kb=kb, formatter=formatter) if msg.text: - await sender.send(f"{header_text}\n{escape(msg.text)}", reply_markup=kb) + await sender.send( + formatter.join_header_body(header_text, formatter.format_content(msg.text)), + reply_markup=kb, + ) log.info("Forwarded message → TG") await _message_sent() return @@ -388,7 +552,11 @@ async def _message_sent() -> None: if link_type == "REPLY": attaches_str, full_header, fwd_text = await _handle_reply_message(link, header_text, resolver) if msg.text: - await sender.send(f"{full_header}\n
{escape(fwd_text)}{attaches_str}
{escape(msg.text)}", reply_markup=kb) + reply_body = formatter.format_content(msg.text, use_blockquote=False) + await sender.send( + f"{full_header}\n
{escape(fwd_text)}{attaches_str}
{reply_body}", + reply_markup=kb, + ) log.info("Forwarded reply → TG") await _message_sent() return @@ -402,7 +570,7 @@ async def _message_sent() -> None: text_sent = False for i, attach in enumerate(meaningful_attaches): if i == 0 and msg.text: - cap = f"{header_text}\n{escape(msg.text)}" + cap = formatter.join_header_body(header_text, formatter.format_content(msg.text)) text_sent = True else: cap = header_text @@ -410,11 +578,17 @@ async def _message_sent() -> None: log.info("Forwarded attach _type=%s → TG", attach.get("_type")) if msg.text and not text_sent: - await sender.send(f"{header_text}\n{escape(msg.text)}", reply_markup=kb) + await sender.send( + formatter.join_header_body(header_text, formatter.format_content(msg.text)), + reply_markup=kb, + ) await _message_sent() else: if msg.text: - await sender.send(f"{header_text}\n{escape(msg.text)}", reply_markup=kb) + await sender.send( + formatter.join_header_body(header_text, formatter.format_content(msg.text)), + reply_markup=kb, + ) log.info("Forwarded text → TG") await _message_sent() else: diff --git a/tests/test_config.py b/tests/test_config.py index f3fe45e..8be49fe 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -142,6 +142,28 @@ def test_max_chat_ids_none_when_empty_string(self): s = _load_settings_with_env(_env(MAX_CHAT_IDS="")) assert s.max_chat_ids is None + def test_tg_format_defaults(self): + s = _load_settings_with_env(_env()) + assert s.tg_format_style == "enhanced" + assert s.tg_format_separator is True + assert s.tg_format_timestamp is True + + def test_tg_format_style_plain(self): + s = _load_settings_with_env(_env(TG_FORMAT_STYLE="plain")) + assert s.tg_format_style == "plain" + + def test_tg_format_style_invalid_falls_back(self): + s = _load_settings_with_env(_env(TG_FORMAT_STYLE="unknown")) + assert s.tg_format_style == "enhanced" + + def test_tg_format_separator_false(self): + s = _load_settings_with_env(_env(TG_FORMAT_SEPARATOR="false")) + assert s.tg_format_separator is False + + def test_tg_format_timestamp_false(self): + s = _load_settings_with_env(_env(TG_FORMAT_TIMESTAMP="0")) + assert s.tg_format_timestamp is False + # --------------------------------------------------------------------------- # load_settings — missing required variables diff --git a/tests/test_max_listener.py b/tests/test_max_listener.py index 0f8440b..0cf8edd 100644 --- a/tests/test_max_listener.py +++ b/tests/test_max_listener.py @@ -1,7 +1,19 @@ """Tests for app/max_listener.py — pure helper functions.""" +from datetime import datetime, timezone + import pytest -from app.max_listener import _human_size, _guess_media_kind +from app.max_listener import ( + SEPARATOR_LINE, + MessageFormatter, + _build_header, + _format_body_text, + _format_time, + _guess_media_kind, + _human_size, + _join_header_body, + _needs_separator, +) # --------------------------------------------------------------------------- @@ -144,3 +156,110 @@ def test_full_path_mp4(self): # Extension appearing in the middle of filename should not trigger false match def test_mp4_in_name_not_extension_is_document(self): assert _guess_media_kind("mp4_notes.txt") == "document" + + +# --------------------------------------------------------------------------- +# _format_time +# --------------------------------------------------------------------------- + +class TestFormatTime: + def test_none_returns_empty(self): + assert _format_time(None) == "" + + def test_invalid_returns_empty(self): + assert _format_time("not-a-number") == "" + + def test_seconds_timestamp(self): + ts = int(datetime(2024, 6, 18, 14, 30, tzinfo=timezone.utc).timestamp()) + assert _format_time(ts) == datetime.fromtimestamp(ts).strftime("%H:%M") + + def test_milliseconds_timestamp(self): + ts = int(datetime(2024, 6, 18, 9, 5, tzinfo=timezone.utc).timestamp()) * 1000 + assert _format_time(ts) == datetime.fromtimestamp(ts // 1000).strftime("%H:%M") + + +# --------------------------------------------------------------------------- +# _needs_separator +# --------------------------------------------------------------------------- + +class TestNeedsSeparator: + def test_disabled(self): + assert _needs_separator(("c", "s"), "c", "s2", False) is False + + def test_first_message_no_separator(self): + assert _needs_separator(None, "c", "s", True) is False + + def test_same_sender_no_separator(self): + assert _needs_separator(("c", "s"), "c", "s", True) is False + + def test_different_sender_separator(self): + assert _needs_separator(("c", "s1"), "c", "s2", True) is True + + +# --------------------------------------------------------------------------- +# _build_header +# --------------------------------------------------------------------------- + +class TestBuildHeader: + def test_plain_dm(self): + assert _build_header("Ivan", "", True, "plain", None, True) == "✉ Ivan" + + def test_plain_group_with_chat(self): + result = _build_header("Ivan", "Work", False, "plain", None, True) + assert result == "💬 Work | Ivan" + + def test_enhanced_dm_with_time(self): + ts = int(datetime(2024, 6, 18, 14, 30, tzinfo=timezone.utc).timestamp()) + result = _build_header("Ivan", "", True, "enhanced", ts, True) + assert result.startswith("✉ Ivan ·") + + def test_enhanced_group_two_lines(self): + ts = int(datetime(2024, 6, 18, 14, 30, tzinfo=timezone.utc).timestamp()) + result = _build_header("Petr", "Work", False, "enhanced", ts, True) + assert "💬 Work" in result + assert "👤 Petr" in result + + +# --------------------------------------------------------------------------- +# _format_body_text +# --------------------------------------------------------------------------- + +class TestFormatBodyText: + def test_plain(self): + assert _format_body_text("hi", "plain") == "hi" + + def test_enhanced_blockquote(self): + assert _format_body_text("hi", "enhanced") == "
hi
" + + def test_no_blockquote_override(self): + assert _format_body_text("hi", "enhanced", use_blockquote=False) == "hi" + + +# --------------------------------------------------------------------------- +# MessageFormatter +# --------------------------------------------------------------------------- + +class TestMessageFormatter: + def test_enhanced_adds_separator_on_sender_change(self): + fmt = MessageFormatter(style="enhanced") + first = fmt.begin_message("c1", "u1", "Ivan", "", False, None) + second = fmt.begin_message("c1", "u2", "Petr", "", False, None) + assert SEPARATOR_LINE not in first + assert SEPARATOR_LINE in second + + def test_compact_hides_repeated_header(self): + fmt = MessageFormatter(style="compact") + first = fmt.format_text_message("c1", "u1", "Ivan", "", False, "hi", None) + second = fmt.format_text_message("c1", "u1", "Ivan", "", False, "again", None) + assert "Ivan" in first + assert "Ivan" not in second + assert "
again
" in second + + def test_plain_matches_legacy_format(self): + fmt = MessageFormatter(style="plain", separator_enabled=True, show_timestamp=True) + text = fmt.format_text_message("c1", "u1", "Ivan", "Work", False, "hello", None) + assert text == "💬 Work | Ivan\nhello" + + def test_join_header_body(self): + assert _join_header_body("hdr", "body") == "hdr\n\nbody" + assert _join_header_body("", "body") == "body"