File size: 6,333 Bytes
9d614f1 | 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 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | """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 # Telegram hard limit is 4096 characters
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 = "?"
# ------------------------------------------------------------------ api
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.")
# ------------------------------------------------------------------ handling
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 |