| """Live Telegram bot (PRIVATE to Serving_Module) - started when TELEGRAM_BOT_TOKEN is set in .env. |
| |
| Plain HTTPS long-polling through `requests` (no extra dependency, works in Codespaces / Kaggle / a 1 GB VPS). |
| Commands: /thinking /non_thinking /max_thinking (per-chat mode) /mode /teach सवाल | जवाब /clear /help |
| Every other message is answered by the model in the chat's current mode. |
| """ |
| from __future__ import annotations |
|
|
| import threading |
| import time |
| from typing import Dict, List, Optional |
|
|
| from .inference_engine import MODES, InferenceEngine, normalise_mode |
|
|
| API = "https://api.telegram.org" |
| LIMIT = 4000 |
|
|
| HELP = ("RSS AI bot\n" |
| "/thinking - reason step by step first\n" |
| "/non_thinking - fast direct answers (default)\n" |
| "/max_thinking - longest, most careful reasoning\n" |
| "/mode - show the current mode\n" |
| "/teach सवाल | जवाब - teach me something right now\n" |
| "/clear - forget this conversation") |
|
|
|
|
| def _chunks(text: str, size: int = LIMIT) -> List[str]: |
| text = text.strip() or "(empty answer)" |
| out: List[str] = [] |
| while len(text) > size: |
| cut = text.rfind("\n", 0, size) |
| cut = cut if cut > size // 2 else size |
| out.append(text[:cut]) |
| text = text[cut:].lstrip("\n") |
| out.append(text) |
| return out |
|
|
|
|
| class TelegramGateway: |
| def __init__(self, token: str, chat_id: str, engine: InferenceEngine, session=None, api_base: str = API, |
| poll_timeout: int = 25): |
| import requests |
| self.token, self.chat_id, self.engine = token.strip(), str(chat_id or "").strip(), engine |
| self.http = session or requests.Session() |
| self.base = f"{api_base}/bot{self.token}" |
| self.poll_timeout = poll_timeout |
| self.modes: Dict[str, str] = {} |
| self.offset = 0 |
| self.stop = threading.Event() |
| self.bot_name = "?" |
|
|
| |
| def call(self, method: str, payload: Optional[dict] = None, timeout: int = 40) -> dict: |
| reply = self.http.post(f"{self.base}/{method}", json=payload or {}, timeout=timeout) |
| data = reply.json() |
| if not data.get("ok"): |
| raise RuntimeError(f"telegram {method} failed: {data.get('description', reply.status_code)}") |
| return data |
|
|
| def verify(self) -> bool: |
| try: |
| self.bot_name = self.call("getMe", timeout=15)["result"].get("username", "?") |
| return True |
| except Exception as exc: |
| print(f" [telegram] token check failed: {type(exc).__name__}: {exc}") |
| return False |
|
|
| def send(self, chat_id, text: str) -> None: |
| for part in _chunks(text): |
| try: |
| self.call("sendMessage", {"chat_id": chat_id, "text": part}, timeout=30) |
| except Exception as exc: |
| print(f" [telegram] send failed: {exc}") |
|
|
| def notify_start(self) -> None: |
| if self.chat_id: |
| self.send(self.chat_id, "RSS AI is online ✅\nModes: /thinking /non_thinking /max_thinking\nSend any message to chat.") |
|
|
| |
| def handle(self, message: dict) -> None: |
| chat = message.get("chat", {}).get("id") |
| text = (message.get("text") or "").strip() |
| if chat is None or not text: |
| return |
| key = str(chat) |
| mode = self.modes.get(key, "non_thinking") |
| if text.startswith("/"): |
| command, _, rest = text.partition(" ") |
| command = command.split("@", 1)[0].lower() |
| if command in ("/start", "/help"): |
| self.send(chat, HELP + f"\n\ncurrent mode: {mode}") |
| elif command.lstrip("/") in MODES: |
| self.modes[key] = command.lstrip("/") |
| self.send(chat, f"mode -> {self.modes[key]}") |
| elif command == "/mode": |
| chosen = normalise_mode(rest) |
| if chosen: |
| self.modes[key] = chosen |
| self.send(chat, f"mode -> {self.modes.get(key, mode)}") |
| elif command == "/clear": |
| self.engine.reset(key) |
| self.send(chat, "conversation cleared.") |
| elif command == "/teach": |
| question, sep, answer = rest.partition("|") |
| self.send(chat, self.engine.teach(question, answer) if sep else "format: /teach सवाल | जवाब") |
| else: |
| self.send(chat, "unknown command - /help") |
| return |
| try: |
| self.call("sendChatAction", {"chat_id": chat, "action": "typing"}, timeout=10) |
| except Exception: |
| pass |
| try: |
| result = self.engine.generate(text, mode=mode, session=key) |
| except Exception as exc: |
| self.send(chat, f"⚠️ error: {type(exc).__name__}: {exc}") |
| return |
| reply = result["answer"] |
| if result["thinking"]: |
| reply = f"💭 {result['thinking']}\n\n✅ {result['answer']}" |
| self.send(chat, reply) |
|
|
| def run(self, max_updates: Optional[int] = None) -> None: |
| print(f" [telegram] bot @{self.bot_name} is live - press Ctrl+C to stop") |
| handled, backoff = 0, 2 |
| while not self.stop.is_set(): |
| try: |
| data = self.call("getUpdates", {"offset": self.offset, "timeout": self.poll_timeout, |
| "allowed_updates": ["message"]}, timeout=self.poll_timeout + 15) |
| backoff = 2 |
| except KeyboardInterrupt: |
| break |
| except Exception as exc: |
| print(f" [telegram] poll error ({type(exc).__name__}: {exc}); retry in {backoff}s") |
| time.sleep(backoff) |
| backoff = min(backoff * 2, 60) |
| continue |
| for update in data.get("result", []): |
| self.offset = update["update_id"] + 1 |
| if "message" in update: |
| self.handle(update["message"]) |
| handled += 1 |
| if max_updates is not None and handled >= max_updates: |
| return |
