File size: 2,949 Bytes
44083db af49805 44083db af49805 44083db af49805 44083db af49805 44083db | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | import logging
import threading
from typing import Callable, Optional
import telebot
class TelegramChatService:
def __init__(self, llm_factory: Callable, token: str, logger: Optional[logging.Logger] = None, jira_connector=None):
self._llm_factory = llm_factory
self._token = token
self._logger = logger or logging.getLogger(__name__)
self._jira_connector = jira_connector
self._bot: Optional[telebot.TeleBot] = None
self._llm = None
self._thread: Optional[threading.Thread] = None
self._lock = threading.Lock()
self._running = False
def _ensure_bot(self) -> telebot.TeleBot:
if self._bot is not None:
return self._bot
bot = telebot.TeleBot(self._token, parse_mode=None)
@bot.message_handler(commands=["start"])
def on_start(message):
bot.reply_to(message, "Telegram chat is ready. Send me a message.")
@bot.message_handler(func=lambda message: True)
def on_text(message):
text = (message.text or "").strip()
if not text:
bot.reply_to(message, "Please send a non-empty message.")
return
try:
answer = None
if self._jira_connector is not None:
answer = self._jira_connector.maybe_handle(text)
if answer is None:
if self._llm is None:
self._llm = self._llm_factory()
answer = self._llm.invoke(text)
bot.reply_to(message, str(answer).strip())
except Exception as exc:
self._logger.exception("Failed to handle Telegram message")
bot.reply_to(message, f"Error: {exc}")
self._bot = bot
return bot
def _poll(self) -> None:
bot = self._ensure_bot()
try:
bot.delete_webhook(drop_pending_updates=False)
bot.infinity_polling(timeout=20, long_polling_timeout=20, skip_pending=True)
except Exception:
self._logger.exception("Telegram polling crashed")
finally:
with self._lock:
self._running = False
def start(self) -> bool:
with self._lock:
if self._running:
return False
if not self._token:
raise ValueError("Missing TELEGRAM_TOKEN")
self._running = True
self._thread = threading.Thread(target=self._poll, daemon=True)
self._thread.start()
return True
def stop(self) -> bool:
with self._lock:
if not self._running:
return False
self._running = False
if self._bot is not None:
self._bot.stop_polling()
return True
@property
def is_running(self) -> bool:
with self._lock:
return self._running
|