fix: avoid Telegram notification probe messages

This commit is contained in:
3252a8
2026-05-31 00:17:20 +03:00
parent 93353db511
commit cf3af17243
2 changed files with 45 additions and 47 deletions
+2 -46
View File
@@ -4,7 +4,6 @@ from typing import Any, Optional
from aiogram import Bot from aiogram import Bot
from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup, WebAppInfo
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from bot.middlewares.i18n import JsonI18n from bot.middlewares.i18n import JsonI18n
@@ -128,37 +127,6 @@ async def mark_telegram_notifications_enabled_for_telegram_user(
) )
def _translate(
i18n: Optional[JsonI18n],
language: str,
key: str,
fallback: str,
**kwargs: Any,
) -> str:
if not i18n:
return fallback.format(**kwargs) if kwargs else fallback
return i18n.gettext(language, key, **kwargs) or fallback
def _probe_keyboard(
settings: Settings,
i18n: Optional[JsonI18n],
language: str,
) -> Optional[InlineKeyboardMarkup]:
app_url = str(getattr(settings, "SUBSCRIPTION_MINI_APP_URL", "") or "").strip()
if not app_url:
return None
text = _translate(
i18n,
language,
"telegram_notifications_open_app_button",
"Open app",
)
return InlineKeyboardMarkup(
inline_keyboard=[[InlineKeyboardButton(text=text, web_app=WebAppInfo(url=app_url))]]
)
async def probe_telegram_notifications( async def probe_telegram_notifications(
*, *,
session: AsyncSession, session: AsyncSession,
@@ -187,20 +155,8 @@ async def probe_telegram_notifications(
"start_link": telegram_notifications_start_link(bot_username), "start_link": telegram_notifications_start_link(bot_username),
} }
language = str(getattr(user, "language_code", "") or settings.DEFAULT_LANGUAGE)
text = _translate(
i18n,
language,
"telegram_notifications_enabled_message",
"Telegram notifications are enabled.",
)
try: try:
await bot.send_message( await bot.get_chat(int(telegram_id))
int(telegram_id),
text,
reply_markup=_probe_keyboard(settings, i18n, language),
disable_web_page_preview=True,
)
except Exception as exc: except Exception as exc:
status = telegram_notification_status_from_error(exc) status = telegram_notification_status_from_error(exc)
if status: if status:
@@ -211,7 +167,7 @@ async def probe_telegram_notifications(
"start_link": telegram_notifications_start_link(bot_username), "start_link": telegram_notifications_start_link(bot_username),
} }
logger.warning( logger.warning(
"Telegram notification probe failed for user %s / telegram %s: %s", "Telegram notification chat probe failed for user %s / telegram %s: %s",
user.user_id, user.user_id,
telegram_id, telegram_id,
exc, exc,
+43 -1
View File
@@ -1,6 +1,12 @@
import asyncio
from types import SimpleNamespace from types import SimpleNamespace
from unittest.mock import AsyncMock
from bot.services.telegram_notifications import telegram_notifications_need_prompt from bot.services import telegram_notifications as module
from bot.services.telegram_notifications import (
TELEGRAM_NOTIFICATIONS_ENABLED,
telegram_notifications_need_prompt,
)
def _user(status: str): def _user(status: str):
@@ -22,3 +28,39 @@ def test_telegram_notifications_prompt_requires_linked_telegram():
) )
assert telegram_notifications_need_prompt(user) is False assert telegram_notifications_need_prompt(user) is False
def test_probe_telegram_notifications_uses_silent_chat_check(monkeypatch):
user = SimpleNamespace(
user_id=42,
telegram_id=123,
telegram_notifications_status="unknown",
)
bot = SimpleNamespace(
get_chat=AsyncMock(return_value=SimpleNamespace(id=123)),
send_message=AsyncMock(),
)
recorded = []
async def fake_mark_status(session, user_id, status, *, telegram_id=None, checked_at=None):
recorded.append((session, user_id, status, telegram_id, checked_at))
return user
monkeypatch.setattr(module, "mark_telegram_notifications_status", fake_mark_status)
result = asyncio.run(
module.probe_telegram_notifications(
session="session",
bot=bot,
settings=SimpleNamespace(DEFAULT_LANGUAGE="ru"),
i18n=None,
user=user,
bot_username="preview_bot",
)
)
bot.get_chat.assert_awaited_once_with(123)
bot.send_message.assert_not_called()
assert result["ok"] is True
assert result["status"] == TELEGRAM_NOTIFICATIONS_ENABLED
assert recorded == [("session", 42, TELEGRAM_NOTIFICATIONS_ENABLED, 123, None)]