diff --git a/src/bot.py b/src/bot.py
index aeb051c..32a3e01 100644
--- a/src/bot.py
+++ b/src/bot.py
@@ -103,8 +103,13 @@ def __init__(
.concurrent_updates(True)
.build()
)
+ # Captured before run_polling() ever runs, on the main thread, so this
+ # is the exact loop object run_polling() and Telethon's client (see
+ # TgClient) end up driving too - see the asyncio migration plan for why
+ # that matters.
+ self.loop = asyncio.get_event_loop()
self.telegram_sender = sender.TelegramSender(
- bot=self.application.bot, tg_config=tg_config
+ bot=self.application.bot, tg_config=tg_config, loop=self.loop
)
try:
self.app_context = AppContext(config_manager, skip_db_update)
diff --git a/src/jobs/tg_analytics_report_job.py b/src/jobs/tg_analytics_report_job.py
index 9c5004d..6a37d73 100644
--- a/src/jobs/tg_analytics_report_job.py
+++ b/src/jobs/tg_analytics_report_job.py
@@ -19,25 +19,9 @@ class TgAnalyticsReportJob(BaseJob):
def _execute(
app_context: AppContext, send: Callable[[str], None], called_from_handler=False
):
- app_context.tg_client.api_client.loop.run_until_complete(
- app_context.tg_client.api_client.connect()
+ stats, entity, messages = app_context.tg_client.get_channel_stats(
+ app_context.tg_client.channel
)
- stats = app_context.tg_client.api_client.loop.run_until_complete(
- app_context.tg_client.api_client.get_stats(app_context.tg_client.channel)
- )
- entity = app_context.tg_client.api_client.loop.run_until_complete(
- app_context.tg_client.api_client.get_entity(app_context.tg_client.channel)
- )
- messages = app_context.tg_client.api_client.loop.run_until_complete(
- app_context.tg_client.api_client.get_messages(
- app_context.tg_client.channel,
- ids=[
- message_stats.msg_id
- for message_stats in stats.recent_message_interactions
- ],
- )
- )
- app_context.tg_client.api_client.disconnect()
post_interactions = TgAnalyticsReportJob._deduplicate_albums(
stats.recent_message_interactions, messages
)
diff --git a/src/tg/sender.py b/src/tg/sender.py
index 1119fc0..0f3a6ac 100644
--- a/src/tg/sender.py
+++ b/src/tg/sender.py
@@ -1,14 +1,11 @@
"""Sends messages"""
import asyncio
-import json
import logging
import re
import time
-from typing import Callable, List
+from typing import Callable, List, Optional
-import nest_asyncio
-import requests
import telegram
from ..app_context import AppContext
@@ -17,21 +14,33 @@
logger = logging.getLogger(__name__)
+SEND_READ_TIMEOUT_SEC = 10
+SEND_WRITE_TIMEOUT_SEC = 10
+SEND_CONNECT_TIMEOUT_SEC = 10
+
+# Outer bridge timeout for the whole send (possibly several paragraphs/photos
+# plus MESSAGE_DELAY_SEC delays between them, plus a possible error-fallback
+# round-trip), kept generously above what a legitimate worst case should take.
+SEND_BRIDGE_TIMEOUT_SEC = 60
+
class TelegramSender(Singleton):
def __init__(
self,
bot: telegram.Bot = None,
tg_config: dict = None,
+ loop: Optional[asyncio.AbstractEventLoop] = None,
):
if self.was_initialized():
return
- if bot is None or tg_config is None:
+ if bot is None or tg_config is None or loop is None:
raise ValueError(
- "On first TelegramSender initialization bot and tg_config must be not None"
+ "On first TelegramSender initialization bot, tg_config and loop "
+ "must be not None"
)
self.bot = bot
+ self._loop = loop
self._tg_config = tg_config
self._update_from_config()
logger.info("TelegramSender successfully initialized")
@@ -75,19 +84,51 @@ def send_to_chat_id(self, message_text: str, chat_id: int, **kwargs) -> bool:
"""
Sends a message to a single chat_id.
"""
+ coro = self._send_to_chat_id_async(message_text, chat_id, **kwargs)
+ try:
+ running_loop = asyncio.get_running_loop()
+ except RuntimeError:
+ running_loop = None
+
+ if running_loop is self._loop:
+ # We're being called synchronously from code that's itself
+ # executing on this exact loop's thread right now (e.g. logging
+ # triggered from an async handler wrapper running directly on
+ # PTB's loop, not routed through the handler executor). Blocking
+ # here would deadlock: the loop can't advance to run our
+ # coroutine while this very callback -- the only thing that could
+ # keep the loop moving -- is stuck waiting on it. Schedule it to
+ # run once we yield back instead of waiting for a result.
+ self._loop.create_task(coro)
+ return None
+
+ if self._loop.is_running():
+ # Normal case: called from a handler/scheduler thread while PTB is
+ # polling on its own loop in the main thread.
+ future = asyncio.run_coroutine_threadsafe(coro, self._loop)
+ return future.result(timeout=SEND_BRIDGE_TIMEOUT_SEC)
+ # Called before run_polling() has started the loop (e.g. the startup
+ # message in app.py, or the AppContext-init-failure error log in
+ # bot.py). We're on the same (main) thread that owns this loop and
+ # nothing else is driving it yet, so run it directly rather than
+ # scheduling work onto a loop nobody is pumping -- that would deadlock.
+ return self._loop.run_until_complete(coro)
+
+ async def _send_to_chat_id_async(
+ self, message_text: str, chat_id: int, **kwargs
+ ) -> bool:
if "poll_options" in kwargs:
# Handle poll sending
poll_options = kwargs["poll_options"]
- resp = requests.get(
- url=f"https://api.telegram.org/bot{self._tg_config['token']}/sendPoll",
- params={
- "chat_id": chat_id,
- "question": poll_options["question"],
- "options": json.dumps(poll_options["options"]),
- "is_anonymous": poll_options["is_anonymous"],
- },
+ await self.bot.send_poll(
+ chat_id=chat_id,
+ question=poll_options["question"],
+ options=poll_options["options"],
+ is_anonymous=poll_options["is_anonymous"],
+ read_timeout=SEND_READ_TIMEOUT_SEC,
+ write_timeout=SEND_WRITE_TIMEOUT_SEC,
+ connect_timeout=SEND_CONNECT_TIMEOUT_SEC,
)
- resp.raise_for_status()
return True
if not message_text:
@@ -98,59 +139,60 @@ def send_to_chat_id(self, message_text: str, chat_id: int, **kwargs) -> bool:
if ".png" in message_text:
for pict in re.findall(r"\S*\.png", message_text):
- self.bot.send_photo(
- photo=open(pict, "rb"),
- chat_id=chat_id,
- disable_notification=self.is_silent,
- **kwargs,
- )
+ with open(pict, "rb") as photo:
+ await self.bot.send_photo(
+ photo=photo,
+ chat_id=chat_id,
+ disable_notification=self.is_silent,
+ read_timeout=SEND_READ_TIMEOUT_SEC,
+ write_timeout=SEND_WRITE_TIMEOUT_SEC,
+ connect_timeout=SEND_CONNECT_TIMEOUT_SEC,
+ **kwargs,
+ )
message_text = re.sub(r"\S*\.png", "", message_text)
if message_text != "":
try:
- loop = asyncio.get_event_loop()
- nest_asyncio.apply(loop)
messages = paragraphs_to_messages([message_text.strip()])
for i, message in enumerate(messages):
if i > 0:
- time.sleep(MESSAGE_DELAY_SEC)
+ await asyncio.sleep(MESSAGE_DELAY_SEC)
if message.startswith("") and "" not in message:
message = message + ""
elif message.endswith("") and "" not in message:
message = "" + message
- payload = {
- "text": message,
- "chat_id": chat_id,
- "silent": self.is_silent,
- "no_webpage": self.disable_web_page_preview,
- "parse_mode": "html",
- }
+ send_kwargs = {}
if "reply_markup" in kwargs:
- payload["reply_markup"] = kwargs["reply_markup"].to_json()
- resp = requests.get(
- url=f"https://api.telegram.org/bot{self._tg_config['token']}/sendMessage",
- json=payload,
+ send_kwargs["reply_markup"] = kwargs["reply_markup"]
+ await self.bot.send_message(
+ text=message,
+ chat_id=chat_id,
+ disable_notification=self.is_silent,
+ disable_web_page_preview=self.disable_web_page_preview,
+ parse_mode=telegram.constants.ParseMode.HTML,
+ read_timeout=SEND_READ_TIMEOUT_SEC,
+ write_timeout=SEND_WRITE_TIMEOUT_SEC,
+ connect_timeout=SEND_CONNECT_TIMEOUT_SEC,
+ **send_kwargs,
)
- resp.raise_for_status()
return True
except telegram.error.TelegramError as e:
logger.error(f"Could not send a message to {chat_id}", exc_info=e)
- loop = asyncio.get_event_loop()
- nest_asyncio.apply(loop)
chat_name = AppContext().db_client.get_chat_name(chat_id)
for error_logs_recipient in self.error_logs_recipients:
try:
# Try redirect unsended message to error_logs_recipients
- loop.run_until_complete(
- self.bot.send_message(
- text=f"Unsended message to {chat_name} {chat_id}\n{message}"[
- : telegram.constants.MessageLimit.MAX_TEXT_LENGTH
- ],
- chat_id=error_logs_recipient,
- disable_notification=self.is_silent,
- disable_web_page_preview=self.disable_web_page_preview,
- parse_mode=telegram.constants.ParseMode.HTML,
- **kwargs,
- )
+ await self.bot.send_message(
+ text=f"Unsended message to {chat_name} {chat_id}\n{message}"[
+ : telegram.constants.MessageLimit.MAX_TEXT_LENGTH
+ ],
+ chat_id=error_logs_recipient,
+ disable_notification=self.is_silent,
+ disable_web_page_preview=self.disable_web_page_preview,
+ parse_mode=telegram.constants.ParseMode.HTML,
+ read_timeout=SEND_READ_TIMEOUT_SEC,
+ write_timeout=SEND_WRITE_TIMEOUT_SEC,
+ connect_timeout=SEND_CONNECT_TIMEOUT_SEC,
+ **kwargs,
)
except telegram.error.TelegramError as e:
logger.error(
@@ -164,16 +206,17 @@ def send_to_chat_id(self, message_text: str, chat_id: int, **kwargs) -> bool:
if "Can't parse entities" in e.message:
try:
# Try sending the plain-text version
- loop.run_until_complete(
- self.bot.send_message(
- text=f"Unsended message to {chat_name} {chat_id}\n{message}"[
- : telegram.constants.MessageLimit.MAX_TEXT_LENGTH
- ],
- chat_id=error_logs_recipient,
- disable_notification=self.is_silent,
- disable_web_page_preview=self.disable_web_page_preview,
- **kwargs,
- )
+ await self.bot.send_message(
+ text=f"Unsended message to {chat_name} {chat_id}\n{message}"[
+ : telegram.constants.MessageLimit.MAX_TEXT_LENGTH
+ ],
+ chat_id=error_logs_recipient,
+ disable_notification=self.is_silent,
+ disable_web_page_preview=self.disable_web_page_preview,
+ read_timeout=SEND_READ_TIMEOUT_SEC,
+ write_timeout=SEND_WRITE_TIMEOUT_SEC,
+ connect_timeout=SEND_CONNECT_TIMEOUT_SEC,
+ **kwargs,
)
return True
except telegram.error.TelegramError as e:
diff --git a/src/tg/tg_client.py b/src/tg/tg_client.py
index e06b0c0..4ef04a9 100644
--- a/src/tg/tg_client.py
+++ b/src/tg/tg_client.py
@@ -1,5 +1,6 @@
+import asyncio
import logging
-from typing import List, Optional, Tuple
+from typing import Optional, Tuple
from telethon import TelegramClient
from telethon.sessions import StringSession
@@ -9,6 +10,19 @@
logger = logging.getLogger(__name__)
+CONNECT_TIMEOUT_SEC = 10
+ENTITY_TIMEOUT_SEC = 10
+STATS_TIMEOUT_SEC = 15
+MESSAGES_TIMEOUT_SEC = 15
+DISCONNECT_TIMEOUT_SEC = 10
+
+# Outer bridge timeouts (see run_coroutine_threadsafe calls below) are kept above
+# the sum of the inner per-call timeouts they wrap, so a legitimate worst-case
+# run doesn't spuriously trip the outer bound right as the inner one would have
+# reported the real failure.
+RESOLVE_USERNAME_BRIDGE_TIMEOUT_SEC = 40
+CHANNEL_STATS_BRIDGE_TIMEOUT_SEC = 90
+
class TgClient(Singleton):
def __init__(self, tg_config=None):
@@ -29,24 +43,54 @@ def _update_from_config(self):
StringSession(self._tg_config["api_session"]),
self._tg_config["api_id"],
self._tg_config["api_hash"],
+ # We connect/disconnect explicitly around each short-lived
+ # operation below rather than holding one persistent connection,
+ # so we don't need (or want) Telethon's own background
+ # auto-reconnect: it spawns its own reconnect task on connection
+ # errors, independent of our connect()/disconnect() calls, and can
+ # race with an explicit disconnect() happening around the same
+ # time (Telethon's own source acknowledges this is unresolved -
+ # see the TODO in MTProtoSender._start_reconnect).
+ auto_reconnect=False,
)
- # we need this to properly reauth in case the tokens need to be updated
- # we need "with" to open and close the event loop
- # removed as part of v20.0 migration
- # with self.api_client as client:
- # client(functions.auth.ResetAuthorizationsRequest())
self.sysblok_chats = self._tg_config["sysblok_chats"]
self.channel = self._tg_config["channel"]
- def _get_chat_users(self, chat_id: str) -> List[User]:
- with self.api_client:
- users = self.api_client.loop.run_until_complete(
- self.api_client.get_participants(chat_id)
+ def _run_coro(self, coro, timeout: float):
+ """
+ Bridges a coroutine onto self.api_client.loop.
+
+ If that loop is already running (the normal case: called from a
+ handler/scheduler thread while PTB is polling on its own loop in the
+ main thread), schedules it via the thread-safe run_coroutine_threadsafe
+ and blocks the calling thread for the result. If the loop isn't
+ running yet (e.g. called during startup, before run_polling() has
+ started it), we're on the same thread that owns the loop and nothing
+ else is driving it, so run it directly instead of scheduling work onto
+ a loop nobody is pumping - that would deadlock.
+ """
+ try:
+ running_loop = asyncio.get_running_loop()
+ except RuntimeError:
+ running_loop = None
+
+ if running_loop is self.api_client.loop:
+ # Called synchronously from code already executing on this exact
+ # loop's thread - blocking here would deadlock the same way a
+ # foreign-thread call would if nothing pumped the loop, except
+ # here it's this very callback that would need to. Unlike sender.py
+ # this method's return value is actually consumed by callers, so
+ # there's no safe fire-and-forget fallback - fail loudly instead of
+ # silently returning a result callers can't use.
+ raise RuntimeError(
+ "TgClient method called reentrantly from its own event loop's "
+ "thread - this would deadlock if we waited for a result."
)
- return users
- def get_main_chat_users(self) -> List[User]:
- return self._get_chat_users(self.sysblok_chats["main_chat"])
+ if self.api_client.loop.is_running():
+ future = asyncio.run_coroutine_threadsafe(coro, self.api_client.loop)
+ return future.result(timeout=timeout)
+ return self.api_client.loop.run_until_complete(coro)
def resolve_telegram_username(self, username: str) -> Optional[Tuple[int, str]]:
"""
@@ -58,20 +102,32 @@ def resolve_telegram_username(self, username: str) -> Optional[Tuple[int, str]]:
Returns:
Tuple of (user_id, username_without_@) or None if not found
"""
- # Normalize: remove @ for telethon API
- normalized = username.lstrip("@")
+ try:
+ return self._run_coro(
+ self._resolve_telegram_username_async(username),
+ timeout=RESOLVE_USERNAME_BRIDGE_TIMEOUT_SEC,
+ )
+ except Exception as e:
+ logger.warning(f"Failed to resolve username {username}: {e}", exc_info=True)
+ return None
+ async def _resolve_telegram_username_async(
+ self, username: str
+ ) -> Optional[Tuple[int, str]]:
+ normalized = username.lstrip("@")
try:
# Check if client is already connected
was_connected = self.api_client.is_connected()
if not was_connected:
# Connect if not already connected
- self.api_client.loop.run_until_complete(self.api_client.connect())
+ await asyncio.wait_for(
+ self.api_client.connect(), timeout=CONNECT_TIMEOUT_SEC
+ )
try:
- entity = self.api_client.loop.run_until_complete(
- self.api_client.get_entity(normalized)
+ entity = await asyncio.wait_for(
+ self.api_client.get_entity(normalized), timeout=ENTITY_TIMEOUT_SEC
)
# Only process User entities, skip channels, groups, bots, etc.
@@ -82,15 +138,56 @@ def resolve_telegram_username(self, username: str) -> Optional[Tuple[int, str]]:
# Entity is not a User (could be Channel, Chat, Bot, etc.)
entity_type = type(entity).__name__
logger.info(
- f"Username {username} resolved to {entity_type} (not a User) - skipping"
+ f"Username {username} resolved to {entity_type} "
+ "(not a User) - skipping"
)
return None
finally:
# Only disconnect if we connected it
if not was_connected:
- self.api_client.disconnect()
+ await asyncio.wait_for(
+ self.api_client.disconnect(), timeout=DISCONNECT_TIMEOUT_SEC
+ )
except Exception as e:
logger.warning(f"Failed to resolve username {username}: {e}", exc_info=True)
return None
return None
+
+ def get_channel_stats(self, channel: str) -> Tuple:
+ """
+ Connects, fetches channel stats/entity/recent-message info, and
+ disconnects, all in one session.
+
+ Returns:
+ Tuple of (stats, entity, messages)
+ """
+ return self._run_coro(
+ self._get_channel_stats_async(channel),
+ timeout=CHANNEL_STATS_BRIDGE_TIMEOUT_SEC,
+ )
+
+ async def _get_channel_stats_async(self, channel: str) -> Tuple:
+ await asyncio.wait_for(self.api_client.connect(), timeout=CONNECT_TIMEOUT_SEC)
+ try:
+ stats = await asyncio.wait_for(
+ self.api_client.get_stats(channel), timeout=STATS_TIMEOUT_SEC
+ )
+ entity = await asyncio.wait_for(
+ self.api_client.get_entity(channel), timeout=ENTITY_TIMEOUT_SEC
+ )
+ messages = await asyncio.wait_for(
+ self.api_client.get_messages(
+ channel,
+ ids=[
+ message_stats.msg_id
+ for message_stats in stats.recent_message_interactions
+ ],
+ ),
+ timeout=MESSAGES_TIMEOUT_SEC,
+ )
+ return stats, entity, messages
+ finally:
+ await asyncio.wait_for(
+ self.api_client.disconnect(), timeout=DISCONNECT_TIMEOUT_SEC
+ )
diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py
index 191eea1..516eac5 100644
--- a/tests/unit/conftest.py
+++ b/tests/unit/conftest.py
@@ -1,3 +1,4 @@
+import asyncio
import logging
import os
from typing import List
@@ -87,7 +88,9 @@ def send_to_chat_id(self, message_text: str, chat_id: int, **kwargs):
monkeypatch.setattr(TelegramSender, "send_to_chat_id", send_to_chat_id)
return TelegramSender(
- bot=mock_telegram_bot, tg_config=mock_config_manager.get_telegram_config()
+ bot=mock_telegram_bot,
+ tg_config=mock_config_manager.get_telegram_config(),
+ loop=asyncio.new_event_loop(),
)