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

feat: prompt users to start Telegram bot for notifications

Browse Source
This commit is contained in:
3252a8
2026-05-30 21:15:17 +03:00
parent 8c0e778388
commit acc222da41
28 changed files with 854 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/bot/app/web/subscription_webapp.py
+2
View File
@@ -16,6 +16,7 @@ from bot.app.web.webapp import (
routes as _routes,
serializers as _serializers,
support as _support,
telegram_notifications as _telegram_notifications,
)
_MODULES = (
@@ -25,6 +26,7 @@ _MODULES = (
_assets,
_auth,
_account,
_telegram_notifications,
_serializers,
_billing,
_devices,
backend/bot/app/web/webapp/account.py
+3
View File
@@ -8,6 +8,7 @@ from .auth import (
_sync_merged_panel_identity_for_user,
)
from .common import _invalidate_webapp_user_caches
from .telegram_notifications import _probe_telegram_notifications_for_user_id
async def account_email_request_route(request: web.Request) -> web.Response:
@@ -416,6 +417,8 @@ async def account_telegram_link_route(request: web.Request) -> web.Response:
except Exception:
logger.exception("Failed to send account Telegram linked notification")
await _probe_telegram_notifications_for_user_id(request, int(final_user_id))
token = create_webapp_session_token(settings, int(final_user_id))
response_payload: Dict[str, Any] = {
"ok": True,
backend/bot/app/web/webapp/auth.py
+5
View File
@@ -1,6 +1,7 @@
# ruff: noqa: F401,F403,F405,I001
from ._runtime import * # noqa: F403,F405
from .common import _invalidate_webapp_user_caches
from .telegram_notifications import _probe_telegram_notifications_for_user_id
def _resolve_telegram_bot_id(bot_token: str) -> Optional[int]:
@@ -436,6 +437,9 @@ async def telegram_oauth_callback_route(request: web.Request) -> web.Response:
first_name=linked_user_for_panel.first_name,
)
if final_user_id:
await _probe_telegram_notifications_for_user_id(request, int(final_user_id))
token = create_webapp_session_token(settings, int(final_user_id))
response = web.HTTPFound(_telegram_oauth_redirect_url(redirect_path, status="success"))
_clear_telegram_oauth_state_cookie(response)
@@ -531,6 +535,7 @@ async def auth_token_route(request: web.Request) -> web.Response:
return _json_error(500, "auth_failed", "Auth failed")
await _invalidate_webapp_user_caches(settings, authenticated_user_id, include_devices=True)
await _probe_telegram_notifications_for_user_id(request, int(authenticated_user_id))
token = create_webapp_session_token(settings, int(authenticated_user_id))
return _build_webapp_auth_response(settings, {"ok": True}, token=token)
backend/bot/app/web/webapp/routes.py
+4
View File
@@ -84,6 +84,10 @@ def setup_subscription_webapp_routes(app: web.Application) -> None:
app.router.add_post("/api/account/password/request", account_password_request_route)
app.router.add_post("/api/account/password/confirm", account_password_confirm_route)
app.router.add_post("/api/account/telegram/link", account_telegram_link_route)
app.router.add_post(
"/api/account/telegram/notifications/probe",
account_telegram_notifications_probe_route,
)
app.router.add_post("/api/promo/apply", apply_promo_route)
app.router.add_post("/api/trial/activate", activate_trial_route)
app.router.add_get("/api/devices", devices_route)
backend/bot/app/web/webapp/serializers.py
+18
View File
@@ -3,6 +3,12 @@ from ._runtime import * # noqa: F403,F405
from config.subscription_guides_config import subscription_guides_available
from config.webapp_themes_config import public_themes_catalog_payload
from bot.services.telegram_notifications import (
TELEGRAM_NOTIFICATIONS_ENABLED,
normalize_telegram_notification_status,
telegram_notifications_need_prompt,
telegram_notifications_start_link,
)
async def _build_user_payload(request: web.Request, user_id: int) -> Dict[str, Any]:
@@ -72,6 +78,12 @@ async def _build_user_payload(request: web.Request, user_id: int) -> Dict[str, A
lang = _normalize_language(db_user.language_code or settings.DEFAULT_LANGUAGE)
admin_ids = {int(x) for x in (settings.ADMIN_IDS or [])}
is_admin = bool(db_user.telegram_id and int(db_user.telegram_id) in admin_ids)
telegram_notifications_status = normalize_telegram_notification_status(
getattr(db_user, "telegram_notifications_status", None)
)
telegram_notifications_link = telegram_notifications_start_link(
request.app.get("bot_username") or ""
)
return {
"user": {
"id": user_id,
@@ -83,6 +95,12 @@ async def _build_user_payload(request: web.Request, user_id: int) -> Dict[str, A
),
"telegram_id": db_user.telegram_id,
"telegram_linked": bool(_telegram_id_for_user(db_user)),
"telegram_notifications_status": telegram_notifications_status,
"telegram_notifications_enabled": (
telegram_notifications_status == TELEGRAM_NOTIFICATIONS_ENABLED
),
"telegram_notifications_need_prompt": telegram_notifications_need_prompt(db_user),
"telegram_notifications_start_link": telegram_notifications_link,
"telegram_photo_url": _telegram_avatar_url(avatar),
"first_name": db_user.first_name,
"language_code": lang,
backend/bot/app/web/webapp/telegram_notifications.py
+70
View File
@@ -0,0 +1,70 @@
# ruff: noqa: F401,F403,F405,I001
from ._runtime import * # noqa: F403,F405
from bot.services.telegram_notifications import (
TELEGRAM_NOTIFICATIONS_ENABLED,
probe_telegram_notifications,
telegram_notifications_start_link,
)
from .common import _invalidate_webapp_user_caches
async def _probe_telegram_notifications_for_user_id(
request: web.Request,
user_id: int,
*,
force: bool = False,
) -> Dict[str, Any]:
settings: Settings = request.app["settings"]
async_session_factory: sessionmaker = request.app["async_session_factory"]
async with async_session_factory() as session:
try:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
await session.rollback()
return {
"ok": False,
"status": "access_denied",
"enabled": False,
"start_link": telegram_notifications_start_link(
request.app.get("bot_username") or ""
),
}
result = await probe_telegram_notifications(
session=session,
bot=request.app["bot"],
settings=settings,
i18n=request.app.get("i18n"),
user=db_user,
bot_username=request.app.get("bot_username") or "",
force=force,
)
await session.commit()
status = str(result.get("status") or "")
await _invalidate_webapp_user_caches(settings, int(db_user.user_id))
return {
"ok": bool(result.get("ok")),
"status": status,
"enabled": status == TELEGRAM_NOTIFICATIONS_ENABLED,
"start_link": result.get("start_link"),
}
except Exception:
await session.rollback()
logger.exception("Telegram notification probe failed")
return {
"ok": False,
"status": "unknown",
"enabled": False,
"start_link": telegram_notifications_start_link(
request.app.get("bot_username") or ""
),
}
async def account_telegram_notifications_probe_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
force = True
result = await _probe_telegram_notifications_for_user_id(request, user_id, force=force)
if result.get("status") == "access_denied":
return _json_error(403, "access_denied", "Access denied")
return web.json_response({"ok": True, "telegram_notifications": result})
backend/bot/handlers/user/start.py
+28 -2
View File
@@ -22,6 +22,7 @@ from bot.services.panel_api_service import PanelApiService
from bot.services.promo_code_service import PromoCodeService
from bot.services.referral_service import ReferralService
from bot.services.subscription_service import SubscriptionService
from bot.services.telegram_notifications import TELEGRAM_NOTIFICATIONS_ENABLED
from bot.utils.callback_answer import safe_answer_callback
from bot.utils.install_links import (
append_install_share_link_text,
@@ -392,11 +393,12 @@ async def ensure_required_channel_subscription(
@router.message(CommandStart(magic=F.args.regexp(r"^promo_(\w+)$").as_("promo_match")))
@router.message(CommandStart(magic=F.args.regexp(r"^admin_user_(\d+)$").as_("admin_user_match")))
@router.message(CommandStart(magic=F.args.regexp(r"^ticket_(\d+)$").as_("ticket_match")))
@router.message(CommandStart(magic=F.args.regexp(r"^notifications$").as_("notifications_match")))
@router.message(CommandStart(magic=F.args.regexp(r"^page_ref$").as_("page_ref_match")))
@router.message(
CommandStart(
magic=F.args.regexp(
r"^(?!ref_|promo_|admin_user_|ticket_|page_ref$|webapp_auth_)([A-Za-z0-9_\-]{2,64})$"
r"^(?!ref_|promo_|admin_user_|ticket_|notifications$|page_ref$|webapp_auth_)([A-Za-z0-9_\-]{2,64})$"
).as_("ad_param_match")
)
)
@@ -414,6 +416,7 @@ async def start_command_handler(
ad_param_match: Optional[re.Match] = None,
admin_user_match: Optional[re.Match] = None,
ticket_match: Optional[re.Match] = None,
notifications_match: Optional[re.Match] = None,
):
await state.clear()
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
@@ -500,6 +503,7 @@ async def start_command_handler(
promo_code_to_apply: Optional[str] = None
should_open_referral_from_start = False
ad_start_param: Optional[str] = None
notifications_start_requested = bool(notifications_match)
if ref_match:
raw_ref_value = ref_match.group(1)
@@ -522,6 +526,8 @@ async def start_command_handler(
elif promo_match:
promo_code_to_apply = promo_match.group(1)
logging.info(f"User {user_id} started with promo code: {promo_code_to_apply}")
elif notifications_start_requested:
logging.info("User %s started bot from notifications deep-link.", user_id)
elif page_ref_match:
should_open_referral_from_start = True
logging.info(f"User {user_id} started with page_ref deep-link.")
@@ -532,18 +538,24 @@ async def start_command_handler(
sanitized_username = sanitize_username(user.username)
sanitized_first_name = sanitize_display_name(user.first_name)
sanitized_last_name = sanitize_display_name(user.last_name)
notification_status_now = datetime.now(timezone.utc)
db_user = await user_dal.get_user_by_id(session, user_id)
is_existing_user = db_user is not None
if not db_user:
user_data_to_create = {
"user_id": user_id,
"telegram_id": user_id,
"username": sanitized_username,
"first_name": sanitized_first_name,
"last_name": sanitized_last_name,
"language_code": current_lang,
"referred_by_id": referred_by_user_id,
"registration_date": datetime.now(timezone.utc),
"telegram_notifications_status": TELEGRAM_NOTIFICATIONS_ENABLED,
"telegram_notifications_checked_at": notification_status_now,
"telegram_notifications_enabled_at": notification_status_now,
"telegram_notifications_blocked_at": None,
}
try:
db_user, created = await user_dal.create_user(session, user_data_to_create)
@@ -631,6 +643,13 @@ async def start_command_handler(
update_payload = {}
if db_user.language_code != current_lang:
update_payload["language_code"] = current_lang
if db_user.telegram_id != user_id:
update_payload["telegram_id"] = user_id
if db_user.telegram_notifications_status != TELEGRAM_NOTIFICATIONS_ENABLED:
update_payload["telegram_notifications_status"] = TELEGRAM_NOTIFICATIONS_ENABLED
update_payload["telegram_notifications_checked_at"] = notification_status_now
update_payload["telegram_notifications_enabled_at"] = notification_status_now
update_payload["telegram_notifications_blocked_at"] = None
# Set referral only if not already set AND user is not currently active.
# This allows previously subscribed but currently inactive users to be attributed.
if referred_by_user_id and db_user.referred_by_id is None:
@@ -684,9 +703,16 @@ async def start_command_handler(
open_referral_page_for_existing_user = should_open_referral_from_start and is_existing_user
# Send welcome message if not disabled
if not settings.DISABLE_WELCOME_MESSAGE and not open_referral_page_for_existing_user:
if (
not settings.DISABLE_WELCOME_MESSAGE
and not open_referral_page_for_existing_user
and not notifications_start_requested
):
await message.answer(_(key="welcome", user_name=hd.quote(user.full_name)))
if notifications_start_requested:
await message.answer(_("telegram_notifications_started"), parse_mode="HTML")
# Auto-apply promo code if provided via start parameter
if promo_code_to_apply:
try:
backend/bot/services/subscription_lifecycle_notifications.py
+34 -32
View File
@@ -13,6 +13,14 @@ from bot.keyboards.inline.user_keyboards import get_subscribe_only_markup
from bot.middlewares.i18n import JsonI18n
from bot.services.email_auth_service import EmailAuthService
from bot.services.email_templates import render_subscription_lifecycle_notification
from bot.services.telegram_notifications import (
TELEGRAM_NOTIFICATIONS_BLOCKED,
TELEGRAM_NOTIFICATIONS_ENABLED,
TELEGRAM_NOTIFICATIONS_NEEDS_START,
mark_telegram_notifications_status,
normalize_telegram_notification_status,
telegram_notification_status_from_error,
)
from config.settings import Settings
from db.dal import subscription_dal
from db.models import Subscription, User
@@ -135,32 +143,31 @@ class SubscriptionLifecycleNotificationService:
chat_id = self._telegram_chat_id(user, getattr(sub, "user_id", None))
if chat_id is None:
return False
if user:
status = normalize_telegram_notification_status(
getattr(user, "telegram_notifications_status", None)
)
if status in {TELEGRAM_NOTIFICATIONS_NEEDS_START, TELEGRAM_NOTIFICATIONS_BLOCKED}:
return False
if await self._already_sent(session, sub.subscription_id, stage.key, "telegram"):
return False
try:
await self.bot.send_message(chat_id, message_text, reply_markup=markup)
except (TelegramBadRequest, TelegramForbiddenError) as exc:
if self._is_terminal_telegram_delivery_error(exc):
delivery_status = telegram_notification_status_from_error(exc)
if user and delivery_status:
await mark_telegram_notifications_status(
session,
int(user.user_id),
delivery_status,
)
if delivery_status:
logging.warning(
"Skipping subscription notification %s for unreachable Telegram user %s: %s",
stage.key,
chat_id,
exc,
)
try:
await subscription_dal.record_subscription_notification(
session,
sub.subscription_id,
self._channel_key(stage.key, "telegram"),
sent_at=sent_at,
)
except Exception:
logging.exception(
"Failed to record skipped subscription notification %s "
"for Telegram user %s",
stage.key,
chat_id,
)
return False
logging.exception(
"Failed to send subscription notification %s to Telegram user %s",
@@ -181,6 +188,18 @@ class SubscriptionLifecycleNotificationService:
self._channel_key(stage.key, "telegram"),
sent_at=sent_at,
)
if user:
status = normalize_telegram_notification_status(
getattr(user, "telegram_notifications_status", None)
)
if status != TELEGRAM_NOTIFICATIONS_ENABLED:
await mark_telegram_notifications_status(
session,
int(user.user_id),
TELEGRAM_NOTIFICATIONS_ENABLED,
telegram_id=chat_id,
checked_at=sent_at,
)
return True
async def _send_email(
@@ -333,23 +352,6 @@ class SubscriptionLifecycleNotificationService:
return chat_id
return None
@staticmethod
def _is_terminal_telegram_delivery_error(
exc: TelegramBadRequest | TelegramForbiddenError,
) -> bool:
if isinstance(exc, TelegramForbiddenError):
return True
message = str(exc).lower()
return any(
token in message
for token in (
"chat not found",
"bot was blocked",
"bot can't initiate conversation",
"user is deactivated",
)
)
@staticmethod
def _as_utc(value: Optional[datetime]) -> Optional[datetime]:
if value is None:
backend/bot/services/subscription_notification_worker.py
+33 -4
View File
@@ -19,6 +19,14 @@ from bot.services.subscription_lifecycle_notifications import (
SubscriptionNotificationStage,
)
from bot.services.subscription_service import SubscriptionService
from bot.services.telegram_notifications import (
TELEGRAM_NOTIFICATIONS_BLOCKED,
TELEGRAM_NOTIFICATIONS_ENABLED,
TELEGRAM_NOTIFICATIONS_NEEDS_START,
mark_telegram_notifications_status,
normalize_telegram_notification_status,
telegram_notification_status_from_error,
)
from bot.services.user_email_notifications import send_user_notification_email
from config.settings import Settings
from db.advisory_locks import acquire_subscription_background_sync_lock
@@ -245,6 +253,7 @@ class SubscriptionNotificationWorker:
if limit <= 0 or used < limit:
continue
delivery = await self._send_trial_traffic_depleted(
session,
sub,
used=used,
limit=limit,
@@ -284,6 +293,7 @@ class SubscriptionNotificationWorker:
async def _send_trial_traffic_depleted(
self,
session: AsyncSession,
sub: Subscription,
*,
used: int,
@@ -304,20 +314,39 @@ class SubscriptionNotificationWorker:
)
telegram_sent = False
email_sent = False
if send_telegram and user_id > 0:
telegram_chat_id = int(getattr(user, "telegram_id", 0) or user_id or 0)
telegram_status = normalize_telegram_notification_status(
getattr(user, "telegram_notifications_status", None)
)
can_try_telegram = telegram_status not in {
TELEGRAM_NOTIFICATIONS_NEEDS_START,
TELEGRAM_NOTIFICATIONS_BLOCKED,
}
if send_telegram and telegram_chat_id > 0 and can_try_telegram:
try:
await self.bot.send_message(
user_id,
telegram_chat_id,
message_text,
reply_markup=get_subscribe_only_markup(lang, self.i18n),
parse_mode="HTML",
)
telegram_sent = True
except Exception:
except Exception as exc:
status = telegram_notification_status_from_error(exc)
if status and user and user_id:
await mark_telegram_notifications_status(session, user_id, status)
logging.exception(
"Failed to send trial traffic depleted warning to user %s",
user_id,
telegram_chat_id,
)
else:
if user and telegram_status != TELEGRAM_NOTIFICATIONS_ENABLED and user_id:
await mark_telegram_notifications_status(
session,
user_id,
TELEGRAM_NOTIFICATIONS_ENABLED,
telegram_id=telegram_chat_id,
)
if send_email and user:
email_sent = await send_user_notification_email(
settings=self.settings,
backend/bot/services/telegram_notifications.py
+240
View File
@@ -0,0 +1,240 @@
import logging
from datetime import datetime, timezone
from typing import Any, Optional
from aiogram import Bot
from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup, WebAppInfo
from sqlalchemy.ext.asyncio import AsyncSession
from bot.middlewares.i18n import JsonI18n
from config.settings import Settings
from db.dal import user_dal
from db.models import User
logger = logging.getLogger(__name__)
TELEGRAM_NOTIFICATIONS_UNKNOWN = "unknown"
TELEGRAM_NOTIFICATIONS_ENABLED = "enabled"
TELEGRAM_NOTIFICATIONS_NEEDS_START = "needs_start"
TELEGRAM_NOTIFICATIONS_BLOCKED = "blocked"
TELEGRAM_NOTIFICATION_STATUSES = {
TELEGRAM_NOTIFICATIONS_UNKNOWN,
TELEGRAM_NOTIFICATIONS_ENABLED,
TELEGRAM_NOTIFICATIONS_NEEDS_START,
TELEGRAM_NOTIFICATIONS_BLOCKED,
}
def normalize_telegram_notification_status(value: Optional[str]) -> str:
status = str(value or "").strip().lower()
return status if status in TELEGRAM_NOTIFICATION_STATUSES else TELEGRAM_NOTIFICATIONS_UNKNOWN
def telegram_notifications_enabled(user: Optional[User]) -> bool:
return (
bool(getattr(user, "telegram_id", None))
and normalize_telegram_notification_status(
getattr(user, "telegram_notifications_status", None)
)
== TELEGRAM_NOTIFICATIONS_ENABLED
)
def telegram_notifications_need_prompt(user: Optional[User]) -> bool:
status = normalize_telegram_notification_status(
getattr(user, "telegram_notifications_status", None)
)
return bool(getattr(user, "telegram_id", None)) and status in {
TELEGRAM_NOTIFICATIONS_NEEDS_START,
TELEGRAM_NOTIFICATIONS_BLOCKED,
}
def telegram_notifications_start_link(bot_username: Optional[str]) -> Optional[str]:
username = str(bot_username or "").strip().lstrip("@")
if not username or username == "your_bot_username":
return None
return f"https://t.me/{username}?start=notifications"
def telegram_notification_status_from_error(exc: Exception) -> Optional[str]:
if isinstance(exc, TelegramForbiddenError):
return TELEGRAM_NOTIFICATIONS_BLOCKED
if not isinstance(exc, TelegramBadRequest):
return None
message = str(exc).lower()
if any(
token in message
for token in (
"bot was blocked",
"user is deactivated",
"forbidden",
)
):
return TELEGRAM_NOTIFICATIONS_BLOCKED
if any(
token in message
for token in (
"chat not found",
"bot can't initiate conversation",
"bot can't initiate",
"user not found",
)
):
return TELEGRAM_NOTIFICATIONS_NEEDS_START
return None
async def mark_telegram_notifications_status(
session: AsyncSession,
user_id: int,
status: str,
*,
telegram_id: Optional[int] = None,
checked_at: Optional[datetime] = None,
) -> Optional[User]:
normalized = normalize_telegram_notification_status(status)
now = checked_at or datetime.now(timezone.utc)
update_data: dict[str, Any] = {
"telegram_notifications_status": normalized,
"telegram_notifications_checked_at": now,
}
if telegram_id:
update_data["telegram_id"] = int(telegram_id)
if normalized == TELEGRAM_NOTIFICATIONS_ENABLED:
update_data["telegram_notifications_enabled_at"] = now
update_data["telegram_notifications_blocked_at"] = None
elif normalized == TELEGRAM_NOTIFICATIONS_BLOCKED:
update_data["telegram_notifications_blocked_at"] = now
return await user_dal.update_user(session, user_id, update_data)
async def mark_telegram_notifications_enabled_for_telegram_user(
session: AsyncSession,
telegram_id: int,
) -> Optional[User]:
db_user = await user_dal.get_user_by_telegram_id(session, telegram_id)
if not db_user:
db_user = await user_dal.get_user_by_id(session, telegram_id)
if not db_user:
return None
return await mark_telegram_notifications_status(
session,
int(db_user.user_id),
TELEGRAM_NOTIFICATIONS_ENABLED,
telegram_id=telegram_id,
)
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(
*,
session: AsyncSession,
bot: Bot,
settings: Settings,
i18n: Optional[JsonI18n],
user: User,
bot_username: Optional[str] = None,
force: bool = False,
) -> dict[str, Any]:
telegram_id = getattr(user, "telegram_id", None)
if not telegram_id:
return {
"ok": False,
"status": TELEGRAM_NOTIFICATIONS_UNKNOWN,
"start_link": telegram_notifications_start_link(bot_username),
}
current_status = normalize_telegram_notification_status(
getattr(user, "telegram_notifications_status", None)
)
if current_status == TELEGRAM_NOTIFICATIONS_ENABLED and not force:
return {
"ok": True,
"status": TELEGRAM_NOTIFICATIONS_ENABLED,
"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:
await bot.send_message(
int(telegram_id),
text,
reply_markup=_probe_keyboard(settings, i18n, language),
disable_web_page_preview=True,
)
except Exception as exc:
status = telegram_notification_status_from_error(exc)
if status:
await mark_telegram_notifications_status(session, int(user.user_id), status)
return {
"ok": False,
"status": status,
"start_link": telegram_notifications_start_link(bot_username),
}
logger.warning(
"Telegram notification probe failed for user %s / telegram %s: %s",
user.user_id,
telegram_id,
exc,
)
await mark_telegram_notifications_status(
session,
int(user.user_id),
TELEGRAM_NOTIFICATIONS_UNKNOWN,
)
return {
"ok": False,
"status": TELEGRAM_NOTIFICATIONS_UNKNOWN,
"start_link": telegram_notifications_start_link(bot_username),
}
await mark_telegram_notifications_status(
session,
int(user.user_id),
TELEGRAM_NOTIFICATIONS_ENABLED,
telegram_id=int(telegram_id),
)
return {
"ok": True,
"status": TELEGRAM_NOTIFICATIONS_ENABLED,
"start_link": telegram_notifications_start_link(bot_username),
}