austnv/remnawave-minishop
1
0
Fork 0
Code Issues 1 Pull Requests Activity

fix: route support tickets to configured topic

Browse Source
This commit is contained in:
3252a8
2026-06-06 23:27:22 +03:00
parent 3d5190639f
commit d263651b48
2 changed files with 176 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/bot/services/notification_service.py
+45 -13
View File
@@ -237,11 +237,12 @@ class NotificationService:
text: str,
path: str,
fallback_url: str,
web_app_button: bool = True,
) -> InlineKeyboardButton:
webapp_url = self._support_webapp_url(path)
if webapp_url:
if webapp_url and web_app_button:
return InlineKeyboardButton(text=text, web_app=WebAppInfo(url=webapp_url))
return InlineKeyboardButton(text=text, url=fallback_url)
return InlineKeyboardButton(text=text, url=webapp_url or fallback_url)
def _support_text(self, language: Optional[str], key: str, fallback: str) -> str:
if not self.i18n:
@@ -316,7 +317,14 @@ class NotificationService:
return enabled
return self._coerce_bool_setting(raw_value, enabled)
def _support_keyboard(self, ticket, user, *, admin: bool = True) -> InlineKeyboardMarkup:
def _support_keyboard(
self,
ticket,
user,
*,
admin: bool = True,
web_app_buttons: bool = True,
) -> InlineKeyboardMarkup:
ticket_path = (
f"/admin/support/{ticket.ticket_id}" if admin else f"/support/{ticket.ticket_id}"
)
@@ -326,6 +334,7 @@ class NotificationService:
text="Открыть тикет",
path=ticket_path,
fallback_url=self._support_ticket_url(ticket.ticket_id, admin=admin),
web_app_button=web_app_buttons,
)
]
]
@@ -342,12 +351,35 @@ class NotificationService:
text="Карточка пользователя",
path=user_card_path,
fallback_url=self._support_ticket_url(ticket.ticket_id, admin=True),
web_app_button=web_app_buttons,
)
)
if profile_row:
rows.append(profile_row)
return InlineKeyboardMarkup(inline_keyboard=rows)
def _support_log_thread_id(self) -> Optional[int]:
return getattr(self.settings, "LOG_SUPPORT_THREAD_ID", None)
def _support_thread_is_configured(self) -> bool:
return bool(getattr(self.settings, "LOG_CHAT_ID", None) and self._support_log_thread_id())
async def _send_admin_support_telegram(
self,
message: str,
*,
admin_markup: InlineKeyboardMarkup,
log_markup: InlineKeyboardMarkup,
) -> None:
thread_id = self._support_log_thread_id()
if not self._support_thread_is_configured():
await self._send_to_admins(message, reply_markup=admin_markup)
await self._send_to_log_channel(
message,
thread_id=thread_id,
reply_markup=log_markup,
)
def _support_user_keyboard(self, ticket, user) -> InlineKeyboardMarkup:
button_text = self._support_text(
getattr(user, "language_code", None),
@@ -415,12 +447,12 @@ class NotificationService:
f"статус: {hd.quote(str(snapshot.get('panel_status') or '—'))}\n\n"
f"<b>Текст обращения</b>\n{hd.quote(preview)}"
)
keyboard = self._support_keyboard(ticket, user, admin=True)
await self._send_to_admins(message, reply_markup=keyboard)
await self._send_to_log_channel(
admin_keyboard = self._support_keyboard(ticket, user, admin=True)
log_keyboard = self._support_keyboard(ticket, user, admin=True, web_app_buttons=False)
await self._send_admin_support_telegram(
message,
thread_id=getattr(self.settings, "LOG_SUPPORT_THREAD_ID", None),
reply_markup=keyboard,
admin_markup=admin_keyboard,
log_markup=log_keyboard,
)
await self._send_admin_support_email(
render_support_new_ticket_admin,
@@ -456,13 +488,13 @@ class NotificationService:
f"💬 <b>Ответ пользователя в тикете #{ticket.ticket_id}</b>\n"
f"{hd.quote(user_display)}{unread_line}\n\n{hd.quote(preview)}"
)
keyboard = self._support_keyboard(ticket, user, admin=True)
if send_telegram and getattr(self.settings, "LOG_SUPPORT", True):
await self._send_to_admins(text, reply_markup=keyboard)
await self._send_to_log_channel(
admin_keyboard = self._support_keyboard(ticket, user, admin=True)
log_keyboard = self._support_keyboard(ticket, user, admin=True, web_app_buttons=False)
await self._send_admin_support_telegram(
text,
thread_id=getattr(self.settings, "LOG_SUPPORT_THREAD_ID", None),
reply_markup=keyboard,
admin_markup=admin_keyboard,
log_markup=log_keyboard,
)
if send_email:
await self._send_admin_support_email(
tests/test_support_notifications.py
+131
View File
@@ -27,6 +27,10 @@ def _i18n():
return JsonI18n(str(REPO_ROOT / "locales"), default="ru")
def _keyboard_buttons(markup):
return [button for row in markup.inline_keyboard for button in row]
def test_support_ticket_closed_email_uses_user_language():
content = render_support_ticket_closed_user(
_settings(DEFAULT_LANGUAGE="en"),
@@ -106,6 +110,25 @@ def test_admin_support_keyboard_uses_consistent_admin_links():
assert user_card_button.web_app.url == "https://app.example.com/app/admin/users/100200300"
def test_admin_support_keyboard_can_use_group_safe_urls():
service = NotificationService(
bot=SimpleNamespace(),
settings=_settings(SUBSCRIPTION_MINI_APP_URL="https://app.example.com/app"),
)
ticket = SimpleNamespace(ticket_id=42)
user = SimpleNamespace(user_id=100200300)
keyboard = service._support_keyboard(ticket, user, admin=True, web_app_buttons=False)
ticket_button = keyboard.inline_keyboard[0][0]
user_card_button = keyboard.inline_keyboard[1][1]
assert ticket_button.web_app is None
assert ticket_button.url == "https://app.example.com/app/admin/support/42"
assert keyboard.inline_keyboard[1][0].url == "tg://user?id=100200300"
assert user_card_button.web_app is None
assert user_card_button.url == "https://app.example.com/app/admin/users/100200300"
def test_admin_support_keyboard_falls_back_to_startapp_url():
service = NotificationService(
bot=SimpleNamespace(),
@@ -364,6 +387,114 @@ def test_disabled_admin_support_email_keeps_telegram_and_log_notifications():
assert emails == []
def test_support_topic_suppresses_admin_dm_and_uses_url_buttons():
channels = []
service = NotificationService(
bot=SimpleNamespace(),
settings=_settings(
LOG_CHAT_ID=-1003918000002,
LOG_SUPPORT_THREAD_ID=77,
SUBSCRIPTION_MINI_APP_URL="https://app.example.com",
),
)
async def send_to_admins(message, reply_markup=None):
channels.append(("admins", None, bool(message), reply_markup))
async def send_to_log_channel(message, thread_id=None, reply_markup=None):
channels.append(("log", thread_id, bool(message), reply_markup))
service._send_to_admins = send_to_admins
service._send_to_log_channel = send_to_log_channel
ticket = SimpleNamespace(
ticket_id=7,
priority="normal",
category="technical",
subject="Connection issue",
)
user = SimpleNamespace(
user_id=100200300,
username="user",
first_name="User",
last_name=None,
email="user@example.com",
)
asyncio.run(
service.notify_new_support_ticket(
ticket,
user,
"Cannot connect",
{"tariff": "Standard", "end_date": "2026-06-01"},
)
)
assert [item[0] for item in channels] == ["log"]
assert channels[0][1] == 77
markup = channels[0][3]
buttons = _keyboard_buttons(markup)
assert all(button.web_app is None for button in buttons)
assert buttons[0].url == "https://app.example.com/admin/support/7"
assert buttons[2].url == "https://app.example.com/admin/users/100200300"
def test_support_user_reply_topic_suppresses_admin_dm_and_uses_url_buttons():
channels = []
service = NotificationService(
bot=SimpleNamespace(),
settings=_settings(
LOG_CHAT_ID=-1003918000002,
LOG_SUPPORT_THREAD_ID=77,
SUBSCRIPTION_MINI_APP_URL="https://app.example.com",
),
)
async def send_to_admins(message, reply_markup=None):
channels.append(("admins", None, bool(message), reply_markup))
async def send_to_log_channel(message, thread_id=None, reply_markup=None):
channels.append(("log", thread_id, bool(message), reply_markup))
service._send_to_admins = send_to_admins
service._send_to_log_channel = send_to_log_channel
ticket = SimpleNamespace(
ticket_id=7,
priority="normal",
category="technical",
subject="Connection issue",
)
message = SimpleNamespace(body="Still cannot connect")
user = SimpleNamespace(
user_id=100200300,
username="user",
first_name="User",
last_name=None,
email="user@example.com",
)
asyncio.run(
service.notify_support_user_reply(
ticket,
message,
user,
{},
unread_count=3,
send_telegram=True,
send_email=False,
)
)
assert [item[0] for item in channels] == ["log"]
assert channels[0][1] == 77
buttons = _keyboard_buttons(channels[0][3])
assert all(button.web_app is None for button in buttons)
assert buttons[0].url == "https://app.example.com/admin/support/7"
def test_support_user_reply_can_send_email_without_telegram_channels():
emails = []
channels = []